1 #include <AvailabilityMacros.h>
2 #include <mach/thread_policy.h>
3
4 #include <pthread.h>
5 #include <stdio.h>
6 #include <stdlib.h>
7 #include <libgen.h>
8 #include <string.h>
9 #include <err.h>
10 #include <unistd.h>
11
12 #include <pthread.h>
13 #include <mach/mach.h>
14 #include <mach/mach_error.h>
15 #include <mach/notify.h>
16 #include <servers/bootstrap.h>
17 #include <sys/event.h>
18 #include <sys/select.h>
19 #include <sys/types.h>
20 #include <sys/time.h>
21 #include <sys/signal.h>
22 #include <errno.h>
23 #include "../unit_tests/tests_common.h"
24
25 #define MAX(A, B) ((A) < (B) ? (B) : (A))
26
27
28 typedef struct {
29 mach_msg_header_t header;
30 mach_msg_trailer_t trailer; // subtract this when sending
31 } ipc_trivial_message;
32
33 typedef struct {
34 mach_msg_header_t header;
35 u_int32_t numbers[0];
36 mach_msg_trailer_t trailer; // subtract this when sending
37 } ipc_inline_message;
38
39 typedef struct {
40 mach_msg_header_t header;
41 mach_msg_body_t body;
42 mach_msg_ool_descriptor_t descriptor;
43 mach_msg_trailer_t trailer; // subtract this when sending
44 } ipc_complex_message;
45
46 enum {
47 msg_type_trivial = 0,
48 msg_type_inline = 1,
49 msg_type_complex = 2
50 };
51
52 struct port_args {
53 int server_num;
54 int req_size;
55 mach_msg_header_t *req_msg;
56 int reply_size;
57 mach_msg_header_t *reply_msg;
58 mach_port_t port;
59 mach_port_t pset;
60 };
61
62 typedef union {
63 pid_t pid;
64 pthread_t tid;
65 } thread_id_t;
66
67 /* Global options */
68 static boolean_t verbose = FALSE;
69 static boolean_t affinity = FALSE;
70 static boolean_t timeshare = FALSE;
71 static boolean_t threaded = FALSE;
72 static boolean_t oneway = FALSE;
73 static boolean_t do_select = FALSE;
74 static boolean_t save_perfdata = FALSE;
75
76 int msg_type;
77 int num_ints;
78 int num_msgs;
79 int num_clients;
80 int num_servers;
81 int client_delay;
82 int client_spin;
83 int client_pages;
84 char **server_port_name;
85
86 void
signal_handler(int sig)87 signal_handler(int sig)
88 {
89 }
90
91 void
usage(const char * progname)92 usage(const char *progname)
93 {
94 fprintf(stderr, "usage: %s [options]\n", progname);
95 fprintf(stderr, "where options are:\n");
96 fprintf(stderr, " -affinity\t\tthreads use affinity\n");
97 fprintf(stderr, " -timeshare\t\tthreads use timeshare\n");
98 fprintf(stderr, " -threaded\t\tuse (p)threads\n");
99 fprintf(stderr, " -verbose\t\tbe verbose\n");
100 fprintf(stderr, " -oneway\t\tdo not request return reply\n");
101 fprintf(stderr, " -count num\t\tnumber of messages to send\n");
102 fprintf(stderr, " -type trivial|inline|complex\ttype of messages to send\n");
103 fprintf(stderr, " -numints num\tnumber of 32-bit ints to send in messages\n");
104 fprintf(stderr, " -servers num\tnumber of servers threads to run\n");
105 fprintf(stderr, " -clients num\tnumber of clients per server\n");
106 fprintf(stderr, " -delay num\t\tmicroseconds to sleep clients between messages\n");
107 fprintf(stderr, " -work num\t\tmicroseconds of client work\n");
108 fprintf(stderr, " -pages num\t\tpages of memory touched by client work\n");
109 fprintf(stderr, " -select \t\tselect prior to calling kevent().\n");
110 fprintf(stderr, " -perf \t\tCreate perfdata files for metrics.\n");
111 fprintf(stderr, "default values are:\n");
112 fprintf(stderr, " . no affinity\n");
113 fprintf(stderr, " . not timeshare\n");
114 fprintf(stderr, " . not verbose\n");
115 fprintf(stderr, " . not oneway\n");
116 fprintf(stderr, " . client sends 100000 messages\n");
117 fprintf(stderr, " . inline message type\n");
118 fprintf(stderr, " . 64 32-bit integers in inline/complex messages\n");
119 fprintf(stderr, " . (num_available_processors+1)%%2 servers\n");
120 fprintf(stderr, " . 4 clients per server\n");
121 fprintf(stderr, " . no delay\n");
122 exit(1);
123 }
124
125 void
parse_args(int argc,char * argv[])126 parse_args(int argc, char *argv[])
127 {
128 host_basic_info_data_t info;
129 mach_msg_type_number_t count;
130 kern_return_t result;
131
132 /* Initialize defaults */
133 msg_type = msg_type_trivial;
134 num_ints = 64;
135 num_msgs = 100000;
136 client_delay = 0;
137 num_clients = 4;
138
139 count = HOST_BASIC_INFO_COUNT;
140 result = host_info(mach_host_self(), HOST_BASIC_INFO,
141 (host_info_t)&info, &count);
142 if (result == KERN_SUCCESS && info.avail_cpus > 1) {
143 num_servers = info.avail_cpus / 2;
144 } else {
145 num_servers = 1;
146 }
147
148 const char *progname = argv[0];
149 argc--; argv++;
150 while (0 < argc) {
151 if (0 == strcmp("-verbose", argv[0])) {
152 verbose = TRUE;
153 argc--; argv++;
154 } else if (0 == strcmp("-affinity", argv[0])) {
155 affinity = TRUE;
156 argc--; argv++;
157 } else if (0 == strcmp("-timeshare", argv[0])) {
158 timeshare = TRUE;
159 argc--; argv++;
160 } else if (0 == strcmp("-threaded", argv[0])) {
161 threaded = TRUE;
162 argc--; argv++;
163 } else if (0 == strcmp("-oneway", argv[0])) {
164 oneway = TRUE;
165 argc--; argv++;
166 } else if (0 == strcmp("-type", argv[0])) {
167 if (argc < 2) {
168 usage(progname);
169 }
170 if (0 == strcmp("trivial", argv[1])) {
171 msg_type = msg_type_trivial;
172 } else if (0 == strcmp("inline", argv[1])) {
173 msg_type = msg_type_inline;
174 } else if (0 == strcmp("complex", argv[1])) {
175 msg_type = msg_type_complex;
176 } else {
177 usage(progname);
178 }
179 argc -= 2; argv += 2;
180 } else if (0 == strcmp("-numints", argv[0])) {
181 if (argc < 2) {
182 usage(progname);
183 }
184 num_ints = strtoul(argv[1], NULL, 0);
185 argc -= 2; argv += 2;
186 } else if (0 == strcmp("-count", argv[0])) {
187 if (argc < 2) {
188 usage(progname);
189 }
190 num_msgs = strtoul(argv[1], NULL, 0);
191 argc -= 2; argv += 2;
192 } else if (0 == strcmp("-clients", argv[0])) {
193 if (argc < 2) {
194 usage(progname);
195 }
196 num_clients = strtoul(argv[1], NULL, 0);
197 argc -= 2; argv += 2;
198 } else if (0 == strcmp("-servers", argv[0])) {
199 if (argc < 2) {
200 usage(progname);
201 }
202 num_servers = strtoul(argv[1], NULL, 0);
203 argc -= 2; argv += 2;
204 } else if (0 == strcmp("-delay", argv[0])) {
205 if (argc < 2) {
206 usage(progname);
207 }
208 client_delay = strtoul(argv[1], NULL, 0);
209 argc -= 2; argv += 2;
210 } else if (0 == strcmp("-spin", argv[0])) {
211 if (argc < 2) {
212 usage(progname);
213 }
214 client_spin = strtoul(argv[1], NULL, 0);
215 argc -= 2; argv += 2;
216 } else if (0 == strcmp("-pages", argv[0])) {
217 if (argc < 2) {
218 usage(progname);
219 }
220 client_pages = strtoul(argv[1], NULL, 0);
221 argc -= 2; argv += 2;
222 } else if (0 == strcmp("-select", argv[0])) {
223 do_select = TRUE;
224 argc--; argv++;
225 } else if (0 == strcmp("-perf", argv[0])) {
226 save_perfdata = TRUE;
227 argc--; argv++;
228 } else {
229 usage(progname);
230 }
231 }
232 }
233
234 void
setup_server_ports(struct port_args * ports)235 setup_server_ports(struct port_args *ports)
236 {
237 kern_return_t ret = 0;
238 mach_port_t bsport;
239
240 ports->req_size = MAX(sizeof(ipc_inline_message) +
241 sizeof(u_int32_t) * num_ints,
242 sizeof(ipc_complex_message));
243 ports->reply_size = sizeof(ipc_trivial_message) -
244 sizeof(mach_msg_trailer_t);
245 ports->req_msg = malloc(ports->req_size);
246 ports->reply_msg = malloc(ports->reply_size);
247
248 ret = mach_port_allocate(mach_task_self(),
249 MACH_PORT_RIGHT_RECEIVE,
250 &(ports->port));
251 if (KERN_SUCCESS != ret) {
252 mach_error("mach_port_allocate(): ", ret);
253 exit(1);
254 }
255
256 ret = mach_port_allocate(mach_task_self(),
257 MACH_PORT_RIGHT_PORT_SET,
258 &(ports->pset));
259 if (KERN_SUCCESS != ret) {
260 mach_error("mach_port_allocate(): ", ret);
261 exit(1);
262 }
263
264 ret = mach_port_insert_member(mach_task_self(),
265 ports->port,
266 ports->pset);
267 if (KERN_SUCCESS != ret) {
268 mach_error("mach_port_insert_member(): ", ret);
269 exit(1);
270 }
271
272 ret = mach_port_insert_right(mach_task_self(),
273 ports->port,
274 ports->port,
275 MACH_MSG_TYPE_MAKE_SEND);
276 if (KERN_SUCCESS != ret) {
277 mach_error("mach_port_insert_right(): ", ret);
278 exit(1);
279 }
280
281 ret = task_get_bootstrap_port(mach_task_self(), &bsport);
282 if (KERN_SUCCESS != ret) {
283 mach_error("task_get_bootstrap_port(): ", ret);
284 exit(1);
285 }
286
287 if (verbose) {
288 printf("server waiting for IPC messages from client on port '%s'.\n",
289 server_port_name[ports->server_num]);
290 }
291 ret = bootstrap_register(bsport,
292 server_port_name[ports->server_num],
293 ports->port);
294 if (KERN_SUCCESS != ret) {
295 mach_error("bootstrap_register(): ", ret);
296 exit(1);
297 }
298 }
299
300 void
setup_client_ports(struct port_args * ports)301 setup_client_ports(struct port_args *ports)
302 {
303 kern_return_t ret = 0;
304 switch (msg_type) {
305 case msg_type_trivial:
306 ports->req_size = sizeof(ipc_trivial_message);
307 break;
308 case msg_type_inline:
309 ports->req_size = sizeof(ipc_inline_message) +
310 sizeof(u_int32_t) * num_ints;
311 break;
312 case msg_type_complex:
313 ports->req_size = sizeof(ipc_complex_message);
314 break;
315 }
316 ports->req_size -= sizeof(mach_msg_trailer_t);
317 ports->reply_size = sizeof(ipc_trivial_message);
318 ports->req_msg = malloc(ports->req_size);
319 ports->reply_msg = malloc(ports->reply_size);
320
321 ret = mach_port_allocate(mach_task_self(),
322 MACH_PORT_RIGHT_RECEIVE,
323 &(ports->port));
324 if (KERN_SUCCESS != ret) {
325 mach_error("mach_port_allocate(): ", ret);
326 exit(1);
327 }
328 if (verbose) {
329 printf("Client sending %d %s IPC messages to port '%s' in %s mode.\n",
330 num_msgs, (msg_type == msg_type_inline) ?
331 "inline" : ((msg_type == msg_type_complex) ?
332 "complex" : "trivial"),
333 server_port_name[ports->server_num],
334 (oneway ? "oneway" : "rpc"));
335 }
336 }
337
338
339 static void
thread_setup(int tag)340 thread_setup(int tag)
341 {
342 kern_return_t ret;
343 thread_extended_policy_data_t epolicy;
344 thread_affinity_policy_data_t policy;
345
346 if (!timeshare) {
347 epolicy.timeshare = FALSE;
348 ret = thread_policy_set(
349 mach_thread_self(), THREAD_EXTENDED_POLICY,
350 (thread_policy_t) &epolicy,
351 THREAD_EXTENDED_POLICY_COUNT);
352 if (ret != KERN_SUCCESS) {
353 printf("thread_policy_set(THREAD_EXTENDED_POLICY) returned %d\n", ret);
354 }
355 }
356
357 if (affinity) {
358 policy.affinity_tag = tag;
359 ret = thread_policy_set(
360 mach_thread_self(), THREAD_AFFINITY_POLICY,
361 (thread_policy_t) &policy,
362 THREAD_AFFINITY_POLICY_COUNT);
363 if (ret != KERN_SUCCESS) {
364 printf("thread_policy_set(THREAD_AFFINITY_POLICY) returned %d\n", ret);
365 }
366 }
367 }
368
369 void *
server(void * serverarg)370 server(void *serverarg)
371 {
372 int kq;
373 struct kevent64_s kev[1];
374 int err;
375 int count;
376 struct port_args args;
377 int idx;
378 kern_return_t ret;
379 int totalmsg = num_msgs * num_clients;
380 fd_set readfds;
381
382 args.server_num = (int) (long) serverarg;
383 setup_server_ports(&args);
384
385 thread_setup(args.server_num + 1);
386
387 kq = kqueue();
388 if (kq == -1) {
389 perror("kqueue");
390 exit(1);
391 }
392 EV_SET64(&kev[0], args.pset, EVFILT_MACHPORT, (EV_ADD | EV_CLEAR | EV_DISPATCH),
393 #if DIRECT_MSG_RCV
394 MACH_RCV_MSG | MACH_RCV_LARGE, 0, 0, (mach_vm_address_t)args.req_msg, args.req_size);
395 #else
396 0, 0, 0, 0, 0);
397 #endif
398 err = kevent64(kq, kev, 1, NULL, 0, 0, NULL);
399 if (err == -1) {
400 perror("kevent");
401 exit(1);
402 }
403
404 for (idx = 0; idx < totalmsg; idx++) {
405 if (verbose) {
406 printf("server awaiting message %d\n", idx);
407 }
408 retry:
409 if (do_select) {
410 FD_ZERO(&readfds);
411 FD_SET(kq, &readfds);
412
413 if (verbose) {
414 printf("Calling select() prior to kevent64().\n");
415 }
416
417 count = select(kq + 1, &readfds, NULL, NULL, NULL);
418 if (count == -1) {
419 perror("select");
420 exit(1);
421 }
422 }
423
424 EV_SET64(&kev[0], args.pset, EVFILT_MACHPORT, EV_ENABLE,
425 #if DIRECT_MSG_RCV
426 MACH_RCV_MSG | MACH_RCV_LARGE, 0, 0, (mach_vm_address_t)args.req_msg, args.req_size);
427 #else
428 0, 0, 0, 0, 0);
429 #endif
430 err = kevent64(kq, kev, 1, kev, 1, 0, NULL);
431 if (err == -1) {
432 perror("kevent64");
433 exit(1);
434 }
435 if (err == 0) {
436 // printf("kevent64: returned zero\n");
437 goto retry;
438 }
439
440 #if DIRECT_MSG_RCV
441 ret = kev[0].fflags;
442 if (MACH_MSG_SUCCESS != ret) {
443 if (verbose) {
444 printf("kevent64() mach_msg_return=%d", ret);
445 }
446 mach_error("kevent64 (msg receive): ", ret);
447 exit(1);
448 }
449 #else
450 if (kev[0].data != args.port) {
451 printf("kevent64(MACH_PORT_NULL) port name (%lld) != expected (0x%x)\n", kev[0].data, args.port);
452 }
453
454 args.req_msg->msgh_bits = 0;
455 args.req_msg->msgh_size = args.req_size;
456 args.req_msg->msgh_local_port = args.port;
457 ret = mach_msg(args.req_msg,
458 MACH_RCV_MSG | MACH_RCV_INTERRUPT | MACH_RCV_LARGE,
459 0,
460 args.req_size,
461 args.pset,
462 MACH_MSG_TIMEOUT_NONE,
463 MACH_PORT_NULL);
464 if (MACH_RCV_INTERRUPTED == ret) {
465 break;
466 }
467 if (MACH_MSG_SUCCESS != ret) {
468 if (verbose) {
469 printf("mach_msg() ret=%d", ret);
470 }
471 mach_error("mach_msg (receive): ", ret);
472 exit(1);
473 }
474 #endif
475 if (verbose) {
476 printf("server received message %d\n", idx);
477 }
478 if (args.req_msg->msgh_bits & MACH_MSGH_BITS_COMPLEX) {
479 ret = vm_deallocate(mach_task_self(),
480 (vm_address_t)((ipc_complex_message *)args.req_msg)->descriptor.address,
481 ((ipc_complex_message *)args.req_msg)->descriptor.size);
482 }
483
484 if (1 == args.req_msg->msgh_id) {
485 if (verbose) {
486 printf("server sending reply %d\n", idx);
487 }
488 args.reply_msg->msgh_bits = MACH_MSGH_BITS(MACH_MSG_TYPE_COPY_SEND,
489 MACH_MSG_TYPE_MAKE_SEND);
490 args.reply_msg->msgh_size = args.reply_size;
491 args.reply_msg->msgh_remote_port = args.req_msg->msgh_remote_port;
492 args.reply_msg->msgh_local_port = args.req_msg->msgh_local_port;
493 args.reply_msg->msgh_id = 2;
494 ret = mach_msg(args.reply_msg,
495 MACH_SEND_MSG,
496 args.reply_size,
497 0,
498 MACH_PORT_NULL,
499 MACH_MSG_TIMEOUT_NONE,
500 MACH_PORT_NULL);
501 if (MACH_MSG_SUCCESS != ret) {
502 mach_error("mach_msg (send): ", ret);
503 exit(1);
504 }
505 }
506 }
507 return NULL;
508 }
509
510 static inline void
client_spin_loop(unsigned count,void (fn)(void))511 client_spin_loop(unsigned count, void(fn)(void))
512 {
513 while (count--) {
514 fn();
515 }
516 }
517
518 static long dummy_memory;
519 static long *client_memory = &dummy_memory;
520 static void
client_work_atom(void)521 client_work_atom(void)
522 {
523 static int i;
524
525 if (++i > client_pages * PAGE_SIZE / sizeof(long)) {
526 i = 0;
527 }
528 client_memory[i] = 0;
529 }
530
531 static int calibration_count = 10000;
532 static int calibration_usec;
533 static void *
calibrate_client_work(void)534 calibrate_client_work(void)
535 {
536 long dummy;
537 struct timeval nowtv;
538 struct timeval warmuptv = { 0, 100 * 1000 }; /* 100ms */
539 struct timeval starttv;
540 struct timeval endtv;
541
542 if (client_spin) {
543 /* Warm-up the stepper first... */
544 gettimeofday(&nowtv, NULL);
545 timeradd(&nowtv, &warmuptv, &endtv);
546 do {
547 client_spin_loop(calibration_count, client_work_atom);
548 gettimeofday(&nowtv, NULL);
549 } while (timercmp(&nowtv, &endtv, < ));
550
551 /* Now do the calibration */
552 while (TRUE) {
553 gettimeofday(&starttv, NULL);
554 client_spin_loop(calibration_count, client_work_atom);
555 gettimeofday(&endtv, NULL);
556 if (endtv.tv_sec - starttv.tv_sec > 1) {
557 calibration_count /= 10;
558 continue;
559 }
560 calibration_usec = endtv.tv_usec - starttv.tv_usec;
561 if (endtv.tv_usec < starttv.tv_usec) {
562 calibration_usec += 1000000;
563 }
564 if (calibration_usec < 1000) {
565 calibration_count *= 10;
566 continue;
567 }
568 calibration_count /= calibration_usec;
569 break;
570 }
571 if (verbose) {
572 printf("calibration_count=%d calibration_usec=%d\n",
573 calibration_count, calibration_usec);
574 }
575 }
576 return NULL;
577 }
578
579 static void *
client_work(void)580 client_work(void)
581 {
582 if (client_spin) {
583 client_spin_loop(calibration_count * client_spin,
584 client_work_atom);
585 }
586
587 if (client_delay) {
588 usleep(client_delay);
589 }
590 return NULL;
591 }
592
593 void *
client(void * threadarg)594 client(void *threadarg)
595 {
596 struct port_args args;
597 int idx;
598 mach_msg_header_t *req, *reply;
599 mach_port_t bsport, servport;
600 kern_return_t ret;
601 int server_num = (int) threadarg;
602 void *ints = malloc(sizeof(u_int32_t) * num_ints);
603
604 if (verbose) {
605 printf("client(%d) started, server port name %s\n",
606 server_num, server_port_name[server_num]);
607 }
608
609 args.server_num = server_num;
610 thread_setup(server_num + 1);
611
612 /* find server port */
613 ret = task_get_bootstrap_port(mach_task_self(), &bsport);
614 if (KERN_SUCCESS != ret) {
615 mach_error("task_get_bootstrap_port(): ", ret);
616 exit(1);
617 }
618 ret = bootstrap_look_up(bsport,
619 server_port_name[server_num],
620 &servport);
621 if (KERN_SUCCESS != ret) {
622 mach_error("bootstrap_look_up(): ", ret);
623 exit(1);
624 }
625
626 setup_client_ports(&args);
627
628 /* Allocate and touch memory */
629 if (client_pages) {
630 unsigned i;
631 client_memory = (long *) malloc(client_pages * PAGE_SIZE);
632 for (i = 0; i < client_pages; i++) {
633 client_memory[i * PAGE_SIZE / sizeof(long)] = 0;
634 }
635 }
636
637 /* start message loop */
638 for (idx = 0; idx < num_msgs; idx++) {
639 req = args.req_msg;
640 reply = args.reply_msg;
641
642 req->msgh_bits = MACH_MSGH_BITS(MACH_MSG_TYPE_COPY_SEND,
643 MACH_MSG_TYPE_MAKE_SEND);
644 req->msgh_size = args.req_size;
645 req->msgh_remote_port = servport;
646 req->msgh_local_port = args.port;
647 req->msgh_id = oneway ? 0 : 1;
648 if (msg_type == msg_type_complex) {
649 (req)->msgh_bits |= MACH_MSGH_BITS_COMPLEX;
650 ((ipc_complex_message *)req)->body.msgh_descriptor_count = 1;
651 ((ipc_complex_message *)req)->descriptor.address = ints;
652 ((ipc_complex_message *)req)->descriptor.size =
653 num_ints * sizeof(u_int32_t);
654 ((ipc_complex_message *)req)->descriptor.deallocate = FALSE;
655 ((ipc_complex_message *)req)->descriptor.copy = MACH_MSG_VIRTUAL_COPY;
656 ((ipc_complex_message *)req)->descriptor.type = MACH_MSG_OOL_DESCRIPTOR;
657 }
658 if (verbose) {
659 printf("client sending message %d\n", idx);
660 }
661 ret = mach_msg(req,
662 MACH_SEND_MSG,
663 args.req_size,
664 0,
665 MACH_PORT_NULL,
666 MACH_MSG_TIMEOUT_NONE,
667 MACH_PORT_NULL);
668 if (MACH_MSG_SUCCESS != ret) {
669 mach_error("mach_msg (send): ", ret);
670 fprintf(stderr, "bailing after %u iterations\n", idx);
671 exit(1);
672 break;
673 }
674 if (!oneway) {
675 if (verbose) {
676 printf("client awaiting reply %d\n", idx);
677 }
678 reply->msgh_bits = 0;
679 reply->msgh_size = args.reply_size;
680 reply->msgh_local_port = args.port;
681 ret = mach_msg(args.reply_msg,
682 MACH_RCV_MSG | MACH_RCV_INTERRUPT,
683 0,
684 args.reply_size,
685 args.port,
686 MACH_MSG_TIMEOUT_NONE,
687 MACH_PORT_NULL);
688 if (MACH_MSG_SUCCESS != ret) {
689 mach_error("mach_msg (receive): ", ret);
690 fprintf(stderr, "bailing after %u iterations\n",
691 idx);
692 exit(1);
693 }
694 if (verbose) {
695 printf("client received reply %d\n", idx);
696 }
697 }
698
699 client_work();
700 }
701
702 free(ints);
703 return NULL;
704 }
705
706 static void
thread_spawn(thread_id_t * thread,void * (fn)(void *),void * arg)707 thread_spawn(thread_id_t *thread, void *(fn)(void *), void *arg)
708 {
709 if (threaded) {
710 kern_return_t ret;
711 ret = pthread_create(
712 &thread->tid,
713 NULL,
714 fn,
715 arg);
716 if (ret != 0) {
717 err(1, "pthread_create()");
718 }
719 if (verbose) {
720 printf("created pthread %p\n", thread->tid);
721 }
722 } else {
723 thread->pid = fork();
724 if (thread->pid == 0) {
725 if (verbose) {
726 printf("calling %p(%p)\n", fn, arg);
727 }
728 fn(arg);
729 exit(0);
730 }
731 if (verbose) {
732 printf("forked pid %d\n", thread->pid);
733 }
734 }
735 }
736
737 static void
thread_join(thread_id_t * thread)738 thread_join(thread_id_t *thread)
739 {
740 if (threaded) {
741 kern_return_t ret;
742 if (verbose) {
743 printf("joining thread %p\n", thread->tid);
744 }
745 ret = pthread_join(thread->tid, NULL);
746 if (ret != KERN_SUCCESS) {
747 err(1, "pthread_join(%p)", thread->tid);
748 }
749 } else {
750 int stat;
751 if (verbose) {
752 printf("waiting for pid %d\n", thread->pid);
753 }
754 waitpid(thread->pid, &stat, 0);
755 }
756 }
757
758 static void
wait_for_servers(void)759 wait_for_servers(void)
760 {
761 int i;
762 int retry_count = 10;
763 mach_port_t bsport, servport;
764 kern_return_t ret;
765
766 /* find server port */
767 ret = task_get_bootstrap_port(mach_task_self(), &bsport);
768 if (KERN_SUCCESS != ret) {
769 mach_error("task_get_bootstrap_port(): ", ret);
770 exit(1);
771 }
772
773 while (retry_count-- > 0) {
774 for (i = 0; i < num_servers; i++) {
775 ret = bootstrap_look_up(bsport,
776 server_port_name[i],
777 &servport);
778 if (ret != KERN_SUCCESS) {
779 break;
780 }
781 }
782 if (ret == KERN_SUCCESS) {
783 return;
784 }
785 usleep(100 * 1000); /* 100ms */
786 }
787 fprintf(stderr, "Server(s) failed to register\n");
788 exit(1);
789 }
790
791
792 int
main(int argc,char * argv[])793 main(int argc, char *argv[])
794 {
795 int i;
796 int j;
797 thread_id_t *client_id;
798 thread_id_t *server_id;
799
800 signal(SIGINT, signal_handler);
801 parse_args(argc, argv);
802
803 calibrate_client_work();
804
805 /*
806 * If we're using affinity create an empty namespace now
807 * so this is shared by all our offspring.
808 */
809 if (affinity) {
810 thread_setup(0);
811 }
812
813 server_id = (thread_id_t *) malloc(num_servers * sizeof(thread_id_t));
814 server_port_name = (char **) malloc(num_servers * sizeof(char *));
815 if (verbose) {
816 printf("creating %d servers\n", num_servers);
817 }
818 for (i = 0; i < num_servers; i++) {
819 server_port_name[i] = (char *) malloc(sizeof("PORT.pppppp.xx"));
820 /* PORT names include pid of main process for disambiguation */
821 sprintf(server_port_name[i], "PORT.%06d.%02d", getpid(), i);
822 thread_spawn(&server_id[i], server, (void *) (long) i);
823 }
824
825 int totalclients = num_servers * num_clients;
826 int totalmsg = num_msgs * totalclients;
827 struct timeval starttv, endtv, deltatv;
828
829 /*
830 * Wait for all servers to have registered all ports before starting
831 * the clients and the clock.
832 */
833 wait_for_servers();
834
835 printf("%d server%s, %d client%s per server (%d total) %u messages...",
836 num_servers, (num_servers > 1)? "s" : "",
837 num_clients, (num_clients > 1)? "s" : "",
838 totalclients,
839 totalmsg);
840 fflush(stdout);
841
842 /* Call gettimeofday() once and throw away result; some implementations
843 * (like Mach's) cache some time zone info on first call.
844 */
845 gettimeofday(&starttv, NULL);
846 gettimeofday(&starttv, NULL);
847
848 client_id = (thread_id_t *) malloc(totalclients * sizeof(thread_id_t));
849 if (verbose) {
850 printf("creating %d clients\n", totalclients);
851 }
852 for (i = 0; i < num_servers; i++) {
853 for (j = 0; j < num_clients; j++) {
854 thread_spawn(
855 &client_id[(i * num_clients) + j],
856 client,
857 (void *) (long) i);
858 }
859 }
860
861 /* Wait for servers to complete */
862 for (i = 0; i < num_servers; i++) {
863 thread_join(&server_id[i]);
864 }
865
866 gettimeofday(&endtv, NULL);
867
868 for (i = 0; i < totalclients; i++) {
869 thread_join(&client_id[i]);
870 }
871
872 /* report results */
873 deltatv.tv_sec = endtv.tv_sec - starttv.tv_sec;
874 deltatv.tv_usec = endtv.tv_usec - starttv.tv_usec;
875 if (endtv.tv_usec < starttv.tv_usec) {
876 deltatv.tv_sec--;
877 deltatv.tv_usec += 1000000;
878 }
879
880 double dsecs = (double) deltatv.tv_sec +
881 1.0E-6 * (double) deltatv.tv_usec;
882
883 double time_in_sec = (double)deltatv.tv_sec + (double)deltatv.tv_usec / 1000.0;
884 double throughput_msg_p_sec = (double) totalmsg / dsecs;
885 double avg_msg_latency = dsecs * 1.0E6 / (double)totalmsg;
886
887 printf(" in %ld.%03u seconds\n",
888 (long)deltatv.tv_sec, deltatv.tv_usec / 1000);
889 printf(" throughput in messages/sec: %g\n",
890 (double)totalmsg / dsecs);
891 printf(" average message latency (usec): %2.3g\n",
892 dsecs * 1.0E6 / (double) totalmsg);
893
894 if (save_perfdata == TRUE) {
895 char name[256];
896 snprintf(name, sizeof(name), "%s_avg_msg_latency", basename(argv[0]));
897 record_perf_data(name, "usec", avg_msg_latency, "Message latency measured in microseconds. Lower is better", stderr);
898 }
899 return 0;
900 }
901