lib/events/events_aio.c

   1 /*
   2    Unix SMB/CIFS implementation.
   3
   4    main select loop and event handling - aio/epoll hybrid implementation
   5
   6    Copyright (C) Andrew Tridgell        2006
   7
   8    based on events_standard.c
   9
  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.
  14
  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.
  19
  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/>.
  22 */
  23 /*
  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
  29
  30   this is _very_ experimental code
  31 */
  32
  33 #include "includes.h"
  34 #include "system/filesys.h"
  35 #include "system/network.h"
  36 #include "lib/util/dlinklist.h"
  37 #include "lib/events/events.h"
  38 #include "lib/events/events_internal.h"
  39 #include <sys/epoll.h>
  40 #include <libaio.h>
  41
  42 #define MAX_AIO_QUEUE_DEPTH     100
  43 #ifndef IOCB_CMD_EPOLL_WAIT
  44 #define IOCB_CMD_EPOLL_WAIT     9
  45 #endif
  46
  47 struct aio_event_context {
  48         /* a pointer back to the generic event_context */
  49         struct event_context *ev;
  50
  51         /* list of filedescriptor events */
  52         struct fd_event *fd_events;
  53
  54         /* number of registered fd event handlers */
  55         int num_fd_events;
  56
  57         uint32_t destruction_count;
  58
  59         io_context_t ioctx;
  60
  61         struct epoll_event epevent[MAX_AIO_QUEUE_DEPTH];
  62
  63         struct iocb *epoll_iocb;
  64
  65         int epoll_fd;
  66         int is_epoll_set;
  67         pid_t pid;
  68 };
  69
  70 struct aio_event {
  71         struct event_context *event_ctx;
  72         struct iocb iocb;
  73         void *private_data;
  74         event_aio_handler_t handler;
  75 };
  76
  77 /*
  78   map from EVENT_FD_* to EPOLLIN/EPOLLOUT
  79 */
  80 static uint32_t epoll_map_flags(uint16_t flags)
  81 {
  82         uint32_t ret = 0;
  83         if (flags & EVENT_FD_READ) ret |= (EPOLLIN | EPOLLERR | EPOLLHUP);
  84         if (flags & EVENT_FD_WRITE) ret |= (EPOLLOUT | EPOLLERR | EPOLLHUP);
  85         return ret;
  86 }
  87
  88 /*
  89  free the epoll fd
  90 */
  91 static int aio_ctx_destructor(struct aio_event_context *aio_ev)
  92 {
  93         io_queue_release(aio_ev->ioctx);
  94         close(aio_ev->epoll_fd);
  95         aio_ev->epoll_fd = -1;
  96         return 0;
  97 }
  98
  99 static void epoll_add_event(struct aio_event_context *aio_ev, struct fd_event *fde);
 100
 101 /*
 102   reopen the epoll handle when our pid changes
 103   see http://junkcode.samba.org/ftp/unpacked/junkcode/epoll_fork.c for an
 104   demonstration of why this is needed
 105  */
 106 static void epoll_check_reopen(struct aio_event_context *aio_ev)
 107 {
 108         struct fd_event *fde;
 109
 110         if (aio_ev->pid == getpid()) {
 111                 return;
 112         }
 113
 114         close(aio_ev->epoll_fd);
 115         aio_ev->epoll_fd = epoll_create(MAX_AIO_QUEUE_DEPTH);
 116         if (aio_ev->epoll_fd == -1) {
 117                 DEBUG(0,("Failed to recreate epoll handle after fork\n"));
 118                 return;
 119         }
 120         aio_ev->pid = getpid();
 121         for (fde=aio_ev->fd_events;fde;fde=fde->next) {
 122                 epoll_add_event(aio_ev, fde);
 123         }
 124 }
 125
 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)
 129
 130 /*
 131  add the epoll event to the given fd_event
 132 */
 133 static void epoll_add_event(struct aio_event_context *aio_ev, struct fd_event *fde)
 134 {
 135         struct epoll_event event;
 136         if (aio_ev->epoll_fd == -1) return;
 137
 138         fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
 139
 140         /* if we don't want events yet, don't add an aio_event */
 141         if (fde->flags == 0) return;
 142
 143         ZERO_STRUCT(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;
 148
 149         /* only if we want to read we want to tell the event handler about errors */
 150         if (fde->flags & EVENT_FD_READ) {
 151                 fde->additional_flags |= EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
 152         }
 153 }
 154
 155 /*
 156  delete the epoll event for given fd_event
 157 */
 158 static void epoll_del_event(struct aio_event_context *aio_ev, struct fd_event *fde)
 159 {
 160         struct epoll_event event;
 161
 162         DLIST_REMOVE(aio_ev->fd_events, fde);
 163
 164         if (aio_ev->epoll_fd == -1) return;
 165
 166         fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
 167
 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;
 170
 171         ZERO_STRUCT(event);
 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);
 175
 176         fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_HAS_EVENT;
 177 }
 178
 179 /*
 180  change the epoll event to the given fd_event
 181 */
 182 static void epoll_mod_event(struct aio_event_context *aio_ev, struct fd_event *fde)
 183 {
 184         struct epoll_event event;
 185         if (aio_ev->epoll_fd == -1) return;
 186
 187         fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
 188
 189         ZERO_STRUCT(event);
 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);
 193
 194         /* only if we want to read we want to tell the event handler about errors */
 195         if (fde->flags & EVENT_FD_READ) {
 196                 fde->additional_flags |= EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
 197         }
 198 }
 199
 200 static void epoll_change_event(struct aio_event_context *aio_ev, struct fd_event *fde)
 201 {
 202         bool got_error = (fde->additional_flags & EPOLL_ADDITIONAL_FD_FLAG_GOT_ERROR);
 203         bool want_read = (fde->flags & EVENT_FD_READ);
 204         bool want_write= (fde->flags & EVENT_FD_WRITE);
 205
 206         if (aio_ev->epoll_fd == -1) return;
 207
 208         fde->additional_flags &= ~EPOLL_ADDITIONAL_FD_FLAG_REPORT_ERROR;
 209
 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);
 214                         return;
 215                 }
 216                 epoll_del_event(aio_ev, fde);
 217                 return;
 218         }
 219
 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);
 224                 return;
 225         }
 226 }
 227
 228 static int setup_epoll_wait(struct aio_event_context *aio_ev)
 229 {
 230         if (aio_ev->is_epoll_set) {
 231                 return 0;
 232         }
 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;
 237
 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;
 241
 242         if (io_submit(aio_ev->ioctx, 1, &aio_ev->epoll_iocb) != 1) {
 243                 return -1;
 244         }
 245         aio_ev->is_epoll_set = 1;
 246
 247         return 0;
 248 }
 249
 250
 251 /*
 252   event loop handling using aio/epoll hybrid
 253 */
 254 static int aio_event_loop(struct aio_event_context *aio_ev, struct timeval *tvalp)
 255 {
 256         int ret, i;
 257         uint32_t destruction_count = ++aio_ev->destruction_count;
 258         struct timespec timeout;
 259         struct io_event events[8];
 260
 261         if (aio_ev->epoll_fd == -1) return -1;
 262
 263         if (aio_ev->ev->num_signal_handlers &&
 264             common_event_check_signal(aio_ev->ev)) {
 265                 return 0;
 266         }
 267
 268         if (tvalp) {
 269                 timeout.tv_sec = tvalp->tv_sec;
 270                 timeout.tv_nsec = tvalp->tv_usec;
 271                 timeout.tv_nsec *= 1000;
 272         }
 273
 274         if (setup_epoll_wait(aio_ev) < 0)
 275                 return -1;
 276
 277         ret = io_getevents(aio_ev->ioctx, 1, 8,
 278                            events, tvalp?&timeout:NULL);
 279
 280         if (ret == -EINTR) {
 281                 if (aio_ev->ev->num_signal_handlers) {
 282                         common_event_check_signal(aio_ev->ev);
 283                 }
 284                 return 0;
 285         }
 286
 287         if (ret == 0 && tvalp) {
 288                 /* we don't care about a possible delay here */
 289                 common_event_loop_timer_delay(aio_ev->ev);
 290                 return 0;
 291         }
 292
 293         for (i=0;i<ret;i++) {
 294                 struct io_event *event = &events[i];
 295                 struct iocb *finished = event->obj;
 296
 297                 switch (finished->aio_lio_opcode) {
 298                 case IO_CMD_PWRITE:
 299                 case IO_CMD_PREAD: {
 300                         struct aio_event *ae = talloc_get_type(finished->data,
 301                                                                struct aio_event);
 302                         if (ae) {
 303                                 talloc_set_destructor(ae, NULL);
 304                                 ae->handler(ae->event_ctx, ae,
 305                                             event->res, ae->private_data);
 306                                 talloc_free(ae);
 307                         }
 308                         break;
 309                 }
 310                 case IOCB_CMD_EPOLL_WAIT: {
 311                         struct epoll_event *ep = (struct epoll_event *)finished->u.c.buf;
 312                         struct fd_event *fde;
 313                         uint16_t flags = 0;
 314                         int j;
 315
 316                         aio_ev->is_epoll_set = 0;
 317
 318                         for (j=0; j<event->res; j++, ep++) {
 319                                 fde = talloc_get_type(ep->data.ptr,
 320                                                       struct fd_event);
 321                                 if (fde == NULL) {
 322                                         return -1;
 323                                 }
 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);
 328                                                 continue;
 329                                         }
 330                                         flags |= EVENT_FD_READ;
 331                                 }
 332                                 if (ep->events & EPOLLIN) flags |= EVENT_FD_READ;
 333                                 if (ep->events & EPOLLOUT) flags |= EVENT_FD_WRITE;
 334                                 if (flags) {
 335                                         fde->handler(aio_ev->ev, fde, flags, fde->private_data);
 336                                 }
 337                         }
 338                         break;
 339                 }
 340                 }
 341                 if (destruction_count != aio_ev->destruction_count) {
 342                         return 0;
 343                 }
 344         }
 345
 346         return 0;
 347 }
 348
 349 /*
 350   create a aio_event_context structure.
 351 */
 352 static int aio_event_context_init(struct event_context *ev)
 353 {
 354         struct aio_event_context *aio_ev;
 355
 356         aio_ev = talloc_zero(ev, struct aio_event_context);
 357         if (!aio_ev) return -1;
 358
 359         aio_ev->ev = ev;
 360         aio_ev->epoll_iocb = talloc(aio_ev, struct iocb);
 361
 362         if (io_queue_init(MAX_AIO_QUEUE_DEPTH, &aio_ev->ioctx) != 0) {
 363                 talloc_free(aio_ev);
 364                 return -1;
 365         }
 366
 367         aio_ev->epoll_fd = epoll_create(MAX_AIO_QUEUE_DEPTH);
 368         if (aio_ev->epoll_fd == -1) {
 369                 talloc_free(aio_ev);
 370                 return -1;
 371         }
 372         aio_ev->pid = getpid();
 373
 374         talloc_set_destructor(aio_ev, aio_ctx_destructor);
 375
 376         ev->additional_data = aio_ev;
 377
 378         if (setup_epoll_wait(aio_ev) < 0) {
 379                 talloc_free(aio_ev);
 380                 return -1;
 381         }
 382
 383         return 0;
 384 }
 385
 386 /*
 387   destroy an fd_event
 388 */
 389 static int aio_event_fd_destructor(struct fd_event *fde)
 390 {
 391         struct event_context *ev = fde->event_ctx;
 392         struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
 393                                                            struct aio_event_context);
 394
 395         epoll_check_reopen(aio_ev);
 396
 397         aio_ev->num_fd_events--;
 398         aio_ev->destruction_count++;
 399
 400         epoll_del_event(aio_ev, fde);
 401
 402         if (fde->flags & EVENT_FD_AUTOCLOSE) {
 403                 close(fde->fd);
 404                 fde->fd = -1;
 405         }
 406
 407         return 0;
 408 }
 409
 410 /*
 411   add a fd based event
 412   return NULL on failure (memory allocation error)
 413 */
 414 static struct fd_event *aio_event_add_fd(struct event_context *ev, TALLOC_CTX *mem_ctx,
 415                                          int fd, uint16_t flags,
 416                                          event_fd_handler_t handler,
 417                                          void *private_data)
 418 {
 419         struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
 420                                                            struct aio_event_context);
 421         struct fd_event *fde;
 422
 423         epoll_check_reopen(aio_ev);
 424
 425         fde = talloc(mem_ctx?mem_ctx:ev, struct fd_event);
 426         if (!fde) return NULL;
 427
 428         fde->event_ctx          = ev;
 429         fde->fd                 = fd;
 430         fde->flags              = flags;
 431         fde->handler            = handler;
 432         fde->private_data       = private_data;
 433         fde->additional_flags   = 0;
 434         fde->additional_data    = NULL;
 435
 436         aio_ev->num_fd_events++;
 437         talloc_set_destructor(fde, aio_event_fd_destructor);
 438
 439         DLIST_ADD(aio_ev->fd_events, fde);
 440         epoll_add_event(aio_ev, fde);
 441
 442         return fde;
 443 }
 444
 445
 446 /*
 447   return the fd event flags
 448 */
 449 static uint16_t aio_event_get_fd_flags(struct fd_event *fde)
 450 {
 451         return fde->flags;
 452 }
 453
 454 /*
 455   set the fd event flags
 456 */
 457 static void aio_event_set_fd_flags(struct fd_event *fde, uint16_t flags)
 458 {
 459         struct event_context *ev;
 460         struct aio_event_context *aio_ev;
 461
 462         if (fde->flags == flags) return;
 463
 464         ev = fde->event_ctx;
 465         aio_ev = talloc_get_type(ev->additional_data, struct aio_event_context);
 466
 467         fde->flags = flags;
 468
 469         epoll_check_reopen(aio_ev);
 470
 471         epoll_change_event(aio_ev, fde);
 472 }
 473
 474 /*
 475   do a single event loop using the events defined in ev
 476 */
 477 static int aio_event_loop_once(struct event_context *ev)
 478 {
 479         struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
 480                                                            struct aio_event_context);
 481         struct timeval tval;
 482
 483         tval = common_event_loop_timer_delay(ev);
 484         if (timeval_is_zero(&tval)) {
 485                 return 0;
 486         }
 487
 488         epoll_check_reopen(aio_ev);
 489
 490         return aio_event_loop(aio_ev, &tval);
 491 }
 492
 493 /*
 494   return on failure or (with 0) if all fd events are removed
 495 */
 496 static int aio_event_loop_wait(struct event_context *ev)
 497 {
 498         struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
 499                                                            struct aio_event_context);
 500         while (aio_ev->num_fd_events) {
 501                 if (aio_event_loop_once(ev) != 0) {
 502                         break;
 503                 }
 504         }
 505
 506         return 0;
 507 }
 508
 509 /*
 510   called when a disk IO event needs to be cancelled
 511 */
 512 static int aio_destructor(struct aio_event *ae)
 513 {
 514         struct event_context *ev = ae->event_ctx;
 515         struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
 516                                                            struct aio_event_context);
 517         struct io_event result;
 518         io_cancel(aio_ev->ioctx, &ae->iocb, &result);
 519         /* TODO: handle errors from io_cancel()! */
 520         return 0;
 521 }
 522
 523 /* submit an aio disk IO event */
 524 static struct aio_event *aio_event_add_aio(struct event_context *ev,
 525                                            TALLOC_CTX *mem_ctx,
 526                                            struct iocb *iocb,
 527                                            event_aio_handler_t handler,
 528                                            void *private_data)
 529 {
 530         struct aio_event_context *aio_ev = talloc_get_type(ev->additional_data,
 531                                                            struct aio_event_context);
 532         struct iocb *iocbp;
 533         struct aio_event *ae = talloc(mem_ctx?mem_ctx:ev, struct aio_event);
 534         if (ae == NULL) return NULL;
 535
 536         ae->event_ctx    = ev;
 537         ae->iocb         = *iocb;
 538         ae->handler      = handler;
 539         ae->private_data = private_data;
 540         iocbp = &ae->iocb;
 541
 542         if (io_submit(aio_ev->ioctx, 1, &iocbp) != 1) {
 543                 talloc_free(ae);
 544                 return NULL;
 545         }
 546         ae->iocb.data = ae;
 547         talloc_set_destructor(ae, aio_destructor);
 548
 549         return ae;
 550 }
 551
 552 static const struct event_ops aio_event_ops = {
 553         .context_init   = aio_event_context_init,
 554         .add_fd         = aio_event_add_fd,
 555         .add_aio        = aio_event_add_aio,
 556         .get_fd_flags   = aio_event_get_fd_flags,
 557         .set_fd_flags   = aio_event_set_fd_flags,
 558         .add_timed      = common_event_add_timed,
 559         .add_signal     = common_event_add_signal,
 560         .loop_once      = aio_event_loop_once,
 561         .loop_wait      = aio_event_loop_wait,
 562 };
 563
 564 bool events_aio_init(void)
 565 {
 566         return event_register_backend("aio", &aio_event_ops);
 567 }
 568
 569 #if _SAMBA_BUILD_
 570 NTSTATUS s4_events_aio_init(void)
 571 {
 572         if (!events_aio_init()) {
 573                 return NT_STATUS_INTERNAL_ERROR;
 574         }
 575         return NT_STATUS_OK;
 576 }
 577 #endif