LRN: scheduler compatible select on top of g_poll for improved W32 portability

1 files changed, 559 insertions, 108 deletions

diff --git a/src/gnunet-gtk.c b/src/gnunet-gtk.c
index fc5916f0..0784432a 100644
--- a/src/gnunet-gtk.c
+++ b/src/gnunet-gtk.c
@@ -138,8 +138,551 @@ struct MainContext
    */
   gint max_priority;
 
+#if WINDOWS
+  /**
+   * Array to hold pipe handles during a select() call
+   */
+  struct GNUNET_DISK_FileHandle **read_array;
+
+  /**
+   * Allocated length of read_array
+   */
+  int read_array_length;
+
+  /**
+   * Event to fire when a socket is ready for reading
+   */
+  HANDLE hEventRead;
+
+  /**
+   * Event to fire when a socket is ready for writing
+   */
+  HANDLE hEventWrite;
+
+  /**
+   * Event to fire when a socket had an error
+   */
+  HANDLE hEventException;
+
+  /**
+   * Event that is permanently enabled and is used to signal a pipe
+   * that is ready for writing (asynchronous pipes are always writable)
+   */
+  HANDLE hEventPipeWrite;
+
+  /**
+   * Event that is permanently enabled and is used to signal a pipe
+   * that is ready for reading (used to wake up early on a pipe that
+   * is known to be readable)
+   */
+  HANDLE hEventReadReady;
+
+  /**
+   * A list to hold file handles that are ready for reading
+   */
+  struct GNUNET_CONTAINER_SList *handles_read;
+
+  /**
+   * A list to hold file handles that are ready for writing
+   */
+  struct GNUNET_CONTAINER_SList *handles_write;
+
+  /**
+   * A list to hold file handles that are broken
+   */
+  struct GNUNET_CONTAINER_SList *handles_except;
+#endif
 };
 
+static void
+gnunet_gtk_dispatch_task (void *cls,
+                         const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct MainContext *mc = (struct MainContext *) cls;
+  g_main_context_dispatch (mc->gmc);
+}
+
+#ifndef FD_COPY
+#define FD_COPY(s, d) (memcpy ((d), (s), sizeof (fd_set)))
+#endif
+
+int
+gnunet_gtk_select (void *cls,
+                   struct GNUNET_NETWORK_FDSet *rfds,
+                   struct GNUNET_NETWORK_FDSet *wfds,
+                   struct GNUNET_NETWORK_FDSet *efds,
+                   const struct GNUNET_TIME_Relative timeout)
+{
+  struct MainContext *mc = (struct MainContext *) cls;
+  int max_nfds;
+  gint poll_result;
+
+  GPollFD *gfds;
+  gint delay;
+
+  guint i;
+  guint fd_counter = 0;
+
+  guint allocated_nfds, need_gfds;
+
+  fd_set aread, awrite, aexcept;
+
+#if WINDOWS
+  int always_ready_write_fd = -1;
+  int sock_read = 0, sock_write = 0, sock_err = 0, socks = 0;
+
+  int pre_ret = 0;
+  int select_ret = 0;
+  int result = 0;
+
+  int read_handles = 0;
+  DWORD waitstatus;
+#endif
+
+  FD_ZERO (&aread);
+  FD_ZERO (&awrite);
+  FD_ZERO (&aexcept);
+  if (rfds)
+    FD_COPY (&rfds->sds, &aread);
+  if (wfds)
+    FD_COPY (&wfds->sds, &awrite);
+  if (efds)
+    FD_COPY (&efds->sds, &aexcept);
+
+#if WINDOWS
+  ResetEvent (mc->hEventRead);
+  ResetEvent (mc->hEventWrite);
+  ResetEvent (mc->hEventException);
+
+  GNUNET_CONTAINER_slist_clear (mc->handles_read);
+  GNUNET_CONTAINER_slist_clear (mc->handles_write);
+  GNUNET_CONTAINER_slist_clear (mc->handles_except);
+#endif
+
+   if (TRUE != g_main_loop_is_running (mc->gml))
+     return GNUNET_NETWORK_socket_select (rfds, wfds, efds, timeout);
+
+  if (rfds != NULL)
+    max_nfds = rfds->nsds;
+  else
+    max_nfds = -1;
+  if (wfds != NULL && max_nfds < wfds->nsds)
+    max_nfds = wfds->nsds;
+  if (efds != NULL && max_nfds < efds->nsds)
+    max_nfds = efds->nsds;
+
+  allocated_nfds = mc->cached_poll_array_size;
+  gfds = mc->cached_poll_array;
+  if (allocated_nfds == 0)
+  {
+    /* TODO: get some statistics, find the maximum number of fds ever
+     * polled during normal gnunet-gtk operation, and set this to that number.
+     */
+    mc->cached_poll_array = gfds = g_new (GPollFD, 30);
+    mc->cached_poll_array_size = allocated_nfds = 30;
+  }
+
+  while (1)
+  {
+    fd_counter = 0;
+#if !WINDOWS
+    gboolean need_realloc = FALSE;
+    for (i = 0; !need_realloc && i < max_nfds; i += 1)
+    {
+      int isset[3];
+      isset[0] = FD_ISSET (i, &rfds->sds);
+      isset[1] = FD_ISSET (i, &wfds->sds);
+      isset[2] = FD_ISSET (i, &efds->sds);
+      if (!isset[0] && !isset[1] && !isset[2])
+        continue;
+      if (fd_counter >= allocated_nfds)
+      {
+        need_realloc = TRUE;
+        break;
+      }
+      gfds[fd_counter].fd = i;
+      gfds[fd_counter].events = (isset[0] ? G_IO_IN | G_IO_HUP | G_IO_ERR : 0)
+          | (isset[1] ? G_IO_OUT | G_IO_ERR : 0) | (issed[2] ? G_IO_ERR : 0);
+      fd_counter += 1;
+    }
+    if (need_realloc)
+    {
+      mc->cached_poll_array = fds = g_renew (GPollFD, fds, mc->cached_poll_array_size * 2);
+      mc->cached_poll_array_size = allocated_nfds = nfds;
+      fd_counter = 0;
+      need_realloc = FALSE;
+    }
+    else
+      break;
+#else
+    struct GNUNET_CONTAINER_SList_Iterator *t;
+    /* We might overshoot a little, but that won't hurt very much */
+    int need_nfds = (rfds->sds.fd_count + rfds->sds.fd_count + rfds->sds.fd_count > 0 ? 3 : 0)
+        + (rfds == NULL ? 0 : GNUNET_CONTAINER_slist_count (rfds->handles))
+        + (wfds == NULL ? 0 : 1)
+        + 1;
+    if (need_nfds >= allocated_nfds)
+    {
+      /* Since there are also gmainloop's own fds, just need_nfds won't be
+       * enough, so make it twice as long.
+       */
+      mc->cached_poll_array = gfds = g_renew (GPollFD, gfds, need_nfds * 2);
+      mc->cached_poll_array_size = allocated_nfds = need_nfds * 2;
+    }
+    if (mc->read_array_length < GNUNET_CONTAINER_slist_count (rfds->handles))
+    {
+      mc->read_array = GNUNET_realloc (mc->read_array, GNUNET_CONTAINER_slist_count (rfds->handles) * sizeof (struct GNUNET_DISK_FileHandle *));
+      mc->read_array_length = GNUNET_CONTAINER_slist_count (rfds->handles);
+    }
+    if (rfds != NULL)
+    {
+      for (t = GNUNET_CONTAINER_slist_begin (rfds->handles), i = 0;
+          GNUNET_CONTAINER_slist_end (t) != GNUNET_YES;
+          GNUNET_CONTAINER_slist_next (t), i += 1)
+      {
+        struct GNUNET_DISK_FileHandle *fh;
+        fh = (struct GNUNET_DISK_FileHandle *) GNUNET_CONTAINER_slist_get (t, NULL);
+        if (fh->type == GNUNET_PIPE)
+        {
+          if (!ReadFile (fh->h, NULL, 0, NULL, fh->oOverlapRead))
+          {
+            DWORD error_code = GetLastError();
+            if (error_code == ERROR_IO_PENDING)
+            {
+#if DEBUG_NETWORK
+              GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Adding the pipe's 0x%x overlapped event to the array as %d\n", fh->h, nhandles);
+#endif
+              gfds[fd_counter].fd = (intptr_t) fh->oOverlapRead->hEvent;
+              /* On W32 .events makes no sense - g_poll will just OR its
+               * contents into .revents when the .fd event fires.
+               * So we'll use it in the way that suits us the best.
+               */
+              gfds[fd_counter].events = G_IO_IN;
+              fd_counter += 1;
+              mc->read_array[read_handles] = fh;
+              read_handles += 1;
+            }
+            else
+            {
+              gfds[fd_counter].fd = (intptr_t) mc->hEventReadReady;
+              gfds[fd_counter].events = G_IO_HUP;
+              fd_counter += 1;
+              mc->read_array[read_handles] = fh;
+              read_handles += 1;
+            }
+          }
+          else
+          {
+#if DEBUG_NETWORK
+            GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Adding the read ready event to the array as %d\n", nhandles);
+#endif
+            gfds[fd_counter].fd = (intptr_t) mc->hEventReadReady;
+            gfds[fd_counter].events = G_IO_IN;
+            fd_counter += 1;
+            mc->read_array[read_handles] = fh;
+            read_handles += 1;
+          }
+        }
+        else
+        {
+          GNUNET_CONTAINER_slist_add (mc->handles_read,
+              GNUNET_CONTAINER_SLIST_DISPOSITION_TRANSIENT,
+              fh, sizeof (struct GNUNET_DISK_FileHandle));
+            pre_ret += 1;
+        }
+      }
+    }
+    if (wfds != NULL && GNUNET_CONTAINER_slist_count (wfds->handles) > 0)
+    {
+      gfds[fd_counter].fd = (intptr_t) mc->hEventPipeWrite;
+      gfds[fd_counter].events = G_IO_OUT;
+      always_ready_write_fd = fd_counter;
+      fd_counter += 1;
+    }
+    if (efds != NULL)
+    {
+      for (t = GNUNET_CONTAINER_slist_begin (efds->handles), i = 0;
+          GNUNET_CONTAINER_slist_end (t) != GNUNET_YES;
+          GNUNET_CONTAINER_slist_next (t), i += 1)
+      {
+        struct GNUNET_DISK_FileHandle *fh;
+        DWORD dwBytes;
+        fh = (struct GNUNET_DISK_FileHandle *) GNUNET_CONTAINER_slist_get (t, NULL);
+        if (fh->type == GNUNET_PIPE)
+        {
+          if (!PeekNamedPipe (fh->h, NULL, 0, NULL, &dwBytes, NULL))
+          {
+            GNUNET_CONTAINER_slist_add (mc->handles_except,
+                GNUNET_CONTAINER_SLIST_DISPOSITION_TRANSIENT,
+                fh, sizeof (struct GNUNET_DISK_FileHandle));
+            pre_ret += 1;
+          }
+        }
+      }
+    }
+    GNUNET_CONTAINER_slist_iter_destroy (t);
+
+    if (rfds != NULL && rfds->sds.fd_count > 0)
+    {
+#if DEBUG_NETWORK
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Adding the socket read event to the array as %d\n", fd_counter);
+#endif
+      gfds[fd_counter].fd = (intptr_t) mc->hEventRead;
+      gfds[fd_counter].events = G_IO_IN;
+      for (i = 0; i < rfds->sds.fd_count; i++)
+        WSAEventSelect (rfds->sds.fd_array[i], mc->hEventRead, FD_ACCEPT | FD_READ | FD_CLOSE);
+      fd_counter += 1;
+      sock_read = rfds->sds.fd_count;
+    }
+    if (wfds != NULL && wfds->sds.fd_count > 0)
+    {
+      int wakeup = 0;
+#if DEBUG_NETWORK
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Adding the socket write event to the array as %d\n", fd_counter);
+#endif
+      gfds[fd_counter].fd = (intptr_t) mc->hEventWrite;
+      gfds[fd_counter].events = G_IO_OUT;
+      for (i = 0; i < wfds->sds.fd_count; i++)
+      {
+        DWORD error;
+        int status;
+        status = send (wfds->sds.fd_array[i], NULL, 0, 0);
+        error = GetLastError ();
+#if DEBUG_NETWORK
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "pre-send to the socket %d returned %d (%u)\n", i, status, error);
+#endif
+        if (status == 0 || (error != WSAEWOULDBLOCK && error != WSAENOTCONN))
+          wakeup = 1;
+        WSAEventSelect (wfds->sds.fd_array[i], mc->hEventWrite, FD_WRITE | FD_CONNECT | FD_CLOSE);
+      }
+      if (wakeup)
+        SetEvent (mc->hEventWrite);
+      fd_counter += 1;
+      sock_write = wfds->sds.fd_count;
+    }
+    if (efds != NULL && efds->sds.fd_count > 0)
+    {
+#if DEBUG_NETWORK
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Adding the socket error event to the array as %d\n", fd_counter);
+#endif
+      gfds[fd_counter].fd = (intptr_t) mc->hEventException;
+      gfds[fd_counter].events = G_IO_ERR;
+      for (i = 0; i < efds->sds.fd_count; i++)
+        WSAEventSelect (efds->sds.fd_array[i], mc->hEventException, FD_OOB | FD_CLOSE);
+      fd_counter += 1;
+      sock_err = efds->sds.fd_count;
+    }
+    break;
+  }
+  socks = sock_read + sock_write + sock_err;
+#endif
+
+  g_main_context_prepare (mc->gmc, &mc->max_priority);
+  while (allocated_nfds < (need_gfds = g_main_context_query (mc->gmc,
+      mc->max_priority, &delay, &gfds[fd_counter], allocated_nfds - fd_counter)))
+  {
+    mc->cached_poll_array = gfds = g_renew (GPollFD, gfds, allocated_nfds - fd_counter + need_gfds);
+    mc->cached_poll_array_size = allocated_nfds = allocated_nfds - fd_counter + need_gfds;
+  }
+  mc->poll_array_active = fd_counter + need_gfds;
+
+  if (timeout.rel_value != GNUNET_TIME_UNIT_FOREVER_REL.rel_value)
+  {
+    if (delay >= 0)
+      delay = MIN(timeout.rel_value / GNUNET_TIME_UNIT_MILLISECONDS.rel_value, delay);
+    else
+      delay = timeout.rel_value / GNUNET_TIME_UNIT_MILLISECONDS.rel_value;
+  }
+
+  if (pre_ret > 0)
+  {
+#if DEBUG_NETWORK
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "pre_ret is %d, setting delay to 0\n", pre_ret);
+#endif
+    delay = 0;
+  }
+
+#if DEBUG_NETWORK
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "We have %d of our FDs and %d of GMC ones, going to wait %6dms\n", fd_counter, need_gfds, delay);
+#endif
+
+  poll_result = g_poll (gfds, fd_counter + need_gfds, delay);
+
+#if DEBUG_NETWORK
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "g_poll returned : %d\n", poll_result);
+#endif
+
+  /* Take care of GUI events.
+   * Dispatching the events here will eventually crash the scheduler, must do this
+   * from within a task (currently we're not in a task, but in a select() call, remember)
+   * Startup reason is used to pass the scheduler sanity check.
+   */
+  if (TRUE == g_main_context_check (mc->gmc, mc->max_priority, &gfds[fd_counter], need_gfds))
+    GNUNET_SCHEDULER_add_continuation (gnunet_gtk_dispatch_task, mc, GNUNET_SCHEDULER_REASON_STARTUP);
+
+#if !WINDOWS
+  if (rfds)
+    FD_ZERO (rfds);
+  if (wfds)
+    FD_ZERO (wfds);
+  if (efds)
+    FD_ZERO (efds);
+  for (i = 0; i < fd_counter; i++)
+  {
+    int set[3];
+    if (set[0] = FD_ISSET (gfds[i].fd, &aread))
+      FD_SET (gfds[i].fd, rfds->sds);
+    if (set[1] = FD_ISSET (gfds[i].fd, &awrite))
+      FD_SET (gfds[i].fd, wfds->sds);
+    if (set[2] = FD_ISSET (gfds[i].fd, &aexcept))
+      FD_SET (gfds[i].fd, efds->sds);
+    if (set[0] || set[1] || set[2])
+      result += 1;
+  }
+#else
+  if (socks > 0)
+  {
+    struct timeval tvslice;
+    tvslice.tv_sec = 0;
+    tvslice.tv_usec = 0;
+    select_ret = select (max_nfds, &aread, &awrite, &aexcept, &tvslice);
+    if (select_ret == -1)
+      select_ret = 0;
+#if DEBUG_NETWORK
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "select() returned %d\n", select_ret);
+#endif
+  }
+  if (always_ready_write_fd >= 0 && gfds[always_ready_write_fd].revents & G_IO_OUT)
+  {
+    GNUNET_CONTAINER_slist_append (mc->handles_write, wfds->handles);
+    result += GNUNET_CONTAINER_slist_count (mc->handles_write);
+#if DEBUG_NETWORK
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Added write pipe\n");
+#endif
+  }
+  for (i = 0; i < read_handles; i++)
+  {
+    DWORD error;
+    BOOL bret;
+    if (!(gfds[i].revents & (G_IO_IN | G_IO_HUP | G_IO_ERR)))
+      continue;
+    SetLastError (0);
+    waitstatus = 0;
+    bret = PeekNamedPipe (mc->read_array[i]->h, NULL, 0, NULL, &waitstatus, NULL);
+    error = GetLastError ();
+#if DEBUG_NETWORK
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peek at read pipe %d (0x%x) returned %d (%d bytes available) GLE %u\n", i, mc->read_array[i]->h, bret, waitstatus, error);
+#endif
+    if (bret == 0 || (gfds[i].revents & G_IO_ERR))
+    {
+      if (efds != NULL)
+      {
+        struct GNUNET_CONTAINER_SList_Iterator *t;
+        for (t = GNUNET_CONTAINER_slist_begin (efds->handles), i = 0;
+            GNUNET_CONTAINER_slist_end (t) != GNUNET_YES;
+            GNUNET_CONTAINER_slist_next (t), i += 1)
+        {
+          struct GNUNET_DISK_FileHandle *fh;
+          fh = (struct GNUNET_DISK_FileHandle *) GNUNET_CONTAINER_slist_get (t, NULL);
+          if (fh == mc->read_array[i])
+          {
+            GNUNET_CONTAINER_slist_add (mc->handles_except,
+                GNUNET_CONTAINER_SLIST_DISPOSITION_TRANSIENT,
+                fh, sizeof (struct GNUNET_DISK_FileHandle));
+            break;
+          }
+        }
+        GNUNET_CONTAINER_slist_iter_destroy (t);
+      }
+    }
+    else if (waitstatus <= 0)
+      continue;
+    GNUNET_CONTAINER_slist_add (mc->handles_read,
+        GNUNET_CONTAINER_SLIST_DISPOSITION_TRANSIENT,
+        mc->read_array[i], sizeof (struct GNUNET_DISK_FileHandle));
+    result += 1;
+#if DEBUG_NETWORK
+    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Added read Pipe 0x%x (0x%x)\n", mc->read_array[i], mc->read_array[i]->h);
+#endif
+  }
+  waitstatus = WaitForSingleObject (mc->hEventWrite, 0);
+#if DEBUG_NETWORK
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Wait for the write event returned %d\n", waitstatus);
+#endif
+  if (waitstatus == WAIT_OBJECT_0)
+  {
+    for (i = 0; i < wfds->sds.fd_count; i++)
+    {
+      DWORD error;
+      int status;
+      int so_error = 0;
+      int sizeof_so_error = sizeof (so_error);
+      int gso_result = getsockopt (wfds->sds.fd_array[i], SOL_SOCKET, SO_ERROR, (char *) &so_error, &sizeof_so_error);
+      status = send (wfds->sds.fd_array[i], NULL, 0, 0);
+      error = GetLastError ();
+#if DEBUG_NETWORK
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "send to the socket %d returned %d (%u)\n", i, status, error);
+#endif
+      if (status == 0
+          || (error != WSAEWOULDBLOCK && error != WSAENOTCONN)
+          || (status == -1 && gso_result == 0 && error == WSAENOTCONN && so_error == WSAECONNREFUSED))
+      {
+        FD_SET (wfds->sds.fd_array[i], &awrite);
+        result += 1;
+      }
+    }
+  }
+  if (rfds)
+  {
+    struct GNUNET_CONTAINER_SList_Iterator *t;
+    for (i = 0; i < rfds->sds.fd_count; i++)
+      WSAEventSelect (rfds->sds.fd_array[i], mc->hEventRead, 0);
+    for (t = GNUNET_CONTAINER_slist_begin (rfds->handles);
+        GNUNET_CONTAINER_slist_end (t) != GNUNET_YES;
+        GNUNET_CONTAINER_slist_next (t))
+    {
+      struct GNUNET_DISK_FileHandle *fh;
+      fh = (struct GNUNET_DISK_FileHandle *) GNUNET_CONTAINER_slist_get (t, NULL);
+      if (fh->type == GNUNET_PIPE)
+        CancelIo (fh->h);
+    }
+    GNUNET_CONTAINER_slist_iter_destroy (t);
+    GNUNET_NETWORK_fdset_zero (rfds);
+    if (select_ret != -1 && socks > 0)
+      GNUNET_NETWORK_fdset_copy_native (rfds, &aread, select_ret);
+    GNUNET_CONTAINER_slist_append (rfds->handles, mc->handles_read);
+  }
+  if (wfds)
+  {
+    for (i = 0; i < wfds->sds.fd_count; i++)
+      WSAEventSelect (wfds->sds.fd_array[i], mc->hEventWrite, 0);
+    GNUNET_NETWORK_fdset_zero (wfds);
+    if (select_ret != -1 && socks > 0)
+      GNUNET_NETWORK_fdset_copy_native (wfds, &awrite, select_ret);
+    GNUNET_CONTAINER_slist_append (wfds->handles, mc->handles_write);
+  }
+  if (efds)
+  {
+    for (i = 0; i < efds->sds.fd_count; i++)
+      WSAEventSelect (efds->sds.fd_array[i], mc->hEventException, 0);
+    GNUNET_NETWORK_fdset_zero (efds);
+    if (select_ret != -1 && socks > 0)
+      GNUNET_NETWORK_fdset_copy_native (efds, &aexcept, select_ret);
+    GNUNET_CONTAINER_slist_append (efds->handles, mc->handles_except);
+    result += GNUNET_CONTAINER_slist_count (mc->handles_except);
+  }
+  if (fd_counter > 0)
+    /* This is not accurate (select_ret counts write-ready sockets,
+     * and result does as well. Anything out there actually cares
+     * about this?
+     */
+    return select_ret + result;
+  else
+    return 0;
+#endif
+  return result;
+}
+
 
 /**
  * Get the name of the directory where all of our package
@@ -178,112 +721,6 @@ free_context (struct MainContext *context)
   GNUNET_free (context);
 }
 
-
-/**
- * Run GTK tasks that are ready.
- *
- * @param cls the 'struct MainContext'
- * @param tc task context
- */
-static void
-run_ready (void *cls,
-           const struct GNUNET_SCHEDULER_TaskContext *tc);
-
-
-/**
- * Schedule the main GTK Event loop with the GNUnet scheduler.
- *
- * @param cls the 'struct MainContext'
- * @param tc task context
- */
-static void
-schedule_main_loop (void *cls,
-                    const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
-  struct MainContext *mc = cls;
-  struct GNUNET_TIME_Relative delay;
-  gint timeout;
-  gint nfds;
-  GPollFD *fds;
-  guint allocated_nfds;
-
-  if ( (tc != NULL) &&
-       (0 != (GNUNET_SCHEDULER_REASON_SHUTDOWN & tc->reason)) )
-    {
-      g_main_loop_quit (mc->gml);
-      GNUNET_SCHEDULER_add_now (&schedule_main_loop,
-                                mc);
-      return;
-    }
-  if (TRUE != g_main_loop_is_running (mc->gml))
-    return;
-  g_main_context_prepare (mc->gmc, &mc->max_priority);
-  allocated_nfds = mc->cached_poll_array_size;
-  fds = mc->cached_poll_array;
-  while ((nfds = g_main_context_query (mc->gmc, mc->max_priority, 
-                                       &timeout, fds, 
-                                       allocated_nfds)) > allocated_nfds)
-    {
-      g_free (fds);
-      mc->cached_poll_array_size = allocated_nfds = nfds;
-      mc->cached_poll_array = fds = g_new (GPollFD, nfds);
-    }
-  mc->poll_array_active = nfds;
-  delay.rel_value = (uint64_t) timeout;
-  GNUNET_NETWORK_fdset_zero (mc->rs);
-  GNUNET_NETWORK_fdset_zero (mc->ws);
-  while (nfds > 0)
-    {
-      nfds--;
-#ifdef __MINGW32__
-      /* TODO: maybe #ifdef WINDOWS? -ndurner */
-      if (fds[nfds].events & (G_IO_IN | G_IO_HUP | G_IO_ERR))
-        GNUNET_NETWORK_fdset_set_native_w32_handle (mc->rs,  (HANDLE) fds[nfds].fd);
-      if (fds[nfds].events & (G_IO_OUT | G_IO_ERR))
-        GNUNET_NETWORK_fdset_set_native_w32_handle (mc->ws,  (HANDLE) fds[nfds].fd);
-#else
-      if (fds[nfds].events & (G_IO_IN | G_IO_HUP | G_IO_ERR))
-        GNUNET_NETWORK_fdset_set_native (mc->rs, fds[nfds].fd);
-      if (fds[nfds].events & (G_IO_OUT | G_IO_ERR))
-        GNUNET_NETWORK_fdset_set_native (mc->ws, fds[nfds].fd);
-#endif
-    }
-  GNUNET_SCHEDULER_add_select (GNUNET_SCHEDULER_PRIORITY_UI,
-                               GNUNET_SCHEDULER_NO_TASK,
-                               delay,
-                               mc->rs,
-                               mc->ws,
-                               &run_ready,
-                               mc);
-}
-
-
-/**
- * Run GTK tasks that are ready.
- *
- * @param cls the 'struct MainContext'
- * @param tc task context
- */
-static void
-run_ready (void *cls,
-           const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
-  struct MainContext *mc = cls;
-  
-  g_poll (mc->cached_poll_array,
-          mc->poll_array_active,
-          0);               
-  if (TRUE == 
-      g_main_context_check (mc->gmc, 
-                            mc->max_priority, 
-                            mc->cached_poll_array, 
-                            mc->poll_array_active))
-    g_main_context_dispatch (mc->gmc);
-  GNUNET_SCHEDULER_add_now (&schedule_main_loop,
-                            mc);
-}
-
-
 /**
  * Main context (global so we can free it on exit and use
  * it for termination).
@@ -562,6 +999,21 @@ run (void *cls,
   if (mc->builder == NULL)
     return;
 
+#if WINDOWS
+  mc->hEventRead = CreateEvent (NULL, TRUE, FALSE, NULL);
+  mc->hEventReadReady = CreateEvent (NULL, TRUE, TRUE, NULL);
+  mc->hEventWrite = CreateEvent (NULL, TRUE, FALSE, NULL);
+  mc->hEventException = CreateEvent (NULL, TRUE, FALSE, NULL);
+
+  mc->hEventPipeWrite = CreateEvent (NULL, TRUE, TRUE, NULL);
+
+  mc->handles_read = GNUNET_CONTAINER_slist_create ();
+  mc->handles_write = GNUNET_CONTAINER_slist_create ();
+  mc->handles_except = GNUNET_CONTAINER_slist_create ();
+  mc->read_array = NULL;
+  mc->read_array_length = 0;
+#endif
+
   /* setup tray icon */
   tray_icon = create_tray_icon();
 
@@ -618,7 +1070,7 @@ run (void *cls,
     }
   /* start the event loop */
   GNUNET_assert (TRUE == g_main_context_acquire (mc->gmc));
-  schedule_main_loop (mc, NULL);
+  GNUNET_SCHEDULER_set_select (gnunet_gtk_select, mc);
 }
 
 
@@ -634,7 +1086,6 @@ GNUNET_GTK_get_main_window_object (const char *name)
   return gtk_builder_get_object (mc->builder, name);
 }
 
-
 int
 main (int argc, char *const *argv)
 {