GDBusMessage: fast-path decoding of fixed arrays
[platform/upstream/glib.git] / 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 }