inotify: rewrite inotify-kernel

Remove the hardwired 1 second event queue logic from inotify-kernel and
replace it with something vastly less complicated.

Events are now reported as soon as is possible instead of after a
delay.

We still must delay IN_MOVED_FROM events in order to look for the
matching IN_MOVED_TO events, and since we want to report events in order
this means that events behind those events can also be delayed.  We
limit ourselves, however:

 - no more than 100 events can be delayed at a time

 - no event can be delayed by more than 10ms

https://bugzilla.gnome.org/show_bug.cgi?id=627285

2 files changed, 183 insertions, 477 deletions

diff --git a/gio/inotify/inotify-kernel.c b/gio/inotify/inotify-kernel.c
index b672cb168..1d20ecf3e 100644
--- a/gio/inotify/inotify-kernel.c
+++ b/gio/inotify/inotify-kernel.c
@@ -1,5 +1,6 @@
 /*
    Copyright (C) 2005 John McCutchan
+   Copyright © 2015 Canonical Limited
 
    The Gnome Library is free software; you can redistribute it and/or
    modify it under the terms of the GNU Library General Public License as
@@ -15,8 +16,9 @@
    License along with the Gnome Library; see the file COPYING.LIB.  If not,
    see <http://www.gnu.org/licenses/>.
 
-   Authors:.
-		John McCutchan <john@johnmccutchan.com>
+   Authors:
+     Ryan Lortie <desrt@desrt.ca>
+     John McCutchan <john@johnmccutchan.com>
 */
 
 #include "config.h"
@@ -29,234 +31,40 @@
 #include <glib.h>
 #include "inotify-kernel.h"
 #include <sys/inotify.h>
+#include <glib/glib-unix.h>
 
 #include "glib-private.h"
 
-/* Timings for pairing MOVED_TO / MOVED_FROM events */
-#define PROCESS_EVENTS_TIME 1000 /* 1000 milliseconds (1 hz) */
-#define DEFAULT_HOLD_UNTIL_TIME 0 /* 0 millisecond */
-#define MOVE_HOLD_UNTIL_TIME 500 /* 500 microseconds or 0.5 milliseconds */
-
-static int inotify_instance_fd = -1;
-static GQueue *events_to_process = NULL;
-static GQueue *event_queue = NULL;
-static GHashTable * cookie_hash = NULL;
-static GIOChannel *inotify_read_ioc;
-static GPollFD ik_poll_fd;
-static gboolean ik_poll_fd_enabled = TRUE;
-static void (*user_cb)(ik_event_t *event);
-
-static gboolean ik_read_callback (gpointer user_data);
-static gboolean ik_process_eq_callback (gpointer user_data);
-
-static guint32 ik_move_matches = 0;
-static guint32 ik_move_misses = 0;
-
-static gboolean process_eq_running = FALSE;
+/* Define limits on the maximum amount of time and maximum amount of
+ * interceding events between FROM/TO that can be merged.
+ */
+#define MOVE_PAIR_DELAY      (10 * G_TIME_SPAN_MILLISECOND)
+#define MOVE_PAIR_DISTANCE   (100)
 
 /* We use the lock from inotify-helper.c
  *
- * There are two places that we take this lock
- *
- * 1) In ik_read_callback
- *
- * 2) ik_process_eq_callback.
- *
+ * We only have to take it on our read callback.
  *
  * The rest of locking is taken care of in inotify-helper.c
  */
 G_LOCK_EXTERN (inotify_lock);
 
-typedef struct ik_event_internal {
-  ik_event_t *event;
-  gboolean seen;
-  gboolean sent;
-  GTimeVal hold_until;
-  struct ik_event_internal *pair;
-} ik_event_internal_t;
-
-/* In order to perform non-sleeping inotify event chunking we need
- * a custom GSource
- */
-static gboolean
-ik_source_prepare (GSource *source,
-		   gint    *timeout)
-{
-  return FALSE;
-}
-
-static gboolean
-ik_source_timeout (gpointer data)
-{
-  GSource *source = (GSource *)data;
-  
-  /* Re-active the PollFD */
-  g_source_add_poll (source, &ik_poll_fd);
-  g_source_unref (source);
-  ik_poll_fd_enabled = TRUE;
-  
-  return FALSE;
-}
-
-#define MAX_PENDING_COUNT 2
-#define PENDING_THRESHOLD(qsize) ((qsize) >> 1)
-#define PENDING_MARGINAL_COST(p) ((unsigned int)(1 << (p)))
-#define MAX_QUEUED_EVENTS 2048
-#define AVERAGE_EVENT_SIZE sizeof (struct inotify_event) + 16
-#define TIMEOUT_MILLISECONDS 10
-
-static gboolean
-ik_source_check (GSource *source)
-{
-  static int prev_pending = 0, pending_count = 0;
-  
-  /* We already disabled the PollFD or
-   * nothing to be read from inotify */
-  if (!ik_poll_fd_enabled || !(ik_poll_fd.revents & G_IO_IN))
-    return FALSE;
-
-  if (pending_count < MAX_PENDING_COUNT)
-    {
-      GSource *timeout_source;
-      unsigned int pending;
-      
-      if (ioctl (inotify_instance_fd, FIONREAD, &pending) == -1)
-	goto do_read;
-      
-      pending /= AVERAGE_EVENT_SIZE;
-      
-      /* Don't wait if the number of pending events is too close
-       * to the maximum queue size.
-       */
-      if (pending > PENDING_THRESHOLD (MAX_QUEUED_EVENTS))
-	goto do_read;
-      
-      /* With each successive iteration, the minimum rate for
-       * further sleep doubles. 
-       */
-      if (pending-prev_pending < PENDING_MARGINAL_COST (pending_count))
-	goto do_read;
-      
-      prev_pending = pending;
-      pending_count++;
-      
-      /* We are going to wait to read the events: */
-      
-      /* Remove the PollFD from the source */
-      g_source_remove_poll (source, &ik_poll_fd);
-      /* To avoid threading issues we need to flag that we've done that */
-      ik_poll_fd_enabled = FALSE;
-      /* Set a timeout to re-add the PollFD to the source */
-      g_source_ref (source);
-
-      timeout_source = g_timeout_source_new (TIMEOUT_MILLISECONDS);
-      g_source_set_callback (timeout_source, ik_source_timeout, source, NULL);
-      g_source_attach (timeout_source, GLIB_PRIVATE_CALL (g_get_worker_context) ());
-      g_source_unref (timeout_source);
-
-      return FALSE;
-    }
-
-do_read:
-  /* We are ready to read events from inotify */
-
-  prev_pending = 0;
-  pending_count = 0;
-  
-  return TRUE;
-}
-
-static gboolean
-ik_source_dispatch (GSource     *source,
-		    GSourceFunc  callback,
-		    gpointer     user_data)
-{
-  if (callback)
-    return callback (user_data);
-  return TRUE;
-}
-
-static GSourceFuncs ik_source_funcs =
-{
-  ik_source_prepare,
-  ik_source_check,
-  ik_source_dispatch,
-  NULL
-};
-
-gboolean _ik_startup (void (*cb)(ik_event_t *event))
-{
-  static gboolean initialized = FALSE;
-  GSource *source;
-  
-  user_cb = cb;
-  /* Ignore multi-calls */
-  if (initialized) 
-    return inotify_instance_fd >= 0;
-
-  initialized = TRUE;
-
-  inotify_instance_fd = inotify_init1 (IN_CLOEXEC);
-
-  if (inotify_instance_fd < 0)
-    inotify_instance_fd = inotify_init ();
-
-  if (inotify_instance_fd < 0)
-    return FALSE;
-
-  inotify_read_ioc = g_io_channel_unix_new (inotify_instance_fd);
-  ik_poll_fd.fd = inotify_instance_fd;
-  ik_poll_fd.events = G_IO_IN | G_IO_HUP | G_IO_ERR;
-  g_io_channel_set_encoding (inotify_read_ioc, NULL, NULL);
-  g_io_channel_set_flags (inotify_read_ioc, G_IO_FLAG_NONBLOCK, NULL);
-
-  source = g_source_new (&ik_source_funcs, sizeof (GSource));
-  g_source_set_name (source, "GIO Inotify");
-  g_source_add_poll (source, &ik_poll_fd);
-  g_source_set_callback (source, ik_read_callback, NULL, NULL);
-  g_source_attach (source, GLIB_PRIVATE_CALL (g_get_worker_context) ());
-  g_source_unref (source);
-
-  cookie_hash = g_hash_table_new (g_direct_hash, g_direct_equal);
-  event_queue = g_queue_new ();
-  events_to_process = g_queue_new ();
-  
-  return TRUE;
-}
-
-static ik_event_internal_t *
-ik_event_internal_new (ik_event_t *event)
-{
-  ik_event_internal_t *internal_event = g_new0 (ik_event_internal_t, 1);
-  GTimeVal tv;
-  
-  g_assert (event);
-  
-  g_get_current_time (&tv);
-  g_time_val_add (&tv, DEFAULT_HOLD_UNTIL_TIME);
-  internal_event->event = event;
-  internal_event->hold_until = tv;
-  
-  return internal_event;
-}
-
 static ik_event_t *
-ik_event_new (char *buffer)
+ik_event_new (struct inotify_event *kevent,
+              gint64                now)
 {
-  struct inotify_event *kevent = (struct inotify_event *)buffer;
   ik_event_t *event = g_new0 (ik_event_t, 1);
-  
-  g_assert (buffer);
-  
+
   event->wd = kevent->wd;
   event->mask = kevent->mask;
   event->cookie = kevent->cookie;
   event->len = kevent->len;
+  event->timestamp = now;
   if (event->len)
     event->name = g_strdup (kevent->name);
   else
     event->name = g_strdup ("");
-  
+
   return event;
 }
 
@@ -265,325 +73,222 @@ _ik_event_free (ik_event_t *event)
 {
   if (event->pair)
     _ik_event_free (event->pair);
+
   g_free (event->name);
   g_free (event);
 }
 
-gint32
-_ik_watch (const char *path, 
-           guint32     mask, 
-           int        *err)
+typedef struct
 {
-  gint32 wd = -1;
-  
-  g_assert (path != NULL);
-  g_assert (inotify_instance_fd >= 0);
-  
-  wd = inotify_add_watch (inotify_instance_fd, path, mask);
-  
-  if (wd < 0)
-    {
-      int e = errno;
-      /* FIXME: debug msg failed to add watch */
-      if (err)
-	*err = e;
-      return wd;
-    }
-  
-  g_assert (wd >= 0);
-  return wd;
+  GSource  source;
+
+  gpointer fd_tag;
+  GQueue   queue;
+  gint     fd;
+} InotifyKernelSource;
+
+static InotifyKernelSource *inotify_source;
+
+static gint64
+ik_source_get_ready_time (InotifyKernelSource *iks)
+{
+  ik_event_t *head;
+
+  head = g_queue_peek_head (&iks->queue);
+
+  /* nothing in the queue: not ready */
+  if (!head)
+    return -1;
+
+  /* if it's not an unpaired move, it is ready now */
+  if (~head->mask & IN_MOVED_FROM || head->pair)
+    return 0;
+
+  /* if the queue is too long then it's ready now */
+  if (iks->queue.length > MOVE_PAIR_DISTANCE)
+    return 0;
+
+  /* otherwise, it's ready after the delay */
+  return head->timestamp + MOVE_PAIR_DELAY;
 }
 
-int
-_ik_ignore (const char *path, 
-            gint32      wd)
+static gboolean
+ik_source_can_dispatch_now (InotifyKernelSource *iks,
+                            gint64               now)
 {
-  g_assert (wd >= 0);
-  g_assert (inotify_instance_fd >= 0);
-  
-  if (inotify_rm_watch (inotify_instance_fd, wd) < 0)
-    {
-      /* int e = errno; */
-      /* failed to rm watch */
-      return -1;
-    }
-  
-  return 0;
+  gint64 ready_time;
+
+  ready_time = ik_source_get_ready_time (iks);
+
+  return 0 <= ready_time && ready_time <= now;
 }
 
 static void
-ik_read_events (gsize  *buffer_size_out, 
-                gchar **buffer_out)
+ik_source_try_to_pair_head (InotifyKernelSource *iks)
 {
-  static gchar *buffer = NULL;
-  static gsize buffer_size;
-  
-  /* Initialize the buffer on our first call */
-  if (buffer == NULL)
-    {
-      buffer_size = AVERAGE_EVENT_SIZE;
-      buffer_size *= MAX_QUEUED_EVENTS;
-      buffer = g_malloc (buffer_size);
-    }
+  ik_event_t *head;
+  GList *node;
+
+  node = g_queue_peek_head_link (&iks->queue);
+
+  if (!node)
+    return;
 
-  *buffer_size_out = 0;
-  *buffer_out = NULL;
-  
-  memset (buffer, 0, buffer_size);
+  head = node->data;
 
-  if (g_io_channel_read_chars (inotify_read_ioc, (char *)buffer, buffer_size, buffer_size_out, NULL) != G_IO_STATUS_NORMAL) {
-    /* error reading */
-  }
-  *buffer_out = buffer;
+  /* we should only be here if... */
+  g_assert (head->mask & IN_MOVED_FROM && !head->pair);
+
+  while ((node = node->next))
+    {
+      ik_event_t *candidate = node->data;
+
+      if (candidate->cookie == head->cookie && candidate->mask & IN_MOVED_TO)
+        {
+          g_queue_remove (&iks->queue, candidate);
+          candidate->is_second_in_pair = TRUE;
+          head->pair = candidate;
+          return;
+        }
+    }
 }
 
 static gboolean
-ik_read_callback (gpointer user_data)
+ik_source_dispatch (GSource     *source,
+                    GSourceFunc  func,
+                    gpointer     user_data)
 {
-  gchar *buffer;
-  gsize buffer_size, buffer_i, events;
-  
-  G_LOCK (inotify_lock);
-  ik_read_events (&buffer_size, &buffer);
-  
-  buffer_i = 0;
-  events = 0;
-  while (buffer_i < buffer_size)
-    {
-      struct inotify_event *event;
-      gsize event_size;
-      event = (struct inotify_event *)&buffer[buffer_i];
-      event_size = sizeof(struct inotify_event) + event->len;
-      g_queue_push_tail (events_to_process, ik_event_internal_new (ik_event_new (&buffer[buffer_i])));
-      buffer_i += event_size;
-      events++;
-    }
-  
-  /* If the event process callback is off, turn it back on */
-  if (!process_eq_running && events)
+  InotifyKernelSource *iks = (InotifyKernelSource *) source;
+  void (*user_callback) (ik_event_t *event) = (void *) func;
+  gint64 now = g_source_get_time (source);
+
+  now = g_source_get_time (source);
+
+  /* Only try to fill the queue if we don't already have work to do. */
+  if (!ik_source_can_dispatch_now (iks, now) &&
+      g_source_query_unix_fd (source, iks->fd_tag))
     {
-      GSource *timeout_source;
+      gchar buffer[sizeof(struct inotify_event) + NAME_MAX + 1];
+      gssize result;
+      gssize offset;
+
+      result = read (iks->fd, buffer, sizeof buffer);
+
+      if (result < 0)
+        g_error ("inotify error: %s\n", g_strerror (errno));
+      else if (result == 0)
+        g_error ("inotify unexpectedly hit eof");
 
-      process_eq_running = TRUE;
-      timeout_source = g_timeout_source_new (PROCESS_EVENTS_TIME);
-      g_source_set_callback (timeout_source, ik_process_eq_callback, NULL, NULL);
-      g_source_attach (timeout_source, GLIB_PRIVATE_CALL (g_get_worker_context ()));
-      g_source_unref (timeout_source);
+      offset = 0;
+
+      while (offset < result)
+        {
+          struct inotify_event *event = (struct inotify_event *) (buffer + offset);
+
+          g_queue_push_tail (&iks->queue, ik_event_new (event, now));
+
+          offset += sizeof (struct inotify_event) + event->len;
+        }
     }
-  
-  G_UNLOCK (inotify_lock);
-  
-  return TRUE;
-}
 
-static gboolean
-g_timeval_lt (GTimeVal *val1,
-              GTimeVal *val2)
-{
-  if (val1->tv_sec < val2->tv_sec)
-    return TRUE;
+  if (!ik_source_can_dispatch_now (iks, now))
+    ik_source_try_to_pair_head (iks);
+
+  if (ik_source_can_dispatch_now (iks, now))
+    {
+      ik_event_t *event;
+
+      /* callback will free the event */
+      event = g_queue_pop_head (&iks->queue);
 
-  if (val1->tv_sec > val2->tv_sec)
-    return FALSE;
+      G_LOCK (inotify_lock);
+      (* user_callback) (event);
+      G_UNLOCK (inotify_lock);
+    }
 
-  /* val1->tv_sec == val2->tv_sec */
-  if (val1->tv_usec < val2->tv_usec)
-    return TRUE;
+  g_source_set_ready_time (source, ik_source_get_ready_time (iks));
 
-  return FALSE;
+  return TRUE;
 }
 
-static gboolean
-g_timeval_le (GTimeVal *val1,
-              GTimeVal *val2)
+static InotifyKernelSource *
+ik_source_new (void (* callback) (ik_event_t *event))
 {
-  if (val1->tv_sec < val2->tv_sec)
-    return TRUE;
+  static GSourceFuncs source_funcs = {
+    NULL, NULL,
+    ik_source_dispatch
+    /* should have a finalize, but it will never happen */
+  };
+  InotifyKernelSource *iks;
+  GSource *source;
 
-  if (val1->tv_sec > val2->tv_sec)
-    return FALSE;
+  source = g_source_new (&source_funcs, sizeof (InotifyKernelSource));
+  iks = (InotifyKernelSource *) source;
 
-  /* val1->tv_sec == val2->tv_sec */
-  if (val1->tv_usec <= val2->tv_usec)
-    return TRUE;
+  g_source_set_name (source, "inotify kernel source");
 
-  return FALSE;
-}
+  iks->fd = inotify_init1 (IN_CLOEXEC);
 
-static void
-ik_pair_events (ik_event_internal_t *event1, 
-                ik_event_internal_t *event2)
-{
-  g_assert (event1 && event2);
-  /* We should only be pairing events that have the same cookie */
-  g_assert (event1->event->cookie == event2->event->cookie);
-  /* We shouldn't pair an event that already is paired */
-  g_assert (event1->pair == NULL && event2->pair == NULL);
+  if (iks->fd < 0)
+    iks->fd = inotify_init ();
 
-  /* Pair the internal structures and the ik_event_t structures */
-  event1->pair = event2;
-  event1->event->pair = event2->event;
-  event2->event->is_second_in_pair = TRUE;
+  if (iks->fd >= 0)
+    iks->fd_tag = g_source_add_unix_fd (source, iks->fd, G_IO_IN);
 
-  if (g_timeval_lt (&event1->hold_until, &event2->hold_until))
-    event1->hold_until = event2->hold_until;
+  g_source_set_callback (source, (GSourceFunc) callback, NULL, NULL);
 
-  event2->hold_until = event1->hold_until;
-}
+  g_source_attach (source, GLIB_PRIVATE_CALL (g_get_worker_context) ());
 
-static void
-ik_event_add_microseconds (ik_event_internal_t *event, 
-                           glong                ms)
-{
-  g_assert (event);
-  g_time_val_add (&event->hold_until, ms);
+  return iks;
 }
 
-static gboolean
-ik_event_ready (ik_event_internal_t *event)
+gboolean
+_ik_startup (void (*cb)(ik_event_t *event))
 {
-  GTimeVal tv;
-  g_assert (event);
-  
-  g_get_current_time (&tv);
-  
-  /* An event is ready if,
-   *
-   * it has no cookie -- there is nothing to be gained by holding it
-   * or, it is already paired -- we don't need to hold it anymore
-   * or, we have held it long enough
-   */
-  return
-    event->event->cookie == 0 ||
-    event->pair != NULL ||
-    g_timeval_le (&event->hold_until, &tv);
-}
+  if (g_once_init_enter (&inotify_source))
+    g_once_init_leave (&inotify_source, ik_source_new (cb));
 
-static void
-ik_pair_moves (gpointer data, 
-               gpointer user_data)
-{
-  ik_event_internal_t *event = (ik_event_internal_t *)data;
-  
-  if (event->seen == TRUE || event->sent == TRUE)
-    return;
-  
-  if (event->event->cookie != 0)
-    {
-      /* When we get a MOVED_FROM event we delay sending the event by
-       * MOVE_HOLD_UNTIL_TIME microseconds. We need to do this because a
-       * MOVED_TO pair _might_ be coming in the near future */
-      if (event->event->mask & IN_MOVED_FROM)
-	{
-	  g_hash_table_insert (cookie_hash, GINT_TO_POINTER (event->event->cookie), event);
-	  /* because we don't deliver move events there is no point in waiting for the match right now. */
-	  ik_event_add_microseconds (event, MOVE_HOLD_UNTIL_TIME);
-	}
-      else if (event->event->mask & IN_MOVED_TO)
-	{
-	  /* We need to check if we are waiting for this MOVED_TO events cookie to pair it with
-	   * a MOVED_FROM */
-	  ik_event_internal_t *match = NULL;
-	  match = g_hash_table_lookup (cookie_hash, GINT_TO_POINTER (event->event->cookie));
-	  if (match)
-	    {
-	      g_hash_table_remove (cookie_hash, GINT_TO_POINTER (event->event->cookie));
-	      ik_pair_events (match, event);
-	    }
-	}
-    }
-  event->seen = TRUE;
+  return inotify_source->fd >= 0;
 }
 
-static void
-ik_process_events (void)
+gint32
+_ik_watch (const char *path,
+           guint32     mask,
+           int        *err)
 {
-  g_queue_foreach (events_to_process, ik_pair_moves, NULL);
+  gint32 wd = -1;
+
+  g_assert (path != NULL);
+  g_assert (inotify_source && inotify_source->fd >= 0);
+
+  wd = inotify_add_watch (inotify_source->fd, path, mask);
 
-  while (!g_queue_is_empty (events_to_process))
+  if (wd < 0)
     {
-      ik_event_internal_t *event = g_queue_peek_head (events_to_process);
-      
-      /* This must have been sent as part of a MOVED_TO/MOVED_FROM */
-      if (event->sent)
-	{
-	  /* Pop event */
-	  g_queue_pop_head (events_to_process);
-	  /* Free the internal event structure */
-	  g_free (event);
-	  continue;
-	}
-      
-      /* The event isn't ready yet */
-      if (!ik_event_ready (event))
-	break;
-      
-      /* Pop it */
-      event = g_queue_pop_head (events_to_process);
-      
-      /* Check if this is a MOVED_FROM that is also sitting in the cookie_hash */
-      if (event->event->cookie && event->pair == NULL &&
-	  g_hash_table_lookup (cookie_hash, GINT_TO_POINTER (event->event->cookie)))
-	g_hash_table_remove (cookie_hash, GINT_TO_POINTER (event->event->cookie));
-      
-      if (event->pair)
-	{
-	  /* We send out paired MOVED_FROM/MOVED_TO events in the same event buffer */
-	  /* g_assert (event->event->mask == IN_MOVED_FROM && event->pair->event->mask == IN_MOVED_TO); */
-	  /* Copy the paired data */
-	  event->pair->sent = TRUE;
-	  event->sent = TRUE;
-	  ik_move_matches++;
-	}
-      else if (event->event->cookie)
-	{
-	  /* If we couldn't pair a MOVED_FROM and MOVED_TO together, we change
-	   * the event masks */
-	  /* Changeing MOVED_FROM to DELETE and MOVED_TO to create lets us make
-	   * the gaurantee that you will never see a non-matched MOVE event */
-	  event->event->original_mask = event->event->mask;
-
-	  if (event->event->mask & IN_MOVED_FROM)
-	    {
-	      event->event->mask = IN_DELETE|(event->event->mask & IN_ISDIR);
-	      ik_move_misses++; /* not super accurate, if we aren't watching the destination it still counts as a miss */
-	    }
-	  if (event->event->mask & IN_MOVED_TO)
-	    event->event->mask = IN_CREATE|(event->event->mask & IN_ISDIR);
-	}
-      
-      /* Push the ik_event_t onto the event queue */
-      g_queue_push_tail (event_queue, event->event);
-      /* Free the internal event structure */
-      g_free (event);
+      int e = errno;
+      /* FIXME: debug msg failed to add watch */
+      if (err)
+        *err = e;
+      return wd;
     }
+
+  g_assert (wd >= 0);
+  return wd;
 }
 
-static gboolean
-ik_process_eq_callback (gpointer user_data)
+int
+_ik_ignore (const char *path,
+            gint32      wd)
 {
-  gboolean res;
-  
-  /* Try and move as many events to the event queue */
-  G_LOCK (inotify_lock);
-  ik_process_events ();
-  
-  while (!g_queue_is_empty (event_queue))
-    {
-      ik_event_t *event = g_queue_pop_head (event_queue);
-      
-      user_cb (event);
-    }
+  g_assert (wd >= 0);
+  g_assert (inotify_source && inotify_source->fd >= 0);
 
-  res = TRUE;
-  
-  if (g_queue_get_length (events_to_process) == 0)
+  if (inotify_rm_watch (inotify_source->fd, wd) < 0)
     {
-      process_eq_running = FALSE;
-      res = FALSE;
+      /* int e = errno; */
+      /* failed to rm watch */
+      return -1;
     }
-  
-  G_UNLOCK (inotify_lock);
-  
-  return res;
+
+  return 0;
 }
diff --git a/gio/inotify/inotify-kernel.h b/gio/inotify/inotify-kernel.h
index e1a2b9c3b..26ab19d28 100644
--- a/gio/inotify/inotify-kernel.h
+++ b/gio/inotify/inotify-kernel.h
@@ -37,6 +37,7 @@ typedef struct ik_event_s {
    * then event1->pair == event2 and event2->pair == NULL.
    * It will result also in event1->pair->is_second_in_pair == TRUE */
   struct ik_event_s *pair;
+  gint64 timestamp; /* monotonic time that this was created */
 } ik_event_t;
 
 gboolean _ik_startup (void (*cb) (ik_event_t *event));