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