diff --git a/include/boost/capy/ex/recycling_memory_resource.hpp b/include/boost/capy/ex/recycling_memory_resource.hpp index 99e02418b..5f8249ccd 100644 --- a/include/boost/capy/ex/recycling_memory_resource.hpp +++ b/include/boost/capy/ex/recycling_memory_resource.hpp @@ -17,6 +17,7 @@ #include #include #include +#include namespace boost { namespace capy { @@ -31,6 +32,17 @@ namespace capy { Size classes: 64, 128, 256, 512, 1024, 2048 bytes. Allocations larger than 2048 bytes bypass the pools entirely. + Each pool caches at most 256 KiB of blocks per size class: + 4096 blocks of the smallest class, 128 of the largest. A freed + block that fits in neither the thread's pool nor the global pool + goes back to `::operator delete`. + + Blocks a thread caches but stops using drain back to the global + pool. The drain runs as the thread goes on freeing blocks or + refilling from the global pool. Other threads can then reuse what a + burst leaves behind. Blocks the global pool has no room for go back + to `::operator delete`. + This is the default allocator used by run_async when no allocator is specified. @@ -52,7 +64,13 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour static constexpr std::size_t num_classes = 6; static constexpr std::size_t min_class_size = 64; // 2^6 static constexpr std::size_t max_class_size = 2048; // 2^11 - static constexpr std::size_t bucket_capacity = 16; + static constexpr int min_class_shift = std::countr_zero(min_class_size); + static constexpr std::size_t max_cached_bytes = 256 * 1024; + + // Frees on one thread, plus blocks it refills from the global pool, + // between decay passes. Allocations served from the thread's pool + // aren't counted, which keeps the counter off that path. + static constexpr std::size_t decay_period = 4096; static std::size_t round_up_pow2(std::size_t n) noexcept @@ -63,39 +81,65 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour static std::size_t get_class_index(std::size_t rounded) noexcept { - std::size_t idx = std::countr_zero(rounded) - 6; // 64 = 2^6 + auto const idx = static_cast( + std::countr_zero(rounded) - min_class_shift); return idx < num_classes ? idx : num_classes; } + static constexpr std::size_t + bucket_capacity(std::size_t idx) noexcept + { + return max_cached_bytes >> (idx + min_class_shift); + } + + // Cached blocks are linked through their own storage, so a bucket + // holds any number of them without a side array. + struct node + { + node* next; + }; + struct bucket { + node* head = nullptr; std::size_t count = 0; - void* ptrs[bucket_capacity] = {}; + + // Fewest blocks cached since the last decay pass: that many sat + // unused through the whole period. + std::size_t low = 0; void* pop() noexcept { - if(count == 0) + node* n = head; + if(n == nullptr) return nullptr; - return ptrs[--count]; + head = n->next; + --count; + if(count < low) + low = count; + return n; } // Peter Dimov's idea + // `b` must be empty: its chain is overwritten. void* pop(bucket& b) noexcept { - if(count == 0) + if(head == nullptr) return nullptr; - for(std::size_t i = 0; i < count; ++i) - b.ptrs[i] = ptrs[i]; - b.count = count - 1; + b.head = head; + b.count = count; + head = nullptr; count = 0; - return b.ptrs[b.count]; + low = 0; + return b.pop(); } - bool push(void* p) noexcept + bool push(void* p, std::size_t capacity) noexcept { - if(count >= bucket_capacity) + if(count >= capacity) return false; - ptrs[count++] = p; + head = ::new(p) node{head}; + ++count; return true; } }; @@ -103,6 +147,7 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour struct pool { bucket buckets[num_classes]; + std::size_t arrivals = 0; // No destructor: a non-trivial dtor forces a guard variable on the // thread_local in local(), checked on every alloc/free. Constant @@ -132,6 +177,11 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour void* allocate_slow(std::size_t rounded, std::size_t idx); void deallocate_slow(void* p, std::size_t idx); + // Move half of each bucket's idle blocks to the global pool, or + // release them when it has no room. Halving, rather than emptying, + // leaves a burst that skips one period half its blocks. + static void decay(pool& lp) noexcept; + // Register a thread-exit callback that drains this thread's local // pool back to the OS. Called only off the hot path: unconditionally // from the slow paths, and from deallocate_fast while the flag below @@ -174,6 +224,11 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour Every block comes from `::operator new`, so blocks carry the implementation's default new alignment and no more. + @par Complexity + Amortized constant. Now and then a call also hands this thread's + idle cached blocks to the global pool, or releases them when it + has no room. + @param bytes The number of bytes to allocate. @return A pointer to a block of at least `bytes` bytes. A pooled @@ -189,8 +244,7 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour std::size_t idx = get_class_index(rounded); if(idx >= num_classes) return ::operator new(bytes); - auto& lp = local(); - if(auto* p = lp.buckets[idx].pop()) + if(auto* p = local().buckets[idx].pop()) return p; return allocate_slow(rounded, idx); } @@ -207,6 +261,11 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour The third parameter is the alignment the block was allocated with, and it is ignored, as it is on allocation. + @par Complexity + Amortized constant. Now and then a call also hands this thread's + idle cached blocks to the global pool, or releases them when it + has no room. + @param p The block to return. It must have come from @ref allocate_fast or @ref do_allocate on this resource. @@ -238,7 +297,9 @@ class BOOST_CAPY_DECL recycling_memory_resource : public std::pmr::memory_resour armed = true; } auto& lp = local(); - if(lp.buckets[idx].push(p)) + if(++lp.arrivals == decay_period) + decay(lp); + if(lp.buckets[idx].push(p, bucket_capacity(idx))) return; deallocate_slow(p, idx); } diff --git a/src/ex/recycling_memory_resource.cpp b/src/ex/recycling_memory_resource.cpp index 2d188adfd..0007156de 100644 --- a/src/ex/recycling_memory_resource.cpp +++ b/src/ex/recycling_memory_resource.cpp @@ -117,13 +117,32 @@ recycling_memory_resource::allocate_slow( // Once this thread's pool was drained at exit, take one block and // leave the pool empty: nothing would release a refill. bool const cache = arm_thread_cleanup(); + void* p = nullptr; + std::size_t refilled = 0; { std::lock_guard lock(global_mutex()); auto& b = global().buckets[idx]; - if(auto* p = cache ? b.pop(local().buckets[idx]) : b.pop()) - return p; + if(cache) + { + auto& lb = local().buckets[idx]; + p = b.pop(lb); + refilled = lb.count; + } + else + { + p = b.pop(); + } } - return ::operator new(rounded); + if(p == nullptr) + return ::operator new(rounded); + + // A thread that only allocates never frees into its pool, so its + // refills are what drive its decay passes. + auto& lp = local(); + lp.arrivals += refilled; + if(lp.arrivals >= decay_period) + decay(lp); + return p; } void @@ -132,12 +151,73 @@ recycling_memory_resource::deallocate_slow( { { std::lock_guard lock(global_mutex()); - if(global().buckets[idx].push(p)) + if(global().buckets[idx].push(p, bucket_capacity(idx))) return; } ::operator delete(p); } +void +recycling_memory_resource::decay(pool& lp) noexcept +{ + lp.arrivals = 0; + + // Unlink each bucket's share before taking the lock, so the walk + // through the blocks happens outside it. + node* heads[num_classes] = {}; + node* tails[num_classes] = {}; + std::size_t counts[num_classes] = {}; + for(std::size_t i = 0; i < num_classes; ++i) + { + auto& b = lp.buckets[i]; + std::size_t const share = (b.low + 1) / 2; + if(share > 0) + { + heads[i] = b.head; + node* t = b.head; + for(std::size_t k = 1; k < share; ++k) + t = t->next; + b.head = t->next; + b.count -= share; + tails[i] = t; + counts[i] = share; + } + b.low = b.count; + } + + node* spill = nullptr; + { + std::lock_guard lock(global_mutex()); + auto& gp = global(); + for(std::size_t i = 0; i < num_classes; ++i) + { + if(counts[i] == 0) + continue; + auto& g = gp.buckets[i]; + // A share that doesn't fit whole is released whole: topping + // off a bucket already at least half full would mean walking + // the chain under the lock. + if(g.count + counts[i] <= bucket_capacity(i)) + { + tails[i]->next = g.head; + g.head = heads[i]; + g.count += counts[i]; + } + else + { + tails[i]->next = spill; + spill = heads[i]; + } + } + } + while(spill != nullptr) + { + node* n = spill; + spill = n->next; + ::operator delete(n); + } +} + void* recycling_memory_resource::do_allocate(std::size_t bytes, std::size_t alignment) { diff --git a/test/unit/ex/recycling_memory_resource.cpp b/test/unit/ex/recycling_memory_resource.cpp index d1bbdb2ba..a6f1ff63e 100644 --- a/test/unit/ex/recycling_memory_resource.cpp +++ b/test/unit/ex/recycling_memory_resource.cpp @@ -10,7 +10,10 @@ // Test that header file is self-contained. #include +#include +#include #include +#include #include #include @@ -37,37 +40,303 @@ class recycling_memory_resource_test testSlowPaths() { // Drive enough same-class allocations/deallocations to exercise - // the bucket overflow (deallocate past a full global pool frees - // directly) and refill (allocate drains the global pool into the - // thread-local one). Counts exceed 2*bucket_capacity (16) so the - // slow paths run regardless of any prior pool state; only block + // releasing what the pools can't keep and refilling (allocate + // drains the global pool into the thread-local one), in every + // size class. Counts exceed twice the class capacity, so neither + // pool can keep every block whatever its prior state; only block // identity, which we do not assert, depends on that state. recycling_memory_resource mr; - constexpr std::size_t bytes = 64; // size class 0 constexpr std::size_t align = 8; - constexpr int n = 50; - std::vector ptrs; - ptrs.reserve(n); + for(std::size_t bytes = 64; bytes <= 2048; bytes *= 2) + { + std::size_t const n = 2 * (256 * 1024 / bytes) + 44; + + std::vector ptrs; + ptrs.reserve(n); + + auto fill = [&] { + for(std::size_t i = 0; i < n; ++i) + { + void* p = mr.allocate_fast(bytes, align); + BOOST_TEST(p != nullptr); + ptrs.push_back(p); + } + }; + auto drain = [&] { + for(void* p : ptrs) + mr.deallocate_fast(p, bytes, align); + ptrs.clear(); + }; + + fill(); + drain(); // more than both pools hold: the excess is released + fill(); // drains the global pool back into the local one + drain(); + } + } - auto fill = [&] { + // A fan-out keeps hundreds of same-sized frames alive at once. Once + // they are freed, a thread must hand every one of them back instead + // of falling through to the upstream allocator. + void + testRecyclesWideFanOut() + { + constexpr std::size_t bytes = 512; // size class 3 + constexpr std::size_t align = 8; + constexpr int n = 256; + + std::thread([] { + recycling_memory_resource mr; + std::vector first; + first.reserve(n); for(int i = 0; i < n; ++i) - { - void* p = mr.allocate_fast(bytes, align); - BOOST_TEST(p != nullptr); - ptrs.push_back(p); - } - }; - auto drain = [&] { + first.push_back(mr.allocate_fast(bytes, align)); + for(void* p : first) + mr.deallocate_fast(p, bytes, align); + + std::vector second; + second.reserve(n); + for(int i = 0; i < n; ++i) + second.push_back(mr.allocate_fast(bytes, align)); + + // Reversed order makes it unlikely that the upstream allocator + // handed the same addresses back after they were freed to it. + BOOST_TEST(std::equal( + second.begin(), second.end(), first.rbegin())); + + for(void* p : second) + mr.deallocate_fast(p, bytes, align); + }).join(); + } + + // Runs `owner` on a thread that keeps its pool alive while a second, + // fresh thread takes every block the global pool can hand it, and + // returns how many of the owner's freed blocks that thread received. + template + static std::size_t + blocks_reaching_other_thread(Owner owner) + { + constexpr std::size_t bytes = 256; // size class 2 + constexpr std::size_t align = 8; + constexpr int takes = 2048; // twice the class capacity + + // Earlier tests may have filled the global pool, which would turn + // the owner's decayed blocks away. A thread's first allocation + // takes the whole global bucket, and its exit frees it. + std::thread([] { + recycling_memory_resource mr; + mr.deallocate_fast(mr.allocate_fast(bytes, align), bytes, align); + }).join(); + + std::vector owned; + std::size_t found = 0; + std::latch ready(1); + std::latch done(1); + + std::thread holder([&] { + recycling_memory_resource mr; + owned = owner(mr, bytes, align); + ready.count_down(); + done.wait(); + }); + ready.wait(); + std::thread([&] { + recycling_memory_resource mr; + std::vector taken; + for(int i = 0; i < takes; ++i) + taken.push_back(mr.allocate_fast(bytes, align)); + for(void* p : taken) + found += static_cast( + std::count(owned.begin(), owned.end(), p)); + for(void* p : taken) + mr.deallocate_fast(p, bytes, align); + }).join(); + done.count_down(); + holder.join(); + return found; + } + + // Allocates and frees a small block many times, enough for many + // decay passes on the calling thread. + static void + churn(recycling_memory_resource& mr) + { + for(int i = 0; i < 1 << 18; ++i) + mr.deallocate_fast(mr.allocate_fast(64, 8), 64, 8); + } + + // A burst cached on one thread and then left idle must become + // available to other threads instead of staying with the thread + // that freed it. + void + testIdleBurstDecays() + { + constexpr int n = 64; + auto found = blocks_reaching_other_thread( + [](recycling_memory_resource& mr, std::size_t bytes, + std::size_t align) { + std::vector ptrs; + for(int i = 0; i < n; ++i) + ptrs.push_back(mr.allocate_fast(bytes, align)); + for(void* p : ptrs) + mr.deallocate_fast(p, bytes, align); + churn(mr); + return ptrs; + }); + BOOST_TEST_EQ(found, std::size_t(n)); + } + + // A burst the thread repeats keeps its blocks cached on that thread. + void + testRecurringBurstStaysCached() + { + constexpr int n = 64; + auto found = blocks_reaching_other_thread( + [](recycling_memory_resource& mr, std::size_t bytes, + std::size_t align) { + std::vector ptrs; + for(int round = 0; round < 1 << 12; ++round) + { + ptrs.clear(); + for(int i = 0; i < n; ++i) + ptrs.push_back(mr.allocate_fast(bytes, align)); + for(void* p : ptrs) + mr.deallocate_fast(p, bytes, align); + } + return ptrs; + }); + BOOST_TEST_EQ(found, std::size_t(0)); + } + + // Idle blocks the full global pool has no room for are released, + // not kept by the thread that cached them. + void + testIdleBurstIntoFullGlobalPool() + { + constexpr std::size_t bytes = 256; // size class 2 + constexpr std::size_t align = 8; + constexpr int cap = 1024; // class capacity + constexpr int n = 64; + + std::vector burst; + std::size_t found = 0; + std::latch held(1); + std::latch full(1); + + std::thread owner([&] { + recycling_memory_resource mr; + for(int i = 0; i < n; ++i) + burst.push_back(mr.allocate_fast(bytes, align)); + for(void* p : burst) + mr.deallocate_fast(p, bytes, align); + held.count_down(); + full.wait(); + churn(mr); + std::vector again; + for(int i = 0; i < n; ++i) + again.push_back(mr.allocate_fast(bytes, align)); + for(void* p : again) + found += static_cast( + std::count(burst.begin(), burst.end(), p)); + for(void* p : again) + mr.deallocate_fast(p, bytes, align); + }); + held.wait(); + + // Fills this thread's pool, then the global one; the exit drain + // empties only the former. + std::thread([] { + recycling_memory_resource mr; + std::vector ptrs; + for(int i = 0; i < 2 * cap + n; ++i) + ptrs.push_back(mr.allocate_fast(bytes, align)); for(void* p : ptrs) mr.deallocate_fast(p, bytes, align); - ptrs.clear(); - }; + }).join(); + full.count_down(); + owner.join(); + + BOOST_TEST_EQ(found, std::size_t(0)); + } + + // Stocks the global pool's bucket for `bytes` from a thread that then + // exits. Returns the blocks freed once that thread's own pool was + // full; if no decay pass ran on it, they fill the whole bucket. + static std::vector + stock_global_pool(std::size_t bytes, int cap) + { + std::vector stocked; + std::thread([&] { + recycling_memory_resource mr; + std::vector ptrs; + for(int i = 0; i < 2 * cap; ++i) + ptrs.push_back(mr.allocate_fast(bytes, 8)); + for(void* p : ptrs) + mr.deallocate_fast(p, bytes, 8); + stocked.assign(ptrs.begin() + cap, ptrs.end()); + }).join(); + return stocked; + } + + // A thread that only allocates still hands back a refill it stopped + // using, as long as it goes on refilling other size classes. + void + testAllocateOnlyThreadDecays() + { + constexpr std::size_t idle_bytes = 256; // size class 2 + constexpr int idle_cap = 1024; + constexpr std::size_t busy_bytes = 64; // size class 0 + constexpr int busy_cap = 4096; + // A refill after the first is nearly a whole bucket, so a decay + // pass runs about every second round. Two passes are needed, as + // the first only marks the idle blocks; the rest is margin. + constexpr int rounds = 8; + + auto stocked = stock_global_pool(idle_bytes, idle_cap); + std::barrier<> sync(2); + + // Takes the whole stocked bucket with one allocation, then only + // allocates from another class. + std::thread owner([&] { + recycling_memory_resource mr; + void* idle = mr.allocate_fast(idle_bytes, 8); + std::vector busy; + for(int r = 0; r < rounds; ++r) + { + sync.arrive_and_wait(); + for(int i = 0; i < busy_cap; ++i) + busy.push_back(mr.allocate_fast(busy_bytes, 8)); + sync.arrive_and_wait(); + } + sync.arrive_and_wait(); + mr.deallocate_fast(idle, idle_bytes, 8); + for(void* p : busy) + mr.deallocate_fast(p, busy_bytes, 8); + }); + for(int r = 0; r < rounds; ++r) + { + stock_global_pool(busy_bytes, busy_cap); + sync.arrive_and_wait(); + sync.arrive_and_wait(); + } + + std::size_t found = 0; + std::thread([&] { + recycling_memory_resource mr; + std::vector taken; + for(int i = 0; i < idle_cap; ++i) + taken.push_back(mr.allocate_fast(idle_bytes, 8)); + for(void* p : taken) + found += static_cast( + std::count(stocked.begin(), stocked.end(), p)); + for(void* p : taken) + mr.deallocate_fast(p, idle_bytes, 8); + }).join(); + sync.arrive_and_wait(); + owner.join(); - fill(); - drain(); // overflows the global pool -> direct delete - fill(); // drains the global pool back into the local one - drain(); + BOOST_TEST(found > 0); } void @@ -130,7 +399,7 @@ class recycling_memory_resource_test }; thread_local late_user user; - // Fill both the thread's pool and the shared one. + // Cache blocks for the exit-time drain to release. recycling_memory_resource mr; std::vector ptrs; for(int i = 0; i < 40; ++i) @@ -147,6 +416,11 @@ class recycling_memory_resource_test { testIsEqual(); testSlowPaths(); + testRecyclesWideFanOut(); + testIdleBurstDecays(); + testRecurringBurstStaysCached(); + testIdleBurstIntoFullGlobalPool(); + testAllocateOnlyThreadDecays(); testThreadExitTeardown(); testUseAfterThreadExitDrain(); }