src/comm/ModPoll.cc

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