Add threading support a-la pthread.h

1 files changed, 257 insertions, 63 deletions

diff --git a/cbs.h b/cbs.h
index aa75101..fed1555 100644
--- a/cbs.h
+++ b/cbs.h
@@ -2,13 +2,21 @@
    POSIX compliant, so it should work on all respectible UNIX-like systems.
 
    All functions and macros are documented.  You can figure out the API pretty
-   easily by just reading the comments in this file.  Any identifier prefixed
-   with a double-underscore (‘__’) is not meant for you to touch, but since this
-   file should be downloaded into your repository, you can touch them anyways if
-   you really want.
+   easily by just reading the comments in this file.
+
+   In many cases you may want to be able to execute commands on multiple threads
+   to speed up compilation, such as the -j option when using Make.  Functions
+   for creating and using thread pools will be made available if the CBS_PTHREAD
+   macro is defined before including this file.  Do note that on POSIX platforms
+   it will require linking to the pthreads library when bootstrapping the build
+   script.
 
    This file does not support C89.  Fuck C89, that shit is ancient.  Move on.
 
+   IMPORTANT NOTE: Any identifiers prefixed with an underscore (e.g. ‘_rebuild’)
+   are meant for internal use only and you should not touch them unless you know
+   what you’re doing.
+
    IMPORTANT NOTE: All the functions and macros in this library will terminate
    the program on error.  If this is undesired behavior, feel free to edit the
    functions to return errors.
@@ -37,6 +45,9 @@
 #include <sys/wait.h>
 
 #include <errno.h>
+#ifdef CBS_PTHREAD
+#	include <pthread.h>
+#endif
 #include <stdarg.h>
 #include <stdint.h>
 #include <stdio.h>
@@ -73,8 +84,8 @@
 
 /* Internal global versions of argc and argv, so our functions and macros can
    access them from anywhere. */
-static int __cbs_argc;
-static char **__cbs_argv;
+static int _cbs_argc;
+static char **_cbs_argv;
 
 /* A wrapper function around realloc().  It behaves exactly the same except
    instead of taking a buffer size as an argument, it takes a count n of
@@ -98,27 +109,31 @@ static void cbsinit(int, char **);
 /* Get the number of items in the array a */
 #define lengthof(a) (sizeof(a) / sizeof(*(a)))
 
-/* Struct representing a CLI command that various functions act on.  You will
-   basically always want to zero-initialize variables of this type before use.
+/* Struct representing a CLI command that various functions act on.  You should
+   always zero-initialize variables of this type before use.
 
    After executing a command, you can reuse the already allocated buffer this
    command holds by calling cmdclr().  When you’re really done with an object of
-   this type, remember to call free() on .argv. */
-struct cmd {
-	char **argv;
-	size_t len, cap;
-};
+   this type, remember to call free() on ._argv.
+
+   The ._argv field is a NULL-terminated list of command arguments of length
+   ._len.  You may safely read from both of these fields but they should NOT be
+   modified without use of cmdadd() and cmdaddv(). */
+typedef struct {
+	char **_argv;
+	size_t _len, _cap;
+} cmd_t;
 
 /* cmdadd() adds the variadic string arguments to the given command.
    Alternatively, the cmdaddv() function adds the n strings pointed to by p to
    the given command. */
-static void cmdaddv(struct cmd *, char **p, size_t n);
+static void cmdaddv(cmd_t *, char **p, size_t n);
 #define cmdadd(cmd, ...) \
 	cmdaddv(cmd, ((char *[]){__VA_ARGS__}), lengthof(((char *[]){__VA_ARGS__})))
 
 /* Clear (but not free) the command c.  Useful for reusing the same command
    struct to minimize allocations. */
-static void cmdclr(struct cmd *c);
+static void cmdclr(cmd_t *c);
 
 /* The cmdexec() function executes the given command and waits for it to
    terminate, returning its exit code.  The cmdexeca() function executes the
@@ -129,9 +144,9 @@ static void cmdclr(struct cmd *c);
    size of the output in *n.
 
    cmdexec() and cmdexecb() have the same return values as cmdwait(). */
-static int cmdexec(struct cmd);
-static pid_t cmdexeca(struct cmd);
-static int cmdexecb(struct cmd, char **p, size_t *n);
+static int cmdexec(cmd_t);
+static pid_t cmdexeca(cmd_t);
+static int cmdexecb(cmd_t, char **p, size_t *n);
 
 /* Wait for the process with the given PID to terminate, and return its exit
    status.  If the process was terminated by a signal 256 is returned. */
@@ -141,8 +156,8 @@ static int cmdwait(pid_t);
    can be used to mimick the echoing behavior of make(1).  The cmdput() function
    is a nice convenience function so you can avoid writing ‘stdout’ all the
    time. */
-static void cmdput(struct cmd);
-static void cmdputf(FILE *, struct cmd);
+static void cmdput(cmd_t);
+static void cmdputf(FILE *, cmd_t);
 
 /* Returns if a file exists at the given path.  A return value of false may also
    mean you don’t have the proper file access permissions, which will also set
@@ -164,9 +179,9 @@ static bool fmdnewer(char *, char *);
 /* Rebuild the build script if either it, or this header file have been
    modified, and execute the newly built script.  You should call the rebuild()
    macro at the very beginning of main(), but right after cbsinit().  You
-   probably don’t want to call __rebuild() directly. */
-static void __rebuild(char *);
-#define rebuild() __rebuild(__FILE__)
+   probably don’t want to call _rebuild() directly. */
+static void _rebuild(char *);
+#define rebuild() _rebuild(__FILE__)
 
 /* Get the number of available CPUs, or -1 on error.  This function also returns
    -1 if the _SC_NPROCESSORS_ONLN flag to sysconf(3) is not available.  In that
@@ -186,14 +201,69 @@ static int nproc(void);
    VALID library name.
 
    The arguments this function appends to the given command are heap-allocated.
-   If you care about free()ing them, you can figure out their indicies in
-   cmd.argv by getting cmd.len both before- and after calling this function. */
-static bool pcquery(struct cmd *, char *lib, int flags);
+   If you care about deallocating them, you can figure out their indicies in
+   the commands ._argv field by getting cmd._len both before and after calling
+   this function. */
+static bool pcquery(cmd_t *, char *lib, int flags);
 enum pkg_config_flags {
 	PKGC_LIBS = 1 << 0,
 	PKGC_CFLAGS = 1 << 1,
 };
 
+#ifdef CBS_PTHREAD
+
+/* A tfunc_t represents a function to be executed by a threads in a thread pool.
+   It takes an argument in the form of a void * and returns nothing. */
+typedef void (*tfunc_t)(void *);
+
+/* A tfunc_free_t represents a function which frees the argument passed to a
+   tfunc_t function. */
+typedef void (*tfree_func_t)(void *);
+
+/* A thread pool job queue.  Meant for internal-use only. */
+struct _tjob {
+	void *arg;
+	tfunc_t fn;
+	tfree_func_t free;
+	struct _tjob *next;
+};
+
+/* A basic thread pool.  None of its fields should really be touched. */
+typedef struct {
+	bool _stop;
+	size_t _tcnt, _left;
+	pthread_t *_thrds;
+	pthread_cond_t _cnd;
+	pthread_mutex_t _mtx;
+	struct _tjob *_head, *_tail;
+} tpool_t;
+
+/* Initialize and destroy a thread pool.  The tpinit() function initializes the
+   given thread pool and creates n threads ready to execute work.  The tpfree()
+   function should be called after a thread pool has been used to release all
+   resources used by the thread pool. */
+static void tpinit(tpool_t *, size_t n);
+static void tpfree(tpool_t *);
+
+/* Wait for all jobs on the given thread pool to be executed.  Note that this
+   function does not destroy the threads or free any resources — those are tasks
+   for the tpfree() function. */
+static void tpwait(tpool_t *);
+
+/* Enqueue and dequeue jobs to the thread pools job queue.  The tpenq() function
+   is threadsafe while the _tpdeq() function is not (so don’t use it).  When
+   calling the tpenq() function, the function fn will be queued to be executed
+   by a thread in the thread pool with the argument arg.  If the given argument
+   needs to be deallocated after the job completes, you can pass the free
+   argument which will be called with the given argument after use.  If free is
+   NULL, it will be ignored.
+
+   The free() function is a valid argument to the free parameter. */
+static void tpenq(tpool_t *, tfunc_t fn, void *arg, tfree_func_t free);
+static struct _tjob *_tpdeq(tpool_t *);
+
+#endif /* CBS_PTHREAD */
+
 /* BEGIN DEFINITIONS */
 
 void *
@@ -216,7 +286,7 @@ die(const char *fmt, ...)
 	va_list ap;
 
 	va_start(ap, fmt);
-	fprintf(stderr, "%s: ", *__cbs_argv);
+	fprintf(stderr, "%s: ", *_cbs_argv);
 	if (fmt) {
 		vfprintf(stderr, fmt, ap);
 		fprintf(stderr, ": ");
@@ -231,7 +301,7 @@ diex(const char *fmt, ...)
 	va_list ap;
 
 	va_start(ap, fmt);
-	fprintf(stderr, "%s: ", *__cbs_argv);
+	fprintf(stderr, "%s: ", *_cbs_argv);
 	if (fmt)
 		vfprintf(stderr, fmt, ap);
 	fputc('\n', stderr);
@@ -241,12 +311,12 @@ diex(const char *fmt, ...)
 void
 cbsinit(int argc, char **argv)
 {
-	__cbs_argc = argc;
-	__cbs_argv = argv;
+	_cbs_argc = argc;
+	_cbs_argv = argv;
 }
 
 static size_t
-__next_powerof2(size_t n)
+_next_powerof2(size_t n)
 {
 	if (n && !(n & (n - 1)))
 		return n;
@@ -258,33 +328,33 @@ __next_powerof2(size_t n)
 }
 
 void
-cmdaddv(struct cmd *cmd, char **xs, size_t n)
+cmdaddv(cmd_t *cmd, char **xs, size_t n)
 {
-	if (cmd->len + n >= cmd->cap) {
-		cmd->cap = __next_powerof2(cmd->len + n) + 2;
-		cmd->argv = bufalloc(cmd->argv, cmd->cap, sizeof(char *));
+	if (cmd->_len + n >= cmd->_cap) {
+		cmd->_cap = _next_powerof2(cmd->_len + n) + 2;
+		cmd->_argv = bufalloc(cmd->_argv, cmd->_cap, sizeof(char *));
 	}
 
-	memcpy(cmd->argv + cmd->len, xs, n * sizeof(*xs));
-	cmd->len += n;
-	cmd->argv[cmd->len] = NULL;
+	memcpy(cmd->_argv + cmd->_len, xs, n * sizeof(*xs));
+	cmd->_len += n;
+	cmd->_argv[cmd->_len] = NULL;
 }
 
 void
-cmdclr(struct cmd *c)
+cmdclr(cmd_t *c)
 {
-	c->len = 0;
-	*c->argv = NULL;
+	c->_len = 0;
+	*c->_argv = NULL;
 }
 
 int
-cmdexec(struct cmd c)
+cmdexec(cmd_t c)
 {
 	return cmdwait(cmdexeca(c));
 }
 
 pid_t
-cmdexeca(struct cmd c)
+cmdexeca(cmd_t c)
 {
 	pid_t pid;
 
@@ -292,15 +362,15 @@ cmdexeca(struct cmd c)
 	case -1:
 		die("fork");
 	case 0:
-		execvp(*c.argv, c.argv);
-		die("execvp: %s", *c.argv);
+		execvp(*c._argv, c._argv);
+		die("execvp: %s", *c._argv);
 	}
 
 	return pid;
 }
 
 int
-cmdexecb(struct cmd c, char **p, size_t *n)
+cmdexecb(cmd_t c, char **p, size_t *n)
 {
 	enum {
 		FD_R,
@@ -322,8 +392,8 @@ cmdexecb(struct cmd c, char **p, size_t *n)
 		close(fds[FD_R]);
 		if (dup2(fds[FD_W], STDOUT_FILENO) == -1)
 			die("dup2");
-		execvp(*c.argv, c.argv);
-		die("execvp: %s", *c.argv);
+		execvp(*c._argv, c._argv);
+		die("execvp: %s", *c._argv);
 	}
 
 	close(fds[FD_W]);
@@ -379,19 +449,19 @@ cmdwait(pid_t pid)
 	"%+,-./0123456789:=@ABCDEFGHIJKLMNOPQRSTUVWXYZ_abcdefghijklmnopqrstuvwxyz"
 
 void
-cmdput(struct cmd c)
+cmdput(cmd_t c)
 {
 	cmdputf(stdout, c);
 }
 
 void
-cmdputf(FILE *stream, struct cmd cmd)
+cmdputf(FILE *stream, cmd_t cmd)
 {
-	for (size_t i = 0; i < cmd.len; i++) {
+	for (size_t i = 0; i < cmd._len; i++) {
 		bool safe = true;
 		char *p, *q;
 
-		p = q = cmd.argv[i];
+		p = q = cmd._argv[i];
 		for (; *q; q++) {
 			if (!strchr(SHELL_SAFE, *q)) {
 				safe = false;
@@ -412,7 +482,7 @@ cmdputf(FILE *stream, struct cmd cmd)
 			putc('\'', stream);
 		}
 
-		putc(i == cmd.len - 1 ? '\n' : ' ', stream);
+		putc(i == cmd._len - 1 ? '\n' : ' ', stream);
 	}
 }
 
@@ -450,22 +520,26 @@ fmdolder(char *lhs, char *rhs)
 }
 
 void
-__rebuild(char *src)
+_rebuild(char *src)
 {
-	struct cmd cmd = {0};
+	cmd_t cmd = {0};
 
-	if (fmdnewer(*__cbs_argv, src) && fmdnewer(*__cbs_argv, __FILE__))
+	if (fmdnewer(*_cbs_argv, src) && fmdnewer(*_cbs_argv, __FILE__))
 		return;
 
-	cmdadd(&cmd, "cc", "-o", *__cbs_argv, src);
+	cmdadd(&cmd, "cc");
+#ifdef CBS_PTHREAD
+	cmdadd(&cmd, "-lpthread");
+#endif
+	cmdadd(&cmd, "-o", *_cbs_argv, src);
 	cmdput(cmd);
 	if (cmdexec(cmd))
 		diex("Compilation of build script failed");
 
 	cmdclr(&cmd);
-	cmdaddv(&cmd, __cbs_argv, __cbs_argc);
-	execvp(*cmd.argv, cmd.argv);
-	die("execvp: %s", *cmd.argv);
+	cmdaddv(&cmd, _cbs_argv, _cbs_argc);
+	execvp(*cmd._argv, cmd._argv);
+	die("execvp: %s", *cmd._argv);
 }
 
 int
@@ -480,12 +554,12 @@ nproc(void)
 }
 
 bool
-pcquery(struct cmd *cmd, char *lib, int flags)
+pcquery(cmd_t *cmd, char *lib, int flags)
 {
 	int ec;
 	char *p, *q, *s;
 	size_t n;
-	struct cmd c = {0};
+	cmd_t c = {0};
 
 	p = NULL;
 
@@ -498,7 +572,7 @@ pcquery(struct cmd *cmd, char *lib, int flags)
 
 	if ((ec = cmdexecb(c, &p, &n))) {
 		if (errno == ENOENT) {
-			free(c.argv);
+			free(c._argv);
 			return false;
 		}
 		diex("pkg-config terminated with exit-code %d", ec);
@@ -513,4 +587,124 @@ pcquery(struct cmd *cmd, char *lib, int flags)
 	return true;
 }
 
+#ifdef CBS_PTHREAD
+
+static void *
+_tpwork(void *arg)
+{
+	tpool_t *tp = arg;
+
+	while (!tp->_stop) {
+		struct _tjob *j;
+
+		pthread_mutex_lock(&tp->_mtx);
+		while (!tp->_stop && !tp->_head)
+			pthread_cond_wait(&tp->_cnd, &tp->_mtx);
+		if (tp->_stop) {
+			pthread_mutex_unlock(&tp->_mtx);
+			break;
+		}
+
+		j = _tpdeq(tp);
+		pthread_mutex_unlock(&tp->_mtx);
+
+		j->fn(j->arg);
+		if (j->free)
+			j->free(j->arg);
+		free(j);
+
+		pthread_mutex_lock(&tp->_mtx);
+		tp->_left--;
+		pthread_cond_broadcast(&tp->_cnd);
+		pthread_mutex_unlock(&tp->_mtx);
+	}
+
+	return NULL;
+}
+
+void
+tpinit(tpool_t *tp, size_t n)
+{
+	tp->_tcnt = n;
+	tp->_stop = false;
+	tp->_left = 0;
+	tp->_head = tp->_tail = NULL;
+	tp->_thrds = bufalloc(NULL, n, sizeof(pthread_t));
+	pthread_cond_init(&tp->_cnd, NULL);
+	pthread_mutex_init(&tp->_mtx, NULL);
+
+	for (size_t i = 0; i < n; i++)
+		pthread_create(tp->_thrds + i, NULL, _tpwork, tp);
+}
+
+void
+tpfree(tpool_t *tp)
+{
+	tp->_stop = true;
+
+	pthread_mutex_lock(&tp->_mtx);
+	pthread_cond_broadcast(&tp->_cnd);
+	pthread_mutex_unlock(&tp->_mtx);
+
+	for (size_t i = 0; i < tp->_tcnt; i++)
+		pthread_join(tp->_thrds[i], NULL);
+
+	free(tp->_thrds);
+	while (tp->_head) {
+		struct _tjob *j = _tpdeq(tp);
+		if (j->free)
+			j->free(j->arg);
+		free(j);
+	}
+
+	pthread_cond_destroy(&tp->_cnd);
+	pthread_mutex_destroy(&tp->_mtx);
+}
+
+struct _tjob *
+_tpdeq(tpool_t *tp)
+{
+	struct _tjob *j = tp->_head;
+
+	if (j) {
+		tp->_head = tp->_head->next;
+		if (!tp->_head)
+			tp->_tail = NULL;
+	}
+
+	return j;
+}
+
+void
+tpenq(tpool_t *tp, tfunc_t fn, void *arg, tfree_func_t free)
+{
+	struct _tjob *j = bufalloc(NULL, 1, sizeof(struct _tjob));
+	*j = (struct _tjob){
+		.fn = fn,
+		.arg = arg,
+		.free = free,
+	};
+
+	pthread_mutex_lock(&tp->_mtx);
+	if (tp->_tail)
+		tp->_tail->next = j;
+	if (!tp->_head)
+		tp->_head = j;
+	tp->_tail = j;
+	tp->_left++;
+	pthread_cond_signal(&tp->_cnd);
+	pthread_mutex_unlock(&tp->_mtx);
+}
+
+void
+tpwait(tpool_t *tp)
+{
+	pthread_mutex_lock(&tp->_mtx);
+	while (!tp->_stop && tp->_left)
+		pthread_cond_wait(&tp->_cnd, &tp->_mtx);
+	pthread_mutex_unlock(&tp->_mtx);
+}
+
+#endif /* CBS_PTHREAD */
+
 #endif /* !C_BUILD_SYSTEM_H */