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