Re: [PATCH] squid 5.4: Latest patch - Bug #5055 - from upstream

Reviewed-by: Peter Müller peter.mueller@ipfire.org

Please note: One day after squid 5.4 was released, this extensive patch followed which I publish separately. It seems to be important - perhaps we should consider taking this one in account.

Building and testing was ok - so far no problems or crashes occurred. Running normal.

Signed-off-by: Matthias Fischer matthias.fischer@ipfire.org

lfs/squid | 1 + ...estinationsEnd_exception_opening_967.patch | 3825 +++++++++++++++++ 2 files changed, 3826 insertions(+) create mode 100644 src/patches/squid/02_Bug_5055_FATAL_FwdState_noteDestinationsEnd_exception_opening_967.patch

diff --git a/lfs/squid b/lfs/squid index 034f4a3e9..6762a0218 100644 --- a/lfs/squid +++ b/lfs/squid @@ -78,6 +78,7 @@ $(TARGET) : $(patsubst %,$(DIR_DL)/%,$(objects)) @rm -rf $(DIR_APP) && cd $(DIR_SRC) && tar xaf $(DIR_DL)/$(DL_FILE)

cd $(DIR_APP) && patch -Np1 < $(DIR_SRC)/src/patches/squid/01_squid-gcc11.patch

• cd $(DIR_APP) && patch -Np1 < $(DIR_SRC)/src/patches/squid/02_Bug_5055_FATAL_FwdState_noteDestinationsEnd_exception_opening_967.patch

  cd $(DIR_APP) && autoreconf -vfi cd $(DIR_APP)/libltdl && autoreconf -vfi

diff --git a/src/patches/squid/02_Bug_5055_FATAL_FwdState_noteDestinationsEnd_exception_opening_967.patch b/src/patches/squid/02_Bug_5055_FATAL_FwdState_noteDestinationsEnd_exception_opening_967.patch new file mode 100644 index 000000000..9f592c7e8 --- /dev/null +++ b/src/patches/squid/02_Bug_5055_FATAL_FwdState_noteDestinationsEnd_exception_opening_967.patch @@ -0,0 +1,3825 @@ +commit 1332f8d606485b5a2f57277634c2f6f7855bd9a6 (refs/remotes/origin/v5, refs/remotes/github/v5, refs/heads/v5) +Author: Alex Rousskov rousskov@measurement-factory.com +Date: 2022-02-08 03:56:43 -0500

• Bug 5055: FATAL FwdState::noteDestinationsEnd exception: opening (#967)
• • Bug 5055: FATAL FwdState::noteDestinationsEnd exception: opening
• The bug was caused by commit 25b0ce4. Other known symptoms are:

•    assertion failed: store.cc:1793: "isEmpty()"
  

•    assertion failed: FwdState.cc:501: "serverConnection() == conn"
  

•    assertion failed: FwdState.cc:1037: "!opening()"
  

• This change has several overlapping parts. Unfortunately, merging
• individual parts is both difficult and likely to cause crashes.
• ## Part 1: Bug 5055.
• FwdState used to check serverConn to decide whether to open a connection
• to forward the request. Since commit 25b0ce4, a nil serverConn pointer
• no longer implies that a new connection should be opened: FwdState
• helper jobs may already be working on preparing an existing open
• connection (e.g., sending a CONNECT request or negotiating encryption).
• Bad serverConn checks in both FwdState::noteDestination() and
• FwdState::noteDestinationsEnd() methods led to extra connectStart()
• calls creating two conflicting concurrent helper jobs.
• To fix this, we replaced direct serverConn inspection with a
• usingDestination() call which also checks whether we are waiting for a
• helper job. Testing that fix exposed another set of bugs: The helper job
• pointers or in-job connections left stale/set after forwarding failures.
• The changes described below addressed (most of) those problems.
• ## Part 2: Connection establishing helper jobs and their callbacks
• A proper fix for Bug 5055 required answering a difficult question: When
• should a dying job call its callbacks? We only found one answer which
• required cooperation from the job creator and led to the following
• rules:
• • AsyncJob destructors must not call any callbacks.
• • AsyncJob::swanSong() is responsible for async-calling any remaining

•  (i.e. set, uncalled, and uncancelled) callbacks.
  

• • AsyncJob::swanSong() is called (only) for started jobs.
• • AsyncJob destructing sequence should validate that no callbacks remain

•  uncalled for started jobs.
  

• ... where an AsyncJob x is considered "started" if AsyncJob::Start(x)
• has returned without throwing.
• A new JobWait class helps job creators follow these rules while keeping
• track on in-progress helper jobs and killing no-longer-needed helpers.
• Also fixed very similar bugs in tunnel.cc code.
• ## Part 3: ConnOpener fixes
• 1. Many ConnOpener users are written to keep a ConnectionPointer to the

•   destination given to ConnOpener. This means that their connection
  

•   magically opens when ConnOpener successfully connects, before
  

•   ConnOpener has a chance to notify the user about the changes. Having
  

•   multiple concurrent connection owners is always dangerous, and the
  

•   user cannot even have a close handler registered for its now-open
  

•   connection. When something happens to ConnOpener or its answer, the
  

•   user job may be in trouble. Now, ConnOpener callers no longer pass
  

•   Connection objects they own, cloning them as needed. That adjustment
  

•   required adjustment 2:
  

• 2. Refactored ConnOpener users to stop assuming that the answer contains

•   a pointer to their connection object. After adjustment 1 above, it
  

•   does not. HappyConnOpener relied on that assumption quite a bit so we
  

•   had to refactor to use two custom callback methods instead of one
  

•   with a complicated if-statement distinguishing prime from spare
  

•   attempts. This refactoring is an overall improvement because it
  

•   simplifies the code. Other ConnOpener users just needed to remove a
  

•   few no longer valid paranoid assertions/Musts.
  

• 3. ConnOpener users were forced to remember to close params.conn when

•   processing negative answers. Some, naturally, forgot, triggering
  

•   warnings about orphaned connections (e.g., Ident and TcpLogger).
  

•   ConnOpener now closes its open connection before sending a negative
  

•   answer.
  

• 4. ConnOpener would trigger orphan connection warnings if the job ended

•   after opening the connection but without supplying the connection to
  

•   the requestor (e.g., because the requestor has gone away). Now,
  

•   ConnOpener explicitly closes its open connection if it has not been
  

•   sent to the requestor.
  

• Also fixed Comm::ConnOpener::cleanFd() debugging that was incorrectly
• saying that the method closes the temporary descriptor.
• Also fixed ConnOpener callback's syncWithComm(): The stale
• CommConnectCbParams override was testing unused (i.e. always negative)
• CommConnectCbParams::fd and was trying to cancel the callback that most
• (possibly all) recipients rely on: ConnOpener users expect a negative
• answer rather than no answer at all.
• Also, after comparing the needs of two old/existing and a temporary
• added ("clone everything") Connection cloning method callers, we decided
• there is no need to maintain three different methods. All existing
• callers should be fine with a single method because none of them suffers
• from "extra" copying of members that others need. Right now, the new
• cloneProfile() method copies everything except FD and a few
• special-purpose members (with documented reasons for not copying).
• Also added Comm::Connection::cloneDestinationDetails() debugging to
• simplify tracking dependencies between half-baked Connection objects
• carrying destination/flags/other metadata and open Connection objects
• created by ConnOpener using that metadata (which are later delivered to
• ConnOpener users and, in some cases, replace those half-baked
• connections mentioned earlier. Long-term, we need to find a better way
• to express these and other Connection states/stages than comments and
• debugging messages.
• ## Part 4: Comm::Connection closure callbacks
• We improved many closure callbacks to make sure (to the extent possible)
• that Connection and other objects are in sync with Comm. There are lots
• of small bugs, inconsistencies, and other problems in Connection closure
• handlers. It is not clear whether any of those problems could result in
• serious runtime errors or leaks. In theory, the rest of the code could
• neutralize their negative side effects. However, even in that case, it
• was just a matter of time before the next bug will bite us due to stale
• Connection::fd and such. These changes themselves carry elevated risk,
• but they get us closer to reliable code as far as Connection maintenance
• is concerned; otherwise, we will keep chasing their deadly side effects.
• Long-term, all these manual efforts to keep things in sync should become
• unnecessary with the introduction of appropriate Connection ownership
• APIs that automatically maintain the corresponding environments (TODO).
• ## Part 5: Other notable improvements in the adjusted code
• Improved Security::PeerConnector::serverConn and
• Http::Tunneler::connection management, especially when sending negative
• answers. When sending a negative answer, we would set answer().conn to
• an open connection, async-send that answer, and then hurry to close the
• connection using our pointer to the shared Connection object. If
• everything went according to the plan, the recipient would get a non-nil
• but closed Connection object. Now, a negative answer simply means no
• connection at all. Same for a tunneled answer.
• Refactored ICAP connection-establishing code to to delay Connection
• ownership until the ICAP connection is fully 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.
• Probably fixed a bug where PeerConnector::negotiate() assumed that a
• sslFinalized() does not return true after callBack(). It may (e.g., when
• CertValidationHelper::Submit() throws). Same for
• PeekingPeerConnector::checkForPeekAndSpliceMatched().
• Fixed FwdState::advanceDestination() bug that did not save
• ERR_GATEWAY_FAILURE details and "lost" the address of that failed
• destination, making it unavailable to future retries (if any).
• Polished PeerPoolMgr, Ident, and Gopher code to be able to fix similar
• job callback and connection management issues.
• Polished AsyncJob::Start() API. Start() used to return a job pointer,
• but that was a bad idea:
• • It implies that a failed Start() will return a nil pointer, and that

•  the caller should check the result. Neither is true.
  

• • It encourages callers to dereference the returned pointer to further

•  adjust the job. That technically works (today) but violates the rules
  

•  of communicating with an async job. The Start() method is the boundary
  

•  after which the job is deemed asynchronous.
  

• Also removed old "and wait for..." post-Start() comments because the
• code itself became clear enough, and those comments were becoming
• increasingly stale (because they duplicated the callback names above
• them).
• Fix Tunneler and PeerConnector handling of last-resort callback
• requirements. Events like handleStopRequest() and callException() stop
• the job but should not be reported as a BUG (e.g., it would be up to the
• callException() to decide how to report the caught exception). There
• might (or there will) be other, similar cases where the job is stopped
• prematurely for some non-BUG reason beyond swanSong() knowledge. The
• existence of non-bug cases does not mean there could be no bugs worth
• reporting here, but until they can be identified more reliably than all
• these benign/irrelevant cases, reporting no BUGs is a (much) lesser
• evil.
• TODO: Revise AsyncJob::doneAll(). Many of its overrides are written to
• check for both positive (i.e. mission accomplished) and negative (i.e.
• mission cancelled or cannot be accomplished) conditions, but the latter
• is usually unnecessary, especially after we added handleStopRequest()
• API to properly support external job cancellation events. Many doneAll()
• overrides can probably be greatly simplified.

• ---

• Cherry picked SQUID-568-premature-serverconn-use-v5 commit 22b5f78.
• • fixup: Cherry-picked SQUID-568-premature-serverconn-use PR-time fixes
• In git log order:
• • e64a6c1: Undone an out-of-scope change and added a missing 'auto'
• • aeaf83d: Fixed an 'unused parameter' error
• • f49d009: fixup: No explicit destructors with implicit copying methods
• • c30c37f: Removed an unnecessary explicit copy constructor
• • 012f5ec: Excluded Connection::rfc931 from cloning
• • 366c78a: ICAP: do not set connect_timeout on the established conn...
• This branch is now in sync with SQUID-568-premature-serverconn-use (S)
• commit e64a6c1 (except for official changes merged from master/v6 into S
• closer to the end of PR 877 work (i.e. S' merge commit 0a7432a).
• • Fix FATAL ServiceRep::putConnection exception: theBusyConns > 0

•    FATAL: check failed: theBusyConns > 0
  

•        exception location: ServiceRep.cc(163) putConnection
  

• Since master/v6 commit 2b6b1bc, a timeout on a ready-to-shovel
• Squid-service ICAP connection was decrementing theBusyConns level one
• extra time because Adaptation::Icap::Xaction::noteCommTimedout() started
• calling both noteConnectionFailed() and closeConnection(). Depending on
• the actual theBusyConns level, the extra decrement could result in FATAL
• errors later, when putConnection() was called (for a different ICAP
• transaction) with zero theBusyConns in an exception-unprotected context.
• Throughout these changes, Xaction still counts the above timeouts as a
• service failure. That is done by calling ServiceRep::noteFailure() from
• Xaction::callException(), including in timeout cases described above.

• ---

• Cherry-picked master/v6 commit a8ac892.

+diff --git a/src/BodyPipe.cc b/src/BodyPipe.cc +index 9abe7d48b..37a3b32cd 100644 +--- a/src/BodyPipe.cc ++++ b/src/BodyPipe.cc +@@ -335,6 +335,7 @@ BodyPipe::startAutoConsumptionIfNeeded()

•     return;
  

• theConsumer = new BodySink(this);
  

++ AsyncJob::Start(theConsumer);

• debugs(91,7, HERE << "starting auto consumption" << status());
  

• scheduleBodyDataNotification();
  

• }

+diff --git a/src/CommCalls.cc b/src/CommCalls.cc +index 4cd503dcb..cca23f6e3 100644 +--- a/src/CommCalls.cc ++++ b/src/CommCalls.cc +@@ -53,6 +53,9 @@ CommAcceptCbParams::CommAcceptCbParams(void *aData):

• {
• }

++// XXX: Add CommAcceptCbParams::syncWithComm(). Adjust syncWithComm() API if all ++// implementations always return true. ++

• void
• CommAcceptCbParams::print(std::ostream &os) const
• {

+@@ -72,13 +75,24 @@ CommConnectCbParams::CommConnectCbParams(void *aData):

• bool
• CommConnectCbParams::syncWithComm()
• {

+- // drop the call if the call was scheduled before comm_close but +- // is being fired after comm_close +- if (fd >= 0 && fd_table[fd].closing()) { +- debugs(5, 3, HERE << "dropping late connect call: FD " << fd); +- return false; ++ assert(conn); ++ ++ // change parameters if this is a successful callback that was scheduled ++ // after its Comm-registered connection started to close ++ ++ if (flag != Comm::OK) { ++ assert(!conn->isOpen()); ++ return true; // not a successful callback; cannot go out of sync

• }
  

+- return true; // now we are in sync and can handle the call ++ ++ assert(conn->isOpen()); ++ if (!fd_table[conn->fd].closing()) ++ return true; // no closing in progress; in sync (for now) ++ ++ debugs(5, 3, "converting to Comm::ERR_CLOSING: " << conn); ++ conn->noteClosure(); ++ flag = Comm::ERR_CLOSING; ++ return true; // now the callback is in sync with Comm again

• }
• /* CommIoCbParams */

+diff --git a/src/Downloader.cc b/src/Downloader.cc +index 298c3a836..012387fb8 100644 +--- a/src/Downloader.cc ++++ b/src/Downloader.cc +@@ -81,6 +81,10 @@ void

• Downloader::swanSong()
• {

• debugs(33, 6, this);
  

++ ++ if (callback_) // job-ending emergencies like handleStopRequest() or callException() ++ callBack(Http::scInternalServerError); ++

• if (context_) {
  

•     context_->finished();
  

•     context_ = nullptr;
  

+@@ -251,6 +255,7 @@ Downloader::downloadFinished()

• void
• Downloader::callBack(Http::StatusCode const statusCode)
• {

++ assert(callback_);

• CbDialer *dialer = dynamic_cast<CbDialer*>(callback_->getDialer());
  

• Must(dialer);
  

• dialer->status = statusCode;
  

+diff --git a/src/FwdState.cc b/src/FwdState.cc +index b721017c3..b69e60c7c 100644 +--- a/src/FwdState.cc ++++ b/src/FwdState.cc +@@ -207,26 +207,22 @@ FwdState::stopAndDestroy(const char *reason)

• {

• debugs(17, 3, "for " << reason);
  

+- if (opening()) +- cancelOpening(reason); ++ cancelStep(reason);

• PeerSelectionInitiator::subscribed = false; // may already be false
  

• self = nullptr; // we hope refcounting destroys us soon; may already be nil
  

• /* do not place any code here as this object may be gone by now */
  

• }

+-/// Notify connOpener that we no longer need connections. We do not have to do +-/// this -- connOpener would eventually notice on its own, but notifying reduces +-/// waste and speeds up spare connection opening for other transactions (that +-/// could otherwise wait for this transaction to use its spare allowance). ++/// Notify a pending subtask, if any, that we no longer need its help. We do not ++/// have to do this -- the subtask job will eventually end -- but ending it ++/// earlier reduces waste and may reduce DoS attack surface.

• void

+-FwdState::cancelOpening(const char *reason) ++FwdState::cancelStep(const char *reason)

• {

+- assert(calls.connector); +- calls.connector->cancel(reason); +- calls.connector = nullptr; +- notifyConnOpener(); +- connOpener.clear(); ++ transportWait.cancel(reason); ++ encryptionWait.cancel(reason); ++ peerWait.cancel(reason);

• }
• #if STRICT_ORIGINAL_DST

+@@ -348,8 +344,7 @@ FwdState::~FwdState()

• entry = NULL;
  

+- if (opening()) +- cancelOpening("~FwdState"); ++ cancelStep("~FwdState");

• if (Comm::IsConnOpen(serverConn))
  

•     closeServerConnection("~FwdState");
  

+@@ -501,8 +496,17 @@ FwdState::fail(ErrorState * errorState)

• if (!errorState->request)
  

•     errorState->request = request;
  

+- if (err->type != ERR_ZERO_SIZE_OBJECT) +- return; ++ if (err->type == ERR_ZERO_SIZE_OBJECT) ++ reactToZeroSizeObject(); ++ ++ destinationReceipt = nullptr; // may already be nil ++} ++ ++/// ERR_ZERO_SIZE_OBJECT requires special adjustments ++void ++FwdState::reactToZeroSizeObject() ++{ ++ assert(err->type == ERR_ZERO_SIZE_OBJECT);

• if (pconnRace == racePossible) {
  

•     debugs(17, 5, HERE << "pconn race happened");
  

+@@ -566,6 +570,8 @@ FwdState::complete()

•     if (Comm::IsConnOpen(serverConn))
  

•         unregister(serverConn);
  

++ serverConn = nullptr; ++ destinationReceipt = nullptr;

•     storedWholeReply_ = nullptr;
  

•     entry->reset();
  

+@@ -584,6 +590,12 @@ FwdState::complete()

• }
  

• }

++bool ++FwdState::usingDestination() const ++{ ++ return encryptionWait || peerWait || Comm::IsConnOpen(serverConn); ++} ++

• void
• FwdState::markStoredReplyAsWhole(const char * const whyWeAreSure)
• {

+@@ -613,19 +625,19 @@ FwdState::noteDestination(Comm::ConnectionPointer path)

• destinations->addPath(path);
  

+- if (Comm::IsConnOpen(serverConn)) { ++ if (usingDestination()) {

•     // We are already using a previously opened connection, so we cannot be
  

+- // waiting for connOpener. We still receive destinations for backup. +- Must(!opening()); ++ // waiting for it. We still receive destinations for backup. ++ Must(!transportWait);

•     return;
  

• }
  

+- if (opening()) { ++ if (transportWait) {

•     notifyConnOpener();
  

•     return; // and continue to wait for FwdState::noteConnection() callback
  

• }
  

+- // This is the first path candidate we have seen. Create connOpener. ++ // This is the first path candidate we have seen. Use it.

• useDestinations();
  

• }

+@@ -653,19 +665,19 @@ FwdState::noteDestinationsEnd(ErrorState *selectionError)

• // if all of them fail, forwarding as whole will fail
  

• Must(!selectionError); // finding at least one path means selection succeeded
  

+- if (Comm::IsConnOpen(serverConn)) { ++ if (usingDestination()) {

•     // We are already using a previously opened connection, so we cannot be
  

+- // waiting for connOpener. We were receiving destinations for backup. +- Must(!opening()); ++ // waiting for it. We were receiving destinations for backup. ++ Must(!transportWait);

•     return;
  

• }
  

+- Must(opening()); // or we would be stuck with nothing to do or wait for ++ Must(transportWait); // or we would be stuck with nothing to do or wait for

• notifyConnOpener();
  

• // and continue to wait for FwdState::noteConnection() callback
  

• }

+-/// makes sure connOpener knows that destinations have changed ++/// makes sure connection opener knows that the destinations have changed

• void
• FwdState::notifyConnOpener()
• {

+@@ -674,7 +686,7 @@ FwdState::notifyConnOpener()

• } else {
  

•     debugs(17, 7, "notifying about " << *destinations);
  

•     destinations->notificationPending = true;
  

+- CallJobHere(17, 5, connOpener, HappyConnOpener, noteCandidatesChange); ++ CallJobHere(17, 5, transportWait.job(), HappyConnOpener, noteCandidatesChange);

• }
  

• }

+@@ -684,7 +696,7 @@ static void

• fwdServerClosedWrapper(const CommCloseCbParams &params)
• {

• FwdState *fwd = (FwdState *)params.data;
  

+- fwd->serverClosed(params.fd); ++ fwd->serverClosed();

• }
• /**** PRIVATE *****************************************************************/

+@@ -754,13 +766,23 @@ FwdState::checkRetriable()

• }
• void

+-FwdState::serverClosed(int fd) ++FwdState::serverClosed()

• {

+- // XXX: fd is often -1 here +- debugs(17, 2, "FD " << fd << " " << entry->url() << " after " << +- (fd >= 0 ? fd_table[fd].pconn.uses : -1) << " requests"); +- if (fd >= 0 && serverConnection()->fd == fd) +- fwdPconnPool->noteUses(fd_table[fd].pconn.uses); ++ // XXX: This method logic attempts to tolerate Connection::close() called ++ // for serverConn earlier, by one of our dispatch()ed jobs. If that happens, ++ // serverConn will already be closed here or, worse, it will already be open ++ // for the next forwarding attempt. The current code prevents us getting ++ // stuck, but the long term solution is to stop sharing serverConn. ++ debugs(17, 2, serverConn); ++ if (Comm::IsConnOpen(serverConn)) { ++ const auto uses = fd_table[serverConn->fd].pconn.uses; ++ debugs(17, 3, "prior uses: " << uses); ++ fwdPconnPool->noteUses(uses); // XXX: May not have come from fwdPconnPool ++ serverConn->noteClosure(); ++ } ++ serverConn = nullptr; ++ closeHandler = nullptr; ++ destinationReceipt = nullptr;

• retryOrBail();
  

• }

+@@ -802,6 +824,8 @@ FwdState::handleUnregisteredServerEnd()

• {

• debugs(17, 2, HERE << "self=" << self << " err=" << err << ' ' << entry->url());
  

• assert(!Comm::IsConnOpen(serverConn));
  

++ serverConn = nullptr; ++ destinationReceipt = nullptr;

• retryOrBail();
  

• }

+@@ -819,6 +843,8 @@ FwdState::advanceDestination(const char *stepDescription, const Comm::Connection

• } catch (...) {
  

•     debugs (17, 2, "exception while trying to " << stepDescription << ": " << CurrentException);
  

•     closePendingConnection(conn, "connection preparation exception");
  

++ if (!err) ++ fail(new ErrorState(ERR_GATEWAY_FAILURE, Http::scInternalServerError, request, al));

•     retryOrBail();
  

• }
  

• }

+@@ -830,8 +856,7 @@ FwdState::noteConnection(HappyConnOpener::Answer &answer)

• {

• assert(!destinationReceipt);
  

+- calls.connector = nullptr; +- connOpener.clear(); ++ transportWait.finish();

• Must(n_tries <= answer.n_tries); // n_tries cannot decrease
  

• n_tries = answer.n_tries;
  

+@@ -843,6 +868,8 @@ FwdState::noteConnection(HappyConnOpener::Answer &answer)

•     Must(!Comm::IsConnOpen(answer.conn));
  

•     answer.error.clear(); // preserve error for errorSendComplete()
  

• } else if (!Comm::IsConnOpen(answer.conn) || fd_table[answer.conn->fd].closing()) {
  

++ // The socket could get closed while our callback was queued. Sync ++ // Connection. XXX: Connection::fd may already be stale/invalid here.

•     // We do not know exactly why the connection got closed, so we play it
  

•     // safe, allowing retries only for persistent (reused) connections
  

•     if (answer.reused) {
  

+@@ -912,23 +939,24 @@ FwdState::establishTunnelThruProxy(const Comm::ConnectionPointer &conn)

• if (!conn->getPeer()->options.no_delay)
  

•     tunneler->setDelayId(entry->mem_obj->mostBytesAllowed());
  

• #endif

+- AsyncJob::Start(tunneler); +- // and wait for the tunnelEstablishmentDone() call ++ peerWait.start(tunneler, callback);

• }
• /// resumes operations after the (possibly failed) HTTP CONNECT exchange
• void
• FwdState::tunnelEstablishmentDone(Http::TunnelerAnswer &answer)
• {

++ peerWait.finish(); ++

• ErrorState *error = nullptr;
  

• if (!answer.positive()) {
  

+- Must(!Comm::IsConnOpen(answer.conn)); ++ Must(!answer.conn);

•     error = answer.squidError.get();
  

•     Must(error);
  

•     answer.squidError.clear(); // preserve error for fail()
  

• } else if (!Comm::IsConnOpen(answer.conn) || fd_table[answer.conn->fd].closing()) {
  

+- // The socket could get closed while our callback was queued. +- // We close Connection here to sync Connection::fd. ++ // The socket could get closed while our callback was queued. Sync ++ // Connection. XXX: Connection::fd may already be stale/invalid here.

•     closePendingConnection(answer.conn, "conn was closed while waiting for tunnelEstablishmentDone");
  

•     error = new ErrorState(ERR_CANNOT_FORWARD, Http::scServiceUnavailable, request, al);
  

• } else if (!answer.leftovers.isEmpty()) {
  

+@@ -998,18 +1026,21 @@ FwdState::secureConnectionToPeer(const Comm::ConnectionPointer &conn)

• #endif

•     connector = new Security::BlindPeerConnector(requestPointer, conn, callback, al, sslNegotiationTimeout);
  

• connector->noteFwdPconnUse = true;
  

+- AsyncJob::Start(connector); // will call our callback ++ encryptionWait.start(connector, callback);

• }
• /// called when all negotiations with the TLS-speaking peer have been completed
• void
• FwdState::connectedToPeer(Security::EncryptorAnswer &answer)
• {

++ encryptionWait.finish(); ++

• ErrorState *error = nullptr;
  

• if ((error = answer.error.get())) {
  

+- Must(!Comm::IsConnOpen(answer.conn)); ++ assert(!answer.conn);

•     answer.error.clear(); // preserve error for errorSendComplete()
  

• } else if (answer.tunneled) {
  

++ assert(!answer.conn);

•     // TODO: When ConnStateData establishes tunnels, its state changes
  

•     // [in ways that may affect logging?]. Consider informing
  

•     // ConnStateData about our tunnel or otherwise unifying tunnel
  

+@@ -1019,6 +1050,8 @@ FwdState::connectedToPeer(Security::EncryptorAnswer &answer)

•     complete(); // destroys us
  

•     return;
  

• } else if (!Comm::IsConnOpen(answer.conn) || fd_table[answer.conn->fd].closing()) {
  

++ // The socket could get closed while our callback was queued. Sync ++ // Connection. XXX: Connection::fd may already be stale/invalid here.

•     closePendingConnection(answer.conn, "conn was closed while waiting for connectedToPeer");
  

•     error = new ErrorState(ERR_CANNOT_FORWARD, Http::scServiceUnavailable, request, al);
  

• }
  

+@@ -1091,22 +1124,20 @@ FwdState::connectStart()

• Must(!request->pinnedConnection());
  

• assert(!destinations->empty());
  

+- assert(!opening()); ++ assert(!usingDestination());

• // Ditch error page if it was created before.
  

• // A new one will be created if there's another problem
  

• delete err;
  

• err = nullptr;
  

• request->clearError();
  

+- serverConn = nullptr; +- destinationReceipt = nullptr;

• request->hier.startPeerClock();
  

+- calls.connector = asyncCall(17, 5, "FwdState::noteConnection", HappyConnOpener::CbDialer<FwdState>(&FwdState::noteConnection, this)); ++ AsyncCall::Pointer callback = asyncCall(17, 5, "FwdState::noteConnection", HappyConnOpener::CbDialer<FwdState>(&FwdState::noteConnection, this));

• HttpRequest::Pointer cause = request;
  

+- const auto cs = new HappyConnOpener(destinations, calls.connector, cause, start_t, n_tries, al); ++ const auto cs = new HappyConnOpener(destinations, callback, cause, start_t, n_tries, al);

• cs->setHost(request->url.host());
  

• bool retriable = checkRetriable();
  

• if (!retriable && Config.accessList.serverPconnForNonretriable) {
  

+@@ -1118,8 +1149,7 @@ FwdState::connectStart()

• cs->setRetriable(retriable);
  

• cs->allowPersistent(pconnRace != raceHappened);
  

• destinations->notificationPending = true; // start() is async
  

+- connOpener = cs; +- AsyncJob::Start(cs); ++ transportWait.start(cs, callback);

• }
• /// send request on an existing connection dedicated to the requesting client

+diff --git a/src/FwdState.h b/src/FwdState.h +index de75f33cc..ebf5f82b1 100644 +--- a/src/FwdState.h ++++ b/src/FwdState.h +@@ -9,13 +9,12 @@

• #ifndef SQUID_FORWARD_H
• #define SQUID_FORWARD_H

+-#include "base/CbcPointer.h"

• #include "base/forward.h"

++#include "base/JobWait.h"

• #include "base/RefCount.h"
• #include "clients/forward.h"
• #include "comm.h"
• #include "comm/Connection.h"

+-#include "comm/ConnOpener.h"

• #include "error/forward.h"
• #include "fde.h"
• #include "http/StatusCode.h"

+@@ -38,7 +37,6 @@ class ResolvedPeers;

• typedef RefCount<ResolvedPeers> ResolvedPeersPointer;
• class HappyConnOpener;

+-typedef CbcPointer<HappyConnOpener> HappyConnOpenerPointer;

• class HappyConnOpenerAnswer;
• /// Sets initial TOS value and Netfilter for the future outgoing connection.

+@@ -87,7 +85,7 @@ public:

• void handleUnregisteredServerEnd();
  

• int reforward();
  

• bool reforwardableStatus(const Http::StatusCode s) const;
  

+- void serverClosed(int fd); ++ void serverClosed();

• void connectStart();
  

• void connectDone(const Comm::ConnectionPointer & conn, Comm::Flag status, int xerrno);
  

• bool checkRetry();
  

+@@ -115,6 +113,9 @@ private:

• /* PeerSelectionInitiator API */
  

• virtual void noteDestination(Comm::ConnectionPointer conn) override;
  

• virtual void noteDestinationsEnd(ErrorState *selectionError) override;
  

++ /// whether the successfully selected path destination or the established ++ /// server connection is still in use ++ bool usingDestination() const;

• void noteConnection(HappyConnOpenerAnswer &);
  

+@@ -157,13 +158,10 @@ private:

• /// \returns the time left for this connection to become connected or 1 second if it is less than one second left
  

• time_t connectingTimeout(const Comm::ConnectionPointer &conn) const;
  

+- /// whether we are waiting for HappyConnOpener +- /// same as calls.connector but may differ from connOpener.valid() +- bool opening() const { return connOpener.set(); } +- +- void cancelOpening(const char *reason); ++ void cancelStep(const char *reason);

• void notifyConnOpener();
  

++ void reactToZeroSizeObject();

• void updateAleWithFinalError();
  

+@@ -182,11 +180,6 @@ private:

• time_t start_t;
  

• int n_tries; ///< the number of forwarding attempts so far
  

+- // AsyncCalls which we set and may need cancelling. +- struct { +- AsyncCall::Pointer connector; ///< a call linking us to the ConnOpener producing serverConn. +- } calls; +-

• struct {
  

•     bool connected_okay; ///< TCP link ever opened properly. This affects retry of POST,PUT,CONNECT,etc
  

•     bool dont_retry;
  

+@@ -194,7 +187,16 @@ private:

•     bool destinationsFound; ///< at least one candidate path found
  

• } flags;
  

+- HappyConnOpenerPointer connOpener; ///< current connection opening job ++ /// waits for a transport connection to the peer to be established/opened ++ JobWait<HappyConnOpener> transportWait; ++ ++ /// waits for the established transport connection to be secured/encrypted ++ JobWaitSecurity::PeerConnector encryptionWait; ++ ++ /// waits for an HTTP CONNECT tunnel through a cache_peer to be negotiated ++ /// over the (encrypted, if needed) transport connection to that cache_peer ++ JobWaitHttp::Tunneler peerWait; ++

• ResolvedPeersPointer destinations; ///< paths for forwarding the request
  

• Comm::ConnectionPointer serverConn; ///< a successfully opened connection to a server.
  

• PeerConnectionPointer destinationReceipt; ///< peer selection result (or nil)
  

+diff --git a/src/HappyConnOpener.cc b/src/HappyConnOpener.cc +index 6e0872506..6d83ff14f 100644 +--- a/src/HappyConnOpener.cc ++++ b/src/HappyConnOpener.cc +@@ -18,6 +18,7 @@

• #include "neighbors.h"
• #include "pconn.h"
• #include "PeerPoolMgr.h"

++#include "sbuf/Stream.h"

• #include "SquidConfig.h"
• CBDATA_CLASS_INIT(HappyConnOpener);

+@@ -330,6 +331,8 @@ HappyConnOpener::HappyConnOpener(const ResolvedPeers::Pointer &dests, const Asyn

• fwdStart(aFwdStart),
  

• callback_(aCall),
  

• destinations(dests),
  

++ prime(&HappyConnOpener::notePrimeConnectDone, "HappyConnOpener::notePrimeConnectDone"), ++ spare(&HappyConnOpener::noteSpareConnectDone, "HappyConnOpener::noteSpareConnectDone"),

• ale(anAle),
  

• cause(request),
  

• n_tries(tries)
  

+@@ -410,33 +413,43 @@ HappyConnOpener::swanSong()

• AsyncJob::swanSong();
  

• }

++/// HappyConnOpener::Attempt printer for debugging ++std::ostream & ++operator <<(std::ostream &os, const HappyConnOpener::Attempt &attempt) ++{ ++ if (!attempt.path) ++ os << '-'; ++ else if (attempt.path->isOpen()) ++ os << "FD " << attempt.path->fd; ++ else if (attempt.connWait) ++ os << attempt.connWait; ++ else // destination is known; connection closed (and we are not opening any) ++ os << attempt.path->id; ++ return os; ++} ++

• const char *
• HappyConnOpener::status() const
• {

+- static MemBuf buf; +- buf.reset(); ++ // TODO: In a redesigned status() API, the caller may mimic this approach. ++ static SBuf buf; ++ buf.clear();

+- buf.append(" [", 2); ++ SBufStream os(buf); ++ ++ os.write(" [", 2);

• if (stopReason)
  

+- buf.appendf("Stopped, reason:%s", stopReason); +- if (prime) { +- if (prime.path && prime.path->isOpen()) +- buf.appendf(" prime FD %d", prime.path->fd); +- else if (prime.connector) +- buf.appendf(" prime call%ud", prime.connector->id.value); +- } +- if (spare) { +- if (spare.path && spare.path->isOpen()) +- buf.appendf(" spare FD %d", spare.path->fd); +- else if (spare.connector) +- buf.appendf(" spare call%ud", spare.connector->id.value); +- } ++ os << "Stopped:" << stopReason; ++ if (prime) ++ os << "prime:" << prime; ++ if (spare) ++ os << "spare:" << spare;

• if (n_tries)
  

+- buf.appendf(" tries %d", n_tries); +- buf.appendf(" %s%u]", id.prefix(), id.value); +- buf.terminate(); ++ os << " tries:" << n_tries; ++ os << ' ' << id << ']';

+- return buf.content(); ++ buf = os.buf(); ++ return buf.c_str();

• }
• /// Create "503 Service Unavailable" or "504 Gateway Timeout" error depending

+@@ -516,7 +529,7 @@ void

• HappyConnOpener::startConnecting(Attempt &attempt, PeerConnectionPointer &dest)
• {

• Must(!attempt.path);
  

+- Must(!attempt.connector); ++ Must(!attempt.connWait);

• Must(dest);
  

• const auto bumpThroughPeer = cause->flags.sslBumped && dest->getPeer();
  

+@@ -552,51 +565,52 @@ HappyConnOpener::openFreshConnection(Attempt &attempt, PeerConnectionPointer &de

• entry->mem_obj->checkUrlChecksum();
  

• #endif

+- GetMarkingsToServer(cause.getRaw(), *dest); ++ const auto conn = dest->cloneProfile(); ++ GetMarkingsToServer(cause.getRaw(), *conn);

+- // ConnOpener modifies its destination argument so we reset the source port +- // in case we are reusing the destination already used by our predecessor. +- dest->local.port(0);

• ++n_tries;
  

• typedef CommCbMemFunT<HappyConnOpener, CommConnectCbParams> Dialer;
  

+- AsyncCall::Pointer callConnect = JobCallback(48, 5, Dialer, this, HappyConnOpener::connectDone); ++ AsyncCall::Pointer callConnect = asyncCall(48, 5, attempt.callbackMethodName, ++ Dialer(this, attempt.callbackMethod));

• const time_t connTimeout = dest->connectTimeout(fwdStart);
  

+- Comm::ConnOpener *cs = new Comm::ConnOpener(dest, callConnect, connTimeout); +- if (!dest->getPeer()) ++ auto cs = new Comm::ConnOpener(conn, callConnect, connTimeout); ++ if (!conn->getPeer())

•     cs->setHost(host_);
  

+- attempt.path = dest; +- attempt.connector = callConnect; +- attempt.opener = cs; ++ attempt.path = dest; // but not the being-opened conn! ++ attempt.connWait.start(cs, callConnect); ++}

+- AsyncJob::Start(cs); ++/// Comm::ConnOpener callback for the prime connection attempt ++void ++HappyConnOpener::notePrimeConnectDone(const CommConnectCbParams &params) ++{ ++ handleConnOpenerAnswer(prime, params, "new prime connection");

• }

+-/// called by Comm::ConnOpener objects after a prime or spare connection attempt +-/// completes (successfully or not) ++/// Comm::ConnOpener callback for the spare connection attempt

• void

+-HappyConnOpener::connectDone(const CommConnectCbParams &params) ++HappyConnOpener::noteSpareConnectDone(const CommConnectCbParams &params)

• {

+- Must(params.conn); +- const bool itWasPrime = (params.conn == prime.path); +- const bool itWasSpare = (params.conn == spare.path); +- Must(itWasPrime != itWasSpare); +- +- PeerConnectionPointer handledPath; +- if (itWasPrime) { +- handledPath = prime.path; +- prime.finish(); +- } else { +- handledPath = spare.path; +- spare.finish(); +- if (gotSpareAllowance) { +- TheSpareAllowanceGiver.jobUsedAllowance(); +- gotSpareAllowance = false; +- } ++ if (gotSpareAllowance) { ++ TheSpareAllowanceGiver.jobUsedAllowance(); ++ gotSpareAllowance = false;

• }
  

++ handleConnOpenerAnswer(spare, params, "new spare connection"); ++} ++ ++/// prime/spare-agnostic processing of a Comm::ConnOpener result ++void ++HappyConnOpener::handleConnOpenerAnswer(Attempt &attempt, const CommConnectCbParams &params, const char *what) ++{ ++ Must(params.conn); ++ ++ // finalize the previously selected path before attempt.finish() forgets it ++ auto handledPath = attempt.path; ++ handledPath.finalize(params.conn); // closed on errors ++ attempt.finish();

+- const char *what = itWasPrime ? "new prime connection" : "new spare connection";

• if (params.flag == Comm::OK) {
  

•     sendSuccess(handledPath, false, what);
  

•     return;
  

+@@ -605,7 +619,6 @@ HappyConnOpener::connectDone(const CommConnectCbParams &params)

• debugs(17, 8, what << " failed: " << params.conn);
  

• if (const auto peer = params.conn->getPeer())
  

•     peerConnectFailed(peer);
  

+- params.conn->close(); // TODO: Comm::ConnOpener should do this instead.

• // remember the last failure (we forward it if we cannot connect anywhere)
  

• lastFailedConnection = handledPath;
  

+@@ -881,13 +894,23 @@ HappyConnOpener::ranOutOfTimeOrAttempts() const

• return false;
  

• }

++HappyConnOpener::Attempt::Attempt(const CallbackMethod method, const char *methodName): ++ callbackMethod(method), ++ callbackMethodName(methodName) ++{ ++} ++ ++void ++HappyConnOpener::Attempt::finish() ++{ ++ connWait.finish(); ++ path = nullptr; ++} ++

• void
• HappyConnOpener::Attempt::cancel(const char *reason)
• {

+- if (connector) { +- connector->cancel(reason); +- CallJobHere(17, 3, opener, Comm::ConnOpener, noteAbort); +- } +- clear(); ++ connWait.cancel(reason); ++ path = nullptr;

• }

+diff --git a/src/HappyConnOpener.h b/src/HappyConnOpener.h +index 4f3135c60..c57c431ad 100644 +--- a/src/HappyConnOpener.h ++++ b/src/HappyConnOpener.h +@@ -156,22 +156,28 @@ private:

• /// a connection opening attempt in progress (or falsy)
  

• class Attempt {
  

• public:
  

++ /// HappyConnOpener method implementing a ConnOpener callback ++ using CallbackMethod = void (HappyConnOpener::*)(const CommConnectCbParams &); ++ ++ Attempt(const CallbackMethod method, const char *methodName); ++

•     explicit operator bool() const { return static_cast<bool>(path); }
  

•     /// reacts to a natural attempt completion (successful or otherwise)
  

+- void finish() { clear(); } ++ void finish();

•     /// aborts an in-progress attempt
  

•     void cancel(const char *reason);
  

•     PeerConnectionPointer path; ///< the destination we are connecting to
  

+- AsyncCall::Pointer connector; ///< our opener callback +- Comm::ConnOpener::Pointer opener; ///< connects to path and calls us

+- private: +- /// cleans up after the attempt ends (successfully or otherwise) +- void clear() { path = nullptr; connector = nullptr; opener = nullptr; } ++ /// waits for a connection to the peer to be established/opened ++ JobWaitComm::ConnOpener connWait; ++ ++ const CallbackMethod callbackMethod; ///< ConnOpener calls this method ++ const char * const callbackMethodName; ///< for callbackMethod debugging

• };
  

++ friend std::ostream &operator <<(std::ostream &, const Attempt &);

• /* AsyncJob API */
  

• virtual void start() override;
  

+@@ -190,7 +196,9 @@ private:

• void openFreshConnection(Attempt &, PeerConnectionPointer &);
  

• bool reuseOldConnection(PeerConnectionPointer &);
  

+- void connectDone(const CommConnectCbParams &); ++ void notePrimeConnectDone(const CommConnectCbParams &); ++ void noteSpareConnectDone(const CommConnectCbParams &); ++ void handleConnOpenerAnswer(Attempt &, const CommConnectCbParams &, const char *connDescription);

• void checkForNewConnection();
  

+diff --git a/src/PeerPoolMgr.cc b/src/PeerPoolMgr.cc +index 2caa09f44..7423dd669 100644 +--- a/src/PeerPoolMgr.cc ++++ b/src/PeerPoolMgr.cc +@@ -43,9 +43,8 @@ public:

• PeerPoolMgr::PeerPoolMgr(CachePeer *aPeer): AsyncJob("PeerPoolMgr"),

• peer(cbdataReference(aPeer)),
  

• request(),
  

+- opener(), +- securer(), +- closer(), ++ transportWait(), ++ encryptionWait(),

• addrUsed(0)
  

• {
• }

+@@ -90,7 +89,7 @@ PeerPoolMgr::doneAll() const

• void
• PeerPoolMgr::handleOpenedConnection(const CommConnectCbParams &params)
• {

+- opener = NULL; ++ transportWait.finish();

• if (!validPeer()) {
  

•     debugs(48, 3, "peer gone");
  

+@@ -100,9 +99,6 @@ PeerPoolMgr::handleOpenedConnection(const CommConnectCbParams &params)

• }
  

• if (params.flag != Comm::OK) {
  

+- /* it might have been a timeout with a partially open link */ +- if (params.conn != NULL) +- params.conn->close();

•     peerConnectFailed(peer);
  

•     checkpoint("conn opening failure"); // may retry
  

•     return;
  

+@@ -112,20 +108,16 @@ PeerPoolMgr::handleOpenedConnection(const CommConnectCbParams &params)

• // Handle TLS peers.
  

• if (peer->secure.encryptTransport) {
  

+- typedef CommCbMemFunT<PeerPoolMgr, CommCloseCbParams> CloserDialer; +- closer = JobCallback(48, 3, CloserDialer, this, +- PeerPoolMgr::handleSecureClosure); +- comm_add_close_handler(params.conn->fd, closer); +- +- securer = asyncCall(48, 4, "PeerPoolMgr::handleSecuredPeer", +- MyAnswerDialer(this, &PeerPoolMgr::handleSecuredPeer)); ++ // XXX: Exceptions orphan params.conn ++ AsyncCall::Pointer callback = asyncCall(48, 4, "PeerPoolMgr::handleSecuredPeer", ++ MyAnswerDialer(this, &PeerPoolMgr::handleSecuredPeer));

•     const int peerTimeout = peerConnectTimeout(peer);
  

•     const int timeUsed = squid_curtime - params.conn->startTime();
  

•     // Use positive timeout when less than one second is left for conn.
  

•     const int timeLeft = positiveTimeout(peerTimeout - timeUsed);
  

+- auto *connector = new Security::BlindPeerConnector(request, params.conn, securer, nullptr, timeLeft); +- AsyncJob::Start(connector); // will call our callback ++ const auto connector = new Security::BlindPeerConnector(request, params.conn, callback, nullptr, timeLeft); ++ encryptionWait.start(connector, callback);

•     return;
  

• }
  

+@@ -144,16 +136,7 @@ PeerPoolMgr::pushNewConnection(const Comm::ConnectionPointer &conn)

• void
• PeerPoolMgr::handleSecuredPeer(Security::EncryptorAnswer &answer)
• {

+- Must(securer != NULL); +- securer = NULL; +- +- if (closer != NULL) { +- if (answer.conn != NULL) +- comm_remove_close_handler(answer.conn->fd, closer); +- else +- closer->cancel("securing completed"); +- closer = NULL; +- } ++ encryptionWait.finish();

• if (!validPeer()) {
  

•     debugs(48, 3, "peer gone");
  

+@@ -162,35 +145,33 @@ PeerPoolMgr::handleSecuredPeer(Security::EncryptorAnswer &answer)

•     return;
  

• }
  

++ assert(!answer.tunneled);

• if (answer.error.get()) {
  

+- if (answer.conn != NULL) +- answer.conn->close(); ++ assert(!answer.conn);

•     // PeerConnector calls peerConnectFailed() for us;
  

•     checkpoint("conn securing failure"); // may retry
  

•     return;
  

• }
  

+- pushNewConnection(answer.conn); +-} ++ assert(answer.conn);

+-void +-PeerPoolMgr::handleSecureClosure(const CommCloseCbParams &params) +-{ +- Must(closer != NULL); +- Must(securer != NULL); +- securer->cancel("conn closed by a 3rd party"); +- securer = NULL; +- closer = NULL; +- // allow the closing connection to fully close before we check again +- Checkpoint(this, "conn closure while securing"); ++ // The socket could get closed while our callback was queued. Sync ++ // Connection. XXX: Connection::fd may already be stale/invalid here. ++ if (answer.conn->isOpen() && fd_table[answer.conn->fd].closing()) { ++ answer.conn->noteClosure(); ++ checkpoint("external connection closure"); // may retry ++ return; ++ } ++ ++ pushNewConnection(answer.conn);

• }
• void
• PeerPoolMgr::openNewConnection()
• {

• // KISS: Do nothing else when we are already doing something.
  

+- if (opener != NULL || securer != NULL || shutting_down) { +- debugs(48, 7, "busy: " << opener << '|' << securer << '|' << shutting_down); ++ if (transportWait || encryptionWait || shutting_down) { ++ debugs(48, 7, "busy: " << transportWait << '|' << encryptionWait << '|' << shutting_down);

•     return; // there will be another checkpoint when we are done opening/securing
  

• }
  

+@@ -227,9 +208,9 @@ PeerPoolMgr::openNewConnection()

• const int ctimeout = peerConnectTimeout(peer);
  

• typedef CommCbMemFunT<PeerPoolMgr, CommConnectCbParams> Dialer;
  

+- opener = JobCallback(48, 5, Dialer, this, PeerPoolMgr::handleOpenedConnection); +- Comm::ConnOpener *cs = new Comm::ConnOpener(conn, opener, ctimeout); +- AsyncJob::Start(cs); ++ AsyncCall::Pointer callback = JobCallback(48, 5, Dialer, this, PeerPoolMgr::handleOpenedConnection); ++ const auto cs = new Comm::ConnOpener(conn, callback, ctimeout); ++ transportWait.start(cs, callback);

• }
• void

+diff --git a/src/PeerPoolMgr.h b/src/PeerPoolMgr.h +index 217994393..6da61b10b 100644 +--- a/src/PeerPoolMgr.h ++++ b/src/PeerPoolMgr.h +@@ -10,6 +10,7 @@

• #define SQUID_PEERPOOLMGR_H
• #include "base/AsyncJob.h"

++#include "base/JobWait.h"

• #include "comm/forward.h"
• #include "security/forward.h"

+@@ -54,18 +55,19 @@ protected:

• /// Security::PeerConnector callback
  

• void handleSecuredPeer(Security::EncryptorAnswer &answer);
  

+- /// called when the connection we are trying to secure is closed by a 3rd party +- void handleSecureClosure(const CommCloseCbParams &params); +-

• /// the final step in connection opening (and, optionally, securing) sequence
  

• void pushNewConnection(const Comm::ConnectionPointer &conn);
  

• private:

• CachePeer *peer; ///< the owner of the pool we manage
  

• RefCount<HttpRequest> request; ///< fake HTTP request for conn opening code
  

+- AsyncCall::Pointer opener; ///< whether we are opening a connection +- AsyncCall::Pointer securer; ///< whether we are securing a connection +- AsyncCall::Pointer closer; ///< monitors conn while we are securing it ++ ++ /// waits for a transport connection to the peer to be established/opened ++ JobWaitComm::ConnOpener transportWait; ++ ++ /// waits for the established transport connection to be secured/encrypted ++ JobWaitSecurity::BlindPeerConnector encryptionWait; ++

• unsigned int addrUsed; ///< counter for cycling through peer addresses
  

• };

+diff --git a/src/ResolvedPeers.cc b/src/ResolvedPeers.cc +index 5b2c6740d..06ab02d23 100644 +--- a/src/ResolvedPeers.cc ++++ b/src/ResolvedPeers.cc +@@ -151,7 +151,7 @@ ResolvedPeers::extractFound(const char *description, const Paths::iterator &foun

•     while (++pathsToSkip < paths_.size() && !paths_[pathsToSkip].available) {}
  

• }
  

+- const auto cleanPath = path.connection->cloneDestinationDetails(); ++ const auto cleanPath = path.connection->cloneProfile();

• return PeerConnectionPointer(cleanPath, found - paths_.begin());
  

• }

+diff --git a/src/adaptation/icap/ModXact.cc b/src/adaptation/icap/ModXact.cc +index bbeebdbb5..982f38f8d 100644 +--- a/src/adaptation/icap/ModXact.cc ++++ b/src/adaptation/icap/ModXact.cc +@@ -187,8 +187,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 f8988ac29..2fda1318b 100644 +--- 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 5ed4fbfbb..2464ea533 100644 +--- 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 3811ab48f..725cd6225 100644 +--- 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 6df475712..fbd896888 100644 +--- a/src/adaptation/icap/ServiceRep.cc ++++ b/src/adaptation/icap/ServiceRep.cc +@@ -112,9 +112,10 @@ void Adaptation::Icap::ServiceRep::noteFailure()

• // should be configurable.
  

• }

+-// returns a persistent or brand new connection; negative int on failures ++// TODO: getIdleConnection() and putConnection()/noteConnectionFailed() manage a ++// "used connection slot" resource. Automate that resource tracking (RAII/etc.).

• Comm::ConnectionPointer

+-Adaptation::Icap::ServiceRep::getConnection(bool retriableXact, bool &reused) ++Adaptation::Icap::ServiceRep::getIdleConnection(const bool retriableXact)

• {

• Comm::ConnectionPointer connection;
  

+@@ -137,7 +138,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 +150,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 f2e893244..cca2df4ab 100644 +--- 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 9aacf2e1b..962ed1703 100644 +--- a/src/adaptation/icap/Xaction.cc ++++ b/src/adaptation/icap/Xaction.cc +@@ -13,6 +13,7 @@

• #include "adaptation/icap/Config.h"
• #include "adaptation/icap/Launcher.h"
• #include "adaptation/icap/Xaction.h"

++#include "base/JobWait.h"

• #include "base/TextException.h"
• #include "comm.h"
• #include "comm/Connection.h"

+@@ -79,7 +80,6 @@ Adaptation::Icap::Xaction::Xaction(const char *aTypeName, Adaptation::Icap::Serv

• icapRequest(NULL),
  

• icapReply(NULL),
  

• attempts(0),
  

+- connection(NULL),

• theService(aService),
  

• commEof(false),
  

• reuseConnection(true),
  

+@@ -87,14 +87,8 @@ Adaptation::Icap::Xaction::Xaction(const char *aTypeName, Adaptation::Icap::Serv

• isRepeatable(true),
  

• ignoreLastWrite(false),
  

• waitingForDns(false),
  

+- stopReason(NULL), +- connector(NULL), +- reader(NULL), +- writer(NULL), +- closer(NULL),

• alep(new AccessLogEntry),
  

+- al(*alep), +- cs(NULL) ++ al(*alep)

• {

• debugs(93,3, typeName << " constructed, this=" << this <<
  

•        " [icapx" << id << ']'); // we should not call virtual status() here
  

+@@ -150,6 +144,8 @@ static void

• icapLookupDnsResults(const ipcache_addrs *ia, const Dns::LookupDetails &, void *data)
• {

• Adaptation::Icap::Xaction *xa = static_cast<Adaptation::Icap::Xaction *>(data);
  

++ /// TODO: refactor with CallJobHere1, passing either std::optional (after upgrading to C++17) ++ /// or OptionalIp::Address (when it can take non-trivial types)

• xa->dnsLookupDone(ia);
  

• }

+@@ -164,21 +160,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)) { +- // Set comm Close handler +- // fake the connect callback +- // TODO: can we sync call Adaptation::Icap::Xaction::noteCommConnected here instead? +- typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> Dialer; +- CbcPointer<Xaction> self(this); +- Dialer dialer(self, &Adaptation::Icap::Xaction::noteCommConnected); +- dialer.params.conn = connection; +- dialer.params.flag = Comm::OK; +- // fake other parameters by copying from the existing connection +- connector = asyncCall(93,3, "Adaptation::Icap::Xaction::noteCommConnected", dialer); +- ScheduleCallHere(connector); ++ if (const auto pconn = s.getIdleConnection(isRetriable)) { ++ useTransportConnection(pconn);

•     return;
  

• }
  

+@@ -207,30 +190,22 @@ Adaptation::Icap::Xaction::dnsLookupDone(const ipcache_addrs *ia)

• #if WHEN_IPCACHE_NBGETHOSTBYNAME_USES_ASYNC_CALLS

•     dieOnConnectionFailure(); // throws
  

• #else // take a step back into protected Async call dialing.

+- // fake the connect callback +- typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommConnectCbParams> Dialer; +- CbcPointer<Xaction> self(this); +- Dialer dialer(self, &Adaptation::Icap::Xaction::noteCommConnected); +- dialer.params.conn = connection; +- dialer.params.flag = Comm::COMM_ERROR; +- // fake other parameters by copying from the existing connection +- connector = asyncCall(93,3, "Adaptation::Icap::Xaction::noteCommConnected", dialer); +- ScheduleCallHere(connector); ++ CallJobHere(93, 3, this, Xaction, Xaction::dieOnConnectionFailure);

• #endif

•     return;
  

• }
  

+- 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;
  

+- connector = JobCallback(93,3, ConnectDialer, this, Adaptation::Icap::Xaction::noteCommConnected); +- cs = new Comm::ConnOpener(connection, connector, TheConfig.connect_timeout(service().cfg().bypass)); ++ AsyncCall::Pointer callback = JobCallback(93, 3, ConnectDialer, this, Adaptation::Icap::Xaction::noteCommConnected); ++ const auto cs = new Comm::ConnOpener(conn, callback, TheConfig.connect_timeout(service().cfg().bypass));

• cs->setHost(s.cfg().host.termedBuf());
  

+- AsyncJob::Start(cs.get()); ++ transportWait.start(cs, callback);

• }
• /*

+@@ -256,6 +231,8 @@ void Adaptation::Icap::Xaction::closeConnection()

•         closer = NULL;
  

•     }
  

++ commUnsetConnTimeout(connection); ++

•     cancelRead(); // may not work
  

•     if (reuseConnection && !doneWithIo()) {
  

+@@ -275,54 +252,65 @@ void Adaptation::Icap::Xaction::closeConnection()

•     writer = NULL;
  

•     reader = NULL;
  

+- connector = NULL;

•     connection = NULL;
  

• }
  

• }

+-// connection with the ICAP service established ++/// called when the connection attempt to an ICAP service completes (successfully or not)

• void Adaptation::Icap::Xaction::noteCommConnected(const CommConnectCbParams &io)
• {

+- cs = NULL; ++ transportWait.finish();

+- if (io.flag == Comm::TIMEOUT) { +- handleCommTimedout(); ++ if (io.flag != Comm::OK) { ++ dieOnConnectionFailure(); // throws

•     return;
  

• }
  

+- Must(connector != NULL); +- connector = NULL; +- +- if (io.flag != Comm::OK) +- dieOnConnectionFailure(); // throws +- +- typedef CommCbMemFunT<Adaptation::Icap::Xaction, CommTimeoutCbParams> TimeoutDialer; +- AsyncCall::Pointer timeoutCall = asyncCall(93, 5, "Adaptation::Icap::Xaction::noteCommTimedout", +- TimeoutDialer(this,&Adaptation::Icap::Xaction::noteCommTimedout)); +- commSetConnTimeout(io.conn, TheConfig.connect_timeout(service().cfg().bypass), timeoutCall); ++ useTransportConnection(io.conn); ++}

+- 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(io.conn->fd, closer); ++/// 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::useTransportConnection(const Comm::ConnectionPointer &conn) ++{ ++ 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[io.conn->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);
  

+- securer = asyncCall(93, 4, "Adaptation::Icap::Xaction::handleSecuredPeer", +- MyIcapAnswerDialer(me, &Adaptation::Icap::Xaction::handleSecuredPeer)); ++ AsyncCall::Pointer callback = asyncCall(93, 4, "Adaptation::Icap::Xaction::handleSecuredPeer", ++ MyIcapAnswerDialer(me, &Adaptation::Icap::Xaction::handleSecuredPeer));

+- auto *sslConnector = new Ssl::IcapPeerConnector(theService, io.conn, securer, masterLogEntry(), TheConfig.connect_timeout(service().cfg().bypass)); +- AsyncJob::Start(sslConnector); // will call our callback ++ const auto sslConnector = new Ssl::IcapPeerConnector(theService, conn, callback, masterLogEntry(), TheConfig.connect_timeout(service().cfg().bypass)); ++ ++ encryptionWait.start(sslConnector, 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, 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()

+@@ -367,40 +355,25 @@ 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 bool whileConnecting = connector != NULL; +- 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()); ++ 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 &)
• {

+- if (securer != NULL) { +- securer->cancel("Connection closed before SSL negotiation finished"); +- securer = NULL; ++ if (connection) { ++ connection->noteClosure(); ++ connection = nullptr;

• }
  

• closer = NULL;
  

+- handleCommClosed(); +-}

+-void Adaptation::Icap::Xaction::handleCommClosed() +-{

• static const auto d = MakeNamedErrorDetail("ICAP_XACT_CLOSE");
  

• detailError(d);
  

• mustStop("ICAP service connection externally closed");
  

+@@ -424,7 +397,8 @@ void Adaptation::Icap::Xaction::callEnd()

• bool Adaptation::Icap::Xaction::doneAll() const
• {

+- return !waitingForDns && !connector && !securer && !reader && !writer && ++ return !waitingForDns && !transportWait && !encryptionWait && ++ !reader && !writer &&

•        Adaptation::Initiate::doneAll();
  

• }

+@@ -568,7 +542,7 @@ bool Adaptation::Icap::Xaction::doneWriting() const

• bool Adaptation::Icap::Xaction::doneWithIo() const
• {

• return haveConnection() &&
  

+- !connector && !reader && !writer && // fast checks, some redundant ++ !transportWait && !reader && !writer && // fast checks, some redundant

•        doneReading() && doneWriting();
  

• }

+@@ -608,10 +582,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 (cs.valid()) { +- debugs(93,6, HERE << id << " about to notify ConnOpener!"); +- CallJobHere(93, 3, cs, Comm::ConnOpener, noteAbort); +- cs = NULL; ++ if (transportWait || encryptionWait) {

•     service().noteConnectionFailed("abort");
  

• }
  

+@@ -750,20 +721,12 @@ Ssl::IcapPeerConnector::noteNegotiationDone(ErrorState *error)

• void
• Adaptation::Icap::Xaction::handleSecuredPeer(Security::EncryptorAnswer &answer)
• {

+- Must(securer != NULL); +- securer = NULL; +- +- if (closer != NULL) { +- if (Comm::IsConnOpen(answer.conn)) +- comm_remove_close_handler(answer.conn->fd, closer); +- else +- closer->cancel("securing completed"); +- closer = NULL; +- } ++ encryptionWait.finish();

++ assert(!answer.tunneled);

• if (answer.error.get()) {
  

+- if (answer.conn != NULL) +- answer.conn->close(); ++ assert(!answer.conn); ++ // TODO: Refactor dieOnConnectionFailure() to be usable here as well.

•     debugs(93, 2, typeName <<
  

•            " TLS negotiation to " << service().cfg().uri << " failed");
  

•     service().noteConnectionFailed("failure");
  

+@@ -774,8 +737,18 @@ Adaptation::Icap::Xaction::handleSecuredPeer(Security::EncryptorAnswer &answer)

• debugs(93, 5, "TLS negotiation to " << service().cfg().uri << " complete");
  

+- service().noteConnectionUse(answer.conn); ++ assert(answer.conn); ++ ++ // The socket could get closed while our callback was queued. Sync ++ // Connection. XXX: Connection::fd may already be stale/invalid here. ++ if (answer.conn->isOpen() && fd_table[answer.conn->fd].closing()) { ++ answer.conn->noteClosure(); ++ service().noteConnectionFailed("external TLS connection closure"); ++ static const auto d = MakeNamedErrorDetail("ICAP_XACT_SSL_CLOSE"); ++ detailError(d); ++ throw TexcHere("external closure of the TLS ICAP service connection"); ++ }

+- handleCommConnected(); ++ useIcapConnection(answer.conn);

• }

+diff --git a/src/adaptation/icap/Xaction.h b/src/adaptation/icap/Xaction.h +index b66044594..31a6e22fc 100644 +--- a/src/adaptation/icap/Xaction.h ++++ b/src/adaptation/icap/Xaction.h +@@ -12,6 +12,7 @@

• #include "AccessLogEntry.h"
• #include "adaptation/icap/ServiceRep.h"
• #include "adaptation/Initiate.h"

++#include "base/JobWait.h"

• #include "comm/ConnOpener.h"
• #include "error/forward.h"
• #include "HttpReply.h"

+@@ -20,6 +21,10 @@

• class MemBuf;

++namespace Ssl { ++class IcapPeerConnector; ++} ++

• namespace Adaptation
• {
• namespace Icap

+@@ -65,12 +70,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
  

+@@ -78,7 +83,6 @@ protected:

• void openConnection();
  

• void closeConnection();
  

+- void dieOnConnectionFailure();

• bool haveConnection() const;
  

• void scheduleRead();
  

+@@ -124,11 +128,13 @@ public:

• ServiceRep &service();
  

• private:

++ 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;
  

+@@ -139,13 +145,8 @@ protected:

• bool ignoreLastWrite;
  

• bool waitingForDns; ///< expecting a ipcache_nbgethostbyname() callback
  

+- const char *stopReason; +- +- // active (pending) comm callbacks for the ICAP server connection +- AsyncCall::Pointer connector;

• AsyncCall::Pointer reader;
  

• AsyncCall::Pointer writer;
  

+- AsyncCall::Pointer closer;

• AccessLogEntry::Pointer alep; ///< icap.log entry
  

• AccessLogEntry &al; ///< short for *alep
  

+@@ -155,8 +156,16 @@ protected:

• timeval icap_tio_finish;   /*time when the last byte of the ICAP responsewas received*/
  

• private:

+- Comm::ConnOpener::Pointer cs; +- AsyncCall::Pointer securer; ///< whether we are securing a connection ++ /// waits for a transport connection to the ICAP server to be established/opened ++ JobWaitComm::ConnOpener transportWait; ++ ++ /// waits for the established transport connection to be secured/encrypted ++ JobWaitSsl::IcapPeerConnector encryptionWait; ++ ++ /// open and, if necessary, secured connection to the ICAP server (or nil) ++ Comm::ConnectionPointer connection; ++ ++ AsyncCall::Pointer closer;

• };
• } // namespace Icap

+diff --git a/src/base/AsyncJob.cc b/src/base/AsyncJob.cc +index 3b9161e4a..111667d5d 100644 +--- a/src/base/AsyncJob.cc ++++ b/src/base/AsyncJob.cc +@@ -20,11 +20,11 @@

• InstanceIdDefinitions(AsyncJob, "job");

+-AsyncJob::Pointer AsyncJob::Start(AsyncJob *j) ++void ++AsyncJob::Start(const Pointer &job)

• {

+- AsyncJob::Pointer job(j);

• CallJobHere(93, 5, job, AsyncJob, start);
  

+- return job; ++ job->started_ = true; // it is the attempt that counts

• }
• AsyncJob::AsyncJob(const char *aTypeName) :

+@@ -38,6 +38,7 @@ AsyncJob::~AsyncJob()

• {

• debugs(93,5, "AsyncJob destructed, this=" << this <<
  

•        " type=" << typeName << " [" << id << ']');
  

++ assert(!started_ || swanSang_);

• }
• void AsyncJob::start()

+@@ -141,9 +142,16 @@ void AsyncJob::callEnd()

•     AsyncCall::Pointer inCallSaved = inCall;
  

•     void *thisSaved = this;
  

++ // TODO: Swallow swanSong() exceptions to reduce memory leaks. ++ ++ // Job callback invariant: swanSong() is (only) called for started jobs. ++ // Here to detect violations in kids that forgot to call our swanSong(). ++ assert(started_); ++ ++ swanSang_ = true; // it is the attempt that counts

•     swanSong();
  

+- delete this; // this is the only place where the object is deleted ++ delete this; // this is the only place where a started job is deleted

•     // careful: this object does not exist any more
  

•     debugs(93, 6, HERE << *inCallSaved << " ended " << thisSaved);
  

+diff --git a/src/base/AsyncJob.h b/src/base/AsyncJob.h +index 4d685dff5..a46e30071 100644 +--- a/src/base/AsyncJob.h ++++ b/src/base/AsyncJob.h +@@ -36,8 +36,13 @@ public:

• public:

• AsyncJob(const char *aTypeName);
  

+- /// starts a freshly created job (i.e., makes the job asynchronous) +- static Pointer Start(AsyncJob *job); ++ /// Promises to start the configured job (eventually). The job is deemed to ++ /// be running asynchronously beyond this point, so the caller should only ++ /// access the job object via AsyncCalls rather than directly. ++ /// ++ /// swanSong() is only called for jobs for which this method has returned ++ /// successfully (i.e. without throwing). ++ static void Start(const Pointer &job);

• protected:

• // XXX: temporary method to replace "delete this" in jobs-in-transition.
  

+@@ -62,6 +67,11 @@ public:

• /// called when the job throws during an async call
  

• virtual void callException(const std::exception &e);
  

++ /// process external request to terminate now (i.e. during this async call) ++ void handleStopRequest() { mustStop("externally aborted"); } ++ ++ const InstanceId<AsyncJob> id; ///< job identifier ++

• protected:

• // external destruction prohibited to ensure swanSong() is called
  

• virtual ~AsyncJob();
  

+@@ -69,7 +79,9 @@ protected:

• const char *stopReason; ///< reason for forcing done() to be true
  

• const char *typeName; ///< kid (leaf) class name, for debugging
  

• AsyncCall::Pointer inCall; ///< the asynchronous call being handled, if any
  

+- const InstanceId<AsyncJob> id; ///< job identifier ++ ++ bool started_ = false; ///< Start() has finished successfully ++ bool swanSang_ = false; ///< swanSong() was called

• };
• #endif /* SQUID_ASYNC_JOB_H */

+diff --git a/src/base/JobWait.cc b/src/base/JobWait.cc +new file mode 100644 +index 000000000..ba6832415 +--- /dev/null ++++ b/src/base/JobWait.cc +@@ -0,0 +1,81 @@ ++/* ++ * Copyright (C) 1996-2020 The Squid Software Foundation and contributors ++ * ++ * Squid software is distributed under GPLv2+ license and includes ++ * contributions from numerous individuals and organizations. ++ * Please see the COPYING and CONTRIBUTORS files for details. ++ */ ++ ++#include "squid.h" ++#include "base/AsyncJobCalls.h" ++#include "base/JobWait.h" ++ ++#include <cassert> ++#include <iostream> ++ ++JobWaitBase::JobWaitBase() = default; ++ ++JobWaitBase::~JobWaitBase() ++{ ++ cancel("owner gone"); ++} ++ ++void ++JobWaitBase::start_(const AsyncJob::Pointer aJob, const AsyncCall::Pointer aCall) ++{ ++ // Invariant: The wait will be over. We cannot guarantee that the job will ++ // call the callback, of course, but we can check these prerequisites. ++ assert(aCall); ++ assert(aJob.valid()); ++ ++ // "Double" waiting state leads to conflicting/mismatching callbacks ++ // detailed in finish(). Detect that bug ASAP. ++ assert(!waiting()); ++ ++ assert(!callback_); ++ assert(!job_); ++ callback_ = aCall; ++ job_ = aJob; ++ ++ AsyncJob::Start(job_.get()); ++} ++ ++void ++JobWaitBase::finish() ++{ ++ // Unexpected callbacks might result in disasters like secrets exposure, ++ // data corruption, or expensive message routing mistakes when the callback ++ // info is applied to the wrong message part or acted upon prematurely. ++ assert(waiting()); ++ clear(); ++} ++ ++void ++JobWaitBase::cancel(const char *reason) ++{ ++ if (callback_) { ++ callback_->cancel(reason); ++ ++ // Instead of AsyncJob, the class parameter could be Job. That would ++ // avoid runtime child-to-parent CbcPointer conversion overheads, but ++ // complicate support for Jobs with virtual AsyncJob bases (GCC error: ++ // "pointer to member conversion via virtual base AsyncJob") and also ++ // cache-log "Job::handleStopRequest()" with a non-existent class name. ++ CallJobHere(callback_->debugSection, callback_->debugLevel, job_, AsyncJob, handleStopRequest); ++ ++ clear(); ++ } ++} ++ ++void ++JobWaitBase::print(std::ostream &os) const ++{ ++ // use a backarrow to emphasize that this is a callback: call24<-job6 ++ if (callback_) ++ os << callback_->id << "<-"; ++ if (const auto rawJob = job_.get()) ++ os << rawJob->id; ++ else ++ os << job_; // raw pointer of a gone job may still be useful for triage ++} ++ +diff --git a/src/base/JobWait.h b/src/base/JobWait.h +new file mode 100644 +index 000000000..6b1131331 +--- /dev/null ++++ b/src/base/JobWait.h +@@ -0,0 +1,91 @@ ++/* ++ * Copyright (C) 1996-2021 The Squid Software Foundation and contributors ++ * ++ * Squid software is distributed under GPLv2+ license and includes ++ * contributions from numerous individuals and organizations. ++ * Please see the COPYING and CONTRIBUTORS files for details. ++ */ ++ ++#ifndef SQUID_BASE_JOBWAIT_H ++#define SQUID_BASE_JOBWAIT_H ++ ++#include "base/AsyncJob.h" ++#include "base/CbcPointer.h" ++ ++#include <iosfwd> ++ ++/// Manages waiting for an AsyncJob callback. Use type-safe JobWait instead. ++/// This base class does not contain code specific to the actual Job type. ++class JobWaitBase ++{ ++public: ++ JobWaitBase(); ++ ~JobWaitBase(); ++ ++ /// no copying of any kind: each waiting context needs a dedicated AsyncCall ++ JobWaitBase(JobWaitBase &&) = delete; ++ ++ explicit operator bool() const { return waiting(); } ++ ++ /// whether we are currently waiting for the job to call us back ++ /// the job itself may be gone even if this returns true ++ bool waiting() const { return bool(callback_); } ++ ++ /// ends wait (after receiving the call back) ++ /// forgets the job which is likely to be gone by now ++ void finish(); ++ ++ /// aborts wait (if any) before receiving the call back ++ /// does nothing if we are not waiting ++ void cancel(const char *reason); ++ ++ /// summarizes what we are waiting for (for debugging) ++ void print(std::ostream &) const; ++ ++protected: ++ /// starts waiting for the given job to call the given callback ++ void start_(AsyncJob::Pointer, AsyncCall::Pointer); ++ ++private: ++ /// the common part of finish() and cancel() ++ void clear() { job_.clear(); callback_ = nullptr; } ++ ++ /// the job that we are waiting to call us back (or nil) ++ AsyncJob::Pointer job_; ++ ++ /// the call we are waiting for the job_ to make (or nil) ++ AsyncCall::Pointer callback_; ++}; ++ ++/// Manages waiting for an AsyncJob callback. ++/// Completes JobWaitBase by providing Job type-specific members. ++template <class Job> ++class JobWait: public JobWaitBase ++{ ++public: ++ typedef CbcPointer<Job> JobPointer; ++ ++ /// starts waiting for the given job to call the given callback ++ void start(const JobPointer &aJob, const AsyncCall::Pointer &aCallback) { ++ start_(aJob, aCallback); ++ typedJob_ = aJob; ++ } ++ ++ /// \returns a cbdata pointer to the job we are waiting for (or nil) ++ /// the returned pointer may be falsy, even if we are still waiting() ++ JobPointer job() const { return waiting() ? typedJob_ : nullptr; } ++ ++private: ++ /// nearly duplicates JobWaitBase::job_ but exposes the actual job type ++ JobPointer typedJob_; ++}; ++ ++inline ++std::ostream &operator <<(std::ostream &os, const JobWaitBase &wait) ++{ ++ wait.print(os); ++ return os; ++} ++ ++#endif /* SQUID_BASE_JOBWAIT_H */ ++ +diff --git a/src/base/Makefile.am b/src/base/Makefile.am +index c9564d755..374ea8798 100644 +--- a/src/base/Makefile.am ++++ b/src/base/Makefile.am +@@ -34,6 +34,8 @@ libbase_la_SOURCES = \

• Here.h \
• InstanceId.cc \
• InstanceId.h \

++ JobWait.cc \ ++ JobWait.h \

• Lock.h \
• LookupTable.h \
• LruMap.h \

+diff --git a/src/base/forward.h b/src/base/forward.h +index 3803e1ea0..46de97c8d 100644 +--- a/src/base/forward.h ++++ b/src/base/forward.h +@@ -17,6 +17,7 @@ class ScopedId;

• template<class Cbc> class CbcPointer;
• template<class RefCountableKid> class RefCount;

++template<class Job> class JobWait;

• typedef CbcPointer<AsyncJob> AsyncJobPointer;
• typedef RefCount<CodeContext> CodeContextPointer;

+diff --git a/src/client_side.cc b/src/client_side.cc +index b8d786423..4eb697696 100644 +--- a/src/client_side.cc ++++ b/src/client_side.cc +@@ -503,6 +503,10 @@ httpRequestFree(void *data)

• /* This is a handler normally called by comm_close() */
• void ConnStateData::connStateClosed(const CommCloseCbParams &)
• {

++ if (clientConnection) { ++ clientConnection->noteClosure(); ++ // keep closed clientConnection for logging, clientdb cleanup, etc. ++ }

• deleteThis("ConnStateData::connStateClosed");
  

• }

+diff --git a/src/clients/FtpClient.cc b/src/clients/FtpClient.cc +index 7874db121..afe7827dc 100644 +--- a/src/clients/FtpClient.cc ++++ b/src/clients/FtpClient.cc +@@ -201,10 +201,6 @@ Ftp::Client::Client(FwdState *fwdState):

• Ftp::Client::~Client()
• {

+- if (data.opener != NULL) { +- data.opener->cancel("Ftp::Client destructed"); +- data.opener = NULL; +- }

• data.close();
  

• safe_free(old_request);
  

+@@ -786,10 +782,10 @@ Ftp::Client::connectDataChannel()

• debugs(9, 3, "connecting to " << conn->remote);
  

• typedef CommCbMemFunT<Client, CommConnectCbParams> Dialer;
  

+- data.opener = JobCallback(9, 3, Dialer, this, Ftp::Client::dataChannelConnected); +- Comm::ConnOpener *cs = new Comm::ConnOpener(conn, data.opener, Config.Timeout.connect); ++ AsyncCall::Pointer callback = JobCallback(9, 3, Dialer, this, Ftp::Client::dataChannelConnected); ++ const auto cs = new Comm::ConnOpener(conn, callback, Config.Timeout.connect);

• cs->setHost(data.host);
  

+- AsyncJob::Start(cs); ++ dataConnWait.start(cs, callback);

• }
• bool

+@@ -811,10 +807,11 @@ void

• Ftp::Client::dataClosed(const CommCloseCbParams &)
• {

• debugs(9, 4, status());
  

++ if (data.conn) ++ data.conn->noteClosure();

• if (data.listenConn != NULL) {
  

•     data.listenConn->close();
  

•     data.listenConn = NULL;
  

+- // NP clear() does the: data.fd = -1;

• }
  

• data.clear();
  

• }

+@@ -879,6 +876,8 @@ void

• Ftp::Client::ctrlClosed(const CommCloseCbParams &)
• {

• debugs(9, 4, status());
  

++ if (ctrl.conn) ++ ctrl.conn->noteClosure();

• ctrl.clear();
  

• doneWithFwd = "ctrlClosed()"; // assume FwdState is monitoring too
  

• mustStop("Ftp::Client::ctrlClosed");
  

+diff --git a/src/clients/FtpClient.h b/src/clients/FtpClient.h +index ac8d22e18..4b2dd61d5 100644 +--- a/src/clients/FtpClient.h ++++ b/src/clients/FtpClient.h +@@ -64,7 +64,6 @@ public:

•  */
  

• Comm::ConnectionPointer listenConn;
  

+- AsyncCall::Pointer opener; ///< Comm opener handler callback.

• private:

• AsyncCall::Pointer closer; ///< Comm close handler callback
  

• };

+@@ -205,6 +204,10 @@ protected:

• virtual void sentRequestBody(const CommIoCbParams &io);
  

• virtual void doneSendingRequestBody();
  

++ /// Waits for an FTP data connection to the server to be established/opened. ++ /// This wait only happens in FTP passive mode (via PASV or EPSV). ++ JobWaitComm::ConnOpener dataConnWait; ++

• private:

• bool parseControlReply(size_t &bytesUsed);
  

+diff --git a/src/clients/FtpGateway.cc b/src/clients/FtpGateway.cc +index 6a7d139ca..aeeaedb48 100644 +--- a/src/clients/FtpGateway.cc ++++ b/src/clients/FtpGateway.cc +@@ -486,11 +486,9 @@ Ftp::Gateway::timeout(const CommTimeoutCbParams &io)

•     flags.pasv_supported = false;
  

•     debugs(9, DBG_IMPORTANT, "FTP Gateway timeout in SENT_PASV state");
  

+- // cancel the data connection setup. +- if (data.opener != NULL) { +- data.opener->cancel("timeout"); +- data.opener = NULL; +- } ++ // cancel the data connection setup, if any ++ dataConnWait.cancel("timeout"); ++

•     data.close();
  

• }
  

+@@ -1723,7 +1721,7 @@ void

• Ftp::Gateway::dataChannelConnected(const CommConnectCbParams &io)
• {

• debugs(9, 3, HERE);
  

+- data.opener = NULL; ++ dataConnWait.finish();

• if (io.flag != Comm::OK) {
  

•     debugs(9, 2, HERE << "Failed to connect. Retrying via another method.");
  

+@@ -2727,9 +2725,9 @@ Ftp::Gateway::mayReadVirginReplyBody() const

• return !doneWithServer();
  

• }

+-AsyncJob::Pointer ++void

• Ftp::StartGateway(FwdState *const fwdState)
• {

+- return AsyncJob::Start(new Ftp::Gateway(fwdState)); ++ AsyncJob::Start(new Ftp::Gateway(fwdState));

• }

+diff --git a/src/clients/FtpRelay.cc b/src/clients/FtpRelay.cc +index 8796a74c8..ce6ba3ba6 100644 +--- a/src/clients/FtpRelay.cc ++++ b/src/clients/FtpRelay.cc +@@ -739,7 +739,7 @@ void

• Ftp::Relay::dataChannelConnected(const CommConnectCbParams &io)
• {

• debugs(9, 3, status());
  

+- data.opener = NULL; ++ dataConnWait.finish();

• if (io.flag != Comm::OK) {
  

•     debugs(9, 2, "failed to connect FTP server data channel");
  

+@@ -804,9 +804,9 @@ Ftp::Relay::HandleStoreAbort(Relay *ftpClient)

•     ftpClient->dataComplete();
  

• }

+-AsyncJob::Pointer ++void

• Ftp::StartRelay(FwdState *const fwdState)
• {

+- return AsyncJob::Start(new Ftp::Relay(fwdState)); ++ AsyncJob::Start(new Ftp::Relay(fwdState));

• }

+diff --git a/src/clients/HttpTunneler.cc b/src/clients/HttpTunneler.cc +index 023a8586c..67a52e662 100644 +--- a/src/clients/HttpTunneler.cc ++++ b/src/clients/HttpTunneler.cc +@@ -101,6 +101,11 @@ void

• Http::Tunneler::handleConnectionClosure(const CommCloseCbParams &params)
• {

• closer = nullptr;
  

++ if (connection) { ++ countFailingConnection(); ++ connection->noteClosure(); ++ connection = nullptr; ++ }

• bailWith(new ErrorState(ERR_CONNECT_FAIL, Http::scBadGateway, request.getRaw(), al));
  

• }

+@@ -360,50 +365,64 @@ Http::Tunneler::bailWith(ErrorState *error)

• Must(error);
  

• answer().squidError = error;
  

+- if (const auto p = connection->getPeer()) +- peerConnectFailed(p); ++ if (const auto failingConnection = connection) { ++ // TODO: Reuse to-peer connections after a CONNECT error response. ++ countFailingConnection(); ++ disconnect(); ++ failingConnection->close(); ++ }

• callBack();
  

+- disconnect(); +- +- if (noteFwdPconnUse) +- fwdPconnPool->noteUses(fd_table[connection->fd].pconn.uses); +- // TODO: Reuse to-peer connections after a CONNECT error response. +- connection->close(); +- connection = nullptr;

• }
• void
• Http::Tunneler::sendSuccess()
• {

• assert(answer().positive());
  

+- callBack(); ++ assert(Comm::IsConnOpen(connection)); ++ answer().conn = connection;

• disconnect();
  

++ callBack(); ++} ++ ++void ++Http::Tunneler::countFailingConnection() ++{ ++ assert(connection); ++ if (const auto p = connection->getPeer()) ++ peerConnectFailed(p); ++ if (noteFwdPconnUse && connection->isOpen()) ++ fwdPconnPool->noteUses(fd_table[connection->fd].pconn.uses);

• }
• void
• Http::Tunneler::disconnect()
• {

++ const auto stillOpen = Comm::IsConnOpen(connection); ++

• if (closer) {
  

+- comm_remove_close_handler(connection->fd, closer); ++ if (stillOpen) ++ comm_remove_close_handler(connection->fd, closer);

•     closer = nullptr;
  

• }
  

• if (reader) {
  

+- Comm::ReadCancel(connection->fd, reader); ++ if (stillOpen) ++ Comm::ReadCancel(connection->fd, reader);

•     reader = nullptr;
  

• }
  

+- // remove connection timeout handler +- commUnsetConnTimeout(connection); ++ if (stillOpen) ++ commUnsetConnTimeout(connection); ++ ++ connection = nullptr; // may still be open

• }
• void
• Http::Tunneler::callBack()
• {

+- debugs(83, 5, connection << status()); +- if (answer().positive()) +- answer().conn = connection; ++ debugs(83, 5, answer().conn << status()); ++ assert(!connection); // returned inside answer() or gone

• auto cb = callback;
  

• callback = nullptr;
  

• ScheduleCallHere(cb);
  

+@@ -415,11 +434,10 @@ Http::Tunneler::swanSong()

• AsyncJob::swanSong();
  

• if (callback) {
  

+- if (requestWritten && tunnelEstablished) { ++ if (requestWritten && tunnelEstablished && Comm::IsConnOpen(connection)) {

•         sendSuccess();
  

•     } else {
  

+- // we should have bailed when we discovered the job-killing problem +- debugs(83, DBG_IMPORTANT, "BUG: Unexpected state while establishing a CONNECT tunnel " << connection << status()); ++ // job-ending emergencies like handleStopRequest() or callException()

•         bailWith(new ErrorState(ERR_GATEWAY_FAILURE, Http::scInternalServerError, request.getRaw(), al));
  

•     }
  

•     assert(!callback);
  

+diff --git a/src/clients/HttpTunneler.h b/src/clients/HttpTunneler.h +index c4c096fe7..e88d17610 100644 +--- a/src/clients/HttpTunneler.h ++++ b/src/clients/HttpTunneler.h +@@ -87,6 +87,7 @@ protected:

• void handleResponse(const bool eof);
  

• void bailOnResponseError(const char *error, HttpReply *);
  

++private:

• /// sends the given error to the initiator
  

• void bailWith(ErrorState*);
  

+@@ -96,12 +97,14 @@ protected:

• /// a bailWith(), sendSuccess() helper: sends results to the initiator
  

• void callBack();
  

+- /// a bailWith(), sendSuccess() helper: stops monitoring the connection ++ /// stops monitoring the connection

• void disconnect();
  

++ /// updates connection usage history before the connection is closed ++ void countFailingConnection(); ++

• TunnelerAnswer &answer();
  

+-private:

• AsyncCall::Pointer writer; ///< called when the request has been written
  

• AsyncCall::Pointer reader; ///< called when the response should be read
  

• AsyncCall::Pointer closer; ///< called when the connection is being closed
  

+diff --git a/src/clients/forward.h b/src/clients/forward.h +index 0351557da..82e5eb9bd 100644 +--- a/src/clients/forward.h ++++ b/src/clients/forward.h +@@ -28,10 +28,10 @@ namespace Ftp

• {
• /// A new FTP Gateway job

+-AsyncJobPointer StartGateway(FwdState *const fwdState); ++void StartGateway(FwdState *const fwdState);

• /// A new FTP Relay job

+-AsyncJobPointer StartRelay(FwdState *const fwdState); ++void StartRelay(FwdState *const fwdState);

• /** Construct an URI with leading / in PATH portion for use by CWD command
• • possibly others. FTP encodes absolute paths as beginning with '/'

+diff --git a/src/comm.cc b/src/comm.cc +index 54ba16271..a0913f324 100644 +--- a/src/comm.cc ++++ b/src/comm.cc +@@ -743,6 +743,10 @@ 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);
  

++ // XXX: Without the following code, callback fd may be -1. ++ // typedef CommCloseCbParams Params; ++ // auto &params = GetCommParams<Params>(call); ++ // params.fd = fd;

•         ScheduleCallHere(call);
  

•     }
  

• }
  

+@@ -1787,6 +1791,10 @@ DeferredReadManager::CloseHandler(const CommCloseCbParams &params)

• CbDataList<DeferredRead> *temp = (CbDataList<DeferredRead> *)params.data;
  

• temp->element.closer = NULL;
  

++ if (temp->element.theRead.conn) { ++ temp->element.theRead.conn->noteClosure(); ++ temp->element.theRead.conn = nullptr; ++ }

• temp->element.markCancelled();
  

• }

+@@ -1860,6 +1868,11 @@ DeferredReadManager::kickARead(DeferredRead const &aRead)

• if (aRead.cancelled)
  

•     return;
  

++ // TODO: This check still allows theReader call with a closed theRead.conn. ++ // If a delayRead() caller has a close connection handler, then such a call ++ // would be useless and dangerous. If a delayRead() caller does not have it, ++ // then the caller will get stuck when an external connection closure makes ++ // aRead.cancelled (checked above) true.

• if (Comm::IsConnOpen(aRead.theRead.conn) && fd_table[aRead.theRead.conn->fd].closing())
  

•     return;
  

+diff --git a/src/comm/ConnOpener.cc b/src/comm/ConnOpener.cc +index dc376b778..19c1237ae 100644 +--- a/src/comm/ConnOpener.cc ++++ b/src/comm/ConnOpener.cc +@@ -41,6 +41,14 @@ Comm::ConnOpener::ConnOpener(const Comm::ConnectionPointer &c, const AsyncCall::

• deadline_(squid_curtime + static_cast<time_t>(ctimeout))
  

• {

• debugs(5, 3, "will connect to " << c << " with " << ctimeout << " timeout");
  

++ assert(conn_); // we know where to go ++ ++ // Sharing a being-modified Connection object with the caller is dangerous, ++ // but we cannot ban (or even check for) that using existing APIs. We do not ++ // want to clone "just in case" because cloning is a bit expensive, and most ++ // callers already have a non-owned Connection object to give us. Until the ++ // APIs improve, we can only check that the connection is not open. ++ assert(!conn_->isOpen());

• }
• Comm::ConnOpener::~ConnOpener()

+@@ -78,6 +86,10 @@ Comm::ConnOpener::swanSong()

• if (temporaryFd_ >= 0)
  

•     closeFd();
  

++ // did we abort while owning an open connection? ++ if (conn_ && conn_->isOpen()) ++ conn_->close(); ++

• // did we abort while waiting between retries?
  

• if (calls_.sleep_)
  

•     cancelSleep();
  

+@@ -131,9 +143,18 @@ Comm::ConnOpener::sendAnswer(Comm::Flag errFlag, int xerrno, const char *why)

•                " [" << callback_->id << ']' );
  

•         // TODO save the pconn to the pconnPool ?
  

•     } else {
  

++ assert(conn_); ++ ++ // free resources earlier and simplify recipients ++ if (errFlag != Comm::OK) ++ conn_->close(); // may not be opened ++ else ++ assert(conn_->isOpen()); ++

•         typedef CommConnectCbParams Params;
  

•         Params &params = GetCommParams<Params>(callback_);
  

•         params.conn = conn_;
  

++ conn_ = nullptr; // release ownership; prevent closure by us

•         params.flag = errFlag;
  

•         params.xerrno = xerrno;
  

•         ScheduleCallHere(callback_);
  

+@@ -152,7 +173,7 @@ Comm::ConnOpener::sendAnswer(Comm::Flag errFlag, int xerrno, const char *why)

• void
• Comm::ConnOpener::cleanFd()
• {

+- debugs(5, 4, HERE << conn_ << " closing temp FD " << temporaryFd_); ++ debugs(5, 4, conn_ << "; temp FD " << temporaryFd_);

• Must(temporaryFd_ >= 0);
  

• fde &f = fd_table[temporaryFd_];
  

+@@ -258,6 +279,7 @@ bool

• Comm::ConnOpener::createFd()
• {

• Must(temporaryFd_ < 0);
  

++ assert(conn_);

• // our initators signal abort by cancelling their callbacks
  

• if (callback_ == NULL || callback_->canceled())
  

+diff --git a/src/comm/ConnOpener.h b/src/comm/ConnOpener.h +index 329d8a3c8..e1e60649d 100644 +--- a/src/comm/ConnOpener.h ++++ b/src/comm/ConnOpener.h +@@ -19,16 +19,13 @@

• namespace Comm
• {

+-/** +- * Async-opener of a Comm connection. +- */ ++/// Asynchronously opens a TCP connection. Returns CommConnectCbParams: either ++/// Comm::OK with an open connection or another Comm::Flag with a closed one.

• class ConnOpener : public AsyncJob
• {

• CBDATA_CLASS(ConnOpener);
  

• public:

+- void noteAbort() { mustStop("externally aborted"); } +-

• typedef CbcPointer<ConnOpener> Pointer;
  

• virtual bool doneAll() const;
  

+diff --git a/src/comm/Connection.cc b/src/comm/Connection.cc +index 4e9719a01..9bc08da05 100644 +--- a/src/comm/Connection.cc ++++ b/src/comm/Connection.cc +@@ -7,6 +7,7 @@

• */
• #include "squid.h"

++#include "base/JobWait.h"

• #include "CachePeer.h"
• #include "cbdata.h"
• #include "comm.h"

+@@ -60,26 +61,44 @@ Comm::Connection::~Connection()

• }
• Comm::ConnectionPointer

+-Comm::Connection::cloneDestinationDetails() const ++Comm::Connection::cloneProfile() const

• {

+- const ConnectionPointer c = new Comm::Connection; +- c->setAddrs(local, remote); +- c->peerType = peerType; +- c->flags = flags; +- c->peer_ = cbdataReference(getPeer()); +- assert(!c->isOpen()); +- return c; +-} ++ const ConnectionPointer clone = new Comm::Connection; ++ auto &c = *clone; // optimization ++ ++ /* ++ * Copy or excuse each data member. Excused members do not belong to a ++ * Connection configuration profile because their values cannot be reused ++ * across (co-existing) Connection objects and/or are tied to their own ++ * object lifetime. ++ */ ++ ++ c.setAddrs(local, remote); ++ c.peerType = peerType; ++ // fd excused ++ c.tos = tos; ++ c.nfmark = nfmark; ++ c.nfConnmark = nfConnmark; ++ // COMM_ORPHANED is not a part of connection opening instructions ++ c.flags = flags & ~COMM_ORPHANED; ++ // rfc931 is excused ++ ++#if USE_SQUID_EUI ++ // These are currently only set when accepting connections and never used ++ // for establishing new ones, so this copying is currently in vain, but, ++ // technically, they can be a part of connection opening instructions. ++ c.remoteEui48 = remoteEui48; ++ c.remoteEui64 = remoteEui64; ++#endif

+-Comm::ConnectionPointer +-Comm::Connection::cloneIdentDetails() const +-{ +- auto c = cloneDestinationDetails(); +- c->tos = tos; +- c->nfmark = nfmark; +- c->nfConnmark = nfConnmark; +- c->startTime_ = startTime_; +- return c; ++ // id excused ++ c.peer_ = cbdataReference(getPeer()); ++ // startTime_ excused ++ // tlsHistory excused ++ ++ debugs(5, 5, this << " made " << c); ++ assert(!c.isOpen()); ++ return clone;

• }
• void

+diff --git a/src/comm/Connection.h b/src/comm/Connection.h +index 5b66943ba..40c22491d 100644 +--- a/src/comm/Connection.h ++++ b/src/comm/Connection.h +@@ -77,13 +77,12 @@ public:

• /** Clear the connection properties and close any open socket. */
  

• virtual ~Connection();
  

+- /// Create a new (closed) IDENT Connection object based on our from-Squid +- /// connection properties. +- ConnectionPointer cloneIdentDetails() const; ++ /// To prevent accidental copying of Connection objects that we started to ++ /// open or that are open, use cloneProfile() instead. ++ Connection(const Connection &&) = delete;

+- /// Create a new (closed) Connection object pointing to the same destination +- /// as this from-Squid connection. +- ConnectionPointer cloneDestinationDetails() const; ++ /// Create a new closed Connection with the same configuration as this one. ++ ConnectionPointer cloneProfile() const;

• /// close the still-open connection when its last reference is gone
  

• void enterOrphanage() { flags |= COMM_ORPHANED; }
  

+@@ -140,17 +139,6 @@ public:

• virtual ScopedId codeContextGist() const override;
  

• virtual std::ostream &detailCodeContext(std::ostream &os) const override;
  

+-private: +- /** These objects may not be exactly duplicated. Use cloneIdentDetails() or +- * cloneDestinationDetails() instead. +- */ +- Connection(const Connection &c); +- +- /** These objects may not be exactly duplicated. Use cloneIdentDetails() or +- * cloneDestinationDetails() instead. +- */ +- Connection & operator =(const Connection &c); +-

• public:

• /** Address/Port for the Squid end of a TCP link. */
  

• Ip::Address local;
  

+diff --git a/src/comm/TcpAcceptor.cc b/src/comm/TcpAcceptor.cc +index 341622e0f..510efa94f 100644 +--- a/src/comm/TcpAcceptor.cc ++++ b/src/comm/TcpAcceptor.cc +@@ -206,7 +206,10 @@ void

• Comm::TcpAcceptor::handleClosure(const CommCloseCbParams &)
• {

• closer_ = NULL;
  

+- conn = NULL; ++ if (conn) { ++ conn->noteClosure(); ++ conn = nullptr; ++ }

• Must(done());
  

• }

+diff --git a/src/dns_internal.cc b/src/dns_internal.cc +index e1f4d3ee7..72078c64e 100644 +--- a/src/dns_internal.cc ++++ b/src/dns_internal.cc +@@ -872,6 +872,10 @@ static void

• idnsVCClosed(const CommCloseCbParams &params)
• {

• nsvc * vc = (nsvc *)params.data;
  

++ if (vc->conn) { ++ vc->conn->noteClosure(); ++ vc->conn = nullptr; ++ }

• delete vc;
  

• }

+diff --git a/src/eui/Eui48.h b/src/eui/Eui48.h +index 11f4e51b1..9aab5bbb7 100644 +--- a/src/eui/Eui48.h ++++ b/src/eui/Eui48.h +@@ -29,10 +29,8 @@ class Eui48

• public:

• Eui48() { clear(); }
  

+- Eui48(const Eui48 &t) { memcpy(this, &t, sizeof(Eui48)); }

• bool operator== (const Eui48 &t) const { return memcmp(eui, t.eui, SZ_EUI48_BUF) == 0; }
  

• bool operator< (const Eui48 &t) const { return memcmp(eui, t.eui, SZ_EUI48_BUF) < 0; }
  

+- ~Eui48() {}

• const unsigned char *get(void);
  

+diff --git a/src/gopher.cc b/src/gopher.cc +index 455220c2e..576a3f7b1 100644 +--- a/src/gopher.cc ++++ b/src/gopher.cc +@@ -98,11 +98,8 @@ public:

•     entry->lock("gopherState");
  

•     *replybuf = 0;
  

• }
  

+- ~GopherStateData() {if(buf) swanSong();}

+- /* AsyncJob API emulated */ +- void deleteThis(const char *aReason); +- void swanSong(); ++ ~GopherStateData();

• public:

• StoreEntry *entry;
  

+@@ -156,30 +153,18 @@ static void

• gopherStateFree(const CommCloseCbParams &params)
• {

• GopherStateData *gopherState = (GopherStateData *)params.data;
  

+- +- if (gopherState == NULL) +- return; +- +- gopherState->deleteThis("gopherStateFree"); ++ // Assume that FwdState is monitoring and calls noteClosure(). See XXX about ++ // Connection sharing with FwdState in gopherStart(). ++ delete gopherState;

• }

+-void +-GopherStateData::deleteThis(const char *) +-{ +- swanSong(); +- delete this; +-} +- +-void +-GopherStateData::swanSong() ++GopherStateData::~GopherStateData()

• {

• if (entry)
  

•     entry->unlock("gopherState");
  

+- if (buf) { ++ if (buf)

•     memFree(buf, MEM_4K_BUF);
  

+- buf = nullptr; +- }

• }
• /**

+@@ -986,6 +971,7 @@ gopherStart(FwdState * fwd)

•     return;
  

• }
  

++ // XXX: Sharing open Connection with FwdState that has its own handlers/etc.

• gopherState->serverConn = fwd->serverConnection();
  

• gopherSendRequest(fwd->serverConnection()->fd, gopherState);
  

• AsyncCall::Pointer timeoutCall = commCbCall(5, 4, "gopherTimeout",
  

+diff --git a/src/ident/Ident.cc b/src/ident/Ident.cc +index 50dc6baeb..3c3ae5e2f 100644 +--- a/src/ident/Ident.cc ++++ b/src/ident/Ident.cc +@@ -11,6 +11,7 @@

• #include "squid.h"
• #if USE_IDENT

++#include "base/JobWait.h"

• #include "comm.h"
• #include "comm/Connection.h"
• #include "comm/ConnOpener.h"

+@@ -53,8 +54,15 @@ public:

• Comm::ConnectionPointer conn;
  

• MemBuf queryMsg;  ///< the lookup message sent to IDENT server
  

+- IdentClient *clients; ++ IdentClient *clients = nullptr;

• char buf[IDENT_BUFSIZE];
  

++ ++ /// waits for a connection to the IDENT server to be established/opened ++ JobWaitComm::ConnOpener connWait; ++ ++private: ++ // use deleteThis() to destroy ++ ~IdentStateData();

• };
• CBDATA_CLASS_INIT(IdentStateData);

+@@ -73,8 +81,9 @@ static void ClientAdd(IdentStateData * state, IDCB * callback, void *callback_da

• Ident::IdentConfig Ident::TheConfig;
• void

+-Ident::IdentStateData::deleteThis(const char *) ++Ident::IdentStateData::deleteThis(const char *reason)

• {

++ debugs(30, 3, reason);

• swanSong();
  

• delete this;
  

• }

+@@ -84,6 +93,10 @@ Ident::IdentStateData::swanSong()

• {

• if (clients != NULL)
  

•     notify(NULL);
  

++} ++ ++Ident::IdentStateData::~IdentStateData() { ++ assert(!clients);

• if (Comm::IsConnOpen(conn)) {
  

•     comm_remove_close_handler(conn->fd, Ident::Close, this);
  

+@@ -112,6 +125,10 @@ void

• Ident::Close(const CommCloseCbParams &params)
• {

• IdentStateData *state = (IdentStateData *)params.data;
  

++ if (state->conn) { ++ state->conn->noteClosure(); ++ state->conn = nullptr; ++ }

• state->deleteThis("connection closed");
  

• }

+@@ -127,6 +144,16 @@ void

• Ident::ConnectDone(const Comm::ConnectionPointer &conn, Comm::Flag status, int, void *data)
• {

• IdentStateData *state = (IdentStateData *)data;
  

++ state->connWait.finish(); ++ ++ // Start owning the supplied connection (so that it is not orphaned if this ++ // function bails early). As a (tiny) optimization or perhaps just diff ++ // minimization, the close handler is added later, when we know we are not ++ // bailing. This delay is safe because comm_remove_close_handler() forgives ++ // missing handlers. ++ assert(conn); // but may be closed ++ assert(!state->conn); ++ state->conn = conn;

• if (status != Comm::OK) {
  

•     if (status == Comm::TIMEOUT)
  

+@@ -149,8 +176,8 @@ Ident::ConnectDone(const Comm::ConnectionPointer &conn, Comm::Flag status, int,

•     return;
  

• }
  

+- assert(conn != NULL && conn == state->conn); +- comm_add_close_handler(conn->fd, Ident::Close, state); ++ assert(state->conn->isOpen()); ++ comm_add_close_handler(state->conn->fd, Ident::Close, state);

• AsyncCall::Pointer writeCall = commCbCall(5,4, "Ident::WriteFeedback",
  

•                                CommIoCbPtrFun(Ident::WriteFeedback, state));
  

+@@ -259,10 +286,10 @@ Ident::Start(const Comm::ConnectionPointer &conn, IDCB * callback, void *data)

• state->hash.key = xstrdup(key);
  

• // copy the conn details. We do not want the original FD to be re-used by IDENT.
  

+- state->conn = conn->cloneIdentDetails(); ++ const auto identConn = conn->cloneProfile();

• // NP: use random port for secure outbound to IDENT_PORT
  

+- state->conn->local.port(0); +- state->conn->remote.port(IDENT_PORT); ++ identConn->local.port(0); ++ identConn->remote.port(IDENT_PORT);

• // build our query from the original connection details
  

• state->queryMsg.init();
  

+@@ -272,7 +299,8 @@ Ident::Start(const Comm::ConnectionPointer &conn, IDCB * callback, void *data)

• hash_join(ident_hash, &state->hash);
  

• AsyncCall::Pointer call = commCbCall(30,3, "Ident::ConnectDone", CommConnectCbPtrFun(Ident::ConnectDone, state));
  

+- AsyncJob::Start(new Comm::ConnOpener(state->conn, call, Ident::TheConfig.timeout)); ++ const auto connOpener = new Comm::ConnOpener(identConn, call, Ident::TheConfig.timeout); ++ state->connWait.start(connOpener, call);

• }
• void

+diff --git a/src/log/TcpLogger.cc b/src/log/TcpLogger.cc +index 2be2f6aea..3be1cc5db 100644 +--- a/src/log/TcpLogger.cc ++++ b/src/log/TcpLogger.cc +@@ -256,13 +256,16 @@ Log::TcpLogger::doConnect()

• typedef CommCbMemFunT<TcpLogger, CommConnectCbParams> Dialer;
  

• AsyncCall::Pointer call = JobCallback(MY_DEBUG_SECTION, 5, Dialer, this, Log::TcpLogger::connectDone);
  

+- AsyncJob::Start(new Comm::ConnOpener(futureConn, call, 2)); ++ const auto cs = new Comm::ConnOpener(futureConn, call, 2); ++ connWait.start(cs, call);

• }
• /// Comm::ConnOpener callback
• void
• Log::TcpLogger::connectDone(const CommConnectCbParams &params)
• {

++ connWait.finish(); ++

• if (params.flag != Comm::OK) {
  

•     const double delay = 0.5; // seconds
  

•     if (connectFailures++ % 100 == 0) {
  

+@@ -367,7 +370,10 @@ Log::TcpLogger::handleClosure(const CommCloseCbParams &)

• {

• assert(inCall != NULL);
  

• closer = NULL;
  

+- conn = NULL; ++ if (conn) { ++ conn->noteClosure(); ++ conn = nullptr; ++ }

• // in all current use cases, we should not try to reconnect
  

• mustStop("Log::TcpLogger::handleClosure");
  

• }

+diff --git a/src/log/TcpLogger.h b/src/log/TcpLogger.h +index 1be2113fe..ec7ca5f7b 100644 +--- a/src/log/TcpLogger.h ++++ b/src/log/TcpLogger.h +@@ -10,6 +10,8 @@

• #define _SQUID_SRC_LOG_TCPLOGGER_H
• #include "base/AsyncJob.h"

++#include "base/JobWait.h" ++#include "comm/forward.h"

• #include "ip/Address.h"
• #include <list>

+@@ -103,6 +105,9 @@ private:

• Ip::Address remote; ///< where the remote logger expects our records
  

• AsyncCall::Pointer closer; ///< handles unexpected/external conn closures
  

++ /// waits for a connection to the remote logger to be established/opened ++ JobWaitComm::ConnOpener connWait; ++

• uint64_t connectFailures; ///< number of sequential connection failures
  

• uint64_t drops; ///< number of records dropped during the current outage
  

• };

+diff --git a/src/mgr/Forwarder.cc b/src/mgr/Forwarder.cc +index 8edba1d78..e4da4ca08 100644 +--- a/src/mgr/Forwarder.cc ++++ b/src/mgr/Forwarder.cc +@@ -100,7 +100,11 @@ void

• Mgr::Forwarder::noteCommClosed(const CommCloseCbParams &)
• {

• debugs(16, 5, HERE);
  

+- conn = NULL; // needed? ++ closer = nullptr; ++ if (conn) { ++ conn->noteClosure(); ++ conn = nullptr; ++ }

• mustStop("commClosed");
  

• }

+diff --git a/src/mgr/Inquirer.cc b/src/mgr/Inquirer.cc +index ba41d7aa6..dea0452de 100644 +--- a/src/mgr/Inquirer.cc ++++ b/src/mgr/Inquirer.cc +@@ -107,11 +107,14 @@ Mgr::Inquirer::noteWroteHeader(const CommIoCbParams& params)

• /// called when the HTTP client or some external force closed our socket
• void

+-Mgr::Inquirer::noteCommClosed(const CommCloseCbParams& params) ++Mgr::Inquirer::noteCommClosed(const CommCloseCbParams &)

• {

• debugs(16, 5, HERE);
  

+- Must(!Comm::IsConnOpen(conn) && params.conn.getRaw() == conn.getRaw()); +- conn = NULL; ++ closer = nullptr; ++ if (conn) { ++ conn->noteClosure(); ++ conn = nullptr; ++ }

• mustStop("commClosed");
  

• }

+diff --git a/src/mgr/StoreToCommWriter.cc b/src/mgr/StoreToCommWriter.cc +index e2cccec2f..d3cf73888 100644 +--- a/src/mgr/StoreToCommWriter.cc ++++ b/src/mgr/StoreToCommWriter.cc +@@ -131,7 +131,11 @@ void

• Mgr::StoreToCommWriter::noteCommClosed(const CommCloseCbParams &)
• {

• debugs(16, 6, HERE);
  

+- Must(!Comm::IsConnOpen(clientConnection)); ++ if (clientConnection) { ++ clientConnection->noteClosure(); ++ clientConnection = nullptr; ++ } ++ closer = nullptr;

• mustStop("commClosed");
  

• }

+diff --git a/src/security/PeerConnector.cc b/src/security/PeerConnector.cc +index 4c1401f6e..494edabb8 100644 +--- a/src/security/PeerConnector.cc ++++ b/src/security/PeerConnector.cc +@@ -89,9 +89,15 @@ Security::PeerConnector::start()

• void
• Security::PeerConnector::commCloseHandler(const CommCloseCbParams &params)
• {

++ debugs(83, 5, "FD " << params.fd << ", Security::PeerConnector=" << params.data); ++

• closeHandler = nullptr;
  

++ if (serverConn) { ++ countFailingConnection(); ++ serverConn->noteClosure(); ++ serverConn = nullptr; ++ }

+- debugs(83, 5, "FD " << params.fd << ", Security::PeerConnector=" << params.data);

• const auto err = new ErrorState(ERR_SECURE_CONNECT_FAIL, Http::scServiceUnavailable, request.getRaw(), al);
  

• static const auto d = MakeNamedErrorDetail("TLS_CONNECT_CLOSE");
  

• err->detailError(d);
  

+@@ -111,6 +117,8 @@ Security::PeerConnector::commTimeoutHandler(const CommTimeoutCbParams &)

• bool
• Security::PeerConnector::initialize(Security::SessionPointer &serverSession)
• {

++ Must(Comm::IsConnOpen(serverConnection())); ++

• Security::ContextPointer ctx(getTlsContext());
  

• debugs(83, 5, serverConnection() << ", ctx=" << (void*)ctx.get());
  

+@@ -162,6 +170,8 @@ Security::PeerConnector::initialize(Security::SessionPointer &serverSession)

• void
• Security::PeerConnector::recordNegotiationDetails()
• {

++ Must(Comm::IsConnOpen(serverConnection())); ++

• const int fd = serverConnection()->fd;
  

• Security::SessionPointer session(fd_table[fd].ssl);
  

+@@ -180,6 +190,8 @@ Security::PeerConnector::recordNegotiationDetails()

• void
• Security::PeerConnector::negotiate()
• {

++ Must(Comm::IsConnOpen(serverConnection())); ++

• const int fd = serverConnection()->fd;
  

• if (fd_table[fd].closing())
  

•     return;
  

+@@ -224,7 +236,7 @@ Security::PeerConnector::handleNegotiationResult(const Security::IoResult &resul

• switch (result.category) {
  

• case Security::IoResult::ioSuccess:
  

•     recordNegotiationDetails();
  

+- if (sslFinalized()) ++ if (sslFinalized() && callback)

•         sendSuccess();
  

•     return; // we may be gone by now
  

+@@ -252,6 +264,7 @@ Security::PeerConnector::sslFinalized()

• {
• #if USE_OPENSSL

• if (Ssl::TheConfig.ssl_crt_validator && useCertValidator_) {
  

++ Must(Comm::IsConnOpen(serverConnection()));

•     const int fd = serverConnection()->fd;
  

•     Security::SessionPointer session(fd_table[fd].ssl);
  

+@@ -295,6 +308,7 @@ void

• Security::PeerConnector::sslCrtvdHandleReply(Ssl::CertValidationResponse::Pointer validationResponse)
• {

• Must(validationResponse != NULL);
  

++ Must(Comm::IsConnOpen(serverConnection()));

• ErrorDetail::Pointer errDetails;
  

• bool validatorFailed = false;
  

+@@ -317,7 +331,8 @@ Security::PeerConnector::sslCrtvdHandleReply(Ssl::CertValidationResponse::Pointe

• if (!errDetails && !validatorFailed) {
  

•     noteNegotiationDone(NULL);
  

+- sendSuccess(); ++ if (callback) ++ sendSuccess();

•     return;
  

• }
  

+@@ -343,6 +358,8 @@ Security::PeerConnector::sslCrtvdHandleReply(Ssl::CertValidationResponse::Pointe

• Security::CertErrors *
• Security::PeerConnector::sslCrtvdCheckForErrors(Ssl::CertValidationResponse const &resp, ErrorDetail::Pointer &errDetails)
• {

++ Must(Comm::IsConnOpen(serverConnection())); ++

• ACLFilledChecklist *check = NULL;
  

• Security::SessionPointer session(fd_table[serverConnection()->fd].ssl);
  

+@@ -418,9 +435,11 @@ Security::PeerConnector::negotiateSsl()

• void
• Security::PeerConnector::noteWantRead()
• {

+- const int fd = serverConnection()->fd;

• debugs(83, 5, serverConnection());
  

++ Must(Comm::IsConnOpen(serverConnection())); ++ const int fd = serverConnection()->fd; ++

• // read timeout to avoid getting stuck while reading from a silent server
  

• typedef CommCbMemFunT<Security::PeerConnector, CommTimeoutCbParams> TimeoutDialer;
  

• AsyncCall::Pointer timeoutCall = JobCallback(83, 5,
  

+@@ -434,8 +453,10 @@ Security::PeerConnector::noteWantRead()

• void
• Security::PeerConnector::noteWantWrite()
• {

+- const int fd = serverConnection()->fd;

• debugs(83, 5, serverConnection());
  

++ Must(Comm::IsConnOpen(serverConnection())); ++ ++ const int fd = serverConnection()->fd;

• Comm::SetSelect(fd, COMM_SELECT_WRITE, &NegotiateSsl, new Pointer(this), 0);
  

• return;
  

• }

+@@ -452,57 +473,76 @@ Security::PeerConnector::noteNegotiationError(const Security::ErrorDetailPointer

• bail(anErr);
  

• }

++Security::EncryptorAnswer & ++Security::PeerConnector::answer() ++{ ++ assert(callback); ++ const auto dialer = dynamic_cast<CbDialer*>(callback->getDialer()); ++ assert(dialer); ++ return dialer->answer(); ++} ++

• void
• Security::PeerConnector::bail(ErrorState *error)
• {

• Must(error); // or the recepient will not know there was a problem
  

+- Must(callback != NULL); +- CbDialer *dialer = dynamic_cast<CbDialer*>(callback->getDialer()); +- Must(dialer); +- dialer->answer().error = error; ++ answer().error = error;

+- if (const auto p = serverConnection()->getPeer()) +- peerConnectFailed(p); ++ if (const auto failingConnection = serverConn) { ++ countFailingConnection(); ++ disconnect(); ++ failingConnection->close(); ++ }

• callBack();
  

+- disconnect(); +- +- if (noteFwdPconnUse) +- fwdPconnPool->noteUses(fd_table[serverConn->fd].pconn.uses); +- serverConn->close(); +- serverConn = nullptr;

• }
• void
• Security::PeerConnector::sendSuccess()
• {

+- callBack(); ++ assert(Comm::IsConnOpen(serverConn)); ++ answer().conn = serverConn;

• disconnect();
  

++ callBack(); ++} ++ ++void ++Security::PeerConnector::countFailingConnection() ++{ ++ assert(serverConn); ++ if (const auto p = serverConn->getPeer()) ++ peerConnectFailed(p); ++ // TODO: Calling PconnPool::noteUses() should not be our responsibility. ++ if (noteFwdPconnUse && serverConn->isOpen()) ++ fwdPconnPool->noteUses(fd_table[serverConn->fd].pconn.uses);

• }
• void
• Security::PeerConnector::disconnect()
• {

++ const auto stillOpen = Comm::IsConnOpen(serverConn); ++

• if (closeHandler) {
  

+- comm_remove_close_handler(serverConnection()->fd, closeHandler); ++ if (stillOpen) ++ comm_remove_close_handler(serverConn->fd, closeHandler);

•     closeHandler = nullptr;
  

• }
  

+- commUnsetConnTimeout(serverConnection()); ++ if (stillOpen) ++ commUnsetConnTimeout(serverConn); ++ ++ serverConn = nullptr;

• }
• void
• Security::PeerConnector::callBack()
• {

+- debugs(83, 5, "TLS setup ended for " << serverConnection()); ++ debugs(83, 5, "TLS setup ended for " << answer().conn);

• AsyncCall::Pointer cb = callback;
  

• // Do this now so that if we throw below, swanSong() assert that we _tried_
  

• // to call back holds.
  

• callback = NULL; // this should make done() true
  

+- CbDialer *dialer = dynamic_cast<CbDialer*>(cb->getDialer()); +- Must(dialer); +- dialer->answer().conn = serverConnection();

• ScheduleCallHere(cb);
  

• }

+@@ -511,8 +551,9 @@ Security::PeerConnector::swanSong()

• {

• // XXX: unregister fd-closure monitoring and CommSetSelect interest, if any
  

• AsyncJob::swanSong();
  

+- if (callback != NULL) { // paranoid: we have left the caller waiting +- debugs(83, DBG_IMPORTANT, "BUG: Unexpected state while connecting to a cache_peer or origin server"); ++ ++ if (callback) { ++ // job-ending emergencies like handleStopRequest() or callException()

•     const auto anErr = new ErrorState(ERR_GATEWAY_FAILURE, Http::scInternalServerError, request.getRaw(), al);
  

•     bail(anErr);
  

•     assert(!callback);
  

+@@ -533,7 +574,7 @@ Security::PeerConnector::status() const

•     buf.append("Stopped, reason:", 16);
  

•     buf.appendf("%s",stopReason);
  

• }
  

+- if (serverConn != NULL) ++ if (Comm::IsConnOpen(serverConn))

•     buf.appendf(" FD %d", serverConn->fd);
  

• buf.appendf(" %s%u]", id.prefix(), id.value);
  

• buf.terminate();
  

+@@ -581,15 +622,18 @@ Security::PeerConnector::startCertDownloading(SBuf &url)

•                                   PeerConnectorCertDownloaderDialer(&Security::PeerConnector::certDownloadingDone, this));
  

• const auto dl = new Downloader(url, certCallback, XactionInitiator::initCertFetcher, certDownloadNestingLevel() + 1);
  

+- AsyncJob::Start(dl); ++ certDownloadWait.start(dl, certCallback);

• }
• void
• Security::PeerConnector::certDownloadingDone(SBuf &obj, int downloadStatus)
• {

++ certDownloadWait.finish(); ++

• ++certsDownloads;
  

• debugs(81, 5, "Certificate downloading status: " << downloadStatus << " certificate size: " << obj.length());
  

++ Must(Comm::IsConnOpen(serverConnection()));

• const auto &sconn = *fd_table[serverConnection()->fd].ssl;
  

• // Parse Certificate. Assume that it is in DER format.
  

+@@ -642,6 +686,7 @@ Security::PeerConnector::certDownloadingDone(SBuf &obj, int downloadStatus)

• void
• Security::PeerConnector::handleMissingCertificates(const Security::IoResult &ioResult)
• {

++ Must(Comm::IsConnOpen(serverConnection()));

• auto &sconn = *fd_table[serverConnection()->fd].ssl;
  

• // We download the missing certificate(s) once. We would prefer to clear
  

+diff --git a/src/security/PeerConnector.h b/src/security/PeerConnector.h +index 942f8627a..10700d796 100644 +--- a/src/security/PeerConnector.h ++++ b/src/security/PeerConnector.h +@@ -12,6 +12,7 @@

• #include "acl/Acl.h"
• #include "base/AsyncCbdataCalls.h"
• #include "base/AsyncJob.h"

++#include "base/JobWait.h"

• #include "CommCalls.h"
• #include "http/forward.h"
• #include "security/EncryptorAnswer.h"

+@@ -24,6 +25,7 @@

• #include <queue>
• class ErrorState;

++class Downloader;

• class AccessLogEntry;
• typedef RefCount<AccessLogEntry> AccessLogEntryPointer;

+@@ -152,6 +154,9 @@ protected:

• /// a bail(), sendSuccess() helper: stops monitoring the connection
  

• void disconnect();
  

++ /// updates connection usage history before the connection is closed ++ void countFailingConnection(); ++

• /// If called the certificates validator will not used
  

• void bypassCertValidator() {useCertValidator_ = false;}
  

+@@ -159,6 +164,9 @@ protected:

• /// logging
  

• void recordNegotiationDetails();
  

++ /// convenience method to get to the answer fields ++ EncryptorAnswer &answer(); ++

• HttpRequestPointer request; ///< peer connection trigger or cause
  

• Comm::ConnectionPointer serverConn; ///< TCP connection to the peer
  

• AccessLogEntryPointer al; ///< info for the future access.log entry
  

+@@ -203,6 +211,8 @@ private:

• /// outcome of the last (failed and) suspended negotiation attempt (or nil)
  

• Security::IoResultPointer suspendedError_;
  

++ ++ JobWait<Downloader> certDownloadWait; ///< waits for the missing certificate to be downloaded

• };
• } // namespace Security

+diff --git a/src/security/forward.h b/src/security/forward.h +index dce66e397..26225aae0 100644 +--- a/src/security/forward.h ++++ b/src/security/forward.h +@@ -172,6 +172,7 @@ class ParsedOptions {}; // we never parse/use TLS options in this case

• typedef long ParsedPortFlags;
• class PeerConnector;

++class BlindPeerConnector;

• class PeerOptions;
• #if USE_OPENSSL

+diff --git a/src/servers/FtpServer.cc b/src/servers/FtpServer.cc +index 53f822d2f..e0a10b6b5 100644 +--- a/src/servers/FtpServer.cc ++++ b/src/servers/FtpServer.cc +@@ -61,7 +61,7 @@ Ftp::Server::Server(const MasterXaction::Pointer &xact):

• dataConn(),
  

• uploadAvailSize(0),
  

• listener(),
  

+- connector(), ++ dataConnWait(),

• reader(),
  

• waitingForOrigin(false),
  

• originDataDownloadAbortedOnError(false)
  

+@@ -1676,11 +1676,11 @@ Ftp::Server::checkDataConnPre()

• // active transfer: open a data connection from Squid to client
  

• typedef CommCbMemFunT<Server, CommConnectCbParams> Dialer;
  

+- connector = JobCallback(17, 3, Dialer, this, Ftp::Server::connectedForData); +- Comm::ConnOpener *cs = new Comm::ConnOpener(dataConn, connector, +- Config.Timeout.connect); +- AsyncJob::Start(cs); +- return false; // ConnStateData::processFtpRequest waits handleConnectDone ++ AsyncCall::Pointer callback = JobCallback(17, 3, Dialer, this, Ftp::Server::connectedForData); ++ const auto cs = new Comm::ConnOpener(dataConn->cloneProfile(), callback, ++ Config.Timeout.connect); ++ dataConnWait.start(cs, callback); ++ return false;

• }
• /// Check that client data connection is ready for immediate I/O.

+@@ -1698,18 +1698,22 @@ Ftp::Server::checkDataConnPost() const

• void
• Ftp::Server::connectedForData(const CommConnectCbParams &params)
• {

+- connector = NULL; ++ dataConnWait.finish();

• if (params.flag != Comm::OK) {
  

+- /* it might have been a timeout with a partially open link */ +- if (params.conn != NULL) +- params.conn->close();

•     setReply(425, "Cannot open data connection.");
  

•     Http::StreamPointer context = pipeline.front();
  

•     Must(context->http);
  

•     Must(context->http->storeEntry() != NULL);
  

++ // TODO: call closeDataConnection() to reset data conn processing?

• } else {
  

+- Must(dataConn == params.conn); ++ // Finalize the details and start owning the supplied connection. ++ assert(params.conn); ++ assert(dataConn); ++ assert(!dataConn->isOpen()); ++ dataConn = params.conn; ++ // XXX: Missing comm_add_close_handler() to track external closures. ++

•     Must(Comm::IsConnOpen(params.conn));
  

•     fd_note(params.conn->fd, "active client ftp data");
  

• }
  

+diff --git a/src/servers/FtpServer.h b/src/servers/FtpServer.h +index fb9ef9081..d67799827 100644 +--- a/src/servers/FtpServer.h ++++ b/src/servers/FtpServer.h +@@ -11,8 +11,10 @@

• #ifndef SQUID_SERVERS_FTP_SERVER_H
• #define SQUID_SERVERS_FTP_SERVER_H

++#include "base/JobWait.h"

• #include "base/Lock.h"
• #include "client_side.h"

++#include "comm/forward.h"

• namespace Ftp
• {

+@@ -188,7 +190,11 @@ private:

• size_t uploadAvailSize; ///< number of yet unused uploadBuf bytes
  

• AsyncCall::Pointer listener; ///< set when we are passively listening
  

+- AsyncCall::Pointer connector; ///< set when we are actively connecting ++ ++ /// Waits for an FTP data connection to the client to be established/opened. ++ /// This wait only happens in FTP active mode (via PORT or EPRT). ++ JobWaitComm::ConnOpener dataConnWait; ++

• AsyncCall::Pointer reader; ///< set when we are reading FTP data
  

• /// whether we wait for the origin data transfer to end
  

+diff --git a/src/snmp/Forwarder.cc b/src/snmp/Forwarder.cc +index 836eb5772..259bb0441 100644 +--- a/src/snmp/Forwarder.cc ++++ b/src/snmp/Forwarder.cc +@@ -53,6 +53,7 @@ Snmp::Forwarder::noteCommClosed(const CommCloseCbParams& params)

• {

• debugs(49, 5, HERE);
  

• Must(fd == params.fd);
  

++ closer = nullptr;

• fd = -1;
  

• mustStop("commClosed");
  

• }

+@@ -68,8 +69,7 @@ void

• Snmp::Forwarder::handleException(const std::exception& e)
• {

• debugs(49, 3, HERE << e.what());
  

+- if (fd >= 0) +- sendError(SNMP_ERR_GENERR); ++ sendError(SNMP_ERR_GENERR);

• Ipc::Forwarder::handleException(e);
  

• }

+@@ -78,6 +78,10 @@ void

• Snmp::Forwarder::sendError(int error)
• {

• debugs(49, 3, HERE);
  

++ ++ if (fd < 0) ++ return; // client gone ++

• Snmp::Request& req = static_cast<Snmp::Request&>(*request);
  

• req.pdu.command = SNMP_PDU_RESPONSE;
  

• req.pdu.errstat = error;
  

+diff --git a/src/snmp/Inquirer.cc b/src/snmp/Inquirer.cc +index 9b6e34405..82f8c4475 100644 +--- a/src/snmp/Inquirer.cc ++++ b/src/snmp/Inquirer.cc +@@ -88,7 +88,11 @@ Snmp::Inquirer::noteCommClosed(const CommCloseCbParams& params)

• {

• debugs(49, 5, HERE);
  

• Must(!Comm::IsConnOpen(conn) || conn->fd == params.conn->fd);
  

+- conn = NULL; ++ closer = nullptr; ++ if (conn) { ++ conn->noteClosure(); ++ conn = nullptr; ++ }

• mustStop("commClosed");
  

• }

+@@ -102,6 +106,10 @@ void

• Snmp::Inquirer::sendResponse()
• {

• debugs(49, 5, HERE);
  

++ ++ if (!Comm::IsConnOpen(conn)) ++ return; // client gone ++

• aggrPdu.fixAggregate();
  

• aggrPdu.command = SNMP_PDU_RESPONSE;
  

• u_char buffer[SNMP_REQUEST_SIZE];
  

+diff --git a/src/ssl/PeekingPeerConnector.cc b/src/ssl/PeekingPeerConnector.cc +index 9a4055355..89e45435b 100644 +--- a/src/ssl/PeekingPeerConnector.cc ++++ b/src/ssl/PeekingPeerConnector.cc +@@ -27,18 +27,18 @@ CBDATA_NAMESPACED_CLASS_INIT(Ssl, PeekingPeerConnector);

• void switchToTunnel(HttpRequest *request, const Comm::ConnectionPointer &clientConn, const Comm::ConnectionPointer &srvConn, const SBuf &preReadServerData);
• void

+-Ssl::PeekingPeerConnector::cbCheckForPeekAndSpliceDone(Acl::Answer answer, void *data) ++Ssl::PeekingPeerConnector::cbCheckForPeekAndSpliceDone(const Acl::Answer aclAnswer, void *data)

• {

• Ssl::PeekingPeerConnector *peerConnect = (Ssl::PeekingPeerConnector *) data;
  

• // Use job calls to add done() checks and other job logic/protections.
  

+- CallJobHere1(83, 7, CbcPointer<PeekingPeerConnector>(peerConnect), Ssl::PeekingPeerConnector, checkForPeekAndSpliceDone, answer); ++ CallJobHere1(83, 7, CbcPointer<PeekingPeerConnector>(peerConnect), Ssl::PeekingPeerConnector, checkForPeekAndSpliceDone, aclAnswer);

• }
• void

+-Ssl::PeekingPeerConnector::checkForPeekAndSpliceDone(Acl::Answer answer) ++Ssl::PeekingPeerConnector::checkForPeekAndSpliceDone(const Acl::Answer aclAnswer)

• {

+- const Ssl::BumpMode finalAction = answer.allowed() ? +- static_castSsl::BumpMode(answer.kind): ++ const Ssl::BumpMode finalAction = aclAnswer.allowed() ? ++ static_castSsl::BumpMode(aclAnswer.kind):

•                                   checkForPeekAndSpliceGuess();
  

• checkForPeekAndSpliceMatched(finalAction);
  

• }

+@@ -106,10 +106,8 @@ Ssl::PeekingPeerConnector::checkForPeekAndSpliceMatched(const Ssl::BumpMode acti

•     splice = true;
  

•     // Ssl Negotiation stops here. Last SSL checks for valid certificates
  

•     // and if done, switch to tunnel mode
  

+- if (sslFinalized()) { +- debugs(83,5, "Abort NegotiateSSL on FD " << serverConn->fd << " and splice the connection"); ++ if (sslFinalized() && callback)

•         callBack();
  

+- }

• }
  

• }

+@@ -272,8 +270,11 @@ Ssl::PeekingPeerConnector::startTunneling()

• auto b = SSL_get_rbio(session.get());
  

• auto srvBio = static_cast<Ssl::ServerBio*>(BIO_get_data(b));
  

++ debugs(83, 5, "will tunnel instead of negotiating TLS");

• switchToTunnel(request.getRaw(), clientConn, serverConn, srvBio->rBufData());
  

+- tunnelInsteadOfNegotiating(); ++ answer().tunneled = true; ++ disconnect(); ++ callBack();

• }
• void

+@@ -397,13 +398,3 @@ Ssl::PeekingPeerConnector::serverCertificateVerified()

• }
  

• }

+-void +-Ssl::PeekingPeerConnector::tunnelInsteadOfNegotiating() +-{ +- Must(callback != NULL); +- CbDialer *dialer = dynamic_cast<CbDialer*>(callback->getDialer()); +- Must(dialer); +- dialer->answer().tunneled = true; +- debugs(83, 5, "The SSL negotiation with server aborted"); +-} +- +diff --git a/src/ssl/PeekingPeerConnector.h b/src/ssl/PeekingPeerConnector.h +index 3c86b887d..0145e77f9 100644 +--- a/src/ssl/PeekingPeerConnector.h ++++ b/src/ssl/PeekingPeerConnector.h +@@ -51,7 +51,7 @@ public:

• void checkForPeekAndSplice();
  

• /// Callback function for ssl_bump acl check in step3  SSL bump step.
  

+- void checkForPeekAndSpliceDone(Acl::Answer answer); ++ void checkForPeekAndSpliceDone(Acl::Answer);

• /// Handles the final bumping decision.
  

• void checkForPeekAndSpliceMatched(const Ssl::BumpMode finalMode);
  

+@@ -67,7 +67,7 @@ public:

• void startTunneling();
  

• /// A wrapper function for checkForPeekAndSpliceDone for use with acl
  

+- static void cbCheckForPeekAndSpliceDone(Acl::Answer answer, void *data); ++ static void cbCheckForPeekAndSpliceDone(Acl::Answer, void *data);

• private:

+diff --git a/src/tests/stub_libcomm.cc b/src/tests/stub_libcomm.cc +index 45aa6deb2..8fdb40bbd 100644 +--- a/src/tests/stub_libcomm.cc ++++ b/src/tests/stub_libcomm.cc +@@ -22,9 +22,9 @@ void Comm::AcceptLimiter::kick() STUB

• #include "comm/Connection.h"
• Comm::Connection::Connection() STUB
• Comm::Connection::~Connection() STUB

+-Comm::ConnectionPointer Comm::Connection::cloneIdentDetails() const STUB_RETVAL(nullptr) +-Comm::ConnectionPointer Comm::Connection::cloneDestinationDetails() const STUB_RETVAL(nullptr) ++Comm::ConnectionPointer Comm::Connection::cloneProfile() const STUB_RETVAL(nullptr)

• void Comm::Connection::close() STUB

++void Comm::Connection::noteClosure() STUB

• CachePeer * Comm::Connection::getPeer() const STUB_RETVAL(NULL)
• void Comm::Connection::setPeer(CachePeer * p) STUB
• ScopedId Comm::Connection::codeContextGist() const STUB_RETVAL(id.detach())

+diff --git a/src/tests/stub_libsecurity.cc b/src/tests/stub_libsecurity.cc +index aa9cf0b36..176cf337e 100644 +--- a/src/tests/stub_libsecurity.cc ++++ b/src/tests/stub_libsecurity.cc +@@ -9,6 +9,7 @@

• #include "squid.h"
• #include "AccessLogEntry.h"
• #include "comm/Connection.h"

++#include "Downloader.h"

• #include "HttpRequest.h"
• #define STUB_API "security/libsecurity.la"

+@@ -88,7 +89,9 @@ void PeerConnector::bail(ErrorState *) STUB

• void PeerConnector::sendSuccess() STUB
• void PeerConnector::callBack() STUB
• void PeerConnector::disconnect() STUB

++void PeerConnector::countFailingConnection() STUB

• void PeerConnector::recordNegotiationDetails() STUB

++EncryptorAnswer &PeerConnector::answer() STUB_RETREF(EncryptorAnswer)

• }
• #include "security/PeerOptions.h"

+diff --git a/src/tunnel.cc b/src/tunnel.cc +index 386a0ecd6..4fc5abdef 100644 +--- a/src/tunnel.cc ++++ b/src/tunnel.cc +@@ -11,6 +11,7 @@

• #include "squid.h"
• #include "acl/FilledChecklist.h"
• #include "base/CbcPointer.h"

++#include "base/JobWait.h"

• #include "CachePeer.h"
• #include "cbdata.h"
• #include "client_side.h"

+@@ -180,9 +181,6 @@ public:

• SBuf preReadClientData;
  

• SBuf preReadServerData;
  

• time_t startTime; ///< object creation time, before any peer selection/connection attempts
  

+- /// Whether we are waiting for the CONNECT request/response exchange with the peer. +- bool waitingForConnectExchange; +- HappyConnOpenerPointer connOpener; ///< current connection opening job

• ResolvedPeersPointer destinations; ///< paths for forwarding the request
  

• bool destinationsFound; ///< At least one candidate path found
  

• /// whether another destination may be still attempted if the TCP connection
  

+@@ -191,10 +189,15 @@ public:

• // TODO: remove after fixing deferred reads in TunnelStateData::copyRead()
  

• CodeContext::Pointer codeContext; ///< our creator context
  

+- // AsyncCalls which we set and may need cancelling. +- struct { +- AsyncCall::Pointer connector; ///< a call linking us to the ConnOpener producing serverConn. +- } calls; ++ /// waits for a transport connection to the peer to be established/opened ++ JobWait<HappyConnOpener> transportWait; ++ ++ /// waits for the established transport connection to be secured/encrypted ++ JobWaitSecurity::PeerConnector encryptionWait; ++ ++ /// waits for an HTTP CONNECT tunnel through a cache_peer to be negotiated ++ /// over the (encrypted, if needed) transport connection to that cache_peer ++ JobWaitHttp::Tunneler peerWait;

• void copyRead(Connection &from, IOCB *completion);
  

+@@ -212,12 +215,6 @@ public:

• /// when all candidate destinations have been tried and all have failed
  

• void noteConnection(HappyConnOpenerAnswer &);
  

+- /// whether we are waiting for HappyConnOpener +- /// same as calls.connector but may differ from connOpener.valid() +- bool opening() const { return connOpener.set(); } +- +- void cancelOpening(const char *reason); +-

• /// Start using an established connection
  

• void connectDone(const Comm::ConnectionPointer &conn, const char *origin, const bool reused);
  

+@@ -266,6 +263,9 @@ private:

• /// \returns whether the request should be retried (nil) or the description why it should not
  

• const char *checkRetry();
  

++ /// whether the successfully selected path destination or the established ++ /// server connection is still in use ++ bool usingDestination() const;

• /// details of the "last tunneling attempt" failure (if it failed)
  

• ErrorState *savedError = nullptr;
  

+@@ -275,6 +275,8 @@ private:

• void deleteThis();
  

++ void cancelStep(const char *reason); ++

• public:

• bool keepGoingAfterRead(size_t len, Comm::Flag errcode, int xerrno, Connection &from, Connection &to);
  

• void copy(size_t len, Connection &from, Connection &to, IOCB *);
  

+@@ -357,7 +359,6 @@ TunnelStateData::deleteThis()

• TunnelStateData::TunnelStateData(ClientHttpRequest *clientRequest) :

• startTime(squid_curtime),
  

+- waitingForConnectExchange(false),

• destinations(new ResolvedPeers()),
  

• destinationsFound(false),
  

• retriable(true),
  

+@@ -390,8 +391,7 @@ TunnelStateData::~TunnelStateData()

• debugs(26, 3, "TunnelStateData destructed this=" << this);
  

• assert(noConnections());
  

• xfree(url);
  

+- if (opening()) +- cancelOpening("~TunnelStateData"); ++ cancelStep("~TunnelStateData");

• delete savedError;
  

• }

+@@ -904,7 +904,10 @@ TunnelStateData::copyServerBytes()

• static void
• tunnelStartShoveling(TunnelStateData *tunnelState)
• {

+- assert(!tunnelState->waitingForConnectExchange); ++ assert(!tunnelState->transportWait); ++ assert(!tunnelState->encryptionWait); ++ assert(!tunnelState->peerWait); ++

• assert(tunnelState->server.conn);
  

• AsyncCall::Pointer timeoutCall = commCbCall(5, 4, "tunnelTimeout",
  

•                                  CommTimeoutCbPtrFun(tunnelTimeout, tunnelState));
  

+@@ -962,6 +965,7 @@ tunnelConnectedWriteDone(const Comm::ConnectionPointer &conn, char *, size_t len

• void
• TunnelStateData::tunnelEstablishmentDone(Http::TunnelerAnswer &answer)
• {

++ peerWait.finish();

• server.len = 0;
  

• if (logTag_ptr)
  

+@@ -970,13 +974,11 @@ TunnelStateData::tunnelEstablishmentDone(Http::TunnelerAnswer &answer)

• if (answer.peerResponseStatus != Http::scNone)
  

•     *status_ptr = answer.peerResponseStatus;
  

+- waitingForConnectExchange = false; +-

• auto sawProblem = false;
  

• if (!answer.positive()) {
  

•     sawProblem = true;
  

+- Must(!Comm::IsConnOpen(answer.conn)); ++ assert(!answer.conn);

• } else if (!Comm::IsConnOpen(answer.conn) || fd_table[answer.conn->fd].closing()) {
  

•     sawProblem = true;
  

•     closePendingConnection(answer.conn, "conn was closed while waiting for tunnelEstablishmentDone");
  

+@@ -1042,8 +1044,7 @@ tunnelErrorComplete(int fd/*const Comm::ConnectionPointer &*/, void *data, size_

• void
• TunnelStateData::noteConnection(HappyConnOpener::Answer &answer)
• {

+- calls.connector = nullptr; +- connOpener.clear(); ++ transportWait.finish();

• ErrorState *error = nullptr;
  

• if ((error = answer.error.get())) {
  

+@@ -1167,7 +1168,7 @@ TunnelStateData::secureConnectionToPeer(const Comm::ConnectionPointer &conn)

• AsyncCall::Pointer callback = asyncCall(5,4, "TunnelStateData::noteSecurityPeerConnectorAnswer",
  

•                                         MyAnswerDialer(&TunnelStateData::noteSecurityPeerConnectorAnswer, this));
  

• const auto connector = new Security::BlindPeerConnector(request, conn, callback, al);
  

+- AsyncJob::Start(connector); // will call our callback ++ encryptionWait.start(connector, callback);

• }
• /// starts a preparation step for an established connection; retries on failures

+@@ -1194,9 +1195,12 @@ TunnelStateData::advanceDestination(const char *stepDescription, const Comm::Con

• void
• TunnelStateData::noteSecurityPeerConnectorAnswer(Security::EncryptorAnswer &answer)
• {

++ encryptionWait.finish(); ++

• ErrorState *error = nullptr;
  

++ assert(!answer.tunneled);

• if ((error = answer.error.get())) {
  

+- Must(!Comm::IsConnOpen(answer.conn)); ++ assert(!answer.conn);

•     answer.error.clear();
  

• } else if (!Comm::IsConnOpen(answer.conn) || fd_table[answer.conn->fd].closing()) {
  

•     error = new ErrorState(ERR_CANNOT_FORWARD, Http::scServiceUnavailable, request.getRaw(), al);
  

+@@ -1223,8 +1227,6 @@ TunnelStateData::connectedToPeer(const Comm::ConnectionPointer &conn)

• void
• TunnelStateData::establishTunnelThruProxy(const Comm::ConnectionPointer &conn)
• {

+- assert(!waitingForConnectExchange); +-

• AsyncCall::Pointer callback = asyncCall(5,4,
  

•                                         "TunnelStateData::tunnelEstablishmentDone",
  

•                                         Http::Tunneler::CbDialer<TunnelStateData>(&TunnelStateData::tunnelEstablishmentDone, this));
  

+@@ -1232,9 +1234,7 @@ TunnelStateData::establishTunnelThruProxy(const Comm::ConnectionPointer &conn)

• #if USE_DELAY_POOLS

• tunneler->setDelayId(server.delayId);
  

• #endif

+- AsyncJob::Start(tunneler); +- waitingForConnectExchange = true; +- // and wait for the tunnelEstablishmentDone() call ++ peerWait.start(tunneler, callback);

• }
• void

+@@ -1252,14 +1252,14 @@ TunnelStateData::noteDestination(Comm::ConnectionPointer path)

• destinations->addPath(path);
  

+- if (Comm::IsConnOpen(server.conn)) { ++ if (usingDestination()) {

•     // We are already using a previously opened connection but also
  

•     // receiving destinations in case we need to re-forward.
  

+- Must(!opening()); ++ Must(!transportWait);

•     return;
  

• }
  

+- if (opening()) { ++ if (transportWait) {

•     notifyConnOpener();
  

•     return; // and continue to wait for tunnelConnectDone() callback
  

• }
  

+@@ -1289,17 +1289,23 @@ TunnelStateData::noteDestinationsEnd(ErrorState *selectionError)

• // if all of them fail, tunneling as whole will fail
  

• Must(!selectionError); // finding at least one path means selection succeeded
  

+- if (Comm::IsConnOpen(server.conn)) { ++ if (usingDestination()) {

•     // We are already using a previously opened connection but also
  

•     // receiving destinations in case we need to re-forward.
  

+- Must(!opening()); ++ Must(!transportWait);

•     return;
  

• }
  

+- Must(opening()); // or we would be stuck with nothing to do or wait for ++ Must(transportWait); // or we would be stuck with nothing to do or wait for

• notifyConnOpener();
  

• }

++bool ++TunnelStateData::usingDestination() const ++{ ++ return encryptionWait || peerWait || Comm::IsConnOpen(server.conn); ++} ++

• /// remembers an error to be used if there will be no more connection attempts
• void
• TunnelStateData::saveError(ErrorState *error)

+@@ -1320,8 +1326,7 @@ TunnelStateData::sendError(ErrorState *finalError, const char *reason)

• if (request)
  

•     request->hier.stopPeerClock(false);
  

+- if (opening()) +- cancelOpening(reason); ++ cancelStep(reason);

• assert(finalError);
  

+@@ -1339,18 +1344,15 @@ TunnelStateData::sendError(ErrorState *finalError, const char *reason)

• errorSend(client.conn, finalError);
  

• }

+-/// Notify connOpener that we no longer need connections. We do not have to do +-/// this -- connOpener would eventually notice on its own, but notifying reduces +-/// waste and speeds up spare connection opening for other transactions (that +-/// could otherwise wait for this transaction to use its spare allowance). ++/// Notify a pending subtask, if any, that we no longer need its help. We do not ++/// have to do this -- the subtask job will eventually end -- but ending it ++/// earlier reduces waste and may reduce DoS attack surface.

• void

+-TunnelStateData::cancelOpening(const char *reason) ++TunnelStateData::cancelStep(const char *reason)

• {

+- assert(calls.connector); +- calls.connector->cancel(reason); +- calls.connector = nullptr; +- notifyConnOpener(); +- connOpener.clear(); ++ transportWait.cancel(reason); ++ encryptionWait.cancel(reason); ++ peerWait.cancel(reason);

• }
• void

+@@ -1360,15 +1362,14 @@ TunnelStateData::startConnecting()

•     request->hier.startPeerClock();
  

• assert(!destinations->empty());
  

+- assert(!opening()); +- calls.connector = asyncCall(17, 5, "TunnelStateData::noteConnection", HappyConnOpener::CbDialer<TunnelStateData>(&TunnelStateData::noteConnection, this)); +- const auto cs = new HappyConnOpener(destinations, calls.connector, request, startTime, 0, al); ++ assert(!usingDestination()); ++ AsyncCall::Pointer callback = asyncCall(17, 5, "TunnelStateData::noteConnection", HappyConnOpener::CbDialer<TunnelStateData>(&TunnelStateData::noteConnection, this)); ++ const auto cs = new HappyConnOpener(destinations, callback, request, startTime, 0, al);

• cs->setHost(request->url.host());
  

• cs->setRetriable(false);
  

• cs->allowPersistent(false);
  

• destinations->notificationPending = true; // start() is async
  

+- connOpener = cs; +- AsyncJob::Start(cs); ++ transportWait.start(cs, callback);

• }
• /// send request on an existing connection dedicated to the requesting client

+@@ -1417,7 +1418,7 @@ TunnelStateData::Connection::setDelayId(DelayId const &newDelay)

• #endif

+-/// makes sure connOpener knows that destinations have changed ++/// makes sure connection opener knows that the destinations have changed

• void
• TunnelStateData::notifyConnOpener()
• {

+@@ -1425,7 +1426,7 @@ TunnelStateData::notifyConnOpener()

•     debugs(17, 7, "reusing pending notification");
  

• } else {
  

•     destinations->notificationPending = true;
  

+- CallJobHere(17, 5, connOpener, HappyConnOpener, noteCandidatesChange); ++ CallJobHere(17, 5, transportWait.job(), HappyConnOpener, noteCandidatesChange);

• }
  

• }

+diff --git a/src/whois.cc b/src/whois.cc +index 38dfbacde..306ea5c62 100644 +--- a/src/whois.cc ++++ b/src/whois.cc +@@ -182,6 +182,7 @@ whoisClose(const CommCloseCbParams &params)

• {

• WhoisState *p = (WhoisState *)params.data;
  

• debugs(75, 3, "whoisClose: FD " << params.fd);
  

++ // We do not own a Connection. Assume that FwdState is also monitoring.

• p->entry->unlock("whoisClose");
  

• delete p;
  

• }