A parallel version of find(1), implementing a DFS-ish over cwd.

parallel_find

//A parallel version of find(1), implementing a DFS-ish over cwd.
//Compile with -lpthread

#include <dirent.h>
#include <fcntl.h>
#include <pthread.h>
#include <stdbool.h>
#include <stddef.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <string.h>
#include <unistd.h>

#define container_of(ptr, type, member) ({			\
	const typeof( ((type *)0)->member ) *__mptr = (ptr);	\
	(type *)( (char *)__mptr - offsetof(type,member) );})

struct singly_linked {
	struct singly_linked * next;
};
typedef struct singly_linked sl;

struct singly_linked_list {
	sl * head;
	pthread_mutex_t head_lock;
	pthread_cond_t cond;
	bool closed;
};
typedef struct singly_linked_list sll;

sl * list_pop_head(sll * list) {
	pthread_mutex_lock(&list->head_lock);
	sl * ret;
	while((ret = list->head) == NULL && !list->closed) {
		pthread_cond_wait(&list->cond, &list->head_lock);
	}

	if(ret != NULL) {
		list->head = ret->next;
	}

	pthread_mutex_unlock(&list->head_lock);
	return ret;
}

void list_push_head(sll * list, sl * item) {
	pthread_mutex_lock(&list->head_lock);
	item->next = list->head;
	list->head = item;

	pthread_cond_signal(&list->cond);
	pthread_mutex_unlock(&list->head_lock);
}

void list_init(sll * list) {
	list->head = NULL;
	list->closed = 0;
	pthread_mutex_init(&list->head_lock, NULL);
	pthread_cond_init(&list->cond, NULL);
}

void list_close(sll * list) {
	list->closed = 1;
	pthread_cond_broadcast(&list->cond);
}

struct dir_info {
	sl list;
	DIR * dir;
	size_t len;
	char * path;
};

void traverse_dir(
	struct dir_info * current,
	sll * work_list
) {
	struct dirent *ep;
	while((ep = readdir(current->dir))) {
		if(strcmp(ep->d_name, "..") == 0 ||
			strcmp(ep->d_name, ".") == 0) {
			continue;
		}

		if(!(ep->d_type & (DT_REG|DT_DIR))) {
			continue;
		}

		if(ep->d_type == DT_LNK) {
			continue;
		}

		int basename_len = strlen(ep->d_name);
		size_t concat_len = basename_len + current->len + 2;
		char * fullpath = malloc(concat_len);

		snprintf(fullpath, concat_len, "%s/%s", current->path, ep->d_name);

		if(ep->d_type & DT_DIR) {
			int down_fd = openat(dirfd(current->dir), ep->d_name, O_RDONLY);
			if(down_fd != -1) {
				//Push the context we were working on to the shared stack
				list_push_head(work_list, &current->list);

				//Create a new context
				current = malloc(sizeof(struct dir_info));
				if(current == NULL) {
					perror(NULL);
					exit(1);
				}

				current->path = fullpath;
				current->dir = fdopendir(down_fd);
				current->len = concat_len;
			}
		} else {
			printf("%s\n", fullpath);
		}
	}

	if(current->len == 1) {
		list_close(work_list);
	}

	closedir(current->dir);
	free(current->path);
	free(current);
}

void traverse_worker(sll * work_list) {
	struct dir_info * work;
	while(true) {
		work = container_of(
			list_pop_head(work_list),
			struct dir_info,
			list
		);
		if(work == NULL) {
			break;
		}
		traverse_dir(
			work,
			work_list
		);
	}
}

int main() {
	sll work_list;

	struct dir_info * work = malloc(sizeof(struct dir_info));
	work->len = 1;
	work->path = malloc(work->len + 1);
	strncpy(work->path, ".", work->len + 1);
	work->dir = opendir(work->path);

	list_init(&work_list);
	list_push_head(&work_list, &work->list);

	int num_threads = 128; //A common value for nr_requests
	pthread_t * threads = calloc(num_threads, sizeof(pthread_t));
	for(uint32_t i = 0; i < num_threads; i++) {
		pthread_create(
			&threads[i],
			NULL,
			(void * (*)(void *)) traverse_worker,
			(void *) &work_list
		);
	}

	for(uint32_t i = 0; i < num_threads; i++) {
		pthread_join(threads[i], NULL);
	}
}