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 +

+ + + + + + +