1 files changed, 517 insertions, 0 deletions
diff --git a/lib/pthreadpool/tests.c b/lib/pthreadpool/tests.c
new file mode 100644
index 0000000..08cb59e
--- /dev/null
+++ b/lib/pthreadpool/tests.c
@@ -0,0 +1,517 @@
+#include <stdio.h>
+#include <string.h>
+#include <poll.h>
+#include <errno.h>
+#include <stdlib.h>
+#include <limits.h>
+#include <pthread.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <signal.h>
+#include "pthreadpool_pipe.h"
+#include "pthreadpool_tevent.h"
+
+static int test_init(void)
+{
+	struct pthreadpool_pipe *p;
+	int ret;
+
+	ret = pthreadpool_pipe_init(1, &p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_init failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+	ret = pthreadpool_pipe_destroy(p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_destroy failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+	return 0;
+}
+
+static void test_sleep(void *ptr)
+{
+	int *ptimeout = (int *)ptr;
+	int ret;
+	ret = poll(NULL, 0, *ptimeout);
+	if (ret != 0) {
+		fprintf(stderr, "poll returned %d (%s)\n",
+			ret, strerror(errno));
+	}
+}
+
+static int test_jobs(int num_threads, int num_jobs)
+{
+	char *finished;
+	struct pthreadpool_pipe *p;
+	int timeout = 1;
+	int i, ret;
+
+	finished = (char *)calloc(1, num_jobs);
+	if (finished == NULL) {
+		fprintf(stderr, "calloc failed\n");
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_init(num_threads, &p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_init failed: %s\n",
+			strerror(ret));
+		free(finished);
+		return -1;
+	}
+
+	for (i=0; i<num_jobs; i++) {
+		ret = pthreadpool_pipe_add_job(p, i, test_sleep, &timeout);
+		if (ret != 0) {
+			fprintf(stderr, "pthreadpool_pipe_add_job failed: "
+				"%s\n", strerror(ret));
+			free(finished);
+			return -1;
+		}
+	}
+
+	for (i=0; i<num_jobs; i++) {
+		int jobid = -1;
+		ret = pthreadpool_pipe_finished_jobs(p, &jobid, 1);
+		if (ret < 0) {
+			fprintf(stderr, "pthreadpool_pipe_finished_jobs "
+				"failed: %s\n", strerror(-ret));
+			free(finished);
+			return -1;
+		}
+		if ((ret != 1) || (jobid >= num_jobs)) {
+			fprintf(stderr, "invalid job number %d\n", jobid);
+			free(finished);
+			return -1;
+		}
+		finished[jobid] += 1;
+	}
+
+	for (i=0; i<num_jobs; i++) {
+		if (finished[i] != 1) {
+			fprintf(stderr, "finished[%d] = %d\n",
+				i, finished[i]);
+			free(finished);
+			return -1;
+		}
+	}
+
+	ret = pthreadpool_pipe_destroy(p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_destroy failed: %s\n",
+			strerror(ret));
+		free(finished);
+		return -1;
+	}
+
+	free(finished);
+	return 0;
+}
+
+static int test_busydestroy(void)
+{
+	struct pthreadpool_pipe *p;
+	int timeout = 50;
+	struct pollfd pfd;
+	int ret, jobid;
+
+	ret = pthreadpool_pipe_init(1, &p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_init failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+	ret = pthreadpool_pipe_add_job(p, 1, test_sleep, &timeout);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_add_job failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+	ret = pthreadpool_pipe_destroy(p);
+	if (ret != EBUSY) {
+		fprintf(stderr, "Could destroy a busy pool\n");
+		return -1;
+	}
+
+	pfd.fd = pthreadpool_pipe_signal_fd(p);
+	pfd.events = POLLIN|POLLERR;
+
+	do {
+		ret = poll(&pfd, 1, -1);
+	} while ((ret == -1) && (errno == EINTR));
+
+	ret = pthreadpool_pipe_finished_jobs(p, &jobid, 1);
+	if (ret < 0) {
+		fprintf(stderr, "pthreadpool_pipe_finished_jobs failed: %s\n",
+			strerror(-ret));
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_destroy(p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_destroy failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+	return 0;
+}
+
+static int test_fork(void)
+{
+	struct pthreadpool_pipe *p;
+	pid_t child, waited;
+	int status, ret;
+
+	ret = pthreadpool_pipe_init(1, &p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_init failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+	ret = pthreadpool_pipe_destroy(p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_destroy failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+
+	child = fork();
+	if (child < 0) {
+		perror("fork failed");
+		return -1;
+	}
+	if (child == 0) {
+		exit(0);
+	}
+	waited = wait(&status);
+	if (waited == -1) {
+		perror("wait failed");
+		return -1;
+	}
+	if (waited != child) {
+		fprintf(stderr, "expected child %d, got %d\n",
+			(int)child, (int)waited);
+		return -1;
+	}
+	return 0;
+}
+
+static void busyfork_job(void *private_data)
+{
+	return;
+}
+
+static int test_busyfork(void)
+{
+	struct pthreadpool_pipe *p;
+	int fds[2];
+	struct pollfd pfd;
+	pid_t child, waitret;
+	int ret, jobnum, wstatus;
+
+	ret = pipe(fds);
+	if (ret == -1) {
+		perror("pipe failed");
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_init(1, &p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_init failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_add_job(p, 1, busyfork_job, NULL);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_add_job failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_finished_jobs(p, &jobnum, 1);
+	if (ret != 1) {
+		fprintf(stderr, "pthreadpool_pipe_finished_jobs failed\n");
+		return -1;
+	}
+
+	ret = poll(NULL, 0, 200);
+	if (ret == -1) {
+		perror("poll failed");
+		return -1;
+	}
+
+	child = fork();
+	if (child < 0) {
+		perror("fork failed");
+		return -1;
+	}
+
+	if (child == 0) {
+		ret = pthreadpool_pipe_destroy(p);
+		if (ret != 0) {
+			fprintf(stderr, "pthreadpool_pipe_destroy failed: "
+				"%s\n", strerror(ret));
+			exit(1);
+		}
+		exit(0);
+	}
+
+	ret = close(fds[1]);
+	if (ret == -1) {
+		perror("close failed");
+		return -1;
+	}
+
+	pfd = (struct pollfd) { .fd = fds[0], .events = POLLIN };
+
+	ret = poll(&pfd, 1, 5000);
+	if (ret == -1) {
+		perror("poll failed");
+		return -1;
+	}
+	if (ret == 0) {
+		fprintf(stderr, "Child did not exit for 5 seconds\n");
+		/*
+		 * The child might hang forever in
+		 * pthread_cond_destroy for example. Be kind to the
+		 * system and kill it.
+		 */
+		kill(child, SIGTERM);
+		return -1;
+	}
+	if (ret != 1) {
+		fprintf(stderr, "poll returned %d -- huh??\n", ret);
+		return -1;
+	}
+
+	ret = poll(NULL, 0, 200);
+	if (ret == -1) {
+		perror("poll failed");
+		return -1;
+	}
+
+	waitret = waitpid(child, &wstatus, WNOHANG);
+	if (waitret != child) {
+		fprintf(stderr, "waitpid returned %d\n", (int)waitret);
+		return -1;
+	}
+
+	if (!WIFEXITED(wstatus)) {
+		fprintf(stderr, "child did not properly exit\n");
+		return -1;
+	}
+
+	ret = WEXITSTATUS(wstatus);
+	if (ret != 0) {
+		fprintf(stderr, "child returned %d\n", ret);
+		return -1;
+	}
+
+	return 0;
+}
+
+static int test_busyfork2(void)
+{
+	struct pthreadpool_pipe *p;
+	pid_t child;
+	int ret, jobnum;
+	struct pollfd pfd;
+
+	ret = pthreadpool_pipe_init(1, &p);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_pipe_init failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_add_job(p, 1, busyfork_job, NULL);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_add_job failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_finished_jobs(p, &jobnum, 1);
+	if (ret != 1) {
+		fprintf(stderr, "pthreadpool_pipe_finished_jobs failed\n");
+		return -1;
+	}
+
+	ret = poll(NULL, 0, 10);
+	if (ret == -1) {
+		perror("poll failed");
+		return -1;
+	}
+
+	ret = pthreadpool_pipe_add_job(p, 1, busyfork_job, NULL);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_add_job failed: %s\n",
+			strerror(ret));
+		return -1;
+	}
+
+	/*
+	 * Do the fork right after the add_job. This tests a race
+	 * where the atfork prepare handler gets all idle threads off
+	 * the condvar. If we are faster doing the fork than the
+	 * existing idle thread could get out of idle and take the
+	 * job, after the fork we end up with no threads to take care
+	 * of the job.
+	 */
+
+	child = fork();
+	if (child < 0) {
+		perror("fork failed");
+		return -1;
+	}
+
+	if (child == 0) {
+		exit(0);
+	}
+
+	pfd = (struct pollfd) {
+		.fd = pthreadpool_pipe_signal_fd(p),
+		.events = POLLIN|POLLERR
+	};
+
+	do {
+		ret = poll(&pfd, 1, 5000);
+	} while ((ret == -1) && (errno == EINTR));
+
+	if (ret == 0) {
+		fprintf(stderr, "job unfinished after 5 seconds\n");
+		return -1;
+	}
+
+	return 0;
+}
+
+static void test_tevent_wait(void *private_data)
+{
+	int *timeout = private_data;
+	poll(NULL, 0, *timeout);
+}
+
+static int test_tevent_1(void)
+{
+	struct tevent_context *ev;
+	struct pthreadpool_tevent *pool;
+	struct tevent_req *req1, *req2;
+	int timeout10 = 10;
+	int timeout100 = 100;
+	int ret;
+	bool ok;
+
+	ev = tevent_context_init(NULL);
+	if (ev == NULL) {
+		ret = errno;
+		fprintf(stderr, "tevent_context_init failed: %s\n",
+			strerror(ret));
+		return ret;
+	}
+	ret = pthreadpool_tevent_init(ev, UINT_MAX, &pool);
+	if (ret != 0) {
+		fprintf(stderr, "pthreadpool_tevent_init failed: %s\n",
+			strerror(ret));
+		TALLOC_FREE(ev);
+		return ret;
+	}
+	req1 = pthreadpool_tevent_job_send(
+		ev, ev, pool, test_tevent_wait, &timeout10);
+	if (req1 == NULL) {
+		fprintf(stderr, "pthreadpool_tevent_job_send failed\n");
+		TALLOC_FREE(ev);
+		return ENOMEM;
+	}
+	req2 = pthreadpool_tevent_job_send(
+		ev, ev, pool, test_tevent_wait, &timeout100);
+	if (req2 == NULL) {
+		fprintf(stderr, "pthreadpool_tevent_job_send failed\n");
+		TALLOC_FREE(ev);
+		return ENOMEM;
+	}
+	ok = tevent_req_poll(req2, ev);
+	if (!ok) {
+		ret = errno;
+		fprintf(stderr, "tevent_req_poll failed: %s\n",
+			strerror(ret));
+		TALLOC_FREE(ev);
+		return ret;
+	}
+	ret = pthreadpool_tevent_job_recv(req1);
+	TALLOC_FREE(req1);
+	if (ret != 0) {
+		fprintf(stderr, "tevent_req_poll failed: %s\n",
+			strerror(ret));
+		TALLOC_FREE(ev);
+		return ret;
+	}
+
+	TALLOC_FREE(req2);
+
+	ret = tevent_loop_wait(ev);
+	if (ret != 0) {
+		fprintf(stderr, "tevent_loop_wait failed\n");
+		return ret;
+	}
+
+	TALLOC_FREE(pool);
+	TALLOC_FREE(ev);
+	return 0;
+}
+
+int main(void)
+{
+	int ret;
+
+	ret = test_tevent_1();
+	if (ret != 0) {
+		fprintf(stderr, "test_event_1 failed: %s\n",
+			strerror(ret));
+		return 1;
+	}
+
+	ret = test_init();
+	if (ret != 0) {
+		fprintf(stderr, "test_init failed\n");
+		return 1;
+	}
+
+	ret = test_fork();
+	if (ret != 0) {
+		fprintf(stderr, "test_fork failed\n");
+		return 1;
+	}
+
+	ret = test_jobs(10, 10000);
+	if (ret != 0) {
+		fprintf(stderr, "test_jobs failed\n");
+		return 1;
+	}
+
+	ret = test_busydestroy();
+	if (ret != 0) {
+		fprintf(stderr, "test_busydestroy failed\n");
+		return 1;
+	}
+
+	ret = test_busyfork();
+	if (ret != 0) {
+		fprintf(stderr, "test_busyfork failed\n");
+		return 1;
+	}
+
+	ret = test_busyfork2();
+	if (ret != 0) {
+		fprintf(stderr, "test_busyfork2 failed\n");
+		return 1;
+	}
+
+	printf("success\n");
+	return 0;
+}