add new AvahiThreadedPoll event loop implementation which runs a main loop in a helper thread

git-svn-id: file:///home/lennart/svn/public/avahi/trunk@1074 941a03a8-eaeb-0310-b9a0-b1bbd8fe43fe

5 files changed, 311 insertions, 7 deletions

diff --git a/Makefile.am b/Makefile.am
index f9e95a1..f338c58 100644
--- a/Makefile.am
+++ b/Makefile.am
@@ -88,6 +88,7 @@ DX_INPUT = \
 	$(srcdir)/avahi-common/domain.h \
 	$(srcdir)/avahi-common/watch.h \
 	$(srcdir)/avahi-common/simple-watch.h \
+	$(srcdir)/avahi-common/thread-watch.h \
 	$(srcdir)/avahi-glib/glib-watch.h \
 	$(srcdir)/avahi-glib/glib-malloc.h \
 	$(srcdir)/avahi-common/timeval.h \
diff --git a/avahi-common/Makefile.am b/avahi-common/Makefile.am
index f9fdef4..2a3a591 100644
--- a/avahi-common/Makefile.am
+++ b/avahi-common/Makefile.am
@@ -46,7 +46,8 @@ noinst_PROGRAMS = \
 	domain-test \
 	alternative-test \
 	timeval-test \
-	watch-test
+	watch-test \
+	watch-test-thread
 endif
 
 lib_LTLIBRARIES = \
@@ -61,6 +62,7 @@ libavahi_common_la_SOURCES = \
 	domain.c domain.h \
 	timeval.c timeval.h \
 	simple-watch.c simple-watch.h \
+	thread-watch.c thread-watch.h \
 	watch.h gccmacro.h \
 	rlist.h rlist.c
 
@@ -95,6 +97,10 @@ watch_test_SOURCES = \
 watch_test_CFLAGS = $(AM_CFLAGS) $(PTHREAD_CFLAGS)
 watch_test_LDADD = $(AM_LDADD) $(PTHREAD_LIBS) $(PTHREAD_CFLAGS)
 
+watch_test_thread_SOURCES = $(watch_test_SOURCES) thread-watch.c thread-watch.h
+watch_test_thread_CFLAGS = $(watch_test_CFLAGS) -DUSE_THREAD
+watch_test_thread_LDADD = $(watch_test_LDADD)
+
 timeval_test_SOURCES = \
 	timeval.c timeval.h \
 	timeval-test.c
diff --git a/avahi-common/thread-watch.c b/avahi-common/thread-watch.c
new file mode 100644
index 0000000..27a0835
--- /dev/null
+++ b/avahi-common/thread-watch.c
@@ -0,0 +1,186 @@
+/* $Id$ */
+
+/***
+  This file is part of avahi.
+ 
+  avahi is free software; you can redistribute it and/or modify it
+  under the terms of the GNU Lesser General Public License as
+  published by the Free Software Foundation; either version 2.1 of the
+  License, or (at your option) any later version.
+ 
+  avahi is distributed in the hope that it will be useful, but WITHOUT
+  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+  or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
+  Public License for more details.
+ 
+  You should have received a copy of the GNU Lesser General Public
+  License along with avahi; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+  USA.
+***/
+
+#ifdef HAVE_CONFIG_H
+#include <config.h>
+#endif
+
+#include <sys/poll.h>
+#include <assert.h>
+#include <string.h>
+#include <errno.h>
+#include <unistd.h>
+#include <fcntl.h>
+#include <stdio.h>
+#include <pthread.h>
+#include <signal.h>
+
+#include "llist.h"
+#include "malloc.h"
+#include "timeval.h"
+#include "simple-watch.h"
+#include "thread-watch.h"
+
+struct AvahiThreadedPoll {
+    AvahiSimplePoll *simple_poll;
+    pthread_t thread_id;
+    pthread_mutex_t mutex;
+    int thread_running;
+    int retval;
+};
+
+static int poll_func(struct pollfd *ufds, unsigned int nfds, int timeout, void *userdata) {
+    pthread_mutex_t *mutex = userdata;
+    int r;
+
+    /* Before entering poll() we unlock the mutex, so that
+     * avahi_simple_poll_quit() can succeed from another thread. */
+    
+    pthread_mutex_unlock(mutex);
+    r = poll(ufds, nfds, timeout);
+    pthread_mutex_lock(mutex);
+
+    return r;
+}
+
+static void* thread(void *userdata){
+    AvahiThreadedPoll *p = userdata;
+    sigset_t mask;
+
+    /* Make sure that signals are delivered to the main thread */
+    sigfillset(&mask);
+    pthread_sigmask(SIG_BLOCK, &mask, NULL);
+    
+    pthread_mutex_lock(&p->mutex);
+    p->retval = avahi_simple_poll_loop(p->simple_poll);
+    pthread_mutex_unlock(&p->mutex);
+    
+    return NULL;
+}
+
+AvahiThreadedPoll *avahi_threaded_poll_new(void) {
+    AvahiThreadedPoll *p;
+
+    if (!(p = avahi_new(AvahiThreadedPoll, 1)))
+        goto fail; /* OOM */
+    
+    if (!(p->simple_poll = avahi_simple_poll_new()))
+        goto fail;
+        
+    pthread_mutex_init(&p->mutex, NULL);
+
+    avahi_simple_poll_set_func(p->simple_poll, poll_func, &p->mutex);
+    
+    p->thread_running = 0;
+
+    return p;
+
+fail:
+    if (p) {
+        if (p->simple_poll) {
+            avahi_simple_poll_free(p->simple_poll);
+            pthread_mutex_destroy(&p->mutex);
+        }
+
+        avahi_free(p);
+    }
+
+    return NULL;
+}
+
+void avahi_threaded_poll_free(AvahiThreadedPoll *p) {
+    assert(p);
+
+    /* Make sure that this function is not called from the helper thread */
+    assert(!p->thread_running || !pthread_equal(pthread_self(), p->thread_id));
+
+    if (p->thread_running)
+        avahi_threaded_poll_quit(p);
+
+    if (p->simple_poll)
+        avahi_simple_poll_free(p->simple_poll);
+
+    pthread_mutex_destroy(&p->mutex);
+    avahi_free(p);
+}
+
+const AvahiPoll* avahi_threaded_poll_get(AvahiThreadedPoll *p) {
+    assert(p);
+
+    return avahi_simple_poll_get(p->simple_poll);
+}
+
+int avahi_threaded_poll_start(AvahiThreadedPoll *p) {
+    assert(p);
+
+    assert(!p->thread_running);
+
+    if (pthread_create(&p->thread_id, NULL, thread, p) < 0)
+        return -1;
+
+    return 0;
+}
+
+int avahi_threaded_poll_stop(AvahiThreadedPoll *p) {
+    assert(p);
+
+    if (!p->thread_running)
+        return -1;
+
+    /* Make sure that this function is not called from the helper thread */
+    assert(!pthread_equal(pthread_self(), p->thread_id));
+
+    pthread_mutex_lock(&p->mutex);
+    avahi_simple_poll_quit(p->simple_poll);
+    pthread_mutex_unlock(&p->mutex);
+    
+    pthread_join(p->thread_id, NULL);
+    p->thread_running = 0;
+
+    return p->retval;
+}
+
+void avahi_threaded_poll_quit(AvahiThreadedPoll *p) {
+    assert(p);
+
+    /* Make sure that this function is called from the helper thread */
+    assert(pthread_equal(pthread_self(), p->thread_id));
+
+    avahi_simple_poll_quit(p->simple_poll);
+}
+
+void avahi_threaded_poll_lock(AvahiThreadedPoll *p) {
+    assert(p);
+
+    /* Make sure that this function is not called from the helper thread */
+    assert(!p->thread_running || !pthread_equal(pthread_self(), p->thread_id));
+    
+    pthread_mutex_lock(&p->mutex);
+}
+
+void avahi_threaded_poll_unlock(AvahiThreadedPoll *p) {
+    assert(p);
+
+    /* Make sure that this function is not called from the helper thread */
+    assert(!p->thread_running || !pthread_equal(pthread_self(), p->thread_id));
+    
+    pthread_mutex_unlock(&p->mutex);
+}
diff --git a/avahi-common/thread-watch.h b/avahi-common/thread-watch.h
new file mode 100644
index 0000000..924dc04
--- /dev/null
+++ b/avahi-common/thread-watch.h
@@ -0,0 +1,82 @@
+#ifndef foothreadedwatchhfoo
+#define foothreadedwatchhfoo
+
+/* $Id$ */
+
+/***
+  This file is part of avahi.
+ 
+  avahi is free software; you can redistribute it and/or modify it
+  under the terms of the GNU Lesser General Public License as
+  published by the Free Software Foundation; either version 2.1 of the
+  License, or (at your option) any later version.
+ 
+  avahi is distributed in the hope that it will be useful, but WITHOUT
+  ANY WARRANTY; without even the implied warranty of MERCHANTABILITY
+  or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General
+  Public License for more details.
+ 
+  You should have received a copy of the GNU Lesser General Public
+  License along with avahi; if not, write to the Free Software
+  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307
+  USA.
+***/
+
+/** \file threaded-watch.h Threaded poll() based main loop implementation */
+
+#include <sys/poll.h>
+#include <avahi-common/cdecl.h>
+#include <avahi-common/watch.h>
+
+AVAHI_C_DECL_BEGIN
+
+/** A main loop object that runs an AvahiSimplePoll in its own thread. */
+typedef struct AvahiThreadedPoll AvahiThreadedPoll;
+
+/** Create a new event loop object. This will allocate the internal
+ * AvahiSimplePoll, but will not start the helper thread. */
+AvahiThreadedPoll *avahi_threaded_poll_new(void);
+
+/** Free an event loop object. Ths will stop the associated evet loop
+ * thread (if it is running). */
+void avahi_threaded_poll_free(AvahiThreadedPoll *p);
+
+/** Return the abstracted poll API object for this event loop
+ * object. The will return the same pointer each time it is
+ * called. */
+const AvahiPoll* avahi_threaded_poll_get(AvahiThreadedPoll *p);
+
+/** Start the event loop helper thread. After the thread has startet
+ * you must make sure to access the event loop object
+ * (AvahiThreadedPoll, AvahiPoll and all its associated objects)
+ * synchronized, i.e. with proper locking. You may want to use
+ * avahi_threaded_poll_lock()/avahi_threaded_poll_unlock() for this,
+ * which will lock the the entire event loop. Please note that event
+ * loop callback functions are called from the event loop helper thread
+ * with that lock held, i.e. avahi_threaded_poll_lock() calls are not
+ * required from event callbacks. */
+int avahi_threaded_poll_start(AvahiThreadedPoll *p);
+
+/** Request that the event loop quits and the associated thread
+ stops. Call this from outside the helper thread if you want to shut
+ it down. */
+int avahi_threaded_poll_stop(AvahiThreadedPoll *p);
+
+/** Request that the event loop quits and the associated thread
+ stops. Call this from inside the helper thread if you want to shut it
+ down.*/
+void avahi_threaded_poll_quit(AvahiThreadedPoll *p);
+
+/** Lock the main loop object. Use this if you want to access the event
+ * loop objects (such as creating a new event source) from anything
+ * else but the event loop helper thread, i.e. from anything else but event
+ * loop callbacks */
+void avahi_threaded_poll_lock(AvahiThreadedPoll *p);
+
+/** Unlock the event loop object, use this as counterpart to
+ * avahi_threaded_poll_lock() */
+void avahi_threaded_poll_unlock(AvahiThreadedPoll *p);
+
+AVAHI_C_DECL_END
+
+#endif
diff --git a/avahi-common/watch-test.c b/avahi-common/watch-test.c
index cd70117..870045d 100644
--- a/avahi-common/watch-test.c
+++ b/avahi-common/watch-test.c
@@ -30,12 +30,18 @@
 #include <string.h>
 
 #include "watch.h"
-#include "simple-watch.h"
 #include "timeval.h"
 #include "gccmacro.h"
 
 static const AvahiPoll *api = NULL;
+
+#ifndef USE_THREAD
+#include "simple-watch.h"
 static AvahiSimplePoll *simple_poll = NULL;
+#else
+#include "thread-watch.h"
+static AvahiThreadedPoll *threaded_poll = NULL;
+#endif
 
 static void callback(AvahiWatch *w, int fd, AvahiWatchEvent event, AVAHI_GCC_UNUSED void *userdata) {
 
@@ -59,9 +65,13 @@ static void wakeup(AvahiTimeout *t, AVAHI_GCC_UNUSED void *userdata) {
 
     printf("Wakeup #%i\n", i++);
 
-    if (i > 10)
+    if (i > 10) {
+#ifndef USE_THREAD
         avahi_simple_poll_quit(simple_poll);
-    else {
+#else
+        avahi_threaded_poll_quit(threaded_poll);
+#endif
+    } else {
         avahi_elapse_time(&tv, 1000, 0);
         api->timeout_update(t, &tv);
     }
@@ -69,20 +79,39 @@ static void wakeup(AvahiTimeout *t, AVAHI_GCC_UNUSED void *userdata) {
 
 int main(AVAHI_GCC_UNUSED int argc, AVAHI_GCC_UNUSED char *argv[]) {
     struct timeval tv;
-    
-    simple_poll = avahi_simple_poll_new();
 
+#ifndef USE_THREAD
+    simple_poll = avahi_simple_poll_new();
+    assert(simple_poll);
     api = avahi_simple_poll_get(simple_poll);
+    assert(api);
+#else
+    threaded_poll = avahi_threaded_poll_new();
+    assert(threaded_poll);
+    api = avahi_threaded_poll_get(threaded_poll);
+    assert(api);
+#endif
+    
     api->watch_new(api, 0, AVAHI_WATCH_IN, callback, NULL);
 
     avahi_elapse_time(&tv, 1000, 0);
     api->timeout_new(api, &tv, wakeup, NULL);
 
+#ifndef USE_THREAD
     /* Our main loop */
     avahi_simple_poll_loop(simple_poll);
+    avahi_simple_poll_free(simple_poll);
 
+#else
+    avahi_threaded_poll_start(threaded_poll);
 
-    avahi_simple_poll_free(simple_poll);
+    fprintf(stderr, "Now doing some stupid stuff ...\n");
+    sleep(20);
+    fprintf(stderr, "... stupid stuff is done.\n");
+
+    avahi_threaded_poll_free(threaded_poll);
+    
+#endif
     
     return 0;
 }