Optimize freeMemorySpace() to not loop in vain

src/MemObject.cc

@@ -100,6 +100,7 @@ MemObject::MemObject()
     ping_reply_callback = nullptr;
     memset(&start_ping, 0, sizeof(start_ping));
     reply_ = new HttpReply;
+    assert(!repl.data);
 }
 
 MemObject::~MemObject()
@@ -132,19 +133,6 @@ MemObject::replaceBaseReply(const HttpReplyPointer &r)
     updatedReply_ = nullptr;
 }
 
-void
-MemObject::write(const StoreIOBuffer &writeBuffer)
-{
-    debugs(19, 6, "memWrite: offset " << writeBuffer.offset << " len " << writeBuffer.length);
-
-    /* We don't separate out mime headers yet, so ensure that the first
-     * write is at offset 0 - where they start
-     */
-    assert (data_hdr.endOffset() || writeBuffer.offset == 0);
-
-    assert (data_hdr.write (writeBuffer));
-}
-
 void
 MemObject::dump() const
 {

src/MemObject.h

@@ -51,7 +51,6 @@ class MemObject
     /// whether setUris() has been called
     bool hasUris() const;
 
-    void write(const StoreIOBuffer &buf);
     void unlinkRequest() { request = nullptr; }
 
     /// HTTP response before 304 (Not Modified) updates

src/MemStore.cc

@@ -569,7 +569,7 @@ MemStore::copyFromShmSlice(StoreEntry &e, const StoreIOBuffer &buf, bool eof)
 
     // local memory stores both headers and body so copy regardless of pstate
     const int64_t offBefore = e.mem_obj->endOffset();
-    assert(e.mem_obj->data_hdr.write(buf)); // from MemObject::write()
+    e.writeData(buf);
     const int64_t offAfter = e.mem_obj->endOffset();
     // expect to write the entire buf because StoreEntry::write() never fails
     assert(offAfter >= 0 && offBefore <= offAfter &&

src/Store.h

@@ -56,6 +56,11 @@ class StoreEntry : public hash_link, public Packable
     /// \see MemObject::freshestReply()
     const HttpReply *hasFreshestReply() const { return mem_obj ? &mem_obj->freshestReply() : nullptr; }
 
+    /// writes to local memory store
+    void writeData(StoreIOBuffer);
+
+    /// writeData() and calls awaiting handlers.
+    /// Should be called for non-completed (STORE_PENDING) entries only.
     void write(StoreIOBuffer);
 
     /** Check if the Store entry is empty
@@ -67,6 +72,9 @@ class StoreEntry : public hash_link, public Packable
     bool isAccepting() const;
     size_t bytesWanted(Range<size_t> const aRange, bool ignoreDelayPool = false) const;
 
+    /// whether a non-completed (STORE_PENDING) entry is being written to the local memory store
+    bool isLocalWriter() const { return locked() && isAccepting(); }
+
     /// Signals that the entire response has been stored and no more append()
     /// calls should be expected; cf. completeTruncated().
     void completeSuccessfully(const char *whyWeAreSureWeStoredTheWholeReply);
@@ -319,6 +327,8 @@ class StoreEntry : public hash_link, public Packable
     /// flags [truncated or too big] entry with ENTRY_BAD_LENGTH and releases it
     void lengthWentBad(const char *reason);
 
+    bool hasReplPolicy() const;
+
     static Mem::Allocator *pool;
 
     unsigned short lock_count;      /* Assume < 65536! */

src/stmem.cc

@@ -15,6 +15,8 @@
 #include "MemObject.h"
 #include "stmem.h"
 
+size_t mem_hdr::ReplPolicyIdleNodesCount = 0;
+
 /*
  * NodeGet() is called to get the data buffer to pass to storeIOWrite().
  * By setting the write_pending flag here we are assuming that there
@@ -58,7 +60,9 @@ mem_hdr::endOffset () const
 void
 mem_hdr::freeContent()
 {
+    const auto initialNodes = size();
     nodes.destroy();
+    updateIdleNodes(initialNodes);
     inmem_hi = 0;
     debugs(19, 9, this << " hi: " << inmem_hi);
 }
@@ -72,8 +76,10 @@ mem_hdr::unlink(mem_node *aNode)
     }
 
     debugs(19, 8, this << " removing " << aNode);
+    const auto initialNodes = size();
     nodes.remove (aNode, NodeCompare);
     delete aNode;
+    updateIdleNodes(initialNodes);
     return true;
 }
 
@@ -320,18 +326,20 @@ mem_hdr::write (StoreIOBuffer const &writeBuffer)
     char *currentSource = writeBuffer.data;
     size_t len = writeBuffer.length;
 
+    const auto initialNodes = size();
     while (len && (target = nodeToRecieve(currentOffset))) {
         size_t wrote = writeAvailable(target, currentOffset, len, currentSource);
         assert (wrote);
         len -= wrote;
         currentOffset += wrote;
         currentSource += wrote;
     }
+    updateIdleNodes(initialNodes);
 
     return true;
 }
 
-mem_hdr::mem_hdr() : inmem_hi(0)
+mem_hdr::mem_hdr() : inmem_hi(0), removableByReplPolicy(false)
 {
     debugs(19, 9, this << " hi: " << inmem_hi);
 }
@@ -376,3 +384,40 @@ mem_hdr::getNodes() const
     return nodes;
 }
 
+void
+mem_hdr::allowedToFreeWithReplPolicy(const bool allowed)
+{
+    if (removableByReplPolicy == allowed)
+        return;
+    removableByReplPolicy = allowed;
+    const auto oldCount = ReplPolicyIdleNodesCount;
+    if (removableByReplPolicy)
+        ReplPolicyIdleNodesCount += size();
+    else {
+        assert(ReplPolicyIdleNodesCount >= size());
+        ReplPolicyIdleNodesCount -= size();
+    }
+    debugs(19, 5, this << " modified ReplPolicyIdleNodesCount from " << oldCount << " to " << ReplPolicyIdleNodesCount);
+}
+
+/// Adjusts the ReplPolicyIdleNodesCount counter by the difference
+/// between the current size() and oldSize.
+void
+mem_hdr::updateIdleNodes(const size_t oldSize)
+{
+    if (!removableByReplPolicy)
+        return;
+    const auto newSize = size();
+    const auto delta = newSize > oldSize ? newSize - oldSize : oldSize - newSize;
+    if (newSize == oldSize)
+        return;
+    const auto oldCount = ReplPolicyIdleNodesCount;
+    if (newSize > oldSize) {
+        ReplPolicyIdleNodesCount += delta;
+    } else {
+        assert(ReplPolicyIdleNodesCount >= delta);
+        ReplPolicyIdleNodesCount -= delta;
+    }
+    debugs(19, 5, this << " updated ReplPolicyIdleNodesCount from " << oldCount << " to " << ReplPolicyIdleNodesCount);
+}
+

src/stmem.h

@@ -28,11 +28,13 @@ class mem_hdr
     int64_t freeDataUpto (int64_t);
     ssize_t copy (StoreIOBuffer const &) const;
     bool hasContigousContentRange(Range<int64_t> const &range) const;
-    /* success or fail */
+    /// Saves the buffer into the internal storage.
+    /// Do not call directly - use StoreEntry::writeData() instead.
     bool write (StoreIOBuffer const &);
     void dump() const;
     size_t size() const;
     mem_node *getBlockContainingLocation (int64_t location) const;
+    void allowedToFreeWithReplPolicy(const bool allowed);
     /* access the contained nodes - easier than punning
      * as a container ourselves
      */
@@ -41,6 +43,10 @@ class mem_hdr
 
     static Splay<mem_node *>::SPLAYCMP NodeCompare;
 
+    /// the total number of pages that allowed to be purged by
+    /// the associated replacement policy
+    static size_t ReplPolicyIdleNodesCount;
+
 private:
     void debugDump() const;
     bool unlink(mem_node *aNode);
@@ -49,8 +55,10 @@ class mem_hdr
     bool unionNotEmpty (StoreIOBuffer const &);
     mem_node *nodeToRecieve(int64_t offset);
     size_t writeAvailable(mem_node *aNode, int64_t location, size_t amount, char const *source);
+    void updateIdleNodes(const size_t oldSize);
     int64_t inmem_hi;
     Splay<mem_node *> nodes;
+    bool removableByReplPolicy; ///< whether nodes allowed to be purged by the associated replacement policy
 };
 
 #endif /* SQUID_STMEM_H */

src/store.cc

@@ -418,6 +418,8 @@ StoreEntry::hashDelete()
 void
 StoreEntry::lock(const char *context)
 {
+    if (!lock_count && hasReplPolicy())
+        mem_obj->data_hdr.allowedToFreeWithReplPolicy(false);
     ++lock_count;
     debugs(20, 3, context << " locked key " << getMD5Text() << ' ' << *this);
 }
@@ -450,6 +452,9 @@ StoreEntry::unlock(const char *context)
     if (lock_count)
         return (int) lock_count;
 
+    if (hasReplPolicy())
+        mem_obj->data_hdr.allowedToFreeWithReplPolicy(true);
+
     abandon(context);
     return 0;
 }
@@ -755,14 +760,14 @@ StoreEntry::write (StoreIOBuffer writeBuffer)
 {
     assert(mem_obj != nullptr);
     /* This assert will change when we teach the store to update */
-    assert(store_status == STORE_PENDING);
+    assert(isLocalWriter());
 
     // XXX: caller uses content offset, but we also store headers
     writeBuffer.offset += mem_obj->baseReply().hdr_sz;
 
     debugs(20, 5, "storeWrite: writing " << writeBuffer.length << " bytes for '" << getMD5Text() << "'");
     storeGetMemSpace(writeBuffer.length);
-    mem_obj->write(writeBuffer);
+    writeData(writeBuffer);
 
     if (EBIT_TEST(flags, ENTRY_FWD_HDR_WAIT) && !mem_obj->readAheadPolicyCanRead()) {
         debugs(20, 3, "allow Store clients to get entry content after buffering too much for " << *this);
@@ -772,6 +777,16 @@ StoreEntry::write (StoreIOBuffer writeBuffer)
     invokeHandlers();
 }
 
+void
+StoreEntry::writeData(StoreIOBuffer writeBuffer)
+{
+    assert(mem_obj);
+    debugs(20, 6, "offset " << writeBuffer.offset << " len " << writeBuffer.length);
+    auto &dataHdr = mem_obj->data_hdr;
+    assert (dataHdr.endOffset() || writeBuffer.offset == 0);
+    assert(dataHdr.write(writeBuffer));
+}
+
 /* Append incoming data from a primary server to an entry. */
 void
 StoreEntry::append(char const *buf, int len)
@@ -1513,15 +1528,20 @@ StoreEntry::setMemStatus(mem_status_t new_status)
         } else {
             mem_policy->Add(mem_policy, this, &mem_obj->repl);
             debugs(20, 4, "inserted " << *this << " key: " << getMD5Text());
+            if (!locked())
+                mem_obj->data_hdr.allowedToFreeWithReplPolicy(true);
         }
 
         ++hot_obj_count; // TODO: maintain for the shared hot cache as well
     } else {
         if (EBIT_TEST(flags, ENTRY_SPECIAL)) {
             debugs(20, 4, "not removing special " << *this << " from policy");
         } else {
-            mem_policy->Remove(mem_policy, this, &mem_obj->repl);
-            debugs(20, 4, "removed " << *this);
+            if (hasReplPolicy()) {
+                mem_obj->data_hdr.allowedToFreeWithReplPolicy(false);
+                mem_policy->Remove(mem_policy, this, &mem_obj->repl);
+                debugs(20, 4, "removed " << *this);
+            }
         }
 
         --hot_obj_count;
@@ -1557,7 +1577,7 @@ void
 StoreEntry::ensureMemObject(const char *aUrl, const char *aLogUrl, const HttpRequestMethod &aMethod)
 {
     if (!mem_obj)
-        mem_obj = new MemObject();
+        createMemObject();
     mem_obj->setUris(aUrl, aLogUrl, aMethod);
 }
 
@@ -1949,6 +1969,19 @@ StoreEntry::checkDisk() const
     }
 }
 
+/// whether the entry was added to a memory replacement policy
+bool
+StoreEntry::hasReplPolicy() const
+{
+    if (!mem_obj)
+        return false;
+    if (mem_obj->repl.data) {
+        assert(!EBIT_TEST(flags, ENTRY_SPECIAL));
+        return true;
+    }
+    return false;
+}
+
 /*
  * return true if the entry is in a state where
  * it can accept more data (ie with write() method)

src/store/Controller.cc

@@ -339,7 +339,7 @@ Store::Controller::checkFoundCandidate(const StoreEntry &entry) const
             // Transients do check when they setCollapsingRequirement().
         } else {
             // a local writer must hold a lock on its writable entry
-            if (!(entry.locked() && entry.isAccepting()))
+            if (!(entry.isLocalWriter()))
                 throw TextException("no local writer", Here());
         }
     }
@@ -534,6 +534,9 @@ Store::Controller::freeMemorySpace(const int bytesRequired)
     if (memoryCacheHasSpaceFor(pagesRequired))
         return;
 
+    if (mem_hdr::ReplPolicyIdleNodesCount < static_cast<size_t>(pagesRequired))
+        return;
+
     // XXX: When store_pages_max is smaller than pagesRequired, we should not
     // look for more space (but we do because we want to abandon idle entries?).
 

src/store_client.cc

@@ -539,7 +539,7 @@ store_client::readBody(const char *, ssize_t len)
              * copyInto.offset includes headers, which is what mem cache needs
              */
             if (copyInto.offset == entry->mem_obj->endOffset()) {
-                entry->mem_obj->write(StoreIOBuffer(len, copyInto.offset, copyInto.data));
+                entry->writeData(StoreIOBuffer(len, copyInto.offset, copyInto.data));
             }
         }
     }

src/tests/stub_MemObject.cc

@@ -29,7 +29,7 @@ MemObject::endOffset() const
 void MemObject::trimSwappable() STUB
 void MemObject::trimUnSwappable() STUB
 int64_t MemObject::policyLowestOffsetToKeep(bool) const STUB_RETVAL(-1)
-MemObject::MemObject() {
+MemObject::MemObject(bool locked) : data_hdr(locked) {
     ping_reply_callback = nullptr;
     memset(&start_ping, 0, sizeof(start_ping));
 } // NOP instead of elided due to Store
@@ -46,7 +46,6 @@ int MemObject::mostBytesWanted(int, bool) const STUB_RETVAL(-1)
 #if USE_DELAY_POOLS
 DelayId MemObject::mostBytesAllowed() const STUB_RETVAL(DelayId())
 #endif
-void MemObject::write(const StoreIOBuffer &) STUB
 int64_t MemObject::lowestMemReaderOffset() const STUB_RETVAL(0)
 void MemObject::kickReads() STUB
 int64_t MemObject::objectBytesOnDisk() const STUB_RETVAL(0)

src/tests/stub_stmem.cc

@@ -12,7 +12,7 @@
 #define STUB_API "stmem.cc"
 #include "tests/STUB.h"
 
-mem_hdr::mem_hdr() STUB
+mem_hdr::mem_hdr(bool) STUB
 mem_hdr::~mem_hdr() STUB
 size_t mem_hdr::size() const STUB_RETVAL(0)
 int64_t mem_hdr::endOffset () const STUB_RETVAL(0)

src/tests/testStore.cc

@@ -112,6 +112,7 @@ TestStore::testStats()
     Store::Init(aStore);
     CPPUNIT_ASSERT_EQUAL(false, aStore->statsCalled);
     StoreEntry entry;
+    entry.lock("TestStore::testStats");
     Store::Stats(&entry);
     CPPUNIT_ASSERT_EQUAL(true, aStore->statsCalled);
     Store::FreeMemory();

src/tests/testStoreController.cc

@@ -30,6 +30,7 @@ TestStoreController::testStats()
 {
     Store::Init();
     StoreEntry *logEntry = new StoreEntry;
+    logEntry->lock("TestStoreController::testStats");
     logEntry->createMemObject("dummy_storeId", nullptr, HttpRequestMethod());
     logEntry->store_status = STORE_PENDING;
     TestSwapDirPointer aStore (new TestSwapDir);

src/tests/testStoreHashIndex.cc

@@ -29,6 +29,7 @@ void
 TestStoreHashIndex::testStats()
 {
     StoreEntry *logEntry = new StoreEntry;
+    logEntry->lock("TestStoreHashIndex::testStats");
     logEntry->createMemObject("dummy_storeId", nullptr, HttpRequestMethod());
     logEntry->store_status = STORE_PENDING;
     Store::Init();