src/comm/ModKqueue.cc

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