src/comm.cc

   1 /*
   2  * Copyright (C) 1996-2015 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 /* DEBUG: section 05    Socket Functions */
  10
  11 #include "squid.h"
  12 #include "ClientInfo.h"
  13 #include "comm/AcceptLimiter.h"
  14 #include "comm/comm_internal.h"
  15 #include "comm/Connection.h"
  16 #include "comm/IoCallback.h"
  17 #include "comm/Loops.h"
  18 #include "comm/Read.h"
  19 #include "comm/TcpAcceptor.h"
  20 #include "comm/Write.h"
  21 #include "CommRead.h"
  22 #include "compat/cmsg.h"
  23 #include "DescriptorSet.h"
  24 #include "event.h"
  25 #include "fd.h"
  26 #include "fde.h"
  27 #include "globals.h"
  28 #include "icmp/net_db.h"
  29 #include "ip/Intercept.h"
  30 #include "ip/QosConfig.h"
  31 #include "ip/tools.h"
  32 #include "pconn.h"
  33 #include "profiler/Profiler.h"
  34 #include "SBuf.h"
  35 #include "SquidConfig.h"
  36 #include "StatCounters.h"
  37 #include "StoreIOBuffer.h"
  38 #include "tools.h"
  39
  40 #if USE_OPENSSL
  41 #include "ssl/support.h"
  42 #endif
  43
  44 #include <cerrno>
  45 #include <cmath>
  46 #if _SQUID_CYGWIN_
  47 #include <sys/ioctl.h>
  48 #endif
  49 #ifdef HAVE_NETINET_TCP_H
  50 #include <netinet/tcp.h>
  51 #endif
  52 #if HAVE_SYS_UN_H
  53 #include <sys/un.h>
  54 #endif
  55
  56 /*
  57  * New C-like simple comm code. This stuff is a mess and doesn't really buy us anything.
  58  */
  59
  60 static IOCB commHalfClosedReader;
  61 static void comm_init_opened(const Comm::ConnectionPointer &conn, const char *note, struct addrinfo *AI);
  62 static int comm_apply_flags(int new_socket, Ip::Address &addr, int flags, struct addrinfo *AI);
  63
  64 #if USE_DELAY_POOLS
  65 CBDATA_CLASS_INIT(CommQuotaQueue);
  66
  67 static void commHandleWriteHelper(void * data);
  68 #endif
  69
  70 /* STATIC */
  71
  72 static DescriptorSet *TheHalfClosed = NULL; /// the set of half-closed FDs
  73 static bool WillCheckHalfClosed = false; /// true if check is scheduled
  74 static EVH commHalfClosedCheck;
  75 static void commPlanHalfClosedCheck();
  76
  77 static Comm::Flag commBind(int s, struct addrinfo &);
  78 static void commSetReuseAddr(int);
  79 static void commSetNoLinger(int);
  80 #ifdef TCP_NODELAY
  81 static void commSetTcpNoDelay(int);
  82 #endif
  83 static void commSetTcpRcvbuf(int, int);
  84
  85 fd_debug_t *fdd_table = NULL;
  86
  87 bool
  88 isOpen(const int fd)
  89 {
  90     return fd >= 0 && fd_table && fd_table[fd].flags.open != 0;
  91 }
  92
  93 /**
  94  * Empty the read buffers
  95  *
  96  * This is a magical routine that empties the read buffers.
  97  * Under some platforms (Linux) if a buffer has data in it before
  98  * you call close(), the socket will hang and take quite a while
  99  * to timeout.
 100  */
 101 static void
 102 comm_empty_os_read_buffers(int fd)
 103 {
 104 #if _SQUID_LINUX_
 105 #if USE_OPENSSL
 106     // Bug 4146: SSL-Bump BIO does not release sockets on close.
 107     if (fd_table[fd].ssl)
 108         return;
 109 #endif
 110
 111     /* prevent those nasty RST packets */
 112     char buf[SQUID_TCP_SO_RCVBUF];
 113     if (fd_table[fd].flags.nonblocking) {
 114         while (FD_READ_METHOD(fd, buf, SQUID_TCP_SO_RCVBUF) > 0) {};
 115     }
 116 #endif
 117 }
 118
 119 /**
 120  * synchronous wrapper around udp socket functions
 121  */
 122 int
 123 comm_udp_recvfrom(int fd, void *buf, size_t len, int flags, Ip::Address &from)
 124 {
 125     ++ statCounter.syscalls.sock.recvfroms;
 126     debugs(5,8, "comm_udp_recvfrom: FD " << fd << " from " << from);
 127     struct addrinfo *AI = NULL;
 128     Ip::Address::InitAddr(AI);
 129     int x = recvfrom(fd, buf, len, flags, AI->ai_addr, &AI->ai_addrlen);
 130     from = *AI;
 131     Ip::Address::FreeAddr(AI);
 132     return x;
 133 }
 134
 135 int
 136 comm_udp_recv(int fd, void *buf, size_t len, int flags)
 137 {
 138     Ip::Address nul;
 139     return comm_udp_recvfrom(fd, buf, len, flags, nul);
 140 }
 141
 142 ssize_t
 143 comm_udp_send(int s, const void *buf, size_t len, int flags)
 144 {
 145     return send(s, buf, len, flags);
 146 }
 147
 148 bool
 149 comm_has_incomplete_write(int fd)
 150 {
 151     assert(isOpen(fd) && COMMIO_FD_WRITECB(fd) != NULL);
 152     return COMMIO_FD_WRITECB(fd)->active();
 153 }
 154
 155 /**
 156  * Queue a write. handler/handler_data are called when the write fully
 157  * completes, on error, or on file descriptor close.
 158  */
 159
 160 /* Return the local port associated with fd. */
 161 unsigned short
 162 comm_local_port(int fd)
 163 {
 164     Ip::Address temp;
 165     struct addrinfo *addr = NULL;
 166     fde *F = &fd_table[fd];
 167
 168     /* If the fd is closed already, just return */
 169
 170     if (!F->flags.open) {
 171         debugs(5, 0, "comm_local_port: FD " << fd << " has been closed.");
 172         return 0;
 173     }
 174
 175     if (F->local_addr.port())
 176         return F->local_addr.port();
 177
 178     if (F->sock_family == AF_INET)
 179         temp.setIPv4();
 180
 181     Ip::Address::InitAddr(addr);
 182
 183     if (getsockname(fd, addr->ai_addr, &(addr->ai_addrlen)) ) {
 184         debugs(50, DBG_IMPORTANT, "comm_local_port: Failed to retrieve TCP/UDP port number for socket: FD " << fd << ": " << xstrerror());
 185         Ip::Address::FreeAddr(addr);
 186         return 0;
 187     }
 188     temp = *addr;
 189
 190     Ip::Address::FreeAddr(addr);
 191
 192     if (F->local_addr.isAnyAddr()) {
 193         /* save the whole local address, not just the port. */
 194         F->local_addr = temp;
 195     } else {
 196         F->local_addr.port(temp.port());
 197     }
 198
 199     debugs(5, 6, "comm_local_port: FD " << fd << ": port " << F->local_addr.port() << "(family=" << F->sock_family << ")");
 200     return F->local_addr.port();
 201 }
 202
 203 static Comm::Flag
 204 commBind(int s, struct addrinfo &inaddr)
 205 {
 206     ++ statCounter.syscalls.sock.binds;
 207
 208     if (bind(s, inaddr.ai_addr, inaddr.ai_addrlen) == 0) {
 209         debugs(50, 6, "commBind: bind socket FD " << s << " to " << fd_table[s].local_addr);
 210         return Comm::OK;
 211     }
 212
 213     debugs(50, 0, "commBind: Cannot bind socket FD " << s << " to " << fd_table[s].local_addr << ": " << xstrerror());
 214
 215     return Comm::COMM_ERROR;
 216 }
 217
 218 /**
 219  * Create a socket. Default is blocking, stream (TCP) socket.  IO_TYPE
 220  * is OR of flags specified in comm.h. Defaults TOS
 221  */
 222 int
 223 comm_open(int sock_type,
 224           int proto,
 225           Ip::Address &addr,
 226           int flags,
 227           const char *note)
 228 {
 229     return comm_openex(sock_type, proto, addr, flags, note);
 230 }
 231
 232 void
 233 comm_open_listener(int sock_type,
 234                    int proto,
 235                    Comm::ConnectionPointer &conn,
 236                    const char *note)
 237 {
 238     /* all listener sockets require bind() */
 239     conn->flags |= COMM_DOBIND;
 240
 241     /* attempt native enabled port. */
 242     conn->fd = comm_openex(sock_type, proto, conn->local, conn->flags, note);
 243 }
 244
 245 int
 246 comm_open_listener(int sock_type,
 247                    int proto,
 248                    Ip::Address &addr,
 249                    int flags,
 250                    const char *note)
 251 {
 252     int sock = -1;
 253
 254     /* all listener sockets require bind() */
 255     flags |= COMM_DOBIND;
 256
 257     /* attempt native enabled port. */
 258     sock = comm_openex(sock_type, proto, addr, flags, note);
 259
 260     return sock;
 261 }
 262
 263 static bool
 264 limitError(int const anErrno)
 265 {
 266     return anErrno == ENFILE || anErrno == EMFILE;
 267 }
 268
 269 void
 270 comm_set_v6only(int fd, int tos)
 271 {
 272 #ifdef IPV6_V6ONLY
 273     if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &tos, sizeof(int)) < 0) {
 274         debugs(50, DBG_IMPORTANT, "comm_open: setsockopt(IPV6_V6ONLY) " << (tos?"ON":"OFF") << " for FD " << fd << ": " << xstrerror());
 275     }
 276 #else
 277     debugs(50, 0, "WARNING: comm_open: setsockopt(IPV6_V6ONLY) not supported on this platform");
 278 #endif /* sockopt */
 279 }
 280
 281 /**
 282  * Set the socket option required for TPROXY spoofing for:
 283  * - Linux TPROXY v4 support,
 284  * - OpenBSD divert-to support,
 285  * - FreeBSD IPFW TPROXY v4 support.
 286  */
 287 void
 288 comm_set_transparent(int fd)
 289 {
 290 #if _SQUID_LINUX_ && defined(IP_TRANSPARENT) // Linux
 291 # define soLevel SOL_IP
 292 # define soFlag  IP_TRANSPARENT
 293     bool doneSuid = false;
 294
 295 #elif defined(SO_BINDANY) // OpenBSD 4.7+ and NetBSD with PF
 296 # define soLevel SOL_SOCKET
 297 # define soFlag  SO_BINDANY
 298     enter_suid();
 299     bool doneSuid = true;
 300
 301 #elif defined(IP_BINDANY) // FreeBSD with IPFW
 302 # define soLevel IPPROTO_IP
 303 # define soFlag  IP_BINDANY
 304     enter_suid();
 305     bool doneSuid = true;
 306
 307 #else
 308     debugs(50, DBG_CRITICAL, "WARNING: comm_open: setsockopt(TPROXY) not supported on this platform");
 309 #endif /* sockopt */
 310
 311 #if defined(soLevel) && defined(soFlag)
 312     int tos = 1;
 313     if (setsockopt(fd, soLevel, soFlag, (char *) &tos, sizeof(int)) < 0) {
 314         debugs(50, DBG_IMPORTANT, "comm_open: setsockopt(TPROXY) on FD " << fd << ": " << xstrerror());
 315     } else {
 316         /* mark the socket as having transparent options */
 317         fd_table[fd].flags.transparent = true;
 318     }
 319     if (doneSuid)
 320         leave_suid();
 321 #endif
 322 }
 323
 324 /**
 325  * Create a socket. Default is blocking, stream (TCP) socket.  IO_TYPE
 326  * is OR of flags specified in defines.h:COMM_*
 327  */
 328 int
 329 comm_openex(int sock_type,
 330             int proto,
 331             Ip::Address &addr,
 332             int flags,
 333             const char *note)
 334 {
 335     int new_socket;
 336     struct addrinfo *AI = NULL;
 337
 338     PROF_start(comm_open);
 339     /* Create socket for accepting new connections. */
 340     ++ statCounter.syscalls.sock.sockets;
 341
 342     /* Setup the socket addrinfo details for use */
 343     addr.getAddrInfo(AI);
 344     AI->ai_socktype = sock_type;
 345     AI->ai_protocol = proto;
 346
 347     debugs(50, 3, "comm_openex: Attempt open socket for: " << addr );
 348
 349     new_socket = socket(AI->ai_family, AI->ai_socktype, AI->ai_protocol);
 350
 351     /* under IPv6 there is the possibility IPv6 is present but disabled. */
 352     /* try again as IPv4-native if possible */
 353     if ( new_socket < 0 && Ip::EnableIpv6 && addr.isIPv6() && addr.setIPv4() ) {
 354         /* attempt to open this IPv4-only. */
 355         Ip::Address::FreeAddr(AI);
 356         /* Setup the socket addrinfo details for use */
 357         addr.getAddrInfo(AI);
 358         AI->ai_socktype = sock_type;
 359         AI->ai_protocol = proto;
 360         debugs(50, 3, "comm_openex: Attempt fallback open socket for: " << addr );
 361         new_socket = socket(AI->ai_family, AI->ai_socktype, AI->ai_protocol);
 362         debugs(50, 2, HERE << "attempt open " << note << " socket on: " << addr);
 363     }
 364
 365     if (new_socket < 0) {
 366         /* Increase the number of reserved fd's if calls to socket()
 367          * are failing because the open file table is full.  This
 368          * limits the number of simultaneous clients */
 369
 370         if (limitError(errno)) {
 371             debugs(50, DBG_IMPORTANT, "comm_open: socket failure: " << xstrerror());
 372             fdAdjustReserved();
 373         } else {
 374             debugs(50, DBG_CRITICAL, "comm_open: socket failure: " << xstrerror());
 375         }
 376
 377         Ip::Address::FreeAddr(AI);
 378
 379         PROF_stop(comm_open);
 380         return -1;
 381     }
 382
 383     // XXX: temporary for the transition. comm_openex will eventually have a conn to play with.
 384     Comm::ConnectionPointer conn = new Comm::Connection;
 385     conn->local = addr;
 386     conn->fd = new_socket;
 387
 388     debugs(50, 3, "comm_openex: Opened socket " << conn << " : family=" << AI->ai_family << ", type=" << AI->ai_socktype << ", protocol=" << AI->ai_protocol );
 389
 390     if ( Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && addr.isIPv6() )
 391         comm_set_v6only(conn->fd, 1);
 392
 393     /* Windows Vista supports Dual-Sockets. BUT defaults them to V6ONLY. Turn it OFF. */
 394     /* Other OS may have this administratively disabled for general use. Same deal. */
 395     if ( Ip::EnableIpv6&IPV6_SPECIAL_V4MAPPING && addr.isIPv6() )
 396         comm_set_v6only(conn->fd, 0);
 397
 398     comm_init_opened(conn, note, AI);
 399     new_socket = comm_apply_flags(conn->fd, addr, flags, AI);
 400
 401     Ip::Address::FreeAddr(AI);
 402
 403     PROF_stop(comm_open);
 404
 405     // XXX transition only. prevent conn from closing the new FD on function exit.
 406     conn->fd = -1;
 407     return new_socket;
 408 }
 409
 410 /// update FD tables after a local or remote (IPC) comm_openex();
 411 void
 412 comm_init_opened(const Comm::ConnectionPointer &conn,
 413                  const char *note,
 414                  struct addrinfo *AI)
 415 {
 416     assert(Comm::IsConnOpen(conn));
 417     assert(AI);
 418
 419     /* update fdstat */
 420     debugs(5, 5, HERE << conn << " is a new socket");
 421
 422     assert(!isOpen(conn->fd)); // NP: global isOpen checks the fde entry for openness not the Comm::Connection
 423     fd_open(conn->fd, FD_SOCKET, note);
 424
 425     fdd_table[conn->fd].close_file = NULL;
 426     fdd_table[conn->fd].close_line = 0;
 427
 428     fde *F = &fd_table[conn->fd];
 429     F->local_addr = conn->local;
 430
 431     F->sock_family = AI->ai_family;
 432 }
 433
 434 /// apply flags after a local comm_open*() call;
 435 /// returns new_socket or -1 on error
 436 static int
 437 comm_apply_flags(int new_socket,
 438                  Ip::Address &addr,
 439                  int flags,
 440                  struct addrinfo *AI)
 441 {
 442     assert(new_socket >= 0);
 443     assert(AI);
 444     const int sock_type = AI->ai_socktype;
 445
 446     if (!(flags & COMM_NOCLOEXEC))
 447         commSetCloseOnExec(new_socket);
 448
 449     if ((flags & COMM_REUSEADDR))
 450         commSetReuseAddr(new_socket);
 451
 452     if (addr.port() > (unsigned short) 0) {
 453 #if _SQUID_WINDOWS_
 454         if (sock_type != SOCK_DGRAM)
 455 #endif
 456             commSetNoLinger(new_socket);
 457
 458         if (opt_reuseaddr)
 459             commSetReuseAddr(new_socket);
 460     }
 461
 462     /* MUST be done before binding or face OS Error: "(99) Cannot assign requested address"... */
 463     if ((flags & COMM_TRANSPARENT)) {
 464         comm_set_transparent(new_socket);
 465     }
 466
 467     if ( (flags & COMM_DOBIND) || addr.port() > 0 || !addr.isAnyAddr() ) {
 468         if ( !(flags & COMM_DOBIND) && addr.isAnyAddr() )
 469             debugs(5, DBG_IMPORTANT,"WARNING: Squid is attempting to bind() port " << addr << " without being a listener.");
 470         if ( addr.isNoAddr() )
 471             debugs(5,0,"CRITICAL: Squid is attempting to bind() port " << addr << "!!");
 472
 473         if (commBind(new_socket, *AI) != Comm::OK) {
 474             comm_close(new_socket);
 475             return -1;
 476         }
 477     }
 478
 479     if (flags & COMM_NONBLOCKING)
 480         if (commSetNonBlocking(new_socket) == Comm::COMM_ERROR) {
 481             comm_close(new_socket);
 482             return -1;
 483         }
 484
 485 #ifdef TCP_NODELAY
 486     if (sock_type == SOCK_STREAM)
 487         commSetTcpNoDelay(new_socket);
 488
 489 #endif
 490
 491     if (Config.tcpRcvBufsz > 0 && sock_type == SOCK_STREAM)
 492         commSetTcpRcvbuf(new_socket, Config.tcpRcvBufsz);
 493
 494     return new_socket;
 495 }
 496
 497 void
 498 comm_import_opened(const Comm::ConnectionPointer &conn,
 499                    const char *note,
 500                    struct addrinfo *AI)
 501 {
 502     debugs(5, 2, HERE << conn);
 503     assert(Comm::IsConnOpen(conn));
 504     assert(AI);
 505
 506     comm_init_opened(conn, note, AI);
 507
 508     if (!(conn->flags & COMM_NOCLOEXEC))
 509         fd_table[conn->fd].flags.close_on_exec = true;
 510
 511     if (conn->local.port() > (unsigned short) 0) {
 512 #if _SQUID_WINDOWS_
 513         if (AI->ai_socktype != SOCK_DGRAM)
 514 #endif
 515             fd_table[conn->fd].flags.nolinger = true;
 516     }
 517
 518     if ((conn->flags & COMM_TRANSPARENT))
 519         fd_table[conn->fd].flags.transparent = true;
 520
 521     if (conn->flags & COMM_NONBLOCKING)
 522         fd_table[conn->fd].flags.nonblocking = true;
 523
 524 #ifdef TCP_NODELAY
 525     if (AI->ai_socktype == SOCK_STREAM)
 526         fd_table[conn->fd].flags.nodelay = true;
 527 #endif
 528
 529     /* no fd_table[fd].flags. updates needed for these conditions:
 530      * if ((flags & COMM_REUSEADDR)) ...
 531      * if ((flags & COMM_DOBIND) ...) ...
 532      */
 533 }
 534
 535 // XXX: now that raw-FD timeouts are only unset for pipes and files this SHOULD be a no-op.
 536 // With handler already unset. Leaving this present until that can be verified for all code paths.
 537 void
 538 commUnsetFdTimeout(int fd)
 539 {
 540     debugs(5, 3, HERE << "Remove timeout for FD " << fd);
 541     assert(fd >= 0);
 542     assert(fd < Squid_MaxFD);
 543     fde *F = &fd_table[fd];
 544     assert(F->flags.open);
 545
 546     F->timeoutHandler = NULL;
 547     F->timeout = 0;
 548 }
 549
 550 int
 551 commSetConnTimeout(const Comm::ConnectionPointer &conn, int timeout, AsyncCall::Pointer &callback)
 552 {
 553     debugs(5, 3, HERE << conn << " timeout " << timeout);
 554     assert(Comm::IsConnOpen(conn));
 555     assert(conn->fd < Squid_MaxFD);
 556     fde *F = &fd_table[conn->fd];
 557     assert(F->flags.open);
 558
 559     if (timeout < 0) {
 560         F->timeoutHandler = NULL;
 561         F->timeout = 0;
 562     } else {
 563         if (callback != NULL) {
 564             typedef CommTimeoutCbParams Params;
 565             Params &params = GetCommParams<Params>(callback);
 566             params.conn = conn;
 567             F->timeoutHandler = callback;
 568         }
 569
 570         F->timeout = squid_curtime + (time_t) timeout;
 571     }
 572
 573     return F->timeout;
 574 }
 575
 576 int
 577 commUnsetConnTimeout(const Comm::ConnectionPointer &conn)
 578 {
 579     debugs(5, 3, HERE << "Remove timeout for " << conn);
 580     AsyncCall::Pointer nil;
 581     return commSetConnTimeout(conn, -1, nil);
 582 }
 583
 584 int
 585 comm_connect_addr(int sock, const Ip::Address &address)
 586 {
 587     Comm::Flag status = Comm::OK;
 588     fde *F = &fd_table[sock];
 589     int x = 0;
 590     int err = 0;
 591     socklen_t errlen;
 592     struct addrinfo *AI = NULL;
 593     PROF_start(comm_connect_addr);
 594
 595     assert(address.port() != 0);
 596
 597     debugs(5, 9, HERE << "connecting socket FD " << sock << " to " << address << " (want family: " << F->sock_family << ")");
 598
 599     /* Handle IPv6 over IPv4-only socket case.
 600      * this case must presently be handled here since the getAddrInfo asserts on bad mappings.
 601      * NP: because commResetFD is private to ConnStateData we have to return an error and
 602      *     trust its handled properly.
 603      */
 604     if (F->sock_family == AF_INET && !address.isIPv4()) {
 605         errno = ENETUNREACH;
 606         return Comm::ERR_PROTOCOL;
 607     }
 608
 609     /* Handle IPv4 over IPv6-only socket case.
 610      * This case is presently handled here as it's both a known case and it's
 611      * uncertain what error will be returned by the IPv6 stack in such case. It's
 612      * possible this will also be handled by the errno checks below after connect()
 613      * but needs carefull cross-platform verification, and verifying the address
 614      * condition here is simple.
 615      */
 616     if (!F->local_addr.isIPv4() && address.isIPv4()) {
 617         errno = ENETUNREACH;
 618         return Comm::ERR_PROTOCOL;
 619     }
 620
 621     address.getAddrInfo(AI, F->sock_family);
 622
 623     /* Establish connection. */
 624     errno = 0;
 625
 626     if (!F->flags.called_connect) {
 627         F->flags.called_connect = true;
 628         ++ statCounter.syscalls.sock.connects;
 629
 630         x = connect(sock, AI->ai_addr, AI->ai_addrlen);
 631
 632         // XXX: ICAP code refuses callbacks during a pending comm_ call
 633         // Async calls development will fix this.
 634         if (x == 0) {
 635             x = -1;
 636             errno = EINPROGRESS;
 637         }
 638
 639         if (x < 0) {
 640             debugs(5,5, "comm_connect_addr: sock=" << sock << ", addrinfo( " <<
 641                    " flags=" << AI->ai_flags <<
 642                    ", family=" << AI->ai_family <<
 643                    ", socktype=" << AI->ai_socktype <<
 644                    ", protocol=" << AI->ai_protocol <<
 645                    ", &addr=" << AI->ai_addr <<
 646                    ", addrlen=" << AI->ai_addrlen <<
 647                    " )" );
 648             debugs(5, 9, "connect FD " << sock << ": (" << x << ") " << xstrerror());
 649             debugs(14,9, "connecting to: " << address );
 650         }
 651     } else {
 652 #if _SQUID_NEWSOS6_
 653         /* Makoto MATSUSHITA <matusita@ics.es.osaka-u.ac.jp> */
 654
 655         connect(sock, AI->ai_addr, AI->ai_addrlen);
 656
 657         if (errno == EINVAL) {
 658             errlen = sizeof(err);
 659             x = getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &errlen);
 660
 661             if (x >= 0)
 662                 errno = x;
 663         }
 664
 665 #else
 666         errlen = sizeof(err);
 667
 668         x = getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &errlen);
 669
 670         if (x == 0)
 671             errno = err;
 672
 673 #if _SQUID_SOLARIS_
 674         /*
 675         * Solaris 2.4's socket emulation doesn't allow you
 676         * to determine the error from a failed non-blocking
 677         * connect and just returns EPIPE.  Create a fake
 678         * error message for connect.   -- fenner@parc.xerox.com
 679         */
 680         if (x < 0 && errno == EPIPE)
 681             errno = ENOTCONN;
 682
 683 #endif
 684 #endif
 685
 686     }
 687
 688     Ip::Address::FreeAddr(AI);
 689
 690     PROF_stop(comm_connect_addr);
 691
 692     if (errno == 0 || errno == EISCONN)
 693         status = Comm::OK;
 694     else if (ignoreErrno(errno))
 695         status = Comm::INPROGRESS;
 696     else if (errno == EAFNOSUPPORT || errno == EINVAL)
 697         return Comm::ERR_PROTOCOL;
 698     else
 699         return Comm::COMM_ERROR;
 700
 701     address.toStr(F->ipaddr, MAX_IPSTRLEN);
 702
 703     F->remote_port = address.port(); /* remote_port is HS */
 704
 705     if (status == Comm::OK) {
 706         debugs(5, DBG_DATA, "comm_connect_addr: FD " << sock << " connected to " << address);
 707     } else if (status == Comm::INPROGRESS) {
 708         debugs(5, DBG_DATA, "comm_connect_addr: FD " << sock << " connection pending");
 709     }
 710
 711     return status;
 712 }
 713
 714 void
 715 commCallCloseHandlers(int fd)
 716 {
 717     fde *F = &fd_table[fd];
 718     debugs(5, 5, "commCallCloseHandlers: FD " << fd);
 719
 720     while (F->closeHandler != NULL) {
 721         AsyncCall::Pointer call = F->closeHandler;
 722         F->closeHandler = call->Next();
 723         call->setNext(NULL);
 724         // If call is not canceled schedule it for execution else ignore it
 725         if (!call->canceled()) {
 726             debugs(5, 5, "commCallCloseHandlers: ch->handler=" << call);
 727             ScheduleCallHere(call);
 728         }
 729     }
 730 }
 731
 732 #if LINGERING_CLOSE
 733 static void
 734 commLingerClose(int fd, void *unused)
 735 {
 736     LOCAL_ARRAY(char, buf, 1024);
 737     int n;
 738     n = FD_READ_METHOD(fd, buf, 1024);
 739
 740     if (n < 0)
 741         debugs(5, 3, "commLingerClose: FD " << fd << " read: " << xstrerror());
 742
 743     comm_close(fd);
 744 }
 745
 746 static void
 747 commLingerTimeout(const FdeCbParams &params)
 748 {
 749     debugs(5, 3, "commLingerTimeout: FD " << params.fd);
 750     comm_close(params.fd);
 751 }
 752
 753 /*
 754  * Inspired by apache
 755  */
 756 void
 757 comm_lingering_close(int fd)
 758 {
 759 #if USE_OPENSSL
 760     if (fd_table[fd].ssl)
 761         ssl_shutdown_method(fd_table[fd].ssl);
 762 #endif
 763
 764     if (shutdown(fd, 1) < 0) {
 765         comm_close(fd);
 766         return;
 767     }
 768
 769     fd_note(fd, "lingering close");
 770     AsyncCall::Pointer call = commCbCall(5,4, "commLingerTimeout", FdeCbPtrFun(commLingerTimeout, NULL));
 771
 772     debugs(5, 3, HERE << "FD " << fd << " timeout " << timeout);
 773     assert(fd_table[fd].flags.open);
 774     if (callback != NULL) {
 775         typedef FdeCbParams Params;
 776         Params &params = GetCommParams<Params>(callback);
 777         params.fd = fd;
 778         fd_table[fd].timeoutHandler = callback;
 779         fd_table[fd].timeout = squid_curtime + static_cast<time_t>(10);
 780     }
 781
 782     Comm::SetSelect(fd, COMM_SELECT_READ, commLingerClose, NULL, 0);
 783 }
 784
 785 #endif
 786
 787 /**
 788  * enable linger with time of 0 so that when the socket is
 789  * closed, TCP generates a RESET
 790  */
 791 void
 792 comm_reset_close(const Comm::ConnectionPointer &conn)
 793 {
 794     struct linger L;
 795     L.l_onoff = 1;
 796     L.l_linger = 0;
 797
 798     if (setsockopt(conn->fd, SOL_SOCKET, SO_LINGER, (char *) &L, sizeof(L)) < 0)
 799         debugs(50, DBG_CRITICAL, "ERROR: Closing " << conn << " with TCP RST: " << xstrerror());
 800
 801     conn->close();
 802 }
 803
 804 // Legacy close function.
 805 void
 806 old_comm_reset_close(int fd)
 807 {
 808     struct linger L;
 809     L.l_onoff = 1;
 810     L.l_linger = 0;
 811
 812     if (setsockopt(fd, SOL_SOCKET, SO_LINGER, (char *) &L, sizeof(L)) < 0)
 813         debugs(50, DBG_CRITICAL, "ERROR: Closing FD " << fd << " with TCP RST: " << xstrerror());
 814
 815     comm_close(fd);
 816 }
 817
 818 #if USE_OPENSSL
 819 void
 820 commStartSslClose(const FdeCbParams &params)
 821 {
 822     assert(fd_table[params.fd].ssl != NULL);
 823     ssl_shutdown_method(fd_table[params.fd].ssl);
 824 }
 825 #endif
 826
 827 void
 828 comm_close_complete(const FdeCbParams &params)
 829 {
 830 #if USE_OPENSSL
 831     fde *F = &fd_table[params.fd];
 832
 833     if (F->ssl) {
 834         SSL_free(F->ssl);
 835         F->ssl = NULL;
 836     }
 837
 838     if (F->dynamicSslContext) {
 839         SSL_CTX_free(F->dynamicSslContext);
 840         F->dynamicSslContext = NULL;
 841     }
 842 #endif
 843     fd_close(params.fd);        /* update fdstat */
 844     close(params.fd);
 845
 846     ++ statCounter.syscalls.sock.closes;
 847
 848     /* When one connection closes, give accept() a chance, if need be */
 849     Comm::AcceptLimiter::Instance().kick();
 850 }
 851
 852 /*
 853  * Close the socket fd.
 854  *
 855  * + call write handlers with ERR_CLOSING
 856  * + call read handlers with ERR_CLOSING
 857  * + call closing handlers
 858  *
 859  * NOTE: Comm::ERR_CLOSING will NOT be called for CommReads' sitting in a
 860  * DeferredReadManager.
 861  */
 862 void
 863 _comm_close(int fd, char const *file, int line)
 864 {
 865     debugs(5, 3, "comm_close: start closing FD " << fd);
 866     assert(fd >= 0);
 867     assert(fd < Squid_MaxFD);
 868
 869     fde *F = &fd_table[fd];
 870     fdd_table[fd].close_file = file;
 871     fdd_table[fd].close_line = line;
 872
 873     if (F->closing())
 874         return;
 875
 876     /* XXX: is this obsolete behind F->closing() ? */
 877     if ( (shutting_down || reconfiguring) && (!F->flags.open || F->type == FD_FILE))
 878         return;
 879
 880     /* The following fails because ipc.c is doing calls to pipe() to create sockets! */
 881     if (!isOpen(fd)) {
 882         debugs(50, DBG_IMPORTANT, HERE << "BUG 3556: FD " << fd << " is not an open socket.");
 883         // XXX: do we need to run close(fd) or fd_close(fd) here?
 884         return;
 885     }
 886
 887     assert(F->type != FD_FILE);
 888
 889     PROF_start(comm_close);
 890
 891     F->flags.close_request = true;
 892
 893 #if USE_OPENSSL
 894     if (F->ssl) {
 895         AsyncCall::Pointer startCall=commCbCall(5,4, "commStartSslClose",
 896                                                 FdeCbPtrFun(commStartSslClose, NULL));
 897         FdeCbParams &startParams = GetCommParams<FdeCbParams>(startCall);
 898         startParams.fd = fd;
 899         ScheduleCallHere(startCall);
 900     }
 901 #endif
 902
 903     // a half-closed fd may lack a reader, so we stop monitoring explicitly
 904     if (commHasHalfClosedMonitor(fd))
 905         commStopHalfClosedMonitor(fd);
 906     commUnsetFdTimeout(fd);
 907
 908     // notify read/write handlers after canceling select reservations, if any
 909     if (COMMIO_FD_WRITECB(fd)->active()) {
 910         Comm::SetSelect(fd, COMM_SELECT_WRITE, NULL, NULL, 0);
 911         COMMIO_FD_WRITECB(fd)->finish(Comm::ERR_CLOSING, errno);
 912     }
 913     if (COMMIO_FD_READCB(fd)->active()) {
 914         Comm::SetSelect(fd, COMM_SELECT_READ, NULL, NULL, 0);
 915         COMMIO_FD_READCB(fd)->finish(Comm::ERR_CLOSING, errno);
 916     }
 917
 918 #if USE_DELAY_POOLS
 919     if (ClientInfo *clientInfo = F->clientInfo) {
 920         if (clientInfo->selectWaiting) {
 921             clientInfo->selectWaiting = false;
 922             // kick queue or it will get stuck as commWriteHandle is not called
 923             clientInfo->kickQuotaQueue();
 924         }
 925     }
 926 #endif
 927
 928     commCallCloseHandlers(fd);
 929
 930     comm_empty_os_read_buffers(fd);
 931
 932     AsyncCall::Pointer completeCall=commCbCall(5,4, "comm_close_complete",
 933                                     FdeCbPtrFun(comm_close_complete, NULL));
 934     FdeCbParams &completeParams = GetCommParams<FdeCbParams>(completeCall);
 935     completeParams.fd = fd;
 936     // must use async call to wait for all callbacks
 937     // scheduled before comm_close() to finish
 938     ScheduleCallHere(completeCall);
 939
 940     PROF_stop(comm_close);
 941 }
 942
 943 /* Send a udp datagram to specified TO_ADDR. */
 944 int
 945 comm_udp_sendto(int fd,
 946                 const Ip::Address &to_addr,
 947                 const void *buf,
 948                 int len)
 949 {
 950     PROF_start(comm_udp_sendto);
 951     ++ statCounter.syscalls.sock.sendtos;
 952
 953     debugs(50, 3, "comm_udp_sendto: Attempt to send UDP packet to " << to_addr <<
 954            " using FD " << fd << " using Port " << comm_local_port(fd) );
 955
 956     struct addrinfo *AI = NULL;
 957     to_addr.getAddrInfo(AI, fd_table[fd].sock_family);
 958     int x = sendto(fd, buf, len, 0, AI->ai_addr, AI->ai_addrlen);
 959     Ip::Address::FreeAddr(AI);
 960
 961     PROF_stop(comm_udp_sendto);
 962
 963     if (x >= 0)
 964         return x;
 965
 966 #if _SQUID_LINUX_
 967
 968     if (ECONNREFUSED != errno)
 969 #endif
 970
 971         debugs(50, DBG_IMPORTANT, "comm_udp_sendto: FD " << fd << ", (family=" << fd_table[fd].sock_family << ") " << to_addr << ": " << xstrerror());
 972
 973     return Comm::COMM_ERROR;
 974 }
 975
 976 void
 977 comm_add_close_handler(int fd, CLCB * handler, void *data)
 978 {
 979     debugs(5, 5, "comm_add_close_handler: FD " << fd << ", handler=" <<
 980            handler << ", data=" << data);
 981
 982     AsyncCall::Pointer call=commCbCall(5,4, "SomeCloseHandler",
 983                                        CommCloseCbPtrFun(handler, data));
 984     comm_add_close_handler(fd, call);
 985 }
 986
 987 void
 988 comm_add_close_handler(int fd, AsyncCall::Pointer &call)
 989 {
 990     debugs(5, 5, "comm_add_close_handler: FD " << fd << ", AsyncCall=" << call);
 991
 992     /*TODO:Check for a similar scheduled AsyncCall*/
 993 //    for (c = fd_table[fd].closeHandler; c; c = c->next)
 994 //        assert(c->handler != handler || c->data != data);
 995
 996     call->setNext(fd_table[fd].closeHandler);
 997
 998     fd_table[fd].closeHandler = call;
 999 }
1000
1001 // remove function-based close handler
1002 void
1003 comm_remove_close_handler(int fd, CLCB * handler, void *data)
1004 {
1005     assert(isOpen(fd));
1006     /* Find handler in list */
1007     debugs(5, 5, "comm_remove_close_handler: FD " << fd << ", handler=" <<
1008            handler << ", data=" << data);
1009
1010     AsyncCall::Pointer p, prev = NULL;
1011     for (p = fd_table[fd].closeHandler; p != NULL; prev = p, p = p->Next()) {
1012         typedef CommCbFunPtrCallT<CommCloseCbPtrFun> Call;
1013         const Call *call = dynamic_cast<const Call*>(p.getRaw());
1014         if (!call) // method callbacks have their own comm_remove_close_handler
1015             continue;
1016
1017         typedef CommCloseCbParams Params;
1018         const Params &params = GetCommParams<Params>(p);
1019         if (call->dialer.handler == handler && params.data == data)
1020             break;      /* This is our handler */
1021     }
1022
1023     // comm_close removes all close handlers so our handler may be gone
1024     if (p != NULL) {
1025         p->dequeue(fd_table[fd].closeHandler, prev);
1026         p->cancel("comm_remove_close_handler");
1027     }
1028 }
1029
1030 // remove method-based close handler
1031 void
1032 comm_remove_close_handler(int fd, AsyncCall::Pointer &call)
1033 {
1034     assert(isOpen(fd));
1035     debugs(5, 5, "comm_remove_close_handler: FD " << fd << ", AsyncCall=" << call);
1036
1037     // comm_close removes all close handlers so our handler may be gone
1038     AsyncCall::Pointer p, prev = NULL;
1039     for (p = fd_table[fd].closeHandler; p != NULL && p != call; prev = p, p = p->Next());
1040
1041     if (p != NULL)
1042         p->dequeue(fd_table[fd].closeHandler, prev);
1043     call->cancel("comm_remove_close_handler");
1044 }
1045
1046 static void
1047 commSetNoLinger(int fd)
1048 {
1049
1050     struct linger L;
1051     L.l_onoff = 0;      /* off */
1052     L.l_linger = 0;
1053
1054     if (setsockopt(fd, SOL_SOCKET, SO_LINGER, (char *) &L, sizeof(L)) < 0)
1055         debugs(50, 0, "commSetNoLinger: FD " << fd << ": " << xstrerror());
1056
1057     fd_table[fd].flags.nolinger = true;
1058 }
1059
1060 static void
1061 commSetReuseAddr(int fd)
1062 {
1063     int on = 1;
1064
1065     if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0)
1066         debugs(50, DBG_IMPORTANT, "commSetReuseAddr: FD " << fd << ": " << xstrerror());
1067 }
1068
1069 static void
1070 commSetTcpRcvbuf(int fd, int size)
1071 {
1072     if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size)) < 0)
1073         debugs(50, DBG_IMPORTANT, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
1074     if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char *) &size, sizeof(size)) < 0)
1075         debugs(50, DBG_IMPORTANT, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
1076 #ifdef TCP_WINDOW_CLAMP
1077     if (setsockopt(fd, SOL_TCP, TCP_WINDOW_CLAMP, (char *) &size, sizeof(size)) < 0)
1078         debugs(50, DBG_IMPORTANT, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
1079 #endif
1080 }
1081
1082 int
1083 commSetNonBlocking(int fd)
1084 {
1085 #if !_SQUID_WINDOWS_
1086     int flags;
1087     int dummy = 0;
1088 #endif
1089 #if _SQUID_WINDOWS_
1090     int nonblocking = TRUE;
1091
1092 #if _SQUID_CYGWIN_
1093     if (fd_table[fd].type != FD_PIPE) {
1094 #endif
1095
1096         if (ioctl(fd, FIONBIO, &nonblocking) < 0) {
1097             debugs(50, 0, "commSetNonBlocking: FD " << fd << ": " << xstrerror() << " " << fd_table[fd].type);
1098             return Comm::COMM_ERROR;
1099         }
1100
1101 #if _SQUID_CYGWIN_
1102     } else {
1103 #endif
1104 #endif
1105 #if !_SQUID_WINDOWS_
1106
1107         if ((flags = fcntl(fd, F_GETFL, dummy)) < 0) {
1108             debugs(50, 0, "FD " << fd << ": fcntl F_GETFL: " << xstrerror());
1109             return Comm::COMM_ERROR;
1110         }
1111
1112         if (fcntl(fd, F_SETFL, flags | SQUID_NONBLOCK) < 0) {
1113             debugs(50, 0, "commSetNonBlocking: FD " << fd << ": " << xstrerror());
1114             return Comm::COMM_ERROR;
1115         }
1116
1117 #endif
1118 #if _SQUID_CYGWIN_
1119     }
1120 #endif
1121     fd_table[fd].flags.nonblocking = true;
1122
1123     return 0;
1124 }
1125
1126 int
1127 commUnsetNonBlocking(int fd)
1128 {
1129 #if _SQUID_WINDOWS_
1130     int nonblocking = FALSE;
1131
1132     if (ioctlsocket(fd, FIONBIO, (unsigned long *) &nonblocking) < 0) {
1133 #else
1134     int flags;
1135     int dummy = 0;
1136
1137     if ((flags = fcntl(fd, F_GETFL, dummy)) < 0) {
1138         debugs(50, 0, "FD " << fd << ": fcntl F_GETFL: " << xstrerror());
1139         return Comm::COMM_ERROR;
1140     }
1141
1142     if (fcntl(fd, F_SETFL, flags & (~SQUID_NONBLOCK)) < 0) {
1143 #endif
1144         debugs(50, 0, "commUnsetNonBlocking: FD " << fd << ": " << xstrerror());
1145         return Comm::COMM_ERROR;
1146     }
1147
1148     fd_table[fd].flags.nonblocking = false;
1149     return 0;
1150 }
1151
1152 void
1153 commSetCloseOnExec(int fd)
1154 {
1155 #ifdef FD_CLOEXEC
1156     int flags;
1157     int dummy = 0;
1158
1159     if ((flags = fcntl(fd, F_GETFD, dummy)) < 0) {
1160         debugs(50, 0, "FD " << fd << ": fcntl F_GETFD: " << xstrerror());
1161         return;
1162     }
1163
1164     if (fcntl(fd, F_SETFD, flags | FD_CLOEXEC) < 0)
1165         debugs(50, 0, "FD " << fd << ": set close-on-exec failed: " << xstrerror());
1166
1167     fd_table[fd].flags.close_on_exec = true;
1168
1169 #endif
1170 }
1171
1172 #ifdef TCP_NODELAY
1173 static void
1174 commSetTcpNoDelay(int fd)
1175 {
1176     int on = 1;
1177
1178     if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char *) &on, sizeof(on)) < 0)
1179         debugs(50, DBG_IMPORTANT, "commSetTcpNoDelay: FD " << fd << ": " << xstrerror());
1180
1181     fd_table[fd].flags.nodelay = true;
1182 }
1183
1184 #endif
1185
1186 void
1187 commSetTcpKeepalive(int fd, int idle, int interval, int timeout)
1188 {
1189     int on = 1;
1190 #ifdef TCP_KEEPCNT
1191     if (timeout && interval) {
1192         int count = (timeout + interval - 1) / interval;
1193         if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &count, sizeof(on)) < 0)
1194             debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
1195     }
1196 #endif
1197 #ifdef TCP_KEEPIDLE
1198     if (idle) {
1199         if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(on)) < 0)
1200             debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
1201     }
1202 #endif
1203 #ifdef TCP_KEEPINTVL
1204     if (interval) {
1205         if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(on)) < 0)
1206             debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
1207     }
1208 #endif
1209     if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (char *) &on, sizeof(on)) < 0)
1210         debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
1211 }
1212
1213 void
1214 comm_init(void)
1215 {
1216     fd_table =(fde *) xcalloc(Squid_MaxFD, sizeof(fde));
1217     fdd_table = (fd_debug_t *)xcalloc(Squid_MaxFD, sizeof(fd_debug_t));
1218
1219     /* make sure the accept() socket FIFO delay queue exists */
1220     Comm::AcceptLimiter::Instance();
1221
1222     // make sure the IO pending callback table exists
1223     Comm::CallbackTableInit();
1224
1225     /* XXX account fd_table */
1226     /* Keep a few file descriptors free so that we don't run out of FD's
1227      * after accepting a client but before it opens a socket or a file.
1228      * Since Squid_MaxFD can be as high as several thousand, don't waste them */
1229     RESERVED_FD = min(100, Squid_MaxFD / 4);
1230
1231     TheHalfClosed = new DescriptorSet;
1232
1233     /* setup the select loop module */
1234     Comm::SelectLoopInit();
1235 }
1236
1237 void
1238 comm_exit(void)
1239 {
1240     delete TheHalfClosed;
1241     TheHalfClosed = NULL;
1242
1243     safe_free(fd_table);
1244     safe_free(fdd_table);
1245     Comm::CallbackTableDestruct();
1246 }
1247
1248 #if USE_DELAY_POOLS
1249 // called when the queue is done waiting for the client bucket to fill
1250 void
1251 commHandleWriteHelper(void * data)
1252 {
1253     CommQuotaQueue *queue = static_cast<CommQuotaQueue*>(data);
1254     assert(queue);
1255
1256     ClientInfo *clientInfo = queue->clientInfo;
1257     // ClientInfo invalidates queue if freed, so if we got here through,
1258     // evenAdd cbdata protections, everything should be valid and consistent
1259     assert(clientInfo);
1260     assert(clientInfo->hasQueue());
1261     assert(clientInfo->hasQueue(queue));
1262     assert(!clientInfo->selectWaiting);
1263     assert(clientInfo->eventWaiting);
1264     clientInfo->eventWaiting = false;
1265
1266     do {
1267         // check that the head descriptor is still relevant
1268         const int head = clientInfo->quotaPeekFd();
1269         Comm::IoCallback *ccb = COMMIO_FD_WRITECB(head);
1270
1271         if (fd_table[head].clientInfo == clientInfo &&
1272                 clientInfo->quotaPeekReserv() == ccb->quotaQueueReserv &&
1273                 !fd_table[head].closing()) {
1274
1275             // wait for the head descriptor to become ready for writing
1276             Comm::SetSelect(head, COMM_SELECT_WRITE, Comm::HandleWrite, ccb, 0);
1277             clientInfo->selectWaiting = true;
1278             return;
1279         }
1280
1281         clientInfo->quotaDequeue(); // remove the no longer relevant descriptor
1282         // and continue looking for a relevant one
1283     } while (clientInfo->hasQueue());
1284
1285     debugs(77,3, HERE << "emptied queue");
1286 }
1287
1288 bool
1289 ClientInfo::hasQueue() const
1290 {
1291     assert(quotaQueue);
1292     return !quotaQueue->empty();
1293 }
1294
1295 bool
1296 ClientInfo::hasQueue(const CommQuotaQueue *q) const
1297 {
1298     assert(quotaQueue);
1299     return quotaQueue == q;
1300 }
1301
1302 /// returns the first descriptor to be dequeued
1303 int
1304 ClientInfo::quotaPeekFd() const
1305 {
1306     assert(quotaQueue);
1307     return quotaQueue->front();
1308 }
1309
1310 /// returns the reservation ID of the first descriptor to be dequeued
1311 unsigned int
1312 ClientInfo::quotaPeekReserv() const
1313 {
1314     assert(quotaQueue);
1315     return quotaQueue->outs + 1;
1316 }
1317
1318 /// queues a given fd, creating the queue if necessary; returns reservation ID
1319 unsigned int
1320 ClientInfo::quotaEnqueue(int fd)
1321 {
1322     assert(quotaQueue);
1323     return quotaQueue->enqueue(fd);
1324 }
1325
1326 /// removes queue head
1327 void
1328 ClientInfo::quotaDequeue()
1329 {
1330     assert(quotaQueue);
1331     quotaQueue->dequeue();
1332 }
1333
1334 void
1335 ClientInfo::kickQuotaQueue()
1336 {
1337     if (!eventWaiting && !selectWaiting && hasQueue()) {
1338         // wait at least a second if the bucket is empty
1339         const double delay = (bucketSize < 1.0) ? 1.0 : 0.0;
1340         eventAdd("commHandleWriteHelper", &commHandleWriteHelper,
1341                  quotaQueue, delay, 0, true);
1342         eventWaiting = true;
1343     }
1344 }
1345
1346 /// calculates how much to write for a single dequeued client
1347 int
1348 ClientInfo::quotaForDequed()
1349 {
1350     /* If we have multiple clients and give full bucketSize to each client then
1351      * clt1 may often get a lot more because clt1->clt2 time distance in the
1352      * select(2) callback order may be a lot smaller than cltN->clt1 distance.
1353      * We divide quota evenly to be more fair. */
1354
1355     if (!rationedCount) {
1356         rationedCount = quotaQueue->size() + 1;
1357
1358         // The delay in ration recalculation _temporary_ deprives clients from
1359         // bytes that should have trickled in while rationedCount was positive.
1360         refillBucket();
1361
1362         // Rounding errors do not accumulate here, but we round down to avoid
1363         // negative bucket sizes after write with rationedCount=1.
1364         rationedQuota = static_cast<int>(floor(bucketSize/rationedCount));
1365         debugs(77,5, HERE << "new rationedQuota: " << rationedQuota <<
1366                '*' << rationedCount);
1367     }
1368
1369     --rationedCount;
1370     debugs(77,7, HERE << "rationedQuota: " << rationedQuota <<
1371            " rations remaining: " << rationedCount);
1372
1373     // update 'last seen' time to prevent clientdb GC from dropping us
1374     last_seen = squid_curtime;
1375     return rationedQuota;
1376 }
1377
1378 ///< adds bytes to the quota bucket based on the rate and passed time
1379 void
1380 ClientInfo::refillBucket()
1381 {
1382     // all these times are in seconds, with double precision
1383     const double currTime = current_dtime;
1384     const double timePassed = currTime - prevTime;
1385
1386     // Calculate allowance for the time passed. Use double to avoid
1387     // accumulating rounding errors for small intervals. For example, always
1388     // adding 1 byte instead of 1.4 results in 29% bandwidth allocation error.
1389     const double gain = timePassed * writeSpeedLimit;
1390
1391     debugs(77,5, HERE << currTime << " clt" << (const char*)hash.key << ": " <<
1392            bucketSize << " + (" << timePassed << " * " << writeSpeedLimit <<
1393            " = " << gain << ')');
1394
1395     // to further combat error accumulation during micro updates,
1396     // quit before updating time if we cannot add at least one byte
1397     if (gain < 1.0)
1398         return;
1399
1400     prevTime = currTime;
1401
1402     // for "first" connections, drain initial fat before refilling but keep
1403     // updating prevTime to avoid bursts after the fat is gone
1404     if (bucketSize > bucketSizeLimit) {
1405         debugs(77,4, HERE << "not refilling while draining initial fat");
1406         return;
1407     }
1408
1409     bucketSize += gain;
1410
1411     // obey quota limits
1412     if (bucketSize > bucketSizeLimit)
1413         bucketSize = bucketSizeLimit;
1414 }
1415
1416 void
1417 ClientInfo::setWriteLimiter(const int aWriteSpeedLimit, const double anInitialBurst, const double aHighWatermark)
1418 {
1419     debugs(77,5, HERE << "Write limits for " << (const char*)hash.key <<
1420            " speed=" << aWriteSpeedLimit << " burst=" << anInitialBurst <<
1421            " highwatermark=" << aHighWatermark);
1422
1423     // set or possibly update traffic shaping parameters
1424     writeLimitingActive = true;
1425     writeSpeedLimit = aWriteSpeedLimit;
1426     bucketSizeLimit = aHighWatermark;
1427
1428     // but some members should only be set once for a newly activated bucket
1429     if (firstTimeConnection) {
1430         firstTimeConnection = false;
1431
1432         assert(!selectWaiting);
1433         assert(!quotaQueue);
1434         quotaQueue = new CommQuotaQueue(this);
1435
1436         bucketSize = anInitialBurst;
1437         prevTime = current_dtime;
1438     }
1439 }
1440
1441 CommQuotaQueue::CommQuotaQueue(ClientInfo *info): clientInfo(info),
1442     ins(0), outs(0)
1443 {
1444     assert(clientInfo);
1445 }
1446
1447 CommQuotaQueue::~CommQuotaQueue()
1448 {
1449     assert(!clientInfo); // ClientInfo should clear this before destroying us
1450 }
1451
1452 /// places the given fd at the end of the queue; returns reservation ID
1453 unsigned int
1454 CommQuotaQueue::enqueue(int fd)
1455 {
1456     debugs(77,5, HERE << "clt" << (const char*)clientInfo->hash.key <<
1457            ": FD " << fd << " with qqid" << (ins+1) << ' ' << fds.size());
1458     fds.push_back(fd);
1459     return ++ins;
1460 }
1461
1462 /// removes queue head
1463 void
1464 CommQuotaQueue::dequeue()
1465 {
1466     assert(!fds.empty());
1467     debugs(77,5, HERE << "clt" << (const char*)clientInfo->hash.key <<
1468            ": FD " << fds.front() << " with qqid" << (outs+1) << ' ' <<
1469            fds.size());
1470     fds.pop_front();
1471     ++outs;
1472 }
1473 #endif
1474
1475 /*
1476  * hm, this might be too general-purpose for all the places we'd
1477  * like to use it.
1478  */
1479 int
1480 ignoreErrno(int ierrno)
1481 {
1482     switch (ierrno) {
1483
1484     case EINPROGRESS:
1485
1486     case EWOULDBLOCK:
1487 #if EAGAIN != EWOULDBLOCK
1488
1489     case EAGAIN:
1490 #endif
1491
1492     case EALREADY:
1493
1494     case EINTR:
1495 #ifdef ERESTART
1496
1497     case ERESTART:
1498 #endif
1499
1500         return 1;
1501
1502     default:
1503         return 0;
1504     }
1505
1506     /* NOTREACHED */
1507 }
1508
1509 void
1510 commCloseAllSockets(void)
1511 {
1512     int fd;
1513     fde *F = NULL;
1514
1515     for (fd = 0; fd <= Biggest_FD; ++fd) {
1516         F = &fd_table[fd];
1517
1518         if (!F->flags.open)
1519             continue;
1520
1521         if (F->type != FD_SOCKET)
1522             continue;
1523
1524         if (F->flags.ipc)   /* don't close inter-process sockets */
1525             continue;
1526
1527         if (F->timeoutHandler != NULL) {
1528             AsyncCall::Pointer callback = F->timeoutHandler;
1529             F->timeoutHandler = NULL;
1530             debugs(5, 5, "commCloseAllSockets: FD " << fd << ": Calling timeout handler");
1531             ScheduleCallHere(callback);
1532         } else {
1533             debugs(5, 5, "commCloseAllSockets: FD " << fd << ": calling comm_reset_close()");
1534             old_comm_reset_close(fd);
1535         }
1536     }
1537 }
1538
1539 static bool
1540 AlreadyTimedOut(fde *F)
1541 {
1542     if (!F->flags.open)
1543         return true;
1544
1545     if (F->timeout == 0)
1546         return true;
1547
1548     if (F->timeout > squid_curtime)
1549         return true;
1550
1551     return false;
1552 }
1553
1554 static bool
1555 writeTimedOut(int fd)
1556 {
1557     if (!COMMIO_FD_WRITECB(fd)->active())
1558         return false;
1559
1560     if ((squid_curtime - fd_table[fd].writeStart) < Config.Timeout.write)
1561         return false;
1562
1563     return true;
1564 }
1565
1566 void
1567 checkTimeouts(void)
1568 {
1569     int fd;
1570     fde *F = NULL;
1571     AsyncCall::Pointer callback;
1572
1573     for (fd = 0; fd <= Biggest_FD; ++fd) {
1574         F = &fd_table[fd];
1575
1576         if (writeTimedOut(fd)) {
1577             // We have an active write callback and we are timed out
1578             debugs(5, 5, "checkTimeouts: FD " << fd << " auto write timeout");
1579             Comm::SetSelect(fd, COMM_SELECT_WRITE, NULL, NULL, 0);
1580             COMMIO_FD_WRITECB(fd)->finish(Comm::COMM_ERROR, ETIMEDOUT);
1581         } else if (AlreadyTimedOut(F))
1582             continue;
1583
1584         debugs(5, 5, "checkTimeouts: FD " << fd << " Expired");
1585
1586         if (F->timeoutHandler != NULL) {
1587             debugs(5, 5, "checkTimeouts: FD " << fd << ": Call timeout handler");
1588             callback = F->timeoutHandler;
1589             F->timeoutHandler = NULL;
1590             ScheduleCallHere(callback);
1591         } else {
1592             debugs(5, 5, "checkTimeouts: FD " << fd << ": Forcing comm_close()");
1593             comm_close(fd);
1594         }
1595     }
1596 }
1597
1598 /// Start waiting for a possibly half-closed connection to close
1599 // by scheduling a read callback to a monitoring handler that
1600 // will close the connection on read errors.
1601 void
1602 commStartHalfClosedMonitor(int fd)
1603 {
1604     debugs(5, 5, HERE << "adding FD " << fd << " to " << *TheHalfClosed);
1605     assert(isOpen(fd) && !commHasHalfClosedMonitor(fd));
1606     (void)TheHalfClosed->add(fd); // could also assert the result
1607     commPlanHalfClosedCheck(); // may schedule check if we added the first FD
1608 }
1609
1610 static
1611 void
1612 commPlanHalfClosedCheck()
1613 {
1614     if (!WillCheckHalfClosed && !TheHalfClosed->empty()) {
1615         eventAdd("commHalfClosedCheck", &commHalfClosedCheck, NULL, 1.0, 1);
1616         WillCheckHalfClosed = true;
1617     }
1618 }
1619
1620 /// iterates over all descriptors that may need half-closed tests and
1621 /// calls comm_read for those that do; re-schedules the check if needed
1622 static
1623 void
1624 commHalfClosedCheck(void *)
1625 {
1626     debugs(5, 5, HERE << "checking " << *TheHalfClosed);
1627
1628     typedef DescriptorSet::const_iterator DSCI;
1629     const DSCI end = TheHalfClosed->end();
1630     for (DSCI i = TheHalfClosed->begin(); i != end; ++i) {
1631         Comm::ConnectionPointer c = new Comm::Connection; // XXX: temporary. make HalfClosed a list of these.
1632         c->fd = *i;
1633         if (!fd_table[c->fd].halfClosedReader) { // not reading already
1634             AsyncCall::Pointer call = commCbCall(5,4, "commHalfClosedReader",
1635                                                  CommIoCbPtrFun(&commHalfClosedReader, NULL));
1636             Comm::Read(c, call);
1637             fd_table[c->fd].halfClosedReader = call;
1638         } else
1639             c->fd = -1; // XXX: temporary. prevent c replacement erase closing listed FD
1640     }
1641
1642     WillCheckHalfClosed = false; // as far as we know
1643     commPlanHalfClosedCheck(); // may need to check again
1644 }
1645
1646 /// checks whether we are waiting for possibly half-closed connection to close
1647 // We are monitoring if the read handler for the fd is the monitoring handler.
1648 bool
1649 commHasHalfClosedMonitor(int fd)
1650 {
1651     return TheHalfClosed->has(fd);
1652 }
1653
1654 /// stop waiting for possibly half-closed connection to close
1655 void
1656 commStopHalfClosedMonitor(int const fd)
1657 {
1658     debugs(5, 5, HERE << "removing FD " << fd << " from " << *TheHalfClosed);
1659
1660     // cancel the read if one was scheduled
1661     AsyncCall::Pointer reader = fd_table[fd].halfClosedReader;
1662     if (reader != NULL)
1663         Comm::ReadCancel(fd, reader);
1664     fd_table[fd].halfClosedReader = NULL;
1665
1666     TheHalfClosed->del(fd);
1667 }
1668
1669 /// I/O handler for the possibly half-closed connection monitoring code
1670 static void
1671 commHalfClosedReader(const Comm::ConnectionPointer &conn, char *, size_t size, Comm::Flag flag, int, void *)
1672 {
1673     // there cannot be more data coming in on half-closed connections
1674     assert(size == 0);
1675     assert(conn != NULL);
1676     assert(commHasHalfClosedMonitor(conn->fd)); // or we would have canceled the read
1677
1678     fd_table[conn->fd].halfClosedReader = NULL; // done reading, for now
1679
1680     // nothing to do if fd is being closed
1681     if (flag == Comm::ERR_CLOSING)
1682         return;
1683
1684     // if read failed, close the connection
1685     if (flag != Comm::OK) {
1686         debugs(5, 3, HERE << "closing " << conn);
1687         conn->close();
1688         return;
1689     }
1690
1691     // continue waiting for close or error
1692     commPlanHalfClosedCheck(); // make sure this fd will be checked again
1693 }
1694
1695 CommRead::CommRead() : conn(NULL), buf(NULL), len(0), callback(NULL) {}
1696
1697 CommRead::CommRead(const Comm::ConnectionPointer &c, char *buf_, int len_, AsyncCall::Pointer &callback_)
1698     : conn(c), buf(buf_), len(len_), callback(callback_) {}
1699
1700 DeferredRead::DeferredRead () : theReader(NULL), theContext(NULL), theRead(), cancelled(false) {}
1701
1702 DeferredRead::DeferredRead (DeferrableRead *aReader, void *data, CommRead const &aRead) : theReader(aReader), theContext (data), theRead(aRead), cancelled(false) {}
1703
1704 DeferredReadManager::~DeferredReadManager()
1705 {
1706     flushReads();
1707     assert (deferredReads.empty());
1708 }
1709
1710 /* explicit instantiation required for some systems */
1711
1712 /// \cond AUTODOCS_IGNORE
1713 template cbdata_type CbDataList<DeferredRead>::CBDATA_CbDataList;
1714 /// \endcond
1715
1716 void
1717 DeferredReadManager::delayRead(DeferredRead const &aRead)
1718 {
1719     debugs(5, 3, "Adding deferred read on " << aRead.theRead.conn);
1720     CbDataList<DeferredRead> *temp = deferredReads.push_back(aRead);
1721
1722     // We have to use a global function as a closer and point to temp
1723     // instead of "this" because DeferredReadManager is not a job and
1724     // is not even cbdata protected
1725     // XXX: and yet we use cbdata protection functions on it??
1726     AsyncCall::Pointer closer = commCbCall(5,4,
1727                                            "DeferredReadManager::CloseHandler",
1728                                            CommCloseCbPtrFun(&CloseHandler, temp));
1729     comm_add_close_handler(aRead.theRead.conn->fd, closer);
1730     temp->element.closer = closer; // remeber so that we can cancel
1731 }
1732
1733 void
1734 DeferredReadManager::CloseHandler(const CommCloseCbParams &params)
1735 {
1736     if (!cbdataReferenceValid(params.data))
1737         return;
1738
1739     CbDataList<DeferredRead> *temp = (CbDataList<DeferredRead> *)params.data;
1740
1741     temp->element.closer = NULL;
1742     temp->element.markCancelled();
1743 }
1744
1745 DeferredRead
1746 DeferredReadManager::popHead(CbDataListContainer<DeferredRead> &deferredReads)
1747 {
1748     assert (!deferredReads.empty());
1749
1750     DeferredRead &read = deferredReads.head->element;
1751
1752     // NOTE: at this point the connection has been paused/stalled for an unknown
1753     //       amount of time. We must re-validate that it is active and usable.
1754
1755     // If the connection has been closed already. Cancel this read.
1756     if (!Comm::IsConnOpen(read.theRead.conn)) {
1757         if (read.closer != NULL) {
1758             read.closer->cancel("Connection closed before.");
1759             read.closer = NULL;
1760         }
1761         read.markCancelled();
1762     }
1763
1764     if (!read.cancelled) {
1765         comm_remove_close_handler(read.theRead.conn->fd, read.closer);
1766         read.closer = NULL;
1767     }
1768
1769     DeferredRead result = deferredReads.pop_front();
1770
1771     return result;
1772 }
1773
1774 void
1775 DeferredReadManager::kickReads(int const count)
1776 {
1777     /* if we had CbDataList::size() we could consolidate this and flushReads */
1778
1779     if (count < 1) {
1780         flushReads();
1781         return;
1782     }
1783
1784     size_t remaining = count;
1785
1786     while (!deferredReads.empty() && remaining) {
1787         DeferredRead aRead = popHead(deferredReads);
1788         kickARead(aRead);
1789
1790         if (!aRead.cancelled)
1791             --remaining;
1792     }
1793 }
1794
1795 void
1796 DeferredReadManager::flushReads()
1797 {
1798     CbDataListContainer<DeferredRead> reads;
1799     reads = deferredReads;
1800     deferredReads = CbDataListContainer<DeferredRead>();
1801
1802     // XXX: For fairness this SHOULD randomize the order
1803     while (!reads.empty()) {
1804         DeferredRead aRead = popHead(reads);
1805         kickARead(aRead);
1806     }
1807 }
1808
1809 void
1810 DeferredReadManager::kickARead(DeferredRead const &aRead)
1811 {
1812     if (aRead.cancelled)
1813         return;
1814
1815     if (Comm::IsConnOpen(aRead.theRead.conn) && fd_table[aRead.theRead.conn->fd].closing())
1816         return;
1817
1818     debugs(5, 3, "Kicking deferred read on " << aRead.theRead.conn);
1819
1820     aRead.theReader(aRead.theContext, aRead.theRead);
1821 }
1822
1823 void
1824 DeferredRead::markCancelled()
1825 {
1826     cancelled = true;
1827 }
1828
1829 int
1830 CommSelectEngine::checkEvents(int timeout)
1831 {
1832     static time_t last_timeout = 0;
1833
1834     /* No, this shouldn't be here. But it shouldn't be in each comm handler. -adrian */
1835     if (squid_curtime > last_timeout) {
1836         last_timeout = squid_curtime;
1837         checkTimeouts();
1838     }
1839
1840     switch (Comm::DoSelect(timeout)) {
1841
1842     case Comm::OK:
1843
1844     case Comm::TIMEOUT:
1845         return 0;
1846
1847     case Comm::IDLE:
1848
1849     case Comm::SHUTDOWN:
1850         return EVENT_IDLE;
1851
1852     case Comm::COMM_ERROR:
1853         return EVENT_ERROR;
1854
1855     default:
1856         fatal_dump("comm.cc: Internal error -- this should never happen.");
1857         return EVENT_ERROR;
1858     };
1859 }
1860
1861 /// Create a unix-domain socket (UDS) that only supports FD_MSGHDR I/O.
1862 int
1863 comm_open_uds(int sock_type,
1864               int proto,
1865               struct sockaddr_un* addr,
1866               int flags)
1867 {
1868     // TODO: merge with comm_openex() when Ip::Address becomes NetAddress
1869
1870     int new_socket;
1871
1872     PROF_start(comm_open);
1873     /* Create socket for accepting new connections. */
1874     ++ statCounter.syscalls.sock.sockets;
1875
1876     /* Setup the socket addrinfo details for use */
1877     struct addrinfo AI;
1878     AI.ai_flags = 0;
1879     AI.ai_family = PF_UNIX;
1880     AI.ai_socktype = sock_type;
1881     AI.ai_protocol = proto;
1882     AI.ai_addrlen = SUN_LEN(addr);
1883     AI.ai_addr = (sockaddr*)addr;
1884     AI.ai_canonname = NULL;
1885     AI.ai_next = NULL;
1886
1887     debugs(50, 3, HERE << "Attempt open socket for: " << addr->sun_path);
1888
1889     if ((new_socket = socket(AI.ai_family, AI.ai_socktype, AI.ai_protocol)) < 0) {
1890         /* Increase the number of reserved fd's if calls to socket()
1891          * are failing because the open file table is full.  This
1892          * limits the number of simultaneous clients */
1893
1894         if (limitError(errno)) {
1895             debugs(50, DBG_IMPORTANT, HERE << "socket failure: " << xstrerror());
1896             fdAdjustReserved();
1897         } else {
1898             debugs(50, DBG_CRITICAL, HERE << "socket failure: " << xstrerror());
1899         }
1900
1901         PROF_stop(comm_open);
1902         return -1;
1903     }
1904
1905     debugs(50, 3, "Opened UDS FD " << new_socket << " : family=" << AI.ai_family << ", type=" << AI.ai_socktype << ", protocol=" << AI.ai_protocol);
1906
1907     /* update fdstat */
1908     debugs(50, 5, HERE << "FD " << new_socket << " is a new socket");
1909
1910     assert(!isOpen(new_socket));
1911     fd_open(new_socket, FD_MSGHDR, NULL);
1912
1913     fdd_table[new_socket].close_file = NULL;
1914
1915     fdd_table[new_socket].close_line = 0;
1916
1917     fd_table[new_socket].sock_family = AI.ai_family;
1918
1919     if (!(flags & COMM_NOCLOEXEC))
1920         commSetCloseOnExec(new_socket);
1921
1922     if (flags & COMM_REUSEADDR)
1923         commSetReuseAddr(new_socket);
1924
1925     if (flags & COMM_NONBLOCKING) {
1926         if (commSetNonBlocking(new_socket) != Comm::OK) {
1927             comm_close(new_socket);
1928             PROF_stop(comm_open);
1929             return -1;
1930         }
1931     }
1932
1933     if (flags & COMM_DOBIND) {
1934         if (commBind(new_socket, AI) != Comm::OK) {
1935             comm_close(new_socket);
1936             PROF_stop(comm_open);
1937             return -1;
1938         }
1939     }
1940
1941 #ifdef TCP_NODELAY
1942     if (sock_type == SOCK_STREAM)
1943         commSetTcpNoDelay(new_socket);
1944
1945 #endif
1946
1947     if (Config.tcpRcvBufsz > 0 && sock_type == SOCK_STREAM)
1948         commSetTcpRcvbuf(new_socket, Config.tcpRcvBufsz);
1949
1950     PROF_stop(comm_open);
1951
1952     return new_socket;
1953 }
1954