]> git.ipfire.org Git - thirdparty/squid.git/blobdiff - src/comm.cc
SourceFormat Enforcement
[thirdparty/squid.git] / src / comm.cc
index b92dc8e92f1808174cafc85d1594641af30d124d..b2bbad35c3a6ccf84a1b688191e3993d846083f9 100644 (file)
  */
 
 #include "squid.h"
-#include "StoreIOBuffer.h"
+#include "base/AsyncCall.h"
 #include "comm.h"
-#include "event.h"
-#include "fde.h"
+#include "ClientInfo.h"
+#include "CommCalls.h"
 #include "comm/AcceptLimiter.h"
 #include "comm/comm_internal.h"
+#include "comm/Connection.h"
 #include "comm/IoCallback.h"
 #include "comm/Loops.h"
 #include "comm/Write.h"
 #include "comm/TcpAcceptor.h"
-#include "CommIO.h"
 #include "CommRead.h"
-#include "ConnectionDetail.h"
-#include "MemBuf.h"
-#include "pconn.h"
-#include "SquidTime.h"
-#include "CommCalls.h"
+#include "compat/cmsg.h"
 #include "DescriptorSet.h"
+#include "event.h"
+#include "fde.h"
+#include "globals.h"
 #include "icmp/net_db.h"
 #include "ip/Address.h"
 #include "ip/Intercept.h"
 #include "ip/QosConfig.h"
 #include "ip/tools.h"
-#include "ClientInfo.h"
+#include "MemBuf.h"
+#include "pconn.h"
+#include "protos.h"
+#include "profiler/Profiler.h"
+#include "SquidTime.h"
+#include "StatCounters.h"
+#include "StoreIOBuffer.h"
 #if USE_SSL
 #include "ssl/support.h"
 #endif
 #ifdef HAVE_NETINET_TCP_H
 #include <netinet/tcp.h>
 #endif
+#if HAVE_SYS_UN_H
+#include <sys/un.h>
+#endif
+#if HAVE_MATH_H
+#include <math.h>
+#endif
+#if HAVE_ERRNO_H
+#include <errno.h>
+#endif
 
 /*
  * New C-like simple comm code. This stuff is a mess and doesn't really buy us anything.
@@ -75,7 +89,7 @@
 
 static void commStopHalfClosedMonitor(int fd);
 static IOCB commHalfClosedReader;
-static void comm_init_opened(int new_socket, Ip::Address &addr, tos_t tos, nfmark_t nfmark, const char *note, struct addrinfo *AI);
+static void comm_init_opened(const Comm::ConnectionPointer &conn, tos_t tos, nfmark_t nfmark, const char *note, struct addrinfo *AI);
 static int comm_apply_flags(int new_socket, Ip::Address &addr, int flags, struct addrinfo *AI);
 
 #if USE_DELAY_POOLS
@@ -84,38 +98,6 @@ CBDATA_CLASS_INIT(CommQuotaQueue);
 static void commHandleWriteHelper(void * data);
 #endif
 
-class ConnectStateData
-{
-
-public:
-    void *operator new (size_t);
-    void operator delete (void *);
-    static void Connect (int fd, void *me);
-    void connect();
-    void callCallback(comm_err_t status, int xerrno);
-    void defaults();
-
-// defaults given by client
-    char *host;
-    u_short default_port;
-    Ip::Address default_addr;
-    // NP: CANNOT store the default addr:port together as it gets set/reset differently.
-
-    DnsLookupDetails dns; ///< host lookup details
-    Ip::Address S;
-    AsyncCall::Pointer callback;
-
-    int fd;
-    int tries;
-    int addrcount;
-    int connstart;
-
-private:
-    int commResetFD();
-    int commRetryConnect();
-    CBDATA_CLASS(ConnectStateData);
-};
-
 /* STATIC */
 
 static DescriptorSet *TheHalfClosed = NULL; /// the set of half-closed FDs
@@ -130,13 +112,6 @@ static void commSetNoLinger(int);
 static void commSetTcpNoDelay(int);
 #endif
 static void commSetTcpRcvbuf(int, int);
-static PF commConnectFree;
-static IPH commConnectDnsHandle;
-
-typedef enum {
-    COMM_CB_READ = 1,
-    COMM_CB_DERIVED
-} comm_callback_t;
 
 static MemAllocator *conn_close_pool = NULL;
 fd_debug_t *fdd_table = NULL;
@@ -144,7 +119,7 @@ fd_debug_t *fdd_table = NULL;
 bool
 isOpen(const int fd)
 {
-    return fd >= 0 && fd_table[fd].flags.open != 0;
+    return fd >= 0 && fd_table && fd_table[fd].flags.open != 0;
 }
 
 /**
@@ -161,7 +136,7 @@ commHandleRead(int fd, void *data)
     assert(data == COMMIO_FD_READCB(fd));
     assert(ccb->active());
     /* Attempt a read */
-    statCounter.syscalls.sock.reads++;
+    ++ statCounter.syscalls.sock.reads;
     errno = 0;
     int retval;
     retval = FD_READ_METHOD(fd, ccb->buf, ccb->size);
@@ -192,35 +167,28 @@ commHandleRead(int fd, void *data)
  * completes, on error, or on file descriptor close.
  */
 void
-comm_read(int fd, char *buf, int size, IOCB *handler, void *handler_data)
+comm_read(const Comm::ConnectionPointer &conn, char *buf, int size, AsyncCall::Pointer &callback)
 {
-    AsyncCall::Pointer call = commCbCall(5,4, "SomeCommReadHandler",
-                                         CommIoCbPtrFun(handler, handler_data));
-    comm_read(fd, buf, size, call);
-}
-
-void
-comm_read(int fd, char *buf, int size, AsyncCall::Pointer &callback)
-{
-    debugs(5, 5, "comm_read, queueing read for FD " << fd << "; asynCall " << callback);
+    debugs(5, 5, "comm_read, queueing read for " << conn << "; asynCall " << callback);
 
     /* Make sure we are open and not closing */
-    assert(isOpen(fd));
-    assert(!fd_table[fd].closing());
-    Comm::IoCallback *ccb = COMMIO_FD_READCB(fd);
+    assert(Comm::IsConnOpen(conn));
+    assert(!fd_table[conn->fd].closing());
+    Comm::IoCallback *ccb = COMMIO_FD_READCB(conn->fd);
 
     // Make sure we are either not reading or just passively monitoring.
     // Active/passive conflicts are OK and simply cancel passive monitoring.
     if (ccb->active()) {
         // if the assertion below fails, we have an active comm_read conflict
-        assert(fd_table[fd].halfClosedReader != NULL);
-        commStopHalfClosedMonitor(fd);
+        assert(fd_table[conn->fd].halfClosedReader != NULL);
+        commStopHalfClosedMonitor(conn->fd);
         assert(!ccb->active());
     }
+    ccb->conn = conn;
 
     /* Queue the read */
     ccb->setCallback(Comm::IOCB_READ, callback, (char *)buf, NULL, size);
-    Comm::SetSelect(fd, COMM_SELECT_READ, commHandleRead, ccb, 0);
+    Comm::SetSelect(conn->fd, COMM_SELECT_READ, commHandleRead, ccb, 0);
 }
 
 /**
@@ -234,7 +202,7 @@ comm_read(int fd, char *buf, int size, AsyncCall::Pointer &callback)
 static void
 comm_empty_os_read_buffers(int fd)
 {
-#ifdef _SQUID_LINUX_
+#if _SQUID_LINUX_
     /* prevent those nasty RST packets */
     char buf[SQUID_TCP_SO_RCVBUF];
 
@@ -244,9 +212,9 @@ comm_empty_os_read_buffers(int fd)
 #endif
 }
 
-
 /**
  * Return whether the FD has a pending completed callback.
+ * NP: does not work.
  */
 int
 comm_has_pending_read_callback(int fd)
@@ -353,14 +321,13 @@ comm_read_cancel(int fd, AsyncCall::Pointer &callback)
     Comm::SetSelect(fd, COMM_SELECT_READ, NULL, NULL, 0);
 }
 
-
 /**
  * synchronous wrapper around udp socket functions
  */
 int
 comm_udp_recvfrom(int fd, void *buf, size_t len, int flags, Ip::Address &from)
 {
-    statCounter.syscalls.sock.recvfroms++;
+    ++ statCounter.syscalls.sock.recvfroms;
     int x = 0;
     struct addrinfo *AI = NULL;
 
@@ -392,7 +359,6 @@ comm_udp_send(int s, const void *buf, size_t len, int flags)
     return send(s, buf, len, flags);
 }
 
-
 bool
 comm_has_incomplete_write(int fd)
 {
@@ -406,7 +372,7 @@ comm_has_incomplete_write(int fd)
  */
 
 /* Return the local port associated with fd. */
-u_short
+unsigned short
 comm_local_port(int fd)
 {
     Ip::Address temp;
@@ -429,7 +395,7 @@ comm_local_port(int fd)
     temp.InitAddrInfo(addr);
 
     if (getsockname(fd, addr->ai_addr, &(addr->ai_addrlen)) ) {
-        debugs(50, 1, "comm_local_port: Failed to retrieve TCP/UDP port number for socket: FD " << fd << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "comm_local_port: Failed to retrieve TCP/UDP port number for socket: FD " << fd << ": " << xstrerror());
         temp.FreeAddrInfo(addr);
         return 0;
     }
@@ -437,16 +403,12 @@ comm_local_port(int fd)
 
     temp.FreeAddrInfo(addr);
 
-    F->local_addr.SetPort(temp.GetPort());
-
-#if 0 // seems to undo comm_open actions on the FD ...
-    // grab default socket information for this address
-    temp.GetAddrInfo(addr);
-
-    F->sock_family = addr->ai_family;
-
-    temp.FreeAddrInfo(addr);
-#endif
+    if (F->local_addr.IsAnyAddr()) {
+        /* save the whole local address, not just the port. */
+        F->local_addr = temp;
+    } else {
+        F->local_addr.SetPort(temp.GetPort());
+    }
 
     debugs(5, 6, "comm_local_port: FD " << fd << ": port " << F->local_addr.GetPort() << "(family=" << F->sock_family << ")");
     return F->local_addr.GetPort();
@@ -455,7 +417,7 @@ comm_local_port(int fd)
 static comm_err_t
 commBind(int s, struct addrinfo &inaddr)
 {
-    statCounter.syscalls.sock.binds++;
+    ++ statCounter.syscalls.sock.binds;
 
     if (bind(s, inaddr.ai_addr, inaddr.ai_addrlen) == 0) {
         debugs(50, 6, "commBind: bind socket FD " << s << " to " << fd_table[s].local_addr);
@@ -481,6 +443,19 @@ comm_open(int sock_type,
     return comm_openex(sock_type, proto, addr, flags, 0, 0, note);
 }
 
+void
+comm_open_listener(int sock_type,
+                   int proto,
+                   Comm::ConnectionPointer &conn,
+                   const char *note)
+{
+    /* all listener sockets require bind() */
+    conn->flags |= COMM_DOBIND;
+
+    /* attempt native enabled port. */
+    conn->fd = comm_openex(sock_type, proto, conn->local, conn->flags, 0, 0, note);
+}
+
 int
 comm_open_listener(int sock_type,
                    int proto,
@@ -510,7 +485,7 @@ comm_set_v6only(int fd, int tos)
 {
 #ifdef IPV6_V6ONLY
     if (setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, (char *) &tos, sizeof(int)) < 0) {
-        debugs(50, 1, "comm_open: setsockopt(IPV6_V6ONLY) " << (tos?"ON":"OFF") << " for FD " << fd << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "comm_open: setsockopt(IPV6_V6ONLY) " << (tos?"ON":"OFF") << " for FD " << fd << ": " << xstrerror());
     }
 #else
     debugs(50, 0, "WARNING: comm_open: setsockopt(IPV6_V6ONLY) not supported on this platform");
@@ -554,7 +529,7 @@ comm_openex(int sock_type,
 
     PROF_start(comm_open);
     /* Create socket for accepting new connections. */
-    statCounter.syscalls.sock.sockets++;
+    ++ statCounter.syscalls.sock.sockets;
 
     /* Setup the socket addrinfo details for use */
     addr.GetAddrInfo(AI);
@@ -597,62 +572,63 @@ comm_openex(int sock_type,
         return -1;
     }
 
-    debugs(50, 3, "comm_openex: Opened socket FD " << new_socket << " : family=" << AI->ai_family << ", type=" << AI->ai_socktype << ", protocol=" << AI->ai_protocol );
+    // XXX: temporary for the transition. comm_openex will eventually have a conn to play with.
+    Comm::ConnectionPointer conn = new Comm::Connection;
+    conn->local = addr;
+    conn->fd = new_socket;
+
+    debugs(50, 3, "comm_openex: Opened socket " << conn << " : family=" << AI->ai_family << ", type=" << AI->ai_socktype << ", protocol=" << AI->ai_protocol );
 
     /* set TOS if needed */
     if (tos)
-        Ip::Qos::setSockTos(new_socket, tos);
+        Ip::Qos::setSockTos(conn, tos);
 
     /* set netfilter mark if needed */
     if (nfmark)
-        Ip::Qos::setSockNfmark(new_socket, nfmark);
+        Ip::Qos::setSockNfmark(conn, nfmark);
 
     if ( Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && addr.IsIPv6() )
-        comm_set_v6only(new_socket, 1);
+        comm_set_v6only(conn->fd, 1);
 
     /* Windows Vista supports Dual-Sockets. BUT defaults them to V6ONLY. Turn it OFF. */
     /* Other OS may have this administratively disabled for general use. Same deal. */
     if ( Ip::EnableIpv6&IPV6_SPECIAL_V4MAPPING && addr.IsIPv6() )
-        comm_set_v6only(new_socket, 0);
+        comm_set_v6only(conn->fd, 0);
 
-    comm_init_opened(new_socket, addr, tos, nfmark, note, AI);
-    new_socket = comm_apply_flags(new_socket, addr, flags, AI);
+    comm_init_opened(conn, tos, nfmark, note, AI);
+    new_socket = comm_apply_flags(conn->fd, addr, flags, AI);
 
     addr.FreeAddrInfo(AI);
 
     PROF_stop(comm_open);
 
+    // XXX transition only. prevent conn from closing the new FD on function exit.
+    conn->fd = -1;
     return new_socket;
 }
 
 /// update FD tables after a local or remote (IPC) comm_openex();
 void
-comm_init_opened(int new_socket,
-                 Ip::Address &addr,
+comm_init_opened(const Comm::ConnectionPointer &conn,
                  tos_t tos,
                  nfmark_t nfmark,
                  const char *note,
                  struct addrinfo *AI)
 {
-    assert(new_socket >= 0);
+    assert(Comm::IsConnOpen(conn));
     assert(AI);
 
-    fde *F = NULL;
-
     /* update fdstat */
-    debugs(5, 5, "comm_open: FD " << new_socket << " is a new socket");
-
-    assert(!isOpen(new_socket));
-    fd_open(new_socket, FD_SOCKET, note);
-
-    fdd_table[new_socket].close_file = NULL;
-
-    fdd_table[new_socket].close_line = 0;
+    debugs(5, 5, HERE << conn << " is a new socket");
 
-    F = &fd_table[new_socket];
+    assert(!isOpen(conn->fd)); // NP: global isOpen checks the fde entry for openness not the Comm::Connection
+    fd_open(conn->fd, FD_SOCKET, note);
 
-    F->local_addr = addr;
+    fdd_table[conn->fd].close_file = NULL;
+    fdd_table[conn->fd].close_line = 0;
 
+    fde *F = &fd_table[conn->fd];
+    F->local_addr = conn->local;
     F->tosToServer = tos;
 
     F->nfmarkToServer = nfmark;
@@ -678,8 +654,8 @@ comm_apply_flags(int new_socket,
     if ((flags & COMM_REUSEADDR))
         commSetReuseAddr(new_socket);
 
-    if (addr.GetPort() > (u_short) 0) {
-#ifdef _SQUID_MSWIN_
+    if (addr.GetPort() > (unsigned short) 0) {
+#if _SQUID_MSWIN_
         if (sock_type != SOCK_DGRAM)
 #endif
             commSetNoLinger(new_socket);
@@ -695,7 +671,7 @@ comm_apply_flags(int new_socket,
 
     if ( (flags & COMM_DOBIND) || addr.GetPort() > 0 || !addr.IsAnyAddr() ) {
         if ( !(flags & COMM_DOBIND) && addr.IsAnyAddr() )
-            debugs(5,1,"WARNING: Squid is attempting to bind() port " << addr << " without being a listener.");
+            debugs(5, DBG_IMPORTANT,"WARNING: Squid is attempting to bind() port " << addr << " without being a listener.");
         if ( addr.IsNoAddr() )
             debugs(5,0,"CRITICAL: Squid is attempting to bind() port " << addr << "!!");
 
@@ -724,37 +700,35 @@ comm_apply_flags(int new_socket,
 }
 
 void
-comm_import_opened(int fd,
-                   Ip::Address &addr,
-                   int flags,
+comm_import_opened(const Comm::ConnectionPointer &conn,
                    const char *note,
                    struct addrinfo *AI)
 {
-    debugs(5, 2, HERE << " FD " << fd << " at " << addr);
-    assert(fd >= 0);
+    debugs(5, 2, HERE << conn);
+    assert(Comm::IsConnOpen(conn));
     assert(AI);
 
-    comm_init_opened(fd, addr, 0, 0, note, AI);
+    comm_init_opened(conn, 0, 0, note, AI);
 
-    if (!(flags & COMM_NOCLOEXEC))
-        fd_table[fd].flags.close_on_exec = 1;
+    if (!(conn->flags & COMM_NOCLOEXEC))
+        fd_table[conn->fd].flags.close_on_exec = 1;
 
-    if (addr.GetPort() > (u_short) 0) {
-#ifdef _SQUID_MSWIN_
-        if (sock_type != SOCK_DGRAM)
+    if (conn->local.GetPort() > (unsigned short) 0) {
+#if _SQUID_MSWIN_
+        if (AI->ai_socktype != SOCK_DGRAM)
 #endif
-            fd_table[fd].flags.nolinger = 1;
+            fd_table[conn->fd].flags.nolinger = 1;
     }
 
-    if ((flags & COMM_TRANSPARENT))
-        fd_table[fd].flags.transparent = 1;
+    if ((conn->flags & COMM_TRANSPARENT))
+        fd_table[conn->fd].flags.transparent = 1;
 
-    if (flags & COMM_NONBLOCKING)
-        fd_table[fd].flags.nonblocking = 1;
+    if (conn->flags & COMM_NONBLOCKING)
+        fd_table[conn->fd].flags.nonblocking = 1;
 
 #ifdef TCP_NODELAY
     if (AI->ai_socktype == SOCK_STREAM)
-        fd_table[fd].flags.nodelay = 1;
+        fd_table[conn->fd].flags.nodelay = 1;
 #endif
 
     /* no fd_table[fd].flags. updates needed for these conditions:
@@ -763,371 +737,28 @@ comm_import_opened(int fd,
      */
 }
 
-
-CBDATA_CLASS_INIT(ConnectStateData);
-
-void *
-ConnectStateData::operator new (size_t size)
-{
-    CBDATA_INIT_TYPE(ConnectStateData);
-    return cbdataAlloc(ConnectStateData);
-}
-
-void
-ConnectStateData::operator delete (void *address)
-{
-    cbdataFree(address);
-}
-
-
-
-void
-commConnectStart(int fd, const char *host, u_short port, AsyncCall::Pointer &cb)
-{
-    debugs(cb->debugSection, cb->debugLevel, "commConnectStart: FD " << fd <<
-           ", cb " << cb << ", " << host << ":" << port); // TODO: just print *cb
-
-    ConnectStateData *cs;
-    cs = new ConnectStateData;
-    cs->fd = fd;
-    cs->host = xstrdup(host);
-    cs->default_port = port;
-    cs->callback = cb;
-
-    comm_add_close_handler(fd, commConnectFree, cs);
-    ipcache_nbgethostbyname(host, commConnectDnsHandle, cs);
-}
-
-// TODO: Remove this and similar callback registration functions by replacing
-// (callback,data) parameters with an AsyncCall so that we do not have to use
-// a generic call name and debug level when creating an AsyncCall. This will
-// also cut the number of callback registration routines in half.
-void
-commConnectStart(int fd, const char *host, u_short port, CNCB * callback, void *data)
-{
-    debugs(5, 5, "commConnectStart: FD " << fd << ", data " << data << ", " << host << ":" << port);
-    AsyncCall::Pointer call = commCbCall(5,3,
-                                         "SomeCommConnectHandler", CommConnectCbPtrFun(callback, data));
-    commConnectStart(fd, host, port, call);
-}
-
-static void
-commConnectDnsHandle(const ipcache_addrs *ia, const DnsLookupDetails &details, void *data)
-{
-    ConnectStateData *cs = (ConnectStateData *)data;
-    cs->dns = details;
-
-    if (ia == NULL) {
-        debugs(5, 3, "commConnectDnsHandle: Unknown host: " << cs->host);
-        cs->callCallback(COMM_ERR_DNS, 0);
-        return;
-    }
-
-    assert(ia->cur < ia->count);
-
-    cs->default_addr = ia->in_addrs[ia->cur];
-
-    if (Config.onoff.balance_on_multiple_ip)
-        ipcacheCycleAddr(cs->host, NULL);
-
-    cs->addrcount = ia->count;
-
-    cs->connstart = squid_curtime;
-
-    cs->connect();
-}
-
+// XXX: now that raw-FD timeouts are only unset for pipes and files this SHOULD be a no-op.
+// With handler already unset. Leaving this present until that can be verified for all code paths.
 void
-ConnectStateData::callCallback(comm_err_t status, int xerrno)
-{
-    debugs(5, 3, "commConnectCallback: FD " << fd);
-
-    comm_remove_close_handler(fd, commConnectFree, this);
-    commSetTimeout(fd, -1, NULL, NULL);
-
-    typedef CommConnectCbParams Params;
-    Params &params = GetCommParams<Params>(callback);
-    params.fd = fd;
-    params.dns = dns;
-    params.flag = status;
-    params.xerrno = xerrno;
-    ScheduleCallHere(callback);
-    callback = NULL;
-
-    commConnectFree(fd, this);
-}
-
-static void
-commConnectFree(int fd, void *data)
-{
-    ConnectStateData *cs = (ConnectStateData *)data;
-    debugs(5, 3, "commConnectFree: FD " << fd);
-//    delete cs->callback;
-    cs->callback = NULL;
-    safe_free(cs->host);
-    delete cs;
-}
-
-static void
-copyFDFlags(int to, fde *F)
+commUnsetFdTimeout(int fd)
 {
-    if (F->flags.close_on_exec)
-        commSetCloseOnExec(to);
-
-    if (F->flags.nonblocking)
-        commSetNonBlocking(to);
-
-#ifdef TCP_NODELAY
-
-    if (F->flags.nodelay)
-        commSetTcpNoDelay(to);
-
-#endif
-
-    if (Config.tcpRcvBufsz > 0)
-        commSetTcpRcvbuf(to, Config.tcpRcvBufsz);
-}
-
-/* Reset FD so that we can connect() again */
-int
-ConnectStateData::commResetFD()
-{
-
-// XXX: do we have to check this?
-//
-//    if (!cbdataReferenceValid(callback.data))
-//        return 0;
-
-    statCounter.syscalls.sock.sockets++;
-
-    fde *F = &fd_table[fd];
-
-    struct addrinfo *AI = NULL;
-    F->local_addr.GetAddrInfo(AI);
-    int new_family = AI->ai_family;
-
-    int fd2 = socket(new_family, AI->ai_socktype, AI->ai_protocol);
-
-    if (fd2 < 0) {
-        debugs(5, DBG_CRITICAL, HERE << "WARNING: FD " << fd2 << " socket failed to allocate: " << xstrerror());
-
-        if (ENFILE == errno || EMFILE == errno)
-            fdAdjustReserved();
-
-        F->local_addr.FreeAddrInfo(AI);
-        return 0;
-    }
-
-#ifdef _SQUID_MSWIN_
-
-    /* On Windows dup2() can't work correctly on Sockets, the          */
-    /* workaround is to close the destination Socket before call them. */
-    close(fd);
-
-#endif
-
-    if (dup2(fd2, fd) < 0) {
-        debugs(5, DBG_CRITICAL, HERE << "WARNING: dup2(FD " << fd2 << ", FD " << fd << ") failed: " << xstrerror());
-
-        if (ENFILE == errno || EMFILE == errno)
-            fdAdjustReserved();
-
-        close(fd2);
-
-        F->local_addr.FreeAddrInfo(AI);
-        return 0;
-    }
-    Comm::ResetSelect(fd);
-
-    close(fd2);
-
-    debugs(50, 3, "commResetFD: Reset socket FD " << fd << "->" << fd2 << " : family=" << new_family );
-
-    /* INET6: copy the new sockets family type to the FDE table */
-    F->sock_family = new_family;
-
-    F->flags.called_connect = 0;
-
-    /*
-     * yuck, this has assumptions about comm_open() arguments for
-     * the original socket
-     */
-
-    /* MUST be done before binding or face OS Error: "(99) Cannot assign requested address"... */
-    if ( F->flags.transparent ) {
-        comm_set_transparent(fd);
-    }
-
-    if (commBind(fd, *AI) != COMM_OK) {
-        debugs(5, DBG_CRITICAL, "WARNING: Reset of FD " << fd << " for " << F->local_addr << " failed to bind: " << xstrerror());
-        F->local_addr.FreeAddrInfo(AI);
-        return 0;
-    }
-    F->local_addr.FreeAddrInfo(AI);
-
-    if (F->tosToServer)
-        Ip::Qos::setSockTos(fd, F->tosToServer);
-
-    if (F->nfmarkToServer)
-        Ip::Qos::setSockNfmark(fd, F->nfmarkToServer);
-
-    if ( Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && F->local_addr.IsIPv6() )
-        comm_set_v6only(fd, 1);
-
-    copyFDFlags(fd, F);
-
-    return 1;
-}
-
-int
-ConnectStateData::commRetryConnect()
-{
-    assert(addrcount > 0);
-
-    if (addrcount == 1) {
-        if (tries >= Config.retry.maxtries)
-            return 0;
-
-        if (squid_curtime - connstart > Config.Timeout.connect)
-            return 0;
-    } else {
-        if (tries > addrcount) {
-            /* Flush bad address count in case we are
-             * skipping over incompatible protocol
-             */
-            ipcacheMarkAllGood(host);
-            return 0;
-        }
-    }
-
-    return commResetFD();
-}
-
-static void
-commReconnect(void *data)
-{
-    ConnectStateData *cs = (ConnectStateData *)data;
-    ipcache_nbgethostbyname(cs->host, commConnectDnsHandle, cs);
-}
-
-/** Connect SOCK to specified DEST_PORT at DEST_HOST. */
-void
-ConnectStateData::Connect(int fd, void *me)
-{
-    ConnectStateData *cs = (ConnectStateData *)me;
-    assert (cs->fd == fd);
-    cs->connect();
-}
-
-void
-ConnectStateData::defaults()
-{
-    S = default_addr;
-    S.SetPort(default_port);
-}
-
-void
-ConnectStateData::connect()
-{
-    defaults();
-
-    debugs(5,5, HERE << "to " << S);
-
-    switch (comm_connect_addr(fd, S) ) {
-
-    case COMM_INPROGRESS:
-        debugs(5, 5, HERE << "FD " << fd << ": COMM_INPROGRESS");
-        Comm::SetSelect(fd, COMM_SELECT_WRITE, ConnectStateData::Connect, this, 0);
-        break;
-
-    case COMM_OK:
-        debugs(5, 5, HERE << "FD " << fd << ": COMM_OK - connected");
-        ipcacheMarkGoodAddr(host, S);
-        callCallback(COMM_OK, 0);
-        break;
-
-    case COMM_ERR_PROTOCOL:
-        debugs(5, 5, HERE "FD " << fd << ": COMM_ERR_PROTOCOL - try again");
-        /* problem using the desired protocol over this socket.
-         * skip to the next address and hope it's more compatible
-         * but do not mark the current address as bad
-         */
-        tries++;
-        if (commRetryConnect()) {
-            /* Force an addr cycle to move forward to the next possible address */
-            ipcacheCycleAddr(host, NULL);
-            eventAdd("commReconnect", commReconnect, this, this->addrcount == 1 ? 0.05 : 0.0, 0);
-        } else {
-            debugs(5, 5, HERE << "FD " << fd << ": COMM_ERR_PROTOCOL - ERR tried too many times already.");
-            callCallback(COMM_ERR_CONNECT, errno);
-        }
-        break;
-
-    default:
-        debugs(5, 5, HERE "FD " << fd << ": * - try again");
-        tries++;
-        ipcacheMarkBadAddr(host, S);
-
-#if USE_ICMP
-        if (Config.onoff.test_reachability)
-            netdbDeleteAddrNetwork(S);
-#endif
-
-        if (commRetryConnect()) {
-            eventAdd("commReconnect", commReconnect, this, this->addrcount == 1 ? 0.05 : 0.0, 0);
-        } else {
-            debugs(5, 5, HERE << "FD " << fd << ": * - ERR tried too many times already.");
-            callCallback(COMM_ERR_CONNECT, errno);
-        }
-    }
-}
-/*
-int
-commSetTimeout_old(int fd, int timeout, PF * handler, void *data)
-{
-    debugs(5, 3, HERE << "FD " << fd << " timeout " << timeout);
+    debugs(5, 3, HERE << "Remove timeout for FD " << fd);
     assert(fd >= 0);
     assert(fd < Squid_MaxFD);
     fde *F = &fd_table[fd];
     assert(F->flags.open);
 
-    if (timeout < 0) {
-        cbdataReferenceDone(F->timeout_data);
-        F->timeout_handler = NULL;
-        F->timeout = 0;
-    } else {
-        if (handler) {
-            cbdataReferenceDone(F->timeout_data);
-            F->timeout_handler = handler;
-            F->timeout_data = cbdataReference(data);
-        }
-
-        F->timeout = squid_curtime + (time_t) timeout;
-    }
-
-    return F->timeout;
+    F->timeoutHandler = NULL;
+    F->timeout = 0;
 }
-*/
 
 int
-commSetTimeout(int fd, int timeout, PF * handler, void *data)
-{
-    AsyncCall::Pointer call;
-    debugs(5, 3, HERE << "FD " << fd << " timeout " << timeout);
-    if (handler != NULL)
-        call=commCbCall(5,4, "SomeTimeoutHandler", CommTimeoutCbPtrFun(handler, data));
-    else
-        call = NULL;
-    return commSetTimeout(fd, timeout, call);
-}
-
-
-int commSetTimeout(int fd, int timeout, AsyncCall::Pointer &callback)
+commSetConnTimeout(const Comm::ConnectionPointer &conn, int timeout, AsyncCall::Pointer &callback)
 {
-    debugs(5, 3, HERE << "FD " << fd << " timeout " << timeout);
-    assert(fd >= 0);
-    assert(fd < Squid_MaxFD);
-    fde *F = &fd_table[fd];
+    debugs(5, 3, HERE << conn << " timeout " << timeout);
+    assert(Comm::IsConnOpen(conn));
+    assert(conn->fd < Squid_MaxFD);
+    fde *F = &fd_table[conn->fd];
     assert(F->flags.open);
 
     if (timeout < 0) {
@@ -1137,7 +768,7 @@ int commSetTimeout(int fd, int timeout, AsyncCall::Pointer &callback)
         if (callback != NULL) {
             typedef CommTimeoutCbParams Params;
             Params &params = GetCommParams<Params>(callback);
-            params.fd = fd;
+            params.conn = conn;
             F->timeoutHandler = callback;
         }
 
@@ -1145,7 +776,14 @@ int commSetTimeout(int fd, int timeout, AsyncCall::Pointer &callback)
     }
 
     return F->timeout;
+}
 
+int
+commUnsetConnTimeout(const Comm::ConnectionPointer &conn)
+{
+    debugs(5, 3, HERE << "Remove timeout for " << conn);
+    AsyncCall::Pointer nil;
+    return commSetConnTimeout(conn, -1, nil);
 }
 
 int
@@ -1192,7 +830,7 @@ comm_connect_addr(int sock, const Ip::Address &address)
 
     if (!F->flags.called_connect) {
         F->flags.called_connect = 1;
-        statCounter.syscalls.sock.connects++;
+        ++ statCounter.syscalls.sock.connects;
 
         x = connect(sock, AI->ai_addr, AI->ai_addrlen);
 
@@ -1216,7 +854,7 @@ comm_connect_addr(int sock, const Ip::Address &address)
             debugs(14,9, "connecting to: " << address );
         }
     } else {
-#if defined(_SQUID_NEWSOS6_)
+#if _SQUID_NEWSOS6_
         /* Makoto MATSUSHITA <matusita@ics.es.osaka-u.ac.jp> */
 
         connect(sock, AI->ai_addr, AI->ai_addrlen);
@@ -1237,7 +875,7 @@ comm_connect_addr(int sock, const Ip::Address &address)
         if (x == 0)
             errno = err;
 
-#if defined(_SQUID_SOLARIS_)
+#if _SQUID_SOLARIS_
         /*
         * Solaris 2.4's socket emulation doesn't allow you
         * to determine the error from a failed non-blocking
@@ -1256,7 +894,7 @@ comm_connect_addr(int sock, const Ip::Address &address)
      * we leak memory on many connect requests because of EINPROGRESS.
      * If you find that this code is needed, please file a bug report. */
 #if 0
-#ifdef _SQUID_LINUX_
+#if _SQUID_LINUX_
     /* 2007-11-27:
      * Linux Debian replaces our allocated AI pointer with garbage when
      * connect() fails. This leads to segmentation faults deallocating
@@ -1287,9 +925,9 @@ comm_connect_addr(int sock, const Ip::Address &address)
     F->remote_port = address.GetPort(); /* remote_port is HS */
 
     if (status == COMM_OK) {
-        debugs(5, 10, "comm_connect_addr: FD " << sock << " connected to " << address);
+        debugs(5, DBG_DATA, "comm_connect_addr: FD " << sock << " connected to " << address);
     } else if (status == COMM_INPROGRESS) {
-        debugs(5, 10, "comm_connect_addr: FD " << sock << " connection pending");
+        debugs(5, DBG_DATA, "comm_connect_addr: FD " << sock << " connection pending");
     }
 
     return status;
@@ -1308,9 +946,6 @@ commCallCloseHandlers(int fd)
         // If call is not canceled schedule it for execution else ignore it
         if (!call->canceled()) {
             debugs(5, 5, "commCallCloseHandlers: ch->handler=" << call);
-            typedef CommCloseCbParams Params;
-            Params &params = GetCommParams<Params>(call);
-            params.fd = fd;
             ScheduleCallHere(call);
         }
     }
@@ -1331,10 +966,10 @@ commLingerClose(int fd, void *unused)
 }
 
 static void
-commLingerTimeout(int fd, void *unused)
+commLingerTimeout(const FdeCbParams &params)
 {
-    debugs(5, 3, "commLingerTimeout: FD " << fd);
-    comm_close(fd);
+    debugs(5, 3, "commLingerTimeout: FD " << params.fd);
+    comm_close(params.fd);
 }
 
 /*
@@ -1344,10 +979,8 @@ void
 comm_lingering_close(int fd)
 {
 #if USE_SSL
-
     if (fd_table[fd].ssl)
-        ssl_shutdown_method(fd);
-
+        ssl_shutdown_method(fd_table[fd].ssl);
 #endif
 
     if (shutdown(fd, 1) < 0) {
@@ -1356,20 +989,44 @@ comm_lingering_close(int fd)
     }
 
     fd_note(fd, "lingering close");
-    commSetTimeout(fd, 10, commLingerTimeout, NULL);
+    AsyncCall::Pointer call = commCbCall(5,4, "commLingerTimeout", FdeCbPtrFun(commLingerTimeout, NULL));
+
+    debugs(5, 3, HERE << "FD " << fd << " timeout " << timeout);
+    assert(fd_table[fd].flags.open);
+    if (callback != NULL) {
+        typedef FdeCbParams Params;
+        Params &params = GetCommParams<Params>(callback);
+        params.fd = fd;
+        fd_table[fd].timeoutHandler = callback;
+        fd_table[fd].timeout = squid_curtime + static_cast<time_t>(10);
+    }
+
     Comm::SetSelect(fd, COMM_SELECT_READ, commLingerClose, NULL, 0);
 }
 
 #endif
 
-/*
+/**
  * enable linger with time of 0 so that when the socket is
  * closed, TCP generates a RESET
  */
 void
-comm_reset_close(int fd)
+comm_reset_close(const Comm::ConnectionPointer &conn)
 {
+    struct linger L;
+    L.l_onoff = 1;
+    L.l_linger = 0;
+
+    if (setsockopt(conn->fd, SOL_SOCKET, SO_LINGER, (char *) &L, sizeof(L)) < 0)
+        debugs(50, DBG_CRITICAL, "ERROR: Closing " << conn << " with TCP RST: " << xstrerror());
+
+    conn->close();
+}
 
+// Legacy close function.
+void
+old_comm_reset_close(int fd)
+{
     struct linger L;
     L.l_onoff = 1;
     L.l_linger = 0;
@@ -1380,23 +1037,20 @@ comm_reset_close(int fd)
     comm_close(fd);
 }
 
+#if USE_SSL
 void
-comm_close_start(int fd, void *data)
+commStartSslClose(const FdeCbParams &params)
 {
-#if USE_SSL
-    fde *F = &fd_table[fd];
-    if (F->ssl)
-        ssl_shutdown_method(fd);
-
-#endif
-
+    assert(&fd_table[params.fd].ssl);
+    ssl_shutdown_method(fd_table[params.fd].ssl);
 }
+#endif
 
 void
-comm_close_complete(int fd, void *data)
+comm_close_complete(const FdeCbParams &params)
 {
 #if USE_SSL
-    fde *F = &fd_table[fd];
+    fde *F = &fd_table[params.fd];
 
     if (F->ssl) {
         SSL_free(F->ssl);
@@ -1408,13 +1062,12 @@ comm_close_complete(int fd, void *data)
         F->dynamicSslContext = NULL;
     }
 #endif
-    fd_close(fd);              /* update fdstat */
+    fd_close(params.fd);               /* update fdstat */
+    close(params.fd);
 
-    close(fd);
+    ++ statCounter.syscalls.sock.closes;
 
-    statCounter.syscalls.sock.closes++;
-
-    /* When an fd closes, give accept() a chance, if need be */
+    /* When one connection closes, give accept() a chance, if need be */
     Comm::AcceptLimiter::Instance().kick();
 }
 
@@ -1447,7 +1100,11 @@ _comm_close(int fd, char const *file, int line)
         return;
 
     /* The following fails because ipc.c is doing calls to pipe() to create sockets! */
-    assert(isOpen(fd));
+    if (!isOpen(fd)) {
+        debugs(50, DBG_IMPORTANT, HERE << "BUG 3556: FD " << fd << " is not an open socket.");
+        // XXX: do we need to run close(fd) or fd_close(fd) here?
+        return;
+    }
 
     assert(F->type != FD_FILE);
 
@@ -1455,17 +1112,20 @@ _comm_close(int fd, char const *file, int line)
 
     F->flags.close_request = 1;
 
-    AsyncCall::Pointer startCall=commCbCall(5,4, "comm_close_start",
-                                            CommCloseCbPtrFun(comm_close_start, NULL));
-    typedef CommCloseCbParams Params;
-    Params &startParams = GetCommParams<Params>(startCall);
-    startParams.fd = fd;
-    ScheduleCallHere(startCall);
+#if USE_SSL
+    if (F->ssl) {
+        AsyncCall::Pointer startCall=commCbCall(5,4, "commStartSslClose",
+                                                FdeCbPtrFun(commStartSslClose, NULL));
+        FdeCbParams &startParams = GetCommParams<FdeCbParams>(startCall);
+        startParams.fd = fd;
+        ScheduleCallHere(startCall);
+    }
+#endif
 
     // a half-closed fd may lack a reader, so we stop monitoring explicitly
     if (commHasHalfClosedMonitor(fd))
         commStopHalfClosedMonitor(fd);
-    commSetTimeout(fd, -1, NULL, NULL);
+    commUnsetFdTimeout(fd);
 
     // notify read/write handlers after canceling select reservations, if any
     if (COMMIO_FD_WRITECB(fd)->active()) {
@@ -1489,15 +1149,14 @@ _comm_close(int fd, char const *file, int line)
 
     commCallCloseHandlers(fd);
 
-    if (F->pconn.uses)
-        F->pconn.pool->count(F->pconn.uses);
+    if (F->pconn.uses && F->pconn.pool)
+        F->pconn.pool->noteUses(F->pconn.uses);
 
     comm_empty_os_read_buffers(fd);
 
-
     AsyncCall::Pointer completeCall=commCbCall(5,4, "comm_close_complete",
-                                    CommCloseCbPtrFun(comm_close_complete, NULL));
-    Params &completeParams = GetCommParams<Params>(completeCall);
+                                    FdeCbPtrFun(comm_close_complete, NULL));
+    FdeCbParams &completeParams = GetCommParams<FdeCbParams>(completeCall);
     completeParams.fd = fd;
     // must use async call to wait for all callbacks
     // scheduled before comm_close() to finish
@@ -1517,7 +1176,7 @@ comm_udp_sendto(int fd,
     struct addrinfo *AI = NULL;
 
     PROF_start(comm_udp_sendto);
-    statCounter.syscalls.sock.sendtos++;
+    ++ statCounter.syscalls.sock.sendtos;
 
     debugs(50, 3, "comm_udp_sendto: Attempt to send UDP packet to " << to_addr <<
            " using FD " << fd << " using Port " << comm_local_port(fd) );
@@ -1537,18 +1196,18 @@ comm_udp_sendto(int fd,
     if (x >= 0)
         return x;
 
-#ifdef _SQUID_LINUX_
+#if _SQUID_LINUX_
 
     if (ECONNREFUSED != errno)
 #endif
 
-        debugs(50, 1, "comm_udp_sendto: FD " << fd << ", (family=" << fd_table[fd].sock_family << ") " << to_addr << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "comm_udp_sendto: FD " << fd << ", (family=" << fd_table[fd].sock_family << ") " << to_addr << ": " << xstrerror());
 
     return COMM_ERROR;
 }
 
 void
-comm_add_close_handler(int fd, PF * handler, void *data)
+comm_add_close_handler(int fd, CLCB * handler, void *data)
 {
     debugs(5, 5, "comm_add_close_handler: FD " << fd << ", handler=" <<
            handler << ", data=" << data);
@@ -1572,10 +1231,9 @@ comm_add_close_handler(int fd, AsyncCall::Pointer &call)
     fd_table[fd].closeHandler = call;
 }
 
-
 // remove function-based close handler
 void
-comm_remove_close_handler(int fd, PF * handler, void *data)
+comm_remove_close_handler(int fd, CLCB * handler, void *data)
 {
     assert (isOpen(fd));
     /* Find handler in list */
@@ -1638,26 +1296,26 @@ commSetReuseAddr(int fd)
     int on = 1;
 
     if (setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, (char *) &on, sizeof(on)) < 0)
-        debugs(50, 1, "commSetReuseAddr: FD " << fd << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "commSetReuseAddr: FD " << fd << ": " << xstrerror());
 }
 
 static void
 commSetTcpRcvbuf(int fd, int size)
 {
     if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, (char *) &size, sizeof(size)) < 0)
-        debugs(50, 1, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
     if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, (char *) &size, sizeof(size)) < 0)
-        debugs(50, 1, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
 #ifdef TCP_WINDOW_CLAMP
     if (setsockopt(fd, SOL_TCP, TCP_WINDOW_CLAMP, (char *) &size, sizeof(size)) < 0)
-        debugs(50, 1, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "commSetTcpRcvbuf: FD " << fd << ", SIZE " << size << ": " << xstrerror());
 #endif
 }
 
 int
 commSetNonBlocking(int fd)
 {
-#ifndef _SQUID_MSWIN_
+#if !_SQUID_MSWIN_
     int flags;
     int dummy = 0;
 #endif
@@ -1677,7 +1335,7 @@ commSetNonBlocking(int fd)
     } else {
 #endif
 #endif
-#ifndef _SQUID_MSWIN_
+#if !_SQUID_MSWIN_
 
         if ((flags = fcntl(fd, F_GETFL, dummy)) < 0) {
             debugs(50, 0, "FD " << fd << ": fcntl F_GETFL: " << xstrerror());
@@ -1701,7 +1359,7 @@ commSetNonBlocking(int fd)
 int
 commUnsetNonBlocking(int fd)
 {
-#ifdef _SQUID_MSWIN_
+#if _SQUID_MSWIN_
     int nonblocking = FALSE;
 
     if (ioctlsocket(fd, FIONBIO, (unsigned long *) &nonblocking) < 0) {
@@ -1751,7 +1409,7 @@ commSetTcpNoDelay(int fd)
     int on = 1;
 
     if (setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, (char *) &on, sizeof(on)) < 0)
-        debugs(50, 1, "commSetTcpNoDelay: FD " << fd << ": " << xstrerror());
+        debugs(50, DBG_IMPORTANT, "commSetTcpNoDelay: FD " << fd << ": " << xstrerror());
 
     fd_table[fd].flags.nodelay = 1;
 }
@@ -1766,23 +1424,23 @@ commSetTcpKeepalive(int fd, int idle, int interval, int timeout)
     if (timeout && interval) {
         int count = (timeout + interval - 1) / interval;
         if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPCNT, &count, sizeof(on)) < 0)
-            debugs(5, 1, "commSetKeepalive: FD " << fd << ": " << xstrerror());
+            debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
     }
 #endif
 #ifdef TCP_KEEPIDLE
     if (idle) {
         if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPIDLE, &idle, sizeof(on)) < 0)
-            debugs(5, 1, "commSetKeepalive: FD " << fd << ": " << xstrerror());
+            debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
     }
 #endif
 #ifdef TCP_KEEPINTVL
     if (interval) {
         if (setsockopt(fd, IPPROTO_TCP, TCP_KEEPINTVL, &interval, sizeof(on)) < 0)
-            debugs(5, 1, "commSetKeepalive: FD " << fd << ": " << xstrerror());
+            debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
     }
 #endif
     if (setsockopt(fd, SOL_SOCKET, SO_KEEPALIVE, (char *) &on, sizeof(on)) < 0)
-        debugs(5, 1, "commSetKeepalive: FD " << fd << ": " << xstrerror());
+        debugs(5, DBG_IMPORTANT, "commSetKeepalive: FD " << fd << ": " << xstrerror());
 }
 
 void
@@ -2089,7 +1747,7 @@ commCloseAllSockets(void)
     int fd;
     fde *F = NULL;
 
-    for (fd = 0; fd <= Biggest_FD; fd++) {
+    for (fd = 0; fd <= Biggest_FD; ++fd) {
         F = &fd_table[fd];
 
         if (!F->flags.open)
@@ -2108,7 +1766,7 @@ commCloseAllSockets(void)
             ScheduleCallHere(callback);
         } else {
             debugs(5, 5, "commCloseAllSockets: FD " << fd << ": calling comm_reset_close()");
-            comm_reset_close(fd);
+            old_comm_reset_close(fd);
         }
     }
 }
@@ -2147,7 +1805,7 @@ checkTimeouts(void)
     fde *F = NULL;
     AsyncCall::Pointer callback;
 
-    for (fd = 0; fd <= Biggest_FD; fd++) {
+    for (fd = 0; fd <= Biggest_FD; ++fd) {
         F = &fd_table[fd];
 
         if (writeTimedOut(fd)) {
@@ -2172,60 +1830,6 @@ checkTimeouts(void)
     }
 }
 
-void CommIO::Initialise()
-{
-    /* Initialize done pipe signal */
-    int DonePipe[2];
-    if (pipe(DonePipe)) {}
-    DoneFD = DonePipe[1];
-    DoneReadFD = DonePipe[0];
-    fd_open(DoneReadFD, FD_PIPE, "async-io completetion event: main");
-    fd_open(DoneFD, FD_PIPE, "async-io completetion event: threads");
-    commSetNonBlocking(DoneReadFD);
-    commSetNonBlocking(DoneFD);
-    Comm::SetSelect(DoneReadFD, COMM_SELECT_READ, NULLFDHandler, NULL, 0);
-    Initialised = true;
-}
-
-void CommIO::NotifyIOClose()
-{
-    /* Close done pipe signal */
-    FlushPipe();
-    close(DoneFD);
-    close(DoneReadFD);
-    fd_close(DoneFD);
-    fd_close(DoneReadFD);
-    Initialised = false;
-}
-
-bool CommIO::Initialised = false;
-bool CommIO::DoneSignalled = false;
-int CommIO::DoneFD = -1;
-int CommIO::DoneReadFD = -1;
-
-void
-CommIO::FlushPipe()
-{
-    char buf[256];
-    FD_READ_METHOD(DoneReadFD, buf, sizeof(buf));
-}
-
-void
-CommIO::NULLFDHandler(int fd, void *data)
-{
-    FlushPipe();
-    Comm::SetSelect(fd, COMM_SELECT_READ, NULLFDHandler, NULL, 0);
-}
-
-void
-CommIO::ResetNotifications()
-{
-    if (DoneSignalled) {
-        FlushPipe();
-        DoneSignalled = false;
-    }
-}
-
 /// Start waiting for a possibly half-closed connection to close
 // by scheduling a read callback to a monitoring handler that
 // will close the connection on read errors.
@@ -2260,13 +1864,15 @@ commHalfClosedCheck(void *)
     typedef DescriptorSet::const_iterator DSCI;
     const DSCI end = TheHalfClosed->end();
     for (DSCI i = TheHalfClosed->begin(); i != end; ++i) {
-        const int fd = *i;
-        if (!fd_table[fd].halfClosedReader) { // not reading already
+        Comm::ConnectionPointer c = new Comm::Connection; // XXX: temporary. make HalfClosed a list of these.
+        c->fd = *i;
+        if (!fd_table[c->fd].halfClosedReader) { // not reading already
             AsyncCall::Pointer call = commCbCall(5,4, "commHalfClosedReader",
                                                  CommIoCbPtrFun(&commHalfClosedReader, NULL));
-            comm_read(fd, NULL, 0, call);
-            fd_table[fd].halfClosedReader = call;
-        }
+            comm_read(c, NULL, 0, call);
+            fd_table[c->fd].halfClosedReader = call;
+        } else
+            c->fd = -1; // XXX: temporary. prevent c replacement erase closing listed FD
     }
 
     WillCheckHalfClosed = false; // as far as we know
@@ -2298,13 +1904,14 @@ commStopHalfClosedMonitor(int const fd)
 
 /// I/O handler for the possibly half-closed connection monitoring code
 static void
-commHalfClosedReader(int fd, char *, size_t size, comm_err_t flag, int, void *)
+commHalfClosedReader(const Comm::ConnectionPointer &conn, char *, size_t size, comm_err_t flag, int, void *)
 {
     // there cannot be more data coming in on half-closed connections
     assert(size == 0);
-    assert(commHasHalfClosedMonitor(fd)); // or we would have canceled the read
+    assert(conn != NULL);
+    assert(commHasHalfClosedMonitor(conn->fd)); // or we would have canceled the read
 
-    fd_table[fd].halfClosedReader = NULL; // done reading, for now
+    fd_table[conn->fd].halfClosedReader = NULL; // done reading, for now
 
     // nothing to do if fd is being closed
     if (flag == COMM_ERR_CLOSING)
@@ -2312,8 +1919,8 @@ commHalfClosedReader(int fd, char *, size_t size, comm_err_t flag, int, void *)
 
     // if read failed, close the connection
     if (flag != COMM_OK) {
-        debugs(5, 3, "commHalfClosedReader: closing FD " << fd);
-        comm_close(fd);
+        debugs(5, 3, HERE << "closing " << conn);
+        conn->close();
         return;
     }
 
@@ -2321,11 +1928,10 @@ commHalfClosedReader(int fd, char *, size_t size, comm_err_t flag, int, void *)
     commPlanHalfClosedCheck(); // make sure this fd will be checked again
 }
 
+CommRead::CommRead() : conn(NULL), buf(NULL), len(0), callback(NULL) {}
 
-CommRead::CommRead() : fd(-1), buf(NULL), len(0), callback(NULL) {}
-
-CommRead::CommRead(int fd_, char *buf_, int len_, AsyncCall::Pointer &callback_)
-        : fd(fd_), buf(buf_), len(len_), callback(callback_) {}
+CommRead::CommRead(const Comm::ConnectionPointer &c, char *buf_, int len_, AsyncCall::Pointer &callback_)
+        : conn(c), buf(buf_), len(len_), callback(callback_) {}
 
 DeferredRead::DeferredRead () : theReader(NULL), theContext(NULL), theRead(), cancelled(false) {}
 
@@ -2346,26 +1952,27 @@ template cbdata_type CbDataList<DeferredRead>::CBDATA_CbDataList;
 void
 DeferredReadManager::delayRead(DeferredRead const &aRead)
 {
-    debugs(5, 3, "Adding deferred read on FD " << aRead.theRead.fd);
+    debugs(5, 3, "Adding deferred read on " << aRead.theRead.conn);
     CbDataList<DeferredRead> *temp = deferredReads.push_back(aRead);
 
     // We have to use a global function as a closer and point to temp
     // instead of "this" because DeferredReadManager is not a job and
     // is not even cbdata protected
+    // XXX: and yet we use cbdata protection functions on it??
     AsyncCall::Pointer closer = commCbCall(5,4,
                                            "DeferredReadManager::CloseHandler",
                                            CommCloseCbPtrFun(&CloseHandler, temp));
-    comm_add_close_handler(aRead.theRead.fd, closer);
+    comm_add_close_handler(aRead.theRead.conn->fd, closer);
     temp->element.closer = closer; // remeber so that we can cancel
 }
 
 void
-DeferredReadManager::CloseHandler(int fd, void *thecbdata)
+DeferredReadManager::CloseHandler(const CommCloseCbParams &params)
 {
-    if (!cbdataReferenceValid (thecbdata))
+    if (!cbdataReferenceValid(params.data))
         return;
 
-    CbDataList<DeferredRead> *temp = (CbDataList<DeferredRead> *)thecbdata;
+    CbDataList<DeferredRead> *temp = (CbDataList<DeferredRead> *)params.data;
 
     temp->element.closer = NULL;
     temp->element.markCancelled();
@@ -2377,8 +1984,21 @@ DeferredReadManager::popHead(CbDataListContainer<DeferredRead> &deferredReads)
     assert (!deferredReads.empty());
 
     DeferredRead &read = deferredReads.head->element;
+
+    // NOTE: at this point the connection has been paused/stalled for an unknown
+    //       amount of time. We must re-validate that it is active and usable.
+
+    // If the connection has been closed already. Cancel this read.
+    if (!Comm::IsConnOpen(read.theRead.conn)) {
+        if (read.closer != NULL) {
+            read.closer->cancel("Connection closed before.");
+            read.closer = NULL;
+        }
+        read.markCancelled();
+    }
+
     if (!read.cancelled) {
-        comm_remove_close_handler(read.theRead.fd, read.closer);
+        comm_remove_close_handler(read.theRead.conn->fd, read.closer);
         read.closer = NULL;
     }
 
@@ -2428,10 +2048,10 @@ DeferredReadManager::kickARead(DeferredRead const &aRead)
     if (aRead.cancelled)
         return;
 
-    if (aRead.theRead.fd>=0 && fd_table[aRead.theRead.fd].closing())
+    if (Comm::IsConnOpen(aRead.theRead.conn) && fd_table[aRead.theRead.conn->fd].closing())
         return;
 
-    debugs(5, 3, "Kicking deferred read on FD " << aRead.theRead.fd);
+    debugs(5, 3, "Kicking deferred read on " << aRead.theRead.conn);
 
     aRead.theReader(aRead.theContext, aRead.theRead);
 }
@@ -2442,10 +2062,6 @@ DeferredRead::markCancelled()
     cancelled = true;
 }
 
-ConnectionDetail::ConnectionDetail() : me(), peer()
-{
-}
-
 int
 CommSelectEngine::checkEvents(int timeout)
 {
@@ -2491,7 +2107,7 @@ comm_open_uds(int sock_type,
 
     PROF_start(comm_open);
     /* Create socket for accepting new connections. */
-    statCounter.syscalls.sock.sockets++;
+    ++ statCounter.syscalls.sock.sockets;
 
     /* Setup the socket addrinfo details for use */
     struct addrinfo AI;