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