From d94bfe64b558d62ab8c50dc6a2362f56492629d0 Mon Sep 17 00:00:00 2001 From: Ichthyostega Date: Wed, 13 May 2026 19:57:50 +0200 Subject: [PATCH] BufferProvider: validate (and fix) concurrent access The test performs a massively concurrent scenario that presses with 100 threads and 1 MiB allocation for each. This highlighted my still limited understanding of the situation and leads to a much better solution, where, on-demand, some worker performs all the accrued allocation and management work. So effectively we do not even need to use the synchronous API --- src/vault/mem/engine-buffer-manager.hpp | 16 +- src/vault/mem/local-mem-pool.hpp | 3 +- .../vault/mem/engine-buffer-manager-test.cpp | 107 +++-- wiki/thinkPad.ichthyo.mm | 366 ++++++++++++++++-- 4 files changed, 401 insertions(+), 91 deletions(-) diff --git a/src/vault/mem/engine-buffer-manager.hpp b/src/vault/mem/engine-buffer-manager.hpp index 08866ae0a..7a082bf24 100644 --- a/src/vault/mem/engine-buffer-manager.hpp +++ b/src/vault/mem/engine-buffer-manager.hpp @@ -75,7 +75,18 @@ namespace mem { Buff* mem; size_t siz; - bool empty() const { return not bool(mem); } + bool + empty() const + { + return not bool(mem); + } + + void + discard() + { + mem = nullptr; + siz = 0; + } }; @@ -100,9 +111,10 @@ namespace mem { * implementation code from a subclass will have to * receive and dispatch those entries */ void - supply (Alloc const& alloc) + supply (Alloc& alloc) { inQueue_.push (alloc); + alloc.discard(); } }; diff --git a/src/vault/mem/local-mem-pool.hpp b/src/vault/mem/local-mem-pool.hpp index 5bd4643f7..a147f0b91 100644 --- a/src/vault/mem/local-mem-pool.hpp +++ b/src/vault/mem/local-mem-pool.hpp @@ -227,7 +227,8 @@ namespace mem { void supply (Buff* mem, size_t siz) { - AllocReceiver::supply ({mem,siz}); + Alloc alloc{mem,siz}; + AllocReceiver::supply (alloc); } /** diff --git a/tests/vault/mem/engine-buffer-manager-test.cpp b/tests/vault/mem/engine-buffer-manager-test.cpp index e3f09925f..c5922f693 100644 --- a/tests/vault/mem/engine-buffer-manager-test.cpp +++ b/tests/vault/mem/engine-buffer-manager-test.cpp @@ -18,25 +18,22 @@ #include "test/run.hpp" #include "test/test-helper.hpp" -//#include "vault/mem/engine-buffer-metadata.hpp" #include "vault/mem/engine-buffer-manager.hpp" #include "vault/mem/engine-buffer-allocator.hpp" #include "lib/scoped-collection.hpp" #include "lib/iter-explorer.hpp" -//#include "lib/depend-inject.hpp" +#include "lib/sync-barrier.hpp" #include "lib/thread.hpp" -//#include "lib/error.hpp" #include "lib/symbol.hpp" -#include "test/diagnostic-output.hpp"////////////TODO #include -//#include +#include #include -using lib::Thread; -using lib::Literal; using lib::explore; -//using lib::ScopedCollection; +using lib::Literal; +using lib::Thread; +using lib::SyncBarrier; using std::this_thread::yield; using std::this_thread::sleep_for; using std::chrono::microseconds; @@ -46,28 +43,14 @@ namespace vault { namespace mem { namespace test { -// using LERR_(LOGIC); -// using LERR_(LIFECYCLE); - - - namespace { // Test helper - - const size_t ALLOC_REQ = 1024*1024; - const size_t NUM_THREADS = 100; -// template -// Buff* -// mark_as_Buffer(X& something) -// { -// return reinterpret_cast (std::addressof(something)); -// } -// + namespace { // Test parameters...... + const size_t ALLOC_REQ = 1024*1024; ///< use test-allocation request of 1 MiB size + const size_t NUM_THREADS = 100; ///< press concurrently } - - /*****************************************************************************//** * @test Show how the central buffer manager of the Render Engine can be accessed. */ @@ -77,12 +60,9 @@ namespace test { virtual void run (Arg) { - seedRand(); - demonstrate_AllocatorInterface(); verify_syncRequest(); verify_asyncRequest(); - verify_lulz(); } @@ -150,7 +130,7 @@ namespace test { // can use that memory char* buf = reinterpret_cast (alloc.mem); - string ranS{lib::randStr(555-1)}; + std::string ranS{lib::randStr(555-1)}; auto* p = std::copy (ranS.begin(), ranS.end(), buf); CHECK (ranS.length() == size_t(p - buf)); *p = '\0'; @@ -175,7 +155,16 @@ namespace test { - /** @test show that an working allocation request passes thread boundaries */ + /** @test show a working allocation request passed over thread boundaries... + * - use a simplified »local pool« that actually just provides an in-queue + * - start a lot of threads that perform the typical allocation sequence + * + first send an asynchronous allocation request + * + some delay while doing other stuff (here: sleep) + * + check if allocation has been serviced + * + if not, then perform global allocation servicing in this worker + * + allocation must be available now, can use it + * + when done, send allocation back to the global pool + */ void verify_asyncRequest() { @@ -196,42 +185,44 @@ namespace test { return inQueue_.empty(); } }; - + EngineBufferManager manager; using WorkBuff = std::array; -std::atomic_int cnt{0}; - auto workSeq = [&, gen = makeRandGen()]// local random generator per thread - () mutable - {// sequence of actions within a worker thread -int i=++cnt; + SyncBarrier afterThread{NUM_THREADS+1}; + auto workSeq = [&]{ // sequence-of-actions-within-worker-thread------------------- LocalTestPool myPool; CHECK (myPool.empty()); - + manager.async_requestAllocation (myPool, ALLOC_REQ); - - // meanwhile the worker does some other "work" - uint delay = 100 + gen.i(800); + + // meanwhile the worker does some other "work"... + auto gen = makeRandGen(); // local random generator per thread + uint delay = 100 + gen.i(900); sleep_for (microseconds(delay)); -SHOW_EXPR(delay) Alloc alloc = myPool.retrieveAlloc(); if (alloc.empty()) - // they let us down ... - // insist harder to be serviced thusly - alloc = manager.requestAllocation (ALLOC_REQ); + { // they let us down ... + // insist harder to be serviced thusly + manager.processPendingRequests(); + alloc = myPool.retrieveAlloc(); + // Explanation: since the async_requestAllocation() »happens-before« the processPendingRequests() + } // our request *must* have been processed and dispatched now, unless there was a failure CHECK (not alloc.empty()); CHECK (ALLOC_REQ <= alloc.siz); // can use that allocation for some "serious work"... auto content = new(alloc.mem) WorkBuff; std::ranges::generate (*content, [&]{ return gen.u64(); }); - sleep_for (10ms); + + // coordinate end of all threads + afterThread.sync(); // pass allocation back to the central hub manager.supply (alloc); -SHOW_EXPR(i) + CHECK (myPool.empty()); // no further pending / duplicate Alloc in-queue }; using Threads = lib::ScopedCollection; @@ -240,31 +231,21 @@ SHOW_EXPR(i) // Start a collection of workers.... Threads threads{NUM_THREADS, startThread}; + // after that barrier, all threads send back their allocation + afterThread.sync(); + // wait for all threads to finish... while (explore(threads).has_any()) yield(); // all the allocations sent back by the workers - // should sit in the EngineBufferManager's input queue -SHOW_EXPR(watch(manager).numAllocs()); -// CHECK (NUM_THREADS == watch(manager).numAllocs()); -SHOW_EXPR(watch(manager).bytesLeased()); -// CHECK (NUM_THREADS*ALLOC_REQ <= watch(manager).bytesLeased()); + // should sit now in the EngineBufferManager's input queue + CHECK (NUM_THREADS == watch(manager).numAllocs()); + CHECK (NUM_THREADS*ALLOC_REQ <= watch(manager).bytesLeased()); // process in-queue and retrieve all allocations.. manager.processPendingRequests(); -SHOW_EXPR(watch(manager).numAllocs()); -SHOW_EXPR(watch(manager).bytesLeased()); -// CHECK (0 == watch(manager).bytesLeased()); - } - - - - /** @test - */ - void - verify_lulz() - { + CHECK (0 == watch(manager).bytesLeased()); } }; diff --git a/wiki/thinkPad.ichthyo.mm b/wiki/thinkPad.ichthyo.mm index df37ccc83..b9d1eab6f 100644 --- a/wiki/thinkPad.ichthyo.mm +++ b/wiki/thinkPad.ichthyo.mm @@ -146679,23 +146679,34 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + - + + + + - - + + + + + + + + + + - + @@ -146705,11 +146716,113 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) + + + + +

+ bezüglich Code-Struktur, da Teile des asynchronen Verarbeitungs-Pfades im LocalMemPool liegen +

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

+ und zwar als zweiter Code-Pfad, nachdem zunächst eine asynchrone Anfrage abgesetzt wurde, und dennoch an der Stelle des Bedarfs die Allokation noch nicht bereitgestellt wurde. Deshalb dient diser »synchrone Code-Pfad« dazu, JETZT eine DEFINITIVE Antwort zu bekommen (ggfs. einen Fehler) +

+ + + + + + + +

+ ...das wäre zwar der naheliegende Ansatz, würde hier aber die Sache sehr komplex machen, da ich nicht nur die definitive Antwort möchte, sondern zusätzlich auch noch Duplikate vermeiden. +

+ + + + + + + +

+ ich hatte bereits die zutreffende Antwort gefunden, diese aber nicht »realisiert« +

+ + + + + +

+ Nochmal: die Lösung besteht darin, daß der asynchrone Code-Pfad in die synchrone Feststellung eingeschlossen wird. Aber eben nicht so, daß ich (am asynchronen Codepfad vorbei) einen synchronen API-Aufruf mache. +

+ + + + + + + + + + + + + + + + - - + + + + + + + + + + + + + + +

+ der async-Request happens-before dem synchronen globalen Processing +

+ + + + + + + + + + + + @@ -146819,9 +146932,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - - +

im Grunde müßte man nur den Zugang zum Memory-Manager selber locken, denn die ganzen Queue-Operationen sind ja schon lock-free und threadsafe; auch könnten diese Locks dann non-recursive sein @@ -146832,9 +146943,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - - +

...andererseits ist jeder (Mutex)-Lock ein Kernel-Aufruf, und wir erwarten durchaus ganze Bündel an Requests. Außerdem wird mit wenigen Locks auf API-Level viel klarer, daß das Locking vollständig, und der Zustand konsistent ist. @@ -146846,9 +146955,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - - +

...und zwar sagt mir mein Bauchgefühl, daß wir hier keine Situation mit starker Contention haben; das hat sich auch bereits im Zusammenhang mit dem Scheduler gezeigt. Im Regelfall erwarte ich, daß die Anfragen durch die Queue gehen, und damit ohne Interferenz aneinander vorbeikommen. Wenn dann tatsächlich mal ein Thread den synchronen Pfad einschlägt, erscheint es mir sinnvoller, alle Arbeit in einem Schwung zu erledigen, selbst wenn dadruch (ohenhin nur ausnahmsweise!) ein anderer Thread mal geblockt wird. Letztlich gilt hier immer: die Jobs sind vergleichsweise groß. @@ -146931,9 +147038,10 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + - + + @@ -146943,9 +147051,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - - +

hatte diese Technik vom incidence-count-test kopiert; der Code dort ist aber nicht ganz korrekt — was dort aber nicht auffällt, da das thread-Microbenchmark seine test-Funktion (das »subjekt«) nochmal kopiert, nämlich in die closure des jeweiligen Thread. Diese Kopie bewirkt dann einfach, daß der Generator nochmal kopiert wird ... bin mir aber trotzdem nicht sicher, so wirklich verschiedene Zufallszahlen gezogen werden. @@ -146963,11 +147069,223 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + + + + + + + + + + + +

+ Genauer: das Szenario ist nicht so, wie von mir geplant. Es ist zwar tatsächlich massiv concurrent, und das müßte das Subject auch verkraften; aber das Verhalten ist dadurch nicht mehr verifzierbar +

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

+ Randomisierung ist hier grundsätzlich nicht reproduzierbar +

+ + + + + +

+ Das liegt daran, daß das relative Verhalten der Threads zueinander eine Rolle spielt. Solche zeitlichen Verschiebungen enthalten ein Element echter Entropie, wenngleich auch die Qualität der Entropie unterschiedlich sein kann. Man kann zwar versuchen, diese Entropie zu reduzieren, und ein solches Unterfangen ist gefährlich, weil es einen Grad der Kontrolle suggeriert, der letztlich nicht gegeben ist. +

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

+ beides ist hier komplett egal +

+ + + + + +

+ solange wir irgendwie verschiedene Werte in den Threads bekommen, is gut +

+ + + + + + + + +

+ Da der Seed irgend ein 64bit-uint ist, ist das Schlimmste, was passieren kann, daß zwei Threads einen lokalen Generator mit dem gleichen Seed und damit der gleichen Sequenz haben. Dieser Fall ist aber nicht die Regel +

+ + + + + + + + + + +

+ ...was auch eigentlich nicht verwunderlich ist, denn wir haben zwar einen Sleep, aber auch eine signifikante CPU-Load (1 Million Speicherzugriffe) +

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

+ wenn ich am Ende in jedem Thread die Queue leere, dann stimmt die Bilanz am Ende +

+ + + + + + + +

+ Typische Werte schwanken zwischen 40 und 80 zusätzlichen Allokationen +

+ + + + + + + +

+ zwischen dem Prüfen auf die leere Queue und dem synchronen Aufruf für eine Zuteilung +

+ + + + + + + +

+ ...da man das Lock erlangen muß, um die synchrone Anfrage zu tätigen, verstärkt sich eine einzige zufällige Kollision: der benachteiligte Thread wartet, erlangt das Lock und bringt dadruch andere Threads zum Warten. Immer wenn ein Thread auf das Lock wartet, ist danach auch sicher seine asynchrone Anfrage bereits bedient +

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

+ ich dachte, ich kann mich dem Problem entwinden, indem ich beide auf den gleichen Basis-Funktionen aufbaue; der Test zeigt aber, daß das nicht aufgeht +

+ + + + + + + + + +

+ Zusammengefaßt: ich brauche einen synchronen Abgleich zur Herstellung der Konsistenz, sofern der zunächt ausgelöste asynchrone Abgleich nicht erfolgreich war. Aber das impliziert nicht einen Aufruf des synchronen API (das ich nun tatsächlich nur für den Test gebaut habe). Vielmehr besteht die elegante Lösung darin, ein »happens-before« herzustellen (in dem wir »jetzt und hier« das globale Processing machen); danach muß der zunächst gesendete asynchrone Request auch tatsächlich behandelt und zugestellt sein, ohne dafür noch extra etwas zu unternehmen.  +

+ + + + + + + + + + + + + + + @@ -171692,9 +172010,7 @@ std::cout << tmpl.render({"what", "World"}) << s - - - +

...nämlich indem der Job letztlich auf den synchronen Fall zurückgreift, dieser aber durch die gleiche Mechanik implementiert ist, wie der asynchrone Fall, und dabei gleich alle aufgesammelten Requests in den Queues mit verarbeitet. Dadruch wird effektiv ein Batch an Allokator-Aufwand in denjenigen Worker geschoben, der rein zufällig als erster auf eine nicht bediente Allokation stößt. Das Schöne an dieser Code-Struktur ist, daß die Möglichkeit offen bleibt, später zur Performance-Optimierung (oder nur in bestimmten Situationen) dennoch etwas speziell einzurichten / schedulen / auszuführen, ohne daß dazu der Code im EngineBufferManager geändert werden muß