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