Honor the RamCache copy flag in LRU and S3-FIFO

src/iocore/cache/CMakeLists.txt

@@ -103,6 +103,7 @@ if(BUILD_TESTING)
   add_cache_test(CacheStripe unit_tests/test_Stripe.cc)
   add_cache_test(CacheAggregateWriteBuffer unit_tests/test_AggregateWriteBuffer.cc)
   add_cache_test(RamCacheCLFUS unit_tests/test_RamCacheCLFUS.cc)
+  add_cache_test(RamCacheCopy unit_tests/test_RamCacheCopy.cc)
   add_cache_test(RamCacheCompressEntries unit_tests/test_RamCacheCompressEntries.cc)
   add_cache_test(RamCacheSeenFilter unit_tests/test_RamCacheSeenFilter.cc)
   # Only the shutdown test attaches a live segment; the rest need no shm syscall.

src/iocore/cache/P_RamCache.h

@@ -28,6 +28,8 @@
 #include "iocore/eventsystem/IOBuffer.h"
 #include "tscore/CryptoHash.h"
 
+#include <cstring>
+
 class StripeSM;
 
 class RamCache
@@ -41,6 +43,33 @@ class RamCache
 
   virtual void init(int64_t max_bytes, StripeSM *stripe) = 0;
   virtual ~RamCache(){};
+
+protected:
+  // put(copy = true) is the caller's promise that it may mutate its buffer
+  // after the put, so buffers are never shared with callers in either
+  // direction for such entries: implementations store a private copy on put
+  // (copy_data_in) and hand out a private copy on get (copy_data_out).
+
+  // Exact-size private copy of the caller's buffer; block_size() on the
+  // result recovers len.
+  static Ptr<IOBufferData>
+  copy_data_in(IOBufferData *data, uint32_t len)
+  {
+    Ptr<IOBufferData> d = make_ptr(new_IOBufferData(BUFFER_SIZE_INDEX_FOR_XMALLOC_SIZE(len), DEFAULT_ALLOC));
+
+    memcpy(d->data(), data->data(), len);
+    return d;
+  }
+
+  // Private copy of a stored buffer to hand to the caller.
+  static Ptr<IOBufferData>
+  copy_data_out(IOBufferData *data, uint32_t len)
+  {
+    Ptr<IOBufferData> d = make_ptr(new_IOBufferData(iobuffer_size_to_index(len, MAX_BUFFER_SIZE_INDEX), MEMALIGNED));
+
+    memcpy(d->data(), data->data(), len);
+    return d;
+  }
 };
 
 RamCache *new_RamCacheLRU();

src/iocore/cache/RamCacheCLFUS.cc

@@ -426,12 +426,11 @@ RamCacheCLFUS::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxkey
           }
           (*ret_data) = data;
         } else {
-          IOBufferData *data = e->data.get();
           if (e->flag_bits.copy) {
-            data = new_IOBufferData(iobuffer_size_to_index(e->len, MAX_BUFFER_SIZE_INDEX), MEMALIGNED);
-            ::memcpy(data->data(), e->data->data(), e->len);
+            (*ret_data) = copy_data_out(e->data.get(), e->len);
+          } else {
+            (*ret_data) = e->data;
           }
-          (*ret_data) = data;
         }
         ts::Metrics::Counter::increment(cache_rsb.ram_cache_hits);
         ts::Metrics::Counter::increment(stripe->cache_vol->vol_rsb.ram_cache_hits);
@@ -811,6 +810,17 @@ RamCacheCLFUS::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy,
       this->_move_compressed(e);
       this->_lru[e->flag_bits.lru].remove(e);
       this->_lru[e->flag_bits.lru].enqueue(e);
+      if (copy && e->flag_bits.copy && !e->flag_bits.compressed) {
+        // Already an uncompressed private copy of this object: get() never
+        // exposes that buffer, so re-copying it (two requests that both missed
+        // and both read the object from disk) changes nothing. Everything
+        // below is a no-op for such an entry apart from the allocation and
+        // memcpy -- size is already len, so delta is 0 -- except when the
+        // entry is compressed, where the swap is what decompresses it and
+        // skipping it would leave compressed bytes behind a cleared flag.
+        DDbg(dbg_ctl_ram_cache, "put %X %" PRId64 " size %d HIT (private, unchanged)", key->slice32(3), auxkey, e->size);
+        return 1;
+      }
       int64_t delta  = (static_cast<int64_t>(size)) - static_cast<int64_t>(e->size);
       this->_bytes  += delta;
       ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, delta);
@@ -819,12 +829,10 @@ RamCacheCLFUS::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy,
         e->size = size;
         e->data = data;
       } else {
-        char *b = static_cast<char *>(ats_malloc(len));
-        memcpy(b, data->data(), len);
-        e->data            = new_xmalloc_IOBufferData(b, len);
-        e->data->_mem_type = DEFAULT_ALLOC;
-        e->size            = size;
+        e->data = copy_data_in(data, len);
+        e->size = size;
       }
+      e->len = len; // get() and the compressor read e->len bytes out of e->data
       check_accounting(this);
       e->flag_bits.copy       = copy;
       e->flag_bits.compressed = 0;
@@ -930,10 +938,7 @@ RamCacheCLFUS::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy,
   if (!copy) {
     e->data = data;
   } else {
-    char *b = static_cast<char *>(ats_malloc(len));
-    memcpy(b, data->data(), len);
-    e->data            = new_xmalloc_IOBufferData(b, len);
-    e->data->_mem_type = DEFAULT_ALLOC;
+    e->data = copy_data_in(data, len);
   }
   e->flag_bits.copy  = copy;
   this->_bytes      += size + entry_overhead;

src/iocore/cache/RamCacheLRU.cc

@@ -33,6 +33,8 @@
 struct RamCacheLRUEntry {
   CryptoHash key;
   uint64_t   auxkey;
+  uint32_t   len;  // data length, so get() does not depend on copy_data_in allocating exactly len
+  bool       copy; // copy-in-copy-out: buffers are never shared with callers
   LINK(RamCacheLRUEntry, lru_link);
   LINK(RamCacheLRUEntry, hash_link);
   Ptr<IOBufferData> data;
@@ -147,7 +149,11 @@ RamCacheLRU::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxkey)
     if (e->key == *key && e->auxkey == auxkey) {
       lru.remove(e);
       lru.enqueue(e);
-      (*ret_data) = e->data;
+      if (e->copy) {
+        (*ret_data) = copy_data_out(e->data.get(), e->len);
+      } else {
+        (*ret_data) = e->data;
+      }
       DDbg(dbg_ctl_ram_cache, "get %X %" PRIu64 " HIT", key->slice32(3), auxkey);
       ts::Metrics::Counter::increment(cache_rsb.ram_cache_hits);
       ts::Metrics::Counter::increment(stripe->cache_vol->vol_rsb.ram_cache_hits);
@@ -181,9 +187,8 @@ RamCacheLRU::remove(RamCacheLRUEntry *e)
   return ret;
 }
 
-// ignore 'copy' since we don't touch the data
 int
-RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t len, bool, uint64_t auxkey)
+RamCacheLRU::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy, uint64_t auxkey)
 {
   if (!max_bytes) {
     return 0;
@@ -196,6 +201,30 @@ RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t
       if (e->auxkey == auxkey) {
         lru.remove(e);
         lru.enqueue(e);
+        if (copy && !e->copy) {
+          // The entry may still be sharing a caller's buffer from a put made
+          // while copy semantics were not requested; refresh it with a
+          // private copy before the caller mutates its buffer. Once it holds
+          // a private copy there is nothing to refresh: get() never exposes
+          // that buffer, so a further copy=true put (two requests that both
+          // missed and both read the object from disk) would only allocate
+          // and copy the object again to the same effect.
+          //
+          // A refresh only ever gives bytes back, so unlike an insert it needs
+          // no eviction pass: the private copy is an exact-size allocation
+          // while a shared buffer was charged its rounded block_size() (always
+          // >= len), and a re-put under the same key and auxkey names the same
+          // on-disk doc, so len itself does not grow.
+          Ptr<IOBufferData> d     = copy_data_in(data, len);
+          int64_t           delta = d->block_size() - e->data->block_size();
+
+          e->data  = d;
+          e->len   = len;
+          e->copy  = true;
+          bytes   += delta;
+          ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, delta);
+          ts::Metrics::Gauge::increment(stripe->cache_vol->vol_rsb.ram_cache_bytes, delta);
+        }
         return 1;
       } else { // discard when aux keys conflict
         e        = remove(e);
@@ -233,13 +262,19 @@ RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t
   e         = THREAD_ALLOC(ramCacheLRUEntryAllocator, this_ethread());
   e->key    = *key;
   e->auxkey = auxkey;
-  e->data   = data;
+  e->len    = len;
+  e->copy   = copy;
+  if (copy) {
+    e->data = copy_data_in(data, len);
+  } else {
+    e->data = data;
+  }
   bucket[i].push(e);
   lru.enqueue(e);
-  bytes += ENTRY_OVERHEAD + data->block_size();
+  bytes += ENTRY_OVERHEAD + e->data->block_size();
   objects++;
-  ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, ENTRY_OVERHEAD + data->block_size());
-  ts::Metrics::Gauge::increment(stripe->cache_vol->vol_rsb.ram_cache_bytes, ENTRY_OVERHEAD + data->block_size());
+  ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, ENTRY_OVERHEAD + e->data->block_size());
+  ts::Metrics::Gauge::increment(stripe->cache_vol->vol_rsb.ram_cache_bytes, ENTRY_OVERHEAD + e->data->block_size());
   while (bytes > max_bytes) {
     RamCacheLRUEntry *ee = lru.dequeue();
     if (ee) {

src/iocore/cache/RamCacheS3FIFO.cc

@@ -61,6 +61,7 @@ struct RamCacheS3FIFOEntry {
   uint32_t   size; // object bytes; resident entries account size + ENTRY_OVERHEAD against the budget
   uint8_t    seg;  // SEG_SMALL / SEG_MAIN / SEG_GHOST
   uint8_t    freq; // 0..FREQ_MAX
+  bool       copy; // copy-in-copy-out: buffers are never shared with callers
   LINK(RamCacheS3FIFOEntry, lru_link);
   LINK(RamCacheS3FIFOEntry, hash_link);
   Ptr<IOBufferData> data; // null for ghost entries
@@ -328,7 +329,13 @@ RamCacheS3FIFO::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxke
     if (e->freq < FREQ_MAX) {
       e->freq++;
     }
-    (*ret_data) = e->data;
+    if (e->copy) {
+      // For copy entries the stored buffer is an exact-size allocation, so
+      // e->size is the data length.
+      (*ret_data) = copy_data_out(e->data.get(), e->size);
+    } else {
+      (*ret_data) = e->data;
+    }
     ts::Metrics::Counter::increment(cache_rsb.ram_cache_hits);
     ts::Metrics::Counter::increment(_stripe->cache_vol->vol_rsb.ram_cache_hits);
     return 1;
@@ -339,12 +346,12 @@ RamCacheS3FIFO::get(CryptoHash *key, Ptr<IOBufferData> *ret_data, uint64_t auxke
 }
 
 int
-RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t len, bool, uint64_t auxkey)
+RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy, uint64_t auxkey)
 {
   if (!_max_bytes) {
     return 0;
   }
-  uint32_t size = data->block_size();
+  uint32_t size = copy ? len : data->block_size();
   uint32_t i    = key->slice32(3) % _nbuckets;
 
   // Walk the hash chain. A resident hit just counts the reference; a ghost hit is admitted fresh to
@@ -359,6 +366,36 @@ RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32
           if (e->freq < FREQ_MAX) {
             e->freq++;
           }
+          if (copy && !e->copy) {
+            // The entry may still be sharing a caller's buffer from a put
+            // made while copy semantics were not requested; refresh it with a
+            // private copy before the caller mutates its buffer. Once it holds
+            // a private copy there is nothing to refresh: get() never exposes
+            // that buffer, so a further copy=true put (two requests that both
+            // missed and both read the object from disk) would only allocate
+            // and copy the object again to the same effect.
+            //
+            // A refresh only ever gives bytes back, so unlike an insert it
+            // needs no eviction pass: the private copy is charged its exact
+            // length while a shared buffer was charged the rounded
+            // block_size() (always >= len), and a re-put under the same key
+            // and auxkey names the same on-disk doc, so len itself does not
+            // grow. That matters here because the insert path keeps the budget
+            // by declining an oversized object outright, which a refresh of an
+            // already-resident entry could not do.
+            int64_t delta = static_cast<int64_t>(len) - e->size;
+
+            e->data = copy_data_in(data, len);
+            e->copy = true;
+            e->size = len;
+            if (e->seg == SEG_SMALL) {
+              _s_bytes += delta;
+            } else {
+              _m_bytes += delta;
+            }
+            ts::Metrics::Gauge::increment(cache_rsb.ram_cache_bytes, delta);
+            ts::Metrics::Gauge::increment(_stripe->cache_vol->vol_rsb.ram_cache_bytes, delta);
+          }
           return 1;
         }
         RamCacheS3FIFOEntry *next = e->hash_link.next; // ghost hit: admit a fresh copy to main
@@ -396,7 +433,12 @@ RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32
   ne->size                = size;
   ne->freq                = 0;
   ne->seg                 = ghost_hit ? SEG_MAIN : SEG_SMALL;
-  ne->data                = data;
+  ne->copy                = copy;
+  if (copy) {
+    ne->data = copy_data_in(data, len);
+  } else {
+    ne->data = data;
+  }
   _bucket[i].push(ne);
   _seg[ne->seg].enqueue(ne);
   if (ghost_hit) {