From 0256ad85280c584cb064fb363c4ceab1d1ac3059 Mon Sep 17 00:00:00 2001 From: Ichthyostega Date: Sun, 24 May 2026 18:44:32 +0200 Subject: [PATCH] BufferProvider: cover capacity anouncement in unit test ...need an elaborate test setup, with the test worker thread and the main thread proceeding in lockstep, otherwise the effect would not be visible. Furthermore, the diagnostic access for LocalMemPool can be re-used here, thereby mapping requests onto the pool for the current thread. --- src/vault/mem/local-buffer-store.hpp | 34 +++++--- src/vault/mem/local-mem-pool.hpp | 13 +++ tests/vault/mem/local-buffer-store-test.cpp | 94 ++++++++++++++++++++- wiki/thinkPad.ichthyo.mm | 66 +++++++++++++-- 4 files changed, 188 insertions(+), 19 deletions(-) diff --git a/src/vault/mem/local-buffer-store.hpp b/src/vault/mem/local-buffer-store.hpp index 3ab46a93d..b93cc6664 100644 --- a/src/vault/mem/local-buffer-store.hpp +++ b/src/vault/mem/local-buffer-store.hpp @@ -53,6 +53,9 @@ namespace mem { using util::_Fmt; + class LocalStoreDiagnostic; + + /** * Production-grade implementation of buffer allocation @@ -98,8 +101,8 @@ namespace mem { } }; - LocalPool localReg_; - GlobalPool globalReg_; + LocalPool localPool_; + GlobalPool globalPool_; public: LocalBufferStore() @@ -118,9 +121,9 @@ namespace mem { uint prepareBuffers (HashVal, uint numBuffers, size_t sizRequest) override { - uint cnt = localReg_->reserve(numBuffers, sizRequest); + uint cnt = localPool_->reserve(numBuffers, sizRequest); for (uint n{cnt}; n < numBuffers; ++n) - globalReg_().async_requestAllocation (*localReg_, sizRequest); + globalPool_().async_requestAllocation (*localPool_, sizRequest); // signal back the number of buffers reserved and thus immediately available return cnt; } @@ -142,16 +145,16 @@ namespace mem { BuffAlloc provideBuffer (HashVal, size_t sizRequest, LocalTag specifics, int64_t) override { - if (not localReg_->canServe(sizRequest)) - globalReg_().processPendingRequests(); + if (not localPool_->canServe(sizRequest)) + globalPool_().processPendingRequests(); // attempt to satisfy allocation request from the local pool... - Alloc alloc = localReg_->retrieve (sizRequest); + Alloc alloc = localPool_->retrieve (sizRequest); auto& [storage,buffSiz] = alloc; // possibly the request was not announced beforehand...? if (not storage) - alloc = globalReg_().requestAllocation (sizRequest); + alloc = globalPool_().requestAllocation (sizRequest); if (not storage) throw err::State{_Fmt{"Unable to provide further %d bytes of buffer memory."} % sizRequest ,LUMIERA_ERROR_BUFFER_MANAGEMENT}; @@ -181,12 +184,23 @@ namespace mem { { auto& [storage,buffSiz,specifics] = storageSlot; REQUIRE (storage); - localReg_->reSupply (storage); + localPool_->reSupply (storage); /////////////////////////////////////////////////////OOO trigger heuristic clean-up here } + + private: + /** diagnostic access for unit testing */ + friend PoolDiagnostic watch (LocalBufferStore const&); }; - + + /** entrance point to inspection for test */ + inline PoolDiagnostic + watch (LocalBufferStore const& lstor) + { + return PoolDiagnostic{*lstor.localPool_}; + } + }} // namespace vault::mem #endif /*VAULT_MEM_LOCAL_BUFFER_STORE_H*/ diff --git a/src/vault/mem/local-mem-pool.hpp b/src/vault/mem/local-mem-pool.hpp index 0b8e104cf..6d11661f4 100644 --- a/src/vault/mem/local-mem-pool.hpp +++ b/src/vault/mem/local-mem-pool.hpp @@ -508,6 +508,19 @@ namespace mem { ,[](Block const& b){ return not b.used; }); } + size_t + cntResd() + { + return std::ranges::count_if (memPool_.blocks_ + ,[](Block const& b){ return b.resd; }); + } + + bool + canServe (size_t sizRequest) + { + return memPool_.canServe (sizRequest); + } // map to the pool's diagnostic API for uniform access + bool isFree (Buff* mem) { diff --git a/tests/vault/mem/local-buffer-store-test.cpp b/tests/vault/mem/local-buffer-store-test.cpp index f96eb83f4..e637b7fd1 100644 --- a/tests/vault/mem/local-buffer-store-test.cpp +++ b/tests/vault/mem/local-buffer-store-test.cpp @@ -27,6 +27,7 @@ #include "test/diagnostic-output.hpp"///////////////////////TODO #include +#include using util::isSameAdr; @@ -97,8 +98,8 @@ namespace test { lib::DependInject::Local globalPool; // provide the *test subject*, - // which implements the BufferStage API - // and uses a thread-local metadata table internally + // which implements the BufferStore API + // and uses a thread-local allocation pool internally LocalBufferStore localStore; BufferProviderSetup::Store& storeAPI = localStore; @@ -184,11 +185,98 @@ namespace test { - /** @test + + /** @test announcement of capacity demand is propagated eventually + * - test thread must proceed in lockstep to make effects observable + * - Step-1: test thread announces demand on the subject (`BufferStore` API) + * - Step-2: „someone“ processes pending requests in the EngineBufferManager + * - Step-3: test thread announces further demand that remains unprocessed + * - test thread then terminates, sends back all allocations + * - the pending request is ignored transparently */ void verify_announce() { + // »central BufferPool« (transient instance for this test) + lib::DependInject::Local globalPool; + + // *test subject* + LocalBufferStore localStore; + BufferProviderSetup::Store& storeAPI = localStore; + + const size_t BUFFSIZ = 1024; + const HashVal TYPEID = 12345; + + // coordination of test steps + std::atomic_uint step{1}; + + Thread testWorker{[&] /* === Scenario: test subject announces capacity demand === */ + { + // Step-1 + uint avail = storeAPI.prepareBuffers(TYPEID, 1, BUFFSIZ); + CHECK (0 == avail); + + CHECK (watch(localStore).isEmpty()); + + // wait for being serviced... + ++step; + while (step < 3) + sleep_for(1ms); + + // Step-3 + CHECK (not watch(localStore).isEmpty()); + CHECK (watch(localStore).canServe(BUFFSIZ)); + CHECK (watch(localStore).cntFree() == 1); + CHECK (watch(localStore).size() == 1); + + // Note: this block was supplied but the local pool does not know why, + // since reservations are only tracked on blocks already available + CHECK (watch(localStore).cntResd() == 0); + + // Now, instead of using this buffer block, + // rather generate another capacity announcement... + avail = storeAPI.prepareBuffers(TYPEID, 2, BUFFSIZ); + CHECK (1 == avail); + + CHECK (watch(localStore).cntResd() == 1); + CHECK (watch(localStore).cntFree() == 1); + CHECK (watch(localStore).size() == 1); + CHECK (watch(localStore).canServe(BUFFSIZ)); + + // Note: while a further request is "out there", + // only a single block was received up to now. + // LocalMemPool winds down automatically at end of thread + }}; + + + /* === Main thread: observe effects in the global pool === */ + + // wait until the first announcement was made + while (step < 2) + sleep_for(1ms); + + // Step-2 + CHECK (watch(*globalPool).numAllocs() == 0); + + // process the in-queue and service the allocation request... + globalPool->processPendingRequests(); + + CHECK (watch(*globalPool).numAllocs() == 1); + CHECK (watch(*globalPool).bytesLeased() >= BUFFSIZ); + + // let the test thread proceed... + step = 3; + + // wait until test thread has wound down + while (testWorker) + sleep_for(1ms); + + // additional delay to be sure the destructor of the local pool has finished + sleep_for(1ms); + + CHECK (watch(*globalPool).numAllocs() == 1); + globalPool->processPendingRequests(); + CHECK (watch(*globalPool).bytesLeased() == 0); } diff --git a/wiki/thinkPad.ichthyo.mm b/wiki/thinkPad.ichthyo.mm index fb4deb534..62b49d7da 100644 --- a/wiki/thinkPad.ichthyo.mm +++ b/wiki/thinkPad.ichthyo.mm @@ -145226,8 +145226,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) Man müßte dazu also den AllocReceiver umformulieren und zudem auch noch die Requests im EngineBufferManager. Vermutlich läuft das auf einen speziellen smart-Handle-Typ hinaus. Dieser würde eine Prüfsumme beinhalten, die dann auf der Empfängerseite genauso im Speicher vorzufinden sein muß. Die Prüfsumme würde auf einer Nonce plus der Ziel-Addresse beruhen, und könnte damit so nur ein einziges Mal pro Programmlauf auftreten (modulo 64-bit Hash-Kollision). Die Sperr-Flag bestünde dann einfach darin, die empfängerseitige Prüfsumme auf Null zu setzen (kein Match, kein Zugriff). Diese empfängerseitige Prüfsumme müßte ein Atomic sein, und beim Zugriff würde eine Lese-Barriere angewendet, beim Ändern eine Schreibbarriere. Das wäre völlig akzeptabel. Darüberhinaus gäbe es keine Contention (außer beim Neuerzeugen eines Objekts, weil dann die Nonce inkrementiert werden muß).

- - + @@ -145245,8 +145244,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)

- + @@ -145592,12 +145591,67 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - - + + + + + + + + + + +

+ ...denn nach dem Thread-Ende wird die Allokation automatisch zurückgeführt, so daß sich der Effekt nicht mehr beobachten läßt +

+ + + + + + + + + + + + + + +

+ ...wenn man das erst mal akzeptiert, ist es auch nicht so schlimm: man inkrementiert einen Atomic +

+ + + + + + + + + + + + +

+ ein weiters Announcement, +

+

+ das nicht mehr zugestellt wird +

+

+ wird transparent ignoriert +

+ + + + + + +