2 * DEBUG: section 05 Socket Functions
3 * AUTHOR: Harvest Derived
5 * SQUID Web Proxy Cache http://www.squid-cache.org/
6 * ----------------------------------------------------------
8 * Squid is the result of efforts by numerous individuals from
9 * the Internet community; see the CONTRIBUTORS file for full
10 * details. Many organizations have provided support for Squid's
11 * development; see the SPONSORS file for full details. Squid is
12 * Copyrighted (C) 2001 by the Regents of the University of
13 * California; see the COPYRIGHT file for full details. Squid
14 * incorporates software developed and/or copyrighted by other
15 * sources; see the CREDITS file for full details.
17 * This program is free software; you can redistribute it and/or modify
18 * it under the terms of the GNU General Public License as published by
19 * the Free Software Foundation; either version 2 of the License, or
20 * (at your option) any later version.
22 * This program is distributed in the hope that it will be useful,
23 * but WITHOUT ANY WARRANTY; without even the implied warranty of
24 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
25 * GNU General Public License for more details.
27 * You should have received a copy of the GNU General Public License
28 * along with this program; if not, write to the Free Software
29 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
32 * Copyright (c) 2003, Robert Collins <robertc@squid-cache.org>
36 #include "base/AsyncCall.h"
37 #include "StoreIOBuffer.h"
41 #include "comm/AcceptLimiter.h"
42 #include "comm/comm_internal.h"
43 #include "comm/Connection.h"
48 #include "SquidTime.h"
49 #include "CommCalls.h"
50 #include "DescriptorSet.h"
51 #include "icmp/net_db.h"
52 #include "ip/Address.h"
53 #include "ip/Intercept.h"
54 #include "ip/QosConfig.h"
57 #if defined(_SQUID_CYGWIN_)
58 #include <sys/ioctl.h>
60 #ifdef HAVE_NETINET_TCP_H
61 #include <netinet/tcp.h>
65 * New C-like simple comm code. This stuff is a mess and doesn't really buy us anything.
74 static void commStopHalfClosedMonitor(int fd
);
75 static IOCB commHalfClosedReader
;
76 static void comm_init_opened(const Comm::ConnectionPointer
&conn
, tos_t tos
, nfmark_t nfmark
, const char *note
, struct addrinfo
*AI
);
77 static int comm_apply_flags(int new_socket
, Ip::Address
&addr
, int flags
, struct addrinfo
*AI
);
80 struct comm_io_callback_t
{
83 AsyncCall::Pointer callback
;
91 bool active() const { return callback
!= NULL
; }
96 comm_io_callback_t readcb
;
97 comm_io_callback_t writecb
;
99 typedef struct _comm_fd comm_fd_t
;
100 comm_fd_t
*commfd_table
;
102 // TODO: make this a comm_io_callback_t method?
104 commio_has_callback(int fd
, iocb_type type
, comm_io_callback_t
*ccb
)
106 assert(ccb
->fd
== fd
);
107 assert(ccb
->type
== type
);
108 return ccb
->active();
112 * Configure comm_io_callback_t for I/O
114 * @param fd filedescriptor
115 * @param ccb comm io callback
117 * @param cbdata callback data (must be cbdata'ed)
118 * @param buf buffer, if applicable
119 * @param freefunc freefunc, if applicable
120 * @param size buffer size
123 commio_set_callback(int fd
, iocb_type type
, comm_io_callback_t
*ccb
,
124 AsyncCall::Pointer
&cb
, char *buf
, FREE
*freefunc
, int size
)
126 assert(!ccb
->active());
127 assert(ccb
->type
== type
);
132 ccb
->freefunc
= freefunc
;
138 // Schedule the callback call and clear the callback
140 commio_finish_callback(int fd
, comm_io_callback_t
*ccb
, comm_err_t code
, int xerrno
)
142 debugs(5, 3, "commio_finish_callback: called for FD " << fd
<< " (" <<
143 code
<< ", " << xerrno
<< ")");
144 assert(ccb
->active());
145 assert(ccb
->fd
== fd
);
147 ccb
->xerrno
= xerrno
;
149 comm_io_callback_t cb
= *ccb
;
151 /* We've got a copy; blow away the real one */
152 /* XXX duplicate code from commio_cancel_callback! */
154 ccb
->callback
= NULL
; // cb has it
162 if (cb
.callback
!= NULL
) {
163 typedef CommIoCbParams Params
;
164 Params
¶ms
= GetCommParams
<Params
>(cb
.callback
);
167 params
.size
= cb
.offset
;
168 params
.flag
= cb
.errcode
;
169 params
.xerrno
= cb
.xerrno
;
170 ScheduleCallHere(cb
.callback
);
176 * Cancel the given callback
178 * Remember that the data is cbdataRef'ed.
180 // TODO: make this a comm_io_callback_t method
182 commio_cancel_callback(int fd
, comm_io_callback_t
*ccb
)
184 debugs(5, 3, "commio_cancel_callback: called for FD " << fd
);
185 assert(ccb
->fd
== fd
);
186 assert(ccb
->active());
189 ccb
->callback
= NULL
;
193 * Call the given comm callback; assumes the callback is valid.
195 * @param ccb io completion callback
198 commio_call_callback(comm_io_callback_t
*ccb
)
204 static DescriptorSet
*TheHalfClosed
= NULL
; /// the set of half-closed FDs
205 static bool WillCheckHalfClosed
= false; /// true if check is scheduled
206 static EVH commHalfClosedCheck
;
207 static void commPlanHalfClosedCheck();
209 static comm_err_t
commBind(int s
, struct addrinfo
&);
210 static void commSetReuseAddr(int);
211 static void commSetNoLinger(int);
213 static void commSetTcpNoDelay(int);
215 static void commSetTcpRcvbuf(int, int);
216 static PF commHandleWrite
;
223 static MemAllocator
*conn_close_pool
= NULL
;
224 fd_debug_t
*fdd_table
= NULL
;
229 return fd_table
[fd
].flags
.open
!= 0;
235 * If the read attempt succeeds or fails, call the callback.
236 * Else, wait for another IO notification.
239 commHandleRead(int fd
, void *data
)
241 comm_io_callback_t
*ccb
= (comm_io_callback_t
*) data
;
243 assert(data
== COMMIO_FD_READCB(fd
));
244 assert(commio_has_callback(fd
, IOCB_READ
, ccb
));
246 statCounter
.syscalls
.sock
.reads
++;
249 retval
= FD_READ_METHOD(fd
, ccb
->buf
, ccb
->size
);
250 debugs(5, 3, "comm_read_try: FD " << fd
<< ", size " << ccb
->size
<< ", retval " << retval
<< ", errno " << errno
);
252 if (retval
< 0 && !ignoreErrno(errno
)) {
253 debugs(5, 3, "comm_read_try: scheduling COMM_ERROR");
255 commio_finish_callback(fd
, ccb
, COMM_ERROR
, errno
);
259 /* See if we read anything */
260 /* Note - read 0 == socket EOF, which is a valid read */
262 fd_bytes(fd
, retval
, FD_READ
);
263 ccb
->offset
= retval
;
264 commio_finish_callback(fd
, ccb
, COMM_OK
, errno
);
268 /* Nope, register for some more IO */
269 commSetSelect(fd
, COMM_SELECT_READ
, commHandleRead
, data
, 0);
273 * Queue a read. handler/handler_data are called when the read
274 * completes, on error, or on file descriptor close.
277 comm_read(const Comm::ConnectionPointer
&conn
, char *buf
, int size
, IOCB
*handler
, void *handler_data
)
279 AsyncCall::Pointer call
= commCbCall(5,4, "SomeCommReadHandler",
280 CommIoCbPtrFun(handler
, handler_data
));
281 comm_read(conn
, buf
, size
, call
);
285 comm_read(const Comm::ConnectionPointer
&conn
, char *buf
, int size
, AsyncCall::Pointer
&callback
)
287 debugs(5, 5, "comm_read, queueing read for " << conn
<< "; asynCall " << callback
);
289 /* Make sure we are open and not closing */
290 assert(Comm::IsConnOpen(conn
));
291 assert(!fd_table
[conn
->fd
].closing());
292 comm_io_callback_t
*ccb
= COMMIO_FD_READCB(conn
->fd
);
294 // Make sure we are either not reading or just passively monitoring.
295 // Active/passive conflicts are OK and simply cancel passive monitoring.
297 // if the assertion below fails, we have an active comm_read conflict
298 assert(fd_table
[conn
->fd
].halfClosedReader
!= NULL
);
299 commStopHalfClosedMonitor(conn
->fd
);
300 assert(!ccb
->active());
304 commio_set_callback(conn
->fd
, IOCB_READ
, ccb
, callback
, (char *)buf
, NULL
, size
);
305 commSetSelect(conn
->fd
, COMM_SELECT_READ
, commHandleRead
, ccb
, 0);
309 * Empty the read buffers
311 * This is a magical routine that empties the read buffers.
312 * Under some platforms (Linux) if a buffer has data in it before
313 * you call close(), the socket will hang and take quite a while
317 comm_empty_os_read_buffers(int fd
)
320 /* prevent those nasty RST packets */
321 char buf
[SQUID_TCP_SO_RCVBUF
];
323 if (fd_table
[fd
].flags
.nonblocking
== 1) {
324 while (FD_READ_METHOD(fd
, buf
, SQUID_TCP_SO_RCVBUF
) > 0) {};
331 * Return whether the FD has a pending completed callback.
334 comm_has_pending_read_callback(int fd
)
337 // XXX: We do not know whether there is a read callback scheduled.
338 // This is used for pconn management that should probably be more
339 // tightly integrated into comm to minimize the chance that a
340 // closing pconn socket will be used for a new transaction.
344 // Does comm check this fd for read readiness?
345 // Note that when comm is not monitoring, there can be a pending callback
346 // call, which may resume comm monitoring once fired.
348 comm_monitors_read(int fd
)
351 // Being active is usually the same as monitoring because we always
352 // start monitoring the FD when we configure comm_io_callback_t for I/O
353 // and we usually configure comm_io_callback_t for I/O when we starting
354 // monitoring a FD for reading. TODO: replace with commio_has_callback
355 return COMMIO_FD_READCB(fd
)->active();
359 * Cancel a pending read. Assert that we have the right parameters,
360 * and that there are no pending read events!
362 * XXX: We do not assert that there are no pending read events and
363 * with async calls it becomes even more difficult.
364 * The whole interface should be reworked to do callback->cancel()
365 * instead of searching for places where the callback may be stored and
366 * updating the state of those places.
368 * AHC Don't call the comm handlers?
371 comm_read_cancel(int fd
, IOCB
*callback
, void *data
)
374 debugs(5, 4, "comm_read_cancel fails: FD " << fd
<< " closed");
378 comm_io_callback_t
*cb
= COMMIO_FD_READCB(fd
);
379 // TODO: is "active" == "monitors FD"?
381 debugs(5, 4, "comm_read_cancel fails: FD " << fd
<< " inactive");
385 typedef CommCbFunPtrCallT
<CommIoCbPtrFun
> Call
;
386 Call
*call
= dynamic_cast<Call
*>(cb
->callback
.getRaw());
388 debugs(5, 4, "comm_read_cancel fails: FD " << fd
<< " lacks callback");
392 call
->cancel("old comm_read_cancel");
394 typedef CommIoCbParams Params
;
395 const Params
¶ms
= GetCommParams
<Params
>(cb
->callback
);
397 /* Ok, we can be reasonably sure we won't lose any data here! */
398 assert(call
->dialer
.handler
== callback
);
399 assert(params
.data
== data
);
401 /* Delete the callback */
402 commio_cancel_callback(fd
, cb
);
404 /* And the IO event */
405 commSetSelect(fd
, COMM_SELECT_READ
, NULL
, NULL
, 0);
409 comm_read_cancel(int fd
, AsyncCall::Pointer
&callback
)
411 callback
->cancel("comm_read_cancel");
414 debugs(5, 4, "comm_read_cancel fails: FD " << fd
<< " closed");
418 comm_io_callback_t
*cb
= COMMIO_FD_READCB(fd
);
421 debugs(5, 4, "comm_read_cancel fails: FD " << fd
<< " inactive");
425 AsyncCall::Pointer call
= cb
->callback
;
426 assert(call
!= NULL
); // XXX: should never fail (active() checks for callback==NULL)
428 /* Ok, we can be reasonably sure we won't lose any data here! */
429 assert(call
== callback
);
431 /* Delete the callback */
432 commio_cancel_callback(fd
, cb
);
434 /* And the IO event */
435 commSetSelect(fd
, COMM_SELECT_READ
, NULL
, NULL
, 0);
440 * synchronous wrapper around udp socket functions
443 comm_udp_recvfrom(int fd
, void *buf
, size_t len
, int flags
, Ip::Address
&from
)
445 statCounter
.syscalls
.sock
.recvfroms
++;
447 struct addrinfo
*AI
= NULL
;
449 debugs(5,8, "comm_udp_recvfrom: FD " << fd
<< " from " << from
);
451 assert( NULL
== AI
);
453 from
.InitAddrInfo(AI
);
455 x
= recvfrom(fd
, buf
, len
, flags
, AI
->ai_addr
, &AI
->ai_addrlen
);
459 from
.FreeAddrInfo(AI
);
465 comm_udp_recv(int fd
, void *buf
, size_t len
, int flags
)
468 return comm_udp_recvfrom(fd
, buf
, len
, flags
, nul
);
472 comm_udp_send(int s
, const void *buf
, size_t len
, int flags
)
474 return send(s
, buf
, len
, flags
);
479 comm_has_incomplete_write(int fd
)
482 return COMMIO_FD_WRITECB(fd
)->active();
486 * Queue a write. handler/handler_data are called when the write fully
487 * completes, on error, or on file descriptor close.
490 /* Return the local port associated with fd. */
492 comm_local_port(int fd
)
495 struct addrinfo
*addr
= NULL
;
496 fde
*F
= &fd_table
[fd
];
498 /* If the fd is closed already, just return */
500 if (!F
->flags
.open
) {
501 debugs(5, 0, "comm_local_port: FD " << fd
<< " has been closed.");
505 if (F
->local_addr
.GetPort())
506 return F
->local_addr
.GetPort();
508 if (F
->sock_family
== AF_INET
)
511 temp
.InitAddrInfo(addr
);
513 if (getsockname(fd
, addr
->ai_addr
, &(addr
->ai_addrlen
)) ) {
514 debugs(50, 1, "comm_local_port: Failed to retrieve TCP/UDP port number for socket: FD " << fd
<< ": " << xstrerror());
515 temp
.FreeAddrInfo(addr
);
520 temp
.FreeAddrInfo(addr
);
522 if (F
->local_addr
.IsAnyAddr()) {
523 /* save the whole local address, not just the port. */
524 F
->local_addr
= temp
;
526 F
->local_addr
.SetPort(temp
.GetPort());
529 debugs(5, 6, "comm_local_port: FD " << fd
<< ": port " << F
->local_addr
.GetPort() << "(family=" << F
->sock_family
<< ")");
530 return F
->local_addr
.GetPort();
534 commBind(int s
, struct addrinfo
&inaddr
)
536 statCounter
.syscalls
.sock
.binds
++;
538 if (bind(s
, inaddr
.ai_addr
, inaddr
.ai_addrlen
) == 0) {
539 debugs(50, 6, "commBind: bind socket FD " << s
<< " to " << fd_table
[s
].local_addr
);
543 debugs(50, 0, "commBind: Cannot bind socket FD " << s
<< " to " << fd_table
[s
].local_addr
<< ": " << xstrerror());
549 * Create a socket. Default is blocking, stream (TCP) socket. IO_TYPE
550 * is OR of flags specified in comm.h. Defaults TOS
553 comm_open(int sock_type
,
559 return comm_openex(sock_type
, proto
, addr
, flags
, 0, 0, note
);
563 comm_open_listener(int sock_type
,
565 Comm::ConnectionPointer
&conn
,
568 /* all listener sockets require bind() */
569 conn
->flags
|= COMM_DOBIND
;
571 /* attempt native enabled port. */
572 conn
->fd
= comm_openex(sock_type
, proto
, conn
->local
, conn
->flags
, 0, note
);
576 comm_open_listener(int sock_type
,
584 /* all listener sockets require bind() */
585 flags
|= COMM_DOBIND
;
587 /* attempt native enabled port. */
588 sock
= comm_openex(sock_type
, proto
, addr
, flags
, 0, 0, note
);
594 limitError(int const anErrno
)
596 return anErrno
== ENFILE
|| anErrno
== EMFILE
;
600 comm_set_v6only(int fd
, int tos
)
603 if (setsockopt(fd
, IPPROTO_IPV6
, IPV6_V6ONLY
, (char *) &tos
, sizeof(int)) < 0) {
604 debugs(50, 1, "comm_open: setsockopt(IPV6_V6ONLY) " << (tos
?"ON":"OFF") << " for FD " << fd
<< ": " << xstrerror());
607 debugs(50, 0, "WARNING: comm_open: setsockopt(IPV6_V6ONLY) not supported on this platform");
612 * Set the socket IP_TRANSPARENT option for Linux TPROXY v4 support.
615 comm_set_transparent(int fd
)
617 #if defined(IP_TRANSPARENT)
619 if (setsockopt(fd
, SOL_IP
, IP_TRANSPARENT
, (char *) &tos
, sizeof(int)) < 0) {
620 debugs(50, DBG_IMPORTANT
, "comm_open: setsockopt(IP_TRANSPARENT) on FD " << fd
<< ": " << xstrerror());
622 /* mark the socket as having transparent options */
623 fd_table
[fd
].flags
.transparent
= 1;
626 debugs(50, DBG_CRITICAL
, "WARNING: comm_open: setsockopt(IP_TRANSPARENT) not supported on this platform");
631 * Create a socket. Default is blocking, stream (TCP) socket. IO_TYPE
632 * is OR of flags specified in defines.h:COMM_*
635 comm_openex(int sock_type
,
644 struct addrinfo
*AI
= NULL
;
646 PROF_start(comm_open
);
647 /* Create socket for accepting new connections. */
648 statCounter
.syscalls
.sock
.sockets
++;
650 /* Setup the socket addrinfo details for use */
651 addr
.GetAddrInfo(AI
);
652 AI
->ai_socktype
= sock_type
;
653 AI
->ai_protocol
= proto
;
655 debugs(50, 3, "comm_openex: Attempt open socket for: " << addr
);
657 new_socket
= socket(AI
->ai_family
, AI
->ai_socktype
, AI
->ai_protocol
);
659 /* under IPv6 there is the possibility IPv6 is present but disabled. */
660 /* try again as IPv4-native if possible */
661 if ( new_socket
< 0 && Ip::EnableIpv6
&& addr
.IsIPv6() && addr
.SetIPv4() ) {
662 /* attempt to open this IPv4-only. */
663 addr
.FreeAddrInfo(AI
);
664 /* Setup the socket addrinfo details for use */
665 addr
.GetAddrInfo(AI
);
666 AI
->ai_socktype
= sock_type
;
667 AI
->ai_protocol
= proto
;
668 debugs(50, 3, "comm_openex: Attempt fallback open socket for: " << addr
);
669 new_socket
= socket(AI
->ai_family
, AI
->ai_socktype
, AI
->ai_protocol
);
670 debugs(50, 2, HERE
<< "attempt open " << note
<< " socket on: " << addr
);
673 if (new_socket
< 0) {
674 /* Increase the number of reserved fd's if calls to socket()
675 * are failing because the open file table is full. This
676 * limits the number of simultaneous clients */
678 if (limitError(errno
)) {
679 debugs(50, DBG_IMPORTANT
, "comm_open: socket failure: " << xstrerror());
682 debugs(50, DBG_CRITICAL
, "comm_open: socket failure: " << xstrerror());
685 addr
.FreeAddrInfo(AI
);
687 PROF_stop(comm_open
);
691 debugs(50, 3, "comm_openex: Opened socket FD " << new_socket
<< " : family=" << AI
->ai_family
<< ", type=" << AI
->ai_socktype
<< ", protocol=" << AI
->ai_protocol
);
693 /* set TOS if needed */
695 Ip::Qos::setSockTos(new_socket
, tos
);
697 /* set netfilter mark if needed */
699 Ip::Qos::setSockNfmark(new_socket
, nfmark
);
701 if ( Ip::EnableIpv6
&IPV6_SPECIAL_SPLITSTACK
&& addr
.IsIPv6() )
702 comm_set_v6only(new_socket
, 1);
704 /* Windows Vista supports Dual-Sockets. BUT defaults them to V6ONLY. Turn it OFF. */
705 /* Other OS may have this administratively disabled for general use. Same deal. */
706 if ( Ip::EnableIpv6
&IPV6_SPECIAL_V4MAPPING
&& addr
.IsIPv6() )
707 comm_set_v6only(new_socket
, 0);
709 // temporary for te transition. comm_openex will eventually have a conn to play with.
710 Comm::ConnectionPointer temp
= new Comm::Connection
;
711 temp
->fd
= new_socket
;
713 comm_init_opened(temp
, tos
, nfmark
, note
, AI
);
714 new_socket
= comm_apply_flags(new_socket
, addr
, flags
, AI
);
716 addr
.FreeAddrInfo(AI
);
718 PROF_stop(comm_open
);
723 /// update FD tables after a local or remote (IPC) comm_openex();
725 comm_init_opened(const Comm::ConnectionPointer
&conn
,
731 assert(Comm::IsConnOpen(conn
));
735 debugs(5, 5, HERE
<< conn
<< " is a new socket");
737 assert(!isOpen(conn
->fd
)); // NP: global isOpen checks the fde entry for openness not the Comm::Connection
738 fd_open(conn
->fd
, FD_SOCKET
, note
);
740 fdd_table
[conn
->fd
].close_file
= NULL
;
741 fdd_table
[conn
->fd
].close_line
= 0;
743 fde
*F
= &fd_table
[conn
->fd
];
744 F
->local_addr
= conn
->local
;
745 F
->tosToServer
= tos
;
746 F
->nfmarkToServer
= nfmark
;
747 F
->sock_family
= AI
->ai_family
;
750 /// apply flags after a local comm_open*() call;
751 /// returns new_socket or -1 on error
753 comm_apply_flags(int new_socket
,
758 assert(new_socket
>= 0);
760 const int sock_type
= AI
->ai_socktype
;
762 if (!(flags
& COMM_NOCLOEXEC
))
763 commSetCloseOnExec(new_socket
);
765 if ((flags
& COMM_REUSEADDR
))
766 commSetReuseAddr(new_socket
);
768 if (addr
.GetPort() > (u_short
) 0) {
770 if (sock_type
!= SOCK_DGRAM
)
772 commSetNoLinger(new_socket
);
775 commSetReuseAddr(new_socket
);
778 /* MUST be done before binding or face OS Error: "(99) Cannot assign requested address"... */
779 if ((flags
& COMM_TRANSPARENT
)) {
780 comm_set_transparent(new_socket
);
783 if ( (flags
& COMM_DOBIND
) || addr
.GetPort() > 0 || !addr
.IsAnyAddr() ) {
784 if ( !(flags
& COMM_DOBIND
) && addr
.IsAnyAddr() )
785 debugs(5,1,"WARNING: Squid is attempting to bind() port " << addr
<< " without being a listener.");
786 if ( addr
.IsNoAddr() )
787 debugs(5,0,"CRITICAL: Squid is attempting to bind() port " << addr
<< "!!");
789 if (commBind(new_socket
, *AI
) != COMM_OK
) {
790 comm_close(new_socket
);
795 if (flags
& COMM_NONBLOCKING
)
796 if (commSetNonBlocking(new_socket
) == COMM_ERROR
) {
797 comm_close(new_socket
);
802 if (sock_type
== SOCK_STREAM
)
803 commSetTcpNoDelay(new_socket
);
807 if (Config
.tcpRcvBufsz
> 0 && sock_type
== SOCK_STREAM
)
808 commSetTcpRcvbuf(new_socket
, Config
.tcpRcvBufsz
);
814 comm_import_opened(const Comm::ConnectionPointer
&conn
,
818 debugs(5, 2, HERE
<< conn
);
819 assert(Comm::IsConnOpen(conn
));
822 comm_init_opened(conn
, 0, 0, note
, AI
);
824 if (!(conn
->flags
& COMM_NOCLOEXEC
))
825 fd_table
[conn
->fd
].flags
.close_on_exec
= 1;
827 if (conn
->local
.GetPort() > (u_short
) 0) {
829 if (AI
->ai_socktype
!= SOCK_DGRAM
)
831 fd_table
[conn
->fd
].flags
.nolinger
= 1;
834 if ((conn
->flags
& COMM_TRANSPARENT
))
835 fd_table
[conn
->fd
].flags
.transparent
= 1;
837 if (conn
->flags
& COMM_NONBLOCKING
)
838 fd_table
[conn
->fd
].flags
.nonblocking
= 1;
841 if (AI
->ai_socktype
== SOCK_STREAM
)
842 fd_table
[conn
->fd
].flags
.nodelay
= 1;
845 /* no fd_table[fd].flags. updates needed for these conditions:
846 * if ((flags & COMM_REUSEADDR)) ...
847 * if ((flags & COMM_DOBIND) ...) ...
853 commSetTimeout_old(int fd
, int timeout
, PF
* handler
, void *data
)
855 debugs(5, 3, HERE
<< "FD " << fd
<< " timeout " << timeout
);
857 assert(fd
< Squid_MaxFD
);
858 fde
*F
= &fd_table
[fd
];
859 assert(F
->flags
.open
);
862 cbdataReferenceDone(F
->timeout_data
);
863 F
->timeout_handler
= NULL
;
867 cbdataReferenceDone(F
->timeout_data
);
868 F
->timeout_handler
= handler
;
869 F
->timeout_data
= cbdataReference(data
);
872 F
->timeout
= squid_curtime
+ (time_t) timeout
;
879 // Legacy pre-AsyncCalls API for FD timeouts.
881 commSetTimeout(int fd
, int timeout
, PF
* handler
, void *data
)
883 AsyncCall::Pointer call
;
884 debugs(5, 3, HERE
<< "FD " << fd
<< " timeout " << timeout
);
886 call
=commCbCall(5,4, "SomeTimeoutHandler", CommTimeoutCbPtrFun(handler
, data
));
889 return commSetTimeout(fd
, timeout
, call
);
892 // Legacy pre-Comm::Connection API for FD timeouts
893 // still used by non-socket FD code dealing with pipes and IPC sockets.
895 commSetTimeout(int fd
, int timeout
, AsyncCall::Pointer
&callback
)
897 debugs(5, 3, HERE
<< "FD " << fd
<< " timeout " << timeout
);
899 assert(fd
< Squid_MaxFD
);
900 fde
*F
= &fd_table
[fd
];
901 assert(F
->flags
.open
);
904 F
->timeoutHandler
= NULL
;
907 if (callback
!= NULL
) {
908 typedef CommTimeoutCbParams Params
;
909 Params
¶ms
= GetCommParams
<Params
>(callback
);
911 F
->timeoutHandler
= callback
;
914 F
->timeout
= squid_curtime
+ (time_t) timeout
;
921 commSetConnTimeout(const Comm::ConnectionPointer
&conn
, int timeout
, AsyncCall::Pointer
&callback
)
923 debugs(5, 3, HERE
<< conn
<< " timeout " << timeout
);
924 assert(Comm::IsConnOpen(conn
));
925 assert(conn
->fd
< Squid_MaxFD
);
926 fde
*F
= &fd_table
[conn
->fd
];
927 assert(F
->flags
.open
);
930 F
->timeoutHandler
= NULL
;
933 if (callback
!= NULL
) {
934 typedef CommTimeoutCbParams Params
;
935 Params
¶ms
= GetCommParams
<Params
>(callback
);
937 F
->timeoutHandler
= callback
;
940 F
->timeout
= squid_curtime
+ (time_t) timeout
;
947 comm_connect_addr(int sock
, const Ip::Address
&address
)
949 comm_err_t status
= COMM_OK
;
950 fde
*F
= &fd_table
[sock
];
954 struct addrinfo
*AI
= NULL
;
955 PROF_start(comm_connect_addr
);
957 assert(address
.GetPort() != 0);
959 debugs(5, 9, HERE
<< "connecting socket FD " << sock
<< " to " << address
<< " (want family: " << F
->sock_family
<< ")");
961 /* Handle IPv6 over IPv4-only socket case.
962 * this case must presently be handled here since the GetAddrInfo asserts on bad mappings.
963 * NP: because commResetFD is private to ConnStateData we have to return an error and
964 * trust its handled properly.
966 if (F
->sock_family
== AF_INET
&& !address
.IsIPv4()) {
968 return COMM_ERR_PROTOCOL
;
971 /* Handle IPv4 over IPv6-only socket case.
972 * This case is presently handled here as it's both a known case and it's
973 * uncertain what error will be returned by the IPv6 stack in such case. It's
974 * possible this will also be handled by the errno checks below after connect()
975 * but needs carefull cross-platform verification, and verifying the address
976 * condition here is simple.
978 if (!F
->local_addr
.IsIPv4() && address
.IsIPv4()) {
980 return COMM_ERR_PROTOCOL
;
983 address
.GetAddrInfo(AI
, F
->sock_family
);
985 /* Establish connection. */
988 if (!F
->flags
.called_connect
) {
989 F
->flags
.called_connect
= 1;
990 statCounter
.syscalls
.sock
.connects
++;
992 x
= connect(sock
, AI
->ai_addr
, AI
->ai_addrlen
);
994 // XXX: ICAP code refuses callbacks during a pending comm_ call
995 // Async calls development will fix this.
1002 debugs(5,5, "comm_connect_addr: sock=" << sock
<< ", addrinfo( " <<
1003 " flags=" << AI
->ai_flags
<<
1004 ", family=" << AI
->ai_family
<<
1005 ", socktype=" << AI
->ai_socktype
<<
1006 ", protocol=" << AI
->ai_protocol
<<
1007 ", &addr=" << AI
->ai_addr
<<
1008 ", addrlen=" << AI
->ai_addrlen
<<
1010 debugs(5, 9, "connect FD " << sock
<< ": (" << x
<< ") " << xstrerror());
1011 debugs(14,9, "connecting to: " << address
);
1014 #if defined(_SQUID_NEWSOS6_)
1015 /* Makoto MATSUSHITA <matusita@ics.es.osaka-u.ac.jp> */
1017 connect(sock
, AI
->ai_addr
, AI
->ai_addrlen
);
1019 if (errno
== EINVAL
) {
1020 errlen
= sizeof(err
);
1021 x
= getsockopt(sock
, SOL_SOCKET
, SO_ERROR
, &err
, &errlen
);
1028 errlen
= sizeof(err
);
1030 x
= getsockopt(sock
, SOL_SOCKET
, SO_ERROR
, &err
, &errlen
);
1035 #if defined(_SQUID_SOLARIS_)
1037 * Solaris 2.4's socket emulation doesn't allow you
1038 * to determine the error from a failed non-blocking
1039 * connect and just returns EPIPE. Create a fake
1040 * error message for connect. -- fenner@parc.xerox.com
1042 if (x
< 0 && errno
== EPIPE
)
1050 /* Squid seems to be working fine without this code. With this code,
1051 * we leak memory on many connect requests because of EINPROGRESS.
1052 * If you find that this code is needed, please file a bug report. */
1054 #ifdef _SQUID_LINUX_
1056 * Linux Debian replaces our allocated AI pointer with garbage when
1057 * connect() fails. This leads to segmentation faults deallocating
1058 * the system-allocated memory when we go to clean up our pointer.
1059 * HACK: is to leak the memory returned since we can't deallocate.
1067 address
.FreeAddrInfo(AI
);
1069 PROF_stop(comm_connect_addr
);
1071 if (errno
== 0 || errno
== EISCONN
)
1073 else if (ignoreErrno(errno
))
1074 status
= COMM_INPROGRESS
;
1075 else if (errno
== EAFNOSUPPORT
|| errno
== EINVAL
)
1076 return COMM_ERR_PROTOCOL
;
1080 address
.NtoA(F
->ipaddr
, MAX_IPSTRLEN
);
1082 F
->remote_port
= address
.GetPort(); /* remote_port is HS */
1084 if (status
== COMM_OK
) {
1085 debugs(5, 10, "comm_connect_addr: FD " << sock
<< " connected to " << address
);
1086 } else if (status
== COMM_INPROGRESS
) {
1087 debugs(5, 10, "comm_connect_addr: FD " << sock
<< " connection pending");
1094 commCallCloseHandlers(int fd
)
1096 fde
*F
= &fd_table
[fd
];
1097 debugs(5, 5, "commCallCloseHandlers: FD " << fd
);
1099 while (F
->closeHandler
!= NULL
) {
1100 AsyncCall::Pointer call
= F
->closeHandler
;
1101 F
->closeHandler
= call
->Next();
1102 call
->setNext(NULL
);
1103 // If call is not canceled schedule it for execution else ignore it
1104 if (!call
->canceled()) {
1105 debugs(5, 5, "commCallCloseHandlers: ch->handler=" << call
);
1106 typedef CommCloseCbParams Params
;
1107 Params
¶ms
= GetCommParams
<Params
>(call
);
1109 ScheduleCallHere(call
);
1116 commLingerClose(int fd
, void *unused
)
1118 LOCAL_ARRAY(char, buf
, 1024);
1120 n
= FD_READ_METHOD(fd
, buf
, 1024);
1123 debugs(5, 3, "commLingerClose: FD " << fd
<< " read: " << xstrerror());
1129 commLingerTimeout(int fd
, void *unused
)
1131 debugs(5, 3, "commLingerTimeout: FD " << fd
);
1136 * Inspired by apache
1139 comm_lingering_close(int fd
)
1143 if (fd_table
[fd
].ssl
)
1144 ssl_shutdown_method(fd
);
1148 if (shutdown(fd
, 1) < 0) {
1153 fd_note(fd
, "lingering close");
1154 commSetTimeout(fd
, 10, commLingerTimeout
, NULL
);
1155 commSetSelect(fd
, COMM_SELECT_READ
, commLingerClose
, NULL
, 0);
1161 * enable linger with time of 0 so that when the socket is
1162 * closed, TCP generates a RESET
1165 comm_reset_close(Comm::ConnectionPointer
&conn
)
1171 if (setsockopt(conn
->fd
, SOL_SOCKET
, SO_LINGER
, (char *) &L
, sizeof(L
)) < 0)
1172 debugs(50, DBG_CRITICAL
, "ERROR: Closing FD " << conn
->fd
<< " with TCP RST: " << xstrerror());
1177 // Legacy close function.
1179 old_comm_reset_close(int fd
)
1185 if (setsockopt(fd
, SOL_SOCKET
, SO_LINGER
, (char *) &L
, sizeof(L
)) < 0)
1186 debugs(50, DBG_CRITICAL
, "ERROR: Closing FD " << fd
<< " with TCP RST: " << xstrerror());
1192 comm_close_start(int fd
, void *data
)
1195 fde
*F
= &fd_table
[fd
];
1197 ssl_shutdown_method(fd
);
1204 comm_close_complete(int fd
, void *data
)
1207 fde
*F
= &fd_table
[fd
];
1215 fd_close(fd
); /* update fdstat */
1219 statCounter
.syscalls
.sock
.closes
++;
1221 /* When an fd closes, give accept() a chance, if need be */
1222 Comm::AcceptLimiter::Instance().kick();
1226 * Close the socket fd.
1228 * + call write handlers with ERR_CLOSING
1229 * + call read handlers with ERR_CLOSING
1230 * + call closing handlers
1232 * NOTE: COMM_ERR_CLOSING will NOT be called for CommReads' sitting in a
1233 * DeferredReadManager.
1236 _comm_close(int fd
, char const *file
, int line
)
1238 debugs(5, 3, "comm_close: start closing FD " << fd
);
1240 assert(fd
< Squid_MaxFD
);
1242 fde
*F
= &fd_table
[fd
];
1243 fdd_table
[fd
].close_file
= file
;
1244 fdd_table
[fd
].close_line
= line
;
1249 /* XXX: is this obsolete behind F->closing() ? */
1250 if ( (shutting_down
|| reconfiguring
) && (!F
->flags
.open
|| F
->type
== FD_FILE
))
1253 /* The following fails because ipc.c is doing calls to pipe() to create sockets! */
1256 assert(F
->type
!= FD_FILE
);
1258 PROF_start(comm_close
);
1260 F
->flags
.close_request
= 1;
1262 AsyncCall::Pointer startCall
=commCbCall(5,4, "comm_close_start",
1263 CommCloseCbPtrFun(comm_close_start
, NULL
));
1264 typedef CommCloseCbParams Params
;
1265 Params
&startParams
= GetCommParams
<Params
>(startCall
);
1266 startParams
.fd
= fd
;
1267 ScheduleCallHere(startCall
);
1269 // a half-closed fd may lack a reader, so we stop monitoring explicitly
1270 if (commHasHalfClosedMonitor(fd
))
1271 commStopHalfClosedMonitor(fd
);
1272 commSetTimeout(fd
, -1, NULL
, NULL
);
1274 // notify read/write handlers after canceling select reservations, if any
1275 if (commio_has_callback(fd
, IOCB_WRITE
, COMMIO_FD_WRITECB(fd
))) {
1276 commSetSelect(fd
, COMM_SELECT_WRITE
, NULL
, NULL
, 0);
1277 commio_finish_callback(fd
, COMMIO_FD_WRITECB(fd
), COMM_ERR_CLOSING
, errno
);
1279 if (commio_has_callback(fd
, IOCB_READ
, COMMIO_FD_READCB(fd
))) {
1280 commSetSelect(fd
, COMM_SELECT_READ
, NULL
, NULL
, 0);
1281 commio_finish_callback(fd
, COMMIO_FD_READCB(fd
), COMM_ERR_CLOSING
, errno
);
1284 commCallCloseHandlers(fd
);
1287 F
->pconn
.pool
->count(F
->pconn
.uses
);
1289 comm_empty_os_read_buffers(fd
);
1292 AsyncCall::Pointer completeCall
=commCbCall(5,4, "comm_close_complete",
1293 CommCloseCbPtrFun(comm_close_complete
, NULL
));
1294 Params
&completeParams
= GetCommParams
<Params
>(completeCall
);
1295 completeParams
.fd
= fd
;
1296 // must use async call to wait for all callbacks
1297 // scheduled before comm_close() to finish
1298 ScheduleCallHere(completeCall
);
1300 PROF_stop(comm_close
);
1303 /* Send a udp datagram to specified TO_ADDR. */
1305 comm_udp_sendto(int fd
,
1306 const Ip::Address
&to_addr
,
1311 struct addrinfo
*AI
= NULL
;
1313 PROF_start(comm_udp_sendto
);
1314 statCounter
.syscalls
.sock
.sendtos
++;
1316 debugs(50, 3, "comm_udp_sendto: Attempt to send UDP packet to " << to_addr
<<
1317 " using FD " << fd
<< " using Port " << comm_local_port(fd
) );
1319 /* BUG: something in the above macro appears to occasionally be setting AI to garbage. */
1320 /* AYJ: 2007-08-27 : or was it because I wasn't then setting 'fd_table[fd].sock_family' to fill properly. */
1321 assert( NULL
== AI
);
1323 to_addr
.GetAddrInfo(AI
, fd_table
[fd
].sock_family
);
1325 x
= sendto(fd
, buf
, len
, 0, AI
->ai_addr
, AI
->ai_addrlen
);
1327 to_addr
.FreeAddrInfo(AI
);
1329 PROF_stop(comm_udp_sendto
);
1334 #ifdef _SQUID_LINUX_
1336 if (ECONNREFUSED
!= errno
)
1339 debugs(50, 1, "comm_udp_sendto: FD " << fd
<< ", (family=" << fd_table
[fd
].sock_family
<< ") " << to_addr
<< ": " << xstrerror());
1345 comm_add_close_handler(int fd
, PF
* handler
, void *data
)
1347 debugs(5, 5, "comm_add_close_handler: FD " << fd
<< ", handler=" <<
1348 handler
<< ", data=" << data
);
1350 AsyncCall::Pointer call
=commCbCall(5,4, "SomeCloseHandler",
1351 CommCloseCbPtrFun(handler
, data
));
1352 comm_add_close_handler(fd
, call
);
1356 comm_add_close_handler(int fd
, AsyncCall::Pointer
&call
)
1358 debugs(5, 5, "comm_add_close_handler: FD " << fd
<< ", AsyncCall=" << call
);
1360 /*TODO:Check for a similar scheduled AsyncCall*/
1361 // for (c = fd_table[fd].closeHandler; c; c = c->next)
1362 // assert(c->handler != handler || c->data != data);
1364 call
->setNext(fd_table
[fd
].closeHandler
);
1366 fd_table
[fd
].closeHandler
= call
;
1370 // remove function-based close handler
1372 comm_remove_close_handler(int fd
, PF
* handler
, void *data
)
1374 assert (isOpen(fd
));
1375 /* Find handler in list */
1376 debugs(5, 5, "comm_remove_close_handler: FD " << fd
<< ", handler=" <<
1377 handler
<< ", data=" << data
);
1379 AsyncCall::Pointer p
;
1380 for (p
= fd_table
[fd
].closeHandler
; p
!= NULL
; p
= p
->Next()) {
1381 typedef CommCbFunPtrCallT
<CommCloseCbPtrFun
> Call
;
1382 const Call
*call
= dynamic_cast<const Call
*>(p
.getRaw());
1383 if (!call
) // method callbacks have their own comm_remove_close_handler
1386 typedef CommCloseCbParams Params
;
1387 const Params
¶ms
= GetCommParams
<Params
>(p
);
1388 if (call
->dialer
.handler
== handler
&& params
.data
== data
)
1389 break; /* This is our handler */
1392 // comm_close removes all close handlers so our handler may be gone
1394 p
->cancel("comm_remove_close_handler");
1395 // TODO: should we remove the handler from the close handlers list?
1398 // remove method-based close handler
1400 comm_remove_close_handler(int fd
, AsyncCall::Pointer
&call
)
1402 assert (isOpen(fd
));
1403 debugs(5, 5, "comm_remove_close_handler: FD " << fd
<< ", AsyncCall=" << call
);
1405 // comm_close removes all close handlers so our handler may be gone
1406 // TODO: should we remove the handler from the close handlers list?
1408 // Check to see if really exist the given AsyncCall in comm_close handlers
1409 // TODO: optimize: this slow code is only needed for the assert() below
1410 AsyncCall::Pointer p
;
1411 for (p
= fd_table
[fd
].closeHandler
; p
!= NULL
&& p
!= call
; p
= p
->Next());
1415 call
->cancel("comm_remove_close_handler");
1419 commSetNoLinger(int fd
)
1423 L
.l_onoff
= 0; /* off */
1426 if (setsockopt(fd
, SOL_SOCKET
, SO_LINGER
, (char *) &L
, sizeof(L
)) < 0)
1427 debugs(50, 0, "commSetNoLinger: FD " << fd
<< ": " << xstrerror());
1429 fd_table
[fd
].flags
.nolinger
= 1;
1433 commSetReuseAddr(int fd
)
1437 if (setsockopt(fd
, SOL_SOCKET
, SO_REUSEADDR
, (char *) &on
, sizeof(on
)) < 0)
1438 debugs(50, 1, "commSetReuseAddr: FD " << fd
<< ": " << xstrerror());
1442 commSetTcpRcvbuf(int fd
, int size
)
1444 if (setsockopt(fd
, SOL_SOCKET
, SO_RCVBUF
, (char *) &size
, sizeof(size
)) < 0)
1445 debugs(50, 1, "commSetTcpRcvbuf: FD " << fd
<< ", SIZE " << size
<< ": " << xstrerror());
1446 if (setsockopt(fd
, SOL_SOCKET
, SO_SNDBUF
, (char *) &size
, sizeof(size
)) < 0)
1447 debugs(50, 1, "commSetTcpRcvbuf: FD " << fd
<< ", SIZE " << size
<< ": " << xstrerror());
1448 #ifdef TCP_WINDOW_CLAMP
1449 if (setsockopt(fd
, SOL_TCP
, TCP_WINDOW_CLAMP
, (char *) &size
, sizeof(size
)) < 0)
1450 debugs(50, 1, "commSetTcpRcvbuf: FD " << fd
<< ", SIZE " << size
<< ": " << xstrerror());
1455 commSetNonBlocking(int fd
)
1457 #ifndef _SQUID_MSWIN_
1461 #ifdef _SQUID_WIN32_
1463 int nonblocking
= TRUE
;
1465 #ifdef _SQUID_CYGWIN_
1467 if (fd_table
[fd
].type
!= FD_PIPE
) {
1470 if (ioctl(fd
, FIONBIO
, &nonblocking
) < 0) {
1471 debugs(50, 0, "commSetNonBlocking: FD " << fd
<< ": " << xstrerror() << " " << fd_table
[fd
].type
);
1475 #ifdef _SQUID_CYGWIN_
1480 #ifndef _SQUID_MSWIN_
1482 if ((flags
= fcntl(fd
, F_GETFL
, dummy
)) < 0) {
1483 debugs(50, 0, "FD " << fd
<< ": fcntl F_GETFL: " << xstrerror());
1487 if (fcntl(fd
, F_SETFL
, flags
| SQUID_NONBLOCK
) < 0) {
1488 debugs(50, 0, "commSetNonBlocking: FD " << fd
<< ": " << xstrerror());
1493 #ifdef _SQUID_CYGWIN_
1498 fd_table
[fd
].flags
.nonblocking
= 1;
1504 commUnsetNonBlocking(int fd
)
1506 #ifdef _SQUID_MSWIN_
1507 int nonblocking
= FALSE
;
1509 if (ioctlsocket(fd
, FIONBIO
, (unsigned long *) &nonblocking
) < 0) {
1514 if ((flags
= fcntl(fd
, F_GETFL
, dummy
)) < 0) {
1515 debugs(50, 0, "FD " << fd
<< ": fcntl F_GETFL: " << xstrerror());
1519 if (fcntl(fd
, F_SETFL
, flags
& (~SQUID_NONBLOCK
)) < 0) {
1521 debugs(50, 0, "commUnsetNonBlocking: FD " << fd
<< ": " << xstrerror());
1525 fd_table
[fd
].flags
.nonblocking
= 0;
1530 commSetCloseOnExec(int fd
)
1536 if ((flags
= fcntl(fd
, F_GETFD
, dummy
)) < 0) {
1537 debugs(50, 0, "FD " << fd
<< ": fcntl F_GETFD: " << xstrerror());
1541 if (fcntl(fd
, F_SETFD
, flags
| FD_CLOEXEC
) < 0)
1542 debugs(50, 0, "FD " << fd
<< ": set close-on-exec failed: " << xstrerror());
1544 fd_table
[fd
].flags
.close_on_exec
= 1;
1551 commSetTcpNoDelay(int fd
)
1555 if (setsockopt(fd
, IPPROTO_TCP
, TCP_NODELAY
, (char *) &on
, sizeof(on
)) < 0)
1556 debugs(50, 1, "commSetTcpNoDelay: FD " << fd
<< ": " << xstrerror());
1558 fd_table
[fd
].flags
.nodelay
= 1;
1564 commSetTcpKeepalive(int fd
, int idle
, int interval
, int timeout
)
1568 if (timeout
&& interval
) {
1569 int count
= (timeout
+ interval
- 1) / interval
;
1570 if (setsockopt(fd
, IPPROTO_TCP
, TCP_KEEPCNT
, &count
, sizeof(on
)) < 0)
1571 debugs(5, 1, "commSetKeepalive: FD " << fd
<< ": " << xstrerror());
1576 if (setsockopt(fd
, IPPROTO_TCP
, TCP_KEEPIDLE
, &idle
, sizeof(on
)) < 0)
1577 debugs(5, 1, "commSetKeepalive: FD " << fd
<< ": " << xstrerror());
1580 #ifdef TCP_KEEPINTVL
1582 if (setsockopt(fd
, IPPROTO_TCP
, TCP_KEEPINTVL
, &interval
, sizeof(on
)) < 0)
1583 debugs(5, 1, "commSetKeepalive: FD " << fd
<< ": " << xstrerror());
1586 if (setsockopt(fd
, SOL_SOCKET
, SO_KEEPALIVE
, (char *) &on
, sizeof(on
)) < 0)
1587 debugs(5, 1, "commSetKeepalive: FD " << fd
<< ": " << xstrerror());
1593 fd_table
=(fde
*) xcalloc(Squid_MaxFD
, sizeof(fde
));
1594 fdd_table
= (fd_debug_t
*)xcalloc(Squid_MaxFD
, sizeof(fd_debug_t
));
1596 /* make sure the accept() socket FIFO delay queue exists */
1597 Comm::AcceptLimiter::Instance();
1599 commfd_table
= (comm_fd_t
*) xcalloc(Squid_MaxFD
, sizeof(comm_fd_t
));
1600 for (int pos
= 0; pos
< Squid_MaxFD
; pos
++) {
1601 commfd_table
[pos
].fd
= pos
;
1602 commfd_table
[pos
].readcb
.fd
= pos
;
1603 commfd_table
[pos
].readcb
.type
= IOCB_READ
;
1604 commfd_table
[pos
].writecb
.fd
= pos
;
1605 commfd_table
[pos
].writecb
.type
= IOCB_WRITE
;
1608 /* XXX account fd_table */
1609 /* Keep a few file descriptors free so that we don't run out of FD's
1610 * after accepting a client but before it opens a socket or a file.
1611 * Since Squid_MaxFD can be as high as several thousand, don't waste them */
1612 RESERVED_FD
= min(100, Squid_MaxFD
/ 4);
1614 conn_close_pool
= memPoolCreate("close_handler", sizeof(close_handler
));
1616 TheHalfClosed
= new DescriptorSet
;
1622 delete TheHalfClosed
;
1623 TheHalfClosed
= NULL
;
1625 safe_free(fd_table
);
1626 safe_free(fdd_table
);
1627 safe_free(commfd_table
);
1632 commHandleWrite(int fd
, void *data
)
1634 comm_io_callback_t
*state
= (comm_io_callback_t
*)data
;
1638 assert(state
== COMMIO_FD_WRITECB(fd
));
1640 PROF_start(commHandleWrite
);
1641 debugs(5, 5, "commHandleWrite: FD " << fd
<< ": off " <<
1642 (long int) state
->offset
<< ", sz " << (long int) state
->size
<< ".");
1644 nleft
= state
->size
- state
->offset
;
1645 len
= FD_WRITE_METHOD(fd
, state
->buf
+ state
->offset
, nleft
);
1646 debugs(5, 5, "commHandleWrite: write() returns " << len
);
1647 fd_bytes(fd
, len
, FD_WRITE
);
1648 statCounter
.syscalls
.sock
.writes
++;
1649 // After each successful partial write,
1650 // reset fde::writeStart to the current time.
1651 fd_table
[fd
].writeStart
= squid_curtime
;
1654 /* Note we even call write if nleft == 0 */
1658 debugs(5, 1, "commHandleWrite: FD " << fd
<< ": write failure: connection closed with " << nleft
<< " bytes remaining.");
1660 commio_finish_callback(fd
, COMMIO_FD_WRITECB(fd
), nleft
? COMM_ERROR
: COMM_OK
, errno
);
1661 } else if (len
< 0) {
1664 if (fd_table
[fd
].flags
.socket_eof
) {
1665 debugs(50, 2, "commHandleWrite: FD " << fd
<< ": write failure: " << xstrerror() << ".");
1666 commio_finish_callback(fd
, COMMIO_FD_WRITECB(fd
), nleft
? COMM_ERROR
: COMM_OK
, errno
);
1667 } else if (ignoreErrno(errno
)) {
1668 debugs(50, 10, "commHandleWrite: FD " << fd
<< ": write failure: " << xstrerror() << ".");
1675 debugs(50, 2, "commHandleWrite: FD " << fd
<< ": write failure: " << xstrerror() << ".");
1676 commio_finish_callback(fd
, COMMIO_FD_WRITECB(fd
), nleft
? COMM_ERROR
: COMM_OK
, errno
);
1679 /* A successful write, continue */
1680 state
->offset
+= len
;
1682 if (state
->offset
< state
->size
) {
1683 /* Not done, reinstall the write handler and write some more */
1690 commio_finish_callback(fd
, COMMIO_FD_WRITECB(fd
), nleft
? COMM_OK
: COMM_ERROR
, errno
);
1694 PROF_stop(commHandleWrite
);
1698 * Queue a write. handler/handler_data are called when the write
1699 * completes, on error, or on file descriptor close.
1701 * free_func is used to free the passed buffer when the write has completed.
1704 comm_write(const Comm::ConnectionPointer
&conn
, const char *buf
, int size
, IOCB
* handler
, void *handler_data
, FREE
* free_func
)
1706 AsyncCall::Pointer call
= commCbCall(5,5, "SomeCommWriteHander",
1707 CommIoCbPtrFun(handler
, handler_data
));
1709 comm_write(conn
, buf
, size
, call
, free_func
);
1713 comm_write(const Comm::ConnectionPointer
&conn
, const char *buf
, int size
, AsyncCall::Pointer
&callback
, FREE
* free_func
)
1715 debugs(5, 5, HERE
<< conn
<< ": sz " << size
<< ": asynCall " << callback
);
1717 /* Make sure we are open, not closing, and not writing */
1718 assert(Comm::IsConnOpen(conn
));
1719 assert(!fd_table
[conn
->fd
].closing());
1720 comm_io_callback_t
*ccb
= COMMIO_FD_WRITECB(conn
->fd
);
1721 assert(!ccb
->active());
1723 fd_table
[conn
->fd
].writeStart
= squid_curtime
;
1724 /* Queue the write */
1725 commio_set_callback(conn
->fd
, IOCB_WRITE
, ccb
, callback
,
1726 (char *)buf
, free_func
, size
);
1727 commSetSelect(conn
->fd
, COMM_SELECT_WRITE
, commHandleWrite
, ccb
, 0);
1731 /* a wrapper around comm_write to allow for MemBuf to be comm_written in a snap */
1734 comm_write_mbuf(int fd
, MemBuf
*mb
, IOCB
* handler
, void *handler_data
)
1736 comm_write(fd
, mb
->buf
, mb
->size
, handler
, handler_data
, mb
->freeFunc());
1740 comm_write_mbuf(int fd
, MemBuf
*mb
, AsyncCall::Pointer
&callback
)
1742 comm_write(fd
, mb
->buf
, mb
->size
, callback
, mb
->freeFunc());
1747 comm_write_mbuf(const Comm::ConnectionPointer
&conn
, MemBuf
*mb
, IOCB
* handler
, void *handler_data
)
1749 comm_write(conn
, mb
->buf
, mb
->size
, handler
, handler_data
, mb
->freeFunc());
1753 comm_write_mbuf(const Comm::ConnectionPointer
&conn
, MemBuf
*mb
, AsyncCall::Pointer
&callback
)
1755 comm_write(conn
, mb
->buf
, mb
->size
, callback
, mb
->freeFunc());
1760 * hm, this might be too general-purpose for all the places we'd
1764 ignoreErrno(int ierrno
)
1771 #if EAGAIN != EWOULDBLOCK
1794 commCloseAllSockets(void)
1799 for (fd
= 0; fd
<= Biggest_FD
; fd
++) {
1805 if (F
->type
!= FD_SOCKET
)
1808 if (F
->flags
.ipc
) /* don't close inter-process sockets */
1811 if (F
->timeoutHandler
!= NULL
) {
1812 AsyncCall::Pointer callback
= F
->timeoutHandler
;
1813 F
->timeoutHandler
= NULL
;
1814 debugs(5, 5, "commCloseAllSockets: FD " << fd
<< ": Calling timeout handler");
1815 ScheduleCallHere(callback
);
1817 debugs(5, 5, "commCloseAllSockets: FD " << fd
<< ": calling comm_reset_close()");
1818 old_comm_reset_close(fd
);
1824 AlreadyTimedOut(fde
*F
)
1829 if (F
->timeout
== 0)
1832 if (F
->timeout
> squid_curtime
)
1839 writeTimedOut(int fd
)
1841 if (!commio_has_callback(fd
, IOCB_WRITE
, COMMIO_FD_WRITECB(fd
)))
1844 if ((squid_curtime
- fd_table
[fd
].writeStart
) < Config
.Timeout
.write
)
1855 AsyncCall::Pointer callback
;
1857 for (fd
= 0; fd
<= Biggest_FD
; fd
++) {
1860 if (writeTimedOut(fd
)) {
1861 // We have an active write callback and we are timed out
1862 debugs(5, 5, "checkTimeouts: FD " << fd
<< " auto write timeout");
1863 commSetSelect(fd
, COMM_SELECT_WRITE
, NULL
, NULL
, 0);
1864 commio_finish_callback(fd
, COMMIO_FD_WRITECB(fd
), COMM_ERROR
, ETIMEDOUT
);
1865 } else if (AlreadyTimedOut(F
))
1868 debugs(5, 5, "checkTimeouts: FD " << fd
<< " Expired");
1870 if (F
->timeoutHandler
!= NULL
) {
1871 debugs(5, 5, "checkTimeouts: FD " << fd
<< ": Call timeout handler");
1872 callback
= F
->timeoutHandler
;
1873 F
->timeoutHandler
= NULL
;
1874 ScheduleCallHere(callback
);
1876 debugs(5, 5, "checkTimeouts: FD " << fd
<< ": Forcing comm_close()");
1882 void CommIO::Initialise()
1884 /* Initialize done pipe signal */
1886 if (pipe(DonePipe
)) {}
1887 DoneFD
= DonePipe
[1];
1888 DoneReadFD
= DonePipe
[0];
1889 fd_open(DoneReadFD
, FD_PIPE
, "async-io completetion event: main");
1890 fd_open(DoneFD
, FD_PIPE
, "async-io completetion event: threads");
1891 commSetNonBlocking(DoneReadFD
);
1892 commSetNonBlocking(DoneFD
);
1893 commSetSelect(DoneReadFD
, COMM_SELECT_READ
, NULLFDHandler
, NULL
, 0);
1897 void CommIO::NotifyIOClose()
1899 /* Close done pipe signal */
1904 fd_close(DoneReadFD
);
1905 Initialised
= false;
1908 bool CommIO::Initialised
= false;
1909 bool CommIO::DoneSignalled
= false;
1910 int CommIO::DoneFD
= -1;
1911 int CommIO::DoneReadFD
= -1;
1917 FD_READ_METHOD(DoneReadFD
, buf
, sizeof(buf
));
1921 CommIO::NULLFDHandler(int fd
, void *data
)
1924 commSetSelect(fd
, COMM_SELECT_READ
, NULLFDHandler
, NULL
, 0);
1928 CommIO::ResetNotifications()
1930 if (DoneSignalled
) {
1932 DoneSignalled
= false;
1936 /// Start waiting for a possibly half-closed connection to close
1937 // by scheduling a read callback to a monitoring handler that
1938 // will close the connection on read errors.
1940 commStartHalfClosedMonitor(int fd
)
1942 debugs(5, 5, HERE
<< "adding FD " << fd
<< " to " << *TheHalfClosed
);
1944 assert(!commHasHalfClosedMonitor(fd
));
1945 (void)TheHalfClosed
->add(fd
); // could also assert the result
1946 commPlanHalfClosedCheck(); // may schedule check if we added the first FD
1951 commPlanHalfClosedCheck()
1953 if (!WillCheckHalfClosed
&& !TheHalfClosed
->empty()) {
1954 eventAdd("commHalfClosedCheck", &commHalfClosedCheck
, NULL
, 1.0, 1);
1955 WillCheckHalfClosed
= true;
1959 /// iterates over all descriptors that may need half-closed tests and
1960 /// calls comm_read for those that do; re-schedules the check if needed
1963 commHalfClosedCheck(void *)
1965 debugs(5, 5, HERE
<< "checking " << *TheHalfClosed
);
1967 typedef DescriptorSet::const_iterator DSCI
;
1968 const DSCI end
= TheHalfClosed
->end();
1969 for (DSCI i
= TheHalfClosed
->begin(); i
!= end
; ++i
) {
1970 Comm::ConnectionPointer c
= new Comm::Connection
; // XXX: temporary. make HalfClosed a list of these.
1972 if (!fd_table
[c
->fd
].halfClosedReader
) { // not reading already
1973 AsyncCall::Pointer call
= commCbCall(5,4, "commHalfClosedReader",
1974 CommIoCbPtrFun(&commHalfClosedReader
, NULL
));
1975 comm_read(c
, NULL
, 0, call
);
1976 fd_table
[c
->fd
].halfClosedReader
= call
;
1978 c
->fd
= -1; // XXX: temporary. prevent c replacement erase closing listed FD
1981 WillCheckHalfClosed
= false; // as far as we know
1982 commPlanHalfClosedCheck(); // may need to check again
1985 /// checks whether we are waiting for possibly half-closed connection to close
1986 // We are monitoring if the read handler for the fd is the monitoring handler.
1988 commHasHalfClosedMonitor(int fd
)
1990 return TheHalfClosed
->has(fd
);
1993 /// stop waiting for possibly half-closed connection to close
1995 commStopHalfClosedMonitor(int const fd
)
1997 debugs(5, 5, HERE
<< "removing FD " << fd
<< " from " << *TheHalfClosed
);
1999 // cancel the read if one was scheduled
2000 AsyncCall::Pointer reader
= fd_table
[fd
].halfClosedReader
;
2002 comm_read_cancel(fd
, reader
);
2003 fd_table
[fd
].halfClosedReader
= NULL
;
2005 TheHalfClosed
->del(fd
);
2008 /// I/O handler for the possibly half-closed connection monitoring code
2010 commHalfClosedReader(const Comm::ConnectionPointer
&conn
, char *, size_t size
, comm_err_t flag
, int, void *)
2012 // there cannot be more data coming in on half-closed connections
2014 assert(conn
!= NULL
);
2015 assert(commHasHalfClosedMonitor(conn
->fd
)); // or we would have canceled the read
2017 fd_table
[conn
->fd
].halfClosedReader
= NULL
; // done reading, for now
2019 // nothing to do if fd is being closed
2020 if (flag
== COMM_ERR_CLOSING
)
2023 // if read failed, close the connection
2024 if (flag
!= COMM_OK
) {
2025 debugs(5, 3, HERE
<< "closing " << conn
);
2030 // continue waiting for close or error
2031 commPlanHalfClosedCheck(); // make sure this fd will be checked again
2035 CommRead::CommRead() : fd(-1), buf(NULL
), len(0), callback(NULL
) {}
2037 CommRead::CommRead(int fd_
, char *buf_
, int len_
, AsyncCall::Pointer
&callback_
)
2038 : fd(fd_
), buf(buf_
), len(len_
), callback(callback_
) {}
2040 DeferredRead::DeferredRead () : theReader(NULL
), theContext(NULL
), theRead(), cancelled(false) {}
2042 DeferredRead::DeferredRead (DeferrableRead
*aReader
, void *data
, CommRead
const &aRead
) : theReader(aReader
), theContext (data
), theRead(aRead
), cancelled(false) {}
2044 DeferredReadManager::~DeferredReadManager()
2047 assert (deferredReads
.empty());
2050 /* explicit instantiation required for some systems */
2052 /// \cond AUTODOCS-IGNORE
2053 template cbdata_type CbDataList
<DeferredRead
>::CBDATA_CbDataList
;
2057 DeferredReadManager::delayRead(DeferredRead
const &aRead
)
2059 debugs(5, 3, "Adding deferred read on FD " << aRead
.theRead
.fd
);
2060 CbDataList
<DeferredRead
> *temp
= deferredReads
.push_back(aRead
);
2062 // We have to use a global function as a closer and point to temp
2063 // instead of "this" because DeferredReadManager is not a job and
2064 // is not even cbdata protected
2065 AsyncCall::Pointer closer
= commCbCall(5,4,
2066 "DeferredReadManager::CloseHandler",
2067 CommCloseCbPtrFun(&CloseHandler
, temp
));
2068 comm_add_close_handler(aRead
.theRead
.fd
, closer
);
2069 temp
->element
.closer
= closer
; // remeber so that we can cancel
2073 DeferredReadManager::CloseHandler(int fd
, void *thecbdata
)
2075 if (!cbdataReferenceValid (thecbdata
))
2078 CbDataList
<DeferredRead
> *temp
= (CbDataList
<DeferredRead
> *)thecbdata
;
2080 temp
->element
.closer
= NULL
;
2081 temp
->element
.markCancelled();
2085 DeferredReadManager::popHead(CbDataListContainer
<DeferredRead
> &deferredReads
)
2087 assert (!deferredReads
.empty());
2089 DeferredRead
&read
= deferredReads
.head
->element
;
2090 if (!read
.cancelled
) {
2091 comm_remove_close_handler(read
.theRead
.fd
, read
.closer
);
2095 DeferredRead result
= deferredReads
.pop_front();
2101 DeferredReadManager::kickReads(int const count
)
2103 /* if we had CbDataList::size() we could consolidate this and flushReads */
2110 size_t remaining
= count
;
2112 while (!deferredReads
.empty() && remaining
) {
2113 DeferredRead aRead
= popHead(deferredReads
);
2116 if (!aRead
.cancelled
)
2122 DeferredReadManager::flushReads()
2124 CbDataListContainer
<DeferredRead
> reads
;
2125 reads
= deferredReads
;
2126 deferredReads
= CbDataListContainer
<DeferredRead
>();
2128 // XXX: For fairness this SHOULD randomize the order
2129 while (!reads
.empty()) {
2130 DeferredRead aRead
= popHead(reads
);
2136 DeferredReadManager::kickARead(DeferredRead
const &aRead
)
2138 if (aRead
.cancelled
)
2141 if (aRead
.theRead
.fd
>=0 && fd_table
[aRead
.theRead
.fd
].closing())
2144 debugs(5, 3, "Kicking deferred read on FD " << aRead
.theRead
.fd
);
2146 aRead
.theReader(aRead
.theContext
, aRead
.theRead
);
2150 DeferredRead::markCancelled()
2156 CommSelectEngine::checkEvents(int timeout
)
2158 static time_t last_timeout
= 0;
2160 /* No, this shouldn't be here. But it shouldn't be in each comm handler. -adrian */
2161 if (squid_curtime
> last_timeout
) {
2162 last_timeout
= squid_curtime
;
2166 switch (comm_select(timeout
)) {
2182 fatal_dump("comm.cc: Internal error -- this should never happen.");
2187 /// Create a unix-domain socket (UDS) that only supports FD_MSGHDR I/O.
2189 comm_open_uds(int sock_type
,
2191 struct sockaddr_un
* addr
,
2194 // TODO: merge with comm_openex() when Ip::Address becomes NetAddress
2198 PROF_start(comm_open
);
2199 /* Create socket for accepting new connections. */
2200 statCounter
.syscalls
.sock
.sockets
++;
2202 /* Setup the socket addrinfo details for use */
2205 AI
.ai_family
= PF_UNIX
;
2206 AI
.ai_socktype
= sock_type
;
2207 AI
.ai_protocol
= proto
;
2208 AI
.ai_addrlen
= SUN_LEN(addr
);
2209 AI
.ai_addr
= (sockaddr
*)addr
;
2210 AI
.ai_canonname
= NULL
;
2213 debugs(50, 3, HERE
<< "Attempt open socket for: " << addr
->sun_path
);
2215 if ((new_socket
= socket(AI
.ai_family
, AI
.ai_socktype
, AI
.ai_protocol
)) < 0) {
2216 /* Increase the number of reserved fd's if calls to socket()
2217 * are failing because the open file table is full. This
2218 * limits the number of simultaneous clients */
2220 if (limitError(errno
)) {
2221 debugs(50, DBG_IMPORTANT
, HERE
<< "socket failure: " << xstrerror());
2224 debugs(50, DBG_CRITICAL
, HERE
<< "socket failure: " << xstrerror());
2227 PROF_stop(comm_open
);
2231 debugs(50, 3, HERE
"Opened UDS FD " << new_socket
<< " : family=" << AI
.ai_family
<< ", type=" << AI
.ai_socktype
<< ", protocol=" << AI
.ai_protocol
);
2234 debugs(50, 5, HERE
<< "FD " << new_socket
<< " is a new socket");
2236 assert(!isOpen(new_socket
));
2237 fd_open(new_socket
, FD_MSGHDR
, NULL
);
2239 fdd_table
[new_socket
].close_file
= NULL
;
2241 fdd_table
[new_socket
].close_line
= 0;
2243 fd_table
[new_socket
].sock_family
= AI
.ai_family
;
2245 if (!(flags
& COMM_NOCLOEXEC
))
2246 commSetCloseOnExec(new_socket
);
2248 if (flags
& COMM_REUSEADDR
)
2249 commSetReuseAddr(new_socket
);
2251 if (flags
& COMM_NONBLOCKING
) {
2252 if (commSetNonBlocking(new_socket
) != COMM_OK
) {
2253 comm_close(new_socket
);
2254 PROF_stop(comm_open
);
2259 if (flags
& COMM_DOBIND
) {
2260 if (commBind(new_socket
, AI
) != COMM_OK
) {
2261 comm_close(new_socket
);
2262 PROF_stop(comm_open
);
2268 if (sock_type
== SOCK_STREAM
)
2269 commSetTcpNoDelay(new_socket
);
2273 if (Config
.tcpRcvBufsz
> 0 && sock_type
== SOCK_STREAM
)
2274 commSetTcpRcvbuf(new_socket
, Config
.tcpRcvBufsz
);
2276 PROF_stop(comm_open
);