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