Implementation of a Concurrent Indexing Algorithm

configure.ac

@@ -165,6 +165,11 @@ The X11 library was not found, which is needed for x11 gui support.
 fi
 
 
+AC_CHECK_LIB([pthread], [pthread_create],, [AC_MSG_ERROR([
+The pthread library was not found, which is needed for concurrency in the indexing algorithm.
+])])
+
+
 AC_CHECK_HEADERS([fcntl.h limits.h stdint.h stdlib.h string.h sys/ioctl.h unistd.h fnmatch.h termios.h])
 AC_CHECK_HEADERS([ncurses.h ncurses/ncurses.h ncursesw/ncurses.h])
 

src/duc/cmd-index.c

@@ -26,6 +26,8 @@ static bool opt_hide_file_names = false;
 static char *opt_username = NULL;
 static int opt_uid = 0;
 static int opt_max_depth = 0;
+static unsigned int opt_thread_count = 1;
+static unsigned int opt_cutoff_depth = 2;
 static bool opt_one_file_system = false;
 static bool opt_progress = false;
 static bool opt_uncompressed = false;
@@ -73,6 +75,8 @@ static int index_main(duc *duc, int argc, char **argv)
 
 	if(opt_force) open_flags |= DUC_OPEN_FORCE;
 	if(opt_max_depth) duc_index_req_set_maxdepth(req, opt_max_depth);
+	if(opt_thread_count) duc_index_set_worker_count(opt_thread_count);
+	if(opt_cutoff_depth) duc_index_set_cutoff_depth(opt_cutoff_depth);
 	if(opt_one_file_system) index_flags |= DUC_INDEX_XDEV;
 	if(opt_hide_file_names) index_flags |= DUC_INDEX_HIDE_FILE_NAMES;
 	if(opt_check_hard_links) index_flags |= DUC_INDEX_CHECK_HARD_LINKS;
@@ -174,11 +178,16 @@ static struct ducrc_option options[] = {
 	  "VAL is a comma separated list of file system types as found in your systems fstab, for example ext3,ext4,dosfs" },
 	{ &opt_hide_file_names, "hide-file-names",  0 , DUCRC_TYPE_BOOL,   "hide file names in index (privacy)", 
 	  "the names of directories will be preserved, but the names of the individual files will be hidden" },
-	{ &opt_uid,             "uid",              'U', DUCRC_TYPE_INT,    "limit index to only files/dirs owned by uid" },
-	{ &opt_username,        "username",         'u', DUCRC_TYPE_STRING, "limit index to only files/dirs owned by username" },
+	{ &opt_uid,             "uid",             'U', DUCRC_TYPE_INT,    "limit index to only files/dirs owned by uid" },
+	{ &opt_username,        "username",        'u', DUCRC_TYPE_STRING, "limit index to only files/dirs owned by username" },
 	{ &opt_max_depth,       "max-depth",       'm', DUCRC_TYPE_INT,    "limit directory names to given depth" ,
 	  "when this option is given duc will traverse the complete file system, but will only the first VAL "
 	  "levels of directories in the database to reduce the size of the index" },
+	{ &opt_thread_count,    "thread-count",    't', DUCRC_TYPE_INT,    "stipulate index to use a given number of threads",
+	"when this option is given duc will multithread the indexing algorithm and use VAL workers to traverse the filesystem. "
+	"Default value is 1."},
+	{ &opt_cutoff_depth,    "cutoff-depth",    'c', DUCRC_TYPE_INT,    "ensures that each worker has at least VAL tasks before "
+	"other workers start taking its tasks", "In general the lower this is, the higher the concurrency. Default value is 2."},
 	{ &opt_one_file_system, "one-file-system", 'x', DUCRC_TYPE_BOOL,   "skip directories on different file systems" },
 	{ &opt_progress,        "progress",        'p', DUCRC_TYPE_BOOL,   "show progress during indexing" },
 	{ &opt_dryrun,          "dry-run",          0 , DUCRC_TYPE_BOOL,   "do not update database, just crawl" },

src/libduc/buffer.c

@@ -6,6 +6,7 @@
 #include <string.h>
 #include <assert.h>
 #include <stdint.h>
+#include <pthread.h>
 
 #include "private.h"
 #include "buffer.h"
@@ -23,6 +24,7 @@ struct buffer *buffer_new(void *data, size_t len)
 
 	b = duc_malloc(sizeof(struct buffer));
 	b->ptr = 0;
+	pthread_rwlock_init(&b->data_lock, NULL);
 
 	if(data) {
 		b->max = len;
@@ -40,6 +42,7 @@ struct buffer *buffer_new(void *data, size_t len)
 
 void buffer_free(struct buffer *b)
 {
+	pthread_rwlock_destroy(&b->data_lock);
 	duc_free(b->data);
 	duc_free(b);
 }
@@ -162,31 +165,42 @@ static void buffer_get_size(struct buffer *b, struct duc_size *size)
 
 void buffer_put_dir(struct buffer *b, const struct duc_devino *devino, time_t mtime)
 {
+	/* Putting a directory into a buffer should be an atomic operation */
+	pthread_rwlock_wrlock(&b->data_lock);
+
 	buffer_put_devino(b, devino);
 	buffer_put_varint(b, mtime);
+
+	pthread_rwlock_unlock(&b->data_lock);
 }
 
 
 void buffer_get_dir(struct buffer *b, struct duc_devino *devino, time_t *mtime)
 {
+	pthread_rwlock_rdlock(&b->data_lock);
 	uint64_t v;
 	buffer_get_devino(b, devino);
 	buffer_get_varint(b, &v); *mtime = v;
+	pthread_rwlock_unlock(&b->data_lock);
 }
 
 void buffer_put_dirent(struct buffer *b, const struct duc_dirent *ent)
 {
+	/* Putting a directory entry into a buffer should be an atomic operation */
+	pthread_rwlock_wrlock(&b->data_lock);
 	buffer_put_string(b, ent->name);
 	buffer_put_size(b, &ent->size);
 	buffer_put_varint(b, ent->type);
 
 	if(ent->type == DUC_FILE_TYPE_DIR) {
 		buffer_put_devino(b, &ent->devino);
 	}
+	pthread_rwlock_unlock(&b->data_lock);
 }
 
 void buffer_get_dirent(struct buffer *b, struct duc_dirent *ent)
 {
+	pthread_rwlock_rdlock(&b->data_lock);
 	uint64_t v;
 
 	buffer_get_string(b, &ent->name);
@@ -196,11 +210,14 @@ void buffer_get_dirent(struct buffer *b, struct duc_dirent *ent)
 	if(ent->type == DUC_FILE_TYPE_DIR) {
 		buffer_get_devino(b, &ent->devino);
 	}
+	pthread_rwlock_unlock(&b->data_lock);
 }
 
 
 void buffer_put_index_report(struct buffer *b, const struct duc_index_report *report)
 {
+	/* Putting the index report into the buffer should be an atomic operation */
+	pthread_rwlock_wrlock(&b->data_lock);
 	buffer_put_string(b, report->path);
 	buffer_put_devino(b, &report->devino);
 	buffer_put_varint(b, report->time_start.tv_sec);
@@ -210,11 +227,13 @@ void buffer_put_index_report(struct buffer *b, const struct duc_index_report *re
 	buffer_put_varint(b, report->file_count);
 	buffer_put_varint(b, report->dir_count);
 	buffer_put_size(b, &report->size);
+	pthread_rwlock_unlock(&b->data_lock);
 }
 
 
 void buffer_get_index_report(struct buffer *b, struct duc_index_report *report)
 {
+	pthread_rwlock_rdlock(&b->data_lock);
 	char *vs = NULL;
 	buffer_get_string(b, &vs);
 	if(vs == NULL) return;
@@ -230,6 +249,7 @@ void buffer_get_index_report(struct buffer *b, struct duc_index_report *report)
 	buffer_get_varint(b, &vi); report->file_count = vi;
 	buffer_get_varint(b, &vi); report->dir_count = vi;
 	buffer_get_size(b, &report->size);
+	pthread_rwlock_unlock(&b->data_lock);
 }
 
 

src/libduc/buffer.h

@@ -3,9 +3,10 @@
 
 struct buffer {
 	uint8_t *data;
-	size_t max;
-	size_t len;
-	size_t ptr;
+	pthread_rwlock_t data_lock;
+	_Atomic size_t max;
+	_Atomic size_t len;
+	_Atomic size_t ptr;
 };
 
 struct buffer *buffer_new(void *data, size_t len);

src/libduc/db.c

@@ -34,9 +34,8 @@ duc_errno db_write_report(duc *duc, const struct duc_index_report *report)
 		} else {
 			db_put(duc->db, "duc_index_reports", 17, report->path, sizeof(report->path));
 		}
-	} else {
-		free(tmp);
 	}
+	free(tmp);
 
 	struct buffer *b = buffer_new(NULL, 0);
 

src/libduc/duc.h

@@ -90,18 +90,18 @@ struct duc_devino {
 };
 
 struct duc_size {
-	off_t actual;
-	off_t apparent;
-	off_t count;
+	_Atomic off_t actual;
+	_Atomic off_t apparent;
+	_Atomic off_t count;
 };
 
 struct duc_index_report {
-	char path[DUC_PATH_MAX];        /* Indexed path */
+	char path[DUC_PATH_MAX];    /* Indexed path */
 	struct duc_devino devino;   /* Index top device id and inode number */
 	struct timeval time_start;  /* Index start time */
 	struct timeval time_stop;   /* Index finished time */
-	size_t file_count;          /* Total number of files indexed */
-	size_t dir_count;           /* Total number of directories indexed */
+	_Atomic size_t file_count;  /* Total number of files indexed */
+	_Atomic size_t dir_count;   /* Total number of directories indexed */
 	struct duc_size size;       /* Total size */
 };
 
@@ -150,6 +150,8 @@ int duc_index_req_add_fstype_include(duc_index_req *req, const char *types);
 int duc_index_req_add_fstype_exclude(duc_index_req *req, const char *types);
 int duc_index_req_set_maxdepth(duc_index_req *req, int maxdepth);
 int duc_index_req_set_progress_cb(duc_index_req *req, duc_index_progress_cb fn, void *ptr);
+int duc_index_set_cutoff_depth(unsigned int cutoff_depth);
+int duc_index_set_worker_count(unsigned int worker_count);
 struct duc_index_report *duc_index(duc_index_req *req, const char *path, duc_index_flags flags);
 int duc_index_req_free(duc_index_req *req);
 int duc_index_report_free(struct duc_index_report *rep);