src/comm/ModPoll.cc

   1 /*
   2  * Copyright (C) 1996-2023 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 "SquidConfig.h"
  23 #include "StatCounters.h"
  24 #include "Store.h"
  25
  26 #include <cerrno>
  27 #if HAVE_POLL_H
  28 #include <poll.h>
  29 #endif
  30
  31 /* Needed for poll() on Linux at least */
  32 #if USE_POLL
  33 #ifndef POLLRDNORM
  34 #define POLLRDNORM POLLIN
  35 #endif
  36 #ifndef POLLWRNORM
  37 #define POLLWRNORM POLLOUT
  38 #endif
  39 #endif
  40
  41 static int MAX_POLL_TIME = 1000;    /* see also Comm::QuickPollRequired() */
  42
  43 #ifndef        howmany
  44 #define howmany(x, y)   (((x)+((y)-1))/(y))
  45 #endif
  46 #ifndef        NBBY
  47 #define        NBBY    8
  48 #endif
  49 #define FD_MASK_BYTES sizeof(fd_mask)
  50 #define FD_MASK_BITS (FD_MASK_BYTES*NBBY)
  51
  52 /* STATIC */
  53 static int fdIsTcpListen(int fd);
  54 static int fdIsUdpListen(int fd);
  55 static int fdIsDns(int fd);
  56 static OBJH commIncomingStats;
  57 static int comm_check_incoming_poll_handlers(int nfds, int *fds);
  58 static void comm_poll_dns_incoming(void);
  59
  60 void
  61 Comm::SetSelect(int fd, unsigned int type, PF * handler, void *client_data, time_t timeout)
  62 {
  63     fde *F = &fd_table[fd];
  64     assert(fd >= 0);
  65     assert(F->flags.open || (!handler && !client_data && !timeout));
  66     debugs(5, 5, "FD " << fd << ", type=" << type <<
  67            ", handler=" << handler << ", client_data=" << client_data <<
  68            ", timeout=" << timeout);
  69
  70     if (type & COMM_SELECT_READ) {
  71         F->read_handler = handler;
  72         F->read_data = client_data;
  73     }
  74
  75     if (type & COMM_SELECT_WRITE) {
  76         F->write_handler = handler;
  77         F->write_data = client_data;
  78     }
  79
  80     if (timeout)
  81         F->timeout = squid_curtime + timeout;
  82 }
  83
  84 static int
  85 fdIsUdpListen(int fd)
  86 {
  87     if (icpIncomingConn != nullptr && icpIncomingConn->fd == fd)
  88         return 1;
  89
  90     if (icpOutgoingConn != nullptr && icpOutgoingConn->fd == fd)
  91         return 1;
  92
  93     return 0;
  94 }
  95
  96 static int
  97 fdIsDns(int fd)
  98 {
  99     if (fd == DnsSocketA)
 100         return 1;
 101
 102     if (fd == DnsSocketB)
 103         return 1;
 104
 105     return 0;
 106 }
 107
 108 static int
 109 fdIsTcpListen(int fd)
 110 {
 111     for (AnyP::PortCfgPointer s = HttpPortList; s != nullptr; s = s->next) {
 112         if (s->listenConn != nullptr && s->listenConn->fd == fd)
 113             return 1;
 114     }
 115
 116     return 0;
 117 }
 118
 119 static int
 120 comm_check_incoming_poll_handlers(int nfds, int *fds)
 121 {
 122     int i;
 123     int fd;
 124     PF *hdl = nullptr;
 125     int npfds;
 126
 127     struct pollfd pfds[3 + MAXTCPLISTENPORTS];
 128     incoming_sockets_accepted = 0;
 129
 130     for (i = npfds = 0; i < nfds; ++i) {
 131         int events;
 132         fd = fds[i];
 133         events = 0;
 134
 135         if (fd_table[fd].read_handler)
 136             events |= POLLRDNORM;
 137
 138         if (fd_table[fd].write_handler)
 139             events |= POLLWRNORM;
 140
 141         if (events) {
 142             pfds[npfds].fd = fd;
 143             pfds[npfds].events = events;
 144             pfds[npfds].revents = 0;
 145             ++npfds;
 146         }
 147     }
 148
 149     if (!nfds)
 150         return -1;
 151
 152     getCurrentTime();
 153     ++ statCounter.syscalls.selects;
 154
 155     if (poll(pfds, npfds, 0) < 1)
 156         return incoming_sockets_accepted;
 157
 158     for (i = 0; i < npfds; ++i) {
 159         int revents;
 160
 161         if (((revents = pfds[i].revents) == 0) || ((fd = pfds[i].fd) == -1))
 162             continue;
 163
 164         if (revents & (POLLRDNORM | POLLIN | POLLHUP | POLLERR)) {
 165             if ((hdl = fd_table[fd].read_handler)) {
 166                 fd_table[fd].read_handler = nullptr;
 167                 hdl(fd, fd_table[fd].read_data);
 168             } else if (pfds[i].events & POLLRDNORM)
 169                 debugs(5, DBG_IMPORTANT, "comm_poll_incoming: FD " << fd << " NULL read handler");
 170         }
 171
 172         if (revents & (POLLWRNORM | POLLOUT | POLLHUP | POLLERR)) {
 173             if ((hdl = fd_table[fd].write_handler)) {
 174                 fd_table[fd].write_handler = nullptr;
 175                 hdl(fd, fd_table[fd].write_data);
 176             } else if (pfds[i].events & POLLWRNORM)
 177                 debugs(5, DBG_IMPORTANT, "comm_poll_incoming: FD " << fd << " NULL write_handler");
 178         }
 179     }
 180
 181     return incoming_sockets_accepted;
 182 }
 183
 184 static void
 185 comm_poll_udp_incoming(void)
 186 {
 187     int nfds = 0;
 188     int fds[2];
 189
 190     if (Comm::IsConnOpen(icpIncomingConn)) {
 191         fds[nfds] = icpIncomingConn->fd;
 192         ++nfds;
 193     }
 194
 195     if (icpIncomingConn != icpOutgoingConn && Comm::IsConnOpen(icpOutgoingConn)) {
 196         fds[nfds] = icpOutgoingConn->fd;
 197         ++nfds;
 198     }
 199
 200     if (statCounter.comm_udp.startPolling(nfds)) {
 201         auto n = comm_check_incoming_poll_handlers(nfds, fds);
 202         statCounter.comm_udp.finishPolling(n, Config.comm_incoming.udp);
 203     }
 204 }
 205
 206 static void
 207 comm_poll_tcp_incoming(void)
 208 {
 209     int nfds = 0;
 210     int fds[MAXTCPLISTENPORTS];
 211
 212     // XXX: only poll sockets that won't be deferred. But how do we identify them?
 213
 214     for (AnyP::PortCfgPointer s = HttpPortList; s != nullptr; s = s->next) {
 215         if (Comm::IsConnOpen(s->listenConn)) {
 216             fds[nfds] = s->listenConn->fd;
 217             ++nfds;
 218         }
 219     }
 220
 221     if (statCounter.comm_tcp.startPolling(nfds)) {
 222         auto n = comm_check_incoming_poll_handlers(nfds, fds);
 223         statCounter.comm_tcp.finishPolling(n, Config.comm_incoming.tcp);
 224     }
 225 }
 226
 227 /* poll all sockets; call handlers for those that are ready. */
 228 Comm::Flag
 229 Comm::DoSelect(int msec)
 230 {
 231     struct pollfd pfds[SQUID_MAXFD];
 232
 233     PF *hdl = nullptr;
 234     int fd;
 235     int maxfd;
 236     unsigned long nfds;
 237     unsigned long npending;
 238     int num;
 239     int calldns = 0, calludp = 0, calltcp = 0;
 240     double timeout = current_dtime + (msec / 1000.0);
 241
 242     do {
 243         double start;
 244         getCurrentTime();
 245         start = current_dtime;
 246
 247         if (statCounter.comm_udp.check())
 248             comm_poll_udp_incoming();
 249
 250         if (statCounter.comm_dns.check())
 251             comm_poll_dns_incoming();
 252
 253         if (statCounter.comm_tcp.check())
 254             comm_poll_tcp_incoming();
 255
 256         calldns = calludp = calltcp = 0;
 257
 258         nfds = 0;
 259
 260         npending = 0;
 261
 262         maxfd = Biggest_FD + 1;
 263
 264         for (int i = 0; i < maxfd; ++i) {
 265             int events;
 266             events = 0;
 267             /* Check each open socket for a handler. */
 268
 269             if (fd_table[i].read_handler)
 270                 events |= POLLRDNORM;
 271
 272             if (fd_table[i].write_handler)
 273                 events |= POLLWRNORM;
 274
 275             if (events) {
 276                 pfds[nfds].fd = i;
 277                 pfds[nfds].events = events;
 278                 pfds[nfds].revents = 0;
 279                 ++nfds;
 280
 281                 if ((events & POLLRDNORM) && fd_table[i].flags.read_pending)
 282                     ++npending;
 283             }
 284         }
 285
 286         if (npending)
 287             msec = 0;
 288
 289         if (msec > MAX_POLL_TIME)
 290             msec = MAX_POLL_TIME;
 291
 292         /* nothing to do
 293          *
 294          * Note that this will only ever trigger when there are no log files
 295          * and stdout/err/in are all closed too.
 296          */
 297         if (nfds == 0 && npending == 0) {
 298             if (shutting_down)
 299                 return Comm::SHUTDOWN;
 300             else
 301                 return Comm::IDLE;
 302         }
 303
 304         for (;;) {
 305             ++ statCounter.syscalls.selects;
 306             num = poll(pfds, nfds, msec);
 307             int xerrno = errno;
 308             ++ statCounter.select_loops;
 309
 310             if (num >= 0 || npending > 0)
 311                 break;
 312
 313             if (ignoreErrno(xerrno))
 314                 continue;
 315
 316             debugs(5, DBG_CRITICAL, MYNAME << "poll failure: " << xstrerr(xerrno));
 317
 318             assert(xerrno != EINVAL);
 319
 320             return Comm::COMM_ERROR;
 321
 322             /* NOTREACHED */
 323         }
 324
 325         getCurrentTime();
 326
 327         debugs(5, num ? 5 : 8, "comm_poll: " << num << "+" << npending << " FDs ready");
 328         statCounter.select_fds_hist.count(num);
 329
 330         if (num == 0 && npending == 0)
 331             continue;
 332
 333         /* scan each socket but the accept socket. Poll this
 334          * more frequently to minimize losses due to the 5 connect
 335          * limit in SunOS */
 336
 337         for (size_t loopIndex = 0; loopIndex < nfds; ++loopIndex) {
 338             fde *F;
 339             int revents = pfds[loopIndex].revents;
 340             fd = pfds[loopIndex].fd;
 341
 342             if (fd == -1)
 343                 continue;
 344
 345             if (fd_table[fd].flags.read_pending)
 346                 revents |= POLLIN;
 347
 348             if (revents == 0)
 349                 continue;
 350
 351             if (fdIsUdpListen(fd)) {
 352                 calludp = 1;
 353                 continue;
 354             }
 355
 356             if (fdIsDns(fd)) {
 357                 calldns = 1;
 358                 continue;
 359             }
 360
 361             if (fdIsTcpListen(fd)) {
 362                 calltcp = 1;
 363                 continue;
 364             }
 365
 366             F = &fd_table[fd];
 367
 368             if (revents & (POLLRDNORM | POLLIN | POLLHUP | POLLERR)) {
 369                 debugs(5, 6, "comm_poll: FD " << fd << " ready for reading");
 370
 371                 if ((hdl = F->read_handler)) {
 372                     F->read_handler = nullptr;
 373                     hdl(fd, F->read_data);
 374                     ++ statCounter.select_fds;
 375
 376                     if (statCounter.comm_udp.check())
 377                         comm_poll_udp_incoming();
 378
 379                     if (statCounter.comm_dns.check())
 380                         comm_poll_dns_incoming();
 381
 382                     if (statCounter.comm_tcp.check())
 383                         comm_poll_tcp_incoming();
 384                 }
 385             }
 386
 387             if (revents & (POLLWRNORM | POLLOUT | POLLHUP | POLLERR)) {
 388                 debugs(5, 6, "comm_poll: FD " << fd << " ready for writing");
 389
 390                 if ((hdl = F->write_handler)) {
 391                     F->write_handler = nullptr;
 392                     hdl(fd, F->write_data);
 393                     ++ statCounter.select_fds;
 394
 395                     if (statCounter.comm_udp.check())
 396                         comm_poll_udp_incoming();
 397
 398                     if (statCounter.comm_dns.check())
 399                         comm_poll_dns_incoming();
 400
 401                     if (statCounter.comm_tcp.check())
 402                         comm_poll_tcp_incoming();
 403                 }
 404             }
 405
 406             if (revents & POLLNVAL) {
 407                 AsyncCall::Pointer ch;
 408                 debugs(5, DBG_CRITICAL, "WARNING: FD " << fd << " has handlers, but it's invalid.");
 409                 debugs(5, DBG_CRITICAL, "FD " << fd << " is a " << fdTypeStr[F->type]);
 410                 debugs(5, DBG_CRITICAL, "--> " << F->desc);
 411                 debugs(5, DBG_CRITICAL, "tmout:" << F->timeoutHandler << "read:" <<
 412                        F->read_handler << " write:" << F->write_handler);
 413
 414                 for (ch = F->closeHandler; ch != nullptr; ch = ch->Next())
 415                     debugs(5, DBG_CRITICAL, " close handler: " << ch);
 416
 417                 if (F->closeHandler != nullptr) {
 418                     commCallCloseHandlers(fd);
 419                 } else if (F->timeoutHandler != nullptr) {
 420                     debugs(5, DBG_CRITICAL, "comm_poll: Calling Timeout Handler");
 421                     ScheduleCallHere(F->timeoutHandler);
 422                 }
 423
 424                 F->closeHandler = nullptr;
 425                 F->timeoutHandler = nullptr;
 426                 F->read_handler = nullptr;
 427                 F->write_handler = nullptr;
 428
 429                 if (F->flags.open)
 430                     fd_close(fd);
 431             }
 432         }
 433
 434         if (calludp)
 435             comm_poll_udp_incoming();
 436
 437         if (calldns)
 438             comm_poll_dns_incoming();
 439
 440         if (calltcp)
 441             comm_poll_tcp_incoming();
 442
 443         getCurrentTime();
 444
 445         statCounter.select_time += (current_dtime - start);
 446
 447         return Comm::OK;
 448     } while (timeout > current_dtime);
 449
 450     debugs(5, 8, "comm_poll: time out: " << squid_curtime << ".");
 451
 452     return Comm::TIMEOUT;
 453 }
 454
 455 static void
 456 comm_poll_dns_incoming(void)
 457 {
 458     int nfds = 0;
 459     int fds[2];
 460
 461     if (DnsSocketA >= 0) {
 462         fds[nfds] = DnsSocketA;
 463         ++nfds;
 464     }
 465
 466     if (DnsSocketB >= 0) {
 467         fds[nfds] = DnsSocketB;
 468         ++nfds;
 469     }
 470
 471     if (statCounter.comm_dns.startPolling(nfds)) {
 472         auto n = comm_check_incoming_poll_handlers(nfds, fds);
 473         statCounter.comm_dns.finishPolling(n, Config.comm_incoming.dns);
 474     }
 475 }
 476
 477 static void
 478 commPollRegisterWithCacheManager(void)
 479 {
 480     Mgr::RegisterAction("comm_poll_incoming",
 481                         "comm_incoming() stats",
 482                         commIncomingStats, 0, 1);
 483 }
 484
 485 void
 486 Comm::SelectLoopInit(void)
 487 {
 488     commPollRegisterWithCacheManager();
 489 }
 490
 491 static void
 492 commIncomingStats(StoreEntry * sentry)
 493 {
 494     storeAppendPrintf(sentry, "Current incoming_udp_interval: %d\n",
 495                       statCounter.comm_udp.interval >> Comm::Incoming::Factor);
 496     storeAppendPrintf(sentry, "Current incoming_dns_interval: %d\n",
 497                       statCounter.comm_dns.interval >> Comm::Incoming::Factor);
 498     storeAppendPrintf(sentry, "Current incoming_tcp_interval: %d\n",
 499                       statCounter.comm_tcp.interval >> Comm::Incoming::Factor);
 500     storeAppendPrintf(sentry, "\n");
 501     storeAppendPrintf(sentry, "Histogram of events per incoming socket type\n");
 502     storeAppendPrintf(sentry, "ICP Messages handled per comm_poll_udp_incoming() call:\n");
 503     statCounter.comm_udp.history.dump(sentry, statHistIntDumper);
 504     storeAppendPrintf(sentry, "DNS Messages handled per comm_poll_dns_incoming() call:\n");
 505     statCounter.comm_dns.history.dump(sentry, statHistIntDumper);
 506     storeAppendPrintf(sentry, "HTTP Messages handled per comm_poll_tcp_incoming() call:\n");
 507     statCounter.comm_tcp.history.dump(sentry, statHistIntDumper);
 508 }
 509
 510 /* Called by async-io or diskd to speed up the polling */
 511 void
 512 Comm::QuickPollRequired(void)
 513 {
 514     MAX_POLL_TIME = 10;
 515 }
 516
 517 #endif /* USE_POLL */
 518