src/comm/ModKqueue.cc

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