2 Unix SMB/CIFS implementation.
4 main select loop and event handling - aio/epoll hybrid implementation
6 Copyright (C) Andrew Tridgell 2006
8 based on events_standard.c
10 This program is free software; you can redistribute it and/or modify
11 it under the terms of the GNU General Public License as published by
12 the Free Software Foundation; either version 3 of the License, or
13 (at your option) any later version.
15 This program is distributed in the hope that it will be useful,
16 but WITHOUT ANY WARRANTY; without even the implied warranty of
17 MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
18 GNU General Public License for more details.
20 You should have received a copy of the GNU General Public License
21 along with this program. If not, see <http://www.gnu.org/licenses/>.
24 this is a very strange beast. The Linux AIO implementation doesn't
25 yet integrate properly with epoll, but there is a kernel patch that
26 allows the aio wait primitives to be used to wait for epoll events,
27 and this can be used to give us a unified event system incorporating
28 both aio events and epoll events
30 this is _very_ experimental code
34 #include "system/filesys.h"
35 #include "system/select.h"
37 #include "tevent_internal.h"
38 #include "tevent_util.h"
41 #define MAX_AIO_QUEUE_DEPTH 100
42 #ifndef IOCB_CMD_EPOLL_WAIT
43 #define IOCB_CMD_EPOLL_WAIT 9
46 struct aio_event_context {
47 /* a pointer back to the generic event_context */
48 struct tevent_context *ev;
50 /* list of filedescriptor events */
51 struct tevent_fd *fd_events;
53 /* number of registered fd event handlers */
56 uint32_t destruction_count;
60 struct epoll_event epevent[MAX_AIO_QUEUE_DEPTH];
62 struct iocb *epoll_iocb;
70 struct tevent_context *event_ctx;
73 tevent_aio_handler_t handler;
77 map from EVENT_FD_* to EPOLLIN/EPOLLOUT
79 static uint32_t epoll_map_flags(uint16_t flags)
82 if (flags & TEVENT_FD_READ) ret |= (EPOLLIN | EPOLLERR | EPOLLHUP);
83 if (flags & TEVENT_FD_WRITE) ret |= (EPOLLOUT | EPOLLERR | EPOLLHUP);
90 static int aio_ctx_destructor(struct aio_event_context *aio_ev)
92 io_queue_release(aio_ev->ioctx);
93 close(aio_ev->epoll_fd);
94 aio_ev->epoll_fd = -1;
98 static void epoll_add_event(struct aio_event_context *aio_ev, struct tevent_fd *fde);
101 reopen the epoll handle when our pid changes
102 see http://junkcode.samba.org/ftp/unpacked/junkcode/epoll_fork.c for an
103 demonstration of why this is needed
105 static void epoll_check_reopen(struct aio_event_context *aio_ev)
107 struct tevent_fd *fde;
109 if (aio_ev->pid == getpid()) {
113 close(aio_ev->epoll_fd);
114 aio_ev->epoll_fd = epoll_create(MAX_AIO_QUEUE_DEPTH);
115 if (aio_ev->epoll_fd == -1) {
116 tevent_debug(aio_ev->ev, TEVENT_DEBUG_FATAL,
117 "Failed to recreate epoll handle after fork\n");
120 aio_ev->pid = getpid();
121 for (fde=aio_ev->fd_events;fde;fde=fde->next) {
122 epoll_add_event(aio_ev, fde);
126 #define EPOLL_ADDITIONAL_FD_FLAG_HAS_EVENT (1<<0)
127 #define EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR (1<<1)
128 #define EPOLL_ADDITIONAL_FD_FLAG_GOT_ERROR (1<<2)
131 add the epoll event to the given fd_event
133 static void epoll_add_event(struct aio_event_context *aio_ev, struct tevent_fd *fde)
135 struct epoll_event event;
136 if (aio_ev->epoll_fd == -1) return;
138 fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
140 /* if we don't want events yet, don't add an aio_event */
141 if (fde->flags == 0) return;
143 memset(&event, 0, sizeof(event));
144 event.events = epoll_map_flags(fde->flags);
145 event.data.ptr = fde;
146 epoll_ctl(aio_ev->epoll_fd, EPOLL_CTL_ADD, fde->fd, &event);
147 fde->additional_flags |= EPOLL_ADDITIONAL_FD_FLAG_HAS_EVENT;
149 /* only if we want to read we want to tell the event handler about errors */
150 if (fde->flags & TEVENT_FD_READ) {
151 fde->additional_flags |= EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
156 delete the epoll event for given fd_event
158 static void epoll_del_event(struct aio_event_context *aio_ev, struct tevent_fd *fde)
160 struct epoll_event event;
162 DLIST_REMOVE(aio_ev->fd_events, fde);
164 if (aio_ev->epoll_fd == -1) return;
166 fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
168 /* if there's no aio_event, we don't need to delete it */
169 if (!(fde->additional_flags & EPOLL_ADDITIONAL_FD_FLAG_HAS_EVENT)) return;
172 event.events = epoll_map_flags(fde->flags);
173 event.data.ptr = fde;
174 epoll_ctl(aio_ev->epoll_fd, EPOLL_CTL_DEL, fde->fd, &event);
176 fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_HAS_EVENT;
180 change the epoll event to the given fd_event
182 static void epoll_mod_event(struct aio_event_context *aio_ev, struct tevent_fd *fde)
184 struct epoll_event event;
185 if (aio_ev->epoll_fd == -1) return;
187 fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
190 event.events = epoll_map_flags(fde->flags);
191 event.data.ptr = fde;
192 epoll_ctl(aio_ev->epoll_fd, EPOLL_CTL_MOD, fde->fd, &event);
194 /* only if we want to read we want to tell the event handler about errors */
195 if (fde->flags & TEVENT_FD_READ) {
196 fde->additional_flags |= EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
200 static void epoll_change_event(struct aio_event_context *aio_ev, struct tevent_fd *fde)
202 bool got_error = (fde->additional_flags & EPOLL_ADDITIONAL_FD_FLAG_GOT_ERROR);
203 bool want_read = (fde->flags & TEVENT_FD_READ);
204 bool want_write= (fde->flags & TEVENT_FD_WRITE);
206 if (aio_ev->epoll_fd == -1) return;
208 fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
210 /* there's already an event */
211 if (fde->additional_flags & EPOLL_ADDITIONAL_FD_FLAG_HAS_EVENT) {
212 if (want_read || (want_write && !got_error)) {
213 epoll_mod_event(aio_ev, fde);
216 epoll_del_event(aio_ev, fde);
220 /* there's no aio_event attached to the fde */
221 if (want_read || (want_write && !got_error)) {
222 DLIST_ADD(aio_ev->fd_events, fde);
223 epoll_add_event(aio_ev, fde);
228 static int setup_epoll_wait(struct aio_event_context *aio_ev)
230 if (aio_ev->is_epoll_set) {
233 memset(aio_ev->epoll_iocb, 0, sizeof(*aio_ev->epoll_iocb));
234 aio_ev->epoll_iocb->aio_fildes = aio_ev->epoll_fd;
235 aio_ev->epoll_iocb->aio_lio_opcode = IOCB_CMD_EPOLL_WAIT;
236 aio_ev->epoll_iocb->aio_reqprio = 0;
238 aio_ev->epoll_iocb->u.c.nbytes = MAX_AIO_QUEUE_DEPTH;
239 aio_ev->epoll_iocb->u.c.offset = -1;
240 aio_ev->epoll_iocb->u.c.buf = aio_ev->epevent;
242 if (io_submit(aio_ev->ioctx, 1, &aio_ev->epoll_iocb) != 1) {
245 aio_ev->is_epoll_set = 1;
252 event loop handling using aio/epoll hybrid
254 static int aio_event_loop(struct aio_event_context *aio_ev, struct timeval *tvalp)
257 uint32_t destruction_count = ++aio_ev->destruction_count;
258 struct timespec timeout;
259 struct io_event events[8];
261 if (aio_ev->epoll_fd == -1) return -1;
263 if (aio_ev->ev->signal_events &&
264 tevent_common_check_signal(aio_ev->ev)) {
269 timeout.tv_sec = tvalp->tv_sec;
270 timeout.tv_nsec = tvalp->tv_usec;
271 timeout.tv_nsec *= 1000;
274 if (setup_epoll_wait(aio_ev) < 0)
277 ret = io_getevents(aio_ev->ioctx, 1, 8,
278 events, tvalp?&timeout:NULL);
281 if (aio_ev->ev->signal_events) {
282 tevent_common_check_signal(aio_ev->ev);
287 if (ret == 0 && tvalp) {
288 /* we don't care about a possible delay here */
289 tevent_common_loop_timer_delay(aio_ev->ev);
293 for (i=0;i<ret;i++) {
294 struct io_event *event = &events[i];
295 struct iocb *finished = event->obj;
297 switch (finished->aio_lio_opcode) {
300 struct tevent_aio *ae = talloc_get_type(finished->data,
303 talloc_set_destructor(ae, NULL);
304 ae->handler(ae->event_ctx, ae,
305 event->res, ae->private_data);
310 case IOCB_CMD_EPOLL_WAIT: {
311 struct epoll_event *ep = (struct epoll_event *)finished->u.c.buf;
312 struct tevent_fd *fde;
316 aio_ev->is_epoll_set = 0;
318 for (j=0; j<event->res; j++, ep++) {
319 fde = talloc_get_type(ep->data.ptr,
324 if (ep->events & (EPOLLHUP|EPOLLERR)) {
325 fde->additional_flags |= EPOLL_ADDITIONAL_FD_FLAG_GOT_ERROR;
326 if (!(fde->additional_flags & EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR)) {
327 epoll_del_event(aio_ev, fde);
330 flags |= TEVENT_FD_READ;
332 if (ep->events & EPOLLIN) flags |= TEVENT_FD_READ;
333 if (ep->events & EPOLLOUT) flags |= TEVENT_FD_WRITE;
335 fde->handler(aio_ev->ev, fde, flags, fde->private_data);
341 if (destruction_count != aio_ev->destruction_count) {
350 create a aio_event_context structure.
352 static int aio_event_context_init(struct tevent_context *ev)
354 struct aio_event_context *aio_ev;
356 aio_ev = talloc_zero(ev, struct aio_event_context);
357 if (!aio_ev) return -1;
360 aio_ev->epoll_iocb = talloc(aio_ev, struct iocb);
362 if (io_queue_init(MAX_AIO_QUEUE_DEPTH, &aio_ev->ioctx) != 0) {
367 aio_ev->epoll_fd = epoll_create(MAX_AIO_QUEUE_DEPTH);
368 if (aio_ev->epoll_fd == -1) {
372 aio_ev->pid = getpid();
374 talloc_set_destructor(aio_ev, aio_ctx_destructor);
376 ev->additional_data = aio_ev;
378 if (setup_epoll_wait(aio_ev) < 0) {
389 static int aio_event_fd_destructor(struct tevent_fd *fde)
391 struct tevent_context *ev = fde->event_ctx;
392 struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
393 struct aio_event_context);
395 epoll_check_reopen(aio_ev);
397 aio_ev->num_fd_events--;
398 aio_ev->destruction_count++;
400 epoll_del_event(aio_ev, fde);
403 fde->close_fn(ev, fde, fde->fd, fde->private_data);
412 return NULL on failure (memory allocation error)
414 static struct tevent_fd *aio_event_add_fd(struct tevent_context *ev, TALLOC_CTX *mem_ctx,
415 int fd, uint16_t flags,
416 tevent_fd_handler_t handler,
418 const char *handler_name,
419 const char *location)
421 struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
422 struct aio_event_context);
423 struct tevent_fd *fde;
425 epoll_check_reopen(aio_ev);
427 fde = talloc(mem_ctx?mem_ctx:ev, struct tevent_fd);
428 if (!fde) return NULL;
433 fde->handler = handler;
434 fde->close_fn = NULL;
435 fde->private_data = private_data;
436 fde->handler_name = handler_name;
437 fde->location = location;
438 fde->additional_flags = 0;
439 fde->additional_data = NULL;
441 aio_ev->num_fd_events++;
442 talloc_set_destructor(fde, aio_event_fd_destructor);
444 DLIST_ADD(aio_ev->fd_events, fde);
445 epoll_add_event(aio_ev, fde);
451 set the fd event flags
453 static void aio_event_set_fd_flags(struct tevent_fd *fde, uint16_t flags)
455 struct tevent_context *ev;
456 struct aio_event_context *aio_ev;
458 if (fde->flags == flags) return;
461 aio_ev = talloc_get_type(ev->additional_data, struct aio_event_context);
465 epoll_check_reopen(aio_ev);
467 epoll_change_event(aio_ev, fde);
471 do a single event loop using the events defined in ev
473 static int aio_event_loop_once(struct tevent_context *ev)
475 struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
476 struct aio_event_context);
479 tval = tevent_common_loop_timer_delay(ev);
480 if (ev_timeval_is_zero(&tval)) {
484 epoll_check_reopen(aio_ev);
486 return aio_event_loop(aio_ev, &tval);
490 return on failure or (with 0) if all fd events are removed
492 static int aio_event_loop_wait(struct tevent_context *ev)
494 struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
495 struct aio_event_context);
496 while (aio_ev->num_fd_events) {
497 if (aio_event_loop_once(ev) != 0) {
506 called when a disk IO event needs to be cancelled
508 static int aio_destructor(struct tevent_aio *ae)
510 struct tevent_context *ev = ae->event_ctx;
511 struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
512 struct aio_event_context);
513 struct io_event result;
514 io_cancel(aio_ev->ioctx, &ae->iocb, &result);
515 /* TODO: handle errors from io_cancel()! */
519 /* submit an aio disk IO event */
520 static struct tevent_aio *aio_event_add_aio(struct tevent_context *ev,
523 tevent_aio_handler_t handler,
525 const char *handler_name,
526 const char *location)
528 struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
529 struct aio_event_context);
531 struct tevent_aio *ae = talloc(mem_ctx?mem_ctx:ev, struct tevent_aio);
532 if (ae == NULL) return NULL;
536 ae->handler = handler;
537 ae->private_data = private_data;
540 if (io_submit(aio_ev->ioctx, 1, &iocbp) != 1) {
545 talloc_set_destructor(ae, aio_destructor);
550 static const struct tevent_ops aio_event_ops = {
551 .context_init = aio_event_context_init,
552 .add_fd = aio_event_add_fd,
553 .add_aio = aio_event_add_aio,
554 .set_fd_close_fn= tevent_common_fd_set_close_fn,
555 .get_fd_flags = tevent_common_fd_get_flags,
556 .set_fd_flags = aio_event_set_fd_flags,
557 .add_timer = tevent_common_add_timer,
558 .add_signal = tevent_common_add_signal,
559 .loop_once = aio_event_loop_once,
560 .loop_wait = aio_event_loop_wait,
563 bool tevent_aio_init(void)
565 return tevent_register_backend("aio", &aio_event_ops);