thread-overhead

1 files changed, 345 insertions, 0 deletions

diff --git a/thread-overhead.c b/thread-overhead.c
new file mode 100644
index 0000000..84cadaa
--- /dev/null
+++ b/thread-overhead.c
@@ -0,0 +1,345 @@
+#define _GNU_SOURCE /* for sched_setaffinity(2) */
+#include <stdlib.h>
+#include <stdio.h>
+#include <unistd.h>
+#include <getopt.h>
+#include <limits.h>
+#include <string.h>
+#include <stdint.h>
+#include <inttypes.h>
+#include <pthread.h>
+#include <errno.h>
+#include <sched.h>
+#include <libaio.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+
+#define min(a, b) (a < b ? a : b)
+
+struct shared {
+	pthread_mutex_t mutex;
+	pthread_cond_t cond;
+	unsigned long nr_to_submit;
+	io_context_t aio_ctx;
+	unsigned long aio_nr;
+};
+
+struct per_thread {
+	pthread_t pthread;
+	struct shared *shared;
+
+	int cpu;
+	unsigned int submit:1,
+	 	     live:1;
+	unsigned long count;
+	int fd;
+};
+
+static void submit_loop(struct per_thread *thd, struct shared *shared)
+{
+	struct iocb *iocbs;
+	struct iocb **ios;
+	char junk;
+	long nr = 0;
+	long i;
+	int ret;
+
+	iocbs = calloc(shared->aio_nr, sizeof(struct iocb));
+	ios = calloc(shared->aio_nr, sizeof(struct iocb *));
+
+	while (shared->aio_ctx) {
+		if (nr == 0) {
+			pthread_mutex_lock(&shared->mutex);
+			while (shared->aio_ctx && shared->nr_to_submit == 0) {
+				pthread_cond_wait(&shared->cond,
+						  &shared->mutex);
+			}
+			nr = min(shared->aio_nr, shared->nr_to_submit);
+			shared->nr_to_submit -= nr;
+			pthread_mutex_unlock(&shared->mutex);
+			if (!shared->aio_ctx)
+				break;
+		}
+
+		for (i = 0; i < nr; i++) {
+			io_prep_pread(&iocbs[i], thd->fd, &junk, 1, 0);
+			ios[i] = &iocbs[i];
+		}
+
+		ret = io_submit(shared->aio_ctx, nr, ios);
+		if (!shared->aio_ctx)
+			break;
+		if (ret <= 0) {
+			printf("nr %lu ret %d ctx %p\n", nr, ret, shared->aio_ctx);
+			perror("io_submit failed");
+			exit(1);
+		}
+
+		nr -= ret;
+	}
+
+	free(iocbs);
+	free(ios);
+}
+
+static void get_loop(struct per_thread *thd, struct shared *shared)
+{
+	struct io_event *events;
+	int ret;
+
+	events = calloc(shared->aio_nr, sizeof(struct io_event));
+
+	while (shared->aio_ctx) {
+		/* just enough to not hang when stopping but never expire */
+		ret = io_getevents(shared->aio_ctx, 1, shared->aio_nr, events,
+				   NULL);
+		if (!shared->aio_ctx)
+			break;
+		if (ret <= 0) {
+			perror("io_getevents failed");
+			exit(1);
+		}
+
+		thd->count += ret;
+
+		pthread_mutex_lock(&shared->mutex);
+		shared->nr_to_submit += ret;
+		pthread_cond_signal(&shared->cond);
+		pthread_mutex_unlock(&shared->mutex);
+	}
+
+	free(events);
+}
+
+static void *thread_func(void *arg)
+{
+	struct per_thread *thd = arg;
+	struct shared *shared = thd->shared;
+
+	printf("thread %p starting: submit %u cpu %d\n",
+	       thd, thd->submit, thd->cpu);
+
+	thd->fd = open("/etc/motd", O_RDONLY);
+	if (thd->fd < 0) {
+		perror("open /dev/null");
+		exit(1);
+	}
+
+	/* would need dynamic cpu sets > 1024 cpus */
+	if (thd->cpu >= 0) {
+		cpu_set_t want;
+		cpu_set_t set;
+
+		CPU_ZERO(&want);
+		CPU_SET(thd->cpu, &want);
+		if (pthread_setaffinity_np(thd->pthread, sizeof(want), &want) ||
+		    pthread_getaffinity_np(thd->pthread, sizeof(set), &set)) {
+			perror("pthread affinity failed");
+			exit(1);
+		}
+
+		if (!CPU_EQUAL(&want, &set)) {
+			printf("couldn't set cpu set\n");
+			exit(1);
+		}
+	}
+
+	if (thd->submit) {
+		submit_loop(thd, shared);
+	} else
+		get_loop(thd, shared);
+
+	printf("thread %p exiting\n", thd);
+	close(thd->fd);
+
+	return NULL;
+}
+
+static int get_cpus(unsigned int *cpus, unsigned int max)
+{
+	unsigned int i;
+	unsigned int nr;
+	cpu_set_t cpuset;
+
+	if (sched_getaffinity(getpid(), sizeof(cpuset), &cpuset))
+		return 0;
+
+	for (i = 0, nr = 0; i < CPU_SETSIZE && nr < max; i++) {
+		if (CPU_ISSET(i, &cpuset))
+			cpus[nr++] = i;
+	}
+
+	return nr;
+}
+
+static char *names[] = {"0", "1", "nr", "many"};
+static char garbage[4096];
+
+int main(int argc, char **argv)
+{
+	unsigned int cpus[CPU_SETSIZE];
+	double per_sec[4][4] = {{0,},};
+	struct shared shared;
+	struct per_thread *threads;
+	struct per_thread *thd;
+	io_context_t ctx;
+	int submit_mode;
+	int get_mode;
+	int cpu_ind;
+	int mode;
+	unsigned int i;
+	uint64_t total;
+	int is_submit;
+	int nr_cpus;
+	int ret;
+	int widest = 0;
+	int nr;
+
+	/* figure out which cpus are possible */
+	nr_cpus = get_cpus(cpus, CPU_SETSIZE);
+	if (nr_cpus == 0) {
+		printf("couldn't find nr cpus\n");
+		exit(1);
+	}
+	if (nr_cpus == 1) {
+		printf("need more than one cpu\n");
+		exit(1);
+	}
+
+	shared.aio_nr = 256;
+
+	printf("will run on %d cpus: ", nr_cpus);
+	for (i = 0; i < nr_cpus; i++)
+		printf("%u%c", cpus[i], i == nr_cpus - 1 ? '\n' : ',');
+
+	threads = calloc((nr_cpus * 4 * 2) + 1, sizeof(struct per_thread));
+	if (!threads) {
+		printf("error: couldn't allocate threads\n");
+		exit(1);
+	}
+
+	for (submit_mode = 0; submit_mode < 4; submit_mode++) {
+		for (get_mode = 0; get_mode < 4; get_mode++) {
+
+			/* get a new context for each mode */
+			if (io_setup(shared.aio_nr, &shared.aio_ctx)) {
+				perror("io_setup");
+				exit(1);
+			}
+			shared.nr_to_submit = 0;
+
+			pthread_mutex_init(&shared.mutex, NULL);
+			pthread_cond_init(&shared.cond, NULL);
+
+			thd = threads;
+
+			for (is_submit = 0; is_submit < 2; is_submit++) {
+
+				if (is_submit)
+					mode = submit_mode;
+				else
+					mode = get_mode;
+
+				if (mode == 0 || mode == 1) {
+					cpu_ind = mode;
+					nr = 1;
+				} else if (mode == 2) {
+					cpu_ind = 0;
+					nr = nr_cpus;
+				} else {
+					cpu_ind = -1;
+					nr = nr_cpus * 4;
+				}
+
+				/* fire off threads */
+				for (i = 0; i < nr; i++, thd++) {
+					thd->shared = &shared;
+					thd->submit = is_submit;
+					thd->live = 1;
+					if (cpu_ind >= 0) {
+						thd->cpu = cpus[cpu_ind++];
+					} else {
+						thd->cpu = -1;
+					}
+
+					printf("starting thread %p\n", thd);
+
+					ret = pthread_create(&thd->pthread,
+							     NULL,
+							     thread_func, thd);
+					if (ret) {
+						perror("pthread_create");
+						exit(1);
+					}
+				}
+			}
+
+			/* tell everyone to get going */
+			pthread_mutex_lock(&shared.mutex);
+			shared.nr_to_submit = shared.aio_nr;
+			pthread_cond_signal(&shared.cond);
+			pthread_mutex_unlock(&shared.mutex);
+
+			printf("2s of burn in, then running for 3s\n");
+			sleep(2);
+			for (thd = threads; thd->live; thd++)
+				thd->count = 0;
+			total = 0;
+
+			/* run for a bit and grab numbers */
+			for (i = 0; i < 3; i++) {
+				sleep(1);
+				for (thd = threads; thd->live; thd++) {
+					total += thd->count;
+					thd->count = 0;
+				}
+				printf("%f\n", (double)total / (i + 1));
+			}
+
+			per_sec[submit_mode][get_mode] = (double)total / i;
+
+			/* tell threads to stop */
+			printf("killing..\n");
+			pthread_mutex_lock(&shared.mutex);
+			ctx = shared.aio_ctx;
+			shared.aio_ctx = NULL;
+			if (io_destroy(ctx)) {
+				perror("io_destroy");
+				exit(1);
+			}
+			pthread_cond_broadcast(&shared.cond);
+			pthread_mutex_unlock(&shared.mutex);
+
+			/* and drain 'em */
+			for (thd = threads; thd->live; thd++) {
+				pthread_join(thd->pthread, NULL);
+				thd->live = 0;
+			}
+		}
+	}
+
+	for (submit_mode = 0; submit_mode < 4; submit_mode++) {
+		for (get_mode = 0; get_mode < 4; get_mode++) {
+			ret = snprintf(garbage, sizeof(garbage), "%.2f",
+				      per_sec[submit_mode][get_mode]);
+			if (ret > widest)
+				widest = ret;
+		}
+	}
+
+	for (submit_mode = -1; submit_mode < 4; submit_mode++) {
+		printf("%*s ", widest,
+		       submit_mode < 0 ? "" : names[submit_mode]);
+		for (get_mode = 0; get_mode < 4; get_mode++) {
+			if (submit_mode == -1)
+				printf("%*s ", widest, names[get_mode]);
+			else
+				printf("%*.2f ", widest,
+				       per_sec[submit_mode][get_mode]);
+		}
+		printf("\n");
+	}
+
+	return 0;
+}