4 * DEBUG: section 05 Socket Functions
6 * SQUID Web Proxy Cache http://www.squid-cache.org/
7 * ----------------------------------------------------------
9 * Squid is the result of efforts by numerous individuals from
10 * the Internet community; see the CONTRIBUTORS file for full
11 * details. Many organizations have provided support for Squid's
12 * development; see the SPONSORS file for full details. Squid is
13 * Copyrighted (C) 2001 by the Regents of the University of
14 * California; see the COPYRIGHT file for full details. Squid
15 * incorporates software developed and/or copyrighted by other
16 * sources; see the CREDITS file for full details.
18 * This program is free software; you can redistribute it and/or modify
19 * it under the terms of the GNU General Public License as published by
20 * the Free Software Foundation; either version 2 of the License, or
21 * (at your option) any later version.
23 * This program is distributed in the hope that it will be useful,
24 * but WITHOUT ANY WARRANTY; without even the implied warranty of
25 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
26 * GNU General Public License for more details.
28 * You should have received a copy of the GNU General Public License
29 * along with this program; if not, write to the Free Software
30 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
37 #include "anyp/PortCfg.h"
38 #include "comm/Connection.h"
39 #include "comm/Loops.h"
43 #include "mgr/Registration.h"
44 #include "SquidTime.h"
45 #include "StatCounters.h"
56 static int MAX_POLL_TIME
= 1000; /* see also Comm::QuickPollRequired() */
59 #define howmany(x, y) (((x)+((y)-1))/(y))
64 #define FD_MASK_BYTES sizeof(fd_mask)
65 #define FD_MASK_BITS (FD_MASK_BYTES*NBBY)
68 static int examine_select(fd_set
*, fd_set
*);
69 static int fdIsTcpListener(int fd
);
70 static int fdIsUdpListener(int fd
);
71 static int fdIsDns(int fd
);
72 static OBJH commIncomingStats
;
73 static int comm_check_incoming_select_handlers(int nfds
, int *fds
);
74 static void comm_select_dns_incoming(void);
75 static void commUpdateReadBits(int fd
, PF
* handler
);
76 static void commUpdateWriteBits(int fd
, PF
* handler
);
78 static struct timeval zero_tv
;
79 static fd_set global_readfds
;
80 static fd_set global_writefds
;
85 * Automatic tuning for incoming requests:
87 * INCOMING sockets are the ICP and HTTP ports. We need to check these
88 * fairly regularly, but how often? When the load increases, we
89 * want to check the incoming sockets more often. If we have a lot
90 * of incoming ICP, then we need to check these sockets more than
91 * if we just have HTTP.
93 * The variables 'incoming_udp_interval' and 'incoming_tcp_interval'
94 * determine how many normal I/O events to process before checking
95 * incoming sockets again. Note we store the incoming_interval
96 * multipled by a factor of (2^INCOMING_FACTOR) to have some
97 * pseudo-floating point precision.
99 * The variable 'udp_io_events' and 'tcp_io_events' counts how many normal
100 * I/O events have been processed since the last check on the incoming
101 * sockets. When io_events > incoming_interval, its time to check incoming
104 * Every time we check incoming sockets, we count how many new messages
105 * or connections were processed. This is used to adjust the
106 * incoming_interval for the next iteration. The new incoming_interval
107 * is calculated as the current incoming_interval plus what we would
108 * like to see as an average number of events minus the number of
109 * events just processed.
111 * incoming_interval = incoming_interval + target_average - number_of_events_processed
113 * There are separate incoming_interval counters for DNS, UDP and TCP events
115 * You can see the current values of the incoming_interval's, as well as
116 * a histogram of 'incoming_events' by asking the cache manager
117 * for 'comm_incoming', e.g.:
119 * % ./client mgr:comm_incoming
123 * - We have MAX_INCOMING_INTEGER as a magic upper limit on
124 * incoming_interval for both types of sockets. At the
125 * largest value the cache will effectively be idling.
127 * - The higher the INCOMING_FACTOR, the slower the algorithm will
128 * respond to load spikes/increases/decreases in demand. A value
129 * between 3 and 8 is recommended.
132 #define MAX_INCOMING_INTEGER 256
133 #define INCOMING_FACTOR 5
134 #define MAX_INCOMING_INTERVAL (MAX_INCOMING_INTEGER << INCOMING_FACTOR)
135 static int udp_io_events
= 0;
136 static int dns_io_events
= 0;
137 static int tcp_io_events
= 0;
138 static int incoming_udp_interval
= 16 << INCOMING_FACTOR
;
139 static int incoming_dns_interval
= 16 << INCOMING_FACTOR
;
140 static int incoming_tcp_interval
= 16 << INCOMING_FACTOR
;
141 #define commCheckUdpIncoming (++udp_io_events > (incoming_udp_interval>> INCOMING_FACTOR))
142 #define commCheckDnsIncoming (++dns_io_events > (incoming_dns_interval>> INCOMING_FACTOR))
143 #define commCheckTcpIncoming (++tcp_io_events > (incoming_tcp_interval>> INCOMING_FACTOR))
146 Comm::SetSelect(int fd
, unsigned int type
, PF
* handler
, void *client_data
, time_t timeout
)
148 fde
*F
= &fd_table
[fd
];
150 assert(F
->flags
.open
);
151 debugs(5, 5, HERE
<< "FD " << fd
<< ", type=" << type
<<
152 ", handler=" << handler
<< ", client_data=" << client_data
<<
153 ", timeout=" << timeout
);
155 if (type
& COMM_SELECT_READ
) {
156 F
->read_handler
= handler
;
157 F
->read_data
= client_data
;
158 commUpdateReadBits(fd
, handler
);
161 if (type
& COMM_SELECT_WRITE
) {
162 F
->write_handler
= handler
;
163 F
->write_data
= client_data
;
164 commUpdateWriteBits(fd
, handler
);
168 F
->timeout
= squid_curtime
+ timeout
;
172 Comm::ResetSelect(int fd
)
177 fdIsUdpListener(int fd
)
179 if (icpIncomingConn
!= NULL
&& fd
== icpIncomingConn
->fd
)
182 if (icpOutgoingConn
!= NULL
&& fd
== icpOutgoingConn
->fd
)
191 if (fd
== DnsSocketA
)
194 if (fd
== DnsSocketB
)
201 fdIsTcpListener(int fd
)
203 for (const AnyP::PortCfg
*s
= Config
.Sockaddr
.http
; s
; s
= s
->next
) {
204 if (s
->listenConn
!= NULL
&& s
->listenConn
->fd
== fd
)
212 comm_check_incoming_select_handlers(int nfds
, int *fds
)
221 FD_ZERO(&write_mask
);
222 incoming_sockets_accepted
= 0;
224 for (i
= 0; i
< nfds
; ++i
) {
227 if (fd_table
[fd
].read_handler
) {
228 FD_SET(fd
, &read_mask
);
234 if (fd_table
[fd
].write_handler
) {
235 FD_SET(fd
, &write_mask
);
247 ++ statCounter
.syscalls
.selects
;
249 if (select(maxfd
, &read_mask
, &write_mask
, NULL
, &zero_tv
) < 1)
250 return incoming_sockets_accepted
;
252 for (i
= 0; i
< nfds
; ++i
) {
255 if (FD_ISSET(fd
, &read_mask
)) {
256 if ((hdl
= fd_table
[fd
].read_handler
) != NULL
) {
257 fd_table
[fd
].read_handler
= NULL
;
258 commUpdateReadBits(fd
, NULL
);
259 hdl(fd
, fd_table
[fd
].read_data
);
261 debugs(5, DBG_IMPORTANT
, "comm_select_incoming: FD " << fd
<< " NULL read handler");
265 if (FD_ISSET(fd
, &write_mask
)) {
266 if ((hdl
= fd_table
[fd
].write_handler
) != NULL
) {
267 fd_table
[fd
].write_handler
= NULL
;
268 commUpdateWriteBits(fd
, NULL
);
269 hdl(fd
, fd_table
[fd
].write_data
);
271 debugs(5, DBG_IMPORTANT
, "comm_select_incoming: FD " << fd
<< " NULL write handler");
276 return incoming_sockets_accepted
;
280 comm_select_udp_incoming(void)
287 if (Comm::IsConnOpen(icpIncomingConn
)) {
288 fds
[nfds
] = icpIncomingConn
->fd
;
292 if (Comm::IsConnOpen(icpOutgoingConn
) && icpIncomingConn
!= icpOutgoingConn
) {
293 fds
[nfds
] = icpOutgoingConn
->fd
;
300 nevents
= comm_check_incoming_select_handlers(nfds
, fds
);
302 incoming_udp_interval
+= Config
.comm_incoming
.udp
.average
- nevents
;
304 if (incoming_udp_interval
< 0)
305 incoming_udp_interval
= 0;
307 if (incoming_udp_interval
> MAX_INCOMING_INTERVAL
)
308 incoming_udp_interval
= MAX_INCOMING_INTERVAL
;
310 if (nevents
> INCOMING_UDP_MAX
)
311 nevents
= INCOMING_UDP_MAX
;
313 statCounter
.comm_udp_incoming
.count(nevents
);
317 comm_select_tcp_incoming(void)
320 int fds
[MAXTCPLISTENPORTS
];
324 // XXX: only poll sockets that won't be deferred. But how do we identify them?
326 for (const AnyP::PortCfg
*s
= Config
.Sockaddr
.http
; s
; s
= s
->next
) {
327 if (Comm::IsConnOpen(s
->listenConn
)) {
328 fds
[nfds
] = s
->listenConn
->fd
;
333 nevents
= comm_check_incoming_select_handlers(nfds
, fds
);
334 incoming_tcp_interval
+= Config
.comm_incoming
.tcp
.average
- nevents
;
336 if (incoming_tcp_interval
< 0)
337 incoming_tcp_interval
= 0;
339 if (incoming_tcp_interval
> MAX_INCOMING_INTERVAL
)
340 incoming_tcp_interval
= MAX_INCOMING_INTERVAL
;
342 if (nevents
> INCOMING_TCP_MAX
)
343 nevents
= INCOMING_TCP_MAX
;
345 statCounter
.comm_tcp_incoming
.count(nevents
);
348 #define DEBUG_FDBITS 0
349 /* Select on all sockets; call handlers for those that are ready. */
351 Comm::DoSelect(int msec
)
362 int calldns
= 0, calludp
= 0, calltcp
= 0;
375 struct timeval poll_time
;
376 double timeout
= current_dtime
+ (msec
/ 1000.0);
382 start
= current_dtime
;
384 if (commCheckUdpIncoming
)
385 comm_select_udp_incoming();
387 if (commCheckDnsIncoming
)
388 comm_select_dns_incoming();
390 if (commCheckTcpIncoming
)
391 comm_select_tcp_incoming();
393 calldns
= calludp
= calltcp
= 0;
395 maxfd
= Biggest_FD
+ 1;
397 memcpy(&readfds
, &global_readfds
,
398 howmany(maxfd
, FD_MASK_BITS
) * FD_MASK_BYTES
);
400 memcpy(&writefds
, &global_writefds
,
401 howmany(maxfd
, FD_MASK_BITS
) * FD_MASK_BYTES
);
403 /* remove stalled FDs, and deal with pending descriptors */
406 FD_ZERO(&pendingfds
);
408 maxindex
= howmany(maxfd
, FD_MASK_BITS
);
410 fdsp
= (fd_mask
*) & readfds
;
412 for (j
= 0; j
< maxindex
; ++j
) {
413 if ((tmask
= fdsp
[j
]) == 0)
414 continue; /* no bits here */
416 for (k
= 0; k
< FD_MASK_BITS
; ++k
) {
417 if (!EBIT_TEST(tmask
, k
))
420 /* Found a set bit */
421 fd
= (j
* FD_MASK_BITS
) + k
;
423 if (FD_ISSET(fd
, &readfds
) && fd_table
[fd
].flags
.read_pending
) {
424 FD_SET(fd
, &pendingfds
);
431 for (i
= 0; i
< maxfd
; ++i
) {
432 /* Check each open socket for a handler. */
434 if (fd_table
[i
].read_handler
) {
435 assert(FD_ISSET(i
, &readfds
));
438 if (fd_table
[i
].write_handler
) {
439 assert(FD_ISSET(i
, &writefds
));
444 if (nreadfds
+ nwritefds
== 0) {
445 assert(shutting_down
);
446 return COMM_SHUTDOWN
;
449 if (msec
> MAX_POLL_TIME
)
450 msec
= MAX_POLL_TIME
;
456 poll_time
.tv_sec
= msec
/ 1000;
457 poll_time
.tv_usec
= (msec
% 1000) * 1000;
458 ++ statCounter
.syscalls
.selects
;
459 num
= select(maxfd
, &readfds
, &writefds
, NULL
, &poll_time
);
460 ++ statCounter
.select_loops
;
462 if (num
>= 0 || pending
> 0)
465 if (ignoreErrno(errno
))
468 debugs(5, DBG_CRITICAL
, "comm_select: select failure: " << xstrerror());
470 examine_select(&readfds
, &writefds
);
477 if (num
< 0 && !pending
)
482 debugs(5, num
? 5 : 8, "comm_select: " << num
<< "+" << pending
<< " FDs ready");
484 statCounter
.select_fds_hist
.count(num
);
486 if (num
== 0 && pending
== 0)
489 /* Scan return fd masks for ready descriptors */
490 fdsp
= (fd_mask
*) & readfds
;
492 pfdsp
= (fd_mask
*) & pendingfds
;
494 maxindex
= howmany(maxfd
, FD_MASK_BITS
);
496 for (j
= 0; j
< maxindex
; ++j
) {
497 if ((tmask
= (fdsp
[j
] | pfdsp
[j
])) == 0)
498 continue; /* no bits here */
500 for (k
= 0; k
< FD_MASK_BITS
; ++k
) {
502 break; /* no more bits left */
504 if (!EBIT_TEST(tmask
, k
))
507 /* Found a set bit */
508 fd
= (j
* FD_MASK_BITS
) + k
;
510 EBIT_CLR(tmask
, k
); /* this will be done */
514 debugs(5, 9, "FD " << fd
<< " bit set for reading");
516 assert(FD_ISSET(fd
, &readfds
));
520 if (fdIsUdpListener(fd
)) {
530 if (fdIsTcpListener(fd
)) {
536 debugs(5, 6, "comm_select: FD " << fd
<< " ready for reading");
538 if (NULL
== (hdl
= F
->read_handler
))
541 F
->read_handler
= NULL
;
542 F
->flags
.read_pending
= 0;
543 commUpdateReadBits(fd
, NULL
);
544 hdl(fd
, F
->read_data
);
545 ++ statCounter
.select_fds
;
547 if (commCheckUdpIncoming
)
548 comm_select_udp_incoming();
550 if (commCheckDnsIncoming
)
551 comm_select_dns_incoming();
553 if (commCheckTcpIncoming
)
554 comm_select_tcp_incoming();
559 fdsp
= (fd_mask
*) & writefds
;
561 for (j
= 0; j
< maxindex
; ++j
) {
562 if ((tmask
= fdsp
[j
]) == 0)
563 continue; /* no bits here */
565 for (k
= 0; k
< FD_MASK_BITS
; ++k
) {
567 break; /* no more bits left */
569 if (!EBIT_TEST(tmask
, k
))
572 /* Found a set bit */
573 fd
= (j
* FD_MASK_BITS
) + k
;
575 EBIT_CLR(tmask
, k
); /* this will be done */
579 debugs(5, 9, "FD " << fd
<< " bit set for writing");
581 assert(FD_ISSET(fd
, &writefds
));
585 if (fdIsUdpListener(fd
)) {
595 if (fdIsTcpListener(fd
)) {
601 debugs(5, 6, "comm_select: FD " << fd
<< " ready for writing");
603 if ((hdl
= F
->write_handler
)) {
604 F
->write_handler
= NULL
;
605 commUpdateWriteBits(fd
, NULL
);
606 hdl(fd
, F
->write_data
);
607 ++ statCounter
.select_fds
;
609 if (commCheckUdpIncoming
)
610 comm_select_udp_incoming();
612 if (commCheckDnsIncoming
)
613 comm_select_dns_incoming();
615 if (commCheckTcpIncoming
)
616 comm_select_tcp_incoming();
622 comm_select_udp_incoming();
625 comm_select_dns_incoming();
628 comm_select_tcp_incoming();
632 statCounter
.select_time
+= (current_dtime
- start
);
635 } while (timeout
> current_dtime
);
636 debugs(5, 8, "comm_select: time out: " << squid_curtime
);
642 comm_select_dns_incoming(void)
649 if (DnsSocketA
< 0 && DnsSocketB
< 0)
652 if (DnsSocketA
>= 0) {
653 fds
[nfds
] = DnsSocketA
;
657 if (DnsSocketB
>= 0) {
658 fds
[nfds
] = DnsSocketB
;
662 nevents
= comm_check_incoming_select_handlers(nfds
, fds
);
667 incoming_dns_interval
+= Config
.comm_incoming
.dns
.average
- nevents
;
669 if (incoming_dns_interval
< Config
.comm_incoming
.dns
.min_poll
)
670 incoming_dns_interval
= Config
.comm_incoming
.dns
.min_poll
;
672 if (incoming_dns_interval
> MAX_INCOMING_INTERVAL
)
673 incoming_dns_interval
= MAX_INCOMING_INTERVAL
;
675 if (nevents
> INCOMING_DNS_MAX
)
676 nevents
= INCOMING_DNS_MAX
;
678 statCounter
.comm_dns_incoming
.count(nevents
);
682 Comm::SelectLoopInit(void)
686 FD_ZERO(&global_readfds
);
687 FD_ZERO(&global_writefds
);
688 nreadfds
= nwritefds
= 0;
690 Mgr::RegisterAction("comm_select_incoming",
691 "comm_incoming() stats",
692 commIncomingStats
, 0, 1);
696 * examine_select - debug routine.
698 * I spend the day chasing this core dump that occurs when both the client
699 * and the server side of a cache fetch simultaneoulsy abort the
700 * connection. While I haven't really studied the code to figure out how
701 * it happens, the snippet below may prevent the cache from exitting:
703 * Call this from where the select loop fails.
706 examine_select(fd_set
* readfds
, fd_set
* writefds
)
713 AsyncCall::Pointer ch
= NULL
;
717 debugs(5, DBG_CRITICAL
, "examine_select: Examining open file descriptors...");
719 for (fd
= 0; fd
< Squid_MaxFD
; ++fd
) {
722 tv
.tv_sec
= tv
.tv_usec
= 0;
724 if (FD_ISSET(fd
, readfds
))
726 else if (FD_ISSET(fd
, writefds
))
727 FD_SET(fd
, &write_x
);
731 ++ statCounter
.syscalls
.selects
;
734 if (!fstat(fd
, &sb
)) {
735 debugs(5, 5, "FD " << fd
<< " is valid.");
740 debugs(5, DBG_CRITICAL
, "FD " << fd
<< ": " << xstrerror());
741 debugs(5, DBG_CRITICAL
, "WARNING: FD " << fd
<< " has handlers, but it's invalid.");
742 debugs(5, DBG_CRITICAL
, "FD " << fd
<< " is a " << fdTypeStr
[F
->type
] << " called '" << F
->desc
<< "'");
743 debugs(5, DBG_CRITICAL
, "tmout:" << F
->timeoutHandler
<< " read:" << F
->read_handler
<< " write:" << F
->write_handler
);
745 for (ch
= F
->closeHandler
; ch
!= NULL
; ch
= ch
->Next())
746 debugs(5, DBG_CRITICAL
, " close handler: " << ch
);
748 if (F
->closeHandler
!= NULL
) {
749 commCallCloseHandlers(fd
);
750 } else if (F
->timeoutHandler
!= NULL
) {
751 debugs(5, DBG_CRITICAL
, "examine_select: Calling Timeout Handler");
752 ScheduleCallHere(F
->timeoutHandler
);
755 F
->closeHandler
= NULL
;
756 F
->timeoutHandler
= NULL
;
757 F
->read_handler
= NULL
;
758 F
->write_handler
= NULL
;
760 FD_CLR(fd
, writefds
);
767 commIncomingStats(StoreEntry
* sentry
)
769 storeAppendPrintf(sentry
, "Current incoming_udp_interval: %d\n",
770 incoming_udp_interval
>> INCOMING_FACTOR
);
771 storeAppendPrintf(sentry
, "Current incoming_dns_interval: %d\n",
772 incoming_dns_interval
>> INCOMING_FACTOR
);
773 storeAppendPrintf(sentry
, "Current incoming_tcp_interval: %d\n",
774 incoming_tcp_interval
>> INCOMING_FACTOR
);
775 storeAppendPrintf(sentry
, "\n");
776 storeAppendPrintf(sentry
, "Histogram of events per incoming socket type\n");
777 storeAppendPrintf(sentry
, "ICP Messages handled per comm_select_udp_incoming() call:\n");
778 statCounter
.comm_udp_incoming
.dump(sentry
, statHistIntDumper
);
779 storeAppendPrintf(sentry
, "DNS Messages handled per comm_select_dns_incoming() call:\n");
780 statCounter
.comm_dns_incoming
.dump(sentry
, statHistIntDumper
);
781 storeAppendPrintf(sentry
, "HTTP Messages handled per comm_select_tcp_incoming() call:\n");
782 statCounter
.comm_tcp_incoming
.dump(sentry
, statHistIntDumper
);
786 commUpdateReadBits(int fd
, PF
* handler
)
788 if (handler
&& !FD_ISSET(fd
, &global_readfds
)) {
789 FD_SET(fd
, &global_readfds
);
791 } else if (!handler
&& FD_ISSET(fd
, &global_readfds
)) {
792 FD_CLR(fd
, &global_readfds
);
798 commUpdateWriteBits(int fd
, PF
* handler
)
800 if (handler
&& !FD_ISSET(fd
, &global_writefds
)) {
801 FD_SET(fd
, &global_writefds
);
803 } else if (!handler
&& FD_ISSET(fd
, &global_writefds
)) {
804 FD_CLR(fd
, &global_writefds
);
809 /* Called by async-io or diskd to speed up the polling */
811 Comm::QuickPollRequired(void)
816 #endif /* USE_SELECT */