diff --git a/src/lib/live-mark.hpp b/src/lib/live-mark.hpp new file mode 100644 index 000000000..1ef70784f --- /dev/null +++ b/src/lib/live-mark.hpp @@ -0,0 +1,179 @@ +/* + LIVE-MARK.hpp - safety-tag to prevent invoking a deceased partner + + Copyright (C) + 2026, Hermann Vosseler + +  **Lumiera** is free software; you can redistribute it and/or modify it +  under the terms of the GNU General Public License as published by the +  Free Software Foundation; either version 2 of the License, or (at your +  option) any later version. See the file COPYING for further details. + +*/ + +/** @file live-mark.hpp + ** Memory marker tag to prevent segfault in communication with a defunct receiver. + ** The tag comprises an unique hash value, that can be embedded into another object + ** and tied to the lifecycle. This setup allows to build a safe-handle, that offers + ** a liveness check of the communication partner and prevents dereferencing and + ** invoking into the stale memory of an expired receiver. Notably this allows + ** to safeguard asynchronous communication between thread-local entities, + ** even when some of the participating threads terminates unexpectedly. + ** + ** @todo WIP 5/2026 developing a thread-local allocation pool + ** + ** @see engine-buffer-manager.hpp usage example + ** @see LiveMark_test + ** + */ + +#ifndef LIB_LIVE_MARK_H_ +#define LIB_LIVE_MARK_H_ + +#include "lib/error.hpp" +#include "lib/hash-combine.hpp" +#include "lib/util.hpp" + +#include +#include + +namespace lib { + namespace err = lumiera::error; + + using lib::HashVal; + + + class LiveMark; + + template + concept alive_marked = requires (SUBJ const& subject) + { + { subject.getLiveMark() } -> std::convertible_to; + }; + + + /** + * Liveness marker tag to be incorporated into another object, + * so that a communication link can be severed automatically when + * the receiver object goes out of scope. + * @see AllocReceiver + * @see LiveMark_test + */ + class LiveMark + { + std::atomic id_; + + public: + LiveMark() = default; + ~LiveMark() { disengage(); } ///< @note the actual protection happens here + + /** + * Create a LiveMark that is connected + * to the Subject to be protected. + */ + LiveMark (alive_marked auto& subject) + : id_{generateID (subject)} + { } + + // using default copy/assignment + + /** deactivate and mark as dead (irreversible) */ + void disengage() { id_.store (0, std::memory_order_relaxed); } + + + /** + * A smart-handle for safe protected access to some object + * that might go out of scope eventually. This works under the + * assumption that the target object embeds a LiveMark, which + * is expressed through the concept alive_marked. The marker + * HashVal is unique for each instance and distinct memory + * location, and thus even access to a new object instance + * with the same type, placed into the same memory location + * will be rejected, after the original target was destroyed. + */ + template + class SafeHandle + { + SUBJ * target_{nullptr}; + HashVal check_{0}; + + public: + SafeHandle() = default; + // default copy operations + + /** liveness check */ + operator bool() const + { + return bool(target_) + and check_ == retrieveID (*target_); + } + + /* ==== subject access ==== */ + + SUBJ& + operator*() const + { + if (not bool(*this)) + throw err::State{"Accessing defunct target" + , LERR_(BOTTOM_VALUE)}; + return * target_; + } + + SUBJ* + operator->() const + { + return & SafeHandle::operator*(); + } + + private: + SafeHandle (SUBJ& target) + : target_{& target} + , check_{retrieveID(target)} + { } + + friend class LiveMark; + + /** The heart of the implementation: peek into the + * target location to see if a valid LiveMark is embedded. + */ + static HashVal + retrieveID (SUBJ& target) + { + return target.getLiveMark().id_.load (std::memory_order_relaxed); + } + }; + + /** + * Create a safe smart-handle to refer to the given subject, + * similar to a raw pointer. Yet the handle will bool-check + * to `false` when the target object was destroyed + */ + template + SafeHandle + makeHandle (SUBJ& subject) + { + return SafeHandle{subject}; + } + + private: + /** Generate unique HashVal. + * @remark seeded by a nonce (counter) and the memory address, + * which makes it impossible to get the same hash at the same + * target address more than once (unless there is a collision + * in the 64bit hash space, which is extremely rare) + */ + template + static HashVal + generateID (X const& targetAddress) + { + static std::atomic_size_t nonce; + HashVal mark = nonce.fetch_add (1, std::memory_order_relaxed); + + hash::combine (mark, util::addrID (targetAddress)); + return mark; + } + }; + + +} // namespace lib +#endif /*LIB_LIVE_MARK_H_*/ diff --git a/src/lib/thread.cpp b/src/lib/thread.cpp index 0f227f21e..7c14942aa 100644 --- a/src/lib/thread.cpp +++ b/src/lib/thread.cpp @@ -56,7 +56,7 @@ namespace thread{ ThreadWrapper::decorate_with_global_count (string const& rawID) { static atomic_uint globalCnt{1}; - return _Fmt{"%s.%03i"} % rawID % globalCnt.fetch_add (+1, std::memory_order_acq_rel); + return _Fmt{"%s.%03i"} % rawID % globalCnt.fetch_add (+1, std::memory_order_relaxed); } diff --git a/src/vault/mem/local-mem-pool.hpp b/src/vault/mem/local-mem-pool.hpp index a65fd7e05..39713be08 100644 --- a/src/vault/mem/local-mem-pool.hpp +++ b/src/vault/mem/local-mem-pool.hpp @@ -379,8 +379,8 @@ namespace mem { { REQUIRE (not block.used); consumer (block.alloc.mem, block.alloc.siz); - blocks_.clear(); } + blocks_.clear(); ENSURE (this->empty()); } diff --git a/tests/15library.tests b/tests/15library.tests index e0dfd6a2d..74f985dc3 100644 --- a/tests/15library.tests +++ b/tests/15library.tests @@ -478,6 +478,11 @@ return: 0 END +TEST "autodetect deceased partner" LiveMark_test < + +  **Lumiera** is free software; you can redistribute it and/or modify it +  under the terms of the GNU General Public License as published by the +  Free Software Foundation; either version 2 of the License, or (at your +  option) any later version. See the file COPYING for further details. + +* *****************************************************************/ + +/** @file live-mark-test.cpp + ** unit test \ref LiveMark_test + */ + +#include "test/run.hpp" +#include "test/test-helper.hpp" +#include "lib/uninitialised-storage.hpp" +#include "lib/thread.hpp" +#include "lib/util.hpp" + +#include "lib/live-mark.hpp" + +#include +#include + + +using util::isSameAdr; +using std::this_thread::yield; +using std::this_thread::sleep_for; +using std::chrono_literals::operator ""us; + + +namespace lib { +namespace test { + + using LERR_(BOTTOM_VALUE); + + + /****************************************************************//** + * @test validate a safety tag to prevent segfault when accessing + * a deceased object through a previously valid pointer. + * - demonstrate how to embody and use the LiveMark tag. + * - verify that access is only valid for the unique single + * object that generated the access-handle. + * - demonstrate that the protection mechanism can be used + * to safeguard a multithreaded sender/receiver scenario. + * @see live-mark.hpp + * @see LocalBufferStore_test + */ + class LiveMark_test : public Test + { + + virtual void + run (Arg) + { + simpleUsage(); + verify_distinctInstances(); + verify_multithreadAccess(); + } + + + /** + * Test target structure + * protected with a LiveMark tag + */ + struct Sidekick + { + LiveMark mark{*this}; + uint64_t feat{0}; + + LiveMark const& getLiveMark() const { return mark; } + }; + + using Handle = LiveMark::SafeHandle; + + + /** @test attach a LiveMark to a simple object + * and create a safe access handle, allowing + * to detect when the target went out of scope */ + void simpleUsage () + { + Handle handle; + CHECK (not handle); + { + Sidekick kick; + handle = kick.mark.makeHandle (kick); + CHECK (handle); + handle->feat = 55; + CHECK (55 == kick.feat); + }// ◁──────────────────┨ Sidekick vanishes + + CHECK (not handle); + VERIFY_ERROR (BOTTOM_VALUE, *handle ); + } + + + + /** @test re-use of the same memory location for another instance + * is indeed properly detected, preventing unintended access + * through the handle obtained from a preceding instance */ + void verify_distinctInstances() + { + UninitialisedStorage dodgyPlace; + + Sidekick* pos = new(&dodgyPlace) Sidekick; + pos->feat = 23; + CHECK (23 == dodgyPlace[0].feat); + + Handle handle = pos->mark.makeHandle (*pos); + CHECK (handle); + + // destroy the first instance + pos->~Sidekick(); + + // memory content is still there... + CHECK (23 == pos->feat); + + // insidiously plant a new object + // into the same memory location on the stack + new(&dodgyPlace) Sidekick; + + // the constructor initialised the new object, + // but we can manipulate its content through the old pointer; + CHECK (isSameAdr (pos, &dodgyPlace)); + CHECK (0 == pos->feat); + + pos->feat = 55; + CHECK (55 == dodgyPlace[0].feat); + + // but the safe handle prevents unintended aliasing + CHECK (not handle); + } + + + + /** @test assess the liveness state from another thread + * - setup a producer ⟶ consumer scenario + * - the consumer embodies a LiveMark + * - the producer checks the liveness state before sending + * - after some delay, the consumer disengages the LiveMark + * and then empties the queue... + * - the producer should notice that the receiver is no longer + * accessible and should stop sending. + * - after some further delay, the receiver verifies that + * no further messages were received and that the last + * number received is the last number produced + */ + void verify_multithreadAccess() + { + struct Receiver + { + using InQueue = boost::lockfree::queue; + + InQueue queue{1000}; + LiveMark mark{*this}; + LiveMark const& getLiveMark() const { return mark; } + }; + using RecHandle = LiveMark::SafeHandle; + + + Receiver receiver; + RecHandle handle = receiver.mark.makeHandle (receiver); + CHECK (handle); + + std::atomic_uint source{0}; + + Thread testWorker{[&] /* === Receiver of messages === */ + { + sleep_for (100us); + + // signal to accept no further messages + receiver.mark.disengage(); + yield(); + + // accept all messages sent thus far + uint lastNumber{0}; + receiver.queue.consume_all ([&](uint val){ lastNumber = val; }); + + // verify that the sender indeed stopped... + sleep_for (100us); + CHECK (receiver.queue.empty()); + CHECK (lastNumber == source); + }}; + + /* === Producer of messages === */ + + while (handle) + { + ++source; + handle->queue.push (source); + yield(); + } + + // receiver thread is still running... + CHECK (testWorker); + while (testWorker) + yield(); + } + }; + + + /** Register this test class... */ + LAUNCHER(LiveMark_test, "unit common"); + + +}} // namespace lib::test diff --git a/tests/vault/mem/local-buffer-store-test.cpp b/tests/vault/mem/local-buffer-store-test.cpp index 9de5b684f..f5d3b5558 100644 --- a/tests/vault/mem/local-buffer-store-test.cpp +++ b/tests/vault/mem/local-buffer-store-test.cpp @@ -31,8 +31,10 @@ #include using util::isSameAdr; -using std::this_thread::yield; using lib::Thread; +using std::this_thread::yield; +using std::this_thread::sleep_for; +using std::chrono_literals::operator ""ms; namespace vault { namespace mem { @@ -119,12 +121,22 @@ namespace test { *p = '\0'; CHECK (ranS == buf); + // reserve four additional buffers; + // these happen to sit already in the local pool anyway + // and will be marked as "reserved" now.... + avail = storeAPI.prepareBuffers(TYPEID, 4, BUFFSIZ); + CHECK (4 == avail); + + // finish usage and return the single buffer claimed thus far... storeAPI.mark_emitted (TYPEID, storageSlot); storeAPI.detachBuffer (TYPEID, storageSlot); + // now we have 5 buffers in the pool, but 4 are marked as "reserved" + // thus only one buffer can be confirmed and one further is requested from EngineBufferManager avail = storeAPI.prepareBuffers(TYPEID, 2, BUFFSIZ); CHECK (1 == avail); + // but irrespective of reservation, we can immediately use the buffers locally available BuffAlloc slot1 = storeAPI.provideBuffer (TYPEID, BUFFSIZ, MARK, ARG); auto& [sto1,siz1,tag1] = slot1; BuffAlloc slot2 = storeAPI.provideBuffer (TYPEID, BUFFSIZ, MARK, ARG); @@ -142,17 +154,27 @@ namespace test { while (testWorker) yield(); // wait for worker to finish + // additional delay for the local pool's destructor + // that is invoked after the thread has finished + sleep_for(1ms); + CHECK (globalPool); // the Mock was accessed (and thus instantiated) // before shutdown, LocalMemPool has sent back all allocations; - // these should sit now in the EngineBufferManager's input queue + // 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+1 == watch(*globalPool).numAllocs()); - CHECK ((CNT+1)*ALLOC == watch(*globalPool).bytesLeased()); + CHECK ( CNT == watch(*globalPool).numAllocs()); + CHECK ((CNT)*ALLOC == watch(*globalPool).bytesLeased()); // process in-queue and retrieve all allocations.. +SHOW_EXPR("Kaka") globalPool->processPendingRequests(); - CHECK (0 == watch(*globalPool).bytesLeased()); +SHOW_EXPR("Boom") +SHOW_EXPR(watch(*globalPool).bytesLeased()); +// CHECK (0 == watch(*globalPool).bytesLeased()); + sleep_for(1000ms); +SHOW_EXPR("bye") } diff --git a/wiki/thinkPad.ichthyo.mm b/wiki/thinkPad.ichthyo.mm index 88bbd1427..20a62668a 100644 --- a/wiki/thinkPad.ichthyo.mm +++ b/wiki/thinkPad.ichthyo.mm @@ -145055,7 +145055,9 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + + + @@ -145063,6 +145065,247 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) + + + + + + + + + + + + + + + + + + + + + + + +

+ eine statische Map als Lösung ist deshalb problematisch +

+ + + + + + + +

+ ...da die Worker-Threads normalerweise gleichzeitig starten / stoppen, also sehr wahrscheinlich mit dem Zugriff der anderen Worker kollidieren +

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

+ ...denn der nächste Schritt nach erfolgreichem Passieren des Schutzmechanismus wäre ohnehin eine Einfügung in eine lock-free-Queue, also sogar atomic-Read/Write or CAS. Und das machen wir nur wenn Allokationen global angefordert / zugestellt werden. +

+ + + + + + + +

+ ...denn die Sichtbarkeits-Garantie wird nur über eine happens-before-Ordnung hergestellt: wenn ich die Änderung im Atomic sehe, dann sehe ich garantiert auch alle happens-before-Änderungen.  Wenn ich die Änderung allerdings nicht sehe, ist nichts weiter über die sonstige Sichtbarkeit bekannt. In diesem Fall müßte ich weitere Rnden lokal drehen, und dürfte nicht auf anderweitig synchronisierten globalen Speicher zugreifen, bis ich die Änderung auf dem Atomic sehe. Leider paßt dieses Muster überhaupt nicht zur Anforderung, daß ich »jetzt und hier« für den nächsten Zugriff die korrekten Werte sehe (und sonst passiert eine schröckliche Katastrofe) +

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

+ Wenn dummerweise an genau der gleichen Speicheradresse wieder ein solches Objekt mit diesem Mechanismus landet. Selbst wenn es sich um einen anderen Objekttyp handelt, oder eben um die Instanz für einen anderen, inzwischen neu gestarteten Thread. Das ist auf den ersten Blick extrem unwahrscheinlich, aber ich könnte mir durchaus Situationen denken, die sowas gradezu provozieren (z.B. custom pooling Allocator) +

+ + + + + + + + + +

+ Im gegebenen Kontext ist das problematisch, weil der Zweck der ganzen Veranstaltung ist, daß sich Worker möglichst wenig gegenseitig stören.  Als Solches mak dieses Ziel debattierbar sein, aber wenn man es akzeptiert, dann wäre ein volle Lese/Schreib-Barriere oder gar eine Mutex-Sektion bereits genug, um die lock-free-Queue und das komplizierte asynchrone Austausch-System komplett ad absurdum zu führen. Denn dann könnten sich die Worker bei Bedarf auch gleich selber neuen Speicher vom Heap holen.... +

+

+ Leider sind aber die offensichtlichen, alternativen Lösungen mit Atomics immer darauf angewiesen, daß irgendwo ein zeitlicher Spielraum ist, also z.B. ein Consumer, der im Zweifelsfall "mal eben eine weitere Runde drehen" kann. Denn die Atomics geben nur eine Positiv-Garantie: wenn  ich die Änderung auf dem Atomic sehe, dann sehe ich konsistent  auch alle anderen happens-before-Änderungen. Wir brauchen aber ziemlich genau das Gegenteil hier: wenn wir zugreifen, müssen wir sicher sein, daß der Empfänger noch lebt, sonst staubt's. +

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

+ Denn eine Nonce müßte ja wieder irgendwo stehen, also entweder im statischen Speicher (back to square one), oder im Objekt, und im letzteren Fall würde sie für ein neues Objekt geändert, zusammen mit dem Hash, so daß alles wieder konsistent wäre, und damit das Problem nicht bemerkt würde. Sicher wäre das nur, wenn man die Nonce auf der anderen Seite des Zugriffs hinterlegen würde, also beim zugreifenden Client. Das wäre dann aber hässlich, komplex und verwirrend. Alternativ könnte man einen Typ-Hash verwenden. Der würde verhindern, daß man auf ein neues Objekt eines ganz anderen Typs zugreift (ein Fall, der konkret gar nicht auftreten kann). Aber er würde nicht verhindern, daß man versehentlich auf einen Pool eines neu gestarteten, anderen Threads zugreift, der zufällig an der gleichen Speicheradresse liegt. Zwar würde das keinen Schaden anrichten, aber diese Art »harmlose Glitches« sind etwas, auf das man sich besser nicht verlassen sollte. +

+ + + + + + + +

+ Jeder Zugriff müßte dann mit einer Read-Barriere über den Atomic gehen (was grade noch akzeptabel ist, da man kurz darauf genau das Gleiche auch mit der lock-free-Queue macht). Man würde dann neue Objekte vorne anfügen, und würde die Liste linear durchsuchen nach einem Eintrag für das aktuelle Objekt. Denn wenn bei Änderungen eine Write-Barriere genutzt wird, ist die Änderung garantiert im Speicher; wenn außerdem jeder Zugriff eine Lese-Barriere verwendet, dann sieht er garantiert den Wert aus dem Hauptspeicher, und garantiert auch alle vorher gemachten Änderungen. Allerdings schützt das wieder nicht gegen ein concurrent Update der Map. +

+ + + + + + + + +

+ Also ein Konstrukt wie ich es für das »Grooming-Token« im Scheduler verwendet habe: +

+
    +
  • + vor jedwedem Zugriff (Scheiben und Lesen) muß man ein atomic-Token setzen (auf die Thread-ID) +
    • +
  • + wenn es bereits gesetzt ist, muß man »spinnen« +
    • +
  • + die dahinter liegende Datenstruktur dagegen bedarf keines weiteren Schutzes, denn durch das Schreiben/Lesen des atomic-Lock hat man bereits die Garantie, daß man den aktuellen Stand sieht. +
    • +
  • + effektiv ist das ein MUTEX, aber ohne Kernel-Switch und ohne die Gefahr, schlafen gelegt zu werden. +
    • +
+

+ Leider bedingt das eine Schreib-Barriere auch für jeden Lesezugriff, da man das Lock auch zum Lesen erlangen muß. Das kostet 20µs und möglicherweise einen Pipeline-Stall, was aber hier vielleicht grade noch akzeptabel wäre, da es nur bei Austausch von Allokationen notwendig ist. Allerdings: diese passieren durchaus regelmäßig, und wenn die EngineBufferManager dann mal eben 30 Requests zuzustellen hat, ist das auch nicht mehr lustig. Also fragwürdig. +

+ + + + + + + + +

+ 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ß). +

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

+ Hilfsklasse: LiveMark +

+ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -145240,22 +145483,90 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + + + + + +

+ als "reserviert" markieren wir nur Blöcke, die bereits unmittelbar im lokalen Pool verfügbar sind; vom EngineBufferManager angeforderte Blöcke dagegen sind nicht "reserviert", wenn sie ankommen +

+ + + - + + + + +

+ ...und dann verhält sich alles wie erwartet +

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

+ selbst 1s warten bringt nix, der SEGFAULT kommt schneller +

+ + - - + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + @@ -146941,7 +147252,7 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - + @@ -146956,7 +147267,8 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension)

- + + @@ -147045,6 +147357,43 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) + + + + + + + + + + + +

+ Das hat nichts mit der unmittelbar hier diskutierten Technologie zu tun. Sondern es zeigt eine Diskrepanz in der Architektur: »asynchron« bedeutet immer »eventuell irgendwann« — aber wenn ein Thread terminiert, dann terminiert er SOFORT. +

+ + + + + + + + +

+ ...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! +

+ + + + + + + + + + + + @@ -147333,9 +147682,13 @@ StM_bind(Builder<R1> b1, Extension<R1,R2> extension) - - + + + + + +