1 /*******************************************************************************
2 Copyright (c) 2011, 2012 Dmitry Matveev <me@dmitrymatveev.co.uk>
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:
11 The above copyright notice and this permission notice shall be included in
12 all copies or substantial portions of the Software.
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
21 *******************************************************************************/
24 #include <sys/types.h>
25 #include <sys/event.h>
31 #include "kqueue-thread.h"
32 #include "kqueue-sub.h"
33 #include "kqueue-utils.h"
35 static gboolean kt_debug_enabled = FALSE;
36 #define KT_W if (kt_debug_enabled) g_warning
38 static GQueue pick_up_fds_queue = G_QUEUE_INIT;
39 G_LOCK_DEFINE_STATIC (pick_up_lock);
41 static GSList *remove_fds_list = NULL;
42 G_LOCK_DEFINE_STATIC (remove_lock);
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;
49 extern int get_kqueue_descriptor(void);
52 * _kqueue_thread_collect_fds:
53 * @events: a #kevents - the list of events to monitor. Will be extended
56 * Picks up new file descriptors for monitoring from a global queue.
58 * To add new items to the list, use _kqueue_thread_push_fd().
61 _kqueue_thread_collect_fds (kevents *events)
63 g_assert (events != NULL);
66 G_LOCK (pick_up_lock);
67 if ((length = g_queue_get_length (&pick_up_fds_queue)) != 0)
70 kevents_extend_sz (events, length);
72 while ((fdp = g_queue_pop_head (&pick_up_fds_queue)) != NULL)
74 struct kevent *pevent = &events->memory[events->kq_size++];
76 GPOINTER_TO_INT (fdp),
78 EV_ADD | EV_ENABLE | EV_ONESHOT,
84 G_UNLOCK (pick_up_lock);
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.
94 * Removes file descriptors from monitoring.
96 * This function will pick up file descriptors from a global list
97 * to cancel monitoring on them. The list will be freed then.
99 * To add new items to the list, use _kqueue_thread_remove_fd().
102 _kqueue_thread_cleanup_fds (kevents *events)
104 g_assert (events != NULL);
106 G_LOCK (remove_lock);
109 size_t oldsize = events->kq_size;
112 for (i = 1, j = 1; i < oldsize; i++)
114 int fd = events->memory[i].ident;
115 GSList *elem = g_slist_find (remove_fds_list, GINT_TO_POINTER (fd));
119 events->memory[j] = events->memory[i];
122 else if (close (fd) == -1)
123 KT_W ("Failed to close fd %d, error %d", fd, errno);
126 KT_W ("FD Clean up complete, kq_size now %d\n", j);
128 kevents_reduce (events);
129 g_slist_free (remove_fds_list);
130 remove_fds_list = NULL;
132 G_UNLOCK (remove_lock);
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.
142 * Removes a concrete file descriptor from monitoring.
145 _kqueue_thread_drop_fd (kevents *events, int fd)
147 g_assert (events != NULL);
150 for (i = 1; i < events->kq_size; i++)
152 if (events->memory[i].ident == fd)
154 if (close (fd) == -1)
155 KT_W ("Failed to close fd %d, error %d", fd, errno);
157 events->memory[i] = events->memory[--events->kq_size];
164 * _kqueue_thread_func:
165 * @arg: a pointer to int -- control file descriptor.
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.
171 * Control commands are single-byte characters:
174 * 'A' - pick up new file descriptors to monitor
177 * 'R' - remove some descriptors from monitoring.
181 * For details, see _kqueue_thread_collect_fds() and
182 * _kqueue_thread_cleanup_fds().
184 * Notifications, that thread writes into the command file descriptor,
185 * are represented with #kqueue_notification objects.
190 _kqueue_thread_func (void *arg)
192 int fd, kqueue_descriptor;
195 g_assert (arg != NULL);
196 kevents_init_sz (&waiting, 1);
200 kqueue_descriptor = get_kqueue_descriptor();
201 if (kqueue_descriptor == -1)
203 KT_W ("fatal: kqueue is not initialized!\n");
207 EV_SET (&waiting.memory[0],
210 EV_ADD | EV_ENABLE | EV_ONESHOT,
218 /* TODO: Provide more items in the 'eventlist' to kqueue(2).
219 * Currently the backend takes notifications from the kernel one
220 * by one, i.e. there will be a lot of system calls and context
221 * switches when the application will monitor a lot of files with
222 * high filesystem activity on each. */
224 struct kevent received;
225 KT_W ("Watching for %zi items", waiting.kq_size);
226 int ret = kevent (kqueue_descriptor, waiting.memory, waiting.kq_size, &received, 1, NULL);
227 int kevent_errno = errno;
232 KT_W ("kevent failed: %d", kevent_errno);
233 if (kevent_errno == EINTR)
239 if (received.ident == fd)
242 if (!_ku_read (fd, &c, 1))
244 KT_W ("Failed to read command, error %d", errno);
248 _kqueue_thread_collect_fds (&waiting);
250 _kqueue_thread_cleanup_fds (&waiting);
254 struct kqueue_notification kn;
255 kn.fd = received.ident;
257 if (received.flags & EV_ERROR)
259 kn.flags = NOTE_REVOKE;
260 _kqueue_thread_drop_fd (&waiting, received.ident);
263 kn.flags = (received.fflags & ~NOTE_REVOKE);
265 if (!_ku_write (fd, &kn, sizeof (struct kqueue_notification)))
266 KT_W ("Failed to write a kqueue notification, error %d", errno);
269 kevents_free (&waiting);
275 * _kqueue_thread_push_fd:
276 * @fd: a file descriptor
278 * Puts a new file descriptor into the pick up list for monitroing.
280 * The kqueue thread will not start monitoring on it immediately, it
281 * should be bumped via its command file descriptor manually.
282 * See kqueue_thread() and _kqueue_thread_collect_fds() for details.
285 _kqueue_thread_push_fd (int fd)
287 G_LOCK (pick_up_lock);
288 g_queue_push_tail (&pick_up_fds_queue, GINT_TO_POINTER (fd));
289 G_UNLOCK (pick_up_lock);
294 * _kqueue_thread_remove_fd:
295 * @fd: a file descriptor
297 * Puts a new file descriptor into the remove list to cancel monitoring
300 * The kqueue thread will not stop monitoring on it immediately, it
301 * should be bumped via its command file descriptor manually.
302 * See kqueue_thread() and _kqueue_thread_collect_fds() for details.
305 _kqueue_thread_remove_fd (int fd)
307 G_LOCK (remove_lock);
308 remove_fds_list = g_slist_prepend (remove_fds_list, GINT_TO_POINTER (fd));
309 G_UNLOCK (remove_lock);