1 /*
2 * This file is part of the Sofia-SIP package
3 *
4 * Copyright (C) 2005, 2006, 2007 Nokia Corporation.
5 *
6 * Contact: Pekka Pessi <pekka.pessi@nokia.com>
7 *
8 * This library is free software; you can redistribute it and/or
9 * modify it under the terms of the GNU Lesser General Public License
10 * as published by the Free Software Foundation; either version 2.1 of
11 * the License, or (at your option) any later version.
12 *
13 * This library is distributed in the hope that it will be useful, but
14 * WITHOUT ANY WARRANTY; without even the implied warranty of
15 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
16 * Lesser General Public License for more details.
17 *
18 * You should have received a copy of the GNU Lesser General Public
19 * License along with this library; if not, write to the Free Software
20 * Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
21 * 02110-1301 USA
22 *
23 */
24
25 /**@ingroup su_wait
26 * @CFILE su_epoll_port.c
27 *
28 * Port implementation using epoll(7)
29 *
30 * @author Pekka Pessi <Pekka.Pessi@nokia.com>
31 * @author Kai Vehmanen <kai.vehmanen@nokia.com>
32 *
33 * @date Created: Fri Jan 26 20:44:14 2007 ppessi
34 * @date Original: Tue Sep 14 15:51:04 1999 ppessi
35 */
36
37 #include "config.h"
38
39 #define su_port_s su_epoll_port_s
40
41 #include "su_port.h"
42
43 #if HAVE_EPOLL
44
45 #include "sofia-sip/su.h"
46 #include "sofia-sip/su_alloc.h"
47
48 #include <stdlib.h>
49 #include <assert.h>
50 #include <stdarg.h>
51 #include <stdio.h>
52 #include <string.h>
53 #include <limits.h>
54 #include <errno.h>
55
56 #include <sys/epoll.h>
57
58 #define POLL2EPOLL_NEEDED \
59 (POLLIN != EPOLLIN || POLLOUT != EPOLLOUT || POLLPRI != EPOLLPRI || \
60 POLLERR != EPOLLERR || POLLHUP != EPOLLHUP)
61
62 #define POLL2EPOLL(e) (e & (POLLIN|POLLOUT|POLLPRI|POLLERR|POLLHUP))
63 #define EPOLL2POLL(e) (e & (POLLIN|POLLOUT|POLLPRI|POLLERR|POLLHUP))
64
65 /** Port based on epoll(). */
66
67 struct su_epoll_port_s {
68 su_socket_port_t sup_base[1];
69
70 /** epoll fd */
71 int sup_epoll;
72 unsigned sup_multishot; /**< Multishot operation? */
73
74 unsigned sup_registers; /** Counter incremented by
75 su_port_register() or
76 su_port_unregister()
77 */
78 int sup_n_registrations;
79 int sup_max_index; /**< Indexes are equal or smaller than this */
80 int sup_size_indices; /**< Size of allocated index table */
81
82 /** Structure containing registration data */
83 struct su_epoll_register {
84 struct su_epoll_register *ser_next; /* Next in free list */
85 su_wakeup_f ser_cb;
86 su_wakeup_arg_t*ser_arg;
87 su_root_t *ser_root;
88 int ser_id; /** registration identifier */
89 su_wait_t ser_wait[1];
90 } **sup_indices;
91 };
92
93 static void su_epoll_port_decref(su_port_t *self,
94 int blocking,
95 char const *who);
96 static int su_epoll_port_register(su_port_t *self,
97 su_root_t *root,
98 su_wait_t *wait,
99 su_wakeup_f callback,
100 su_wakeup_arg_t *arg,
101 int priority);
102 static int su_epoll_port_unregister(su_port_t *port,
103 su_root_t *root,
104 su_wait_t *wait,
105 su_wakeup_f callback,
106 su_wakeup_arg_t *arg);
107 static int su_epoll_port_deregister(su_port_t *self, int i);
108 static int su_epoll_port_unregister_all(su_port_t *self, su_root_t *root);
109 static int su_epoll_port_eventmask(su_port_t *self,
110 int index,
111 int socket,
112 int events);
113 static int su_epoll_port_multishot(su_port_t *self, int multishot);
114 static int su_epoll_port_wait_events(su_port_t *self, su_duration_t tout);
115 static char const *su_epoll_port_name(su_port_t const *self);
116
117 su_port_vtable_t const su_epoll_port_vtable[1] =
118 {{
119 /* su_vtable_size: */ sizeof su_epoll_port_vtable,
120 su_pthread_port_lock,
121 su_pthread_port_unlock,
122 su_base_port_incref,
123 su_epoll_port_decref,
124 su_base_port_gsource,
125 su_base_port_send,
126 su_epoll_port_register,
127 su_epoll_port_unregister,
128 su_epoll_port_deregister,
129 su_epoll_port_unregister_all,
130 su_epoll_port_eventmask,
131 su_base_port_run,
132 su_base_port_break,
133 su_base_port_step,
134 su_pthread_port_thread,
135 su_base_port_add_prepoll,
136 su_base_port_remove_prepoll,
137 su_base_port_timers,
138 su_epoll_port_multishot,
139 su_epoll_port_wait_events,
140 su_base_port_getmsgs,
141 su_base_port_getmsgs_from,
142 su_epoll_port_name,
143 su_base_port_start_shared,
144 su_pthread_port_wait,
145 su_pthread_port_execute,
146 su_base_port_deferrable,
147 su_base_port_max_defer,
148 su_socket_port_wakeup,
149 su_base_port_is_running,
150 }};
151
su_epoll_port_name(su_port_t const * self)152 static char const *su_epoll_port_name(su_port_t const *self)
153 {
154 return "epoll";
155 }
156
su_epoll_port_decref(su_port_t * self,int blocking,char const * who)157 static void su_epoll_port_decref(su_port_t *self, int blocking, char const *who)
158 {
159 (void)su_base_port_decref(self, blocking, who);
160 }
161
su_epoll_port_deinit(void * arg)162 static void su_epoll_port_deinit(void *arg)
163 {
164 su_port_t *self = arg;
165
166 SU_DEBUG_9(("%s(%p) called\n", "su_epoll_port_deinit", (void* )self));
167
168 su_socket_port_deinit(self->sup_base);
169
170 close(self->sup_epoll), self->sup_epoll = -1;
171 }
172
173 /** @internal
174 *
175 * Register a #su_wait_t object. The wait object, a callback function and
176 * an argument pointer is stored in the port object. The callback function
177 * will be called when the wait object is signaled.
178 *
179 * Please note if identical wait objects are inserted, only first one is
180 * ever signalled.
181 *
182 * @param self pointer to port
183 * @param root pointer to root object
184 * @param waits pointer to wait object
185 * @param callback callback function pointer
186 * @param arg argument given to callback function when it is invoked
187 * @param priority relative priority of the wait object
188 * (0 is normal, 1 important, 2 realtime)
189 *
190 * @return
191 * Positive index of the wait object,
192 * or -1 upon an error.
193 */
su_epoll_port_register(su_port_t * self,su_root_t * root,su_wait_t * wait,su_wakeup_f callback,su_wakeup_arg_t * arg,int priority)194 int su_epoll_port_register(su_port_t *self,
195 su_root_t *root,
196 su_wait_t *wait,
197 su_wakeup_f callback,
198 su_wakeup_arg_t *arg,
199 int priority)
200 {
201 int i, j, n;
202 struct epoll_event ev;
203 struct su_epoll_register *ser;
204 struct su_epoll_register **indices = self->sup_indices;
205
206 assert(su_port_own_thread(self));
207
208 n = self->sup_size_indices;
209
210 if (n >= SU_WAIT_MAX)
211 return su_seterrno(ENOMEM);
212
213 ser = indices[0];
214
215 if (!ser) {
216 su_home_t *h = su_port_home(self);
217
218 i = self->sup_max_index, j = i == 0 ? 15 : i + 16;
219
220 if (j >= self->sup_size_indices) {
221 /* Reallocate index table */
222 n = n < 1024 ? 2 * n : n + 1024;
223 indices = su_realloc(h, indices, n * sizeof(indices[0]));
224 if (!indices)
225 return -1;
226 self->sup_indices = indices;
227 self->sup_size_indices = n;
228 }
229
230 /* Allocate registrations */
231 ser = su_zalloc(h, (j - i) * (sizeof *ser));
232 if (!ser)
233 return -1;
234
235 indices[0] = ser;
236
237 for (i++; i <= j; i++) {
238 ser->ser_id = i;
239 ser->ser_next = i < j ? ser + 1 : NULL;
240 indices[i] = ser++;
241 }
242
243 self->sup_max_index = j;
244
245 ser = indices[0];
246 }
247
248 i = ser->ser_id;
249
250 ev.events = POLL2EPOLL(wait->events);
251 ev.data.u64 = 0;
252 ev.data.u32 = (uint32_t)i;
253
254 if (epoll_ctl(self->sup_epoll, EPOLL_CTL_ADD, wait->fd, &ev) == -1) {
255 SU_DEBUG_0(("EPOLL_CTL_ADD(%u, %u) failed: %s\n",
256 wait->fd, ev.events, strerror(errno)));
257 return -1;
258 }
259
260 indices[0] = ser->ser_next;
261
262 ser->ser_next = NULL;
263 *ser->ser_wait = *wait;
264 ser->ser_cb = callback;
265 ser->ser_arg = arg;
266 ser->ser_root = root;
267
268 self->sup_registers++;
269 self->sup_n_registrations++;
270
271 return i; /* return index */
272 }
273
274 /** Deregister a su_wait_t object. */
su_epoll_port_deregister0(su_port_t * self,int i,int destroy_wait)275 static int su_epoll_port_deregister0(su_port_t *self, int i, int destroy_wait)
276 {
277 struct su_epoll_register **indices = self->sup_indices;
278 struct su_epoll_register *ser;
279
280 ser = self->sup_indices[i];
281 if (ser == NULL || ser->ser_cb == NULL) {
282 su_seterrno(ENOENT);
283 return -1;
284 }
285
286 assert(ser->ser_id == i);
287
288 if (epoll_ctl(self->sup_epoll, EPOLL_CTL_DEL, ser->ser_wait->fd, NULL) == -1) {
289 SU_DEBUG_1(("su_port(%p): EPOLL_CTL_DEL(%u): %s\n", (void *)self,
290 ser->ser_wait->fd, su_strerror(su_errno())));
291 }
292
293 if (destroy_wait)
294 su_wait_destroy(ser->ser_wait);
295
296 memset(ser, 0, sizeof *ser);
297 ser->ser_id = i;
298 ser->ser_next = indices[0], indices[0] = ser;
299
300 self->sup_n_registrations--;
301 self->sup_registers++;
302
303 return i;
304 }
305
306
307 /** Unregister a su_wait_t object.
308 *
309 * The function su_epoll_port_unregister() unregisters a su_wait_t object. The
310 * wait object, a callback function and a argument are removed from the
311 * port object.
312 *
313 * @param self - pointer to port object
314 * @param root - pointer to root object
315 * @param wait - pointer to wait object
316 * @param callback - callback function pointer (may be NULL)
317 * @param arg - argument given to callback function when it is invoked
318 * (may be NULL)
319 *
320 * @deprecated Use su_epoll_port_deregister() instead.
321 *
322 * @return Nonzero index of the wait object, or -1 upon an error.
323 */
su_epoll_port_unregister(su_port_t * self,su_root_t * root,su_wait_t * wait,su_wakeup_f callback,su_wakeup_arg_t * arg)324 int su_epoll_port_unregister(su_port_t *self,
325 su_root_t *root,
326 su_wait_t *wait,
327 su_wakeup_f callback, /* XXX - ignored */
328 su_wakeup_arg_t *arg)
329 {
330 int i, I;
331
332 struct su_epoll_register *ser;
333
334 assert(self);
335 assert(su_port_own_thread(self));
336
337 I = self->sup_max_index;
338
339 for (i = 1; i <= I; i++) {
340 ser = self->sup_indices[i];
341
342 if (ser->ser_cb &&
343 arg == ser->ser_arg &&
344 SU_WAIT_CMP(wait[0], ser->ser_wait[0]) == 0)
345 return su_epoll_port_deregister0(self, ser->ser_id, 0);
346 }
347
348 su_seterrno(ENOENT);
349
350 return -1;
351 }
352
353 /** Deregister a su_wait_t object.
354 *
355 * Deregisters a registration by index. The wait object, a callback
356 * function and a argument are removed from the port object. The wait
357 * object is destroyed.
358 *
359 * @param self - pointer to port object
360 * @param i - registration index
361 *
362 * @return Index of the wait object, or -1 upon an error.
363 */
su_epoll_port_deregister(su_port_t * self,int i)364 int su_epoll_port_deregister(su_port_t *self, int i)
365 {
366 struct su_epoll_register *ser;
367
368 if (i <= 0 || i > self->sup_max_index)
369 return su_seterrno(EBADF);
370
371 ser = self->sup_indices[i];
372 if (!ser->ser_cb)
373 return su_seterrno(EBADF);
374
375 return su_epoll_port_deregister0(self, i, 1);
376 }
377
378
379 /** @internal
380 * Unregister all su_wait_t objects of given su_root_t instance.
381 *
382 * The function su_epoll_port_unregister_all() unregisters all su_wait_t
383 * objects associated with given root object.
384 *
385 * @param self - pointer to port object
386 * @param root - pointer to root object
387 *
388 * @return Number of wait objects removed.
389 */
su_epoll_port_unregister_all(su_port_t * self,su_root_t * root)390 int su_epoll_port_unregister_all(su_port_t *self, su_root_t *root)
391 {
392 int i, I, n;
393
394 struct su_epoll_register *ser;
395
396 assert(self); assert(root);
397 assert(su_port_own_thread(self));
398
399 I = self->sup_max_index;
400
401 for (i = 1, n = 0; i <= I; i++) {
402 ser = self->sup_indices[i];
403 if (ser->ser_root != root)
404 continue;
405 su_epoll_port_deregister0(self, ser->ser_id, 0);
406 n++;
407 }
408
409 return n;
410 }
411
412 /**Set mask for a registered event. @internal
413 *
414 * The function su_epoll_port_eventmask() sets the mask describing events that can
415 * signal the registered callback.
416 *
417 * @param port pointer to port object
418 * @param index registration index
419 * @param socket socket
420 * @param events new event mask
421 *
422 * @retval 0 when successful,
423 * @retval -1 upon an error.
424 */
su_epoll_port_eventmask(su_port_t * self,int index,int socket,int events)425 int su_epoll_port_eventmask(su_port_t *self, int index, int socket, int events)
426 {
427 struct su_epoll_register *ser;
428 struct epoll_event ev;
429
430 if (index <= 0 || index > self->sup_max_index)
431 return su_seterrno(EBADF);
432
433 ser = self->sup_indices[index];
434 if (!ser->ser_cb)
435 return su_seterrno(EBADF);
436
437 ser->ser_wait->events = events;
438
439 ev.events = POLL2EPOLL(events);
440 ev.data.u64 = (uint64_t)0;
441 ev.data.u32 = (uint32_t)index;
442
443 if (epoll_ctl(self->sup_epoll, EPOLL_CTL_MOD, socket, &ev) == -1) {
444 SU_DEBUG_1(("su_port(%p): EPOLL_CTL_MOD(%u): %s\n", (void *)self,
445 socket, su_strerror(su_errno())));
446 return -1;
447 }
448
449 return 0;
450 }
451
452 /** @internal Enable multishot mode.
453 *
454 * Enables, disables or queries the multishot mode for the port. The
455 * multishot mode determines how the events are scheduled by port. If
456 * multishot mode is enabled, port serves all the sockets that have received
457 * network events. If it is disabled, only first socket event is served.
458 *
459 * @param self pointer to port object
460 * @param multishot multishot mode (0 => disables, 1 => enables, -1 => query)
461 *
462 * @retval 0 multishot mode is disabled
463 * @retval 1 multishot mode is enabled
464 * @retval -1 an error occurred
465 */
466 static
su_epoll_port_multishot(su_port_t * self,int multishot)467 int su_epoll_port_multishot(su_port_t *self, int multishot)
468 {
469 if (multishot < 0)
470 return self->sup_multishot;
471 else if (multishot == 0 || multishot == 1)
472 return self->sup_multishot = multishot;
473 else
474 return (errno = EINVAL), -1;
475 }
476
477
478 /** @internal
479 * Wait (poll()) for wait objects in port.
480 *
481 * @param self pointer to port
482 * @param tout timeout in milliseconds
483 *
484 * @return number of events handled
485 */
486 static
su_epoll_port_wait_events(su_port_t * self,su_duration_t tout)487 int su_epoll_port_wait_events(su_port_t *self, su_duration_t tout)
488 {
489 int j, n, events = 0, index;
490 unsigned version = self->sup_registers;
491
492 int const M = 4;
493 struct epoll_event ev[M];
494
495 n = epoll_wait(self->sup_epoll, ev, self->sup_multishot ? M : 1, tout);
496
497 assert(n <= M);
498
499 for (j = 0; j < n; j++) {
500 struct su_epoll_register *ser;
501 su_root_magic_t *magic;
502
503 index = (int)ev[j].data.u32;
504 if (!ev[j].events || index <= 0 || self->sup_max_index < index)
505 continue;
506 ser = self->sup_indices[index];
507
508 magic = ser->ser_root ? su_root_magic(ser->ser_root) : NULL;
509 ser->ser_wait->revents = ev[j].events;
510 ser->ser_cb(magic, ser->ser_wait, ser->ser_arg);
511 events++;
512 if (version != self->sup_registers)
513 /* Callback function used su_register()/su_deregister() */
514 return events;
515 }
516
517 return n;
518 }
519
520 /** Create a port using epoll() or poll().
521 */
su_epoll_port_create(void)522 su_port_t *su_epoll_port_create(void)
523 {
524 su_port_t *self;
525 int epoll = epoll_create(su_root_size_hint);
526
527 if (epoll == -1) {
528 /* Fallback to poll() */
529 SU_DEBUG_3(("%s(): epoll_create() => %u: %s\n",
530 "su_port_create", epoll, strerror(errno)));
531 return su_poll_port_create();
532 }
533
534 self = su_home_new(sizeof *self);
535 if (!self) {
536 close(epoll);
537 return self;
538 }
539
540 SU_DEBUG_9(("%s(%p): epoll_create() => %u: %s\n",
541 "su_port_create", (void *)self, self->sup_epoll, "OK"));
542
543 if (su_home_destructor(su_port_home(self), su_epoll_port_deinit) < 0 ||
544 !(self->sup_indices =
545 su_zalloc(su_port_home(self),
546 (sizeof self->sup_indices[0]) *
547 (self->sup_size_indices = 64)))) {
548 su_home_unref(su_port_home(self));
549 close(epoll);
550 return NULL;
551 }
552
553 self->sup_epoll = epoll;
554 self->sup_multishot = SU_ENABLE_MULTISHOT_POLL;
555
556 if (su_socket_port_init(self->sup_base, su_epoll_port_vtable) < 0) {
557 close(epoll);
558 return su_home_unref(su_port_home(self)), NULL;
559 }
560
561 return self;
562 }
563
su_epoll_clone_start(su_root_t * parent,su_clone_r return_clone,su_root_magic_t * magic,su_root_init_f init,su_root_deinit_f deinit)564 int su_epoll_clone_start(su_root_t *parent,
565 su_clone_r return_clone,
566 su_root_magic_t *magic,
567 su_root_init_f init,
568 su_root_deinit_f deinit)
569 {
570 return su_pthreaded_port_start(su_epoll_port_create,
571 parent, return_clone, magic, init, deinit);
572 }
573
574 #else
575
su_epoll_port_create(void)576 su_port_t *su_epoll_port_create(void)
577 {
578 return su_default_port_create();
579 }
580
su_epoll_clone_start(su_root_t * parent,su_clone_r return_clone,su_root_magic_t * magic,su_root_init_f init,su_root_deinit_f deinit)581 int su_epoll_clone_start(su_root_t *parent,
582 su_clone_r return_clone,
583 su_root_magic_t *magic,
584 su_root_init_f init,
585 su_root_deinit_f deinit)
586 {
587 return su_default_clone_start(parent, return_clone, magic, init, deinit);
588 }
589
590 #endif /* HAVE_EPOLL */
591