gio/kqueue/kqueue-thread.c

   1 /*******************************************************************************
   2   Copyright (c) 2011, 2012 Dmitry Matveev <me@dmitrymatveev.co.uk>
   3
   4   Permission is hereby granted, free of charge, to any person obtaining a copy
   5   of this software and associated documentation files (the "Software"), to deal
   6   in the Software without restriction, including without limitation the rights
   7   to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
   8   copies of the Software, and to permit persons to whom the Software is
   9   furnished to do so, subject to the following conditions:
  10
  11   The above copyright notice and this permission notice shall be included in
  12   all copies or substantial portions of the Software.
  13
  14   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  15   IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  16   FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  17   AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  18   LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
  19   OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
  20   THE SOFTWARE.
  21 *******************************************************************************/
  22
  23 #include "config.h"
  24 #include <sys/types.h>
  25 #include <sys/event.h>
  26 #include <sys/time.h>
  27 #include <unistd.h>
  28 #include <errno.h>
  29 #include <glib.h>
  30
  31 #include "kqueue-thread.h"
  32 #include "kqueue-sub.h"
  33 #include "kqueue-utils.h"
  34
  35 static gboolean kt_debug_enabled = FALSE;
  36 #define KT_W if (kt_debug_enabled) g_warning
  37
  38 static GQueue pick_up_fds_queue = G_QUEUE_INIT;
  39 G_LOCK_DEFINE_STATIC (pick_up_lock);
  40
  41 static GSList *remove_fds_list = NULL;
  42 G_LOCK_DEFINE_STATIC (remove_lock);
  43
  44 /* GIO does not have analogues for NOTE_LINK and(?) NOTE_REVOKE, so
  45  * we do not ask kqueue() to watch for these events for now. */
  46 const uint32_t KQUEUE_VNODE_FLAGS =
  47   NOTE_DELETE | NOTE_WRITE | NOTE_EXTEND | NOTE_ATTRIB | NOTE_RENAME;
  48
  49 extern int get_kqueue_descriptor(void);
  50
  51 /*
  52  * _kqueue_thread_collect_fds:
  53  * @events: a #kevents - the list of events to monitor. Will be extended
  54  *     with new items.
  55  *
  56  * Picks up new file descriptors for monitoring from a global queue.
  57  *
  58  * To add new items to the list, use _kqueue_thread_push_fd().
  59  */
  60 static void
  61 _kqueue_thread_collect_fds (kevents *events)
  62 {
  63   g_assert (events != NULL);
  64   gint length = 0;
  65
  66   G_LOCK (pick_up_lock);
  67   if ((length = g_queue_get_length (&pick_up_fds_queue)) != 0)
  68     {
  69       gpointer fdp = NULL;
  70       kevents_extend_sz (events, length);
  71
  72       while ((fdp = g_queue_pop_head (&pick_up_fds_queue)) != NULL)
  73         {
  74           struct kevent *pevent = &events->memory[events->kq_size++];
  75           EV_SET (pevent,
  76                   GPOINTER_TO_INT (fdp),
  77                   EVFILT_VNODE,
  78                   EV_ADD | EV_ENABLE | EV_ONESHOT,
  79                   KQUEUE_VNODE_FLAGS,
  80                   0,
  81                   0);
  82         }
  83     }
  84   G_UNLOCK (pick_up_lock);
  85 }
  86
  87
  88 /*
  89  * _kqueue_thread_cleanup_fds:
  90  * @events: a #kevents -- list of events to monitor. Cancelled
  91  *     subscriptions will be removed from it, and its size
  92  *     probably will be reduced.
  93  *
  94  * Removes file descriptors from monitoring.
  95  *
  96  * This function will pick up file descriptors from a global list
  97  * to cancel monitoring on them. The list will be freed then.
  98  *
  99  * To add new items to the list, use _kqueue_thread_remove_fd().
 100  */
 101 static void
 102 _kqueue_thread_cleanup_fds (kevents *events)
 103 {
 104   g_assert (events != NULL);
 105
 106   G_LOCK (remove_lock);
 107   if (remove_fds_list)
 108     {
 109       size_t oldsize = events->kq_size;
 110       int i, j;
 111
 112       for (i = 1, j = 1; i < oldsize; i++)
 113         {
 114           int fd = events->memory[i].ident;
 115           GSList *elem = g_slist_find (remove_fds_list, GINT_TO_POINTER (fd));
 116           if (elem == NULL)
 117             {
 118               if (i != j)
 119                 events->memory[j] = events->memory[i];
 120               ++j;
 121             }
 122           else if (close (fd) == -1)
 123             KT_W ("Failed to close fd %d, error %d", fd, errno);
 124         }
 125
 126       KT_W ("FD Clean up complete, kq_size now %d\n", j);
 127       events->kq_size = j;
 128       kevents_reduce (events);
 129       g_slist_free (remove_fds_list);
 130       remove_fds_list = NULL;
 131     }
 132   G_UNLOCK (remove_lock);
 133 }
 134
 135
 136 /*
 137  * _kqueue_thread_drop_fd:
 138  * @events: a #kevents -- list of events to monitor. Cancelled
 139  *     subscriptions will be removed from it, and its size
 140  *     probably will be reduced.
 141  *
 142  * Removes a concrete file descriptor from monitoring.
 143  */
 144 static void
 145 _kqueue_thread_drop_fd (kevents *events, int fd)
 146 {
 147   g_assert (events != NULL);
 148
 149   int i;
 150   for (i = 1; i < events->kq_size; i++)
 151     {
 152       if (events->memory[i].ident == fd)
 153         {
 154           if (close (fd) == -1)
 155             KT_W ("Failed to close fd %d, error %d", fd, errno);
 156
 157           events->memory[i] = events->memory[--events->kq_size];
 158           return;
 159         }
 160     }
 161 }
 162
 163 /*
 164  * _kqueue_thread_func:
 165  * @arg: a pointer to int -- control file descriptor.
 166  *
 167  * The thread communicates with the outside world through a so-called
 168  * command file descriptor. The thread reads control commands from it
 169  * and writes the notifications into it.
 170  *
 171  * Control commands are single-byte characters:
 172  * - 'A' - pick up new file descriptors to monitor
 173  * - 'R' - remove some descriptors from monitoring.
 174  *
 175  * For details, see _kqueue_thread_collect_fds() and
 176  * _kqueue_thread_cleanup_fds().
 177  *
 178  * Notifications, that thread writes into the command file descriptor,
 179  * are represented with #kqueue_notification objects.
 180  *
 181  * Returns: %NULL
 182  */
 183 void*
 184 _kqueue_thread_func (void *arg)
 185 {
 186   int fd, kqueue_descriptor;
 187   kevents waiting;
 188
 189   g_assert (arg != NULL);
 190   kevents_init_sz (&waiting, 1);
 191
 192   fd = *(int *) arg;
 193
 194   kqueue_descriptor = get_kqueue_descriptor();
 195   if (kqueue_descriptor == -1)
 196     {
 197       KT_W ("fatal: kqueue is not initialized!\n");
 198       return NULL;
 199     }
 200
 201   EV_SET (&waiting.memory[0],
 202           fd,
 203           EVFILT_READ,
 204           EV_ADD | EV_ENABLE | EV_ONESHOT,
 205           NOTE_LOWAT,
 206           1,
 207           0);
 208   waiting.kq_size = 1;
 209
 210   for (;;)
 211     {
 212       /* TODO: Provide more items in the 'eventlist' to kqueue(2).
 213        * Currently the backend takes notifications from the kernel one
 214        * by one, i.e. there will be a lot of system calls and context
 215        * switches when the application will monitor a lot of files with
 216        * high filesystem activity on each. */
 217
 218       struct kevent received;
 219       KT_W ("Watching for %zi items", waiting.kq_size);
 220       int ret = kevent (kqueue_descriptor, waiting.memory, waiting.kq_size, &received, 1, NULL);
 221       int kevent_errno = errno;
 222       KT_W ("Awoken.");
 223
 224       if (ret == -1)
 225         {
 226           KT_W ("kevent failed: %d", kevent_errno);
 227           if (kevent_errno == EINTR)
 228             continue;
 229           else
 230             return NULL;
 231         }
 232
 233       if (received.ident == fd)
 234         {
 235           char c;
 236             if (!_ku_read (fd, &c, 1))
 237               {
 238                 KT_W ("Failed to read command, error %d", errno);
 239                 continue;
 240               }
 241           if (c == 'A')
 242             _kqueue_thread_collect_fds (&waiting);
 243           else if (c == 'R')
 244             _kqueue_thread_cleanup_fds (&waiting);
 245         }
 246       else
 247         {
 248           struct kqueue_notification kn;
 249           kn.fd = received.ident;
 250
 251           if (received.flags & EV_ERROR)
 252             {
 253               kn.flags = NOTE_REVOKE;
 254               _kqueue_thread_drop_fd (&waiting, received.ident);
 255             }
 256           else
 257             kn.flags = (received.fflags & ~NOTE_REVOKE);
 258
 259           if (!_ku_write (fd, &kn, sizeof (struct kqueue_notification)))
 260             KT_W ("Failed to write a kqueue notification, error %d", errno);
 261         }
 262     }
 263   kevents_free (&waiting);
 264   return NULL;
 265 }
 266
 267
 268 /*
 269  * _kqueue_thread_push_fd:
 270  * @fd: a file descriptor
 271  *
 272  * Puts a new file descriptor into the pick up list for monitroing.
 273  *
 274  * The kqueue thread will not start monitoring on it immediately, it
 275  * should be bumped via its command file descriptor manually.
 276  * See kqueue_thread() and _kqueue_thread_collect_fds() for details.
 277  */
 278 void
 279 _kqueue_thread_push_fd (int fd)
 280 {
 281   G_LOCK (pick_up_lock);
 282   g_queue_push_tail (&pick_up_fds_queue, GINT_TO_POINTER (fd));
 283   G_UNLOCK (pick_up_lock);
 284 }
 285
 286
 287 /*
 288  * _kqueue_thread_remove_fd:
 289  * @fd: a file descriptor
 290  *
 291  * Puts a new file descriptor into the remove list to cancel monitoring
 292  * on it.
 293  *
 294  * The kqueue thread will not stop monitoring on it immediately, it
 295  * should be bumped via its command file descriptor manually.
 296  * See kqueue_thread() and _kqueue_thread_collect_fds() for details.
 297  */
 298 void
 299 _kqueue_thread_remove_fd (int fd)
 300 {
 301   G_LOCK (remove_lock);
 302   remove_fds_list = g_slist_prepend (remove_fds_list, GINT_TO_POINTER (fd));
 303   G_UNLOCK (remove_lock);
 304 }