diff --git a/src/iocore/cache/CMakeLists.txt b/src/iocore/cache/CMakeLists.txt index f8a252b430c..5a04183c0f1 100644 --- a/src/iocore/cache/CMakeLists.txt +++ b/src/iocore/cache/CMakeLists.txt @@ -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(RamCacheCompressEntries unit_tests/test_RamCacheCompressEntries.cc) # Unit Tests without unit_tests/main.cc add_executable(test_ConfigVolumes unit_tests/test_ConfigVolumes.cc) diff --git a/src/iocore/cache/RamCacheCLFUS.cc b/src/iocore/cache/RamCacheCLFUS.cc index feb09781eb8..dcabd4c85c3 100644 --- a/src/iocore/cache/RamCacheCLFUS.cc +++ b/src/iocore/cache/RamCacheCLFUS.cc @@ -24,6 +24,7 @@ // Clocked Least Frequently Used by Size (CLFUS) replacement policy // See https://cwiki.apache.org/confluence/display/TS/RamCache +#include "RamCacheCLFUS.h" #include "P_RamCache.h" #include "P_CacheInternal.h" #include "StripeSM.h" @@ -63,27 +64,6 @@ DbgCtl dbg_ctl_ram_cache_compare{"ram_cache_compare"}; #endif -struct RamCacheCLFUSEntry { - CryptoHash key; - uint64_t auxkey; - uint64_t hits; - uint32_t size; // memory used including padding in buffer - uint32_t len; // actual data length - uint32_t compressed_len; - union { - struct { - uint32_t compressed : 3; // compression type - uint32_t incompressible : 1; - uint32_t lru : 1; - uint32_t copy : 1; // copy-in-copy-out - } flag_bits; - uint32_t flags; - }; - LINK(RamCacheCLFUSEntry, lru_link); - LINK(RamCacheCLFUSEntry, hash_link); - Ptr data; -}; - constexpr uint64_t requeue_hits(const uint64_t hits) { @@ -102,46 +82,6 @@ cache_value(const RamCacheCLFUSEntry *const e) return cache_value_hits_size(e->hits, e->size); } -class RamCacheCLFUS : public RamCache -{ -public: - RamCacheCLFUS() {} - - // returns 1 on found/stored, 0 on not found/stored, if provided auxkey1 and auxkey2 must match - int get(CryptoHash *key, Ptr *ret_data, uint64_t auxkey = 0) override; - int put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy = false, uint64_t auxkey = 0) override; - int fixup(const CryptoHash *key, uint64_t old_auxkey, uint64_t new_auxkey) override; - int64_t size() const override; - - void init(int64_t max_bytes, StripeSM *stripe) override; - - void compress_entries(EThread *thread, int do_at_most = INT_MAX); - - // TODO move it to private. - StripeSM *stripe = nullptr; // for stats -private: - int64_t _max_bytes = 0; - int64_t _bytes = 0; - int64_t _objects = 0; - - double _average_value = 0; - int64_t _history = 0; - int _ibuckets = 0; - int _nbuckets = 0; - DList(RamCacheCLFUSEntry, hash_link) *_bucket = nullptr; - Que(RamCacheCLFUSEntry, lru_link) _lru[2]; - uint16_t *_seen = nullptr; - int _ncompressed = 0; - RamCacheCLFUSEntry *_compressed = nullptr; // first uncompressed lru[0] entry - - void _resize_hashtable(); - void _victimize(RamCacheCLFUSEntry *e); - void _move_compressed(RamCacheCLFUSEntry *e); - RamCacheCLFUSEntry *_destroy(RamCacheCLFUSEntry *e); - void _requeue_victims(Que(RamCacheCLFUSEntry, lru_link) & victims); - void _tick(); // move CLOCK on history -}; - int64_t RamCacheCLFUS::size() const { @@ -475,6 +415,12 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) default: goto Lcontinue; case CACHE_COMPRESSION_FASTLZ: + if (e->len < 16) { + // fastlz cannot compress inputs this small; decide while the entry + // is still lock-protected rather than in the unlocked region below. + e->flag_bits.incompressible = 1; + goto Lcontinue; + } l = static_cast(static_cast(e->len) * 1.05 + 66); break; case CACHE_COMPRESSION_LIBZ: @@ -495,11 +441,11 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) bool failed = false; switch (ctype) { default: - goto Lfailed; + // The bound switch above filtered unknown types; this is unreachable, + // but must not jump to Lfailed from this unlocked region. + failed = true; + break; case CACHE_COMPRESSION_FASTLZ: - if (e->len < 16) { - goto Lfailed; - } if ((l = fastlz_compress(edata->data(), elen, b)) <= 0) { failed = true; } @@ -526,11 +472,10 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) #endif } MUTEX_TAKE_LOCK(stripe->mutex, thread); - // see if the entry is till around + // See if the entry is still around; it may have been freed while the + // lock was dropped, so this must be checked before anything writes to + // it (including the failure marking below). { - if (failed) { - goto Lfailed; - } uint32_t i = key.slice32(3) % this->_nbuckets; RamCacheCLFUSEntry *ee = this->_bucket[i].head; while (ee) { @@ -540,11 +485,18 @@ RamCacheCLFUS::compress_entries(EThread *thread, int do_at_most) ee = ee->hash_link.next; } if (!ee || ee != e) { - e = this->_compressed; ats_free(b); + e = this->_compressed; + if (!e) { + // The cursor was invalidated while the lock was dropped. + break; + } goto Lcontinue; } } + if (failed) { + goto Lfailed; + } if (l > required_compression * e->len) { e->flag_bits.incompressible = true; } diff --git a/src/iocore/cache/RamCacheCLFUS.h b/src/iocore/cache/RamCacheCLFUS.h new file mode 100644 index 00000000000..037e0cfb159 --- /dev/null +++ b/src/iocore/cache/RamCacheCLFUS.h @@ -0,0 +1,99 @@ +/** @file + + Clocked Least Frequently Used by Size (CLFUS) RAM cache replacement policy. + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + */ + +#pragma once + +// See https://cwiki.apache.org/confluence/display/TS/RamCache + +#include "P_RamCache.h" + +#include "iocore/eventsystem/IOBuffer.h" +#include "tscore/CryptoHash.h" +#include "tscore/List.h" + +#include +#include + +class EThread; +class StripeSM; + +struct RamCacheCLFUSEntry { + CryptoHash key; + uint64_t auxkey; + uint64_t hits; + uint32_t size; // memory used including padding in buffer + uint32_t len; // actual data length + uint32_t compressed_len; + union { + struct { + uint32_t compressed : 3; // compression type + uint32_t incompressible : 1; + uint32_t lru : 1; + uint32_t copy : 1; // copy-in-copy-out + } flag_bits; + uint32_t flags; + }; + LINK(RamCacheCLFUSEntry, lru_link); + LINK(RamCacheCLFUSEntry, hash_link); + Ptr data; +}; + +class RamCacheCLFUS : public RamCache +{ +public: + RamCacheCLFUS() {} + + // returns 1 on found/stored, 0 on not found/stored, if provided auxkey1 and auxkey2 must match + int get(CryptoHash *key, Ptr *ret_data, uint64_t auxkey = 0) override; + int put(CryptoHash *key, IOBufferData *data, uint32_t len, bool copy = false, uint64_t auxkey = 0) override; + int fixup(const CryptoHash *key, uint64_t old_auxkey, uint64_t new_auxkey) override; + int64_t size() const override; + + void init(int64_t max_bytes, StripeSM *stripe) override; + + void compress_entries(EThread *thread, int do_at_most = INT_MAX); + + // TODO move it to private. + StripeSM *stripe = nullptr; // for stats +private: + int64_t _max_bytes = 0; + int64_t _bytes = 0; + int64_t _objects = 0; + + double _average_value = 0; + int64_t _history = 0; + int _ibuckets = 0; + int _nbuckets = 0; + DList(RamCacheCLFUSEntry, hash_link) *_bucket = nullptr; + Que(RamCacheCLFUSEntry, lru_link) _lru[2]; + uint16_t *_seen = nullptr; + int _ncompressed = 0; + RamCacheCLFUSEntry *_compressed = nullptr; // first uncompressed lru[0] entry + + void _resize_hashtable(); + void _victimize(RamCacheCLFUSEntry *e); + void _move_compressed(RamCacheCLFUSEntry *e); + RamCacheCLFUSEntry *_destroy(RamCacheCLFUSEntry *e); + void _requeue_victims(Que(RamCacheCLFUSEntry, lru_link) & victims); + void _tick(); // move CLOCK on history +}; diff --git a/src/iocore/cache/unit_tests/test_RamCacheCompressEntries.cc b/src/iocore/cache/unit_tests/test_RamCacheCompressEntries.cc new file mode 100644 index 00000000000..9644f5acf0b --- /dev/null +++ b/src/iocore/cache/unit_tests/test_RamCacheCompressEntries.cc @@ -0,0 +1,182 @@ +/** @file + + Catch-based regression tests for RamCacheCLFUS::compress_entries(). + + @section license License + + Licensed to the Apache Software Foundation (ASF) under one + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. + */ + +#include "main.h" + +#include "../RamCacheCLFUS.h" +#include "../P_CacheInternal.h" + +#include +#include +#include +#include + +// Required by main.h +int cache_vols = 1; +bool reuse_existing_cache = false; + +namespace +{ + +// Minimal CacheDisk wiring needed to construct a StripeSM. Mirrors the helper +// in test_Stripe.cc. +void +init_disk(CacheDisk &disk) +{ + disk.path = static_cast(ats_malloc(1)); + disk.path[0] = '\0'; + disk.disk_stripes = static_cast(ats_malloc(sizeof(DiskStripe *))); + disk.disk_stripes[0] = nullptr; + disk.header = static_cast(ats_malloc(sizeof(DiskHeader))); + disk.header->num_volumes = 0; +} + +// The CLFUS get/put/compress paths touch only these metrics and the stripe +// mutex, so that is all the stripe needs for these tests. +void +wire_stripe(StripeSM &stripe, CacheVol &cache_vol) +{ + stripe.cache_vol = &cache_vol; + + cache_rsb.ram_cache_bytes = ts::Metrics::Gauge::createPtr("unit_test.clfus.ram_cache.bytes"); + cache_rsb.ram_cache_hits = ts::Metrics::Counter::createPtr("unit_test.clfus.ram_cache.hits"); + cache_rsb.ram_cache_misses = ts::Metrics::Counter::createPtr("unit_test.clfus.ram_cache.misses"); + cache_vol.vol_rsb.ram_cache_bytes = ts::Metrics::Gauge::createPtr("unit_test.clfus.vol.ram_cache.bytes"); + cache_vol.vol_rsb.ram_cache_hits = ts::Metrics::Counter::createPtr("unit_test.clfus.vol.ram_cache.hits"); + cache_vol.vol_rsb.ram_cache_misses = ts::Metrics::Counter::createPtr("unit_test.clfus.vol.ram_cache.misses"); +} + +RamCacheCLFUS * +make_cache(StripeSM &stripe) +{ + // Initialize with compression disabled so init() does not schedule the + // background compressor continuation, which would retain a pointer to the + // cache; compression is driven synchronously by the tests instead. The + // caches are kept reachable for the life of the process because the policy + // has no destructor (entries are pool-allocated). + cache_config_ram_cache_compress = CACHE_COMPRESSION_NONE; + cache_config_ram_cache_compress_percent = 100; + cache_config_ram_cache_use_seen_filter = 0; + + static std::vector &all_caches = *new std::vector; + auto *rc = new RamCacheCLFUS; + + all_caches.push_back(rc); + rc->init(1 << 20, &stripe); + return rc; +} + +std::vector +pattern_bytes(std::size_t len) +{ + std::vector bytes(len); + for (std::size_t i = 0; i < len; i++) { + bytes[i] = static_cast('A' + (i % 26)); + } + return bytes; +} + +Ptr +make_buffer(const std::vector &bytes) +{ + int64_t idx = iobuffer_size_to_index(bytes.size(), MAX_BUFFER_SIZE_INDEX); + Ptr data{make_ptr(new_IOBufferData(idx, MEMALIGNED))}; + + std::memcpy(data->data(), bytes.data(), bytes.size()); + return data; +} + +CryptoHash +fresh_key() +{ + static uint64_t salt = 0; + + ++salt; + CryptoHash key; + key.u64[0] = 0xc0ffee00 + salt; + key.u64[1] = 0xdeadbeef + salt; + return key; +} + +} // namespace + +// fastlz cannot compress inputs under 16 bytes. Such entries must be decided +// under the stripe lock and left intact. The races fixed alongside this test +// are only observable under concurrency, so this pins the single-threaded +// behavior of the restructured code: undersized entries are marked +// incompressible while locked, skipped on later passes, and stay readable. +TEST_CASE("CLFUS compress_entries leaves undersized payloads intact", "[cache][ramcache][compress]") +{ + CacheDisk disk; + init_disk(disk); + StripeSM stripe{&disk, 10, 0}; + CacheVol cache_vol; + wire_stripe(stripe, cache_vol); + + auto *rc = make_cache(stripe); + auto payload = pattern_bytes(8); + auto buf = make_buffer(payload); + auto key = fresh_key(); + + REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1); + + cache_config_ram_cache_compress = CACHE_COMPRESSION_FASTLZ; + // Two passes: the first marks the entry incompressible; the second must + // skip it (on the former code, the first pass's loop tail ran unlocked). + rc->compress_entries(this_ethread(), INT_MAX); + rc->compress_entries(this_ethread(), INT_MAX); + + Ptr got; + + REQUIRE(rc->get(&key, &got) == RAM_HIT_COMPRESS_NONE); + REQUIRE(got.get() != nullptr); + CHECK(std::memcmp(got->data(), payload.data(), payload.size()) == 0); +} + +TEST_CASE("CLFUS compress_entries still compresses eligible payloads", "[cache][ramcache][compress]") +{ + CacheDisk disk; + init_disk(disk); + StripeSM stripe{&disk, 10, 0}; + CacheVol cache_vol; + wire_stripe(stripe, cache_vol); + + auto *rc = make_cache(stripe); + auto payload = pattern_bytes(8192); + auto buf = make_buffer(payload); + auto key = fresh_key(); + + REQUIRE(rc->put(&key, buf.get(), payload.size(), true) == 1); + + cache_config_ram_cache_compress = CACHE_COMPRESSION_FASTLZ; + int64_t size_before = rc->size(); + + rc->compress_entries(this_ethread(), INT_MAX); + CHECK(rc->size() < size_before); + + Ptr got; + + REQUIRE(rc->get(&key, &got) == RAM_HIT_COMPRESS_FASTLZ); + REQUIRE(got.get() != nullptr); + CHECK(std::memcmp(got->data(), payload.data(), payload.size()) == 0); +}