tools/tests/MPMMTest/KQMPMMtest.c

   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
  87 signal_handler(int sig)
  88 {
  89 }
  90
  91 void
  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
 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
 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
 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
 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 *
 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
 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
 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 *
 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 *
 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 *
 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
 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
 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
 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
 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 }