diff --git a/src/vault/mem/engine-buffer-manager.cpp b/src/vault/mem/engine-buffer-manager.cpp index c2d576754..732dcbaf8 100644 --- a/src/vault/mem/engine-buffer-manager.cpp +++ b/src/vault/mem/engine-buffer-manager.cpp @@ -33,13 +33,10 @@ #include "lib/error.hpp" #include "lib/util-quant.hpp" #include "lib/allocator-handle.hpp" -//#include "lib/hash-value.h" #include "vault/mem/buffhandle.hpp" #include "vault/mem/engine-buffer-manager.hpp" #include "vault/mem/engine-buffer-allocator.hpp" -//#include "vault/mem/buffer-local-tag.hpp" #include "lib/nocopy.hpp" -//#include "lib/util.hpp" #include #include @@ -47,23 +44,11 @@ namespace vault{ namespace mem { - -// using lib::HashVal; -// using util::unConst; - namespace err = lumiera::error; using std::byte; - namespace { // config and internal helpers... - - } - - - - - using BufferAllocator = HeapBufferAllocator; @@ -84,10 +69,11 @@ namespace mem { inQueue_ .consume_all ([this](Alloc alloc){ doRedeemAllocation (alloc); }); requestQueue_.consume_all ([this](AllocRequest request) { - REQUIRE (request.receiver); + if (not request.receiver) return; Alloc alloc = doPerformAllocation (request.sizRequest); ENSURE (not alloc.empty()); - request.receiver->supply (alloc); + if (request.receiver) + request.receiver->supply (alloc); }); } diff --git a/src/vault/mem/engine-buffer-manager.hpp b/src/vault/mem/engine-buffer-manager.hpp index 7a082bf24..a161d54a4 100644 --- a/src/vault/mem/engine-buffer-manager.hpp +++ b/src/vault/mem/engine-buffer-manager.hpp @@ -40,23 +40,16 @@ #define VAULT_MEM_ENGINE_BUFFR_MANAGA_H -//#include "lib/hash-value.h" #include "vault/mem/buffhandle.hpp" -//#include "vault/mem/type-handler.hpp" -//#include "vault/mem/buffer-local-tag.hpp" +#include "lib/live-mark.hpp" #include "lib/nocopy.hpp" #include "lib/sync.hpp" -//#include "lib/util.hpp" #include namespace vault{ namespace mem { - -// using lib::HashVal; -// using util::unConst; - namespace err = lumiera::error; @@ -98,6 +91,8 @@ namespace mem { : util::NonCopyable { protected: + lib::LiveMark mark_{*this}; + using InQueue = boost::lockfree::queue; InQueue inQueue_; @@ -116,6 +111,21 @@ namespace mem { inQueue_.push (alloc); alloc.discard(); } + + /** allow safe access through a smart-handle, + * to prevent interaction with an deceased partner */ + auto + getSafeHandle() + { + return mark_.makeHandle (*this); + } + + /** accessor to allow verification of liveness state */ + lib::LiveMark const& + getLiveMark() const + { + return mark_; + } }; @@ -130,7 +140,8 @@ namespace mem { { struct AllocRequest { - AllocReceiver* receiver; + using Handle = lib::LiveMark::SafeHandle; + Handle receiver; size_t sizRequest; }; @@ -196,7 +207,7 @@ namespace mem { EngineBufferManager::async_requestAllocation (AllocReceiver& receiver, size_t sizRequest) { REQUIRE (sizRequest); - requestQueue_.push ({&receiver, sizRequest}); + requestQueue_.push ({receiver.getSafeHandle(), sizRequest}); } diff --git a/src/vault/mem/local-buffer-store.hpp b/src/vault/mem/local-buffer-store.hpp index 5623afea0..3ab46a93d 100644 --- a/src/vault/mem/local-buffer-store.hpp +++ b/src/vault/mem/local-buffer-store.hpp @@ -13,7 +13,23 @@ /** @file local-buffer-store.hpp ** Production-grade implementation of engine::BufferProvider memory handling. - ** + ** All functionality is delegated to further components. Notably, each thread + ** maintains a LocalMemPool so that allocations can preferably be satisfied + ** by reusing memory blocks currently associated to this thread. Excess capacity + ** or further allocation demand is coordinated with the EngineBufferManager via + ** asynchronous messaging. + ** + ** However, in accordance with the architecture of the Render Engine, there is + ** no dedicated _memory manager thread_ — rather, the worker threads perform + ** the accrued memory management effort on-demand. Allocation requests are + ** placed into the queue, yet whenever some worker runs into some actual + ** memory demand not yet satisfied, it acquires a global lock and performs + ** all the requests queued thus far, in one batch. This scheme allows to + ** allocate all available cores to the workers, while preventing global + ** management efforts to push aside some worker. The same pattern is + ** also used for the scheduler: each worker performs a chunk of the + ** necessary scheduler work, until the next runnable job is found. + ** ** @see LocalBufferStore_test ** @see buffer-provider.hpp ** @see buffer-provider-protocol-test.cpp @@ -39,7 +55,22 @@ namespace mem { /** - * @todo type comment + * Production-grade implementation of buffer allocation + * based on a thread-local pool that is connected asynchronously + * to the central EngineBufferManager (global pooling allocator). + * - announcement will send an allocation request to the global pool + * - allocations will either be served from the local pool or force + * a blocking round-trip into the EngineBufferManagert to perform + * all currently pending allocation requests for all worker threads. + * - the `emit()` call has no effect here; this differs from the + * OutputSlot and the cache manager case, where emit() is crucial. + * - buffer de-allocation will trigger a heuristic clean-up, taking + * into account how successfully each allocation in the pool was + * used recently, to prevent accumulating excess allocations. + * This implementation of the BufferProvider::BufferStore interface + * deliberately skips further sanity checks, beyond what is enforced + * through the BufferMetadata lifecycle. Allocation requests are + * passed to the LocalMemPool for the current thread. */ class LocalBufferStore : public BufferProviderSetup::Store @@ -75,6 +106,7 @@ namespace mem { { } + /* ==== Implementation of the BufferStore interface ==== */ /** API: announce demand for some buffer allocations. diff --git a/src/vault/mem/local-mem-pool.hpp b/src/vault/mem/local-mem-pool.hpp index 39713be08..0b8e104cf 100644 --- a/src/vault/mem/local-mem-pool.hpp +++ b/src/vault/mem/local-mem-pool.hpp @@ -102,6 +102,8 @@ namespace mem { using std::invocable; using util::unConst; using util::isLimited; + using std::this_thread::sleep_for; + using std::chrono_literals::operator ""us; class PoolDiagnostic; @@ -367,14 +369,21 @@ namespace mem { } /** unconditionally dispose of the pool's entire contents. - * @warning there must not be any allocations in use when invoking this function + * @warning there must not be any allocations in use when invoking this function + * @remark This function is intended to be used when shutting down this pool; + * in theory, a dangerous race could happen when a worker thread unwinds + * due to an unexpected error in the middle of a computation. + * EngineBufferManager might process pending allocation requests sent + * shortly before the failure, thereby accessing an already destroyed pool. + * The lib::LiveMark is used to protect against that situation */ template FUN> void purge (FUN&& consumer) { + mark_.disengage(); // »lock down« the in-queue at that point (do not accept further allocations) + sleep_for (100us); // pragmatic solution for race with a dispatch of further allocations; ingest(); - ///////////////////////////////////////OOO need a way to »lock down« the inqueue at that point for (Block& block : blocks_) { REQUIRE (not block.used); diff --git a/tests/vault/mem/local-buffer-store-test.cpp b/tests/vault/mem/local-buffer-store-test.cpp index f5d3b5558..f96eb83f4 100644 --- a/tests/vault/mem/local-buffer-store-test.cpp +++ b/tests/vault/mem/local-buffer-store-test.cpp @@ -18,19 +18,18 @@ #include "test/run.hpp" #include "test/test-helper.hpp" -//#include "vault/mem/engine-buffer-metadata.hpp" #include "vault/mem/local-buffer-store.hpp" #include "vault/mem/engine-buffer-manager.hpp" #include "vault/mem/engine-buffer-allocator.hpp" #include "lib/depend-inject.hpp" #include "lib/thread.hpp" -//#include "lib/error.hpp" #include "lib/util.hpp" -#include "test/diagnostic-output.hpp" +#include "test/diagnostic-output.hpp"///////////////////////TODO #include using util::isSameAdr; + using lib::Thread; using std::this_thread::yield; using std::this_thread::sleep_for; @@ -80,7 +79,17 @@ namespace test { - /** @test */ + /** @test demonstrate a typical usage cycle of the BufferStore in concurrent setup + * - the main thread provides a mock instance of the EngineBufferManager + * - furthermore, the LocalBufferStore instance is created here + * - yet note: LocalBufferStore includes a LocalMemPool per thread + * - a test thread is launched to issue some allocation requests + * - announce some memory allocation demand + * - claim and use one buffer + * - generate further demand to trigger further allocation supply + * - terminate the thread + * - verify the number of allocations seen in the EngineBufferManager + */ void verify_API() { @@ -163,18 +172,14 @@ namespace test { // before shutdown, LocalMemPool has sent back all allocations; // these should sit now in the EngineBufferManager's input queue, // unprocessed yet; same for the additional buffer requested later. -SHOW_EXPR(watch(*globalPool).numAllocs()); - CHECK ( CNT == watch(*globalPool).numAllocs()); - CHECK ((CNT)*ALLOC == watch(*globalPool).bytesLeased()); + // EngineBufferManager only sees the first CNT allocations actually dispatched + CHECK (CNT == watch(*globalPool).numAllocs()); + CHECK (CNT*ALLOC == watch(*globalPool).bytesLeased()); - // process in-queue and retrieve all allocations.. -SHOW_EXPR("Kaka") + // process the in-queue and retrieve all allocations... globalPool->processPendingRequests(); -SHOW_EXPR("Boom") -SHOW_EXPR(watch(*globalPool).bytesLeased()); -// CHECK (0 == watch(*globalPool).bytesLeased()); - sleep_for(1000ms); -SHOW_EXPR("bye") + // everything should be cleaned-up by now... + CHECK (0 == watch(*globalPool).bytesLeased()); } diff --git a/wiki/thinkPad.ichthyo.mm b/wiki/thinkPad.ichthyo.mm index 20a62668a..fb4deb534 100644 --- a/wiki/thinkPad.ichthyo.mm +++ b/wiki/thinkPad.ichthyo.mm @@ -144998,7 +144998,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -145012,7 +145012,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -145024,7 +145024,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -145052,7 +145052,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -145064,11 +145064,14 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) + + - + + - + @@ -145082,7 +145085,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -145149,7 +145152,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -145215,12 +145218,12 @@ 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+der Ziel-Addresse beruhen, und könnte damit so nur ein einziges Mal pro Programmlauf auftreten (modulo 63-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ß). + 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ß).

@@ -145233,7 +145236,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)
- + @@ -145242,6 +145245,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)

+ + @@ -145295,13 +145300,27 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + + + + + + + +

+ ...da es sich um einen Grenzfall handelt, den ich durch theoretische Überlegung identifiziert habe, und der im regulären Betrieb niemals auftreten sollte. Wenn es dazu kommt, wird sehr wahrscheinlich die Applikation insgesamt terminieren, d.h. in dem Fall verhindern wir nur diese lästige Art von nicht-reproduzierbaren SEGFAULT on shutdown. Falls es aber doch mal ein anderes Szenario gibt, würde das zu Speicherlecks führen. Daher denke ich, es genügt, das Restrisiko zu drücken (denn eine tatsächliche Lösung wäre ziemlich schwierig und würde das ganze Konzept der asynchronen Kommunikation konterkarrieren) +

+ +
+ +
+
@@ -145472,8 +145491,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - + + @@ -145562,11 +145581,14 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + - - + + + + + @@ -145739,7 +145761,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)
- + + @@ -145891,7 +145914,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -145970,7 +145993,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -146000,7 +146023,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -146168,16 +146191,24 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - + + + + + - + + + + + - + + @@ -146192,6 +146223,16 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) + + + + + + + + + + @@ -146847,7 +146888,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -146900,7 +146941,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -147058,9 +147099,9 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + - + @@ -147083,8 +147124,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - + + @@ -147097,10 +147138,14 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) + + + - + + - + @@ -147120,7 +147165,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -147163,7 +147208,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -147213,7 +147258,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + + @@ -147242,8 +147288,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - + + @@ -147252,7 +147298,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -147267,8 +147313,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)

+ -
@@ -147344,8 +147390,17 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)
+ + + + + + + - + + + @@ -147357,7 +147412,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -147383,8 +147438,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) ...denn letztlich dichte ich hier ein Restrisiko ab; es würde genügen, wenn der Thread nach dem Sperren des Empfangs noch einen Moment wartet, und dann erst die Eingangsqueue ausräumt. Wir können aber letztlich nicht verhindern, daß der Empfänger-Thread möglicherweise stirbt während der Sender-Thread grade mitten in der Zustellung ist, und durch äußere Umstände dabei ausgebremst wird. Um das zu verhindern, bräuchten wir einen Handshake — und dann wäre die Lösung nicht mehr asynchron!

- - + @@ -147395,7 +147449,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)
- + @@ -147436,7 +147490,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - +