From: Amos Jeffries <squid3@treenet.co.nz>
Date: Fri, 27 Aug 2010 09:33:34 +0000 (+1200)
Subject: Merged from trunk
X-Git-Tag: take08~55^2~124^2~79
X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=d1c7f781273c912a7928b8fb985a5dfc310de383;p=thirdparty%2Fsquid.git

Merged from trunk
---

d1c7f781273c912a7928b8fb985a5dfc310de383
diff --cc src/Server.cc
index cb743b33d9,bc70bf2fd9..9af6116dd1
--- a/src/Server.cc
+++ b/src/Server.cc
@@@ -421,9 -417,9 +421,9 @@@ ServerStateData::sendMoreRequestBody(
      if (requestBodySource->getMoreData(buf)) {
          debugs(9,3, HERE << "will write " << buf.contentSize() << " request body bytes");
          typedef CommCbMemFunT<ServerStateData, CommIoCbParams> Dialer;
-         requestSender = asyncCall(93,3, "ServerStateData::sentRequestBody",
-                                   Dialer(this, &ServerStateData::sentRequestBody));
+         requestSender = JobCallback(93,3,
+                                     Dialer, this, ServerStateData::sentRequestBody);
 -        comm_write_mbuf(fd, &buf, requestSender);
 +        comm_write_mbuf(conn->fd, &buf, requestSender);
      } else {
          debugs(9,3, HERE << "will wait for more request body bytes or eof");
          requestSender = NULL;
diff --cc src/adaptation/icap/Xaction.cc
index 4cfce774f1,9c2784eec0..9a3c64bf65
--- a/src/adaptation/icap/Xaction.cc
+++ b/src/adaptation/icap/Xaction.cc
@@@ -25,9 -24,10 +25,9 @@@ static PconnPool *icapPconnPool = new P
  
  //CBDATA_NAMESPACED_CLASS_INIT(Adaptation::Icap, Xaction);
  
- Adaptation::Icap::Xaction::Xaction(const char *aTypeName, Adaptation::Initiator *anInitiator, Adaptation::Icap::ServiceRep::Pointer &aService):
 -Adaptation::Icap::Xaction::Xaction(const char *aTypeName,
 -                                   Adaptation::Icap::ServiceRep::Pointer &aService):
++Adaptation::Icap::Xaction::Xaction(const char *aTypeName, Adaptation::Icap::ServiceRep::Pointer &aService):
          AsyncJob(aTypeName),
-         Adaptation::Initiate(aTypeName, anInitiator),
+         Adaptation::Initiate(aTypeName),
          icapRequest(NULL),
          icapReply(NULL),
          attempts(0),
@@@ -114,8 -106,9 +114,9 @@@ void Adaptation::Icap::Xaction::openCon
          // fake the connect callback
          // TODO: can we sync call Adaptation::Icap::Xaction::noteCommConnected here instead?
          typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> Dialer;
-         Dialer dialer(this, &Adaptation::Icap::Xaction::noteCommConnected);
+         CbcPointer<Xaction> self(this);
+         Dialer dialer(self, &Adaptation::Icap::Xaction::noteCommConnected);
 -        dialer.params.fd = connection;
 +        dialer.params.fd = connection->fd;
          dialer.params.flag = COMM_OK;
          // fake other parameters by copying from the existing connection
          connector = asyncCall(93,3, "Adaptation::Icap::Xaction::noteCommConnected", dialer);
@@@ -125,13 -118,40 +126,11 @@@
  
      disableRetries(); // we only retry pconn failures
  
 -    Ip::Address outgoing;
 -    if (!Ip::EnableIpv6 && !outgoing.SetIPv4()) {
 -        debugs(31, DBG_CRITICAL, "ERROR: IPv6 is disabled. " << outgoing << " is not an IPv4 address.");
 -        dieOnConnectionFailure(); // throws
 -    }
 -    /* split-stack for now requires default IPv4-only socket */
 -    if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && outgoing.IsAnyAddr() && !s.cfg().ipv6) {
 -        outgoing.SetIPv4();
 -    }
 -
 -    connection = comm_open(SOCK_STREAM, 0, outgoing,
 -                           COMM_NONBLOCKING, s.cfg().uri.termedBuf());
 -
 -    if (connection < 0)
 -        dieOnConnectionFailure(); // throws
 -
 -    debugs(93,3, typeName << " opens connection to " << s.cfg().host << ":" << s.cfg().port);
 -
 -    // TODO: service bypass status may differ from that of a transaction
 -    typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommTimeoutCbParams> TimeoutDialer;
 -    AsyncCall::Pointer timeoutCall = JobCallback(93, 5,
 -                                     TimeoutDialer, this, Adaptation::Icap::Xaction::noteCommTimedout);
 -    commSetTimeout(connection, TheConfig.connect_timeout(
 -                       service().cfg().bypass), timeoutCall);
 -
 -    typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommCloseCbParams> CloseDialer;
 -    closer = JobCallback(93, 5,
 -                         CloseDialer, this, Adaptation::Icap::Xaction::noteCommClosed);
 -    comm_add_close_handler(connection, closer);
 -
      typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> ConnectDialer;
-     connector = asyncCall(93,3, "Adaptation::Icap::Xaction::noteCommConnected",
-                           ConnectDialer(this, &Adaptation::Icap::Xaction::noteCommConnected));
- 
 -    connector = JobCallback(93,3,
 -                            ConnectDialer, this, Adaptation::Icap::Xaction::noteCommConnected);
 -    commConnectStart(connection, s.cfg().host.termedBuf(), s.cfg().port, connector);
++    connector = JobCallback(93,3, ConnectDialer, this, Adaptation::Icap::Xaction::noteCommConnected);
 +    Comm::ConnOpener *cs = new Comm::ConnOpener(connection, connector, TheConfig.connect_timeout(service().cfg().bypass));
 +    cs->setHost(s.cfg().host.termedBuf());
 +    AsyncJob::AsyncStart(cs);
  }
  
  /*
@@@ -222,14 -231,12 +221,12 @@@ void Adaptation::Icap::Xaction::dieOnCo
  
  void Adaptation::Icap::Xaction::scheduleWrite(MemBuf &buf)
  {
 +    Must(haveConnection());
 +
      // comm module will free the buffer
      typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommIoCbParams> Dialer;
-     writer = asyncCall(93,3, "Adaptation::Icap::Xaction::noteCommWrote",
-                        Dialer(this, &Adaptation::Icap::Xaction::noteCommWrote));
 -    writer = JobCallback(93,3,
 -                         Dialer, this, Adaptation::Icap::Xaction::noteCommWrote);
--
 -    comm_write_mbuf(connection, &buf, writer);
++    writer = JobCallback(93, 3, Dialer, this, Adaptation::Icap::Xaction::noteCommWrote);
 +    comm_write_mbuf(connection->fd, &buf, writer);
      updateTimeout();
  }
  
@@@ -310,11 -315,11 +307,8 @@@ void Adaptation::Icap::Xaction::updateT
          // XXX: why does Config.Timeout lacks a write timeout?
          // TODO: service bypass status may differ from that of a transaction
          typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommTimeoutCbParams> TimeoutDialer;
-         AsyncCall::Pointer call =  asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommTimedout",
-                                              TimeoutDialer(this,&Adaptation::Icap::Xaction::noteCommTimedout));
 -        AsyncCall::Pointer call = JobCallback(93,5,
 -                                              TimeoutDialer, this, Adaptation::Icap::Xaction::noteCommTimedout);
--
-         commSetTimeout(connection->fd,
 -        commSetTimeout(connection,
--                       TheConfig.io_timeout(service().cfg().bypass), call);
++        AsyncCall::Pointer call = JobCallback(93, 5, TimeoutDialer, this, Adaptation::Icap::Xaction::noteCommTimedout);
++        commSetTimeout(connection->fd, TheConfig.io_timeout(service().cfg().bypass), call);
      } else {
          // clear timeout when there is no I/O
          // Do we need a lifetime timeout?
@@@ -334,10 -339,10 +328,8 @@@ void Adaptation::Icap::Xaction::schedul
       * here instead of reading directly into readBuf.buf.
       */
      typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommIoCbParams> Dialer;
-     reader = asyncCall(93,3, "Adaptation::Icap::Xaction::noteCommRead",
-                        Dialer(this, &Adaptation::Icap::Xaction::noteCommRead));
 -    reader = JobCallback(93,3,
 -                         Dialer, this, Adaptation::Icap::Xaction::noteCommRead);
--
 -    comm_read(connection, commBuf, readBuf.spaceSize(), reader);
++    reader = JobCallback(93, 3, Dialer, this, Adaptation::Icap::Xaction::noteCommRead);
 +    comm_read(connection->fd, commBuf, readBuf.spaceSize(), reader);
      updateTimeout();
  }
  
diff --cc src/base/AsyncJob.cc
index 25f3b9099e,3fd5557138..6b50ff6973
--- a/src/base/AsyncJob.cc
+++ b/src/base/AsyncJob.cc
@@@ -27,15 -28,8 +28,10 @@@ AsyncJob::AsyncJob(const char *aTypeNam
  
  AsyncJob::~AsyncJob()
  {
 +    debugs(93,3, "AsyncJob of type " << typeName << " destructed, this=" << this <<
 +           " [async" << id << ']');
  }
  
- void AsyncJob::noteStart()
- {
-     start();
- }
- 
  void AsyncJob::start()
  {
  }
diff --cc src/client_side.cc
index 22919f60f0,ee33f79cee..026b5cbb76
--- a/src/client_side.cc
+++ b/src/client_side.cc
@@@ -244,9 -242,9 +244,8 @@@ ConnStateData::readSomeData(
      makeSpaceAvailable();
  
      typedef CommCbMemFunT<ConnStateData, CommIoCbParams> Dialer;
-     reader = asyncCall(33, 5, "ConnStateData::clientReadRequest",
-                        Dialer(this, &ConnStateData::clientReadRequest));
 -    reader = JobCallback(33, 5,
 -                         Dialer, this, ConnStateData::clientReadRequest);
 -    comm_read(fd, in.addressToReadInto(), getAvailableBufferLength(), reader);
++    reader = JobCallback(33, 5, Dialer, this, ConnStateData::clientReadRequest);
 +    comm_read(clientConn->fd, in.addressToReadInto(), getAvailableBufferLength(), reader);
  }
  
  
@@@ -1402,9 -1422,9 +1422,9 @@@ ConnStateData::readNextRequest(
       * Set the timeout BEFORE calling clientReadRequest().
       */
      typedef CommCbMemFunT<ConnStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(33, 5, "ConnStateData::requestTimeout",
-                                       TimeoutDialer(this, &ConnStateData::requestTimeout));
+     AsyncCall::Pointer timeoutCall = JobCallback(33, 5,
+                                      TimeoutDialer, this, ConnStateData::requestTimeout);
 -    commSetTimeout(fd, Config.Timeout.persistent_request, timeoutCall);
 +    commSetTimeout(clientConn->fd, Config.Timeout.persistent_request, timeoutCall);
  
      readSomeData();
      /** Please don't do anything with the FD past here! */
@@@ -3097,22 -3121,22 +3121,21 @@@ httpAccept(int sock, int unused, Comm::
          return;
      }
  
 -    debugs(33, 4, "httpAccept: FD " << newfd << ": accepted");
 -    fd_note(newfd, "client http connect");
 -    connState = connStateCreate(&details->peer, &details->me, newfd, s);
 +    debugs(33, 4, HERE << details << ": accepted");
 +    fd_note(details->fd, "client http connect");
 +    connState = connStateCreate(details, s);
  
      typedef CommCbMemFunT<ConnStateData, CommCloseCbParams> Dialer;
-     AsyncCall::Pointer call = asyncCall(33, 5, "ConnStateData::connStateClosed",
-                                         Dialer(connState, &ConnStateData::connStateClosed));
 -    AsyncCall::Pointer call = JobCallback(33, 5,
 -                                          Dialer, connState, ConnStateData::connStateClosed);
 -    comm_add_close_handler(newfd, call);
++    AsyncCall::Pointer call = JobCallback(33, 5, Dialer, connState, ConnStateData::connStateClosed);
 +    comm_add_close_handler(details->fd, call);
  
      if (Config.onoff.log_fqdn)
 -        fqdncache_gethostbyaddr(details->peer, FQDN_LOOKUP_IF_MISS);
 +        fqdncache_gethostbyaddr(details->remote, FQDN_LOOKUP_IF_MISS);
  
      typedef CommCbMemFunT<ConnStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(33, 5, "ConnStateData::requestTimeout",
-                                       TimeoutDialer(connState,&ConnStateData::requestTimeout));
+     AsyncCall::Pointer timeoutCall =  JobCallback(33, 5,
+                                       TimeoutDialer, connState, ConnStateData::requestTimeout);
 -    commSetTimeout(newfd, Config.Timeout.read, timeoutCall);
 +    commSetTimeout(details->fd, Config.Timeout.read, timeoutCall);
  
  #if USE_IDENT
      if (Ident::TheConfig.identLookup) {
@@@ -3305,24 -3329,25 +3328,23 @@@ httpsAccept(int sock, int newfd, Comm::
      }
  
      SSL *ssl = NULL;
 -    if (!(ssl = httpsCreate(newfd, details, sslContext)))
 +    if (!(ssl = httpsCreate(details, sslContext)))
          return;
  
 -    debugs(33, 5, "httpsAccept: FD " << newfd << " accepted, starting SSL negotiation.");
 -    fd_note(newfd, "client https connect");
 -    ConnStateData *connState = connStateCreate(details->peer, details->me,
 -                               newfd, &s->http);
 +    debugs(33, 5, HERE << details << " accepted, starting SSL negotiation.");
 +    fd_note(details->fd, "client https connect");
 +    ConnStateData *connState = connStateCreate(details, &s->http);
      typedef CommCbMemFunT<ConnStateData, CommCloseCbParams> Dialer;
-     AsyncCall::Pointer call = asyncCall(33, 5, "ConnStateData::connStateClosed",
-                                         Dialer(connState, &ConnStateData::connStateClosed));
 -    AsyncCall::Pointer call = JobCallback(33, 5,
 -                                          Dialer, connState, ConnStateData::connStateClosed);
 -    comm_add_close_handler(newfd, call);
++    AsyncCall::Pointer call = JobCallback(33, 5, Dialer, connState, ConnStateData::connStateClosed);
 +    comm_add_close_handler(details->fd, call);
  
      if (Config.onoff.log_fqdn)
 -        fqdncache_gethostbyaddr(details->peer, FQDN_LOOKUP_IF_MISS);
 +        fqdncache_gethostbyaddr(details->remote, FQDN_LOOKUP_IF_MISS);
  
      typedef CommCbMemFunT<ConnStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(33, 5, "ConnStateData::requestTimeout",
-                                       TimeoutDialer(connState,&ConnStateData::requestTimeout));
+     AsyncCall::Pointer timeoutCall =  JobCallback(33, 5,
+                                       TimeoutDialer, connState, ConnStateData::requestTimeout);
 -    commSetTimeout(newfd, Config.Timeout.request, timeoutCall);
 +    commSetTimeout(details->fd, Config.Timeout.request, timeoutCall);
  
  #if USE_IDENT
      if (Ident::TheConfig.identLookup) {
diff --cc src/ftp.cc
index 6925dd711d,758bc5ec38..2cdfa32497
--- a/src/ftp.cc
+++ b/src/ftp.cc
@@@ -489,9 -479,9 +489,8 @@@ FtpStateData::FtpStateData(FwdState *th
      flags.rest_supported = 1;
  
      typedef CommCbMemFunT<FtpStateData, CommCloseCbParams> Dialer;
-     AsyncCall::Pointer closer = asyncCall(9, 5, "FtpStateData::ctrlClosed",
-                                           Dialer(this, &FtpStateData::ctrlClosed));
 -    AsyncCall::Pointer closer = JobCallback(9, 5,
 -                                            Dialer, this, FtpStateData::ctrlClosed);
 -    ctrl.opened(theFwdState->server_fd, closer);
++    AsyncCall::Pointer closer = JobCallback(9, 5, Dialer, this, FtpStateData::ctrlClosed);
 +    ctrl.opened(conn, closer);
  
      if (request->method == METHOD_PUT)
          flags.put = 1;
@@@ -613,22 -603,6 +612,21 @@@ FtpStateData::loginParser(const char *l
      debugs(9, 9, HERE << ": OUT: login='" << login << "', escaped=" << escaped << ", user=" << user << ", password=" << password);
  }
  
 +/**
 + * Cancel the timeout on the Control socket and establish one
 + * on the data socket
 + */
 +void
 +FtpStateData::switchTimeoutToDataChannel()
 +{
 +    AsyncCall::Pointer nullCall = NULL;
 +    commSetTimeout(ctrl.conn->fd, -1, nullCall);
 +
 +    typedef CommCbMemFunT<FtpStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(9, 5, "FtpStateData::ftpTimeout",
-                                       TimeoutDialer(this, &FtpStateData::ftpTimeout));
++    AsyncCall::Pointer timeoutCall =  JobCallback(9, 5, TimeoutDialer, this, FtpStateData::ftpTimeout);
 +    commSetTimeout(data.conn->fd, Config.Timeout.read, timeoutCall);
 +}
 +
  void
  FtpStateData::ftpTimeout(const CommTimeoutCbParams &io)
  {
@@@ -1183,16 -1158,15 +1181,15 @@@ FtpStateData::maybeReadVirginBody(
      data.read_pending = true;
  
      typedef CommCbMemFunT<FtpStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(9, 5, "FtpStateData::ftpTimeout",
-                                       TimeoutDialer(this,&FtpStateData::ftpTimeout));
+     AsyncCall::Pointer timeoutCall =  JobCallback(9, 5,
+                                       TimeoutDialer, this, FtpStateData::ftpTimeout);
 -    commSetTimeout(data.fd, Config.Timeout.read, timeoutCall);
 +    commSetTimeout(data.conn->fd, Config.Timeout.read, timeoutCall);
  
 -    debugs(9,5,HERE << "queueing read on FD " << data.fd);
 +    debugs(9,5,HERE << "queueing read on FD " << data.conn->fd);
  
      typedef CommCbMemFunT<FtpStateData, CommIoCbParams> Dialer;
 -    entry->delayAwareRead(data.fd, data.readBuf->space(), read_sz,
 +    entry->delayAwareRead(data.conn->fd, data.readBuf->space(), read_sz,
-                           asyncCall(9, 5, "FtpStateData::dataRead",
-                                     Dialer(this, &FtpStateData::dataRead)));
+                           JobCallback(9, 5, Dialer, this, FtpStateData::dataRead));
  }
  
  void
@@@ -1548,9 -1528,9 +1545,8 @@@ FtpStateData::writeCommand(const char *
      }
  
      typedef CommCbMemFunT<FtpStateData, CommIoCbParams> Dialer;
-     AsyncCall::Pointer call = asyncCall(9, 5, "FtpStateData::ftpWriteCommandCallback",
-                                         Dialer(this, &FtpStateData::ftpWriteCommandCallback));
 -    AsyncCall::Pointer call = JobCallback(9, 5,
 -                                          Dialer, this, FtpStateData::ftpWriteCommandCallback);
 -    comm_write(ctrl.fd,
++    AsyncCall::Pointer call = JobCallback(9, 5, Dialer, this, FtpStateData::ftpWriteCommandCallback);
 +    comm_write(ctrl.conn->fd,
                 ctrl.last_command,
                 strlen(ctrl.last_command),
                 call);
@@@ -1694,15 -1680,10 +1690,12 @@@ FtpStateData::scheduleReadControlReply(
          }
  
          typedef CommCbMemFunT<FtpStateData, CommTimeoutCbParams> TimeoutDialer;
-         AsyncCall::Pointer timeoutCall =  asyncCall(9, 5, "FtpStateData::ftpTimeout",
-                                           TimeoutDialer(this,&FtpStateData::ftpTimeout));
- 
 -        AsyncCall::Pointer timeoutCall = JobCallback(9, 5,
 -                                         TimeoutDialer, this, FtpStateData::ftpTimeout);
++        AsyncCall::Pointer timeoutCall = JobCallback(9, 5, TimeoutDialer, this, FtpStateData::ftpTimeout);
 +        commSetTimeout(ctrl.conn->fd, Config.Timeout.read, timeoutCall);
  
 -        commSetTimeout(ctrl.fd, Config.Timeout.read, timeoutCall);
 +        typedef CommCbMemFunT<FtpStateData, CommIoCbParams> Dialer;
-         AsyncCall::Pointer reader=asyncCall(9, 5, "FtpStateData::ftpReadControlReply",
-                                             Dialer(this, &FtpStateData::ftpReadControlReply));
++        AsyncCall::Pointer reader = JobCallback(9, 5, Dialer, this, FtpStateData::ftpReadControlReply);
 +        comm_read(ctrl.conn->fd, ctrl.buf + ctrl.offset, ctrl.size - ctrl.offset, reader);
      }
  }
  
@@@ -2551,13 -2561,13 +2544,13 @@@ ftpSendPassive(FtpStateData * ftpState
  
      /*
       * ugly hack for ftp servers like ftp.netscape.com that sometimes
 -     * dont acknowledge PASV commands.
 +     * dont acknowledge PASV commands. Use connect timeout to be faster then read timeout (minutes).
       */
      typedef CommCbMemFunT<FtpStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(9, 5, "FtpStateData::ftpTimeout",
-                                       TimeoutDialer(ftpState,&FtpStateData::ftpTimeout));
+     AsyncCall::Pointer timeoutCall =  JobCallback(9, 5,
+                                       TimeoutDialer, ftpState, FtpStateData::ftpTimeout);
  
 -    commSetTimeout(ftpState->data.fd, 15, timeoutCall);
 +    commSetTimeout(ftpState->ctrl.conn->fd, Config.Timeout.connect, timeoutCall);
  }
  
  void
@@@ -2738,22 -2763,19 +2731,21 @@@ ftpOpenListenSocket(FtpStateData * ftpS
      }
  
      typedef CommCbMemFunT<FtpStateData, CommAcceptCbParams> acceptDialer;
-     AsyncCall::Pointer acceptCall = asyncCall(11, 5, "FtpStateData::ftpAcceptDataConnection",
-                                     acceptDialer(ftpState, &FtpStateData::ftpAcceptDataConnection));
 -    AsyncCall::Pointer acceptCall = JobCallback(11, 5,
 -                                    acceptDialer, ftpState, FtpStateData::ftpAcceptDataConnection);
 -    ftpState->data.listener = new Comm::ListenStateData(fd, acceptCall, false);
++    AsyncCall::Pointer acceptCall = JobCallback(11, 5, acceptDialer, ftpState, FtpStateData::ftpAcceptDataConnection);
 +    ftpState->data.listener = new Comm::ConnAcceptor(conn, false, ftpState->entry->url());
 +    ftpState->data.listener->subscribe(acceptCall);
  
 -    if (!ftpState->data.listener || ftpState->data.listener->errcode != 0) {
 -        comm_close(fd);
 -        return -1;
 +    if (ftpState->data.listener->errcode != 0) {
 +        conn->close();
 +    } else {
 +
 +        if (!fallback)
 +            conn->local.SetPort(comm_local_port(conn->fd));
 +        ftpState->data.host = NULL;
 +        AsyncJob::AsyncStart(ftpState->data.listener);
      }
  
 -    ftpState->data.opened(fd, ftpState->dataCloser());
 -    ftpState->data.port = comm_local_port(fd);
 -    ftpState->data.host = NULL;
 -    return fd;
 +    ftpState->data.listen_conn = conn;
  }
  
  /// \ingroup ServerProtocolFTPInternal
@@@ -2901,24 -2933,22 +2893,23 @@@ FtpStateData::ftpAcceptDataConnection(c
       * This prevents third-party hacks, but also third-party load balancing handshakes.
       */
      if (Config.Ftp.sanitycheck) {
 -        io.details.peer.NtoA(ntoapeer,MAX_IPSTRLEN);
 +        io.details->remote.NtoA(ntoapeer,MAX_IPSTRLEN);
  
 -        if (strcmp(fd_table[ctrl.fd].ipaddr, ntoapeer) != 0) {
 +        // accept if either our data or ctrl connection is talking to this remote peer.
 +        if (data.listen_conn->remote != io.details->remote && ctrl.conn->remote != io.details->remote) {
              debugs(9, DBG_IMPORTANT,
                     "FTP data connection from unexpected server (" <<
 -                   io.details.peer << "), expecting " <<
 -                   fd_table[ctrl.fd].ipaddr);
 +                   io.details->remote << "), expecting " <<
 +                   data.listen_conn->remote << " or " << ctrl.conn->remote);
  
 -            /* close the bad soures connection down ASAP. */
 -            comm_close(io.nfd);
 +            /* drop the bad connection (io) by ignoring. */
  
 -            /* we are ony accepting once, so need to re-open the listener socket. */
 +            /* we are ony accepting once, so need to reset the listener socket. */
              typedef CommCbMemFunT<FtpStateData, CommAcceptCbParams> acceptDialer;
-             AsyncCall::Pointer acceptCall = asyncCall(11, 5, "FtpStateData::ftpAcceptDataConnection",
-                                             acceptDialer(this, &FtpStateData::ftpAcceptDataConnection));
 -            AsyncCall::Pointer acceptCall = JobCallback(11, 5,
 -                                            acceptDialer, this, FtpStateData::ftpAcceptDataConnection);
 -            data.listener = new Comm::ListenStateData(data.fd, acceptCall, false);
++            AsyncCall::Pointer acceptCall = JobCallback(11, 5, acceptDialer, this, FtpStateData::ftpAcceptDataConnection);
 +            data.listener = new Comm::ConnAcceptor(data.listen_conn, false, data.host);
 +            data.listener->subscribe(acceptCall);
 +            AsyncJob::AsyncStart(data.listener);
              return;
          }
      }
@@@ -3017,15 -3078,14 +3008,15 @@@ void FtpStateData::readStor(
          state = WRITING_DATA;
          debugs(9, 3, HERE << "writing data channel");
      } else if (code == 150) {
 -        /*\par
 -         * When client code is 150 with a hostname, Accept data channel. */
 +        /* When client code is 150 with no data channel, Accept data channel. */
          debugs(9, 3, "ftpReadStor: accepting data channel");
          typedef CommCbMemFunT<FtpStateData, CommAcceptCbParams> acceptDialer;
-         AsyncCall::Pointer acceptCall = asyncCall(11, 5, "FtpStateData::ftpAcceptDataConnection",
-                                         acceptDialer(this, &FtpStateData::ftpAcceptDataConnection));
+         AsyncCall::Pointer acceptCall = JobCallback(11, 5,
+                                         acceptDialer, this, FtpStateData::ftpAcceptDataConnection);
  
 -        data.listener = new Comm::ListenStateData(data.fd, acceptCall, false);
 +        data.listener = new Comm::ConnAcceptor(data.conn, false, data.host);
 +        data.listener->subscribe(acceptCall);
 +        AsyncJob::AsyncStart(data.listener);
      } else {
          debugs(9, DBG_IMPORTANT, HERE << "Unexpected reply code "<< std::setfill('0') << std::setw(3) << code);
          ftpFail(this);
@@@ -3143,25 -3203,36 +3134,25 @@@ ftpReadList(FtpStateData * ftpState
      int code = ftpState->ctrl.replycode;
      debugs(9, 3, HERE);
  
 -    if (code == 125 || (code == 150 && ftpState->data.host)) {
 +    if (code == 125 || (code == 150 && Comm::IsConnOpen(ftpState->data.conn))) {
 +        debugs(9, 3, HERE << "begin data transfer from " << ftpState->data.conn->remote);
          /* Begin data transfer */
 -        /* XXX what about Config.Timeout.read? */
 +        ftpState->switchTimeoutToDataChannel();
          ftpState->maybeReadVirginBody();
          ftpState->state = READING_DATA;
 -        /*
 -         * Cancel the timeout on the Control socket and establish one
 -         * on the data socket
 -         */
 -        AsyncCall::Pointer nullCall = NULL;
 -        commSetTimeout(ftpState->ctrl.fd, -1, nullCall);
          return;
      } else if (code == 150) {
 +        debugs(9, 3, HERE << "accept data channel from " << ftpState->ctrl.conn->remote);
 +        ftpState->switchTimeoutToDataChannel();
 +
          /* Accept data channel */
          typedef CommCbMemFunT<FtpStateData, CommAcceptCbParams> acceptDialer;
-         AsyncCall::Pointer acceptCall = asyncCall(11, 5, "FtpStateData::ftpAcceptDataConnection",
-                                         acceptDialer(ftpState, &FtpStateData::ftpAcceptDataConnection));
+         AsyncCall::Pointer acceptCall = JobCallback(11, 5,
+                                         acceptDialer, ftpState, FtpStateData::ftpAcceptDataConnection);
  
 -        ftpState->data.listener = new Comm::ListenStateData(ftpState->data.fd, acceptCall, false);
 -        /*
 -         * Cancel the timeout on the Control socket and establish one
 -         * on the data socket
 -         */
 -        AsyncCall::Pointer nullCall = NULL;
 -        commSetTimeout(ftpState->ctrl.fd, -1, nullCall);
 -
 -        typedef CommCbMemFunT<FtpStateData, CommTimeoutCbParams> TimeoutDialer;
 -        AsyncCall::Pointer timeoutCall =  JobCallback(9, 5,
 -                                          TimeoutDialer, ftpState,FtpStateData::ftpTimeout);
 -        commSetTimeout(ftpState->data.fd, Config.Timeout.read, timeoutCall);
 +        ftpState->data.listener = new Comm::ConnAcceptor(ftpState->data.conn, false, ftpState->data.host);
 +        ftpState->data.listener->subscribe(acceptCall);
 +        AsyncJob::AsyncStart(ftpState->data.listener);
          return;
      } else if (!ftpState->flags.tried_nlst && code > 300) {
          ftpSendNlst(ftpState);
@@@ -3194,21 -3265,35 +3185,21 @@@ ftpReadRetr(FtpStateData * ftpState
      int code = ftpState->ctrl.replycode;
      debugs(9, 3, HERE);
  
 -    if (code == 125 || (code == 150 && ftpState->data.host)) {
 +    if (code == 125 || (code == 150 && Comm::IsConnOpen(ftpState->data.conn))) {
          /* Begin data transfer */
          debugs(9, 3, HERE << "reading data channel");
 -        /* XXX what about Config.Timeout.read? */
 +        ftpState->switchTimeoutToDataChannel();
          ftpState->maybeReadVirginBody();
          ftpState->state = READING_DATA;
 -        /*
 -         * Cancel the timeout on the Control socket and establish one
 -         * on the data socket
 -         */
 -        AsyncCall::Pointer nullCall = NULL;
 -        commSetTimeout(ftpState->ctrl.fd, -1, nullCall);
      } else if (code == 150) {
          /* Accept data channel */
 +        ftpState->switchTimeoutToDataChannel();
          typedef CommCbMemFunT<FtpStateData, CommAcceptCbParams> acceptDialer;
-         AsyncCall::Pointer acceptCall = asyncCall(11, 5, "FtpStateData::ftpAcceptDataConnection",
-                                         acceptDialer(ftpState, &FtpStateData::ftpAcceptDataConnection));
+         AsyncCall::Pointer acceptCall = JobCallback(11, 5,
+                                         acceptDialer, ftpState, FtpStateData::ftpAcceptDataConnection);
 -        ftpState->data.listener = new Comm::ListenStateData(ftpState->data.fd, acceptCall, false);
 -        /*
 -         * Cancel the timeout on the Control socket and establish one
 -         * on the data socket
 -         */
 -        AsyncCall::Pointer nullCall = NULL;
 -        commSetTimeout(ftpState->ctrl.fd, -1, nullCall);
 -
 -        typedef CommCbMemFunT<FtpStateData, CommTimeoutCbParams> TimeoutDialer;
 -        AsyncCall::Pointer timeoutCall =  JobCallback(9, 5,
 -                                          TimeoutDialer, ftpState,FtpStateData::ftpTimeout);
 -        commSetTimeout(ftpState->data.fd, Config.Timeout.read, timeoutCall);
 +        ftpState->data.listener = new Comm::ConnAcceptor(ftpState->data.conn, false, ftpState->data.host);
 +        ftpState->data.listener->subscribe(acceptCall);
 +        AsyncJob::AsyncStart(ftpState->data.listener);
      } else if (code >= 300) {
          if (!ftpState->flags.try_slash_hack) {
              /* Try this as a directory missing trailing slash... */
diff --cc src/http.cc
index e9c2b2ed7c,3d1d8bdc78..df012c91b2
--- a/src/http.cc
+++ b/src/http.cc
@@@ -142,9 -142,9 +142,8 @@@ HttpStateData::HttpStateData(FwdState *
       * register the handler to free HTTP state data when the FD closes
       */
      typedef CommCbMemFunT<HttpStateData, CommCloseCbParams> Dialer;
-     closeHandler = asyncCall(9, 5, "httpStateData::httpStateConnClosed",
-                              Dialer(this,&HttpStateData::httpStateConnClosed));
 -    closeHandler = JobCallback(9, 5,
 -                               Dialer, this, HttpStateData::httpStateConnClosed);
 -    comm_add_close_handler(fd, closeHandler);
++    closeHandler = JobCallback(9, 5, Dialer, this, HttpStateData::httpStateConnClosed);
 +    comm_add_close_handler(serverConnection->fd, closeHandler);
  }
  
  HttpStateData::~HttpStateData()
@@@ -1404,9 -1402,8 +1403,8 @@@ HttpStateData::maybeReadVirginBody(
      if (flags.do_next_read) {
          flags.do_next_read = 0;
          typedef CommCbMemFunT<HttpStateData, CommIoCbParams> Dialer;
 -        entry->delayAwareRead(fd, readBuf->space(read_size), read_size,
 +        entry->delayAwareRead(serverConnection->fd, readBuf->space(read_size), read_size,
-                               asyncCall(11, 5, "HttpStateData::readReply",
-                                         Dialer(this, &HttpStateData::readReply)));
+                               JobCallback(11, 5, Dialer, this,  HttpStateData::readReply));
      }
  }
  
@@@ -1449,10 -1446,10 +1447,10 @@@ HttpStateData::sendComplete(const CommI
       * request bodies.
       */
      typedef CommCbMemFunT<HttpStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(11, 5, "HttpStateData::httpTimeout",
-                                       TimeoutDialer(this,&HttpStateData::httpTimeout));
+     AsyncCall::Pointer timeoutCall =  JobCallback(11, 5,
+                                       TimeoutDialer, this, HttpStateData::httpTimeout);
  
 -    commSetTimeout(fd, Config.Timeout.read, timeoutCall);
 +    commSetTimeout(serverConnection->fd, Config.Timeout.read, timeoutCall);
  
      flags.request_sent = 1;
  
@@@ -1992,9 -1988,9 +1987,9 @@@ HttpStateData::sendRequest(
      }
  
      typedef CommCbMemFunT<HttpStateData, CommTimeoutCbParams> TimeoutDialer;
-     AsyncCall::Pointer timeoutCall =  asyncCall(11, 5, "HttpStateData::httpTimeout",
-                                       TimeoutDialer(this,&HttpStateData::httpTimeout));
-     commSetTimeout(serverConnection->fd, Config.Timeout.lifetime, timeoutCall);
+     AsyncCall::Pointer timeoutCall =  JobCallback(11, 5,
+                                       TimeoutDialer, this, HttpStateData::httpTimeout);
 -    commSetTimeout(fd, Config.Timeout.lifetime, timeoutCall);
++    commSetTimeout(srverConnection->fd, Config.Timeout.lifetime, timeoutCall);
      flags.do_next_read = 1;
      maybeReadVirginBody();
  
@@@ -2102,9 -2098,9 +2097,8 @@@ HttpStateData::doneSendingRequestBody(
              }
  
              typedef CommCbMemFunT<HttpStateData, CommIoCbParams> Dialer;
-             Dialer dialer(this, &HttpStateData::sendComplete);
-             AsyncCall::Pointer call= asyncCall(11,5, "HttpStateData::SendComplete", dialer);
 -            AsyncCall::Pointer call = JobCallback(11,5,
 -                                                  Dialer, this, HttpStateData::sendComplete);
 -            comm_write(fd, "\r\n", 2, call);
++            AsyncCall::Pointer call = JobCallback(11, 5, Dialer, this, HttpStateData::sendComplete);
 +            comm_write(serverConnection->fd, "\r\n", 2, call);
          }
          return;
      }
diff --cc src/tunnel.cc
index 970388ced4,2f9223d3c1..2f8c05a5cd
--- a/src/tunnel.cc
+++ b/src/tunnel.cc
@@@ -609,11 -640,54 +609,13 @@@ tunnelStart(ClientHttpRequest * http, i
      debugs(26, 3, "tunnelStart: '" << RequestMethodStr(request->method) << " " << url << "'");
      statCounter.server.all.requests++;
      statCounter.server.other.requests++;
 -    /* Create socket. */
 -    Ip::Address temp = getOutgoingAddr(request,NULL);
 -
 -    // if IPv6 is disabled try to force IPv4-only outgoing.
 -    if (!Ip::EnableIpv6 && !temp.SetIPv4()) {
 -        debugs(50, 4, "tunnelStart: IPv6 is Disabled. Tunnel failed from " << temp);
 -        ErrorState *anErr = errorCon(ERR_CONNECT_FAIL, HTTP_SERVICE_UNAVAILABLE, request);
 -        anErr->xerrno = EAFNOSUPPORT;
 -        errorSend(fd, anErr);
 -        return;
 -    }
 -
 -    // if IPv6 is split-stack, prefer IPv4
 -    if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK) {
 -        // NP: This is not a great choice of default,
 -        // but with the current Internet being IPv4-majority has a higher success rate.
 -        // if setting to IPv4 fails we dont care, that just means to use IPv6 outgoing.
 -        temp.SetIPv4();
 -    }
 -
 -    int flags = COMM_NONBLOCKING;
 -    if (request->flags.spoof_client_ip) {
 -        flags |= COMM_TRANSPARENT;
 -    }
 -    sock = comm_openex(SOCK_STREAM,
 -                       IPPROTO_TCP,
 -                       temp,
 -                       flags,
 -                       getOutgoingTOS(request),
 -                       url);
 -
 -    if (sock == COMM_ERROR) {
 -        debugs(26, 4, "tunnelStart: Failed because we're out of sockets.");
 -        err = errorCon(ERR_SOCKET_FAILURE, HTTP_INTERNAL_SERVER_ERROR, request);
 -        *status_ptr = HTTP_INTERNAL_SERVER_ERROR;
 -        err->xerrno = errno;
 -        errorSend(fd, err);
 -        return;
 -    }
  
+     request->hier.peer_local_port = comm_local_port(sock); // for %<lp logging
+ 
      tunnelState = new TunnelStateData;
  #if DELAY_POOLS
 -
      tunnelState->server.setDelayId(DelayId::DelayClient(http));
  #endif
 -
      tunnelState->url = xstrdup(url);
      tunnelState->request = HTTPMSGLOCK(request);
      tunnelState->server.size_ptr = size_ptr;