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