Refactored ICAP connection-establishing code

... to delay Connection ownership until the ICAP connection is ready.

This change addresses primary Connection ownership concerns (as they
apply to this ICAP code) except orphaning of the temporary Connection
object by helper job start exceptions (now an explicit XXX). For
example, the transaction no longer shares a Connection object with
ConnOpener and IcapPeerConnector jobs.

Also removed (unused except for debugging) Icap::Xaction::stopReason
that was shadowing AsyncJob::stopReason, effectively breaking debugging.

diff --git a/src/adaptation/icap/ModXact.cc b/src/adaptation/icap/ModXact.cc
index a73f32c547e39599c667992b168fad22f706aa23..ed01289c5fb4c272e99838e65c03ffb16c6846c4 100644 (file)
--- a/src/adaptation/icap/ModXact.cc
+++ b/src/adaptation/icap/ModXact.cc
@@ -186,8 +186,7 @@ void Adaptation::Icap::ModXact::startWriting()
     openConnection();
 }
 
-// connection with the ICAP service established
-void Adaptation::Icap::ModXact::handleCommConnected()
+void Adaptation::Icap::ModXact::startShoveling()
 {
     Must(state.writing == State::writingConnect);
 

diff --git a/src/adaptation/icap/ModXact.h b/src/adaptation/icap/ModXact.h
index d6c541bdb3dec45ed3bc78dacec259c0c75ea6cc..06ceb251a4f66f8fb62f183cc0de4ea22d52c3fa 100644 (file)
--- a/src/adaptation/icap/ModXact.h
+++ b/src/adaptation/icap/ModXact.h
@@ -155,10 +155,11 @@ public:
     virtual void noteBodyProductionEnded(BodyPipe::Pointer);
     virtual void noteBodyProducerAborted(BodyPipe::Pointer);
 
-    // comm handlers
-    virtual void handleCommConnected();
+    /* Xaction API */
+    virtual void startShoveling();
     virtual void handleCommWrote(size_t size);
     virtual void handleCommRead(size_t size);
+
     void handleCommWroteHeaders();
     void handleCommWroteBody();
 

diff --git a/src/adaptation/icap/OptXact.cc b/src/adaptation/icap/OptXact.cc
index d463faf31630fbd39aa076faea740c6de92b26b3..24510d2113605088a1a16cf8e082074265384ee8 100644 (file)
--- a/src/adaptation/icap/OptXact.cc
+++ b/src/adaptation/icap/OptXact.cc
@@ -37,7 +37,7 @@ void Adaptation::Icap::OptXact::start()
     openConnection();
 }
 
-void Adaptation::Icap::OptXact::handleCommConnected()
+void Adaptation::Icap::OptXact::startShoveling()
 {
     scheduleRead();
 

diff --git a/src/adaptation/icap/OptXact.h b/src/adaptation/icap/OptXact.h
index 6acc531257760f56c4790d8e50930768c10bdcfb..ea2fc374bcb9fc94cad7a93cde7980971bac788f 100644 (file)
--- a/src/adaptation/icap/OptXact.h
+++ b/src/adaptation/icap/OptXact.h
@@ -30,8 +30,9 @@ public:
     OptXact(ServiceRep::Pointer &aService);
 
 protected:
+    /* Xaction API */
     virtual void start();
-    virtual void handleCommConnected();
+    virtual void startShoveling();
     virtual void handleCommWrote(size_t size);
     virtual void handleCommRead(size_t size);
 

diff --git a/src/adaptation/icap/ServiceRep.cc b/src/adaptation/icap/ServiceRep.cc
index 0a6cc18242149d3a2b6a6ea3282858bda5c552e1..6d46d08f0204e864005f3cd9b53e91eb2256ea46 100644 (file)
--- a/src/adaptation/icap/ServiceRep.cc
+++ b/src/adaptation/icap/ServiceRep.cc
@@ -112,9 +112,8 @@ void Adaptation::Icap::ServiceRep::noteFailure()
     // should be configurable.
 }
 
-// returns a persistent or brand new connection; negative int on failures
 Comm::ConnectionPointer
-Adaptation::Icap::ServiceRep::getConnection(bool retriableXact, bool &reused)
+Adaptation::Icap::ServiceRep::getIdleConnection(const bool retriableXact)
 {
     Comm::ConnectionPointer connection;
 
@@ -137,7 +136,6 @@ Adaptation::Icap::ServiceRep::getConnection(bool retriableXact, bool &reused)
     else
         theIdleConns->closeN(1);
 
-    reused = Comm::IsConnOpen(connection);
     ++theBusyConns;
     debugs(93,3, HERE << "got connection: " << connection);
     return connection;
@@ -150,7 +148,6 @@ void Adaptation::Icap::ServiceRep::putConnection(const Comm::ConnectionPointer &
     // do not pool an idle connection if we owe connections
     if (isReusable && excessConnections() == 0) {
         debugs(93, 3, HERE << "pushing pconn" << comment);
-        commUnsetConnTimeout(conn);
         theIdleConns->push(conn);
     } else {
         debugs(93, 3, HERE << (sendReset ? "RST" : "FIN") << "-closing " <<

diff --git a/src/adaptation/icap/ServiceRep.h b/src/adaptation/icap/ServiceRep.h
index fd1e12096e31fc68c9f8c69ed7a84a2cc85f70c2..dde6fd6facd1107de0420ad008e0df32535e88f5 100644 (file)
--- a/src/adaptation/icap/ServiceRep.h
+++ b/src/adaptation/icap/ServiceRep.h
@@ -85,7 +85,8 @@ public:
     bool wantsPreview(const SBuf &urlPath, size_t &wantedSize) const;
     bool allows204() const;
     bool allows206() const;
-    Comm::ConnectionPointer getConnection(bool isRetriable, bool &isReused);
+    /// \returns an idle persistent ICAP connection or nil
+    Comm::ConnectionPointer getIdleConnection(bool isRetriable);
     void putConnection(const Comm::ConnectionPointer &conn, bool isReusable, bool sendReset, const char *comment);
     void noteConnectionUse(const Comm::ConnectionPointer &conn);
     void noteConnectionFailed(const char *comment);

diff --git a/src/adaptation/icap/Xaction.cc b/src/adaptation/icap/Xaction.cc
index fec4b12c19e7e2f2a63b9efa66ed085d1e023ff3..3a6bc94df894762aa4611fd8eacdba7704752361 100644 (file)
--- a/src/adaptation/icap/Xaction.cc
+++ b/src/adaptation/icap/Xaction.cc
@@ -80,17 +80,12 @@ Adaptation::Icap::Xaction::Xaction(const char *aTypeName, Adaptation::Icap::Serv
     icapRequest(NULL),
     icapReply(NULL),
     attempts(0),
-    connection(NULL),
     theService(aService),
     commEof(false),
     reuseConnection(true),
     isRetriable(true),
     isRepeatable(true),
     ignoreLastWrite(false),
-    stopReason(NULL),
-    reader(NULL),
-    writer(NULL),
-    closer(NULL),
     alep(new AccessLogEntry),
     al(*alep)
 {
@@ -164,11 +159,8 @@ Adaptation::Icap::Xaction::openConnection()
     if (!TheConfig.reuse_connections)
         disableRetries(); // this will also safely drain pconn pool
 
-    bool wasReused = false;
-    connection = s.getConnection(isRetriable, wasReused);
-
-    if (wasReused && Comm::IsConnOpen(connection)) {
-        successfullyConnected();
+    if (const auto pconn = s.getIdleConnection(isRetriable)) {
+        useTransportConnection(pconn);
         return;
     }
 
@@ -197,16 +189,15 @@ Adaptation::Icap::Xaction::dnsLookupDone(const ipcache_addrs *ia)
         return;
     }
 
-    // XXX: Do not save this half-baked connection. Just pass it to ConnOpener.
-    connection = new Comm::Connection;
-    connection->remote = ia->current();
-    connection->remote.port(s.cfg().port);
-    getOutgoingAddress(NULL, connection);
+    const Comm::ConnectionPointer conn = new Comm::Connection();
+    conn->remote = ia->current();
+    conn->remote.port(s.cfg().port);
+    getOutgoingAddress(nullptr, conn);
 
     // TODO: service bypass status may differ from that of a transaction
     typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> ConnectDialer;
     AsyncCall::Pointer callback = JobCallback(93, 3, ConnectDialer, this, Adaptation::Icap::Xaction::noteCommConnected);
-    const auto cs = new Comm::ConnOpener(connection, callback, TheConfig.connect_timeout(service().cfg().bypass));
+    const auto cs = new Comm::ConnOpener(conn, callback, TheConfig.connect_timeout(service().cfg().bypass));
     cs->setHost(s.cfg().host.termedBuf());
     connWait.start(cs, callback);
     AsyncJob::Start(cs);
@@ -235,6 +226,8 @@ void Adaptation::Icap::Xaction::closeConnection()
             closer = NULL;
         }
 
+        commUnsetConnTimeout(connection);
+
         cancelRead(); // may not work
 
         if (reuseConnection && !doneWithIo()) {
@@ -263,63 +256,62 @@ void Adaptation::Icap::Xaction::noteCommConnected(const CommConnectCbParams &io)
 {
     connWait.finish();
 
-    if (io.flag == Comm::TIMEOUT) {
-        handleCommTimedout();
-        return;
-    }
-
     if (io.flag != Comm::OK) {
         dieOnConnectionFailure(); // throws
         return;
     }
 
-    // Finalize the details and start owning the supplied connection, possibly
-    // prematurely -- see XXX in successfullyConnected().
-    assert(io.conn);
-    assert(connection);
-    assert(!connection->isOpen());
-    connection = io.conn;
-    successfullyConnected();
+    useTransportConnection(io.conn);
 }
 
-/// called when successfully connected to an ICAP service
+/// React to the availability of a transport connection to the ICAP service.
+/// The given connection may (or may not) be secured already.
 void
-Adaptation::Icap::Xaction::successfullyConnected()
+Adaptation::Icap::Xaction::useTransportConnection(const Comm::ConnectionPointer &conn)
 {
-    assert(Comm::IsConnOpen(connection));
-
-    // XXX: We should not set timeout and closure handlers here if we are going
-    // to negotiate TLS. Only Ssl::IcapPeerConnector should own the connection.
-
-    typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommTimeoutCbParams> TimeoutDialer;
-    AsyncCall::Pointer timeoutCall =  asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommTimedout",
-                                      TimeoutDialer(this,&Adaptation::Icap::Xaction::noteCommTimedout));
-    commSetConnTimeout(connection, TheConfig.connect_timeout(service().cfg().bypass), timeoutCall);
-
-    typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommCloseCbParams> CloseDialer;
-    closer =  asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommClosed",
-                        CloseDialer(this,&Adaptation::Icap::Xaction::noteCommClosed));
-    comm_add_close_handler(connection->fd, closer);
+    assert(Comm::IsConnOpen(conn));
+    assert(!connection);
 
     // If it is a reused connection and the TLS object is built
     // we should not negotiate new TLS session
-    const auto &ssl = fd_table[connection->fd].ssl;
+    const auto &ssl = fd_table[conn->fd].ssl;
     if (!ssl && service().cfg().secure.encryptTransport) {
+        // XXX: Exceptions orphan conn.
         CbcPointer<Adaptation::Icap::Xaction> me(this);
         AsyncCall::Pointer callback = asyncCall(93, 4, "Adaptation::Icap::Xaction::handleSecuredPeer",
                                                 MyIcapAnswerDialer(me, &Adaptation::Icap::Xaction::handleSecuredPeer));
 
-        const auto sslConnector = new Ssl::IcapPeerConnector(theService, connection, callback, masterLogEntry(), TheConfig.connect_timeout(service().cfg().bypass));
+        const auto sslConnector = new Ssl::IcapPeerConnector(theService, conn, callback, masterLogEntry(), TheConfig.connect_timeout(service().cfg().bypass));
 
         encryptionWait.start(sslConnector, callback);
         AsyncJob::Start(sslConnector); // will call our callback
         return;
     }
 
-// ??    fd_table[io.conn->fd].noteUse(icapPconnPool);
+    useIcapConnection(conn);
+}
+
+/// react to the availability of a fully-ready ICAP connection
+void
+Adaptation::Icap::Xaction::useIcapConnection(const Comm::ConnectionPointer &conn)
+{
+    assert(!connection);
+    assert(conn);
+    assert(Comm::IsConnOpen(conn));
+    connection = conn;
     service().noteConnectionUse(connection);
 
-    handleCommConnected();
+    typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommTimeoutCbParams> TimeoutDialer;
+    AsyncCall::Pointer timeoutCall =  asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommTimedout",
+                                      TimeoutDialer(this,&Adaptation::Icap::Xaction::noteCommTimedout));
+    commSetConnTimeout(connection, TheConfig.connect_timeout(service().cfg().bypass), timeoutCall);
+
+    typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommCloseCbParams> CloseDialer;
+    closer =  asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommClosed",
+                        CloseDialer(this,&Adaptation::Icap::Xaction::noteCommClosed));
+    comm_add_close_handler(connection->fd, closer);
+
+    startShoveling();
 }
 
 void Adaptation::Icap::Xaction::dieOnConnectionFailure()
@@ -363,36 +355,21 @@ void Adaptation::Icap::Xaction::noteCommWrote(const CommIoCbParams &io)
 
 // communication timeout with the ICAP service
 void Adaptation::Icap::Xaction::noteCommTimedout(const CommTimeoutCbParams &)
-{
-    handleCommTimedout();
-}
-
-void Adaptation::Icap::Xaction::handleCommTimedout()
 {
     debugs(93, 2, HERE << typeName << " failed: timeout with " <<
            theService->cfg().methodStr() << " " <<
            theService->cfg().uri << status());
     reuseConnection = false;
-    const auto whileConnecting = bool(connWait);
-    if (whileConnecting) {
-        assert(!haveConnection());
-        theService->noteConnectionFailed("timedout");
-    } else
-        closeConnection(); // so that late Comm callbacks do not disturb bypass
-    throw TexcHere(whileConnecting ?
-                   "timed out while connecting to the ICAP service" :
-                   "timed out while talking to the ICAP service");
+    assert(haveConnection());
+    theService->noteConnectionFailed("timedout");
+    closeConnection();
+    throw TextException("timed out while talking to the ICAP service", Here());
 }
 
 // unexpected connection close while talking to the ICAP service
 void Adaptation::Icap::Xaction::noteCommClosed(const CommCloseCbParams &)
 {
     closer = NULL;
-    handleCommClosed();
-}
-
-void Adaptation::Icap::Xaction::handleCommClosed()
-{
     detailError(ERR_DETAIL_ICAP_XACT_CLOSE);
     mustStop("ICAP service connection externally closed");
 }
@@ -597,7 +574,7 @@ void Adaptation::Icap::Xaction::setOutcome(const Adaptation::Icap::XactOutcome &
 void Adaptation::Icap::Xaction::swanSong()
 {
     // kids should sing first and then call the parent method.
-    if (connWait) {
+    if (connWait || encryptionWait) {
         service().noteConnectionFailed("abort");
     }
 
@@ -735,17 +712,11 @@ Adaptation::Icap::Xaction::handleSecuredPeer(Security::EncryptorAnswer &answer)
 {
     encryptionWait.finish();
 
-    if (closer != NULL) {
-        if (Comm::IsConnOpen(answer.conn))
-            comm_remove_close_handler(answer.conn->fd, closer);
-        else
-            closer->cancel("securing completed");
-        closer = NULL;
-    }
-
     if (answer.error.get()) {
+        // XXX: Security::PeerConnector should do that for negative answers instead.
         if (answer.conn != NULL)
             answer.conn->close();
+        // TODO: Refactor dieOnConnectionFailure() to be usable here as well.
         debugs(93, 2, typeName <<
                " TLS negotiation to " << service().cfg().uri << " failed");
         service().noteConnectionFailed("failure");
@@ -755,8 +726,7 @@ Adaptation::Icap::Xaction::handleSecuredPeer(Security::EncryptorAnswer &answer)
 
     debugs(93, 5, "TLS negotiation to " << service().cfg().uri << " complete");
 
-    service().noteConnectionUse(answer.conn);
-
-    handleCommConnected();
+    // XXX: answer.conn could be closing here. Missing a syncWithComm equivalent?
+    useIcapConnection(answer.conn);
 }
 

diff --git a/src/adaptation/icap/Xaction.h b/src/adaptation/icap/Xaction.h
index 88181c28557cdb7cbce6d9c3e7c5740a8a2e34e6..3e340349f732de58e723417ee6f20572868241ab 100644 (file)
--- a/src/adaptation/icap/Xaction.h
+++ b/src/adaptation/icap/Xaction.h
@@ -69,12 +69,12 @@ protected:
     virtual void start();
     virtual void noteInitiatorAborted(); // TODO: move to Adaptation::Initiate
 
+    /// starts sending/receiving ICAP messages
+    virtual void startShoveling() = 0;
+
     // comm hanndlers; called by comm handler wrappers
-    virtual void handleCommConnected() = 0;
     virtual void handleCommWrote(size_t sz) = 0;
     virtual void handleCommRead(size_t sz) = 0;
-    virtual void handleCommTimedout();
-    virtual void handleCommClosed();
 
     void handleSecuredPeer(Security::EncryptorAnswer &answer);
     /// record error detail if possible
@@ -82,7 +82,6 @@ protected:
 
     void openConnection();
     void closeConnection();
-    void dieOnConnectionFailure();
     bool haveConnection() const;
 
     void scheduleRead();
@@ -128,12 +127,13 @@ public:
     ServiceRep &service();
 
 private:
-    void successfullyConnected();
+    void useTransportConnection(const Comm::ConnectionPointer &);
+    void useIcapConnection(const Comm::ConnectionPointer &);
+    void dieOnConnectionFailure();
     void tellQueryAborted();
     void maybeLog();
 
 protected:
-    Comm::ConnectionPointer connection;     ///< ICAP server connection
     Adaptation::Icap::ServiceRep::Pointer theService;
 
     SBuf readBuf;
@@ -143,11 +143,8 @@ protected:
     bool isRepeatable; ///< can repeat if no or unsatisfactory response
     bool ignoreLastWrite;
 
-    const char *stopReason;
-
     AsyncCall::Pointer reader;
     AsyncCall::Pointer writer;
-    AsyncCall::Pointer closer;
 
     AccessLogEntry::Pointer alep; ///< icap.log entry
     AccessLogEntry &al; ///< short for *alep
@@ -162,6 +159,11 @@ private:
 
     /// encrypts an established transport connection
     JobWait<Ssl::IcapPeerConnector> encryptionWait;
+
+    /// open and, if necessary, secured connection to the ICAP server (or nil)
+    Comm::ConnectionPointer connection;
+
+    AsyncCall::Pointer closer;
 };
 
 } // namespace Icap