Attachment 947987 Details for Bug 850426 – Multithreaded userspace program to readahead directory inodes

Multithreaded userspace program to readahead directory inodes

userspace_ra.c (text/x-csrc), 11.30 KB, created by Abhijith Das on 2014-10-17 17:34:12 UTC

(hide)

Description:

Filename:

MIME Type:

Creator: Abhijith Das

Created: 2014-10-17 17:34:12 UTC

Size: 11.30 KB

patch

obsolete

>#include <stdio.h>
>#include <fcntl.h>
>#include <stdlib.h>
>#include <stdarg.h>
>#include <unistd.h>
>#include <dirent.h>
>#include <sys/types.h>
>#include <sys/syscall.h>
>#include <sys/stat.h>
>#include <signal.h>
>#include <errno.h>
>#include <time.h>
>#include <string.h>
>#include <pthread.h>
>#include <time.h>
>#include <assert.h>
>
>unsigned int bufsize;
>unsigned int maxthreads;
>unsigned int namelen;
>int debug, quiet, sort;
>
>const char *dirname;
>typedef unsigned int u32;
>
>void vp(const char *fmt, ...)
>{
>	va_list ap;
>	va_start(ap, fmt);
>	fprintf(stdout, "%lu - ", time(NULL));
>	vfprintf(stdout, fmt, ap);
>	va_end(ap);
>}
>
>void pdebug(const char *fmt, ...)
>{
>	if (debug)
>		vp(fmt);
>}
>
>void print(const char *fmt, ...)
>{
>	if (!quiet)
>		vp(fmt);
>}
>
>void set_defaults()
>{
>	bufsize = 4096;
>	namelen = 5;
>	debug = 0;
>	quiet = 0;
>	sort = 0;
>	maxthreads = (unsigned int) sysconf(_SC_NPROCESSORS_ONLN);
>}
>
>void print_usage(const char *prog)
>{
>	fprintf(stderr, " Usage: %s [OPTIONS] <dir>\n\n", prog);
>	fprintf(stderr, " -b SIZE  Buffer size. Size of buffer to use to retrieve entries.\n");
>	fprintf(stderr, "          Default 4096 bytes\n");
>	fprintf(stderr, " -D       Enable debug\n");
>	fprintf(stderr, " -l LEN   Approximate average filename length in the directory.\n");
>	fprintf(stderr, "          Defaults to 5\n");
>	fprintf(stderr, " -q       Quiet. Don't print anything\n");
>	fprintf(stderr, " -S       sort entries by inode numbers.\n");
>	fprintf(stderr, " -t NUM   Number of threads to use to process directory entries.\n");
>	fprintf(stderr, "          Defaults to the number of active CPUs, max:256\n");
>}
>
>void parse_opts(int argc, char **argv)
>{
>	int opt;
>	while ((opt = getopt(argc, argv, "b:Dl:qsSt:")) != -1) {
>		switch (opt) {
>		case 'b':
>			bufsize = atoi(optarg);
>			break;
>		case 'D':
>			debug = 1;
>			break;
>		case 'l':
>			namelen = atoi(optarg);
>			break;
>		case 'q':
>			quiet = 1;
>			break;
>		case 'S':
>			sort = 1;
>			break;
>		case 't':
>			maxthreads = atoi(optarg);
>			if (maxthreads > 256) {
>				fprintf(stderr, "Invalid maxthreads:%u value\n", maxthreads);
>				goto err;
>			}
>			break;
>		default:
>			fprintf(stderr, "Unrecognized option: %c\n", opt);
>			goto err;
>		}
>	}
>	if (optind >= argc) {
>		fprintf(stderr, "Expected <dir> argument\n");
>		goto err;
>	}
>
>	dirname = argv[optind];
>	return;
>
>err:
>	print_usage(argv[0]);
>	exit(EXIT_FAILURE);
>}
>
>#define THREAD_SLEEP_NS 100000000
>
>#define FL_TERMINATE 1
>#define FL_READY     2
>#define FL_NEWDATA   4
>#define FL_SORT      8
>
>struct thread {
>	pthread_t thread;
>	pthread_spinlock_t spin;
>	int nread;
>	unsigned int flags;
>	unsigned int count;
>	char *buffer;
>	struct linux_dirent **entries;
>};
>typedef struct thread thread_t;
>
>struct linux_dirent {
>	long           d_ino;
>	off_t          d_off;
>	unsigned short d_reclen;
>	char           d_name[];
>};
>
>static void u32_swap(void *a, void *b, int size)
>{
>	u32 t = *(u32 *)a;
>	*(u32 *)a = *(u32 *)b;
>	*(u32 *)b = t;
>}
>
>static void generic_swap(void *a, void *b, int size)
>{
>	char t;
>
>	do {
>		t = *(char *)a;
>		*(char *)a++ = *(char *)b;
>		*(char *)b++ = t;
>	} while (--size > 0);
>}
>
>static int ino_compare(const void *a, const void *b)
>{
>	struct linux_dirent *lda = *(struct linux_dirent **)a;
>	struct linux_dirent *ldb = *(struct linux_dirent **)b;
>
>	pdebug("%ld >> compare a:%ld b:%ld\n",
>	       syscall(SYS_gettid), lda->d_ino, ldb->d_ino);
>
>	return lda->d_ino - ldb->d_ino;
>}
>
>/**
> * sort - sort an array of elements
> * @base: pointer to data to sort
> * @num: number of elements
> * @size: size of each element
> * @cmp_func: pointer to comparison function
> * @swap_func: pointer to swap function or NULL
> *
> * This function does a heapsort on the given array. You may provide a
> * swap_func function optimized to your element type.
> *
> * Sorting time is O(n log n) both on average and worst-case. While
> * qsort is about 20% faster on average, it suffers from exploitable
> * O(n*n) worst-case behavior and extra memory requirements that make
> * it less suitable for kernel use.
> */
>
>void heapsort(void *base, size_t num, size_t size,
>	  int (*cmp_func)(const void *, const void *),
>	  void (*swap_func)(void *, void *, int size))
>{
>	/* pre-scale counters for performance */
>	int i = (num/2 - 1) * size, n = num * size, c, r;
>
>	if (!swap_func)
>		swap_func = (size == 4 ? u32_swap : generic_swap);
>
>	/* heapify */
>	for ( ; i >= 0; i -= size) {
>		for (r = i; r * 2 + size < n; r  = c) {
>			c = r * 2 + size;
>			if (c < n - size &&
>					cmp_func(base + c, base + c + size) < 0)
>				c += size;
>			if (cmp_func(base + r, base + c) >= 0)
>				break;
>			swap_func(base + r, base + c, size);
>		}
>	}
>
>	/* sort */
>	for (i = n - size; i > 0; i -= size) {
>		swap_func(base, base + i, size);
>		for (r = 0; r * 2 + size < i; r = c) {
>			c = r * 2 + size;
>			if (c < i - size &&
>					cmp_func(base + c, base + c + size) < 0)
>				c += size;
>			if (cmp_func(base + r, base + c) >= 0)
>				break;
>			swap_func(base + r, base + c, size);
>		}
>	}
>}
>
>int thread_init(thread_t *t)
>{
>	assert(t != NULL);
>	pthread_spin_init(&t->spin, 0);
>	t->count = t->flags = 0;
>	t->flags |= FL_READY;
>	t->buffer = malloc(bufsize);
>	if (t->buffer == NULL)
>		return -1;
>	return 0;
>}
>
>int thread_run(thread_t *t, void *(*start_routine) (void*))
>{
>	return pthread_create(&t->thread, NULL, start_routine, (void *)t);
>}
>
>void thread_consume(thread_t *t)
>{
>	int i;
>	char d_type;
>	for (i=0; i<t->count; i++) {
>		struct linux_dirent *d = t->entries[i];
>		d_type = *((char*)d + d->d_reclen - 1);
>		print("    %8ld  %-10s %4d %10lld  %s\n", d->d_ino,
>		      (d_type == DT_REG) ?  "regular" :
>		      (d_type == DT_DIR) ?  "directory" :
>		      (d_type == DT_FIFO) ? "FIFO" :
>		      (d_type == DT_SOCK) ? "socket" :
>		      (d_type == DT_LNK) ?  "symlink" :
>		      (d_type == DT_BLK) ?  "block dev" :
>		      (d_type == DT_CHR) ?  "char dev" : "???",
>		      d->d_reclen, (long long) d->d_off, d->d_name);
>		if (d_type == DT_REG) { /* 'readahead' by open()ing the files */
>			int fd;
>			char filename[1024];
>			sprintf(filename, "%s/%s", dirname, d->d_name);
>			fd = open(filename, O_RDONLY);
>			if (fd < 0)
>				fprintf(stderr, "%ld >> open(%s) failed: %s\n",
>					syscall(SYS_gettid), filename, strerror(errno));
>			else
>				close(fd);
>		}
>	}
>}
>
>int thread_prep_buffer(thread_t *t)
>{
>	/* Create the pointers to the dirents, using some creative
>	 * estimation. sizeof(struct linux_dirent) gives us the 
>	 * size of a dirent w/o accounting for the name length. We
>	 * shall use the namelen provided to start with */
>	int bpos, ptrarrlen = t->nread / (sizeof(struct linux_dirent) + namelen);
>	pdebug("%ld >> ptrarrlen:%d nread:%d namelen:%u sizeof:%lu\n", syscall(SYS_gettid),
>	       ptrarrlen, t->nread, namelen, sizeof(struct linux_dirent));
>	t->entries = calloc(sizeof(struct linux_dirent *), ptrarrlen);
>	if (t->entries == NULL)
>		return -1;
>
>	t->count = 0;
>	for (bpos = 0; bpos < t->nread; t->count++) {
>		if (t->count >= ptrarrlen) { /* Our ptr array isn't big enough */
>			ptrarrlen <<= 1; /* Double the len */
>			pdebug("%ld >> entries before realloc: %p\n", syscall(SYS_gettid), t->entries);
>			t->entries = realloc(t->entries, sizeof(struct linux_dirent *) * ptrarrlen);
>			if (t->entries == NULL)
>				return -1;
>			pdebug("%ld >> realloced - new ptrarrlen: %d entries:%p entries[count]:%p\n",
>			       syscall(SYS_gettid), ptrarrlen, t->entries, t->entries[t->count]);
>		}
>		t->entries[t->count] = (struct linux_dirent *) (t->buffer + bpos);
>		bpos += t->entries[t->count]->d_reclen;
>	}
>	pdebug("%ld >> Adjusted pointers for %u entries\n", syscall(SYS_gettid),
>	       t->count);
>	if (sort) {
>		heapsort(t->entries, t->count, sizeof(struct linux_dirent*), ino_compare, NULL);
>		pdebug("%ld >> Sorted pointers by inode number\n", syscall(SYS_gettid));
>	}
>	return 0;
>}
>
>void* thread_work_func(void *context)
>{
>	thread_t *me = (thread_t *) context;
>	const struct timespec rgtp = {.tv_sec = 0, .tv_nsec = THREAD_SLEEP_NS, };
>
>	while (1) {
>		pthread_spin_lock(&me->spin);
>		if (me->flags & FL_NEWDATA) {
>			pthread_spin_unlock(&me->spin);
>			pdebug("%ld >> new data on thread:%p\n",
>			       syscall(SYS_gettid), me);
>			/* Consume the entries in the buffer */
>			if (thread_prep_buffer(me)) {
>				fprintf(stderr, "%ld >> Thread prep failed: %s\n",
>					syscall(SYS_gettid), strerror(errno));
>				break;
>			}
>			thread_consume(me);
>			pthread_spin_lock(&me->spin);
>			me->flags &= ~FL_NEWDATA;
>			me->flags |= FL_READY;
>		}
>		if (me->flags & FL_TERMINATE) {
>			pdebug("%ld >> Received terminate flag, exiting thread\n",
>			       syscall(SYS_gettid));
>			pthread_spin_unlock(&me->spin);
>			break;
>		}
>		pthread_spin_unlock(&me->spin);
>		nanosleep(&rgtp, NULL);
>	}
>	
>	return NULL;
>}
>
>void thread_cleanup(thread_t *t)
>{
>	free(t->buffer);
>	free(t->entries);
>	memset(t, 0, sizeof(thread_t));
>}
>
>struct thread_pool {
>	thread_t *threads;
>	unsigned int active_count;
>};
>typedef struct thread_pool thread_pool_t;
>
>void thread_pool_cleanup(thread_pool_t *tpool)
>{
>	int i;
>	for (i=0; i<tpool->active_count; i++)
>		thread_cleanup(&tpool->threads[i]);
>}
>
>thread_t* thread_find_free(thread_pool_t *tpool)
>{
>	int i;
>	thread_t *t;
>
>	for (i=0; i<tpool->active_count; i++) {
>		t = &(tpool->threads[i]);
>		pthread_spin_lock(&t->spin);
>		if (t->flags & FL_READY)
>			return t; /* caller must unlock spinlock */
>		pthread_spin_unlock(&t->spin);
>	}
>	return NULL; /* Couldn't find a free thread */
>}
>
>int fd;
>thread_pool_t tpool;
>pid_t master;
>
>void sigint_cleanup(int sig)
>{
>	if (syscall(SYS_gettid) == master) {
>		int i;
>		close(fd);
>		for (i=0; i<tpool.active_count; i++)
>			pthread_kill(tpool.threads[i].thread, sig);
>		for (i=0; i<tpool.active_count; i++)
>			pthread_join(tpool.threads[i].thread, NULL);
>		thread_pool_cleanup(&tpool);
>		pthread_exit(NULL);
>	}
>	pthread_exit(NULL);
>}
>
>int main(int argc, char **argv)
>{
>	int err = 0, i;
>
>	master = syscall(SYS_gettid);
>	memset(&tpool, 0, sizeof(thread_pool_t));
>
>	set_defaults();
>	parse_opts(argc, argv);
>
>	fd = open(dirname, O_RDONLY | O_DIRECTORY);
>	if (fd == -1) {
>		fprintf(stderr, "open(%s) failed:%s\n", dirname, strerror(errno));
>		exit(EXIT_FAILURE);
>	}
>
>	tpool.threads = malloc(sizeof(thread_t) * maxthreads);
>	for (i=0; i<maxthreads; i++) {
>		err = thread_init(&tpool.threads[i]);
>		if (err) {
>			fprintf(stderr, "Thread initialization failed: %s\n", strerror(errno));
>			goto thread_cleanup;
>		}
>		err = thread_run(&tpool.threads[i], thread_work_func);
>		if (err) {
>			fprintf(stderr, "Thread run failed: %s\n", strerror(errno));
>			goto thread_cleanup;
>		}
>		tpool.active_count++;
>	}
>
>	signal(SIGINT, sigint_cleanup);
>	
>	for ( ; ; ) {
>		thread_t *t = thread_find_free(&tpool);
>		if (t == NULL)
>			continue;
>
>		pdebug("picked thread %p\n", t);
>		t->nread = syscall(SYS_getdents, fd, t->buffer, bufsize);
>		pdebug("nread in main: %d\n", t->nread);
>		if (t->nread > 0) {
>			t->flags |= FL_NEWDATA;
>			t->flags &= ~FL_READY;
>			pthread_spin_unlock(&t->spin);
>			pdebug("%ld >> new data on thread:%p\n", syscall(SYS_gettid), t);
>			continue;
>
>		} else 	if (t->nread == -1) {
>			pthread_spin_unlock(&t->spin);
>			fprintf(stderr, "getdents() syscall failed: %s\n", strerror(errno));
>			goto thread_cleanup;
>
>		} else 	if (t->nread == 0) {
>			pthread_spin_unlock(&t->spin);
>			/* No more entries available */
>			for (i=0; i<tpool.active_count; i++) {
>				thread_t *tt = &tpool.threads[i];
>				pthread_spin_lock(&tt->spin);
>				tt->flags |= FL_TERMINATE;
>				pthread_spin_unlock(&tt->spin);
>			}
>			break;
>		}		
>	}
>
>	close(fd);
>	for (i=0; i<tpool.active_count; i++)
>		pthread_join(tpool.threads[i].thread, NULL);
>
>thread_cleanup:
>	thread_pool_cleanup(&tpool);
>
>	return 0;
>}

Actions: View

Attachments on bug 850426: 605960 | 617073 | 625592 | 678841 | 680222 | 680282 | 708389 | 731269 | 731270 | 731272 | 731273 | 731274 | 731278 | 731279 | 732739 | 732740 | 732845 | 834840 | 853502 | 909528 | 909529 | 909534 | 912745 | 917238 | 917239 | 920895 | 920896 | 920897 | 920898 | 920899 | 920902 | 920903 | 947987