Skip to content

Honor the RamCache copy flag in LRU and S3-FIFO #13380

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
phongn wants to merge 1 commit into
base: master
Choose a base branch
from
Draft

Honor the RamCache copy flag in LRU and S3-FIFO #13380

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions src/iocore/cache/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ if(BUILD_TESTING)
add_cache_test(Update_Header unit_tests/test_Update_header.cc)
add_cache_test(CacheStripe unit_tests/test_Stripe.cc)
add_cache_test(CacheAggregateWriteBuffer unit_tests/test_AggregateWriteBuffer.cc)
add_cache_test(RamCacheCopy unit_tests/test_RamCacheCopy.cc)

# Unit Tests without unit_tests/main.cc
add_executable(test_ConfigVolumes unit_tests/test_ConfigVolumes.cc)
Expand Down
33 changes: 33 additions & 0 deletions src/iocore/cache/P_RamCache.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@

#include "iocore/eventsystem/IOBuffer.h"
#include "tscore/CryptoHash.h"
#include "tscore/ink_memory.h"

#include <cstring>

class StripeSM;

Expand All @@ -41,6 +44,36 @@ 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)
{
char *b = static_cast<char *>(ats_malloc(len));

memcpy(b, data->data(), len);
Ptr<IOBufferData> d = make_ptr(new_xmalloc_IOBufferData(b, len));

d->_mem_type = DEFAULT_ALLOC;
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();
Expand Down
19 changes: 6 additions & 13 deletions src/iocore/cache/RamCacheCLFUS.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -314,12 +314,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);
Expand Down Expand Up @@ -624,11 +623,8 @@ 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;
}
check_accounting(this);
e->flag_bits.copy = copy;
Expand Down Expand Up @@ -735,10 +731,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;
Expand Down
35 changes: 28 additions & 7 deletions src/iocore/cache/RamCacheLRU.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
struct RamCacheLRUEntry {
CryptoHash key;
uint64_t auxkey;
bool copy; // copy-in-copy-out: buffers are never shared with callers
LINK(RamCacheLRUEntry, lru_link);
LINK(RamCacheLRUEntry, hash_link);
Ptr<IOBufferData> data;
Expand Down Expand Up @@ -147,7 +148,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->data->block_size());
} 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);
Expand Down Expand Up @@ -181,9 +186,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;
Expand Down Expand Up @@ -213,6 +217,18 @@ RamCacheLRU::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32_t
if (e->auxkey == auxkey) {
lru.remove(e);
lru.enqueue(e);
if (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.
int64_t delta = static_cast<int64_t>(len) - e->data->block_size();

e->data = copy_data_in(data, 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);
Expand All @@ -224,13 +240,18 @@ 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->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) {
Expand Down
37 changes: 33 additions & 4 deletions src/iocore/cache/RamCacheS3FIFO.cc
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand All @@ -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
Expand All @@ -359,6 +366,23 @@ RamCacheS3FIFO::put(CryptoHash *key, IOBufferData *data, [[maybe_unused]] uint32
if (e->freq < FREQ_MAX) {
e->freq++;
}
if (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.
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
Expand Down Expand Up @@ -396,7 +420,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) {
Expand Down
Loading