Author: Alex Rousskov <rousskov@measurement-factory.com>

Bug 2964: Prevent memory leaks when ICAP transactions fail.

We now make sure that heap-allocated objects are deleted if an exception
is thrown before the object pointers are saved/registered in a safe location
like a data member.

Assigning state.serviceWaiting=true after calling callWhenReady() in ModXact
prevents ModXact leak when callWhenReady() throws. This may need more work
to mark ModXact state appropriately for the adaptation log.

Based on lp 3p1-rock branch, r9610.

Added doneWithRetries() and used it to inform the request body sender that
there will be no more body consumers. This allows the sender (e.g., an ICAP
REQMOD transaction) to quit instead of waiting for body buffer space forever.

Moved !self check into checkRetry() because we need to call doneWithRetries()
even if self is nil. The move should not affect the old code.

Based on lp 3p1-rock branch, r9613.

At the end of preview, do not go into the writingPaused state if we already
received the final response from the ICAP server. Instead, stop writing so
that we do not get stuck waiting for the response that has already come.

May also handle header-only (zero-size) ieof Preview better.

TODO: Convert other HttpMsg pointer members to use safe HttpMsg::Pointer.

diff --git a/src/adaptation/icap/ModXact.cc b/src/adaptation/icap/ModXact.cc
index e6747b46268a746ce06da29e1fadd59b680385b1..b7acfa5421bad59899fdaf5502fc1cb2813e5820 100644 (file)
--- a/src/adaptation/icap/ModXact.cc
+++ b/src/adaptation/icap/ModXact.cc
@@ -61,7 +61,7 @@ Adaptation::Icap::ModXact::ModXact(HttpMsg *virginHeader,
     // nothing to do because we are using temporary buffers
 
     // parsing; TODO: do not set until we parse, see ICAPOptXact
-    icapReply = HTTPMSGLOCK(new HttpReply);
+    icapReply = new HttpReply;
     icapReply->protoPrefix = "ICAP/"; // TODO: make an IcapReply class?
 
     debugs(93,7, HERE << "initialized." << status());
@@ -93,11 +93,11 @@ void Adaptation::Icap::ModXact::waitForService()
 {
     Must(!state.serviceWaiting);
     debugs(93, 7, HERE << "will wait for the ICAP service" << status());
-    state.serviceWaiting = true;
     typedef NullaryMemFunT<ModXact> Dialer;
     AsyncCall::Pointer call = JobCallback(93,5,
                                           Dialer, this, Adaptation::Icap::ModXact::noteServiceReady);
     service().callWhenReady(call);
+    state.serviceWaiting = true; // after callWhenReady() which may throw
 }
 
 void Adaptation::Icap::ModXact::noteServiceReady()
@@ -159,11 +159,14 @@ void Adaptation::Icap::ModXact::handleCommWroteHeaders()
     Must(state.writing == State::writingHeaders);
 
     // determine next step
-    if (preview.enabled())
-        state.writing = preview.done() ? State::writingPaused : State::writingPreview;
-    else if (virginBody.expected())
+    if (preview.enabled()) {
+        if (preview.done())
+            decideWritingAfterPreview("zero-size");
+        else
+            state.writing = State::writingPreview;
+    } else if (virginBody.expected()) {
         state.writing = State::writingPrime;
-    else {
+    } else {
         stopWriting(true);
         return;
     }
@@ -222,14 +225,22 @@ void Adaptation::Icap::ModXact::writePreviewBody()
 
     // change state once preview is written
 
-    if (preview.done()) {
-        debugs(93, 7, HERE << "wrote entire Preview body" << status());
+    if (preview.done())
+        decideWritingAfterPreview("body");
+}
 
-        if (preview.ieof())
-            stopWriting(true);
-        else
-            state.writing = State::writingPaused;
-    }
+/// determine state.writing after we wrote the entire preview
+void Adaptation::Icap::ModXact::decideWritingAfterPreview(const char *kind)
+{
+    if (preview.ieof()) // nothing more to write
+        stopWriting(true);
+    else if (state.parsing == State::psIcapHeader) // did not get a reply yet
+        state.writing = State::writingPaused; // wait for the ICAP server reply
+    else
+        stopWriting(true); // ICAP server reply implies no post-preview writing
+
+    debugs(93, 6, HERE << "decided on writing after " << kind << " preview" <<
+           status());
 }
 
 void Adaptation::Icap::ModXact::writePrimeBody()
@@ -850,32 +861,34 @@ void Adaptation::Icap::ModXact::prepEchoing()
 
     // allocate the adapted message and copy metainfo
     Must(!adapted.header);
-    HttpMsg *newHead = NULL;
+    {
+    HttpMsg::Pointer newHead;
     if (const HttpRequest *oldR = dynamic_cast<const HttpRequest*>(oldHead)) {
-        HttpRequest *newR = new HttpRequest;
+        HttpRequest::Pointer newR(new HttpRequest);
         newR->canonical = oldR->canonical ?
                           xstrdup(oldR->canonical) : NULL; // parse() does not set it
         newHead = newR;
     } else if (dynamic_cast<const HttpReply*>(oldHead)) {
-        HttpReply *newRep = new HttpReply;
-        newHead = newRep;
+        newHead = new HttpReply;
     }
-    Must(newHead);
+    Must(newHead != NULL);
+
     newHead->inheritProperties(oldHead);
 
     adapted.setHeader(newHead);
+    }
 
     // parse the buffer back
     http_status error = HTTP_STATUS_NONE;
 
-    Must(newHead->parse(&httpBuf, true, &error));
+    Must(adapted.header->parse(&httpBuf, true, &error));
 
-    Must(newHead->hdr_sz == httpBuf.contentSize()); // no leftovers
+    Must(adapted.header->hdr_sz == httpBuf.contentSize()); // no leftovers
 
     httpBuf.clean();
 
     debugs(93, 7, HERE << "cloned virgin message " << oldHead << " to " <<
-           newHead);
+           adapted.header);
 
     // setup adapted body pipe if needed
     if (oldHead->body_pipe != NULL) {
@@ -1085,6 +1098,11 @@ void Adaptation::Icap::ModXact::noteBodyConsumerAborted(BodyPipe::Pointer)
     mustStop("adapted body consumer aborted");
 }
 
+Adaptation::Icap::ModXact::~ModXact()
+{
+    delete bodyParser;
+}
+
 // internal cleanup
 void Adaptation::Icap::ModXact::swanSong()
 {
@@ -1296,21 +1314,20 @@ void Adaptation::Icap::ModXact::encapsulateHead(MemBuf &icapBuf, const char *sec
     icapBuf.Printf("%s=%d, ", section, (int) httpBuf.contentSize());
 
     // begin cloning
-    HttpMsg *headClone = NULL;
+    HttpMsg::Pointer headClone;
 
     if (const HttpRequest* old_request = dynamic_cast<const HttpRequest*>(head)) {
-        HttpRequest* new_request = new HttpRequest;
-        assert(old_request->canonical);
+        HttpRequest::Pointer new_request(new HttpRequest);
+        Must(old_request->canonical);
         urlParse(old_request->method, old_request->canonical, new_request);
         new_request->http_ver = old_request->http_ver;
         headClone = new_request;
     } else if (const HttpReply *old_reply = dynamic_cast<const HttpReply*>(head)) {
-        HttpReply* new_reply = new HttpReply;
+        HttpReply::Pointer new_reply(new HttpReply);
         new_reply->sline = old_reply->sline;
         headClone = new_reply;
     }
-
-    Must(headClone);
+    Must(headClone != NULL);
     headClone->inheritProperties(head);
 
     HttpHeaderPos pos = HttpHeaderInitPos;
@@ -1327,7 +1344,7 @@ void Adaptation::Icap::ModXact::encapsulateHead(MemBuf &icapBuf, const char *sec
     // pack polished HTTP header
     packHead(httpBuf, headClone);
 
-    delete headClone;
+    // headClone unlocks and, hence, deletes the message we packed
 }
 
 void Adaptation::Icap::ModXact::packHead(MemBuf &httpBuf, const HttpMsg *head)

diff --git a/src/adaptation/icap/ModXact.h b/src/adaptation/icap/ModXact.h
index fd5bbf08f7f797c3a64bd1dec43d40e9bf82d5cb..0bde0b25d8f56c9a7edea04b6a06e3aaa4b8b33e 100644 (file)
--- a/src/adaptation/icap/ModXact.h
+++ b/src/adaptation/icap/ModXact.h
@@ -137,6 +137,7 @@ class ModXact: public Xaction, public BodyProducer, public BodyConsumer
 
 public:
     ModXact(HttpMsg *virginHeader, HttpRequest *virginCause, ServiceRep::Pointer &s);
+    virtual ~ModXact();
 
     // BodyProducer methods
     virtual void noteMoreBodySpaceAvailable(BodyPipe::Pointer);
@@ -183,6 +184,7 @@ private:
     void writePreviewBody();
     void writePrimeBody();
     void writeSomeBody(const char *label, size_t size);
+    void decideWritingAfterPreview(const char *previewKind);
 
     void startReading();
     void readMore();

diff --git a/src/adaptation/icap/OptXact.cc b/src/adaptation/icap/OptXact.cc
index 09d217292283b9a806b4a40ce952e19b9d1058ff..c8e85509ea527915ccee5572f89d4921580f9ad8 100644 (file)
--- a/src/adaptation/icap/OptXact.cc
+++ b/src/adaptation/icap/OptXact.cc
@@ -65,11 +65,10 @@ void Adaptation::Icap::OptXact::handleCommWrote(size_t size)
 // comm module read a portion of the ICAP response for us
 void Adaptation::Icap::OptXact::handleCommRead(size_t)
 {
-    if (HttpMsg *r = parseResponse()) {
+    if (parseResponse()) {
         icap_tio_finish = current_time;
         setOutcome(xoOpt);
-        sendAnswer(r);
-        icapReply = HTTPMSGLOCK(dynamic_cast<HttpReply*>(r));
+        sendAnswer(icapReply);
         Must(done()); // there should be nothing else to do
         return;
     }
@@ -77,24 +76,23 @@ void Adaptation::Icap::OptXact::handleCommRead(size_t)
     scheduleRead();
 }
 
-HttpMsg *Adaptation::Icap::OptXact::parseResponse()
+bool Adaptation::Icap::OptXact::parseResponse()
 {
     debugs(93, 5, HERE << "have " << readBuf.contentSize() << " bytes to parse" <<
            status());
     debugs(93, 5, HERE << "\n" << readBuf.content());
 
-    HttpReply *r = HTTPMSGLOCK(new HttpReply);
+    HttpReply::Pointer r(new HttpReply);
     r->protoPrefix = "ICAP/"; // TODO: make an IcapReply class?
 
-    if (!parseHttpMsg(r)) { // throws on errors
-        HTTPMSGUNLOCK(r);
-        return 0;
-    }
+    if (!parseHttpMsg(r)) // throws on errors
+        return false;
 
     if (httpHeaderHasConnDir(&r->header, "close"))
         reuseConnection = false;
 
-    return r;
+    icapReply = r;
+    return true;
 }
 
 void Adaptation::Icap::OptXact::swanSong()

diff --git a/src/adaptation/icap/OptXact.h b/src/adaptation/icap/OptXact.h
index 35561996dc5f3f8855741b3f91a6b59bac1206e2..0fee9a78cc9f9c05738ef750e16bfe8300e40c4d 100644 (file)
--- a/src/adaptation/icap/OptXact.h
+++ b/src/adaptation/icap/OptXact.h
@@ -60,7 +60,7 @@ protected:
     virtual void handleCommRead(size_t size);
 
     void makeRequest(MemBuf &buf);
-    HttpMsg *parseResponse();
+    bool parseResponse();
 
     void startReading();
 

diff --git a/src/adaptation/icap/Xaction.cc b/src/adaptation/icap/Xaction.cc
index d063d276ab9845f90dcd364efb233259472e29d7..87a1ca0e55cb8f7142cc325be2264d674bfe2ac7 100644 (file)
--- a/src/adaptation/icap/Xaction.cc
+++ b/src/adaptation/icap/Xaction.cc
@@ -52,7 +52,6 @@ Adaptation::Icap::Xaction::~Xaction()
     debugs(93,3, typeName << " destructed, this=" << this <<
            " [icapx" << id << ']'); // we should not call virtual status() here
     HTTPMSGUNLOCK(icapRequest);
-    HTTPMSGUNLOCK(icapReply);
 }
 
 Adaptation::Icap::ServiceRep &

diff --git a/src/adaptation/icap/Xaction.h b/src/adaptation/icap/Xaction.h
index 4e81dfca917ac1ee96e0ced391fc76f5fc2b2a45..7eb8d010c26d67a96cc8a54b2b8084092553a38c 100644 (file)
--- a/src/adaptation/icap/Xaction.h
+++ b/src/adaptation/icap/Xaction.h
@@ -40,8 +40,8 @@
 #include "adaptation/icap/ServiceRep.h"
 #include "adaptation/Initiate.h"
 #include "AccessLogEntry.h"
+#include "HttpReply.h"
 
-class HttpMsg;
 class CommConnectCbParams;
 
 namespace Adaptation
@@ -80,7 +80,7 @@ public:
 
     // TODO: create these only when actually sending/receiving
     HttpRequest *icapRequest; ///< sent (or at least created) ICAP request
-    HttpReply *icapReply; ///< received ICAP reply, if any
+    HttpReply::Pointer icapReply; ///< received ICAP reply, if any
 
     /// the number of times we tried to get to the service, including this time
     int attempts;

diff --git a/src/client_side.cc b/src/client_side.cc
index a03518a8e6e3322fd6e39211140735b497dc39c2..d6576489e9180cab807dec4647dd9bebbd47ca10 100644 (file)
--- a/src/client_side.cc
+++ b/src/client_side.cc
@@ -575,9 +575,9 @@ ClientHttpRequest::logRequest()
         }
 
         delete checklist;
-
-        accessLogFreeMemory(&al);
     }
+
+    accessLogFreeMemory(&al);
 }
 
 void

diff --git a/src/forward.cc b/src/forward.cc
index 16f5d5b7415d4018be94fabf264091607d278f78..06c90ede799db2988718b4ee4ba5b0724eca3145 100644 (file)
--- a/src/forward.cc
+++ b/src/forward.cc
@@ -164,6 +164,8 @@ FwdState::~FwdState()
 
     serversFree(&servers);
 
+    doneWithRetries();
+
     HTTPMSGUNLOCK(request);
 
     if (err)
@@ -414,6 +416,12 @@ FwdState::checkRetry()
     if (shutting_down)
         return false;
 
+    if (!self) { // we have aborted before the server called us back
+        debugs(17, 5, HERE << "not retrying because of earlier abort");
+        // we will be destroyed when the server clears its Pointer to us
+        return false;
+    }
+
     if (entry->store_status != STORE_PENDING)
         return false;
 
@@ -497,12 +505,6 @@ FwdState::serverClosed(int fd)
 void
 FwdState::retryOrBail()
 {
-    if (!self) { // we have aborted before the server called us back
-        debugs(17, 5, HERE << "not retrying because of earlier abort");
-        // we will be destroyed when the server clears its Pointer to us
-        return;
-    }
-
     if (checkRetry()) {
         int originserver = (servers->_peer == NULL);
         debugs(17, 3, "fwdServerClosed: re-forwarding (" << n_tries << " tries, " << (squid_curtime - start_t) << " secs)");
@@ -539,13 +541,26 @@ FwdState::retryOrBail()
         return;
     }
 
-    if (!err && shutting_down) {
+    // TODO: should we call completed() here and move doneWithRetries there?
+    doneWithRetries();
+
+    if (self != NULL && !err && shutting_down) {
         errorCon(ERR_SHUTTING_DOWN, HTTP_SERVICE_UNAVAILABLE, request);
     }
 
     self = NULL;       // refcounted
 }
 
+// If the Server quits before nibbling at the request body, the body sender
+// will not know (so that we can retry). Call this if we will not retry. We
+// will notify the sender so that it does not get stuck waiting for space.
+void
+FwdState::doneWithRetries()
+{
+    if (request && request->body_pipe != NULL)
+        request->body_pipe->expectNoConsumption();
+}
+
 // called by the server that failed after calling unregister()
 void
 FwdState::handleUnregisteredServerEnd()

diff --git a/src/forward.h b/src/forward.h
index 6dedc73b55e47a8d15823839aae94f6c9d555d4d..d0ff0f3c7de75821f199391b3e3e0e5db0062187 100644 (file)
--- a/src/forward.h
+++ b/src/forward.h
@@ -63,6 +63,7 @@ private:
 
     static void logReplyStatus(int tries, http_status status);
     void updateHierarchyInfo();
+    void doneWithRetries();
     void completed();
     void retryOrBail();
     static void RegisterWithCacheManager(void);