lib/silcutil/unix/silcunixschedule.c

   1 /*
   2
   3   silcunixschedule.c
   4
   5   Author: Pekka Riikonen <priikone@silcnet.org>
   6
   7   Copyright (C) 1998 - 2008 Pekka Riikonen
   8
   9   This program is free software; you can redistribute it and/or modify
  10   it under the terms of the GNU General Public License as published by
  11   the Free Software Foundation; version 2 of the License.
  12
  13   This program is distributed in the hope that it will be useful,
  14   but WITHOUT ANY WARRANTY; without even the implied warranty of
  15   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  16   GNU General Public License for more details.
  17
  18 */
  19 /* $Id$ */
  20
  21 #include "silc.h"
  22
  23 #if defined(HAVE_EPOLL_WAIT)
  24 #include <sys/epoll.h>
  25 #elif defined(HAVE_POLL) && defined(HAVE_SETRLIMIT) && defined(RLIMIT_NOFILE)
  26 #include <poll.h>
  27 #endif
  28
  29 const SilcScheduleOps schedule_ops;
  30
  31 /* Internal context. */
  32 typedef struct {
  33 #if defined(HAVE_EPOLL_WAIT)
  34   struct epoll_event *fds;
  35   SilcUInt32 fds_count;
  36   int epfd;
  37 #elif defined(HAVE_POLL) && defined(HAVE_SETRLIMIT) && defined(RLIMIT_NOFILE)
  38   struct rlimit nofile;
  39   struct pollfd *fds;
  40   SilcUInt32 fds_count;
  41 #endif /* HAVE_POLL && HAVE_SETRLIMIT && RLIMIT_NOFILE */
  42   void *app_context;
  43   int wakeup_pipe[2];
  44   SilcTask wakeup_task;
  45   sigset_t signals;
  46   sigset_t signals_blocked;
  47 } *SilcUnixScheduler;
  48
  49 typedef struct {
  50   SilcUInt32 sig;
  51   SilcTaskCallback callback;
  52   void *context;
  53   SilcBool call;
  54   SilcSchedule schedule;
  55 } SilcUnixSignal;
  56
  57 #define SIGNAL_COUNT 32
  58 SilcUnixSignal signal_call[SIGNAL_COUNT];
  59
  60 #if defined(HAVE_EPOLL_WAIT)
  61
  62 /* Linux's fast epoll system (level triggered) */
  63
  64 int silc_epoll(SilcSchedule schedule, void *context)
  65 {
  66   SilcUnixScheduler internal = context;
  67   SilcTaskFd task;
  68   struct epoll_event *fds = internal->fds;
  69   SilcUInt32 fds_count = internal->fds_count;
  70   int ret, i, timeout = -1;
  71
  72   /* Allocate larger fd table if needed */
  73   i = silc_hash_table_count(schedule->fd_queue);
  74   if (i > fds_count) {
  75     fds = silc_realloc(internal->fds, sizeof(*internal->fds) *
  76                        (fds_count + (i / 2)));
  77     if (silc_likely(fds)) {
  78       internal->fds = fds;
  79       internal->fds_count = fds_count = fds_count + (i / 2);
  80     }
  81   }
  82
  83   if (schedule->has_timeout)
  84     timeout = ((schedule->timeout.tv_sec * 1000) +
  85                (schedule->timeout.tv_usec / 1000));
  86
  87   SILC_SCHEDULE_UNLOCK(schedule);
  88   ret = epoll_wait(internal->epfd, fds, fds_count, timeout);
  89   SILC_SCHEDULE_LOCK(schedule);
  90   if (ret <= 0)
  91     return ret;
  92
  93   silc_list_init(schedule->fd_dispatch, struct SilcTaskStruct, next);
  94
  95   for (i = 0; i < ret; i++) {
  96     task = fds[i].data.ptr;
  97     task->revents = 0;
  98     if (!task->header.valid || !task->events) {
  99       epoll_ctl(internal->epfd, EPOLL_CTL_DEL, task->fd, &fds[i]);
 100       continue;
 101     }
 102     if (fds[i].events & (EPOLLIN | EPOLLPRI | EPOLLHUP | EPOLLERR))
 103       task->revents |= SILC_TASK_READ;
 104     if (fds[i].events & EPOLLOUT)
 105       task->revents |= SILC_TASK_WRITE;
 106     silc_list_add(schedule->fd_dispatch, task);
 107   }
 108
 109   return ret;
 110 }
 111
 112 #elif defined(HAVE_POLL) && defined(HAVE_SETRLIMIT) && defined(RLIMIT_NOFILE)
 113
 114 /* Calls normal poll() system call. */
 115
 116 int silc_poll(SilcSchedule schedule, void *context)
 117 {
 118   SilcUnixScheduler internal = context;
 119   SilcHashTableList htl;
 120   SilcTaskFd task;
 121   struct pollfd *fds = internal->fds;
 122   SilcUInt32 fds_count = internal->fds_count;
 123   int fd, ret, i = 0, timeout = -1;
 124   void *fdp;
 125
 126   silc_hash_table_list(schedule->fd_queue, &htl);
 127   while (silc_hash_table_get(&htl, &fdp, (void *)&task)) {
 128     if (!task->events)
 129       continue;
 130     fd = SILC_PTR_TO_32(fdp);
 131
 132     /* Allocate larger fd table if needed */
 133     if (i >= fds_count) {
 134       struct rlimit nofile;
 135
 136       fds = silc_realloc(internal->fds, sizeof(*internal->fds) *
 137                          (fds_count + (fds_count / 2)));
 138       if (silc_unlikely(!fds))
 139         break;
 140       internal->fds = fds;
 141       internal->fds_count = fds_count = fds_count + (fds_count / 2);
 142       internal->nofile.rlim_cur = fds_count;
 143       if (fds_count > internal->nofile.rlim_max)
 144         internal->nofile.rlim_max = fds_count;
 145       if (setrlimit(RLIMIT_NOFILE, &nofile) < 0)
 146         break;
 147     }
 148
 149     fds[i].fd = fd;
 150     fds[i].events = 0;
 151     task->revents = fds[i].revents = 0;
 152
 153     if (task->events & SILC_TASK_READ)
 154       fds[i].events |= (POLLIN | POLLPRI);
 155     if (task->events & SILC_TASK_WRITE)
 156       fds[i].events |= POLLOUT;
 157     i++;
 158   }
 159   silc_hash_table_list_reset(&htl);
 160   silc_list_init(schedule->fd_dispatch, struct SilcTaskStruct, next);
 161
 162   if (schedule->has_timeout)
 163     timeout = ((schedule->timeout.tv_sec * 1000) +
 164                (schedule->timeout.tv_usec / 1000));
 165
 166   fds_count = i;
 167   SILC_SCHEDULE_UNLOCK(schedule);
 168   ret = poll(fds, fds_count, timeout);
 169   SILC_SCHEDULE_LOCK(schedule);
 170   if (ret <= 0)
 171     return ret;
 172
 173   for (i = 0; i < fds_count; i++) {
 174     if (!fds[i].revents)
 175       continue;
 176     if (!silc_hash_table_find(schedule->fd_queue, SILC_32_TO_PTR(fds[i].fd),
 177                               NULL, (void *)&task))
 178       continue;
 179     if (!task->header.valid || !task->events)
 180       continue;
 181
 182     fd = fds[i].revents;
 183     if (fd & (POLLIN | POLLPRI | POLLERR | POLLHUP | POLLNVAL))
 184       task->revents |= SILC_TASK_READ;
 185     if (fd & POLLOUT)
 186       task->revents |= SILC_TASK_WRITE;
 187     silc_list_add(schedule->fd_dispatch, task);
 188   }
 189
 190   return ret;
 191 }
 192
 193 #else
 194
 195 /* Calls normal select() system call. */
 196
 197 int silc_select(SilcSchedule schedule, void *context)
 198 {
 199   SilcHashTableList htl;
 200   SilcTaskFd task;
 201   fd_set in, out;
 202   int fd, max_fd = 0, ret;
 203   void *fdp;
 204
 205   FD_ZERO(&in);
 206   FD_ZERO(&out);
 207
 208   silc_hash_table_list(schedule->fd_queue, &htl);
 209   while (silc_hash_table_get(&htl, &fdp, (void *)&task)) {
 210     if (!task->events)
 211       continue;
 212     fd = SILC_PTR_TO_32(fdp);
 213
 214 #ifdef FD_SETSIZE
 215     if (fd >= FD_SETSIZE)
 216       break;
 217 #endif /* FD_SETSIZE */
 218
 219     if (fd > max_fd)
 220       max_fd = fd;
 221
 222     if (task->events & SILC_TASK_READ)
 223       FD_SET(fd, &in);
 224     if (task->events & SILC_TASK_WRITE)
 225       FD_SET(fd, &out);
 226
 227     task->revents = 0;
 228   }
 229   silc_hash_table_list_reset(&htl);
 230   silc_list_init(schedule->fd_dispatch, struct SilcTaskStruct, next);
 231
 232   SILC_SCHEDULE_UNLOCK(schedule);
 233   ret = select(max_fd + 1, &in, &out, NULL, (schedule->has_timeout ?
 234                                              &schedule->timeout : NULL));
 235   SILC_SCHEDULE_LOCK(schedule);
 236   if (ret <= 0)
 237     return ret;
 238
 239   silc_hash_table_list(schedule->fd_queue, &htl);
 240   while (silc_hash_table_get(&htl, &fdp, (void *)&task)) {
 241     if (!task->header.valid || !task->events)
 242       continue;
 243     fd = SILC_PTR_TO_32(fdp);
 244
 245 #ifdef FD_SETSIZE
 246     if (fd >= FD_SETSIZE)
 247       break;
 248 #endif /* FD_SETSIZE */
 249
 250     if (FD_ISSET(fd, &in))
 251       task->revents |= SILC_TASK_READ;
 252     if (FD_ISSET(fd, &out))
 253       task->revents |= SILC_TASK_WRITE;
 254     silc_list_add(schedule->fd_dispatch, task);
 255   }
 256   silc_hash_table_list_reset(&htl);
 257
 258   return ret;
 259 }
 260
 261 #endif /* HAVE_POLL && HAVE_SETRLIMIT && RLIMIT_NOFILE */
 262
 263 /* Schedule `task' with events `event_mask'. Zero `event_mask' unschedules. */
 264
 265 SilcBool silc_schedule_internal_schedule_fd(SilcSchedule schedule,
 266                                             void *context,
 267                                             SilcTaskFd task,
 268                                             SilcTaskEvent event_mask)
 269 {
 270 #if defined(HAVE_EPOLL_WAIT)
 271   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 272   struct epoll_event event;
 273
 274   if (!internal)
 275     return TRUE;
 276
 277   SILC_LOG_DEBUG(("Scheduling fd %lu, mask %x", task->fd, event_mask));
 278
 279   memset(&event, 0, sizeof(event));
 280   if (event_mask & SILC_TASK_READ)
 281     event.events |= (EPOLLIN | EPOLLPRI);
 282   if (event_mask & SILC_TASK_WRITE)
 283     event.events |= EPOLLOUT;
 284
 285   /* Zero mask unschedules task */
 286   if (silc_unlikely(!event.events)) {
 287     if (epoll_ctl(internal->epfd, EPOLL_CTL_DEL, task->fd, &event)) {
 288       SILC_LOG_DEBUG(("epoll_ctl (DEL): %s", strerror(errno)));
 289       return FALSE;
 290     }
 291     task->scheduled = FALSE;
 292     return TRUE;
 293   }
 294
 295   /* Schedule the task */
 296   if (silc_unlikely(!task->scheduled)) {
 297     event.data.ptr = task;
 298     if (epoll_ctl(internal->epfd, EPOLL_CTL_ADD, task->fd, &event)) {
 299       SILC_LOG_DEBUG(("epoll_ctl (ADD): %s", strerror(errno)));
 300       return FALSE;
 301     }
 302     task->scheduled = TRUE;
 303     return TRUE;
 304   }
 305
 306   /* Schedule for specific mask */
 307   event.data.ptr = task;
 308   if (epoll_ctl(internal->epfd, EPOLL_CTL_MOD, task->fd, &event)) {
 309     SILC_LOG_DEBUG(("epoll_ctl (MOD): %s", strerror(errno)));
 310     return FALSE;
 311   }
 312 #endif /* HAVE_EPOLL_WAIT */
 313   return TRUE;
 314 }
 315
 316 #ifdef SILC_THREADS
 317
 318 SILC_TASK_CALLBACK(silc_schedule_wakeup_cb)
 319 {
 320   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 321   unsigned char c;
 322
 323   SILC_LOG_DEBUG(("Wokeup"));
 324
 325   (void)read(internal->wakeup_pipe[0], &c, 1);
 326 }
 327
 328 SILC_TASK_CALLBACK(silc_schedule_wakeup_init)
 329 {
 330   SilcUnixScheduler internal = schedule->internal;
 331
 332   internal->wakeup_task =
 333     silc_schedule_task_add(schedule, internal->wakeup_pipe[0],
 334                            silc_schedule_wakeup_cb, internal,
 335                            0, 0, SILC_TASK_FD);
 336   if (!internal->wakeup_task) {
 337     SILC_LOG_WARNING(("Could not add a wakeup task, threads won't work"));
 338     close(internal->wakeup_pipe[0]);
 339     return;
 340   }
 341   silc_schedule_internal_schedule_fd(schedule, internal,
 342                                      (SilcTaskFd)internal->wakeup_task,
 343                                      SILC_TASK_READ);
 344 }
 345 #endif /* SILC_THREADS */
 346
 347 /* Initializes the platform specific scheduler.  This for example initializes
 348    the wakeup mechanism of the scheduler.  In multi-threaded environment
 349    the scheduler needs to be woken up when tasks are added or removed from
 350    the task queues.  Returns context to the platform specific scheduler. */
 351
 352 void *silc_schedule_internal_init(SilcSchedule schedule,
 353                                   void *app_context)
 354 {
 355   SilcUnixScheduler internal;
 356   int i;
 357
 358   internal = silc_calloc(1, sizeof(*internal));
 359   if (!internal)
 360     return NULL;
 361
 362 #if defined(HAVE_EPOLL_WAIT)
 363   internal->epfd = epoll_create(4);
 364   if (internal->epfd < 0) {
 365     SILC_LOG_ERROR(("epoll_create() failed: %s", strerror(errno)));
 366     return NULL;
 367   }
 368   internal->fds = silc_calloc(4, sizeof(*internal->fds));
 369   if (!internal->fds) {
 370     close(internal->epfd);
 371     return NULL;
 372   }
 373   internal->fds_count = 4;
 374 #elif defined(HAVE_POLL) && defined(HAVE_SETRLIMIT) && defined(RLIMIT_NOFILE)
 375   getrlimit(RLIMIT_NOFILE, &internal->nofile);
 376
 377   if (schedule->max_tasks > 0) {
 378     internal->nofile.rlim_cur = schedule->max_tasks;
 379     if (schedule->max_tasks > internal->nofile.rlim_max)
 380       internal->nofile.rlim_max = schedule->max_tasks;
 381     setrlimit(RLIMIT_NOFILE, &internal->nofile);
 382     getrlimit(RLIMIT_NOFILE, &internal->nofile);
 383     schedule->max_tasks = internal->nofile.rlim_max;
 384   }
 385
 386   internal->fds = silc_calloc(internal->nofile.rlim_cur,
 387                               sizeof(*internal->fds));
 388   if (!internal->fds)
 389     return NULL;
 390   internal->fds_count = internal->nofile.rlim_cur;
 391 #endif /* HAVE_POLL && HAVE_SETRLIMIT && RLIMIT_NOFILE */
 392
 393   sigemptyset(&internal->signals);
 394
 395 #ifdef SILC_THREADS
 396   if (pipe(internal->wakeup_pipe)) {
 397     SILC_LOG_ERROR(("pipe() fails: %s", strerror(errno)));
 398     silc_free(internal);
 399     return NULL;
 400   }
 401
 402   silc_schedule_task_add_timeout(schedule, silc_schedule_wakeup_init,
 403                                  internal, 0, 0);
 404 #endif /* SILC_THREADS */
 405
 406   internal->app_context = app_context;
 407
 408   for (i = 0; i < SIGNAL_COUNT; i++) {
 409     signal_call[i].sig = 0;
 410     signal_call[i].call = FALSE;
 411     signal_call[i].schedule = schedule;
 412   }
 413
 414   return (void *)internal;
 415 }
 416
 417 void silc_schedule_internal_signals_block(SilcSchedule schedule,
 418                                           void *context);
 419 void silc_schedule_internal_signals_unblock(SilcSchedule schedule,
 420                                             void *context);
 421
 422 /* Uninitializes the platform specific scheduler context. */
 423
 424 void silc_schedule_internal_uninit(SilcSchedule schedule, void *context)
 425 {
 426   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 427
 428   if (!internal)
 429     return;
 430
 431 #ifdef SILC_THREADS
 432   close(internal->wakeup_pipe[0]);
 433   close(internal->wakeup_pipe[1]);
 434 #endif
 435
 436 #if defined(HAVE_EPOLL_WAIT)
 437   close(internal->epfd);
 438   silc_free(internal->fds);
 439 #elif defined(HAVE_POLL) && defined(HAVE_SETRLIMIT) && defined(RLIMIT_NOFILE)
 440   silc_free(internal->fds);
 441 #endif /* HAVE_POLL && HAVE_SETRLIMIT && RLIMIT_NOFILE */
 442
 443   silc_free(internal);
 444 }
 445
 446 /* Wakes up the scheduler */
 447
 448 void silc_schedule_internal_wakeup(SilcSchedule schedule, void *context)
 449 {
 450 #ifdef SILC_THREADS
 451   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 452
 453   if (!internal || !internal->wakeup_task)
 454     return;
 455
 456   SILC_LOG_DEBUG(("Wakeup"));
 457
 458   (void)write(internal->wakeup_pipe[1], "!", 1);
 459 #endif
 460 }
 461
 462 /* Signal handler */
 463
 464 static void silc_schedule_internal_sighandler(int signal)
 465 {
 466   int i;
 467
 468   SILC_LOG_DEBUG(("Start"));
 469
 470   for (i = 0; i < SIGNAL_COUNT; i++) {
 471     if (signal_call[i].sig == signal) {
 472       signal_call[i].call = TRUE;
 473       signal_call[i].schedule->signal_tasks = TRUE;
 474       SILC_LOG_DEBUG(("Scheduling signal %d to be called",
 475                       signal_call[i].sig));
 476       break;
 477     }
 478   }
 479 }
 480
 481 void silc_schedule_internal_signal_register(SilcSchedule schedule,
 482                                             void *context,
 483                                             SilcUInt32 sig,
 484                                             SilcTaskCallback callback,
 485                                             void *callback_context)
 486 {
 487   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 488   int i;
 489
 490   if (!internal)
 491     return;
 492
 493   SILC_LOG_DEBUG(("Registering signal %d", sig));
 494
 495   silc_schedule_internal_signals_block(schedule, context);
 496
 497   for (i = 0; i < SIGNAL_COUNT; i++) {
 498     if (!signal_call[i].sig) {
 499       signal_call[i].sig = sig;
 500       signal_call[i].callback = callback;
 501       signal_call[i].context = callback_context;
 502       signal_call[i].schedule = schedule;
 503       signal_call[i].call = FALSE;
 504       signal(sig, silc_schedule_internal_sighandler);
 505       break;
 506     }
 507   }
 508
 509   silc_schedule_internal_signals_unblock(schedule, context);
 510   sigaddset(&internal->signals, sig);
 511 }
 512
 513 void silc_schedule_internal_signal_unregister(SilcSchedule schedule,
 514                                               void *context,
 515                                               SilcUInt32 sig)
 516 {
 517   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 518   int i;
 519
 520   if (!internal)
 521     return;
 522
 523   SILC_LOG_DEBUG(("Unregistering signal %d", sig));
 524
 525   silc_schedule_internal_signals_block(schedule, context);
 526
 527   for (i = 0; i < SIGNAL_COUNT; i++) {
 528     if (signal_call[i].sig == sig) {
 529       signal_call[i].sig = 0;
 530       signal_call[i].callback = NULL;
 531       signal_call[i].context = NULL;
 532       signal_call[i].schedule = NULL;
 533       signal_call[i].call = FALSE;
 534       signal(sig, SIG_DFL);
 535     }
 536   }
 537
 538   silc_schedule_internal_signals_unblock(schedule, context);
 539   sigdelset(&internal->signals, sig);
 540 }
 541
 542 /* Call all signals */
 543
 544 void silc_schedule_internal_signals_call(SilcSchedule schedule, void *context)
 545 {
 546   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 547   int i;
 548
 549   SILC_LOG_DEBUG(("Start"));
 550
 551   if (!internal)
 552     return;
 553
 554   silc_schedule_internal_signals_block(schedule, context);
 555
 556   for (i = 0; i < SIGNAL_COUNT; i++) {
 557     if (signal_call[i].call &&
 558         signal_call[i].callback) {
 559       SILC_LOG_DEBUG(("Calling signal %d callback",
 560                       signal_call[i].sig));
 561       silc_schedule_internal_signals_unblock(schedule, context);
 562       signal_call[i].callback(schedule, internal->app_context,
 563                               SILC_TASK_INTERRUPT,
 564                               signal_call[i].sig,
 565                               signal_call[i].context);
 566       signal_call[i].call = FALSE;
 567       silc_schedule_internal_signals_block(schedule, context);
 568     }
 569   }
 570
 571   silc_schedule_internal_signals_unblock(schedule, context);
 572 }
 573
 574 /* Block registered signals in scheduler. */
 575
 576 void silc_schedule_internal_signals_block(SilcSchedule schedule, void *context)
 577 {
 578   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 579
 580   if (!internal)
 581     return;
 582
 583   sigprocmask(SIG_BLOCK, &internal->signals, &internal->signals_blocked);
 584 }
 585
 586 /* Unblock registered signals in schedule. */
 587
 588 void silc_schedule_internal_signals_unblock(SilcSchedule schedule,
 589                                             void *context)
 590 {
 591   SilcUnixScheduler internal = (SilcUnixScheduler)context;
 592
 593   if (!internal)
 594     return;
 595
 596   sigprocmask(SIG_SETMASK, &internal->signals_blocked, NULL);
 597 }
 598
 599 const SilcScheduleOps schedule_ops =
 600 {
 601   silc_schedule_internal_init,
 602   silc_schedule_internal_uninit,
 603 #if defined(HAVE_EPOLL_WAIT)
 604   silc_epoll,
 605 #elif defined(HAVE_POLL) && defined(HAVE_SETRLIMIT) && defined(RLIMIT_NOFILE)
 606   silc_poll,
 607 #else
 608   silc_select,
 609 #endif /* HAVE_POLL && HAVE_SETRLIMIT && RLIMIT_NOFILE */
 610   silc_schedule_internal_schedule_fd,
 611   silc_schedule_internal_wakeup,
 612   silc_schedule_internal_signal_register,
 613   silc_schedule_internal_signal_unregister,
 614   silc_schedule_internal_signals_call,
 615   silc_schedule_internal_signals_block,
 616   silc_schedule_internal_signals_unblock,
 617 };