src/comm/ModPoll.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 #include "squid.h"
  32
  33 #if USE_POLL
  34 #include "anyp/PortCfg.h"
  35 #include "comm/Connection.h"
  36 #include "comm/Loops.h"
  37 #include "fd.h"
  38 #include "fde.h"
  39 #include "globals.h"
  40 #include "ICP.h"
  41 #include "mgr/Registration.h"
  42 #include "profiler/Profiler.h"
  43 #include "SquidConfig.h"
  44 #include "SquidTime.h"
  45 #include "StatCounters.h"
  46 #include "Store.h"
  47
  48 #if HAVE_POLL_H
  49 #include <poll.h>
  50 #endif
  51 #if HAVE_ERRNO_H
  52 #include <errno.h>
  53 #endif
  54
  55 /* Needed for poll() on Linux at least */
  56 #if USE_POLL
  57 #ifndef POLLRDNORM
  58 #define POLLRDNORM POLLIN
  59 #endif
  60 #ifndef POLLWRNORM
  61 #define POLLWRNORM POLLOUT
  62 #endif
  63 #endif
  64
  65 static int MAX_POLL_TIME = 1000;        /* see also Comm::QuickPollRequired() */
  66
  67 #ifndef        howmany
  68 #define howmany(x, y)   (((x)+((y)-1))/(y))
  69 #endif
  70 #ifndef        NBBY
  71 #define        NBBY    8
  72 #endif
  73 #define FD_MASK_BYTES sizeof(fd_mask)
  74 #define FD_MASK_BITS (FD_MASK_BYTES*NBBY)
  75
  76 /* STATIC */
  77 static int fdIsTcpListen(int fd);
  78 static int fdIsUdpListen(int fd);
  79 static int fdIsDns(int fd);
  80 static OBJH commIncomingStats;
  81 static int comm_check_incoming_poll_handlers(int nfds, int *fds);
  82 static void comm_poll_dns_incoming(void);
  83
  84 /*
  85  * Automatic tuning for incoming requests:
  86  *
  87  * INCOMING sockets are the ICP and HTTP ports.  We need to check these
  88  * fairly regularly, but how often?  When the load increases, we
  89  * want to check the incoming sockets more often.  If we have a lot
  90  * of incoming ICP, then we need to check these sockets more than
  91  * if we just have HTTP.
  92  *
  93  * The variables 'incoming_icp_interval' and 'incoming_http_interval'
  94  * determine how many normal I/O events to process before checking
  95  * incoming sockets again.  Note we store the incoming_interval
  96  * multipled by a factor of (2^INCOMING_FACTOR) to have some
  97  * pseudo-floating point precision.
  98  *
  99  * The variable 'udp_io_events' and 'tcp_io_events' counts how many normal
 100  * I/O events have been processed since the last check on the incoming
 101  * sockets.  When io_events > incoming_interval, its time to check incoming
 102  * sockets.
 103  *
 104  * Every time we check incoming sockets, we count how many new messages
 105  * or connections were processed.  This is used to adjust the
 106  * incoming_interval for the next iteration.  The new incoming_interval
 107  * is calculated as the current incoming_interval plus what we would
 108  * like to see as an average number of events minus the number of
 109  * events just processed.
 110  *
 111  *  incoming_interval = incoming_interval + target_average - number_of_events_processed
 112  *
 113  * There are separate incoming_interval counters for TCP-based, UDP-based, and DNS events
 114  *
 115  * You can see the current values of the incoming_interval's, as well as
 116  * a histogram of 'incoming_events' by asking the cache manager
 117  * for 'comm_incoming', e.g.:
 118  *
 119  *      % ./client mgr:comm_poll_incoming
 120  *
 121  * Caveats:
 122  *
 123  *      - We have MAX_INCOMING_INTEGER as a magic upper limit on
 124  *        incoming_interval for both types of sockets.  At the
 125  *        largest value the cache will effectively be idling.
 126  *
 127  *      - The higher the INCOMING_FACTOR, the slower the algorithm will
 128  *        respond to load spikes/increases/decreases in demand. A value
 129  *        between 3 and 8 is recommended.
 130  */
 131
 132 #define MAX_INCOMING_INTEGER 256
 133 #define INCOMING_FACTOR 5
 134 #define MAX_INCOMING_INTERVAL (MAX_INCOMING_INTEGER << INCOMING_FACTOR)
 135 static int udp_io_events = 0; ///< I/O events passed since last UDP receiver socket poll
 136 static int dns_io_events = 0; ///< I/O events passed since last DNS socket poll
 137 static int tcp_io_events = 0; ///< I/O events passed since last TCP listening socket poll
 138 static int incoming_udp_interval = 16 << INCOMING_FACTOR;
 139 static int incoming_dns_interval = 16 << INCOMING_FACTOR;
 140 static int incoming_tcp_interval = 16 << INCOMING_FACTOR;
 141 #define commCheckUdpIncoming (++udp_io_events > (incoming_udp_interval>> INCOMING_FACTOR))
 142 #define commCheckDnsIncoming (++dns_io_events > (incoming_dns_interval>> INCOMING_FACTOR))
 143 #define commCheckTcpIncoming (++tcp_io_events > (incoming_tcp_interval>> INCOMING_FACTOR))
 144
 145 void
 146 Comm::SetSelect(int fd, unsigned int type, PF * handler, void *client_data, time_t timeout)
 147 {
 148     fde *F = &fd_table[fd];
 149     assert(fd >= 0);
 150     assert(F->flags.open);
 151     debugs(5, 5, HERE << "FD " << fd << ", type=" << type <<
 152            ", handler=" << handler << ", client_data=" << client_data <<
 153            ", timeout=" << timeout);
 154
 155     if (type & COMM_SELECT_READ) {
 156         F->read_handler = handler;
 157         F->read_data = client_data;
 158     }
 159
 160     if (type & COMM_SELECT_WRITE) {
 161         F->write_handler = handler;
 162         F->write_data = client_data;
 163     }
 164
 165     if (timeout)
 166         F->timeout = squid_curtime + timeout;
 167 }
 168
 169 void
 170 Comm::ResetSelect(int fd)
 171 {
 172 }
 173
 174 static int
 175 fdIsUdpListen(int fd)
 176 {
 177     if (icpIncomingConn != NULL && icpIncomingConn->fd == fd)
 178         return 1;
 179
 180     if (icpOutgoingConn != NULL && icpOutgoingConn->fd == fd)
 181         return 1;
 182
 183     return 0;
 184 }
 185
 186 static int
 187 fdIsDns(int fd)
 188 {
 189     if (fd == DnsSocketA)
 190         return 1;
 191
 192     if (fd == DnsSocketB)
 193         return 1;
 194
 195     return 0;
 196 }
 197
 198 static int
 199 fdIsTcpListen(int fd)
 200 {
 201     for (const AnyP::PortCfg *s = Config.Sockaddr.http; s; s = s->next) {
 202         if (s->listenConn != NULL && s->listenConn->fd == fd)
 203             return 1;
 204     }
 205
 206     return 0;
 207 }
 208
 209 static int
 210 comm_check_incoming_poll_handlers(int nfds, int *fds)
 211 {
 212     int i;
 213     int fd;
 214     PF *hdl = NULL;
 215     int npfds;
 216
 217     struct pollfd pfds[3 + MAXTCPLISTENPORTS];
 218     PROF_start(comm_check_incoming);
 219     incoming_sockets_accepted = 0;
 220
 221     for (i = npfds = 0; i < nfds; ++i) {
 222         int events;
 223         fd = fds[i];
 224         events = 0;
 225
 226         if (fd_table[fd].read_handler)
 227             events |= POLLRDNORM;
 228
 229         if (fd_table[fd].write_handler)
 230             events |= POLLWRNORM;
 231
 232         if (events) {
 233             pfds[npfds].fd = fd;
 234             pfds[npfds].events = events;
 235             pfds[npfds].revents = 0;
 236             ++npfds;
 237         }
 238     }
 239
 240     if (!nfds) {
 241         PROF_stop(comm_check_incoming);
 242         return -1;
 243     }
 244
 245     getCurrentTime();
 246     ++ statCounter.syscalls.selects;
 247
 248     if (poll(pfds, npfds, 0) < 1) {
 249         PROF_stop(comm_check_incoming);
 250         return incoming_sockets_accepted;
 251     }
 252
 253     for (i = 0; i < npfds; ++i) {
 254         int revents;
 255
 256         if (((revents = pfds[i].revents) == 0) || ((fd = pfds[i].fd) == -1))
 257             continue;
 258
 259         if (revents & (POLLRDNORM | POLLIN | POLLHUP | POLLERR)) {
 260             if ((hdl = fd_table[fd].read_handler)) {
 261                 fd_table[fd].read_handler = NULL;
 262                 hdl(fd, fd_table[fd].read_data);
 263             } else if (pfds[i].events & POLLRDNORM)
 264                 debugs(5, DBG_IMPORTANT, "comm_poll_incoming: FD " << fd << " NULL read handler");
 265         }
 266
 267         if (revents & (POLLWRNORM | POLLOUT | POLLHUP | POLLERR)) {
 268             if ((hdl = fd_table[fd].write_handler)) {
 269                 fd_table[fd].write_handler = NULL;
 270                 hdl(fd, fd_table[fd].write_data);
 271             } else if (pfds[i].events & POLLWRNORM)
 272                 debugs(5, DBG_IMPORTANT, "comm_poll_incoming: FD " << fd << " NULL write_handler");
 273         }
 274     }
 275
 276     PROF_stop(comm_check_incoming);
 277     return incoming_sockets_accepted;
 278 }
 279
 280 static void
 281 comm_poll_udp_incoming(void)
 282 {
 283     int nfds = 0;
 284     int fds[2];
 285     int nevents;
 286     udp_io_events = 0;
 287
 288     if (Comm::IsConnOpen(icpIncomingConn)) {
 289         fds[nfds] = icpIncomingConn->fd;
 290         ++nfds;
 291     }
 292
 293     if (icpIncomingConn != icpOutgoingConn && Comm::IsConnOpen(icpOutgoingConn)) {
 294         fds[nfds] = icpOutgoingConn->fd;
 295         ++nfds;
 296     }
 297
 298     if (nfds == 0)
 299         return;
 300
 301     nevents = comm_check_incoming_poll_handlers(nfds, fds);
 302
 303     incoming_udp_interval += Config.comm_incoming.udp.average - nevents;
 304
 305     if (incoming_udp_interval < Config.comm_incoming.udp.min_poll)
 306         incoming_udp_interval = Config.comm_incoming.udp.min_poll;
 307
 308     if (incoming_udp_interval > MAX_INCOMING_INTERVAL)
 309         incoming_udp_interval = MAX_INCOMING_INTERVAL;
 310
 311     if (nevents > INCOMING_UDP_MAX)
 312         nevents = INCOMING_UDP_MAX;
 313
 314     statCounter.comm_udp_incoming.count(nevents);
 315 }
 316
 317 static void
 318 comm_poll_tcp_incoming(void)
 319 {
 320     int nfds = 0;
 321     int fds[MAXTCPLISTENPORTS];
 322     int j;
 323     int nevents;
 324     tcp_io_events = 0;
 325
 326     // XXX: only poll sockets that won't be deferred. But how do we identify them?
 327
 328     for (j = 0; j < NHttpSockets; ++j) {
 329         if (HttpSockets[j] < 0)
 330             continue;
 331
 332         fds[nfds] = HttpSockets[j];
 333         ++nfds;
 334     }
 335
 336     nevents = comm_check_incoming_poll_handlers(nfds, fds);
 337     incoming_tcp_interval = incoming_tcp_interval
 338                             + Config.comm_incoming.tcp.average - nevents;
 339
 340     if (incoming_tcp_interval < Config.comm_incoming.tcp.min_poll)
 341         incoming_tcp_interval = Config.comm_incoming.tcp.min_poll;
 342
 343     if (incoming_tcp_interval > MAX_INCOMING_INTERVAL)
 344         incoming_tcp_interval = MAX_INCOMING_INTERVAL;
 345
 346     if (nevents > INCOMING_TCP_MAX)
 347         nevents = INCOMING_TCP_MAX;
 348
 349     statCounter.comm_tcp_incoming.count(nevents);
 350 }
 351
 352 /* poll all sockets; call handlers for those that are ready. */
 353 comm_err_t
 354 Comm::DoSelect(int msec)
 355 {
 356     struct pollfd pfds[SQUID_MAXFD];
 357
 358     PF *hdl = NULL;
 359     int fd;
 360     int maxfd;
 361     unsigned long nfds;
 362     unsigned long npending;
 363     int num;
 364     int calldns = 0, calludp = 0, calltcp = 0;
 365     double timeout = current_dtime + (msec / 1000.0);
 366
 367     do {
 368         double start;
 369         getCurrentTime();
 370         start = current_dtime;
 371
 372         if (commCheckUdpIncoming)
 373             comm_poll_udp_incoming();
 374
 375         if (commCheckDnsIncoming)
 376             comm_poll_dns_incoming();
 377
 378         if (commCheckTcpIncoming)
 379             comm_poll_tcp_incoming();
 380
 381         PROF_start(comm_poll_prep_pfds);
 382
 383         calldns = calludp = calltcp = 0;
 384
 385         nfds = 0;
 386
 387         npending = 0;
 388
 389         maxfd = Biggest_FD + 1;
 390
 391         for (int i = 0; i < maxfd; ++i) {
 392             int events;
 393             events = 0;
 394             /* Check each open socket for a handler. */
 395
 396             if (fd_table[i].read_handler)
 397                 events |= POLLRDNORM;
 398
 399             if (fd_table[i].write_handler)
 400                 events |= POLLWRNORM;
 401
 402             if (events) {
 403                 pfds[nfds].fd = i;
 404                 pfds[nfds].events = events;
 405                 pfds[nfds].revents = 0;
 406                 ++nfds;
 407
 408                 if ((events & POLLRDNORM) && fd_table[i].flags.read_pending)
 409                     ++npending;
 410             }
 411         }
 412
 413         PROF_stop(comm_poll_prep_pfds);
 414
 415         if (npending)
 416             msec = 0;
 417
 418         if (msec > MAX_POLL_TIME)
 419             msec = MAX_POLL_TIME;
 420
 421         /* nothing to do
 422          *
 423          * Note that this will only ever trigger when there are no log files
 424          * and stdout/err/in are all closed too.
 425          */
 426         if (nfds == 0 && npending == 0) {
 427             if (shutting_down)
 428                 return COMM_SHUTDOWN;
 429             else
 430                 return COMM_IDLE;
 431         }
 432
 433         for (;;) {
 434             PROF_start(comm_poll_normal);
 435             ++ statCounter.syscalls.selects;
 436             num = poll(pfds, nfds, msec);
 437             ++ statCounter.select_loops;
 438             PROF_stop(comm_poll_normal);
 439
 440             if (num >= 0 || npending > 0)
 441                 break;
 442
 443             if (ignoreErrno(errno))
 444                 continue;
 445
 446             debugs(5, DBG_CRITICAL, "comm_poll: poll failure: " << xstrerror());
 447
 448             assert(errno != EINVAL);
 449
 450             return COMM_ERROR;
 451
 452             /* NOTREACHED */
 453         }
 454
 455         getCurrentTime();
 456
 457         debugs(5, num ? 5 : 8, "comm_poll: " << num << "+" << npending << " FDs ready");
 458         statCounter.select_fds_hist.count(num);
 459
 460         if (num == 0 && npending == 0)
 461             continue;
 462
 463         /* scan each socket but the accept socket. Poll this
 464          * more frequently to minimize losses due to the 5 connect
 465          * limit in SunOS */
 466         PROF_start(comm_handle_ready_fd);
 467
 468         for (size_t loopIndex = 0; loopIndex < nfds; ++loopIndex) {
 469             fde *F;
 470             int revents = pfds[loopIndex].revents;
 471             fd = pfds[loopIndex].fd;
 472
 473             if (fd == -1)
 474                 continue;
 475
 476             if (fd_table[fd].flags.read_pending)
 477                 revents |= POLLIN;
 478
 479             if (revents == 0)
 480                 continue;
 481
 482             if (fdIsUdpListen(fd)) {
 483                 calludp = 1;
 484                 continue;
 485             }
 486
 487             if (fdIsDns(fd)) {
 488                 calldns = 1;
 489                 continue;
 490             }
 491
 492             if (fdIsTcpListen(fd)) {
 493                 calltcp = 1;
 494                 continue;
 495             }
 496
 497             F = &fd_table[fd];
 498
 499             if (revents & (POLLRDNORM | POLLIN | POLLHUP | POLLERR)) {
 500                 debugs(5, 6, "comm_poll: FD " << fd << " ready for reading");
 501
 502                 if ((hdl = F->read_handler)) {
 503                     PROF_start(comm_read_handler);
 504                     F->read_handler = NULL;
 505                     F->flags.read_pending = 0;
 506                     hdl(fd, F->read_data);
 507                     PROF_stop(comm_read_handler);
 508                     ++ statCounter.select_fds;
 509
 510                     if (commCheckUdpIncoming)
 511                         comm_poll_udp_incoming();
 512
 513                     if (commCheckDnsIncoming)
 514                         comm_poll_dns_incoming();
 515
 516                     if (commCheckTcpIncoming)
 517                         comm_poll_tcp_incoming();
 518                 }
 519             }
 520
 521             if (revents & (POLLWRNORM | POLLOUT | POLLHUP | POLLERR)) {
 522                 debugs(5, 6, "comm_poll: FD " << fd << " ready for writing");
 523
 524                 if ((hdl = F->write_handler)) {
 525                     PROF_start(comm_write_handler);
 526                     F->write_handler = NULL;
 527                     hdl(fd, F->write_data);
 528                     PROF_stop(comm_write_handler);
 529                     ++ statCounter.select_fds;
 530
 531                     if (commCheckUdpIncoming)
 532                         comm_poll_udp_incoming();
 533
 534                     if (commCheckDnsIncoming)
 535                         comm_poll_dns_incoming();
 536
 537                     if (commCheckTcpIncoming)
 538                         comm_poll_tcp_incoming();
 539                 }
 540             }
 541
 542             if (revents & POLLNVAL) {
 543                 AsyncCall::Pointer ch;
 544                 debugs(5, DBG_CRITICAL, "WARNING: FD " << fd << " has handlers, but it's invalid.");
 545                 debugs(5, DBG_CRITICAL, "FD " << fd << " is a " << fdTypeStr[F->type]);
 546                 debugs(5, DBG_CRITICAL, "--> " << F->desc);
 547                 debugs(5, DBG_CRITICAL, "tmout:" << F->timeoutHandler << "read:" <<
 548                        F->read_handler << " write:" << F->write_handler);
 549
 550                 for (ch = F->closeHandler; ch != NULL; ch = ch->Next())
 551                     debugs(5, DBG_CRITICAL, " close handler: " << ch);
 552
 553                 if (F->closeHandler != NULL) {
 554                     commCallCloseHandlers(fd);
 555                 } else if (F->timeoutHandler != NULL) {
 556                     debugs(5, DBG_CRITICAL, "comm_poll: Calling Timeout Handler");
 557                     ScheduleCallHere(F->timeoutHandler);
 558                 }
 559
 560                 F->closeHandler = NULL;
 561                 F->timeoutHandler = NULL;
 562                 F->read_handler = NULL;
 563                 F->write_handler = NULL;
 564
 565                 if (F->flags.open)
 566                     fd_close(fd);
 567             }
 568         }
 569
 570         PROF_stop(comm_handle_ready_fd);
 571
 572         if (calludp)
 573             comm_poll_udp_incoming();
 574
 575         if (calldns)
 576             comm_poll_dns_incoming();
 577
 578         if (calltcp)
 579             comm_poll_tcp_incoming();
 580
 581         getCurrentTime();
 582
 583         statCounter.select_time += (current_dtime - start);
 584
 585         return COMM_OK;
 586     } while (timeout > current_dtime);
 587
 588     debugs(5, 8, "comm_poll: time out: " << squid_curtime << ".");
 589
 590     return COMM_TIMEOUT;
 591 }
 592
 593 static void
 594 comm_poll_dns_incoming(void)
 595 {
 596     int nfds = 0;
 597     int fds[2];
 598     int nevents;
 599     dns_io_events = 0;
 600
 601     if (DnsSocketA < 0 && DnsSocketB < 0)
 602         return;
 603
 604     if (DnsSocketA >= 0) {
 605         fds[nfds] = DnsSocketA;
 606         ++nfds;
 607     }
 608
 609     if (DnsSocketB >= 0) {
 610         fds[nfds] = DnsSocketB;
 611         ++nfds;
 612     }
 613
 614     nevents = comm_check_incoming_poll_handlers(nfds, fds);
 615
 616     if (nevents < 0)
 617         return;
 618
 619     incoming_dns_interval += Config.comm_incoming.dns.average - nevents;
 620
 621     if (incoming_dns_interval < Config.comm_incoming.dns.min_poll)
 622         incoming_dns_interval = Config.comm_incoming.dns.min_poll;
 623
 624     if (incoming_dns_interval > MAX_INCOMING_INTERVAL)
 625         incoming_dns_interval = MAX_INCOMING_INTERVAL;
 626
 627     if (nevents > INCOMING_DNS_MAX)
 628         nevents = INCOMING_DNS_MAX;
 629
 630     statCounter.comm_dns_incoming.count(nevents);
 631 }
 632
 633 static void
 634 commPollRegisterWithCacheManager(void)
 635 {
 636     Mgr::RegisterAction("comm_poll_incoming",
 637                         "comm_incoming() stats",
 638                         commIncomingStats, 0, 1);
 639 }
 640
 641 void
 642 Comm::SelectLoopInit(void)
 643 {
 644     commPollRegisterWithCacheManager();
 645 }
 646
 647 static void
 648 commIncomingStats(StoreEntry * sentry)
 649 {
 650     storeAppendPrintf(sentry, "Current incoming_udp_interval: %d\n",
 651                       incoming_udp_interval >> INCOMING_FACTOR);
 652     storeAppendPrintf(sentry, "Current incoming_dns_interval: %d\n",
 653                       incoming_dns_interval >> INCOMING_FACTOR);
 654     storeAppendPrintf(sentry, "Current incoming_tcp_interval: %d\n",
 655                       incoming_tcp_interval >> INCOMING_FACTOR);
 656     storeAppendPrintf(sentry, "\n");
 657     storeAppendPrintf(sentry, "Histogram of events per incoming socket type\n");
 658     storeAppendPrintf(sentry, "ICP Messages handled per comm_poll_udp_incoming() call:\n");
 659     statCounter.comm_udp_incoming.dump(sentry, statHistIntDumper);
 660     storeAppendPrintf(sentry, "DNS Messages handled per comm_poll_dns_incoming() call:\n");
 661     statCounter.comm_dns_incoming.dump(sentry, statHistIntDumper);
 662     storeAppendPrintf(sentry, "HTTP Messages handled per comm_poll_tcp_incoming() call:\n");
 663     statCounter.comm_tcp_incoming.dump(sentry, statHistIntDumper);
 664 }
 665
 666 /* Called by async-io or diskd to speed up the polling */
 667 void
 668 Comm::QuickPollRequired(void)
 669 {
 670     MAX_POLL_TIME = 10;
 671 }
 672
 673 #endif /* USE_POLL */