src/comm/ModKqueue.cc

   1 /*
   2  * Copyright (C) 1996-2021 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 /* DEBUG: section 05    Socket Functions */
  10
  11 /*
  12  * This code was originally written by Benno Rice and hacked on quite
  13  * a bit by Adrian. Adrian then took it to the hybrid-ircd project to use
  14  * in their new IO subsystem. After a year of modifications and some
  15  * rather interesting changes (event aggregation) its back in squid.
  16  * Thanks to the ircd-hybrid guys.
  17  */
  18
  19 /*
  20  * XXX Currently not implemented / supported by this module XXX
  21  *
  22  * - delay pools
  23  * - deferred reads
  24  * - flags.read_pending
  25  *
  26  * So, its not entirely useful in a production setup since if a read
  27  * is meant to be deferred it isn't (we're not even throwing the event
  28  * away here). Eventually the rest of the code will be rewritten
  29  * so deferred reads aren't required.
  30  *  -- adrian
  31  */
  32 #include "squid.h"
  33
  34 #if USE_KQUEUE
  35 #include "comm/Loops.h"
  36 #include "fde.h"
  37 #include "globals.h"
  38 #include "SquidTime.h"
  39 #include "StatCounters.h"
  40 #include "Store.h"
  41
  42 #include <cerrno>
  43 #if HAVE_SYS_EVENT_H
  44 #include <sys/event.h>
  45 #endif
  46
  47 #define KE_LENGTH        128
  48
  49 /* jlemon goofed up and didn't add EV_SET until fbsd 4.3 */
  50
  51 #ifndef EV_SET
  52 #define EV_SET(kevp, a, b, c, d, e, f) do {     \
  53         (kevp)->ident = (a);                    \
  54         (kevp)->filter = (b);                   \
  55         (kevp)->flags = (c);                    \
  56         (kevp)->fflags = (d);                   \
  57         (kevp)->data = (e);                     \
  58         (kevp)->udata = (f);                    \
  59 } while(0)
  60 #endif
  61
  62 static void kq_update_events(int, short, PF *);
  63 static int kq;
  64
  65 static struct timespec zero_timespec;
  66
  67 static struct kevent *kqlst;        /* kevent buffer */
  68 static int kqmax;                /* max structs to buffer */
  69 static int kqoff;                /* offset into the buffer */
  70 static int max_poll_time = 1000;
  71
  72 static void commKQueueRegisterWithCacheManager(void);
  73
  74 /* XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX */
  75 /* Private functions */
  76
  77 void
  78 kq_update_events(int fd, short filter, PF * handler)
  79 {
  80     PF *cur_handler;
  81     int kep_flags;
  82
  83     switch (filter) {
  84
  85     case EVFILT_READ:
  86         cur_handler = fd_table[fd].read_handler;
  87         break;
  88
  89     case EVFILT_WRITE:
  90         cur_handler = fd_table[fd].write_handler;
  91         break;
  92
  93     default:
  94         /* XXX bad! -- adrian */
  95         return;
  96         break;
  97     }
  98
  99     if ((cur_handler == NULL && handler != NULL)
 100             || (cur_handler != NULL && handler == NULL)) {
 101
 102         struct kevent *kep;
 103
 104         kep = kqlst + kqoff;
 105
 106         if (handler != NULL) {
 107             kep_flags = (EV_ADD | EV_ONESHOT);
 108         } else {
 109             kep_flags = EV_DELETE;
 110         }
 111
 112         EV_SET(kep, (uintptr_t) fd, filter, kep_flags, 0, 0, 0);
 113
 114         /* Check if we've used the last one. If we have then submit them all */
 115         if (kqoff == kqmax - 1) {
 116             int ret;
 117
 118             ret = kevent(kq, kqlst, kqmax, NULL, 0, &zero_timespec);
 119             /* jdc -- someone needs to do error checking... */
 120
 121             if (ret == -1) {
 122                 perror("kq_update_events(): kevent()");
 123                 return;
 124             }
 125
 126             kqoff = 0;
 127         } else {
 128             ++kqoff;
 129         }
 130     }
 131 }
 132
 133 /* XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX */
 134 /* Public functions */
 135
 136 /*
 137  * comm_select_init
 138  *
 139  * This is a needed exported function which will be called to initialise
 140  * the network loop code.
 141  */
 142 void
 143 Comm::SelectLoopInit(void)
 144 {
 145     kq = kqueue();
 146
 147     if (kq < 0) {
 148         fatal("comm_select_init: Couldn't open kqueue fd!\n");
 149     }
 150
 151     kqmax = getdtablesize();
 152
 153     kqlst = (struct kevent *)xmalloc(sizeof(*kqlst) * kqmax);
 154     zero_timespec.tv_sec = 0;
 155     zero_timespec.tv_nsec = 0;
 156
 157     commKQueueRegisterWithCacheManager();
 158 }
 159
 160 /*
 161  * comm_setselect
 162  *
 163  * This is a needed exported function which will be called to register
 164  * and deregister interest in a pending IO state for a given FD.
 165  */
 166 void
 167 Comm::SetSelect(int fd, unsigned int type, PF * handler, void *client_data, time_t timeout)
 168 {
 169     fde *F = &fd_table[fd];
 170     assert(fd >= 0);
 171     assert(F->flags.open || (!handler && !client_data && !timeout));
 172     debugs(5, 5, HERE << "FD " << fd << ", type=" << type <<
 173            ", handler=" << handler << ", client_data=" << client_data <<
 174            ", timeout=" << timeout);
 175
 176     if (type & COMM_SELECT_READ) {
 177         if (F->flags.read_pending)
 178             kq_update_events(fd, EVFILT_WRITE, handler);
 179
 180         kq_update_events(fd, EVFILT_READ, handler);
 181
 182         F->read_handler = handler;
 183         F->read_data = client_data;
 184     }
 185
 186     if (type & COMM_SELECT_WRITE) {
 187         kq_update_events(fd, EVFILT_WRITE, handler);
 188         F->write_handler = handler;
 189         F->write_data = client_data;
 190     }
 191
 192     if (timeout)
 193         F->timeout = squid_curtime + timeout;
 194
 195 }
 196
 197 /*
 198  * Check all connections for new connections and input data that is to be
 199  * processed. Also check for connections with data queued and whether we can
 200  * write it out.
 201  */
 202
 203 /*
 204  * comm_select
 205  *
 206  * Called to do the new-style IO, courtesy of of squid (like most of this
 207  * new IO code). This routine handles the stuff we've hidden in
 208  * comm_setselect and fd_table[] and calls callbacks for IO ready
 209  * events.
 210  */
 211
 212 Comm::Flag
 213 Comm::DoSelect(int msec)
 214 {
 215     int num, i;
 216
 217     static struct kevent ke[KE_LENGTH];
 218
 219     struct timespec poll_time;
 220
 221     if (msec > max_poll_time)
 222         msec = max_poll_time;
 223
 224     poll_time.tv_sec = msec / 1000;
 225
 226     poll_time.tv_nsec = (msec % 1000) * 1000000;
 227
 228     for (;;) {
 229         num = kevent(kq, kqlst, kqoff, ke, KE_LENGTH, &poll_time);
 230         ++statCounter.select_loops;
 231         kqoff = 0;
 232
 233         if (num >= 0)
 234             break;
 235
 236         if (ignoreErrno(errno))
 237             break;
 238
 239         getCurrentTime();
 240
 241         return Comm::COMM_ERROR;
 242
 243         /* NOTREACHED */
 244     }
 245
 246     getCurrentTime();
 247
 248     if (num == 0)
 249         return Comm::OK;        /* No error.. */
 250
 251     for (i = 0; i < num; ++i) {
 252         int fd = (int) ke[i].ident;
 253         PF *hdl = NULL;
 254         fde *F = &fd_table[fd];
 255
 256         if (ke[i].flags & EV_ERROR) {
 257             errno = ke[i].data;
 258             /* XXX error == bad! -- adrian */
 259             continue;        /* XXX! */
 260         }
 261
 262         if (ke[i].filter == EVFILT_READ || F->flags.read_pending) {
 263             if ((hdl = F->read_handler) != NULL) {
 264                 F->read_handler = NULL;
 265                 hdl(fd, F->read_data);
 266             }
 267         }
 268
 269         if (ke[i].filter == EVFILT_WRITE) {
 270             if ((hdl = F->write_handler) != NULL) {
 271                 F->write_handler = NULL;
 272                 hdl(fd, F->write_data);
 273             }
 274         }
 275
 276         if (ke[i].filter != EVFILT_WRITE && ke[i].filter != EVFILT_READ) {
 277             /* Bad! -- adrian */
 278             debugs(5, DBG_IMPORTANT, "comm_select: kevent returned " << ke[i].filter << "!");
 279         }
 280     }
 281
 282     return Comm::OK;
 283 }
 284
 285 void
 286 Comm::QuickPollRequired(void)
 287 {
 288     max_poll_time = 10;
 289 }
 290
 291 static void
 292 commKQueueRegisterWithCacheManager(void)
 293 {
 294 }
 295
 296 #endif /* USE_KQUEUE */
 297