ircd/engine_kqueue.c

   1 /*
   2  * IRC - Internet Relay Chat, ircd/engine_kqueue.c
   3  * Copyright (C) 2001 Kevin L. Mitchell <klmitch@mit.edu>
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 1, or (at your option)
   8  * any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License
  16  * along with this program; if not, write to the Free Software
  17  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  18  *
  19  * $Id$
  20  */
  21 #include "config.h"
  22
  23 #include "ircd_events.h"
  24
  25 #include "ircd.h"
  26 #include "ircd_alloc.h"
  27 #include "ircd_features.h"
  28 #include "ircd_log.h"
  29 #include "s_debug.h"
  30
  31 #include <assert.h>
  32 #include <errno.h>
  33 #include <signal.h>
  34 #include <sys/event.h>
  35 #include <sys/socket.h>
  36 #include <sys/time.h>
  37 #include <sys/types.h>
  38 #include <time.h>
  39 #include <unistd.h>
  40
  41 #define KQUEUE_ERROR_THRESHOLD  20      /* after 20 kqueue errors, restart */
  42 #define ERROR_EXPIRE_TIME       3600    /* expire errors after an hour */
  43
  44 static struct Socket** sockList;
  45 static int kqueue_max;
  46 static int kqueue_id;
  47
  48 static int errors = 0;
  49 static struct Timer clear_error;
  50
  51 /* decrements the error count once per hour */
  52 static void
  53 error_clear(struct Event* ev)
  54 {
  55   if (!--errors) /* remove timer when error count reaches 0 */
  56     timer_del(ev_timer(ev));
  57 }
  58
  59 /* initialize the kqueue engine */
  60 static int
  61 engine_init(int max_sockets)
  62 {
  63   int i;
  64
  65   if ((kqueue_id = kqueue()) < 0) { /* initialize... */
  66     log_write(LS_SYSTEM, L_WARNING, 0,
  67               "kqueue() engine cannot initialize: %m");
  68     return 0;
  69   }
  70
  71   /* allocate necessary memory */
  72   sockList = (struct Socket**) MyMalloc(sizeof(struct Socket*) * max_sockets);
  73
  74   /* initialize the data */
  75   for (i = 0; i < max_sockets; i++)
  76     sockList[i] = 0;
  77
  78   kqueue_max = max_sockets; /* number of sockets allocated */
  79
  80   return 1; /* success! */
  81 }
  82
  83 /* add a signel to be watched for */
  84 static void
  85 engine_signal(struct Signal* sig)
  86 {
  87   struct kevent sigevent;
  88   struct sigaction act;
  89
  90   assert(0 != signal);
  91
  92   Debug((DEBUG_ENGINE, "kqueue: Adding filter for signal %d [%p]",
  93          sig_signal(sig), sig));
  94
  95   sigevent.ident = sig_signal(sig); /* set up the kqueue event */
  96   sigevent.filter = EVFILT_SIGNAL; /* looking for signals... */
  97   sigevent.flags = EV_ADD | EV_ENABLE; /* add and enable it */
  98   sigevent.fflags = 0;
  99   sigevent.data = 0;
 100   sigevent.udata = sig; /* store our user data */
 101
 102   if (kevent(kqueue_id, &sigevent, 1, 0, 0, 0) < 0) { /* add event */
 103     log_write(LS_SYSTEM, L_WARNING, 0, "Unable to trap signal %d",
 104               sig_signal(sig));
 105     return;
 106   }
 107
 108   act.sa_handler = SIG_IGN; /* ignore the signal */
 109   act.sa_flags = 0;
 110   sigemptyset(&act.sa_mask);
 111   sigaction(sig_signal(sig), &act, 0);
 112 }
 113
 114 /* Figure out what events go with a given state */
 115 static unsigned int
 116 state_to_events(enum SocketState state, unsigned int events)
 117 {
 118   switch (state) {
 119   case SS_CONNECTING: /* connecting socket */
 120     return SOCK_EVENT_WRITABLE;
 121     break;
 122
 123   case SS_LISTENING: /* listening socket */
 124   case SS_NOTSOCK: /* our signal socket--just in case */
 125     return SOCK_EVENT_READABLE;
 126     break;
 127
 128   case SS_CONNECTED: case SS_DATAGRAM: case SS_CONNECTDG:
 129     return events; /* ordinary socket */
 130     break;
 131   }
 132
 133   /*NOTREACHED*/
 134   return 0;
 135 }
 136
 137 /* Activate kqueue filters as appropriate */
 138 static void
 139 set_or_clear(struct Socket* sock, unsigned int clear, unsigned int set)
 140 {
 141   int i = 0;
 142   struct kevent chglist[2];
 143
 144   assert(0 != sock);
 145   assert(-1 < s_fd(sock));
 146
 147   if ((clear ^ set) & SOCK_EVENT_READABLE) { /* readable has changed */
 148     chglist[i].ident = s_fd(sock); /* set up the change list */
 149     chglist[i].filter = EVFILT_READ; /* readable filter */
 150     chglist[i].flags = EV_ADD; /* adding it */
 151     chglist[i].fflags = 0;
 152     chglist[i].data = 0;
 153     chglist[i].udata = 0; /* I love udata, but it can't really be used here */
 154
 155     if (set & SOCK_EVENT_READABLE) /* it's set */
 156       chglist[i].flags |= EV_ENABLE;
 157     else /* clear it */
 158       chglist[i].flags |= EV_DISABLE;
 159
 160     i++; /* advance to next element */
 161   }
 162
 163   if ((clear ^ set) & SOCK_EVENT_WRITABLE) { /* writable has changed */
 164     chglist[i].ident = s_fd(sock); /* set up the change list */
 165     chglist[i].filter = EVFILT_WRITE; /* writable filter */
 166     chglist[i].flags = EV_ADD; /* adding it */
 167     chglist[i].fflags = 0;
 168     chglist[i].data = 0;
 169     chglist[i].udata = 0;
 170
 171     if (set & SOCK_EVENT_WRITABLE) /* it's set */
 172       chglist[i].flags |= EV_ENABLE;
 173     else /* clear it */
 174       chglist[i].flags |= EV_DISABLE;
 175
 176     i++; /* advance count... */
 177   }
 178
 179   if (kevent(kqueue_id, chglist, i, 0, 0, 0) < 0 && errno != EBADF)
 180     event_generate(ET_ERROR, sock, errno); /* report error */
 181 }
 182
 183 /* add a socket to be listened on */
 184 static int
 185 engine_add(struct Socket* sock)
 186 {
 187   assert(0 != sock);
 188   assert(0 == sockList[s_fd(sock)]);
 189
 190   /* bounds-check... */
 191   if (sock->s_fd >= kqueue_max) {
 192     log_write(LS_SYSTEM, L_ERROR, 0,
 193               "Attempt to add socket %d (> %d) to event engine", s_fd(sock),
 194               kqueue_max);
 195     return 0;
 196   }
 197
 198   sockList[s_fd(sock)] = sock; /* add to list */
 199
 200   Debug((DEBUG_ENGINE, "kqueue: Adding socket %d [%p], state %s, to engine",
 201          s_fd(sock), sock, state_to_name(s_state(sock))));
 202
 203   /* Add socket to queue */
 204   set_or_clear(sock, 0, state_to_events(s_state(sock), s_events(sock)));
 205
 206   return 1; /* success */
 207 }
 208
 209 /* socket switching to new state */
 210 static void
 211 engine_state(struct Socket* sock, enum SocketState new_state)
 212 {
 213   assert(0 != sock);
 214   assert(sock == sockList[s_fd(sock)]);
 215
 216   Debug((DEBUG_ENGINE, "kqueue: Changing state for socket %p to %s", sock,
 217          state_to_name(new_state)));
 218
 219   /* set the correct events */
 220   set_or_clear(sock,
 221                state_to_events(s_state(sock), s_events(sock)), /* old state */
 222                state_to_events(new_state, s_events(sock))); /* new state */
 223
 224 }
 225
 226 /* socket events changing */
 227 static void
 228 engine_events(struct Socket* sock, unsigned int new_events)
 229 {
 230   assert(0 != sock);
 231   assert(sock == sockList[s_fd(sock)]);
 232
 233   Debug((DEBUG_ENGINE, "kqueue: Changing event mask for socket %p to [%s]",
 234          sock, sock_flags(new_events)));
 235
 236   /* set the correct events */
 237   set_or_clear(sock,
 238                state_to_events(s_state(sock), s_events(sock)), /* old events */
 239                state_to_events(s_state(sock), new_events)); /* new events */
 240 }
 241
 242 /* socket going away */
 243 static void
 244 engine_delete(struct Socket* sock)
 245 {
 246   struct kevent dellist[2];
 247
 248   assert(0 != sock);
 249   assert(sock == sockList[s_fd(sock)]);
 250
 251   Debug((DEBUG_ENGINE, "kqueue: Deleting socket %d [%p], state %s",
 252          s_fd(sock), sock, state_to_name(s_state(sock))));
 253
 254   dellist[0].ident = s_fd(sock); /* set up the delete list */
 255   dellist[0].filter = EVFILT_READ; /* readable filter */
 256   dellist[0].flags = EV_DELETE; /* delete it */
 257   dellist[0].fflags = 0;
 258   dellist[0].data = 0;
 259   dellist[0].udata = 0;
 260
 261   dellist[1].ident = s_fd(sock);
 262   dellist[1].filter = EVFILT_WRITE; /* writable filter */
 263   dellist[1].flags = EV_DELETE; /* delete it */
 264   dellist[1].fflags = 0;
 265   dellist[1].data = 0;
 266   dellist[1].udata = 0;
 267
 268   /* make it all go away */
 269   if (kevent(kqueue_id, dellist, 2, 0, 0, 0) < 0)
 270     log_write(LS_SOCKET, L_WARNING, 0,
 271               "Unable to delete kevent items for socket %d", s_fd(sock));
 272
 273   sockList[s_fd(sock)] = 0;
 274 }
 275
 276 /* engine event loop */
 277 static void
 278 engine_loop(struct Generators* gen)
 279 {
 280   struct kevent *events;
 281   int events_count;
 282   struct Socket* sock;
 283   struct timespec wait;
 284   int nevs;
 285   int i;
 286   int errcode;
 287   size_t codesize;
 288
 289   if ((events_count = feature_int(FEAT_POLLS_PER_LOOP)) < 20)
 290     events_count = 20;
 291   events = (struct kevent *)MyMalloc(sizeof(struct kevent) * events_count);
 292
 293   while (running) {
 294     if ((i = feature_int(FEAT_POLLS_PER_LOOP)) >= 20 && i != events_count) {
 295       events = (struct kevent *)MyRealloc(events, sizeof(struct kevent) * i);
 296       events_count = i;
 297     }
 298
 299     /* set up the sleep time */
 300     wait.tv_sec = timer_next(gen) ? (timer_next(gen) - CurrentTime) : -1;
 301     wait.tv_nsec = 0;
 302
 303     Debug((DEBUG_INFO, "kqueue: delay: %Tu (%Tu) %Tu", timer_next(gen),
 304            CurrentTime, wait.tv_sec));
 305
 306     /* check for active events */
 307     nevs = kevent(kqueue_id, 0, 0, events, events_count,
 308                   wait.tv_sec < 0 ? 0 : &wait);
 309
 310     CurrentTime = time(0); /* set current time... */
 311
 312     if (nevs < 0) {
 313       if (errno != EINTR) { /* ignore kevent interrupts */
 314         /* Log the kqueue error */
 315         log_write(LS_SOCKET, L_ERROR, 0, "kevent() error: %m");
 316         if (!errors++)
 317           timer_add(timer_init(&clear_error), error_clear, 0, TT_PERIODIC,
 318                     ERROR_EXPIRE_TIME);
 319         else if (errors > KQUEUE_ERROR_THRESHOLD) /* too many errors... */
 320           server_restart("too many kevent errors");
 321       }
 322       /* old code did a sleep(1) here; with usage these days,
 323        * that may be too expensive
 324        */
 325       continue;
 326     }
 327
 328     for (i = 0; i < nevs; i++) {
 329       if (events[i].filter == EVFILT_SIGNAL) {
 330         /* it's a signal; deal appropriately */
 331         event_generate(ET_SIGNAL, events[i].udata, events[i].ident);
 332         continue; /* skip socket processing loop */
 333       }
 334
 335       assert(events[i].filter == EVFILT_READ ||
 336              events[i].filter == EVFILT_WRITE);
 337
 338       sock = sockList[events[i].ident];
 339       if (!sock) /* slots may become empty while processing events */
 340         continue;
 341
 342       assert(s_fd(sock) == events[i].ident);
 343
 344       gen_ref_inc(sock); /* can't have it going away on us */
 345
 346       Debug((DEBUG_ENGINE, "kqueue: Checking socket %p (fd %d) state %s, "
 347              "events %s", sock, s_fd(sock), state_to_name(s_state(sock)),
 348              sock_flags(s_events(sock))));
 349
 350       if (s_state(sock) != SS_NOTSOCK) {
 351         errcode = 0; /* check for errors on socket */
 352         codesize = sizeof(errcode);
 353         if (getsockopt(s_fd(sock), SOL_SOCKET, SO_ERROR, &errcode,
 354                        &codesize) < 0)
 355           errcode = errno; /* work around Solaris implementation */
 356
 357         if (errcode) { /* an error occurred; generate an event */
 358           Debug((DEBUG_ENGINE, "kqueue: Error %d on fd %d, socket %p", errcode,
 359                  s_fd(sock), sock));
 360           event_generate(ET_ERROR, sock, errcode);
 361           gen_ref_dec(sock); /* careful not to leak reference counts */
 362           continue;
 363         }
 364       }
 365
 366       switch (s_state(sock)) {
 367       case SS_CONNECTING:
 368         if (events[i].filter == EVFILT_WRITE) { /* connection completed */
 369           Debug((DEBUG_ENGINE, "kqueue: Connection completed"));
 370           event_generate(ET_CONNECT, sock, 0);
 371         }
 372         break;
 373
 374       case SS_LISTENING:
 375         if (events[i].filter == EVFILT_READ) { /* connect. to be accept. */
 376           Debug((DEBUG_ENGINE, "kqueue: Ready for accept"));
 377           event_generate(ET_ACCEPT, sock, 0);
 378         }
 379         break;
 380
 381       case SS_NOTSOCK: /* doing nothing socket-specific */
 382       case SS_CONNECTED:
 383         if (events[i].filter == EVFILT_READ) { /* data on socket */
 384           Debug((DEBUG_ENGINE, "kqueue: EOF or data to be read"));
 385           event_generate(events[i].flags & EV_EOF ? ET_EOF : ET_READ, sock, 0);
 386         }
 387         if (events[i].filter == EVFILT_WRITE) { /* socket writable */
 388           Debug((DEBUG_ENGINE, "kqueue: Data can be written"));
 389           event_generate(ET_WRITE, sock, 0);
 390         }
 391         break;
 392
 393       case SS_DATAGRAM: case SS_CONNECTDG:
 394         if (events[i].filter == EVFILT_READ) { /* socket readable */
 395           Debug((DEBUG_ENGINE, "kqueue: Datagram to be read"));
 396           event_generate(ET_READ, sock, 0);
 397         }
 398         if (events[i].filter == EVFILT_WRITE) { /* socket writable */
 399           Debug((DEBUG_ENGINE, "kqueue: Datagram can be written"));
 400           event_generate(ET_WRITE, sock, 0);
 401         }
 402         break;
 403       }
 404
 405       assert(s_fd(sock) == events[i].ident);
 406
 407       gen_ref_dec(sock); /* we're done with it */
 408     }
 409
 410     timer_run(); /* execute any pending timers */
 411   }
 412 }
 413
 414 struct Engine engine_kqueue = {
 415   "kqueue()",           /* Engine name */
 416   engine_init,          /* Engine initialization function */
 417   engine_signal,        /* Engine signal registration function */
 418   engine_add,           /* Engine socket registration function */
 419   engine_state,         /* Engine socket state change function */
 420   engine_events,        /* Engine socket events mask function */
 421   engine_delete,        /* Engine socket deletion function */
 422   engine_loop           /* Core engine event loop */
 423 };