milkway: add mw-main-context, steal it from glib again

6 files changed, 1808 insertions, 8 deletions

diff --git a/milkway/Makefile.am b/milkway/Makefile.am
index 6130ad4..ce75f78 100644
--- a/milkway/Makefile.am
+++ b/milkway/Makefile.am
@@ -30,7 +30,8 @@ milkway_headers = \
 	mw-slist.h \
 	mw-source.h \
 	mw-poll.h \
-	mw-poll-imp.h
+	mw-poll-imp.h \
+	mw-main-context.h
 
 libmilkway_la_LDFLAGS = -version-info  $(LT_VERSION_INFO) -no-undefined
 libmilkway_la_LIBADD = @DEP_LIBS@ $(PTHREAD_LIBS)
@@ -75,7 +76,9 @@ libmilkway_la_SOURCES = \
 	mw-poll-select.c \
 	mw-poll-win32-private.h \
 	mw-poll-win32.c \
-	mw-poll-imp.c
+	mw-poll-imp.c \
+	mw-main-context-private.h \
+	mw-main-context.c
 
 milkwayincludedir = $(includedir)/milkway/milkway
 milkwayinclude_HEADERS = $(milkway_headers)
diff --git a/milkway/mw-main-context-private.h b/milkway/mw-main-context-private.h
new file mode 100644
index 0000000..ca02f4a
--- /dev/null
+++ b/milkway/mw-main-context-private.h
@@ -0,0 +1,50 @@
+/* Milkway
+ *
+ * Copyright (C) 2008- Luo Jinghua <sunmoon1997@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+#ifndef MW_MAIN_CONTEXT_PRIVATE_H
+#define MW_MAIN_CONTEXT_PRIVATE_H
+
+mw_public void
+mw_main_context_lock(mw_main_context_t *self);
+
+mw_public void
+mw_main_context_unlock(mw_main_context_t *self);
+
+mw_private void
+mw_main_context_add_poll_unlocked (mw_main_context_t *context,
+				   int                priority,
+				   mw_pollfd_t       *fd);
+
+mw_private void
+mw_main_context_remove_poll_unlocked (mw_main_context_t *context,
+				      mw_pollfd_t       *fd);
+
+mw_private void
+mw_main_context_wakeup_unlocked (mw_main_context_t *context);
+
+mw_private void
+mw_main_context_remove_source (mw_main_context_t *context,
+			       mw_source_t       *source);
+
+mw_private void
+mw_main_context_set_source_priority (mw_main_context_t *context,
+				     mw_source_t       *source,
+				     int                priority);
+
+#endif
diff --git a/milkway/mw-main-context.c b/milkway/mw-main-context.c
new file mode 100644
index 0000000..d0c31bb
--- /dev/null
+++ b/milkway/mw-main-context.c
@@ -0,0 +1,1427 @@
+/* Milkway
+ *
+ * Copyright (C) 2008- Luo Jinghua <sunmoon1997@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+/* GLIB - Library of useful routines for C programming
+ * Copyright (C) 1995-1997  Peter Mattis, Spencer Kimball and Josh MacDonald
+ *
+ * gmain.c: Main loop abstraction, timeouts, and idle functions
+ * Copyright 1998 Owen Taylor
+ *
+ * This library 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 of the License, or (at your option) any later version.
+ *
+ * This library 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 this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+/*
+ * Modified by the GLib Team and others 1997-2000.  See the AUTHORS
+ * file for a list of people on the GLib Team.  See the ChangeLog
+ * files for a list of changes.  These files are distributed with
+ * GLib at ftp://ftp.gtk.org/pub/gtk/.
+ */
+
+/*
+ * MT safe
+ */
+
+#include "milkwayint.h"
+#include "milkway/mw-poll-imp.h"
+#include "milkway/mw-main-context.h"
+#include "milkway/mw-main-context-private.h"
+#include "milkway/mw-ptr-array.h"
+
+/* Uncomment the next line (and the corresponding line in gpoll.c) to
+ * enable debugging printouts if the environment variable
+ * MW_MAIN_POLL_DEBUG is set to some value.
+ */
+/* #define MW_MAIN_POLL_DEBUG */
+
+#ifdef _WIN32
+/* Always enable debugging printout on Windows, as it is more often
+ * needed there...
+ */
+#define MW_MAIN_POLL_DEBUG
+#endif
+
+#include <sys/types.h>
+#include <time.h>
+#include <stdlib.h>
+#include <stdio.h>
+
+#ifdef HAVE_SYS_TIME_H
+#include <sys/time.h>
+#endif /* HAVE_SYS_TIME_H */
+#ifdef HAVE_UNISTD_H
+#include <unistd.h>
+#endif /* HAVE_UNISTD_H */
+#include <errno.h>
+
+#ifdef MW_OS_WINDOWS
+#define STRICT
+#include <windows.h>
+#endif /* MW_OS_WIN32 */
+
+#ifndef MW_OS_WINDOWs
+#include <fcntl.h>
+#include <sys/wait.h>
+#endif
+
+#ifndef MIN
+#define MIN(a, b) (a) > (b) ? (b) : (a)
+#endif
+
+typedef struct _mw_poll_rec mw_poll_rec_t;
+
+struct _mw_main_context_type {
+    mw_object_type_t base;
+};
+
+typedef struct _mw_main_waiter mw_main_waiter_t;
+struct _mw_main_waiter
+{
+    mw_cond_t *cond;
+    mw_mutex_t *mutex;
+};
+
+
+#ifdef MW_MAIN_POLL_DEBUG
+mw_bool_t _mw_main_poll_debug = MW_FALSE;
+#endif
+
+struct _mw_main_context {
+    mw_object_t base;
+
+    /* The following lock is used for both the list of sources
+     * and the list of poll records
+     */
+    mw_static_mutex_t mutex;
+    mw_cond_t *cond;
+    mw_thread_t *owner;
+    mw_uint_t owner_count;
+    mw_slist_t *waiters;
+
+    mw_ptr_array_t *pending_dispatches;
+    int timeout;			/* Timeout for current iteration */
+
+    mw_uint_t next_id;
+    mw_source_t *source_list;
+    int in_check_or_prepare;
+
+    mw_poll_rec_t *poll_records;
+    mw_uint_t n_poll_records;
+    mw_pollfd_t *cached_poll_array;
+    mw_uint_t cached_poll_array_size;
+
+#ifndef MW_OS_WINDOWS
+    /* this pipe is used to wake up the main loop when a source is added.
+     */
+    int wake_up_pipe[2];
+#else /* MW_OS_WIN32 */
+    HANDLE wake_up_semaphore;
+#endif /* MW_OS_WIN32 */
+
+    mw_pollfd_t wake_up_rec;
+    mw_bool_t poll_waiting;
+
+    /* Flag indicating whether the set of fd's changed during a poll */
+    mw_bool_t poll_changed;
+
+    mw_timeval_t current_time;
+    mw_bool_t time_is_current;
+
+    mw_poll_t *poll;
+};
+
+struct _mw_poll_rec
+{
+    mw_pollfd_t *fd;
+    mw_poll_rec_t *next;
+    int prio;
+};
+
+#define LOCK_CONTEXT(context)        mw_static_mutex_lock (&context->mutex)
+#define UNLOCK_CONTEXT(context)      mw_static_mutex_unlock (&context->mutex)
+#define MW_THREAD_SELF               mw_thread_self ()
+
+/* Forward declarations */
+static void mw_main_context_poll                 (mw_main_context_t *context,
+						  int                timeout,
+						  int                priority,
+						  mw_pollfd_t       *fds,
+						  int                n_fds);
+
+static mw_inline void
+poll_rec_list_free (mw_main_context_t *context,
+		    mw_poll_rec_t     *list)
+{
+  mw_poll_rec_t *last = list;
+  mw_poll_rec_t *next;
+
+  if (!list)
+    return;
+
+  for (; last; last = next) {
+      next = last->next;
+      free (last);
+  }
+}
+
+static void
+mw_main_context_finalize (mw_object_t *super)
+{
+    mw_main_context_t *context = (mw_main_context_t*)super;
+    mw_source_t *source;
+
+    source = context->source_list;
+    while (source) {
+	mw_source_t *next = source->next;
+	mw_source_destroy (source);
+	source = next;
+    }
+
+    mw_static_mutex_free (&context->mutex);
+
+    mw_object_unref (context->pending_dispatches);
+    free (context->cached_poll_array);
+
+    poll_rec_list_free (context, context->poll_records);
+
+#ifndef MW_OS_WINDOWS
+    close (context->wake_up_pipe[0]);
+    close (context->wake_up_pipe[1]);
+#else
+    CloseHandle (context->wake_up_semaphore);
+#endif
+
+    if (context->cond != NULL)
+	mw_cond_free (context->cond);
+
+    MW_SUPER_FINALIZE(super, MW_MAIN_CONTEXT_TYPE)(super);
+}
+
+static void
+mw_main_context_init_pipe (mw_main_context_t *context)
+{
+# ifndef MW_OS_WINDOWS
+    if (context->wake_up_pipe[0] != -1)
+	return;
+
+    if (pipe (context->wake_up_pipe) < 0)
+	fprintf (stderr, "Cannot create pipe main loop wake-up: %s\n",
+		 strerror (errno));
+
+#ifdef FD_CLOEXEC
+    fcntl (context->wake_up_pipe[0], F_SETFD, FD_CLOEXEC);
+    fcntl (context->wake_up_pipe[1], F_SETFD, FD_CLOEXEC);
+#endif
+
+    context->wake_up_rec.fd = context->wake_up_pipe[0];
+    context->wake_up_rec.events = MW_IO_IN;
+# else
+    if (context->wake_up_semaphore != NULL)
+	return;
+    context->wake_up_semaphore = CreateSemaphore (NULL, 0, 100, NULL);
+    if (context->wake_up_semaphore == NULL)
+	fprintf (stderr, "Cannot create wake-up semaphore\n");
+    context->wake_up_rec.fd = (intptr) context->wake_up_semaphore;
+    context->wake_up_rec.events = MW_IO_IN;
+
+    if (_mw_main_poll_debug)
+	printf ("wake-up semaphore: %p\n", context->wake_up_semaphore);
+# endif
+    mw_main_context_add_poll_unlocked (context, 0, &context->wake_up_rec);
+}
+
+static mw_main_context_t *
+mw_main_context_init (mw_main_context_t *context)
+{
+    if (!mw_object_init(&context->base))
+	return NULL;
+
+    mw_static_mutex_init (&context->mutex);
+
+    context->owner = NULL;
+    context->waiters = NULL;
+
+# ifndef MW_OS_WINDOWS
+    context->wake_up_pipe[0] = -1;
+    context->wake_up_pipe[1] = -1;
+# else
+    context->wake_up_semaphore = NULL;
+# endif
+
+    context->next_id = 1;
+
+    context->source_list = NULL;
+
+    context->cached_poll_array = NULL;
+    context->cached_poll_array_size = 0;
+
+    context->pending_dispatches = mw_ptr_array_new ();
+
+    context->time_is_current = MW_FALSE;
+
+    mw_main_context_init_pipe (context);
+
+    return context;
+}
+
+/**
+ * mw_main_context_new:
+ *
+ * Creates a new #mw_main_context_t structure.
+ *
+ * Return value: the new #mw_main_context_t
+ **/
+mw_main_context_t *
+mw_main_context_new (void)
+{
+    mw_main_context_t *context;
+
+    context = (mw_main_context_t*)mw_object_alloc(MW_MAIN_CONTEXT_TYPE);
+    if (!context)
+	return NULL;
+
+    return mw_main_context_init(context);
+}
+
+void
+mw_main_context_lock(mw_main_context_t *self)
+{
+    LOCK_CONTEXT(self);
+}
+
+mw_public void
+mw_main_context_unlock(mw_main_context_t *self)
+{
+    UNLOCK_CONTEXT(self);
+}
+
+/* Holds context's lock
+ */
+static void
+mw_source_list_add (mw_source_t       *source,
+		    mw_main_context_t *context)
+{
+    mw_source_t *tmp_source, *last_source;
+
+    last_source = NULL;
+    tmp_source = context->source_list;
+    while (tmp_source && tmp_source->prio <= source->prio) {
+	last_source = tmp_source;
+	tmp_source = tmp_source->next;
+    }
+
+    source->next = tmp_source;
+    if (tmp_source)
+	tmp_source->prev = source;
+
+    source->prev = last_source;
+    if (last_source)
+	last_source->next = source;
+    else
+	context->source_list = source;
+}
+
+/* Holds context's lock
+ */
+static void
+mw_source_list_remove (mw_source_t       *source,
+		       mw_main_context_t *context)
+{
+    if (source->prev)
+	source->prev->next = source->next;
+    else
+	context->source_list = source->next;
+
+    if (source->next)
+	source->next->prev = source->prev;
+
+    source->prev = NULL;
+    source->next = NULL;
+}
+
+/**
+ * mw_source_add_source:
+ * @context: a #mw_main_context_t (if %NULL, the default context will be used)
+ * @source: a #mw_source_t
+ *
+ * Adds a #mw_source_t to a @context so that it will be executed within
+ * that context. Remove it by calling mw_source_destroy().
+ *
+ * Return value: the ID (greater than 0) for the source within the
+ *   #mw_main_context_t.
+ **/
+mw_uint_t
+mw_main_context_add_source (mw_main_context_t *context,
+			    mw_source_t       *source)
+{
+    mw_uint_t result = 0;
+    mw_slist_t *tmp_list;
+
+
+    LOCK_CONTEXT (context);
+
+    source->context = context;
+    result = source->id = context->next_id++;
+
+    mw_object_ref (source);
+    mw_source_list_add (source, context);
+
+    tmp_list = source->fds;
+    while (tmp_list) {
+	mw_main_context_add_poll_unlocked (context, source->prio, tmp_list->data);
+	tmp_list = tmp_list->next;
+    }
+
+    /* Now wake up the main loop if it is waiting in the poll() */
+    mw_main_context_wakeup_unlocked (context);
+
+    UNLOCK_CONTEXT (context);
+
+    return result;
+}
+
+void
+mw_main_context_set_source_priority (mw_main_context_t *context,
+				     mw_source_t       *source,
+				     int                priority)
+{
+    mw_slist_t *tmp_list;
+
+    mw_assert (context && context == source->context);
+
+    LOCK_CONTEXT (context);
+
+    /* Remove the source from the context's source and then
+     * add it back so it is sorted in the correct plcae
+     */
+    mw_source_list_remove (source, source->context);
+    mw_source_list_add (source, source->context);
+
+    if (!mw_source_is_blocked (source)) {
+	tmp_list = source->fds;
+	while (tmp_list) {
+	    mw_main_context_remove_poll_unlocked (context, tmp_list->data);
+	    mw_main_context_add_poll_unlocked (context, priority, tmp_list->data);
+
+	    tmp_list = tmp_list->next;
+	}
+    }
+
+    UNLOCK_CONTEXT (context);
+}
+
+/* HOLDS: context's lock */
+void
+mw_main_context_remove_source (mw_main_context_t *context,
+			       mw_source_t       *source)
+{
+
+
+    if (!mw_source_is_blocked (source)) {
+	mw_slist_t *tmp_list;
+
+	tmp_list = source->fds;
+	while (tmp_list) {
+	    mw_main_context_remove_poll_unlocked (context, tmp_list->data);
+	    tmp_list = tmp_list->next;
+	}
+    }
+
+    mw_source_list_remove (source, context);
+}
+
+/* Temporarily remove all this source's file descriptors from the
+ * poll(), so that if data comes available for one of the file descriptors
+ * we don't continually spin in the poll()
+ */
+/* HOLDS: source->context's lock */
+static void
+block_source (mw_source_t *source)
+{
+    mw_slist_t *tmp_list;
+
+    mw_assert (!mw_source_is_blocked (source));
+
+    tmp_list = source->fds;
+    while (tmp_list) {
+	mw_main_context_remove_poll_unlocked (source->context, tmp_list->data);
+	tmp_list = tmp_list->next;
+    }
+}
+
+/* HOLDS: source->context's lock */
+static void
+unblock_source (mw_source_t *source)
+{
+    mw_slist_t *tmp_list;
+
+    mw_assert (!mw_source_is_blocked (source)); /* Source already unblocked */
+    mw_assert (mw_source_is_active (source));
+
+    tmp_list = source->fds;
+    while (tmp_list) {
+	mw_main_context_add_poll_unlocked (source->context, source->prio, tmp_list->data);
+	tmp_list = tmp_list->next;
+    }
+}
+
+/* HOLDS: context's lock */
+static void
+mw_main_dispatch (mw_main_context_t *context)
+{
+    mw_uint_t i;
+
+    for (i = 0; i < context->pending_dispatches->len; i++) {
+	mw_source_t *source = context->pending_dispatches->pdata[i];
+
+	context->pending_dispatches->pdata[i] = NULL;
+	mw_assert (source);
+
+	source->flags &= ~MW_SOURCE_READY;
+
+	if (mw_source_is_active (source)) {
+	    mw_bool_t was_in_call;
+	    mw_bool_t need_destroy;
+
+	    mw_object_ref(source);
+
+	    if ((source->flags & MW_SOURCE_CAN_RECURSE) == 0)
+		block_source (source);
+
+	    was_in_call = source->flags & MW_SOURCE_DISPATCHING;
+	    source->flags |= MW_SOURCE_DISPATCHING;
+
+	    UNLOCK_CONTEXT (context);
+
+	    need_destroy = !mw_source_dispatch(source);
+
+	    mw_object_unref(source);
+
+	    LOCK_CONTEXT (context);
+
+	    if (!was_in_call)
+		source->flags &= ~MW_SOURCE_DISPATCHING;
+
+	    if ((source->flags & MW_SOURCE_CAN_RECURSE) == 0 &&
+		mw_source_is_active (source))
+		unblock_source (source);
+
+	    /* Note: this depends on the fact that we can't switch
+	     * sources from one main context to another
+	     */
+	    if (need_destroy && mw_source_is_active (source)) {
+		mw_assert (source->context == context);
+		UNLOCK_CONTEXT(context);
+		mw_source_destroy (source);
+		LOCK_CONTEXT(context);
+	    }
+	}
+
+	mw_object_unref(source);
+    }
+
+    mw_ptr_array_set_size (context->pending_dispatches, 0);
+}
+
+/* Holds context's lock */
+static inline mw_source_t *
+next_valid_source (mw_main_context_t *context,
+		   mw_source_t      *source)
+{
+    mw_source_t *new_source = source ? source->next : context->source_list;
+
+    while (new_source) {
+	if (!mw_source_is_active (new_source)) {
+	    mw_object_ref (new_source);
+	    break;
+	}
+
+	new_source = new_source->next;
+    }
+
+    if (source)
+	mw_object_unref (source);
+
+    return new_source;
+}
+
+static mw_inline mw_bool_t
+mw_main_context_is_internal_mutex(mw_main_context_t *context,
+				  mw_mutex_t        *mutex)
+{
+    return mutex == mw_static_mutex_get_mutex (&context->mutex);
+}
+
+/**
+ * mw_main_context_acquire:
+ * @context: a #mw_main_context_t
+ *
+ * Tries to become the owner of the specified context.
+ * If some other thread is the owner of the context,
+ * returns %MW_FALSE immediately. Ownership is properly
+ * recursive: the owner can require ownership again
+ * and will release ownership when mw_main_context_release()
+ * is called as many times as mw_main_context_acquire().
+ *
+ * You must be the owner of a context before you
+ * can call mw_main_context_prepare(), mw_main_context_query(),
+ * mw_main_context_check(), mw_main_context_dispatch().
+ *
+ * Return value: %MW_TRUE if the operation succeeded, and
+ *   this thread is now the owner of @context.
+ **/
+mw_bool_t
+mw_main_context_acquire (mw_main_context_t *context)
+{
+    mw_bool_t result = MW_FALSE;
+    mw_thread_t *self = MW_THREAD_SELF;
+
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+
+    if (!context->owner) {
+	context->owner = self;
+	mw_assert (context->owner_count == 0);
+    }
+
+    if (context->owner == self) {
+	context->owner_count++;
+	result = MW_TRUE;
+    }
+
+    UNLOCK_CONTEXT (context);
+
+    return result;
+}
+
+/**
+ * mw_main_context_release:
+ * @context: a #mw_main_context_t
+ *
+ * Releases ownership of a context previously acquired by this thread
+ * with mw_main_context_acquire(). If the context was acquired multiple
+ * times, the ownership will be released only when mw_main_context_release()
+ * is called as many times as it was acquired.
+ **/
+void
+mw_main_context_release (mw_main_context_t *context)
+{
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+
+    context->owner_count--;
+    if (context->owner_count == 0) {
+	context->owner = NULL;
+
+	if (context->waiters) {
+	    mw_main_waiter_t *waiter = context->waiters->data;
+	    mw_bool_t loop_internal_waiter =
+		mw_main_context_is_internal_mutex(context, waiter->mutex);
+	    context->waiters = mw_slist_delete_link (context->waiters,
+						     context->waiters);
+	    if (!loop_internal_waiter)
+		mw_mutex_lock (waiter->mutex);
+
+	    mw_cond_signal (waiter->cond);
+
+	    if (!loop_internal_waiter)
+		mw_mutex_unlock (waiter->mutex);
+	}
+    }
+
+    UNLOCK_CONTEXT (context);
+}
+
+/**
+ * mw_main_context_wait:
+ * @context: a #mw_main_context_t
+ * @cond: a condition variable
+ * @mutex: a mutex, currently held
+ *
+ * Tries to become the owner of the specified context,
+ * as with mw_main_context_acquire(). But if another thread
+ * is the owner, atomically drop @mutex and wait on @cond until
+ * that owner releases ownership or until @cond is signaled, then
+ * try again (once) to become the owner.
+ *
+ * Return value: %MW_TRUE if the operation succeeded, and
+ *   this thread is now the owner of @context.
+ **/
+mw_bool_t
+mw_main_context_wait (mw_main_context_t *context,
+		      mw_cond_t         *cond,
+		      mw_mutex_t        *mutex)
+{
+    mw_bool_t result = MW_FALSE;
+    mw_thread_t *self = MW_THREAD_SELF;
+    mw_bool_t loop_internal_waiter;
+
+    mw_assert (context != NULL);
+
+    loop_internal_waiter = mw_main_context_is_internal_mutex(context, mutex);
+
+    if (!loop_internal_waiter)
+	LOCK_CONTEXT (context);
+
+    if (context->owner && context->owner != self) {
+	mw_main_waiter_t waiter;
+
+	waiter.cond = cond;
+	waiter.mutex = mutex;
+
+	context->waiters = mw_slist_append (context->waiters, &waiter);
+
+	if (!loop_internal_waiter)
+	    UNLOCK_CONTEXT (context);
+	mw_cond_wait (cond, mutex);
+	if (!loop_internal_waiter)
+	    LOCK_CONTEXT (context);
+
+	context->waiters = mw_slist_remove (context->waiters, &waiter);
+    }
+
+    if (!context->owner) {
+	context->owner = self;
+	mw_assert (context->owner_count == 0);
+    }
+
+    if (context->owner == self) {
+	context->owner_count++;
+	result = MW_TRUE;
+    }
+
+    if (!loop_internal_waiter)
+	UNLOCK_CONTEXT (context);
+
+    return result;
+}
+
+/**
+ * mw_main_context_prepare:
+ * @context: a #mw_main_context_t
+ * @priority: location to store priority of highest priority
+ *            source already ready.
+ *
+ * Prepares to poll sources within a main loop. The resulting information
+ * for polling is determined by calling mw_main_context_query ().
+ *
+ * Return value: %MW_TRUE if some source is ready to be dispatched
+ *               prior to polling.
+ **/
+mw_bool_t
+mw_main_context_prepare (mw_main_context_t *context,
+			 int               *priority)
+{
+    mw_uint_t i;
+    int n_ready = 0;
+    int current_priority = INT_MAX;
+    mw_source_t *source;
+
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+
+    context->time_is_current = MW_FALSE;
+
+    if (context->in_check_or_prepare) {
+	fprintf (stderr, "mw_main_context_prepare() called recursively from within a source's check() or "
+		 "prepare() member.");
+	UNLOCK_CONTEXT (context);
+	return MW_FALSE;
+    }
+
+    if (context->poll_waiting) {
+	fprintf(stderr, "mw_main_context_prepare(): main loop already active in another thread");
+	UNLOCK_CONTEXT (context);
+	return MW_FALSE;
+    }
+
+    context->poll_waiting = MW_TRUE;
+
+    for (i = 0; i < context->pending_dispatches->len; i++) {
+	if (context->pending_dispatches->pdata[i])
+	    mw_object_unref ((mw_source_t *)context->pending_dispatches->pdata[i]);
+    }
+    mw_ptr_array_set_size (context->pending_dispatches, 0);
+
+    /* Prepare all sources */
+    context->timeout = -1;
+
+    source = next_valid_source (context, NULL);
+    while (source) {
+	int source_timeout = -1;
+
+	if ((n_ready > 0) && (source->prio > current_priority)) {
+	    mw_object_unref (source);
+	    break;
+	}
+	if (mw_source_is_blocked (source))
+	    goto next;
+
+	if (!mw_source_is_ready(source)) {
+	    mw_bool_t result;
+
+	    context->in_check_or_prepare++;
+	    UNLOCK_CONTEXT (context);
+
+	    result = mw_source_prepare (source, &source_timeout);
+
+	    LOCK_CONTEXT (context);
+	    context->in_check_or_prepare--;
+
+	    if (result)
+		mw_source_set_ready (source);
+	}
+
+	if (mw_source_is_ready (source)) {
+	    n_ready++;
+	    current_priority = source->prio;
+	    context->timeout = 0;
+	}
+
+	if (source_timeout >= 0) {
+	    if (context->timeout < 0)
+		context->timeout = source_timeout;
+	    else
+		context->timeout = MIN (context->timeout, source_timeout);
+	}
+
+      next:
+	source = next_valid_source (context, source);
+    }
+
+    UNLOCK_CONTEXT (context);
+
+    if (priority)
+	*priority = current_priority;
+
+    return n_ready > 0;
+}
+
+/**
+ * mw_main_context_query:
+ * @context: a #mw_main_context_t
+ * @max_priority: maximum priority source to check
+ * @timeout_: location to store timeout to be used in polling
+ * @fds: location to store #mw_pollfd_t records that need to be polled.
+ * @n_fds: length of @fds.
+ *
+ * Determines information necessary to poll this main loop.
+ *
+ * Return value: the number of records actually stored in @fds,
+ *   or, if more than @n_fds records need to be stored, the number
+ *   of records that need to be stored.
+ **/
+int
+mw_main_context_query (mw_main_context_t *context,
+		       int                max_priority,
+		       int               *timeout,
+		       mw_pollfd_t       *fds,
+		       int                n_fds)
+{
+    int n_poll;
+    mw_poll_rec_t *pollrec;
+
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+
+    pollrec = context->poll_records;
+    n_poll = 0;
+    while (pollrec && max_priority >= pollrec->prio) {
+	/* We need to include entries with fd->events == 0 in the array because
+	 * otherwise if the application changes fd->events behind our back and
+	 * makes it non-zero, we'll be out of sync when we check the fds[] array.
+	 * (Changing fd->events after adding an FD wasn't an anticipated use of
+	 * this API, but it occurs in practice.) */
+	if (n_poll < n_fds) {
+	    fds[n_poll].fd = pollrec->fd->fd;
+	    /* In direct contradiction to the Unix98 spec, IRIX runs into
+	     * difficulty if you pass in POLLERR, POLLHUP or POLLNVAL
+	     * flags in the events field of the pollfd while it should
+	     * just ignoring them. So we mask them out here.
+	     */
+	    fds[n_poll].events = pollrec->fd->events & ~(MW_IO_ERR | MW_IO_HUP | MW_IO_NVAL);
+	    fds[n_poll].revents = 0;
+	}
+
+	pollrec = pollrec->next;
+	n_poll++;
+    }
+
+    context->poll_changed = MW_FALSE;
+    if (timeout) {
+	*timeout = context->timeout;
+	if (*timeout != 0)
+	    context->time_is_current = MW_FALSE;
+    }
+
+    UNLOCK_CONTEXT (context);
+
+    return n_poll;
+}
+
+/**
+ * mw_main_context_check:
+ * @context: a #mw_main_context_t
+ * @max_priority: the maximum numerical priority of sources to check
+ * @fds: array of #mw_pollfd_t's that was passed to the last call to
+ *       mw_main_context_query()
+ * @n_fds: return value of mw_main_context_query()
+ *
+ * Passes the results of polling back to the main loop.
+ *
+ * Return value: %MW_TRUE if some sources are ready to be dispatched.
+ **/
+mw_bool_t
+mw_main_context_check (mw_main_context_t *context,
+		       int                max_priority,
+		       mw_pollfd_t       *fds,
+		       int                n_fds)
+{
+    mw_source_t *source;
+    mw_poll_rec_t *pollrec;
+    int n_ready = 0;
+    int i;
+
+    LOCK_CONTEXT (context);
+
+    if (context->in_check_or_prepare) {
+	fprintf (stderr, "mw_main_context_check() called recursively from within a source's check() or "
+		 "prepare() member.");
+	UNLOCK_CONTEXT (context);
+	return MW_FALSE;
+    }
+
+    if (!context->poll_waiting) {
+#ifndef MW_OS_WIN32
+	char a;
+	read (context->wake_up_pipe[0], &a, 1);
+#endif
+    } else {
+	context->poll_waiting = MW_FALSE;
+    }
+
+    /* If the set of poll file descriptors changed, bail out
+     * and let the main loop rerun
+     */
+    if (context->poll_changed) {
+	UNLOCK_CONTEXT (context);
+	return MW_FALSE;
+    }
+
+    pollrec = context->poll_records;
+    i = 0;
+    while (i < n_fds) {
+	if (pollrec->fd->events)
+	    pollrec->fd->revents = fds[i].revents;
+
+	pollrec = pollrec->next;
+	i++;
+    }
+
+    source = next_valid_source (context, NULL);
+    while (source) {
+	if ((n_ready > 0) && (source->prio > max_priority)) {
+	    mw_object_unref (source);
+	    break;
+	}
+	if (mw_source_is_blocked (source))
+	    goto next;
+
+	if (!mw_source_is_ready (source)) {
+	    mw_bool_t result;
+
+	    context->in_check_or_prepare++;
+	    UNLOCK_CONTEXT (context);
+
+	    result = mw_source_check (source);
+
+	    LOCK_CONTEXT (context);
+	    context->in_check_or_prepare--;
+
+	    if (result)
+		mw_source_set_ready (source);
+	}
+
+	if (mw_source_is_ready (source)) {
+	    mw_ptr_array_add (context->pending_dispatches, mw_object_ref(source));
+
+	    n_ready++;
+
+	    /* never dispatch sources with less priority than the first
+	     * one we choose to dispatch
+	     */
+	    max_priority = source->prio;
+	}
+
+      next:
+	source = next_valid_source (context, source);
+    }
+
+    UNLOCK_CONTEXT (context);
+
+    return n_ready > 0;
+}
+
+/**
+ * mw_main_context_dispatch:
+ * @context: a #mw_main_context_t
+ *
+ * Dispatches all pending sources.
+ **/
+void
+mw_main_context_dispatch (mw_main_context_t *context)
+{
+    LOCK_CONTEXT (context);
+
+    if (context->pending_dispatches->len > 0)
+	mw_main_dispatch (context);
+
+    UNLOCK_CONTEXT (context);
+}
+
+/* HOLDS context lock */
+static mw_bool_t
+mw_main_context_iterate (mw_main_context_t *context,
+			 mw_bool_t      block,
+			 mw_bool_t      dispatch,
+			 mw_thread_t      *self)
+{
+    int max_priority;
+    int timeout;
+    mw_bool_t some_ready;
+    int nfds, allocated_nfds;
+    mw_pollfd_t *fds = NULL;
+
+    UNLOCK_CONTEXT (context);
+
+    if (!mw_main_context_acquire (context)) {
+	mw_bool_t got_ownership;
+
+	LOCK_CONTEXT (context);
+
+	if (!block)
+	    return MW_FALSE;
+
+	if (!context->cond)
+	    context->cond = mw_cond_new ();
+
+	got_ownership = mw_main_context_wait (context,
+					      context->cond,
+					      mw_static_mutex_get_mutex (&context->mutex));
+
+	if (!got_ownership)
+	    return MW_FALSE;
+    } else {
+	LOCK_CONTEXT (context);
+    }
+
+    if (!context->cached_poll_array) {
+	context->cached_poll_array_size = context->n_poll_records;
+	context->cached_poll_array = calloc (sizeof (mw_pollfd_t), context->n_poll_records);
+    }
+
+    allocated_nfds = context->cached_poll_array_size;
+    fds = context->cached_poll_array;
+
+    UNLOCK_CONTEXT (context);
+
+    mw_main_context_prepare (context, &max_priority);
+
+    while ((nfds = mw_main_context_query (context, max_priority, &timeout, fds,
+					  allocated_nfds)) > allocated_nfds) {
+	LOCK_CONTEXT (context);
+	free (fds);
+	context->cached_poll_array_size = allocated_nfds = nfds;
+	context->cached_poll_array = fds = calloc (sizeof(mw_pollfd_t), nfds);
+	UNLOCK_CONTEXT (context);
+    }
+
+    if (!block)
+	timeout = 0;
+
+    mw_main_context_poll (context, timeout, max_priority, fds, nfds);
+
+    some_ready = mw_main_context_check (context, max_priority, fds, nfds);
+
+    if (dispatch)
+	mw_main_context_dispatch (context);
+
+    mw_main_context_release (context);
+
+    LOCK_CONTEXT (context);
+
+    return some_ready;
+}
+
+/**
+ * mw_main_context_pending:
+ * @context: a #mw_main_context_t (if %NULL, the default context will be used)
+ *
+ * Checks if any sources have pending events for the given context.
+ *
+ * Return value: %MW_TRUE if events are pending.
+ **/
+mw_bool_t
+mw_main_context_pending (mw_main_context_t *context)
+{
+    mw_bool_t retval;
+
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+    retval = mw_main_context_iterate (context, MW_FALSE, MW_FALSE, MW_THREAD_SELF);
+    UNLOCK_CONTEXT (context);
+
+    return retval;
+}
+
+/**
+ * mw_main_context_iteration:
+ * @context: a #mw_main_context_t (if %NULL, the default context will be used)
+ * @may_block: whether the call may block.
+ *
+ * Runs a single iteration for the given main loop. This involves
+ * checking to see if any event sources are ready to be processed,
+ * then if no events sources are ready and @may_block is %MW_TRUE, waiting
+ * for a source to become ready, then dispatching the highest priority
+ * events sources that are ready. Otherwise, if @may_block is %MW_FALSE
+ * sources are not waited to become ready, only those highest priority
+ * events sources will be dispatched (if any), that are ready at this
+ * given moment without further waiting.
+ *
+ * Note that even when @may_block is %MW_TRUE, it is still possible for
+ * mw_main_context_iteration() to return %MW_FALSE, since the the wait may
+ * be interrupted for other reasons than an event source becoming ready.
+ *
+ * Return value: %MW_TRUE if events were dispatched.
+ **/
+mw_bool_t
+mw_main_context_iteration (mw_main_context_t *context, mw_bool_t may_block)
+{
+    mw_bool_t retval;
+
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+    retval = mw_main_context_iterate (context, may_block, MW_TRUE, MW_THREAD_SELF);
+    UNLOCK_CONTEXT (context);
+
+    return retval;
+}
+
+/* HOLDS: context's lock */
+static void
+mw_main_context_poll (mw_main_context_t *context,
+		      int                timeout,
+		      int                priority,
+		      mw_pollfd_t       *fds,
+		      int                n_fds)
+{
+#ifdef  MW_MAIN_POLL_DEBUG
+    mw_poll_rec_t *pollrec;
+    int i;
+#endif
+
+    mw_poll_t *poll;
+
+
+    if (n_fds || timeout != 0) {
+#ifdef	MW_MAIN_POLL_DEBUG
+	if (_mw_main_poll_debug) {
+	    mw_print ("polling context=%p n=%d timeout=%d\n",
+		      context, n_fds, timeout);
+	    poll_timer = mw_timer_new ();
+	}
+#endif
+
+	LOCK_CONTEXT (context);
+
+	poll = mw_object_ref (context->poll);
+
+	UNLOCK_CONTEXT (context);
+	if (mw_poll_poll (poll, fds, n_fds, timeout) < 0 && errno != EINTR) {
+#ifndef MW_OS_WINDOWS
+	    fprintf (stderr, "poll(2) failed due to: %s.",
+		     strerror (errno));
+#else
+	    /* If mw_poll () returns -1, it has already called mw_warning() */
+#endif
+	}
+	mw_object_unref (poll);
+
+#ifdef	MW_MAIN_POLL_DEBUG
+	if (_mw_main_poll_debug) {
+	    LOCK_CONTEXT (context);
+
+	    pollrec = context->poll_records;
+	    while (pollrec != NULL) {
+		i = 0;
+		while (i < n_fds) {
+		    if (fds[i].fd == pollrec->fd->fd &&
+			pollrec->fd->events &&
+			fds[i].revents) {
+			printf (" [ 0x%ld :", (long)fds[i].fd);
+			if (fds[i].revents & MW_IO_IN)
+			    print ("i");
+			if (fds[i].revents & MW_IO_OUT)
+			    printf ("o");
+			if (fds[i].revents & MW_IO_PRI)
+			    printf ("p");
+			if (fds[i].revents & MW_IO_ERR)
+			    printf ("e");
+			if (fds[i].revents & MW_IO_HUP)
+			    printf ("h");
+			if (fds[i].revents & MW_IO_NVAL)
+			    printf ("n");
+			printf ("]");
+		    }
+		    i++;
+		}
+		pollrec = pollrec->next;
+	    }
+	    printf ("\n");
+
+	    UNLOCK_CONTEXT (context);
+	}
+#endif
+    } /* if (n_fds || timeout != 0) */
+}
+
+/**
+ * mw_main_context_add_poll:
+ * @context: a #mw_main_context_t (or %NULL for the default context)
+ * @fd: a #mw_pollfd_t structure holding information about a file
+ *      descriptor to watch.
+ * @priority: the priority for this file descriptor which should be
+ *      the same as the priority used for mw_source_attach() to ensure that the
+ *      file descriptor is polled whenever the results may be needed.
+ *
+ * Adds a file descriptor to the set of file descriptors polled for
+ * this context. This will very seldomly be used directly. Instead
+ * a typical event source will use mw_source_add_poll() instead.
+ **/
+void
+mw_main_context_add_poll (mw_main_context_t *context,
+			  mw_pollfd_t       *fd,
+			  int                priority)
+{
+    mw_assert (context && fd);
+
+    LOCK_CONTEXT (context);
+    mw_main_context_add_poll_unlocked (context, priority, fd);
+    UNLOCK_CONTEXT (context);
+}
+
+/* HOLDS: main_loop_lock */
+void
+mw_main_context_add_poll_unlocked (mw_main_context_t *context,
+				   int          priority,
+				   mw_pollfd_t      *fd)
+{
+    mw_poll_rec_t *lastrec, *pollrec;
+    mw_poll_rec_t *newrec = malloc (sizeof(mw_poll_rec_t));
+
+    /* This file descriptor may be checked before we ever poll */
+    fd->revents = 0;
+    newrec->fd = fd;
+    newrec->prio = priority;
+
+    lastrec = NULL;
+    pollrec = context->poll_records;
+    while (pollrec && priority >= pollrec->prio) {
+	lastrec = pollrec;
+	pollrec = pollrec->next;
+    }
+
+    if (lastrec)
+	lastrec->next = newrec;
+    else
+	context->poll_records = newrec;
+
+    newrec->next = pollrec;
+
+    context->n_poll_records++;
+
+    context->poll_changed = MW_TRUE;
+
+    /* Now wake up the main loop if it is waiting in the poll() */
+    mw_main_context_wakeup_unlocked (context);
+}
+
+/**
+ * mw_main_context_remove_poll:
+ * @context:a #mw_main_context_t
+ * @fd: a #mw_pollfd_t descriptor previously added with mw_main_context_add_poll()
+ *
+ * Removes file descriptor from the set of file descriptors to be
+ * polled for a particular context.
+ **/
+void
+mw_main_context_remove_poll (mw_main_context_t *context,
+			     mw_pollfd_t       *fd)
+{
+    mw_assert (context && fd);
+
+    LOCK_CONTEXT (context);
+    mw_main_context_remove_poll_unlocked (context, fd);
+    UNLOCK_CONTEXT (context);
+}
+
+void
+mw_main_context_remove_poll_unlocked (mw_main_context_t *context,
+				      mw_pollfd_t       *fd)
+{
+    mw_poll_rec_t *pollrec, *lastrec;
+
+    lastrec = NULL;
+    pollrec = context->poll_records;
+
+    while (pollrec) {
+	if (pollrec->fd == fd) {
+	    if (lastrec != NULL)
+		lastrec->next = pollrec->next;
+	    else
+		context->poll_records = pollrec->next;
+
+	    free (pollrec);
+
+	    context->n_poll_records--;
+	    break;
+	}
+	lastrec = pollrec;
+	pollrec = pollrec->next;
+    }
+
+    context->poll_changed = MW_TRUE;
+
+    /* Now wake up the main loop if it is waiting in the poll() */
+    mw_main_context_wakeup_unlocked (context);
+}
+
+/**
+ * mw_main_context_get_current_time:
+ * @context:  a #mw_main_context_t
+ * @timeval: #mw_timeval_t structure in which to store current time.
+ *
+ * Gets the "current time" to be used when checking
+ * this source. The advantage of calling this function over
+ * calling mw_get_system_time() directly is that when
+ * checking multiple sources, Milkway can cache a single value
+ * instead of having to repeatedly get the system time.
+ **/
+void
+mw_main_context_get_current_time (mw_main_context_t *context,
+				  mw_timeval_t      *timeval)
+{
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+
+    if (!context->time_is_current) {
+	mw_get_system_time (&context->current_time);
+	context->time_is_current = MW_TRUE;
+    }
+
+    *timeval = context->current_time;
+
+    UNLOCK_CONTEXT (context);
+}
+
+/* HOLDS: context's lock */
+/* Wake the main loop up from a poll() */
+void
+mw_main_context_wakeup_unlocked (mw_main_context_t *context)
+{
+    if (context->poll_waiting) {
+	context->poll_waiting = MW_FALSE;
+#ifndef MW_OS_WINDOWNS
+	write (context->wake_up_pipe[1], "A", 1);
+#else
+	ReleaseSemaphore (context->wake_up_semaphore, 1, NULL);
+#endif
+    }
+}
+
+/**
+ * mw_main_context_wakeup:
+ * @context: a #mw_main_context_t
+ *
+ * If @context is currently waiting in a poll(), interrupt
+ * the poll(), and continue the iteration process.
+ **/
+void
+mw_main_context_wakeup (mw_main_context_t *context)
+{
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+    mw_main_context_wakeup_unlocked (context);
+    UNLOCK_CONTEXT (context);
+}
+
+/**
+ * mw_main_context_is_owner:
+ * @context: a #mw_main_context_t
+ *
+ * Determines whether this thread holds the (recursive)
+ * ownership of this #GMaincontext. This is useful to
+ * know before waiting on another thread that may be
+ * blocking to get ownership of @context.
+ *
+ * Returns: %MW_TRUE if current thread is owner of @context.
+ *
+ * Since: 2.10
+ **/
+mw_bool_t
+mw_main_context_is_owner (mw_main_context_t *context)
+{
+    mw_bool_t is_owner;
+
+    mw_assert (context != NULL);
+
+    LOCK_CONTEXT (context);
+    is_owner = context->owner == MW_THREAD_SELF;
+    UNLOCK_CONTEXT (context);
+    return is_owner;
+}
+
+static void
+mw_main_context_type_init(mw_main_context_type_t *self)
+{
+    self->base.finalize = mw_main_context_finalize;
+}
+
+MW_DEFINE_GET_TYPE(mw_main_context, mw_main_context_type_t,
+		   MW_OBJECT_TYPE, "MWMainContext", 0);
diff --git a/milkway/mw-main-context.h b/milkway/mw-main-context.h
new file mode 100644
index 0000000..c355947
--- /dev/null
+++ b/milkway/mw-main-context.h
@@ -0,0 +1,125 @@
+/* Milkway
+ *
+ * Copyright (C) 2008- Luo Jinghua <sunmoon1997@gmail.com>
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+/* gmain.h - the GLib Main loop
+ * Copyright (C) 1998-2000 Red Hat, Inc.
+ *
+ * This library is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Library General Public
+ * License as published by the Free Software Foundation; either
+ * version 2 of the License, or (at your option) any later version.
+ *
+ * This library 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
+ * Library General Public License for more details.
+ *
+ * You should have received a copy of the GNU Library General Public
+ * License along with this library; if not, write to the
+ * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+ * Boston, MA 02111-1307, USA.
+ */
+
+#ifndef __MW_MAIN_H__
+#define __MW_MAIN_H__
+
+#include <milkway/mw-poll.h>
+#include <milkway/mw-source.h>
+
+MW_BEGIN_DECLS
+
+typedef struct _mw_main_context_type    mw_main_context_type_t;
+typedef struct _mw_main_context	        mw_main_context_t;	/* Opaque */
+
+#define MW_MAIN_CONTEXT_TYPE mw_main_context_get_type()
+
+mw_public mw_main_context_type_t*
+mw_main_context_get_type(void);
+
+mw_public mw_main_context_t*
+mw_main_context_new       (void);
+
+mw_public mw_bool_t
+mw_main_context_iteration (mw_main_context_t *context,
+			   mw_bool_t          may_block);
+
+mw_public mw_bool_t
+mw_main_context_pending   (mw_main_context_t *context);
+
+/* Low level functions for implementing custom main loops.
+ */
+mw_public void
+mw_main_context_wakeup  (mw_main_context_t *context);
+
+mw_public mw_bool_t
+mw_main_context_acquire (mw_main_context_t *context);
+
+mw_public void
+mw_main_context_release (mw_main_context_t *context);
+
+mw_public mw_bool_t
+mw_main_context_is_owner (mw_main_context_t *context);
+
+mw_public mw_bool_t
+mw_main_context_wait    (mw_main_context_t *context,
+			 mw_cond_t         *cond,
+			 mw_mutex_t        *mutex);
+
+mw_public mw_bool_t
+mw_main_context_prepare  (mw_main_context_t *context,
+			  int               *priority);
+
+mw_public int
+mw_main_context_query    (mw_main_context_t *context,
+			  int                max_priority,
+			  int               *timeout_,
+			  mw_pollfd_t       *fds,
+			  int                n_fds);
+
+mw_public int
+mw_main_context_check    (mw_main_context_t *context,
+			  int                max_priority,
+			  mw_pollfd_t       *fds,
+			  int                n_fds);
+
+mw_public void
+mw_main_context_dispatch (mw_main_context_t *context);
+
+mw_public mw_uint_t
+mw_main_context_add_source (mw_main_context_t *context,
+			    mw_source_t       *source);
+
+/* Low level functions for use by source implementations
+ */
+mw_public void
+mw_main_context_add_poll    (mw_main_context_t *context,
+			     mw_pollfd_t       *fd,
+			     int                priority);
+
+mw_public void
+mw_main_context_remove_poll (mw_main_context_t *context,
+			     mw_pollfd_t       *fd);
+
+mw_public void
+mw_main_context_get_current_time (mw_main_context_t *context,
+				  mw_timeval_t      *timeval);
+
+MW_END_DECLS
+
+#endif /* __MW_MAIN_H__ */
diff --git a/milkway/mw-source.c b/milkway/mw-source.c
index 70895fc..f2a581b 100644
--- a/milkway/mw-source.c
+++ b/milkway/mw-source.c
@@ -19,6 +19,8 @@
  */
 #include "milkwayint.h"
 #include "milkway/mw-source.h"
+#include "milkway/mw-main-context.h"
+#include "milkway/mw-main-context-private.h"
 
 mw_source_t*
 mw_source_init(mw_source_t *self)
@@ -27,6 +29,12 @@ mw_source_init(mw_source_t *self)
 	return NULL;
 
     self->prio = 0;
+    self->id = 0;
+    self->flags = MW_SOURCE_ACTIVE;
+    self->prev = NULL;
+    self->next = NULL;
+    self->fds = NULL;
+    self->context = NULL;
     return self;
 }
 
@@ -35,14 +43,120 @@ mw_source_set_priority(mw_source_t *self,
 		       int prio)
 {
     self->prio = prio;
+    if (self->context)
+	mw_main_context_set_source_priority (self->context, self, prio);
+}
+
+/**
+ * mw_source_add_poll:
+ * @source:a #mw_source_t
+ * @fd: a #mw_pollfd_t structure holding information about a file
+ *      descriptor to watch.
+ *
+ * Adds a file descriptor to the set of file descriptors polled for
+ * this source. This is usually combined with mw_source_new() to add an
+ * event source. The event source's check function will typically test
+ * the @revents field in the #mw_pollfd_t struct and return %MW_TRUE if events need
+ * to be processed.
+ **/
+void
+mw_source_add_poll (mw_source_t *source,
+		    mw_pollfd_t *fd)
+{
+    mw_main_context_t *context;
+
+    context = source->context;
+    if (context)
+	mw_main_context_lock (context);
+
+    source->fds = mw_slist_prepend (source->fds, fd);
+
+    if (context) {
+	if (!mw_source_is_blocked (source))
+	    mw_main_context_add_poll_unlocked (context, source->prio, fd);
+	mw_main_context_unlock (context);
+    }
+}
+
+/**
+ * mw_source_remove_poll:
+ * @source:a #mw_source_t
+ * @fd: a #mw_pollfd_t structure previously passed to mw_source_add_poll().
+ *
+ * Removes a file descriptor from the set of file descriptors polled for
+ * this source.
+ **/
+void
+mw_source_remove_poll (mw_source_t *source,
+		       mw_pollfd_t *fd)
+{
+    mw_main_context_t *context;
+
+    context = source->context;
+
+    if (context)
+	mw_main_context_lock (context);
+
+    source->fds = mw_slist_remove (source->fds, fd);
+
+    if (context) {
+	if (!mw_source_is_blocked (source))
+	    mw_main_context_remove_poll_unlocked (context, fd);
+	mw_main_context_unlock (context);
+    }
+}
+
+void
+mw_source_get_current_time (mw_source_t  *source,
+			    mw_timeval_t *timeval)
+{
+    mw_main_context_t *context = source->context;
+
+    if (!context)
+	mw_get_system_time(timeval);
+    else
+	mw_main_context_get_current_time(context, timeval);
+}
+
+/**
+ * mw_source_destroy:
+ * @source: a #mw_source_t
+ *
+ * Removes a source from its #mw_main_context_t, if any, and mark it as
+ * destroyed.  The source cannot be subsequently added to another
+ * context.
+ **/
+void
+mw_source_destroy(mw_source_t *source)
+{
+    mw_main_context_t *context = source->context;
+
+    if (context)
+	mw_main_context_lock(context);
+
+    if (mw_source_is_active (source)) {
+	mw_assert (context != NULL);
+
+	source->flags &= ~MW_SOURCE_ACTIVE;
+
+	if (context)
+	    mw_main_context_remove_source (context, source);
+	source->context = NULL;
+
+	mw_object_unref (source);
+    }
+
+    if (context)
+	mw_main_context_unlock (context);
 }
 
 static void
-mw_source_finalize(mw_object_t *self)
+mw_source_finalize(mw_object_t *super)
 {
-    mw_object_type_t *stype = MW_SUPER_TYPE_BASE(self, MW_SOURCE_TYPE);
+    mw_source_t *self = (mw_source_t*)super;
 
-    stype->finalize(self);
+    mw_assert (self->context == NULL);
+    MW_SUPER_FINALIZE(super, MW_SOURCE_TYPE)(super);
 }
 
 static void
diff --git a/milkway/mw-source.h b/milkway/mw-source.h
index a273526..dcd6fec 100644
--- a/milkway/mw-source.h
+++ b/milkway/mw-source.h
@@ -20,11 +20,14 @@
 #ifndef MW_SOURCE_H
 #define MW_SOURCE_H
 
-#include <milkway/mw-object.h>
+#include <milkway/mw-poll.h>
+#include <milkway/mw-slist.h>
 
 typedef struct mw_source_type mw_source_type_t;
 typedef struct mw_source mw_source_t;
+
 #define MW_SOURCE_TYPE mw_source_get_type()
+
 struct mw_source_type {
     mw_object_type_t base;
 
@@ -34,11 +37,26 @@ struct mw_source_type {
     mw_bool_t (*dispatch) (mw_source_t *self);
 };
 
+typedef enum {
+    MW_SOURCE_ACTIVE      = 1 << 0,
+    MW_SOURCE_BLOCKED     = 1 << 1,
+    MW_SOURCE_READY       = 1 << 2,
+    MW_SOURCE_CAN_RECURSE = 1 << 3,
+    MW_SOURCE_DISPATCHING = 1 << 4
+} mw_source_flags_t;
+
 struct mw_source {
     mw_object_t base;
 
     /** priority */
     int prio;
+    mw_source_flags_t flags;
+    mw_slist_t *fds;
+    void       *context;
+    mw_uint_t  id;
+
+    mw_source_t *next;
+    mw_source_t *prev;
 };
 
 #define MW_PRIORITY_HIGH            -100
@@ -63,8 +81,7 @@ mw_source_prepare(mw_source_t *self,
 }
 
 static mw_inline mw_bool_t
-mw_source_check(mw_source_t *self,
-		  int timeout_)
+mw_source_check(mw_source_t *self)
 {
     mw_source_type_t *type = (mw_source_type_t *)MW_TYPE(self);
 
@@ -79,8 +96,72 @@ mw_source_dispatch(mw_source_t *self)
     return type->dispatch(self);
 }
 
+static mw_inline mw_bool_t
+mw_source_is_active(mw_source_t *self)
+{
+    return self->flags & MW_SOURCE_ACTIVE;
+}
+
+static mw_inline mw_bool_t
+mw_source_is_dispatching(mw_source_t *self)
+{
+    return self->flags & MW_SOURCE_DISPATCHING;
+}
+
+static mw_inline mw_bool_t
+mw_source_get_can_recurse(mw_source_t *self)
+{
+    return self->flags & MW_SOURCE_CAN_RECURSE;
+}
+
+static mw_inline void
+mw_source_set_can_recurse(mw_source_t *self)
+{
+    self->flags |= MW_SOURCE_CAN_RECURSE;
+}
+
+static mw_inline mw_bool_t
+mw_source_is_blocked(mw_source_t *self)
+{
+    return self->flags & MW_SOURCE_DISPATCHING &&
+	!(self->flags & MW_SOURCE_CAN_RECURSE);
+}
+
+static mw_inline mw_bool_t
+mw_source_is_ready(mw_source_t *self)
+{
+    return self->flags & MW_SOURCE_READY;
+}
+
+static mw_inline void
+mw_source_set_ready(mw_source_t *self)
+{
+    self->flags |= MW_SOURCE_READY;
+}
+
+static mw_inline mw_uint_t
+mw_source_get_id(mw_source_t *self)
+{
+    return self->id;
+}
+
 mw_public void
 mw_source_set_priority(mw_source_t *self,
 		       int          prio);
 
+mw_public void
+mw_source_add_poll (mw_source_t *source,
+		    mw_pollfd_t *fd);
+
+mw_public void
+mw_source_remove_poll (mw_source_t *source,
+		       mw_pollfd_t *fd);
+
+mw_public void
+mw_source_get_current_time (mw_source_t *source,
+			    mw_timeval_t *timeval);
+
+mw_public void
+mw_source_destroy(mw_source_t *self);
+
 #endif