2 * Copyright (C) 1996-2017 The Squid Software Foundation and contributors
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.
9 /* DEBUG: section 05 Socket Functions */
14 #include "anyp/PortCfg.h"
15 #include "comm/Connection.h"
16 #include "comm/Loops.h"
21 #include "mgr/Registration.h"
22 #include "profiler/Profiler.h"
23 #include "SquidConfig.h"
24 #include "SquidTime.h"
25 #include "StatCounters.h"
33 /* Needed for poll() on Linux at least */
36 #define POLLRDNORM POLLIN
39 #define POLLWRNORM POLLOUT
43 static int MAX_POLL_TIME
= 1000; /* see also Comm::QuickPollRequired() */
46 #define howmany(x, y) (((x)+((y)-1))/(y))
51 #define FD_MASK_BYTES sizeof(fd_mask)
52 #define FD_MASK_BITS (FD_MASK_BYTES*NBBY)
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);
63 * Automatic tuning for incoming requests:
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.
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.
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
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.
89 * incoming_interval = incoming_interval + target_average - number_of_events_processed
91 * There are separate incoming_interval counters for TCP-based, UDP-based, and DNS events
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.:
97 * % ./client mgr:comm_poll_incoming
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.
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.
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))
124 Comm::SetSelect(int fd
, unsigned int type
, PF
* handler
, void *client_data
, time_t timeout
)
126 fde
*F
= &fd_table
[fd
];
128 assert(F
->flags
.open
);
129 debugs(5, 5, HERE
<< "FD " << fd
<< ", type=" << type
<<
130 ", handler=" << handler
<< ", client_data=" << client_data
<<
131 ", timeout=" << timeout
);
133 if (type
& COMM_SELECT_READ
) {
134 F
->read_handler
= handler
;
135 F
->read_data
= client_data
;
138 if (type
& COMM_SELECT_WRITE
) {
139 F
->write_handler
= handler
;
140 F
->write_data
= client_data
;
144 F
->timeout
= squid_curtime
+ timeout
;
148 Comm::ResetSelect(int fd
)
153 fdIsUdpListen(int fd
)
155 if (icpIncomingConn
!= NULL
&& icpIncomingConn
->fd
== fd
)
158 if (icpOutgoingConn
!= NULL
&& icpOutgoingConn
->fd
== fd
)
167 if (fd
== DnsSocketA
)
170 if (fd
== DnsSocketB
)
177 fdIsTcpListen(int fd
)
179 for (AnyP::PortCfgPointer s
= HttpPortList
; s
!= NULL
; s
= s
->next
) {
180 if (s
->listenConn
!= NULL
&& s
->listenConn
->fd
== fd
)
188 comm_check_incoming_poll_handlers(int nfds
, int *fds
)
195 struct pollfd pfds
[3 + MAXTCPLISTENPORTS
];
196 PROF_start(comm_check_incoming
);
197 incoming_sockets_accepted
= 0;
199 for (i
= npfds
= 0; i
< nfds
; ++i
) {
204 if (fd_table
[fd
].read_handler
)
205 events
|= POLLRDNORM
;
207 if (fd_table
[fd
].write_handler
)
208 events
|= POLLWRNORM
;
212 pfds
[npfds
].events
= events
;
213 pfds
[npfds
].revents
= 0;
219 PROF_stop(comm_check_incoming
);
224 ++ statCounter
.syscalls
.selects
;
226 if (poll(pfds
, npfds
, 0) < 1) {
227 PROF_stop(comm_check_incoming
);
228 return incoming_sockets_accepted
;
231 for (i
= 0; i
< npfds
; ++i
) {
234 if (((revents
= pfds
[i
].revents
) == 0) || ((fd
= pfds
[i
].fd
) == -1))
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");
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");
254 PROF_stop(comm_check_incoming
);
255 return incoming_sockets_accepted
;
259 comm_poll_udp_incoming(void)
266 if (Comm::IsConnOpen(icpIncomingConn
)) {
267 fds
[nfds
] = icpIncomingConn
->fd
;
271 if (icpIncomingConn
!= icpOutgoingConn
&& Comm::IsConnOpen(icpOutgoingConn
)) {
272 fds
[nfds
] = icpOutgoingConn
->fd
;
279 nevents
= comm_check_incoming_poll_handlers(nfds
, fds
);
281 incoming_udp_interval
+= Config
.comm_incoming
.udp
.average
- nevents
;
283 if (incoming_udp_interval
< Config
.comm_incoming
.udp
.min_poll
)
284 incoming_udp_interval
= Config
.comm_incoming
.udp
.min_poll
;
286 if (incoming_udp_interval
> MAX_INCOMING_INTERVAL
)
287 incoming_udp_interval
= MAX_INCOMING_INTERVAL
;
289 if (nevents
> INCOMING_UDP_MAX
)
290 nevents
= INCOMING_UDP_MAX
;
292 statCounter
.comm_udp_incoming
.count(nevents
);
296 comm_poll_tcp_incoming(void)
299 int fds
[MAXTCPLISTENPORTS
];
304 // XXX: only poll sockets that won't be deferred. But how do we identify them?
306 for (j
= 0; j
< NHttpSockets
; ++j
) {
307 if (HttpSockets
[j
] < 0)
310 fds
[nfds
] = HttpSockets
[j
];
314 nevents
= comm_check_incoming_poll_handlers(nfds
, fds
);
315 incoming_tcp_interval
= incoming_tcp_interval
316 + Config
.comm_incoming
.tcp
.average
- nevents
;
318 if (incoming_tcp_interval
< Config
.comm_incoming
.tcp
.min_poll
)
319 incoming_tcp_interval
= Config
.comm_incoming
.tcp
.min_poll
;
321 if (incoming_tcp_interval
> MAX_INCOMING_INTERVAL
)
322 incoming_tcp_interval
= MAX_INCOMING_INTERVAL
;
324 if (nevents
> INCOMING_TCP_MAX
)
325 nevents
= INCOMING_TCP_MAX
;
327 statCounter
.comm_tcp_incoming
.count(nevents
);
330 /* poll all sockets; call handlers for those that are ready. */
332 Comm::DoSelect(int msec
)
334 struct pollfd pfds
[SQUID_MAXFD
];
340 unsigned long npending
;
342 int calldns
= 0, calludp
= 0, calltcp
= 0;
343 double timeout
= current_dtime
+ (msec
/ 1000.0);
348 start
= current_dtime
;
350 if (commCheckUdpIncoming
)
351 comm_poll_udp_incoming();
353 if (commCheckDnsIncoming
)
354 comm_poll_dns_incoming();
356 if (commCheckTcpIncoming
)
357 comm_poll_tcp_incoming();
359 PROF_start(comm_poll_prep_pfds
);
361 calldns
= calludp
= calltcp
= 0;
367 maxfd
= Biggest_FD
+ 1;
369 for (int i
= 0; i
< maxfd
; ++i
) {
372 /* Check each open socket for a handler. */
374 if (fd_table
[i
].read_handler
)
375 events
|= POLLRDNORM
;
377 if (fd_table
[i
].write_handler
)
378 events
|= POLLWRNORM
;
382 pfds
[nfds
].events
= events
;
383 pfds
[nfds
].revents
= 0;
386 if ((events
& POLLRDNORM
) && fd_table
[i
].flags
.read_pending
)
391 PROF_stop(comm_poll_prep_pfds
);
396 if (msec
> MAX_POLL_TIME
)
397 msec
= MAX_POLL_TIME
;
401 * Note that this will only ever trigger when there are no log files
402 * and stdout/err/in are all closed too.
404 if (nfds
== 0 && npending
== 0) {
406 return Comm::SHUTDOWN
;
412 PROF_start(comm_poll_normal
);
413 ++ statCounter
.syscalls
.selects
;
414 num
= poll(pfds
, nfds
, msec
);
416 ++ statCounter
.select_loops
;
417 PROF_stop(comm_poll_normal
);
419 if (num
>= 0 || npending
> 0)
422 if (ignoreErrno(xerrno
))
425 debugs(5, DBG_CRITICAL
, MYNAME
<< "poll failure: " << xstrerr(xerrno
));
427 assert(xerrno
!= EINVAL
);
429 return Comm::COMM_ERROR
;
436 debugs(5, num
? 5 : 8, "comm_poll: " << num
<< "+" << npending
<< " FDs ready");
437 statCounter
.select_fds_hist
.count(num
);
439 if (num
== 0 && npending
== 0)
442 /* scan each socket but the accept socket. Poll this
443 * more frequently to minimize losses due to the 5 connect
445 PROF_start(comm_handle_ready_fd
);
447 for (size_t loopIndex
= 0; loopIndex
< nfds
; ++loopIndex
) {
449 int revents
= pfds
[loopIndex
].revents
;
450 fd
= pfds
[loopIndex
].fd
;
455 if (fd_table
[fd
].flags
.read_pending
)
461 if (fdIsUdpListen(fd
)) {
471 if (fdIsTcpListen(fd
)) {
478 if (revents
& (POLLRDNORM
| POLLIN
| POLLHUP
| POLLERR
)) {
479 debugs(5, 6, "comm_poll: FD " << fd
<< " ready for reading");
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
;
489 if (commCheckUdpIncoming
)
490 comm_poll_udp_incoming();
492 if (commCheckDnsIncoming
)
493 comm_poll_dns_incoming();
495 if (commCheckTcpIncoming
)
496 comm_poll_tcp_incoming();
500 if (revents
& (POLLWRNORM
| POLLOUT
| POLLHUP
| POLLERR
)) {
501 debugs(5, 6, "comm_poll: FD " << fd
<< " ready for writing");
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
;
510 if (commCheckUdpIncoming
)
511 comm_poll_udp_incoming();
513 if (commCheckDnsIncoming
)
514 comm_poll_dns_incoming();
516 if (commCheckTcpIncoming
)
517 comm_poll_tcp_incoming();
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
);
529 for (ch
= F
->closeHandler
; ch
!= NULL
; ch
= ch
->Next())
530 debugs(5, DBG_CRITICAL
, " close handler: " << ch
);
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
);
539 F
->closeHandler
= NULL
;
540 F
->timeoutHandler
= NULL
;
541 F
->read_handler
= NULL
;
542 F
->write_handler
= NULL
;
549 PROF_stop(comm_handle_ready_fd
);
552 comm_poll_udp_incoming();
555 comm_poll_dns_incoming();
558 comm_poll_tcp_incoming();
562 statCounter
.select_time
+= (current_dtime
- start
);
565 } while (timeout
> current_dtime
);
567 debugs(5, 8, "comm_poll: time out: " << squid_curtime
<< ".");
569 return Comm::TIMEOUT
;
573 comm_poll_dns_incoming(void)
580 if (DnsSocketA
< 0 && DnsSocketB
< 0)
583 if (DnsSocketA
>= 0) {
584 fds
[nfds
] = DnsSocketA
;
588 if (DnsSocketB
>= 0) {
589 fds
[nfds
] = DnsSocketB
;
593 nevents
= comm_check_incoming_poll_handlers(nfds
, fds
);
598 incoming_dns_interval
+= Config
.comm_incoming
.dns
.average
- nevents
;
600 if (incoming_dns_interval
< Config
.comm_incoming
.dns
.min_poll
)
601 incoming_dns_interval
= Config
.comm_incoming
.dns
.min_poll
;
603 if (incoming_dns_interval
> MAX_INCOMING_INTERVAL
)
604 incoming_dns_interval
= MAX_INCOMING_INTERVAL
;
606 if (nevents
> INCOMING_DNS_MAX
)
607 nevents
= INCOMING_DNS_MAX
;
609 statCounter
.comm_dns_incoming
.count(nevents
);
613 commPollRegisterWithCacheManager(void)
615 Mgr::RegisterAction("comm_poll_incoming",
616 "comm_incoming() stats",
617 commIncomingStats
, 0, 1);
621 Comm::SelectLoopInit(void)
623 commPollRegisterWithCacheManager();
627 commIncomingStats(StoreEntry
* sentry
)
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
);
645 /* Called by async-io or diskd to speed up the polling */
647 Comm::QuickPollRequired(void)
652 #endif /* USE_POLL */