1 /* 2 * 3 * sched-messaging.c 4 * 5 * messaging: Benchmark for scheduler and IPC mechanisms 6 * 7 * Based on hackbench by Rusty Russell <rusty@rustcorp.com.au> 8 * Ported to perf by Hitoshi Mitake <mitake@dcl.info.waseda.ac.jp> 9 * 10 */ 11 12 #include "../perf.h" 13 #include "../util/util.h" 14 #include "../util/parse-options.h" 15 #include "../builtin.h" 16 #include "bench.h" 17 18 /* Test groups of 20 processes spraying to 20 receivers */ 19 #include <pthread.h> 20 #include <stdio.h> 21 #include <stdlib.h> 22 #include <string.h> 23 #include <errno.h> 24 #include <unistd.h> 25 #include <sys/types.h> 26 #include <sys/socket.h> 27 #include <sys/wait.h> 28 #include <sys/time.h> 29 #include <sys/poll.h> 30 #include <limits.h> 31 32 #define DATASIZE 100 33 34 static bool use_pipes = false; 35 static unsigned int loops = 100; 36 static bool thread_mode = false; 37 static unsigned int num_groups = 10; 38 39 struct sender_context { 40 unsigned int num_fds; 41 int ready_out; 42 int wakefd; 43 int out_fds[0]; 44 }; 45 46 struct receiver_context { 47 unsigned int num_packets; 48 int in_fds[2]; 49 int ready_out; 50 int wakefd; 51 }; 52 53 static void barf(const char *msg) 54 { 55 fprintf(stderr, "%s (error: %s)\n", msg, strerror(errno)); 56 exit(1); 57 } 58 59 static void fdpair(int fds[2]) 60 { 61 if (use_pipes) { 62 if (pipe(fds) == 0) 63 return; 64 } else { 65 if (socketpair(AF_UNIX, SOCK_STREAM, 0, fds) == 0) 66 return; 67 } 68 69 barf(use_pipes ? "pipe()" : "socketpair()"); 70 } 71 72 /* Block until we're ready to go */ 73 static void ready(int ready_out, int wakefd) 74 { 75 char dummy; 76 struct pollfd pollfd = { .fd = wakefd, .events = POLLIN }; 77 78 /* Tell them we're ready. */ 79 if (write(ready_out, &dummy, 1) != 1) 80 barf("CLIENT: ready write"); 81 82 /* Wait for "GO" signal */ 83 if (poll(&pollfd, 1, -1) != 1) 84 barf("poll"); 85 } 86 87 /* Sender sprays loops messages down each file descriptor */ 88 static void *sender(struct sender_context *ctx) 89 { 90 char data[DATASIZE]; 91 unsigned int i, j; 92 93 ready(ctx->ready_out, ctx->wakefd); 94 95 /* Now pump to every receiver. */ 96 for (i = 0; i < loops; i++) { 97 for (j = 0; j < ctx->num_fds; j++) { 98 int ret, done = 0; 99 100 again: 101 ret = write(ctx->out_fds[j], data + done, 102 sizeof(data)-done); 103 if (ret < 0) 104 barf("SENDER: write"); 105 done += ret; 106 if (done < DATASIZE) 107 goto again; 108 } 109 } 110 111 return NULL; 112 } 113 114 115 /* One receiver per fd */ 116 static void *receiver(struct receiver_context* ctx) 117 { 118 unsigned int i; 119 120 if (!thread_mode) 121 close(ctx->in_fds[1]); 122 123 /* Wait for start... */ 124 ready(ctx->ready_out, ctx->wakefd); 125 126 /* Receive them all */ 127 for (i = 0; i < ctx->num_packets; i++) { 128 char data[DATASIZE]; 129 int ret, done = 0; 130 131 again: 132 ret = read(ctx->in_fds[0], data + done, DATASIZE - done); 133 if (ret < 0) 134 barf("SERVER: read"); 135 done += ret; 136 if (done < DATASIZE) 137 goto again; 138 } 139 140 return NULL; 141 } 142 143 static pthread_t create_worker(void *ctx, void *(*func)(void *)) 144 { 145 pthread_attr_t attr; 146 pthread_t childid; 147 int err; 148 149 if (!thread_mode) { 150 /* process mode */ 151 /* Fork the receiver. */ 152 switch (fork()) { 153 case -1: 154 barf("fork()"); 155 break; 156 case 0: 157 (*func) (ctx); 158 exit(0); 159 break; 160 default: 161 break; 162 } 163 164 return (pthread_t)0; 165 } 166 167 if (pthread_attr_init(&attr) != 0) 168 barf("pthread_attr_init:"); 169 170 #ifndef __ia64__ 171 if (pthread_attr_setstacksize(&attr, PTHREAD_STACK_MIN) != 0) 172 barf("pthread_attr_setstacksize"); 173 #endif 174 175 err = pthread_create(&childid, &attr, func, ctx); 176 if (err != 0) { 177 fprintf(stderr, "pthread_create failed: %s (%d)\n", 178 strerror(err), err); 179 exit(-1); 180 } 181 return childid; 182 } 183 184 static void reap_worker(pthread_t id) 185 { 186 int proc_status; 187 void *thread_status; 188 189 if (!thread_mode) { 190 /* process mode */ 191 wait(&proc_status); 192 if (!WIFEXITED(proc_status)) 193 exit(1); 194 } else { 195 pthread_join(id, &thread_status); 196 } 197 } 198 199 /* One group of senders and receivers */ 200 static unsigned int group(pthread_t *pth, 201 unsigned int num_fds, 202 int ready_out, 203 int wakefd) 204 { 205 unsigned int i; 206 struct sender_context *snd_ctx = malloc(sizeof(struct sender_context) 207 + num_fds * sizeof(int)); 208 209 if (!snd_ctx) 210 barf("malloc()"); 211 212 for (i = 0; i < num_fds; i++) { 213 int fds[2]; 214 struct receiver_context *ctx = malloc(sizeof(*ctx)); 215 216 if (!ctx) 217 barf("malloc()"); 218 219 220 /* Create the pipe between client and server */ 221 fdpair(fds); 222 223 ctx->num_packets = num_fds * loops; 224 ctx->in_fds[0] = fds[0]; 225 ctx->in_fds[1] = fds[1]; 226 ctx->ready_out = ready_out; 227 ctx->wakefd = wakefd; 228 229 pth[i] = create_worker(ctx, (void *)receiver); 230 231 snd_ctx->out_fds[i] = fds[1]; 232 if (!thread_mode) 233 close(fds[0]); 234 } 235 236 /* Now we have all the fds, fork the senders */ 237 for (i = 0; i < num_fds; i++) { 238 snd_ctx->ready_out = ready_out; 239 snd_ctx->wakefd = wakefd; 240 snd_ctx->num_fds = num_fds; 241 242 pth[num_fds+i] = create_worker(snd_ctx, (void *)sender); 243 } 244 245 /* Close the fds we have left */ 246 if (!thread_mode) 247 for (i = 0; i < num_fds; i++) 248 close(snd_ctx->out_fds[i]); 249 250 /* Return number of children to reap */ 251 return num_fds * 2; 252 } 253 254 static const struct option options[] = { 255 OPT_BOOLEAN('p', "pipe", &use_pipes, 256 "Use pipe() instead of socketpair()"), 257 OPT_BOOLEAN('t', "thread", &thread_mode, 258 "Be multi thread instead of multi process"), 259 OPT_UINTEGER('g', "group", &num_groups, "Specify number of groups"), 260 OPT_UINTEGER('l', "loop", &loops, "Specify number of loops"), 261 OPT_END() 262 }; 263 264 static const char * const bench_sched_message_usage[] = { 265 "perf bench sched messaging <options>", 266 NULL 267 }; 268 269 int bench_sched_messaging(int argc, const char **argv, 270 const char *prefix __maybe_unused) 271 { 272 unsigned int i, total_children; 273 struct timeval start, stop, diff; 274 unsigned int num_fds = 20; 275 int readyfds[2], wakefds[2]; 276 char dummy; 277 pthread_t *pth_tab; 278 279 argc = parse_options(argc, argv, options, 280 bench_sched_message_usage, 0); 281 282 pth_tab = malloc(num_fds * 2 * num_groups * sizeof(pthread_t)); 283 if (!pth_tab) 284 barf("main:malloc()"); 285 286 fdpair(readyfds); 287 fdpair(wakefds); 288 289 total_children = 0; 290 for (i = 0; i < num_groups; i++) 291 total_children += group(pth_tab+total_children, num_fds, 292 readyfds[1], wakefds[0]); 293 294 /* Wait for everyone to be ready */ 295 for (i = 0; i < total_children; i++) 296 if (read(readyfds[0], &dummy, 1) != 1) 297 barf("Reading for readyfds"); 298 299 gettimeofday(&start, NULL); 300 301 /* Kick them off */ 302 if (write(wakefds[1], &dummy, 1) != 1) 303 barf("Writing to start them"); 304 305 /* Reap them all */ 306 for (i = 0; i < total_children; i++) 307 reap_worker(pth_tab[i]); 308 309 gettimeofday(&stop, NULL); 310 311 timersub(&stop, &start, &diff); 312 313 switch (bench_format) { 314 case BENCH_FORMAT_DEFAULT: 315 printf("# %d sender and receiver %s per group\n", 316 num_fds, thread_mode ? "threads" : "processes"); 317 printf("# %d groups == %d %s run\n\n", 318 num_groups, num_groups * 2 * num_fds, 319 thread_mode ? "threads" : "processes"); 320 printf(" %14s: %lu.%03lu [sec]\n", "Total time", 321 diff.tv_sec, 322 (unsigned long) (diff.tv_usec/1000)); 323 break; 324 case BENCH_FORMAT_SIMPLE: 325 printf("%lu.%03lu\n", diff.tv_sec, 326 (unsigned long) (diff.tv_usec/1000)); 327 break; 328 default: 329 /* reaching here is something disaster */ 330 fprintf(stderr, "Unknown format:%d\n", bench_format); 331 exit(1); 332 break; 333 } 334 335 return 0; 336 } 337