src/comm_poll.cc

   1
   2 /*
   3  * $Id: comm_poll.cc,v 1.26 2008/02/12 23:02:13 rousskov Exp $
   4  *
   5  * DEBUG: section 5     Socket Functions
   6  *
   7  * SQUID Web Proxy Cache          http://www.squid-cache.org/
   8  * ----------------------------------------------------------
   9  *
  10  *  Squid is the result of efforts by numerous individuals from
  11  *  the Internet community; see the CONTRIBUTORS file for full
  12  *  details.   Many organizations have provided support for Squid's
  13  *  development; see the SPONSORS file for full details.  Squid is
  14  *  Copyrighted (C) 2001 by the Regents of the University of
  15  *  California; see the COPYRIGHT file for full details.  Squid
  16  *  incorporates software developed and/or copyrighted by other
  17  *  sources; see the CREDITS file for full details.
  18  *
  19  *  This program is free software; you can redistribute it and/or modify
  20  *  it under the terms of the GNU General Public License as published by
  21  *  the Free Software Foundation; either version 2 of the License, or
  22  *  (at your option) any later version.
  23  *
  24  *  This program is distributed in the hope that it will be useful,
  25  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  26  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  27  *  GNU General Public License for more details.
  28  *
  29  *  You should have received a copy of the GNU General Public License
  30  *  along with this program; if not, write to the Free Software
  31  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  32  *
  33  */
  34
  35 #include "squid.h"
  36 #include "comm_poll.h"
  37 #include "CacheManager.h"
  38 #include "SquidTime.h"
  39 #include "Store.h"
  40 #include "fde.h"
  41
  42 #ifdef USE_POLL
  43
  44 static int MAX_POLL_TIME = 1000;        /* see also comm_quick_poll_required() */
  45
  46 #ifndef        howmany
  47 #define howmany(x, y)   (((x)+((y)-1))/(y))
  48 #endif
  49 #ifndef        NBBY
  50 #define        NBBY    8
  51 #endif
  52 #define FD_MASK_BYTES sizeof(fd_mask)
  53 #define FD_MASK_BITS (FD_MASK_BYTES*NBBY)
  54
  55 /* STATIC */
  56 static int fdIsHttp(int fd);
  57 static int fdIsIcp(int fd);
  58 static int fdIsDns(int fd);
  59 static OBJH commIncomingStats;
  60 static int comm_check_incoming_poll_handlers(int nfds, int *fds);
  61 static void comm_poll_dns_incoming(void);
  62
  63 /*
  64  * Automatic tuning for incoming requests:
  65  *
  66  * INCOMING sockets are the ICP and HTTP ports.  We need to check these
  67  * fairly regularly, but how often?  When the load increases, we
  68  * want to check the incoming sockets more often.  If we have a lot
  69  * of incoming ICP, then we need to check these sockets more than
  70  * if we just have HTTP.
  71  *
  72  * The variables 'incoming_icp_interval' and 'incoming_http_interval'
  73  * determine how many normal I/O events to process before checking
  74  * incoming sockets again.  Note we store the incoming_interval
  75  * multipled by a factor of (2^INCOMING_FACTOR) to have some
  76  * pseudo-floating point precision.
  77  *
  78  * The variable 'icp_io_events' and 'http_io_events' counts how many normal
  79  * I/O events have been processed since the last check on the incoming
  80  * sockets.  When io_events > incoming_interval, its time to check incoming
  81  * sockets.
  82  *
  83  * Every time we check incoming sockets, we count how many new messages
  84  * or connections were processed.  This is used to adjust the
  85  * incoming_interval for the next iteration.  The new incoming_interval
  86  * is calculated as the current incoming_interval plus what we would
  87  * like to see as an average number of events minus the number of
  88  * events just processed.
  89  *
  90  *  incoming_interval = incoming_interval + target_average - number_of_events_processed
  91  *
  92  * There are separate incoming_interval counters for both HTTP and ICP events
  93  *
  94  * You can see the current values of the incoming_interval's, as well as
  95  * a histogram of 'incoming_events' by asking the cache manager
  96  * for 'comm_incoming', e.g.:
  97  *
  98  *      % ./client mgr:comm_poll_incoming
  99  *
 100  * Caveats:
 101  *
 102  *      - We have MAX_INCOMING_INTEGER as a magic upper limit on
 103  *        incoming_interval for both types of sockets.  At the
 104  *        largest value the cache will effectively be idling.
 105  *
 106  *      - The higher the INCOMING_FACTOR, the slower the algorithm will
 107  *        respond to load spikes/increases/decreases in demand. A value
 108  *        between 3 and 8 is recommended.
 109  */
 110
 111 #define MAX_INCOMING_INTEGER 256
 112 #define INCOMING_FACTOR 5
 113 #define MAX_INCOMING_INTERVAL (MAX_INCOMING_INTEGER << INCOMING_FACTOR)
 114 static int icp_io_events = 0;
 115 static int dns_io_events = 0;
 116 static int http_io_events = 0;
 117 static int incoming_icp_interval = 16 << INCOMING_FACTOR;
 118 static int incoming_dns_interval = 16 << INCOMING_FACTOR;
 119 static int incoming_http_interval = 16 << INCOMING_FACTOR;
 120 #define commCheckICPIncoming (++icp_io_events > (incoming_icp_interval>> INCOMING_FACTOR))
 121 #define commCheckDNSIncoming (++dns_io_events > (incoming_dns_interval>> INCOMING_FACTOR))
 122 #define commCheckHTTPIncoming (++http_io_events > (incoming_http_interval>> INCOMING_FACTOR))
 123
 124
 125 void
 126 commSetSelect(int fd, unsigned int type, PF * handler, void *client_data,
 127               time_t timeout)
 128 {
 129     fde *F = &fd_table[fd];
 130     assert(fd >= 0);
 131     assert(F->flags.open);
 132     debugs(5, 5, "commSetSelect: FD " << fd << " type " << type);
 133
 134     if (type & COMM_SELECT_READ) {
 135         F->read_handler = handler;
 136         F->read_data = client_data;
 137     }
 138
 139     if (type & COMM_SELECT_WRITE) {
 140         F->write_handler = handler;
 141         F->write_data = client_data;
 142     }
 143
 144     if (timeout)
 145         F->timeout = squid_curtime + timeout;
 146 }
 147
 148 void
 149 commResetSelect(int fd)
 150 {
 151 }
 152
 153 static int
 154 fdIsIcp(int fd)
 155 {
 156     if (fd == theInIcpConnection)
 157         return 1;
 158
 159     if (fd == theOutIcpConnection)
 160         return 1;
 161
 162     return 0;
 163 }
 164
 165 static int
 166 fdIsDns(int fd)
 167 {
 168     if (fd == DnsSocket)
 169         return 1;
 170
 171     return 0;
 172 }
 173
 174 static int
 175 fdIsHttp(int fd)
 176 {
 177     int j;
 178
 179     for (j = 0; j < NHttpSockets; j++) {
 180         if (fd == HttpSockets[j])
 181             return 1;
 182     }
 183
 184     return 0;
 185 }
 186
 187 static int
 188 comm_check_incoming_poll_handlers(int nfds, int *fds)
 189 {
 190     int i;
 191     int fd;
 192     PF *hdl = NULL;
 193     int npfds;
 194
 195     struct pollfd pfds[3 + MAXHTTPPORTS];
 196     PROF_start(comm_check_incoming);
 197     incoming_sockets_accepted = 0;
 198
 199     for (i = npfds = 0; i < nfds; i++) {
 200         int events;
 201         fd = fds[i];
 202         events = 0;
 203
 204         if (fd_table[fd].read_handler)
 205             events |= POLLRDNORM;
 206
 207         if (fd_table[fd].write_handler)
 208             events |= POLLWRNORM;
 209
 210         if (events) {
 211             pfds[npfds].fd = fd;
 212             pfds[npfds].events = events;
 213             pfds[npfds].revents = 0;
 214             npfds++;
 215         }
 216     }
 217
 218     if (!nfds) {
 219         PROF_stop(comm_check_incoming);
 220         return -1;
 221     }
 222
 223     getCurrentTime();
 224     statCounter.syscalls.selects++;
 225
 226     if (poll(pfds, npfds, 0) < 1) {
 227         PROF_stop(comm_check_incoming);
 228         return incoming_sockets_accepted;
 229     }
 230
 231     for (i = 0; i < npfds; i++) {
 232         int revents;
 233
 234         if (((revents = pfds[i].revents) == 0) || ((fd = pfds[i].fd) == -1))
 235             continue;
 236
 237         if (revents & (POLLRDNORM | POLLIN | POLLHUP | POLLERR)) {
 238             if ((hdl = fd_table[fd].read_handler)) {
 239                 fd_table[fd].read_handler = NULL;
 240                 hdl(fd, fd_table[fd].read_data);
 241             } else if (pfds[i].events & POLLRDNORM)
 242                 debugs(5, 1, "comm_poll_incoming: FD " << fd << " NULL read handler");
 243         }
 244
 245         if (revents & (POLLWRNORM | POLLOUT | POLLHUP | POLLERR)) {
 246             if ((hdl = fd_table[fd].write_handler)) {
 247                 fd_table[fd].write_handler = NULL;
 248                 hdl(fd, fd_table[fd].write_data);
 249             } else if (pfds[i].events & POLLWRNORM)
 250                 debugs(5, 1, "comm_poll_incoming: FD " << fd << " NULL write_handler");
 251         }
 252     }
 253
 254     PROF_stop(comm_check_incoming);
 255     return incoming_sockets_accepted;
 256 }
 257
 258 static void
 259 comm_poll_icp_incoming(void)
 260 {
 261     int nfds = 0;
 262     int fds[2];
 263     int nevents;
 264     icp_io_events = 0;
 265
 266     if (theInIcpConnection >= 0)
 267         fds[nfds++] = theInIcpConnection;
 268
 269     if (theInIcpConnection != theOutIcpConnection)
 270         if (theOutIcpConnection >= 0)
 271             fds[nfds++] = theOutIcpConnection;
 272
 273     if (nfds == 0)
 274         return;
 275
 276     nevents = comm_check_incoming_poll_handlers(nfds, fds);
 277
 278     incoming_icp_interval += Config.comm_incoming.icp_average - nevents;
 279
 280     if (incoming_icp_interval < Config.comm_incoming.icp_min_poll)
 281         incoming_icp_interval = Config.comm_incoming.icp_min_poll;
 282
 283     if (incoming_icp_interval > MAX_INCOMING_INTERVAL)
 284         incoming_icp_interval = MAX_INCOMING_INTERVAL;
 285
 286     if (nevents > INCOMING_ICP_MAX)
 287         nevents = INCOMING_ICP_MAX;
 288
 289     statHistCount(&statCounter.comm_icp_incoming, nevents);
 290 }
 291
 292 static void
 293 comm_poll_http_incoming(void)
 294 {
 295     int nfds = 0;
 296     int fds[MAXHTTPPORTS];
 297     int j;
 298     int nevents;
 299     http_io_events = 0;
 300
 301     /* only poll sockets that won't be deferred */
 302
 303     for (j = 0; j < NHttpSockets; j++) {
 304         if (HttpSockets[j] < 0)
 305             continue;
 306
 307         fds[nfds++] = HttpSockets[j];
 308     }
 309
 310     nevents = comm_check_incoming_poll_handlers(nfds, fds);
 311     incoming_http_interval = incoming_http_interval
 312                              + Config.comm_incoming.http_average - nevents;
 313
 314     if (incoming_http_interval < Config.comm_incoming.http_min_poll)
 315         incoming_http_interval = Config.comm_incoming.http_min_poll;
 316
 317     if (incoming_http_interval > MAX_INCOMING_INTERVAL)
 318         incoming_http_interval = MAX_INCOMING_INTERVAL;
 319
 320     if (nevents > INCOMING_HTTP_MAX)
 321         nevents = INCOMING_HTTP_MAX;
 322
 323     statHistCount(&statCounter.comm_http_incoming, nevents);
 324 }
 325
 326 /* poll all sockets; call handlers for those that are ready. */
 327 comm_err_t
 328 comm_select(int msec)
 329 {
 330
 331     struct pollfd pfds[SQUID_MAXFD];
 332
 333     PF *hdl = NULL;
 334     int fd;
 335     int maxfd;
 336     unsigned long nfds;
 337     unsigned long npending;
 338     int num;
 339     int callicp = 0, callhttp = 0;
 340     int calldns = 0;
 341     double timeout = current_dtime + (msec / 1000.0);
 342
 343     do {
 344         double start;
 345         getCurrentTime();
 346         start = current_dtime;
 347
 348         if (commCheckICPIncoming)
 349             comm_poll_icp_incoming();
 350
 351         if (commCheckDNSIncoming)
 352             comm_poll_dns_incoming();
 353
 354         if (commCheckHTTPIncoming)
 355             comm_poll_http_incoming();
 356
 357         PROF_start(comm_poll_prep_pfds);
 358
 359         callicp = calldns = callhttp = 0;
 360
 361         nfds = 0;
 362
 363         npending = 0;
 364
 365         maxfd = Biggest_FD + 1;
 366
 367         for (int i = 0; i < maxfd; i++) {
 368             int events;
 369             events = 0;
 370             /* Check each open socket for a handler. */
 371
 372             if (fd_table[i].read_handler)
 373                 events |= POLLRDNORM;
 374
 375             if (fd_table[i].write_handler)
 376                 events |= POLLWRNORM;
 377
 378             if (events) {
 379                 pfds[nfds].fd = i;
 380                 pfds[nfds].events = events;
 381                 pfds[nfds].revents = 0;
 382                 nfds++;
 383
 384                 if ((events & POLLRDNORM) && fd_table[i].flags.read_pending)
 385                     npending++;
 386             }
 387         }
 388
 389         PROF_stop(comm_poll_prep_pfds);
 390
 391         if (npending)
 392             msec = 0;
 393
 394         if (msec > MAX_POLL_TIME)
 395             msec = MAX_POLL_TIME;
 396
 397         /* nothing to do
 398          *
 399          * Note that this will only ever trigger when there are no log files
 400          * and stdout/err/in are all closed too.
 401          */
 402         if (nfds == 0 && !npending) {
 403             if (shutting_down)
 404                 return COMM_SHUTDOWN;
 405             else
 406                 return COMM_IDLE;
 407         }
 408
 409         for (;;) {
 410             PROF_start(comm_poll_normal);
 411             statCounter.syscalls.selects++;
 412             num = poll(pfds, nfds, msec);
 413             statCounter.select_loops++;
 414             PROF_stop(comm_poll_normal);
 415
 416             if (num >= 0 || npending >= 0)
 417                 break;
 418
 419             if (ignoreErrno(errno))
 420                 continue;
 421
 422             debugs(5, 0, "comm_poll: poll failure: " << xstrerror());
 423
 424             assert(errno != EINVAL);
 425
 426             return COMM_ERROR;
 427
 428             /* NOTREACHED */
 429         }
 430
 431         getCurrentTime();
 432
 433         debugs(5, num ? 5 : 8, "comm_poll: " << num << "+" << npending << " FDs ready");
 434         statHistCount(&statCounter.select_fds_hist, num);
 435
 436         if (num == 0 && npending == 0)
 437             continue;
 438
 439         /* scan each socket but the accept socket. Poll this
 440          * more frequently to minimize losses due to the 5 connect
 441          * limit in SunOS */
 442         PROF_start(comm_handle_ready_fd);
 443
 444         for (size_t loopIndex = 0; loopIndex < nfds; loopIndex++) {
 445             fde *F;
 446             int revents = pfds[loopIndex].revents;
 447             fd = pfds[loopIndex].fd;
 448
 449             if (fd == -1)
 450                 continue;
 451
 452             if (fd_table[fd].flags.read_pending)
 453                 revents |= POLLIN;
 454
 455             if (revents == 0)
 456                 continue;
 457
 458             if (fdIsIcp(fd)) {
 459                 callicp = 1;
 460                 continue;
 461             }
 462
 463             if (fdIsDns(fd)) {
 464                 calldns = 1;
 465                 continue;
 466             }
 467
 468             if (fdIsHttp(fd)) {
 469                 callhttp = 1;
 470                 continue;
 471             }
 472
 473             F = &fd_table[fd];
 474
 475             if (revents & (POLLRDNORM | POLLIN | POLLHUP | POLLERR)) {
 476                 debugs(5, 6, "comm_poll: FD " << fd << " ready for reading");
 477
 478                 if (NULL == (hdl = F->read_handler))
 479                     (void) 0;
 480                 else {
 481                     PROF_start(comm_read_handler);
 482                     F->read_handler = NULL;
 483                     F->flags.read_pending = 0;
 484                     hdl(fd, F->read_data);
 485                     PROF_stop(comm_read_handler);
 486                     statCounter.select_fds++;
 487
 488                     if (commCheckICPIncoming)
 489                         comm_poll_icp_incoming();
 490
 491                     if (commCheckDNSIncoming)
 492                         comm_poll_dns_incoming();
 493
 494                     if (commCheckHTTPIncoming)
 495                         comm_poll_http_incoming();
 496                 }
 497             }
 498
 499             if (revents & (POLLWRNORM | POLLOUT | POLLHUP | POLLERR)) {
 500                 debugs(5, 5, "comm_poll: FD " << fd << " ready for writing");
 501
 502                 if ((hdl = F->write_handler)) {
 503                     PROF_start(comm_write_handler);
 504                     F->write_handler = NULL;
 505                     hdl(fd, F->write_data);
 506                     PROF_stop(comm_write_handler);
 507                     statCounter.select_fds++;
 508
 509                     if (commCheckICPIncoming)
 510                         comm_poll_icp_incoming();
 511
 512                     if (commCheckDNSIncoming)
 513                         comm_poll_dns_incoming();
 514
 515                     if (commCheckHTTPIncoming)
 516                         comm_poll_http_incoming();
 517                 }
 518             }
 519
 520             if (revents & POLLNVAL) {
 521                 AsyncCall::Pointer ch;
 522                 debugs(5, 0, "WARNING: FD " << fd << " has handlers, but it's invalid.");
 523                 debugs(5, 0, "FD " << fd << " is a " << fdTypeStr[F->type]);
 524                 debugs(5, 0, "--> " << F->desc);
 525                 debugs(5, 0, "tmout:" << F->timeoutHandler << "read:" <<
 526                        F->read_handler << " write:" << F->write_handler);
 527
 528                 for (ch = F->closeHandler; ch != NULL; ch = ch->Next())
 529                     debugs(5, 0, " close handler: " << ch);
 530
 531                 if (F->closeHandler != NULL) {
 532                     commCallCloseHandlers(fd);
 533                 } else if (F->timeoutHandler != NULL) {
 534                     debugs(5, 0, "comm_poll: Calling Timeout Handler");
 535                     ScheduleCallHere(F->timeoutHandler);
 536                 }
 537
 538                 F->closeHandler = NULL;
 539                 F->timeoutHandler = NULL;
 540                 F->read_handler = NULL;
 541                 F->write_handler = NULL;
 542
 543                 if (F->flags.open)
 544                     fd_close(fd);
 545             }
 546         }
 547
 548         PROF_stop(comm_handle_ready_fd);
 549
 550         if (callicp)
 551             comm_poll_icp_incoming();
 552
 553         if (calldns)
 554             comm_poll_dns_incoming();
 555
 556         if (callhttp)
 557             comm_poll_http_incoming();
 558
 559         getCurrentTime();
 560
 561         statCounter.select_time += (current_dtime - start);
 562
 563         return COMM_OK;
 564     } while (timeout > current_dtime);
 565
 566     debugs(5, 8, "comm_poll: time out: " << squid_curtime << ".");
 567
 568     return COMM_TIMEOUT;
 569 }
 570
 571
 572 static void
 573 comm_poll_dns_incoming(void)
 574 {
 575     int nfds = 0;
 576     int fds[2];
 577     int nevents;
 578     dns_io_events = 0;
 579
 580     if (DnsSocket < 0)
 581         return;
 582
 583     fds[nfds++] = DnsSocket;
 584
 585     nevents = comm_check_incoming_poll_handlers(nfds, fds);
 586
 587     if (nevents < 0)
 588         return;
 589
 590     incoming_dns_interval += Config.comm_incoming.dns_average - nevents;
 591
 592     if (incoming_dns_interval < Config.comm_incoming.dns_min_poll)
 593         incoming_dns_interval = Config.comm_incoming.dns_min_poll;
 594
 595     if (incoming_dns_interval > MAX_INCOMING_INTERVAL)
 596         incoming_dns_interval = MAX_INCOMING_INTERVAL;
 597
 598     if (nevents > INCOMING_DNS_MAX)
 599         nevents = INCOMING_DNS_MAX;
 600
 601     statHistCount(&statCounter.comm_dns_incoming, nevents);
 602 }
 603
 604 void
 605 comm_select_init(void)
 606 {}
 607
 608 void
 609 commPollRegisterWithCacheManager(CacheManager & manager)
 610 {
 611     manager.registerAction("comm_poll_incoming",
 612                            "comm_incoming() stats",
 613                            commIncomingStats, 0, 1);
 614 }
 615
 616
 617 static void
 618 commIncomingStats(StoreEntry * sentry)
 619 {
 620     StatCounters *f = &statCounter;
 621     storeAppendPrintf(sentry, "Current incoming_icp_interval: %d\n",
 622                       incoming_icp_interval >> INCOMING_FACTOR);
 623     storeAppendPrintf(sentry, "Current incoming_dns_interval: %d\n",
 624                       incoming_dns_interval >> INCOMING_FACTOR);
 625     storeAppendPrintf(sentry, "Current incoming_http_interval: %d\n",
 626                       incoming_http_interval >> INCOMING_FACTOR);
 627     storeAppendPrintf(sentry, "\n");
 628     storeAppendPrintf(sentry, "Histogram of events per incoming socket type\n");
 629     storeAppendPrintf(sentry, "ICP Messages handled per comm_poll_icp_incoming() call:\n");
 630     statHistDump(&f->comm_icp_incoming, sentry, statHistIntDumper);
 631     storeAppendPrintf(sentry, "DNS Messages handled per comm_poll_dns_incoming() call:\n");
 632     statHistDump(&f->comm_dns_incoming, sentry, statHistIntDumper);
 633     storeAppendPrintf(sentry, "HTTP Messages handled per comm_poll_http_incoming() call:\n");
 634     statHistDump(&f->comm_http_incoming, sentry, statHistIntDumper);
 635 }
 636
 637 /* Called by async-io or diskd to speed up the polling */
 638 void
 639 comm_quick_poll_required(void)
 640 {
 641     MAX_POLL_TIME = 10;
 642 }
 643
 644 #endif /* USE_POLL */