Nicer thread alternatives.

git-svn-id: file:///svn/unbound/trunk@150 be551aaa-1e26-0410-a405-d3ace91eadb9

daemon/daemon.c

@@ -88,15 +88,7 @@ daemon_create_workers(struct daemon* daemon)
 {
 	int i;
 	log_assert(daemon && daemon->cfg);
-	/* only one thread for now */
 	daemon->num = daemon->cfg->num_threads;
-#if !defined(HAVE_PTHREAD) && !defined(HAVE_SOLARIS_THREADS)
-	if(daemon->num != 1) {
-		log_err("configed %d threads, but executable was compiled "
-		"with no thread support. Continuing with 1.", daemon->num);
-		daemon->num = 1;
-	}
-#endif /* no threads */
 	daemon->workers = (struct worker**)calloc((size_t)daemon->num, 
 		sizeof(struct worker*));
 	for(i=0; i<daemon->num; i++) {
@@ -105,6 +97,27 @@ daemon_create_workers(struct daemon* daemon)
 	}
 }
 
+/**
+ * Close all pipes except for the numbered thread.
+ * @param daemon: daemon to close pipes in.
+ * @param thr: thread number 0..num-1 of thread to skip.
+ */
+void close_other_pipes(struct daemon* daemon, int thr)
+{
+	int i;
+	for(i=0; i<daemon->num; i++)
+		if(i!=thr) {
+			if(daemon->workers[i]->cmd_send_fd != -1) {
+				close(daemon->workers[i]->cmd_send_fd);
+				daemon->workers[i]->cmd_send_fd = -1;
+			}
+			if(daemon->workers[i]->cmd_recv_fd != -1) {
+				close(daemon->workers[i]->cmd_recv_fd);
+				daemon->workers[i]->cmd_recv_fd = -1;
+			}
+		}
+}
+
 /**
  * Function to start one thread. 
  * @param arg: user argument.
@@ -116,6 +129,12 @@ thread_start(void* arg)
 	struct worker* worker = (struct worker*)arg;
 	int num = worker->thread_num;
 	ub_thread_blocksigs();
+#if !defined(HAVE_PTHREAD) && !defined(HAVE_SOLARIS_THREADS)
+	/* close pipe ends used by main */
+	close(worker->cmd_send_fd);
+	worker->cmd_send_fd = -1;
+	close_other_pipes(worker->daemon, worker->thread_num);
+#endif /* no threads */
 	if(!worker_init(worker, worker->daemon->cfg, worker->daemon->ports,
 		BUFSZ, 0))
 		fatal_exit("Could not initialize thread #%d", num);
@@ -138,6 +157,11 @@ daemon_start_others(struct daemon* daemon)
 	for(i=1; i<daemon->num; i++) {
 		ub_thread_create(&daemon->workers[i]->thr_id,
 			thread_start, daemon->workers[i]);
+#if !defined(HAVE_PTHREAD) && !defined(HAVE_SOLARIS_THREADS)
+		/* close pipe end of child */
+		close(daemon->workers[i]->cmd_recv_fd);
+		daemon->workers[i]->cmd_recv_fd = -1;
+#endif /* no threads */
 	}
 }
 
@@ -148,7 +172,7 @@ daemon_start_others(struct daemon* daemon)
 static void
 daemon_stop_others(struct daemon* daemon)
 {
-	int i, err;
+	int i;
 	log_assert(daemon);
 	log_info("stop others");
 	/* skip i=0, is this thread */
@@ -162,8 +186,7 @@ daemon_stop_others(struct daemon* daemon)
 	for(i=1; i<daemon->num; i++) {
 		/* join it to make sure its dead */
 		log_info("join %d", i);
-		if((err=pthread_join(daemon->workers[i]->thr_id, NULL)))
+		ub_thread_join(daemon->workers[i]->thr_id);
-			log_err("pthread_join: %s", strerror(err));
 		log_info("join success %d", i);
 	}
 }

doc/Changelog

@@ -1,3 +1,7 @@
+27 February 2007: Wouter
+	- ub_thread_join portable definition.
+	- forking is used if no threading is available.
+
 26 February 2007: Wouter
 	- ub_random code used to select ID and port.
 	- log code prints thread id.

util/locks.c

@@ -42,27 +42,27 @@
 #include "config.h"
 #include "util/locks.h"
 #include <signal.h>
+#include <sys/wait.h>
 
 /** block all signals, masks them away. */
 void 
 ub_thread_blocksigs()
 {
-#ifdef HAVE_PTHREAD
 	int err;
 	sigset_t sigset;
 	sigfillset(&sigset);
 	log_info("blocking signals");
+#ifdef HAVE_PTHREAD
 	if((err=pthread_sigmask(SIG_SETMASK, &sigset, NULL)))
 		fatal_exit("pthread_sigmask: %s", strerror(err));
 #else
 #  ifdef HAVE_SOLARIS_THREADS
-	int err;
-	sigset_t sigset;
-	sigfillset(&sigset);
 	if((err=thr_sigsetmask(SIG_SETMASK, &sigset, NULL)))
 		fatal_exit("thr_sigsetmask: %s", strerror(err));
 #  else 
-	/* have nothing, do nothing */
+	/* have nothing, do single process signal mask */
+	if((err=sigprocmask(SIG_SETMASK, &sigset, NULL)))
+		fatal_exit("sigprocmask: %s", strerror(errno));
 #  endif /* HAVE_SOLARIS_THREADS */
 #endif /* HAVE_PTHREAD */
 }
@@ -70,25 +70,62 @@ ub_thread_blocksigs()
 /** unblock one signal, so we can catch it. */
 void ub_thread_sig_unblock(int sig)
 {
-#ifdef HAVE_PTHREAD
 	int err;
 	sigset_t sigset;
 	sigemptyset(&sigset);
 	sigaddset(&sigset, sig);
 	log_info("unblocking signal %d", sig);
+#ifdef HAVE_PTHREAD
 	if((err=pthread_sigmask(SIG_UNBLOCK, &sigset, NULL)))
 		fatal_exit("pthread_sigmask: %s", strerror(err));
 #else
 #  ifdef HAVE_SOLARIS_THREADS
-	int err;
-	sigset_t sigset;
-	sigemptyset(&sigset);
-	sigaddset(&sigset, sig);
 	if((err=thr_sigsetmask(SIG_UNBLOCK, &sigset, NULL)))
 		fatal_exit("thr_sigsetmask: %s", strerror(err));
 #  else 
 	/* have nothing, do nothing */
+	if((err=sigprocmask(SIG_UNBLOCK, &sigset, NULL)))
+		fatal_exit("sigprocmask: %s", strerror(errno));
 #  endif /* HAVE_SOLARIS_THREADS */
 #endif /* HAVE_PTHREAD */
 }
 
+/**
+ * No threading available: fork a new process.
+ * This means no shared data structure, and no locking.
+ * Only the main thread ever returns. Exits on errors.
+ * @param thr: the location where to store the thread-id.
+ * @param func: function body of the thread. Return value of func is lost.
+ * @param arg: user argument to func.
+ */
+void 
+ub_thr_fork_create(ub_thread_t* thr, void* (*func)(void*), void* arg)
+{
+	pid_t pid = fork();
+	switch(pid) {
+	default:	/* main */
+			*thr = pid;
+			return;
+	case 0: 	/* child */
+			*thr = getpid();
+			(void)(*func)(arg);
+			exit(0);
+	case -1:	/* error */
+			fatal_exit("could not fork: %s", strerror(errno));
+	}
+}
+
+/**
+ * There is no threading. Wait for a process to terminate.
+ * Note that ub_thread_t is defined as pid_t.
+ * @param thread: the process id to wait for.
+ */
+void ub_thr_fork_wait(ub_thread_t thread)
+{
+	int status = 0;
+	if(waitpid(thread, &status, 0) == -1)
+		log_err("waitpid(%d): %s", (int)thread, strerror(errno));
+	if(status != 0)
+		log_warn("process %d abnormal exit with status %d",
+			(int)thread, status);
+}

util/locks.h

@@ -123,6 +123,8 @@ typedef pthread_t ub_thread_t;
 #define ub_thread_create(thr, func, arg) LOCKRET(pthread_create(thr, NULL, func, arg))
 /** get self id. */
 #define ub_thread_self() pthread_self()
+/** wait for another thread to terminate */
+#define ub_thread_join(thread) LOCKRET(pthread_join(thread, NULL))
 
 #else /* we do not HAVE_PTHREAD */
 #ifdef HAVE_SOLARIS_THREADS
@@ -153,6 +155,7 @@ typedef mutex_t lock_quick_t;
 typedef thread_t ub_thread_t;
 #define ub_thread_create(thr, func, arg) LOCKRET(thr_create(NULL, NULL, func, arg, NULL, thr))
 #define ub_thread_self() thr_self()
+#define ub_thread_join(thread) LOCKRET(thr_join(thread, NULL, NULL))
 
 #else /* we do not HAVE_SOLARIS_THREADS and no PTHREADS */
 
@@ -180,12 +183,14 @@ typedef int lock_quick_t;
 #define lock_quick_unlock(lock) /* nop */
 
 /** Thread creation, threads do not exist */
-typedef int ub_thread_t;
+typedef pid_t ub_thread_t;
 /** ub_thread_create gives an error, it should not be called. */
 #define ub_thread_create(thr, func, arg) \
+	ub_thr_fork_create(thr, func, arg)
 	fatal_exit("%s %d called thread create, but no thread support "  \
 		"has been compiled in.",  __FILE__, __LINE__)
-#define ub_thread_self() 0
+#define ub_thread_self() getpid()
+#define ub_thread_join(thread) ub_thr_fork_wait(thread)
 
 #endif /* HAVE_SOLARIS_THREADS */
 #endif /* HAVE_PTHREAD */