2 * DEBUG: section 05 Socket Functions
4 * SQUID Web Proxy Cache http://www.squid-cache.org/
5 * ----------------------------------------------------------
7 * Squid is the result of efforts by numerous individuals from
8 * the Internet community; see the CONTRIBUTORS file for full
9 * details. Many organizations have provided support for Squid's
10 * development; see the SPONSORS file for full details. Squid is
11 * Copyrighted (C) 2001 by the Regents of the University of
12 * California; see the COPYRIGHT file for full details. Squid
13 * incorporates software developed and/or copyrighted by other
14 * sources; see the CREDITS file for full details.
16 * This program is free software; you can redistribute it and/or modify
17 * it under the terms of the GNU General Public License as published by
18 * the Free Software Foundation; either version 2 of the License, or
19 * (at your option) any later version.
21 * This program is distributed in the hope that it will be useful,
22 * but WITHOUT ANY WARRANTY; without even the implied warranty of
23 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
24 * GNU General Public License for more details.
26 * You should have received a copy of the GNU General Public License
27 * along with this program; if not, write to the Free Software
28 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
34 #include "anyp/PortCfg.h"
35 #include "comm/Connection.h"
36 #include "comm/Loops.h"
41 #include "mgr/Registration.h"
42 #include "profiler/Profiler.h"
43 #include "SquidConfig.h"
44 #include "SquidTime.h"
45 #include "StatCounters.h"
53 /* Needed for poll() on Linux at least */
56 #define POLLRDNORM POLLIN
59 #define POLLWRNORM POLLOUT
63 static int MAX_POLL_TIME
= 1000; /* see also Comm::QuickPollRequired() */
66 #define howmany(x, y) (((x)+((y)-1))/(y))
71 #define FD_MASK_BYTES sizeof(fd_mask)
72 #define FD_MASK_BITS (FD_MASK_BYTES*NBBY)
75 static int fdIsTcpListen(int fd
);
76 static int fdIsUdpListen(int fd
);
77 static int fdIsDns(int fd
);
78 static OBJH commIncomingStats
;
79 static int comm_check_incoming_poll_handlers(int nfds
, int *fds
);
80 static void comm_poll_dns_incoming(void);
83 * Automatic tuning for incoming requests:
85 * INCOMING sockets are the ICP and HTTP ports. We need to check these
86 * fairly regularly, but how often? When the load increases, we
87 * want to check the incoming sockets more often. If we have a lot
88 * of incoming ICP, then we need to check these sockets more than
89 * if we just have HTTP.
91 * The variables 'incoming_icp_interval' and 'incoming_http_interval'
92 * determine how many normal I/O events to process before checking
93 * incoming sockets again. Note we store the incoming_interval
94 * multipled by a factor of (2^INCOMING_FACTOR) to have some
95 * pseudo-floating point precision.
97 * The variable 'udp_io_events' and 'tcp_io_events' counts how many normal
98 * I/O events have been processed since the last check on the incoming
99 * sockets. When io_events > incoming_interval, its time to check incoming
102 * Every time we check incoming sockets, we count how many new messages
103 * or connections were processed. This is used to adjust the
104 * incoming_interval for the next iteration. The new incoming_interval
105 * is calculated as the current incoming_interval plus what we would
106 * like to see as an average number of events minus the number of
107 * events just processed.
109 * incoming_interval = incoming_interval + target_average - number_of_events_processed
111 * There are separate incoming_interval counters for TCP-based, UDP-based, and DNS events
113 * You can see the current values of the incoming_interval's, as well as
114 * a histogram of 'incoming_events' by asking the cache manager
115 * for 'comm_incoming', e.g.:
117 * % ./client mgr:comm_poll_incoming
121 * - We have MAX_INCOMING_INTEGER as a magic upper limit on
122 * incoming_interval for both types of sockets. At the
123 * largest value the cache will effectively be idling.
125 * - The higher the INCOMING_FACTOR, the slower the algorithm will
126 * respond to load spikes/increases/decreases in demand. A value
127 * between 3 and 8 is recommended.
130 #define MAX_INCOMING_INTEGER 256
131 #define INCOMING_FACTOR 5
132 #define MAX_INCOMING_INTERVAL (MAX_INCOMING_INTEGER << INCOMING_FACTOR)
133 static int udp_io_events
= 0; ///< I/O events passed since last UDP receiver socket poll
134 static int dns_io_events
= 0; ///< I/O events passed since last DNS socket poll
135 static int tcp_io_events
= 0; ///< I/O events passed since last TCP listening socket poll
136 static int incoming_udp_interval
= 16 << INCOMING_FACTOR
;
137 static int incoming_dns_interval
= 16 << INCOMING_FACTOR
;
138 static int incoming_tcp_interval
= 16 << INCOMING_FACTOR
;
139 #define commCheckUdpIncoming (++udp_io_events > (incoming_udp_interval>> INCOMING_FACTOR))
140 #define commCheckDnsIncoming (++dns_io_events > (incoming_dns_interval>> INCOMING_FACTOR))
141 #define commCheckTcpIncoming (++tcp_io_events > (incoming_tcp_interval>> INCOMING_FACTOR))
144 Comm::SetSelect(int fd
, unsigned int type
, PF
* handler
, void *client_data
, time_t timeout
)
146 fde
*F
= &fd_table
[fd
];
148 assert(F
->flags
.open
);
149 debugs(5, 5, HERE
<< "FD " << fd
<< ", type=" << type
<<
150 ", handler=" << handler
<< ", client_data=" << client_data
<<
151 ", timeout=" << timeout
);
153 if (type
& COMM_SELECT_READ
) {
154 F
->read_handler
= handler
;
155 F
->read_data
= client_data
;
158 if (type
& COMM_SELECT_WRITE
) {
159 F
->write_handler
= handler
;
160 F
->write_data
= client_data
;
164 F
->timeout
= squid_curtime
+ timeout
;
168 Comm::ResetSelect(int fd
)
173 fdIsUdpListen(int fd
)
175 if (icpIncomingConn
!= NULL
&& icpIncomingConn
->fd
== fd
)
178 if (icpOutgoingConn
!= NULL
&& icpOutgoingConn
->fd
== fd
)
187 if (fd
== DnsSocketA
)
190 if (fd
== DnsSocketB
)
197 fdIsTcpListen(int fd
)
199 for (AnyP::PortCfgPointer s
= HttpPortList
; s
!= NULL
; s
= s
->next
) {
200 if (s
->listenConn
!= NULL
&& s
->listenConn
->fd
== fd
)
208 comm_check_incoming_poll_handlers(int nfds
, int *fds
)
215 struct pollfd pfds
[3 + MAXTCPLISTENPORTS
];
216 PROF_start(comm_check_incoming
);
217 incoming_sockets_accepted
= 0;
219 for (i
= npfds
= 0; i
< nfds
; ++i
) {
224 if (fd_table
[fd
].read_handler
)
225 events
|= POLLRDNORM
;
227 if (fd_table
[fd
].write_handler
)
228 events
|= POLLWRNORM
;
232 pfds
[npfds
].events
= events
;
233 pfds
[npfds
].revents
= 0;
239 PROF_stop(comm_check_incoming
);
244 ++ statCounter
.syscalls
.selects
;
246 if (poll(pfds
, npfds
, 0) < 1) {
247 PROF_stop(comm_check_incoming
);
248 return incoming_sockets_accepted
;
251 for (i
= 0; i
< npfds
; ++i
) {
254 if (((revents
= pfds
[i
].revents
) == 0) || ((fd
= pfds
[i
].fd
) == -1))
257 if (revents
& (POLLRDNORM
| POLLIN
| POLLHUP
| POLLERR
)) {
258 if ((hdl
= fd_table
[fd
].read_handler
)) {
259 fd_table
[fd
].read_handler
= NULL
;
260 hdl(fd
, fd_table
[fd
].read_data
);
261 } else if (pfds
[i
].events
& POLLRDNORM
)
262 debugs(5, DBG_IMPORTANT
, "comm_poll_incoming: FD " << fd
<< " NULL read handler");
265 if (revents
& (POLLWRNORM
| POLLOUT
| POLLHUP
| POLLERR
)) {
266 if ((hdl
= fd_table
[fd
].write_handler
)) {
267 fd_table
[fd
].write_handler
= NULL
;
268 hdl(fd
, fd_table
[fd
].write_data
);
269 } else if (pfds
[i
].events
& POLLWRNORM
)
270 debugs(5, DBG_IMPORTANT
, "comm_poll_incoming: FD " << fd
<< " NULL write_handler");
274 PROF_stop(comm_check_incoming
);
275 return incoming_sockets_accepted
;
279 comm_poll_udp_incoming(void)
286 if (Comm::IsConnOpen(icpIncomingConn
)) {
287 fds
[nfds
] = icpIncomingConn
->fd
;
291 if (icpIncomingConn
!= icpOutgoingConn
&& Comm::IsConnOpen(icpOutgoingConn
)) {
292 fds
[nfds
] = icpOutgoingConn
->fd
;
299 nevents
= comm_check_incoming_poll_handlers(nfds
, fds
);
301 incoming_udp_interval
+= Config
.comm_incoming
.udp
.average
- nevents
;
303 if (incoming_udp_interval
< Config
.comm_incoming
.udp
.min_poll
)
304 incoming_udp_interval
= Config
.comm_incoming
.udp
.min_poll
;
306 if (incoming_udp_interval
> MAX_INCOMING_INTERVAL
)
307 incoming_udp_interval
= MAX_INCOMING_INTERVAL
;
309 if (nevents
> INCOMING_UDP_MAX
)
310 nevents
= INCOMING_UDP_MAX
;
312 statCounter
.comm_udp_incoming
.count(nevents
);
316 comm_poll_tcp_incoming(void)
319 int fds
[MAXTCPLISTENPORTS
];
324 // XXX: only poll sockets that won't be deferred. But how do we identify them?
326 for (j
= 0; j
< NHttpSockets
; ++j
) {
327 if (HttpSockets
[j
] < 0)
330 fds
[nfds
] = HttpSockets
[j
];
334 nevents
= comm_check_incoming_poll_handlers(nfds
, fds
);
335 incoming_tcp_interval
= incoming_tcp_interval
336 + Config
.comm_incoming
.tcp
.average
- nevents
;
338 if (incoming_tcp_interval
< Config
.comm_incoming
.tcp
.min_poll
)
339 incoming_tcp_interval
= Config
.comm_incoming
.tcp
.min_poll
;
341 if (incoming_tcp_interval
> MAX_INCOMING_INTERVAL
)
342 incoming_tcp_interval
= MAX_INCOMING_INTERVAL
;
344 if (nevents
> INCOMING_TCP_MAX
)
345 nevents
= INCOMING_TCP_MAX
;
347 statCounter
.comm_tcp_incoming
.count(nevents
);
350 /* poll all sockets; call handlers for those that are ready. */
352 Comm::DoSelect(int msec
)
354 struct pollfd pfds
[SQUID_MAXFD
];
360 unsigned long npending
;
362 int calldns
= 0, calludp
= 0, calltcp
= 0;
363 double timeout
= current_dtime
+ (msec
/ 1000.0);
368 start
= current_dtime
;
370 if (commCheckUdpIncoming
)
371 comm_poll_udp_incoming();
373 if (commCheckDnsIncoming
)
374 comm_poll_dns_incoming();
376 if (commCheckTcpIncoming
)
377 comm_poll_tcp_incoming();
379 PROF_start(comm_poll_prep_pfds
);
381 calldns
= calludp
= calltcp
= 0;
387 maxfd
= Biggest_FD
+ 1;
389 for (int i
= 0; i
< maxfd
; ++i
) {
392 /* Check each open socket for a handler. */
394 if (fd_table
[i
].read_handler
)
395 events
|= POLLRDNORM
;
397 if (fd_table
[i
].write_handler
)
398 events
|= POLLWRNORM
;
402 pfds
[nfds
].events
= events
;
403 pfds
[nfds
].revents
= 0;
406 if ((events
& POLLRDNORM
) && fd_table
[i
].flags
.read_pending
)
411 PROF_stop(comm_poll_prep_pfds
);
416 if (msec
> MAX_POLL_TIME
)
417 msec
= MAX_POLL_TIME
;
421 * Note that this will only ever trigger when there are no log files
422 * and stdout/err/in are all closed too.
424 if (nfds
== 0 && npending
== 0) {
426 return Comm::SHUTDOWN
;
432 PROF_start(comm_poll_normal
);
433 ++ statCounter
.syscalls
.selects
;
434 num
= poll(pfds
, nfds
, msec
);
435 ++ statCounter
.select_loops
;
436 PROF_stop(comm_poll_normal
);
438 if (num
>= 0 || npending
> 0)
441 if (ignoreErrno(errno
))
444 debugs(5, DBG_CRITICAL
, "comm_poll: poll failure: " << xstrerror());
446 assert(errno
!= EINVAL
);
448 return Comm::COMM_ERROR
;
455 debugs(5, num
? 5 : 8, "comm_poll: " << num
<< "+" << npending
<< " FDs ready");
456 statCounter
.select_fds_hist
.count(num
);
458 if (num
== 0 && npending
== 0)
461 /* scan each socket but the accept socket. Poll this
462 * more frequently to minimize losses due to the 5 connect
464 PROF_start(comm_handle_ready_fd
);
466 for (size_t loopIndex
= 0; loopIndex
< nfds
; ++loopIndex
) {
468 int revents
= pfds
[loopIndex
].revents
;
469 fd
= pfds
[loopIndex
].fd
;
474 if (fd_table
[fd
].flags
.read_pending
)
480 if (fdIsUdpListen(fd
)) {
490 if (fdIsTcpListen(fd
)) {
497 if (revents
& (POLLRDNORM
| POLLIN
| POLLHUP
| POLLERR
)) {
498 debugs(5, 6, "comm_poll: FD " << fd
<< " ready for reading");
500 if ((hdl
= F
->read_handler
)) {
501 PROF_start(comm_read_handler
);
502 F
->read_handler
= NULL
;
503 F
->flags
.read_pending
= false;
504 hdl(fd
, F
->read_data
);
505 PROF_stop(comm_read_handler
);
506 ++ statCounter
.select_fds
;
508 if (commCheckUdpIncoming
)
509 comm_poll_udp_incoming();
511 if (commCheckDnsIncoming
)
512 comm_poll_dns_incoming();
514 if (commCheckTcpIncoming
)
515 comm_poll_tcp_incoming();
519 if (revents
& (POLLWRNORM
| POLLOUT
| POLLHUP
| POLLERR
)) {
520 debugs(5, 6, "comm_poll: FD " << fd
<< " ready for writing");
522 if ((hdl
= F
->write_handler
)) {
523 PROF_start(comm_write_handler
);
524 F
->write_handler
= NULL
;
525 hdl(fd
, F
->write_data
);
526 PROF_stop(comm_write_handler
);
527 ++ statCounter
.select_fds
;
529 if (commCheckUdpIncoming
)
530 comm_poll_udp_incoming();
532 if (commCheckDnsIncoming
)
533 comm_poll_dns_incoming();
535 if (commCheckTcpIncoming
)
536 comm_poll_tcp_incoming();
540 if (revents
& POLLNVAL
) {
541 AsyncCall::Pointer ch
;
542 debugs(5, DBG_CRITICAL
, "WARNING: FD " << fd
<< " has handlers, but it's invalid.");
543 debugs(5, DBG_CRITICAL
, "FD " << fd
<< " is a " << fdTypeStr
[F
->type
]);
544 debugs(5, DBG_CRITICAL
, "--> " << F
->desc
);
545 debugs(5, DBG_CRITICAL
, "tmout:" << F
->timeoutHandler
<< "read:" <<
546 F
->read_handler
<< " write:" << F
->write_handler
);
548 for (ch
= F
->closeHandler
; ch
!= NULL
; ch
= ch
->Next())
549 debugs(5, DBG_CRITICAL
, " close handler: " << ch
);
551 if (F
->closeHandler
!= NULL
) {
552 commCallCloseHandlers(fd
);
553 } else if (F
->timeoutHandler
!= NULL
) {
554 debugs(5, DBG_CRITICAL
, "comm_poll: Calling Timeout Handler");
555 ScheduleCallHere(F
->timeoutHandler
);
558 F
->closeHandler
= NULL
;
559 F
->timeoutHandler
= NULL
;
560 F
->read_handler
= NULL
;
561 F
->write_handler
= NULL
;
568 PROF_stop(comm_handle_ready_fd
);
571 comm_poll_udp_incoming();
574 comm_poll_dns_incoming();
577 comm_poll_tcp_incoming();
581 statCounter
.select_time
+= (current_dtime
- start
);
584 } while (timeout
> current_dtime
);
586 debugs(5, 8, "comm_poll: time out: " << squid_curtime
<< ".");
588 return Comm::TIMEOUT
;
592 comm_poll_dns_incoming(void)
599 if (DnsSocketA
< 0 && DnsSocketB
< 0)
602 if (DnsSocketA
>= 0) {
603 fds
[nfds
] = DnsSocketA
;
607 if (DnsSocketB
>= 0) {
608 fds
[nfds
] = DnsSocketB
;
612 nevents
= comm_check_incoming_poll_handlers(nfds
, fds
);
617 incoming_dns_interval
+= Config
.comm_incoming
.dns
.average
- nevents
;
619 if (incoming_dns_interval
< Config
.comm_incoming
.dns
.min_poll
)
620 incoming_dns_interval
= Config
.comm_incoming
.dns
.min_poll
;
622 if (incoming_dns_interval
> MAX_INCOMING_INTERVAL
)
623 incoming_dns_interval
= MAX_INCOMING_INTERVAL
;
625 if (nevents
> INCOMING_DNS_MAX
)
626 nevents
= INCOMING_DNS_MAX
;
628 statCounter
.comm_dns_incoming
.count(nevents
);
632 commPollRegisterWithCacheManager(void)
634 Mgr::RegisterAction("comm_poll_incoming",
635 "comm_incoming() stats",
636 commIncomingStats
, 0, 1);
640 Comm::SelectLoopInit(void)
642 commPollRegisterWithCacheManager();
646 commIncomingStats(StoreEntry
* sentry
)
648 storeAppendPrintf(sentry
, "Current incoming_udp_interval: %d\n",
649 incoming_udp_interval
>> INCOMING_FACTOR
);
650 storeAppendPrintf(sentry
, "Current incoming_dns_interval: %d\n",
651 incoming_dns_interval
>> INCOMING_FACTOR
);
652 storeAppendPrintf(sentry
, "Current incoming_tcp_interval: %d\n",
653 incoming_tcp_interval
>> INCOMING_FACTOR
);
654 storeAppendPrintf(sentry
, "\n");
655 storeAppendPrintf(sentry
, "Histogram of events per incoming socket type\n");
656 storeAppendPrintf(sentry
, "ICP Messages handled per comm_poll_udp_incoming() call:\n");
657 statCounter
.comm_udp_incoming
.dump(sentry
, statHistIntDumper
);
658 storeAppendPrintf(sentry
, "DNS Messages handled per comm_poll_dns_incoming() call:\n");
659 statCounter
.comm_dns_incoming
.dump(sentry
, statHistIntDumper
);
660 storeAppendPrintf(sentry
, "HTTP Messages handled per comm_poll_tcp_incoming() call:\n");
661 statCounter
.comm_tcp_incoming
.dump(sentry
, statHistIntDumper
);
664 /* Called by async-io or diskd to speed up the polling */
666 Comm::QuickPollRequired(void)
671 #endif /* USE_POLL */