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