From a420484195c14ea28ba9b910ca2566e65e18c1bb Mon Sep 17 00:00:00 2001 From: Philippe Leduc Date: Fri, 11 Sep 2026 11:17:49 +0200 Subject: [PATCH] Refactor and improve blackboard API - error management - list key and map extraction API - DRYer --- ARCHITECTURE.md | 62 ++- examples/hello_blackboard.cc | 24 +- examples/python/hello_blackboard.py | 6 +- include/kickmsg/Blackboard.h | 167 ++++--- py_bindings/src/kickmsg_py.cc | 113 +++-- python/kickmsg/__init__.py | 2 - src/Blackboard.cc | 678 ++++++++++++++++------------ tests/blackboard_crash_test.cc | 28 +- tests/python/test_blackboard.py | 31 +- tests/stress/blackboard.cc | 8 +- tests/tsan.supp | 8 +- tests/unit/blackboard-t.cc | 249 +++++++--- tests/unit/node-t.cc | 4 +- 13 files changed, 857 insertions(+), 523 deletions(-) diff --git a/ARCHITECTURE.md b/ARCHITECTURE.md index f7934c1..1215cda 100644 --- a/ARCHITECTURE.md +++ b/ARCHITECTURE.md @@ -1921,7 +1921,7 @@ resolve the new key and read the previous tenant's bytes. ``` t1 = tenancy.load(acquire); if (t1 != cached_tenancy) -> re-resolve -v1 = publish.load(acquire); k = v1 >> 1; if (k == 0) -> Unset +v1 = publish.load(acquire); k = v1 >> 1; if (k == 0) -> ENOMSG len = cell[k & 1].value_len.load(relaxed) if (len > value_capacity) { len = value_capacity } // clamp BEFORE the memcpy memcpy(out, cell[k & 1].payload, len) @@ -1935,10 +1935,38 @@ retry iff v2 - (v1 & ~1ULL) >= 2 * CELLS_PER_KEY - 1 The unsigned retry predicate is exact for both parities of `v1` and cannot overflow on a corrupt word. Retries are **bounded**: after -`READ_RETRY_BUDGET` attempts the read reports `Busy` rather than -spinning. `Busy` is transient by construction -- it means a writer +`READ_RETRY_BUDGET` attempts the read reports `EAGAIN` rather than +spinning. `EAGAIN` is transient by construction -- it means a writer outran the reader, never that anything is broken. +### Errors are standard errno + +`Writer::write`, `Writer::release`, `Blackboard::wait` and +`ReadOutcome::ec` are `std::error_code` over `std::errc`: every outcome +the blackboard can report already has a standard errno that names it. + +| errno | means | +|---|---| +| `ENOENT` | no entry for this key | +| `ENOMSG` | key declared, never written | +| `EMSGSIZE` | read buffer too small, or written value over `max_value_size` | +| `EAGAIN` | retry budget exhausted under a hot writer; transient | +| `EBADMSG` | typed read: the value is not `sizeof(T)` bytes | +| `EBADF` | default-constructed or moved-from handle | +| `EPERM` | caller is a `fork()` child; the claim is the parent's | +| `ENOTRECOVERABLE` | the entry was swept and re-tenanted -- the claim is gone | +| `EBUSY` | the board lock stayed held for the whole bounded wait | +| `ETIMEDOUT` | `wait()` returned without `change_seq` moving | + +Ignoring the result is the caller's choice: no board invariant depends +on it. + +The Python bindings map the same outcomes onto Python's own standard: +a failing `write()` or `release()` raises `OSError` with `.errno` set, +and `ReadOutcome` exposes `.errno` to compare against the stdlib `errno` +module. `wait()` stays a `bool` there -- a timeout is an ordinary answer, +not a failure. + **Values are always copied.** There is deliberately no zero-copy view: a writer may overwrite the cell mid-read, so unlike `SampleView`'s refcount-pinned slot there is nothing safe to point at. Pinning would @@ -1954,6 +1982,34 @@ suppression names one frame and leaves `write()` and `read()` themselves checked. The blackboard stress scenario asserts that no torn value ever escapes, over hundreds of thousands of reads. +### Listing the board + +`Reader` answers "what is under this key". Three calls answer "what keys +are there": `snapshot()`, `keys()` and `read_all()`. + +`snapshot()` is the diagnostic one. It reports ownership -- `owner_pid`, +`owner_node`, `owner_alive` -- and pays for it twice: it takes the board +lock, because a takeover rewrites `owner_node` in place with no seqlock +over those bytes, and it probes liveness with one OS call per active key +once the lock is dropped. + +`keys()` and `read_all()` walk the entry array **unlocked**, running the +read protocol above over each `Active` entry: `keys()` copies the key, +`read_all()` copies the key and the value. A key overtaken by its writer +mid-read is dropped rather than returned torn. `keys()` lists a +declared-but-never-written key; `read_all()` has no value to return for +one. + +The unlocked walk copies key bytes that `claim_free_slot()` may be +writing -- the race `bb_key_equals` already carries for +`Reader::resolve()`. `bb_read_key()` is the `noinline` helper that names +that one frame for `tests/tsan.supp`; the tenancy re-check after the copy +discards a torn key. + +Neither call is atomic across the board: a key claimed or written during +the walk may or may not appear. `change_seq()` tells a caller whether the +board moved under it -- read it first, list, then compare. + ### Key claim and uniqueness `declare()` first scans for an `Active` entry holding the key (takeover), diff --git a/examples/hello_blackboard.cc b/examples/hello_blackboard.cc index c487eb7..ea7d108 100644 --- a/examples/hello_blackboard.cc +++ b/examples/hello_blackboard.cc @@ -61,20 +61,6 @@ namespace return elapsed_time(nanoseconds{static_cast(updated_at_ns)}); } - char const* status_name(blackboard::Status status) - { - char const* name = "?"; - switch (status) - { - case blackboard::Ok: { name = "Ok"; break; } - case blackboard::Missing: { name = "Missing"; break; } - case blackboard::Unset: { name = "Unset"; break; } - case blackboard::Truncated: { name = "Truncated"; break; } - case blackboard::Busy: { name = "Busy"; break; } - case blackboard::SizeMismatch: { name = "SizeMismatch"; break; } - } - return name; - } } int main() @@ -110,7 +96,7 @@ int main() ArmState seen{}; auto out = state_view.read(seen); std::printf("[reader] arm/state -> %s (%s, fault %u, %.1f C) age %.3fs owner_alive=%d\n", - status_name(out.status), lifecycle_name(seen.lifecycle), + out.ec.message().c_str(), lifecycle_name(seen.lifecycle), seen.fault_code, static_cast(seen.temperature_c), age_of(out.updated_at_ns).count(), static_cast(state_view.owner_alive())); @@ -118,15 +104,15 @@ int main() uint32_t mode = 0; out = mode_view.read(mode); std::printf("[reader] arm/mode -> %s (%u) age %.3fs\n", - status_name(out.status), mode, age_of(out.updated_at_ns).count()); + out.ec.message().c_str(), mode, age_of(out.updated_at_ns).count()); std::printf("[reader] a Subscriber here would have received nothing at all.\n\n"); // --- Two states a topic cannot express -------------------------------- ArmState ignored{}; std::printf("[reader] arm/gripper -> %s (no writer ever declared it)\n", - status_name(hmi_board.observe("arm/gripper").read(ignored).status)); + hmi_board.observe("arm/gripper").read(ignored).ec.message().c_str()); std::printf("[reader] arm/calibration -> %s (declared, never written)\n\n", - status_name(hmi_board.observe("arm/calibration").read(ignored).status)); + hmi_board.observe("arm/calibration").read(ignored).ec.message().c_str()); // --- Change notification, without polling ----------------------------- // Read the sequence BEFORE acting on the current values, then wait on it: @@ -145,7 +131,7 @@ int main() for (int i = 0; i < 3; ++i) { uint64_t seq = hmi_board.change_seq(); - if (hmi_board.wait(seq, 250ms)) + if (not hmi_board.wait(seq, 250ms)) { state_view.read(seen); std::printf("[reader] woke on change: arm/state = %s fault %u\n", diff --git a/examples/python/hello_blackboard.py b/examples/python/hello_blackboard.py index fe2ebd4..2b9240e 100644 --- a/examples/python/hello_blackboard.py +++ b/examples/python/hello_blackboard.py @@ -52,15 +52,15 @@ def main() -> int: out = state_view.read() lifecycle, fault, temperature = struct.unpack(" {out.status.name} " + print(f"[reader] arm/state -> {out.error} " f"({_LIFECYCLE[lifecycle]}, fault {fault}, {temperature:.1f} C) " f"age {age:.3f}s owner_alive={state_view.owner_alive()}") print("[reader] a Subscriber here would have received nothing at all.\n") # --- Two states a topic cannot express ------------------------------ - print(f"[reader] arm/gripper -> {hmi_board.observe('arm/gripper').read().status.name}" + print(f"[reader] arm/gripper -> {hmi_board.observe('arm/gripper').read().error}" " (no writer ever declared it)") - print(f"[reader] arm/calibration -> {hmi_board.observe('arm/calibration').read().status.name}" + print(f"[reader] arm/calibration -> {hmi_board.observe('arm/calibration').read().error}" " (declared, never written)\n") # --- Change notification, without polling --------------------------- diff --git a/include/kickmsg/Blackboard.h b/include/kickmsg/Blackboard.h index d150b6d..5b65892 100644 --- a/include/kickmsg/Blackboard.h +++ b/include/kickmsg/Blackboard.h @@ -7,7 +7,10 @@ #include #include #include +#include +#include #include +#include #include #include "kickmsg/types.h" @@ -46,17 +49,8 @@ namespace kickmsg Active = 2, }; - /// A read has several distinguishable outcomes, so it reports a status - /// rather than collapsing them into a std::optional. - enum Status : uint32_t - { - Ok = 0, ///< `len` bytes were copied into the caller's buffer - Missing = 1, ///< No entry for this key - Unset = 2, ///< Key is declared but no value has ever been written - Truncated = 3, ///< Buffer too small; `len` is the true size, nothing copied - Busy = 4, ///< Retry budget exhausted under a very hot writer; transient - SizeMismatch = 5, ///< Typed read only: the value is not sizeof(T) bytes - }; + /// ENOENT: no entry for this key. ENOMSG: declared, never written. + /// EMSGSIZE: `len` is the true size and nothing was copied. /// Geometry of a blackboard region. Only the creator's values are /// stamped; an opener's are checked against the stamped config_hash. @@ -74,10 +68,10 @@ namespace kickmsg struct ReadOutcome { - Status status {Missing}; - std::size_t len {0}; - uint64_t updated_at_ns{0}; ///< monotonic_ns of the last write, 0 if never - uint64_t update_count {0}; + std::error_code ec; + std::size_t len {0}; + uint64_t updated_at_ns{0}; ///< monotonic_ns of the last write, 0 if never + uint64_t update_count {0}; }; /// Diagnostic view of one key. owner_alive costs one OS probe, so @@ -200,29 +194,23 @@ namespace kickmsg static_assert(std::is_standard_layout::value, "BlackboardHeader is placed in shared memory via reinterpret_cast"); - BlackboardEntry* bb_entry_at(void* base, BlackboardHeader const* h, uint32_t idx); - BlackboardCell* bb_cell_at(void* base, BlackboardHeader const* h, - uint32_t idx, uint64_t parity); + BlackboardEntry* bb_entry_at(void* base, uint32_t idx); + BlackboardCell* bb_cell_at(void* base, uint32_t idx, uint64_t parity); uint8_t* bb_cell_payload(BlackboardCell* cell); uint64_t bb_config_hash(blackboard::Config const& cfg); - /// Shared-memory key/value state store. - /// - /// A Subscriber attaches at its ring's current write_pos and never sees - /// anything published before it, so a node broadcasting lifecycle state - /// must heartbeat forever and a late listener still waits a full period. - /// A blackboard reader instead observes the current value of every key the - /// instant it attaches. State, not stream: writers publish once and stop. + /// Shared-memory key/value state store. State, not stream: a reader + /// observes the current value of every key the instant it attaches, so + /// writers publish once and stop. /// /// One region per board at `/{namespace}_bb_{name}`, with its own MAGIC /// and blackboard::VERSION -- independent of the channel ABI in types.h. /// Persists beyond any single process; remove with unlink(). /// - /// Mechanism, not policy. The library never interprets a value's bytes, - /// never defines a staleness threshold, and never re-declares a key on a - /// writer's behalf. It exposes updated_at_ns, the owner pid, and a change - /// counter; what counts as "too old" is the caller's. + /// Mechanism, not policy: the library never interprets a value's bytes. It + /// exposes updated_at_ns, the owner pid and a change counter; what counts + /// as "too old" is the caller's. /// /// Lifetime: Writer and Reader hold raw pointers into the mapping. They /// MUST NOT outlive the Blackboard, and the Blackboard MUST NOT be moved @@ -236,23 +224,20 @@ namespace kickmsg Blackboard(Blackboard const&) = delete; Blackboard& operator=(Blackboard const&) = delete; - // Hand-written like SharedRegion's: a defaulted move would leave the - // source aliasing the destination's live mapping. + // A defaulted move would leave the source aliasing the destination's + // live mapping. Blackboard(Blackboard&& other) noexcept; Blackboard& operator=(Blackboard&& other) noexcept; /// `owner_name` labels every key this board declares, for diagnostics. - /// Node passes its own node name, so a Node-owned board never makes - /// the caller repeat it. static Blackboard open_or_create(std::string const& kmsg_namespace, std::string const& name, blackboard::Config const& cfg = {}, char const* owner_name = ""); - /// Returns nullopt when the region does not exist -- for read-only - /// tools that must not create one as a side effect of inspection. - /// Throws on magic / version / geometry / identity mismatch. - /// The mapping is still read/write: "try" means "does not create". + /// Returns nullopt when the region does not exist. Throws on magic / + /// version / geometry / identity mismatch. The mapping is still + /// read/write: "try" means "does not create". static std::optional try_open(std::string const& kmsg_namespace, std::string const& name); @@ -278,35 +263,31 @@ namespace kickmsg Writer(Writer&& other) noexcept; Writer& operator=(Writer&& other) noexcept; - /// Publish a new value. Returns false when `len` exceeds the - /// board's max_value_size, when this writer no longer owns the - /// key (its entry was swept and re-tenanted after a false death - /// verdict), or when the caller is a fork() child of the declaring - /// process. In every case the previous value is untouched. - bool write(void const* data, std::size_t len); + /// Publish a new value. The previous value is untouched on every + /// failure. ENOTRECOVERABLE: the entry was swept and re-tenanted + /// after a false death verdict, so this claim is gone. + std::error_code write(void const* data, std::size_t len); /// Typed convenience, gated exactly like hash::fnv1a_64. template auto write(T const& value) -> std::enable_if_t and not std::is_pointer_v - and not std::is_null_pointer_v, bool> + and not std::is_null_pointer_v, + std::error_code> { return write(&value, sizeof(T)); } /// Drop ownership now rather than at destruction. Same - /// value-preserving semantics as the destructor. A no-op in a - /// fork() child, and a no-op if the board lock stays held by a - /// wedged peer for the whole (bounded) wait -- the key is then - /// reclaimed when this process exits. - void release(); + /// value-preserving semantics as the destructor, and idempotent. + /// EBUSY: the key is reclaimed when this process exits instead. + std::error_code release(); std::string const& key() const { return key_; } - /// False only for a default-constructed or moved-from Writer. A - /// live claim that later loses its entry surfaces through write() - /// returning false, not through this. + /// False only for a default-constructed or moved-from Writer; a + /// claim that later loses its entry surfaces as a write() error. bool valid() const { return base_ != nullptr; } private: @@ -336,17 +317,14 @@ namespace kickmsg Reader& operator=(Reader&&) noexcept = default; /// Copy the current value into the caller's buffer. No - /// allocation, no syscall. - /// - /// Always a copy: a writer may overwrite the cell mid-read, so - /// unlike SampleView's pinned slot there is nothing safe to point - /// at. Do not add a zero-copy view. + /// allocation, no syscall. Always a copy: a writer may overwrite + /// the cell mid-read, so there is nothing safe to point at. blackboard::ReadOutcome read(void* out, std::size_t cap) const; - /// Typed form. Reports SizeMismatch when the stored value is not - /// exactly sizeof(T) bytes -- a uint32_t read as a uint64_t would - /// otherwise return Ok over a half-filled object. `out` is - /// written only on Ok. + /// Typed form. EBADMSG when the stored value is not exactly + /// sizeof(T) bytes -- a uint32_t read as a uint64_t would otherwise + /// succeed over a half-filled object. `out` is written only on + /// success. template auto read(T& out) const -> std::enable_if_t @@ -356,13 +334,13 @@ namespace kickmsg { alignas(T) unsigned char staging[sizeof(T)]; blackboard::ReadOutcome result = read(staging, sizeof(T)); - if (result.status != blackboard::Ok) + if (result.ec) { return result; } if (result.len != sizeof(T)) { - result.status = blackboard::SizeMismatch; + result.ec = std::make_error_code(std::errc::bad_message); return result; } std::memcpy(&out, staging, sizeof(T)); @@ -370,7 +348,7 @@ namespace kickmsg } /// Owning form: resizes `out` to the value length, reusing its - /// capacity. Never returns Truncated. + /// capacity. Never reports errc::message_size. blackboard::ReadOutcome read(std::vector& out) const; /// Probe whether the key's current owner process still exists. @@ -411,8 +389,9 @@ namespace kickmsg Writer declare(char const* key, char const* owner_node = nullptr); /// Track `key` for O(1) reads. Never creates it: a Reader on a key - /// that does not exist yet reads Missing, and starts returning Ok as - /// soon as some writer declares and writes it -- no second call. + /// that does not exist yet reads errc::no_such_file_or_directory, and + /// starts succeeding as soon as some writer declares and writes it -- + /// no second call. Reader observe(char const* key); /// Monotonic count of value updates and key claims across this board. @@ -429,7 +408,9 @@ namespace kickmsg /// The timeout is mandatory and finite: futex_wait compares only the /// low 32 bits of change_seq, so an infinite wait could in principle /// miss a wakeup forever. - bool wait(uint64_t last_seen, nanoseconds timeout); + /// + /// {} once change_seq moved, errc::timed_out when it did not. + std::error_code wait(uint64_t last_seen, nanoseconds timeout); /// Diagnostic copy of every active key. Probes owner liveness, so it /// costs one OS call per active key. Safe under live traffic. @@ -439,6 +420,44 @@ namespace kickmsg /// std::runtime_error if the board lock cannot be taken. std::vector snapshot() const; + /// Names of every active key, `prefix`-filtered, in entry order. Takes + /// no board lock and probes no owner, unlike snapshot(). Includes a key + /// declared but never written. + std::vector keys(std::string_view prefix = {}) const; + + /// Every `prefix`-matching key that holds a value, with a copy of it. + /// One unlocked pass; a key overtaken by its writer mid-read is dropped + /// rather than returned torn. + /// + /// Not atomic across the board: a key claimed or written during the pass + /// may or may not appear. change_seq() tells you whether it moved. + std::unordered_map> + read_all(std::string_view prefix = {}) const; + + /// Typed form, gated exactly like Writer::write. A key whose value is + /// not sizeof(T) bytes is skipped, the listing counterpart of the + /// EBADMSG Reader::read(T&) reports for one. + template + auto read_all(std::string_view prefix = {}) const + -> std::enable_if_t + and not std::is_pointer_v + and not std::is_null_pointer_v, + std::unordered_map> + { + std::unordered_map out; + for (auto const& [key, bytes] : read_all(prefix)) + { + if (bytes.size() != sizeof(T)) + { + continue; + } + T value; + std::memcpy(&value, bytes.data(), sizeof(T)); + out.emplace(key, value); + } + return out; + } + /// Reclaim crash residue. Frees keys whose owner process is provably /// dead -- destroying their values -- and recovers entries left in a /// transient state by a process that died mid-operation. @@ -475,6 +494,20 @@ namespace kickmsg /// Sweep body, run by a caller that already holds the board lock. uint32_t sweep_locked(BlackboardHeader* h); + struct EntryRead + { + std::string key; + std::size_t value_len {0}; + uint64_t updated_at_ns{0}; + uint64_t update_count {0}; + }; + + /// Coherent read of entry `i`, the seqlock Reader::read() runs over a + /// resolved entry. Copies the value into `value` when it is non-null. + /// ENOMSG: active but never written. ENOENT: not active. + std::error_code read_entry(uint32_t i, EntryRead& out, + std::vector* value) const; + SharedMemory shm_; std::string name_; std::string owner_name_; diff --git a/py_bindings/src/kickmsg_py.cc b/py_bindings/src/kickmsg_py.cc index 69d827f..e8b0034 100644 --- a/py_bindings/src/kickmsg_py.cc +++ b/py_bindings/src/kickmsg_py.cc @@ -18,10 +18,9 @@ /// Participant — registry snapshot entry /// Registry — per-namespace participant discovery /// Node — high-level topic / broadcast / mailbox -/// BlackboardStatus — read outcome enum /// BlackboardConfig — blackboard::Config /// KeyStatus — Blackboard.snapshot() entry -/// ReadOutcome — Blackboard reader result (status + bytes) +/// ReadOutcome — Blackboard reader result (errno + bytes) /// BlackboardWriter — declared key owner: .write(bytes) / .release() /// BlackboardReader — declared read interest: .read() / .owner_alive() /// Blackboard — key/value state; late readers see current values @@ -77,6 +76,7 @@ #include #include #include +#include #include #include "kickmsg/Blackboard.h" @@ -113,12 +113,28 @@ namespace kickmsg // there is nothing safe to expose through the buffer protocol. struct PyReadOutcome { - blackboard::Status status; - nb::bytes data; - uint64_t updated_at_ns; - uint64_t update_count; + std::error_code ec; + nb::bytes data; + uint64_t updated_at_ns; + uint64_t update_count; }; + /// The blackboard reports std::error_code; Python's standard for the same + /// thing is OSError carrying an errno, so failures are raised, not returned. + /// Built by hand because nanobind's builtin_exception set has no OSError, + /// and the (errno, strerror) pair is what populates e.errno for the caller. + void raise_if(std::error_code ec, char const* what) + { + if (ec) + { + std::string const msg = std::string{what} + ": " + ec.message(); + PyObject* args = Py_BuildValue("(is)", ec.value(), msg.c_str()); + PyErr_SetObject(PyExc_OSError, args); + Py_XDECREF(args); + throw nb::python_error(); + } + } + struct PyAllocatedSlot { Publisher* publisher; @@ -901,16 +917,6 @@ namespace kickmsg // Blackboard // ------------------------------------------------------------------- - nb::enum_(m, "BlackboardStatus") - .value("Ok", blackboard::Ok) - .value("Missing", blackboard::Missing) - .value("Unset", blackboard::Unset) - .value("Truncated", blackboard::Truncated) - .value("Busy", blackboard::Busy) - // Only the C++ typed read can produce this; listed so a Python - // caller comparing against the enum sees the full set. - .value("SizeMismatch", blackboard::SizeMismatch); - nb::class_(m, "BlackboardConfig") .def(nb::init<>()) .def_rw("capacity", &blackboard::Config::capacity) @@ -939,31 +945,35 @@ namespace kickmsg }); nb::class_(m, "ReadOutcome") - .def_ro("status", &PyReadOutcome::status) + .def_prop_ro("errno", [](PyReadOutcome const& r) + { return r.ec.value(); }, + "0 on success, otherwise the standard errno -- compare against " + "the stdlib errno module (ENOENT: no such key, ENOMSG: declared " + "but never written, EAGAIN: transient, retry).") + .def_prop_ro("error", [](PyReadOutcome const& r) { return r.ec.message(); }) .def_ro("data", &PyReadOutcome::data) .def_ro("updated_at_ns", &PyReadOutcome::updated_at_ns) .def_ro("update_count", &PyReadOutcome::update_count) .def("__len__", [](PyReadOutcome const& r) { return r.data.size(); }) - .def("__bool__", [](PyReadOutcome const& r) - { - return r.status == blackboard::Ok; - }) + .def("__bool__", [](PyReadOutcome const& r) { return not r.ec; }) .def("__repr__", [](PyReadOutcome const& r) { - return std::string{"ReadOutcome(status="} - + std::to_string(static_cast(r.status)) + return std::string{"ReadOutcome(errno="} + + std::to_string(r.ec.value()) + ", len=" + std::to_string(r.data.size()) + ")"; }); nb::class_(m, "BlackboardWriter") .def("write", [](Blackboard::Writer& w, nb::bytes const& data) - { return w.write(data.c_str(), data.size()); }, + { raise_if(w.write(data.c_str(), data.size()), "write"); }, "data"_a, - "Publish a new value. Returns False if the value exceeds the " - "board's max_value_size, or if this writer no longer owns the " - "key; the previous value is left untouched either way.") - .def("release", &Blackboard::Writer::release, + "Publish a new value. Raises OSError -- EMSGSIZE if the value " + "exceeds the board's max_value_size, ENOTRECOVERABLE if this " + "writer no longer owns the key; the previous value is left " + "untouched either way.") + .def("release", + [](Blackboard::Writer& w) { raise_if(w.release(), "release"); }, "Drop ownership now instead of at destruction. The value, its " "timestamp and its update count all survive.") .def_prop_ro("key", &Blackboard::Writer::key) @@ -980,7 +990,7 @@ namespace kickmsg std::vector buf; auto out = r.read(buf); return PyReadOutcome{ - out.status, + out.ec, nb::bytes(reinterpret_cast(buf.data()), buf.size()), out.updated_at_ns, out.update_count}; }, @@ -1028,24 +1038,59 @@ namespace kickmsg .def("observe", &Blackboard::observe, "key"_a, nb::rv_policy::move, nb::keep_alive<0, 1>(), "Track `key` for O(1) reads. Never creates it: a reader on a " - "key that does not exist yet reads Missing, and starts " - "returning Ok as soon as a writer declares and writes it.") + "key that does not exist yet reads ENOENT, and starts " + "succeeding as soon as a writer declares and writes it.") .def_prop_ro("change_seq", &Blackboard::change_seq) .def("wait", [](Blackboard& b, uint64_t last_seen, nanoseconds timeout) { - bool changed = false; + std::error_code ec; { nb::gil_scoped_release release; - changed = b.wait(last_seen, timeout); + ec = b.wait(last_seen, timeout); } - return changed; + // A timeout is an ordinary answer here, not a failure. + return not ec; }, "last_seen"_a, "timeout"_a, "Block until change_seq differs from `last_seen`, or `timeout` " "(a timedelta) elapses. Releases the GIL while blocked. Pass " "the change_seq you last acted on -- that is what closes the " "lost-wakeup window.") + .def("keys", + [](Blackboard const& b, std::string const& prefix) + { + std::vector out; + { + nb::gil_scoped_release release; + out = b.keys(prefix); + } + return out; + }, + "prefix"_a = std::string{}, + "Names of every active key, prefix-filtered. Takes no board " + "lock and probes no owner, unlike snapshot().") + .def("read_all", + [](Blackboard const& b, std::string const& prefix) + { + std::unordered_map> raw; + { + nb::gil_scoped_release release; + raw = b.read_all(prefix); + } + std::unordered_map out; + out.reserve(raw.size()); + for (auto const& [key, value] : raw) + { + out.emplace(key, + nb::bytes(reinterpret_cast(value.data()), value.size())); + } + return out; + }, + "prefix"_a = std::string{}, + "Every prefix-matching key that holds a value, as a dict of " + "key -> bytes. A key overtaken by its writer mid-read is " + "dropped rather than returned torn.") .def("snapshot", [](Blackboard const& b) { diff --git a/python/kickmsg/__init__.py b/python/kickmsg/__init__.py index 571d5f7..0fe5d3b 100644 --- a/python/kickmsg/__init__.py +++ b/python/kickmsg/__init__.py @@ -9,7 +9,6 @@ Blackboard, BlackboardConfig, BlackboardReader, - BlackboardStatus, BlackboardWriter, BroadcastHandle, ChannelType, @@ -45,7 +44,6 @@ "Blackboard", "BlackboardConfig", "BlackboardReader", - "BlackboardStatus", "BlackboardWriter", "BroadcastHandle", "ChannelType", diff --git a/src/Blackboard.cc b/src/Blackboard.cc index 68d1566..d366168 100644 --- a/src/Blackboard.cc +++ b/src/Blackboard.cc @@ -36,9 +36,9 @@ namespace kickmsg /// Copy a value payload. A reader overtaken by CELLS_PER_KEY writes /// races the writer's copy here and discards the result; the payload /// race lives in this function alone, and tests/tsan.supp names it. - KICKMSG_BB_NOINLINE void bb_copy_payload(void* dst, void const* src, std::size_t n) + KICKMSG_BB_NOINLINE void bb_copy_payload(void* dst, void const* src, std::size_t len) { - std::memcpy(dst, src, n); + std::memcpy(dst, src, len); } /// Compare a stored key against \p key. Reader::resolve() runs this @@ -51,6 +51,14 @@ namespace kickmsg and std::memcmp(stored, key, key_len) == 0; } + /// Runs unlocked against claim_free_slot()'s copy_field, the race + /// bb_key_equals also carries; the caller's tenancy re-check discards a + /// torn copy. + KICKMSG_BB_NOINLINE std::string bb_read_key(char const* stored, std::size_t size) + { + return std::string(stored, ::strnlen(stored, size)); + } + std::size_t stride_for(std::size_t max_value_size) { return align_up(sizeof(BlackboardCell) + max_value_size, CACHE_LINE); @@ -67,14 +75,14 @@ namespace kickmsg /// The value limit is the creator's configured size, never the padded /// stride: handing out the alignment slack would let one peer write /// more than its correctly-sized readers can hold. - std::size_t value_capacity(BlackboardHeader const* h) + std::size_t value_capacity(BlackboardHeader const* header) { - return static_cast(h->max_value_size); + return static_cast(header->max_value_size); } - std::size_t value_stride(BlackboardHeader const* h) + std::size_t value_stride(BlackboardHeader const* header) { - return stride_for(static_cast(h->max_value_size)); + return stride_for(static_cast(header->max_value_size)); } void copy_field(char* dst, std::size_t dst_size, char const* src) @@ -84,20 +92,15 @@ namespace kickmsg { return; } - std::size_t n = ::strnlen(src, dst_size - 1); - std::memcpy(dst, src, n); - } - - std::string read_field(char const* src, std::size_t size) - { - return std::string(src, ::strnlen(src, size)); + std::size_t len = ::strnlen(src, dst_size - 1); + std::memcpy(dst, src, len); } - void clear_identity(BlackboardEntry* e) + void clear_identity(BlackboardEntry* entry) { - e->key_hash.store(0, std::memory_order_relaxed); - e->owner_pid.store(0, std::memory_order_relaxed); - e->owner_starttime.store(0, std::memory_order_relaxed); + entry->key_hash.store(0, std::memory_order_relaxed); + entry->owner_pid.store(0, std::memory_order_relaxed); + entry->owner_starttime.store(0, std::memory_order_relaxed); } // Board mutex token: (start-time fingerprint : 32 | pid : 32). @@ -153,7 +156,7 @@ namespace kickmsg return make_token(pid, live.starttime) != token; } - void repair_board(void* base, BlackboardHeader* h); + void repair_board(void* base, BlackboardHeader* header); /// Returns false if the wait ran out, which means a live holder. /// `budget` bounds yields and `limit` bounds wall clock; zero disables @@ -161,7 +164,7 @@ namespace kickmsg /// /// An abandoned lock transfers directly from the dead holder's token /// to ours, never through zero, so nothing can slip in mid-repair. - bool board_lock(void* base, BlackboardHeader* h, uint64_t my_token, + bool board_lock(void* base, BlackboardHeader* header, uint64_t my_token, int budget, nanoseconds limit) { nanoseconds start = monotonic_ns(); @@ -175,11 +178,11 @@ namespace kickmsg return false; } } - uint64_t held = h->lock_token.load(std::memory_order_acquire); + uint64_t held = header->lock_token.load(std::memory_order_acquire); if (held == 0) { uint64_t expected = 0; - if (h->lock_token.compare_exchange_strong( + if (header->lock_token.compare_exchange_strong( expected, my_token, std::memory_order_acq_rel, std::memory_order_relaxed)) { @@ -195,20 +198,20 @@ namespace kickmsg continue; } uint64_t expected = held; - if (h->lock_token.compare_exchange_strong( + if (header->lock_token.compare_exchange_strong( expected, my_token, std::memory_order_acq_rel, std::memory_order_relaxed)) { - repair_board(base, h); + repair_board(base, header); return true; } } return false; } - void board_unlock(BlackboardHeader* h) + void board_unlock(BlackboardHeader* header) { - h->lock_token.store(0, std::memory_order_release); + header->lock_token.store(0, std::memory_order_release); } /// The only way to hold the board lock: [[nodiscard]] so acquisition @@ -223,15 +226,15 @@ namespace kickmsg /// The token is always the caller's own, so the guard derives it /// rather than taking one -- a parameter that can only be passed /// one way is a hazard, not a knob. - BoardGuard(void* base, BlackboardHeader* h, int budget) - : h_{h} - , held_{board_lock(base, h, self_token(), budget, nanoseconds::zero())} + BoardGuard(void* base, BlackboardHeader* header, int budget) + : h_{header} + , held_{board_lock(base, header, self_token(), budget, nanoseconds::zero())} { } - BoardGuard(void* base, BlackboardHeader* h, nanoseconds limit) - : h_{h} - , held_{board_lock(base, h, self_token(), 0, limit)} + BoardGuard(void* base, BlackboardHeader* header, nanoseconds limit) + : h_{header} + , held_{board_lock(base, header, self_token(), 0, limit)} { } @@ -264,52 +267,52 @@ namespace kickmsg /// Lock recovery and sweep_stale() must apply exactly the same rules -- /// a rule added to one and not the other silently diverges them -- so /// they share this rather than each carrying a copy. - bool normalize_entry(BlackboardEntry* e) + bool normalize_entry(BlackboardEntry* entry) { - uint32_t st = e->state.load(std::memory_order_acquire); + uint32_t state = entry->state.load(std::memory_order_acquire); - if (st == blackboard::Claiming) + if (state == blackboard::Claiming) { // A claim that never committed never handed out a Writer. - clear_identity(e); - e->state.store(blackboard::Free, std::memory_order_release); + clear_identity(entry); + entry->state.store(blackboard::Free, std::memory_order_release); return true; } - if (st == blackboard::Active - and e->key_hash.load(std::memory_order_relaxed) == 0) + if (state == blackboard::Active + and entry->key_hash.load(std::memory_order_relaxed) == 0) { // Half-finished publish_free(). Active with no key is a // phantom: matches no reader, has no owner to sweep. - clear_identity(e); - e->state.store(blackboard::Free, std::memory_order_release); + clear_identity(entry); + entry->state.store(blackboard::Free, std::memory_order_release); return true; } - if (st == blackboard::Free - and (e->key_hash.load(std::memory_order_relaxed) != 0 - or e->owner_pid.load(std::memory_order_relaxed) != 0)) + if (state == blackboard::Free + and (entry->key_hash.load(std::memory_order_relaxed) != 0 + or entry->owner_pid.load(std::memory_order_relaxed) != 0)) { // The mirror case: Free published before identity cleared. - clear_identity(e); + clear_identity(entry); return true; } return false; } - void repair_board(void* base, BlackboardHeader* h) + void repair_board(void* base, BlackboardHeader* header) { - for (uint32_t i = 0; i < h->capacity; ++i) + for (uint32_t i = 0; i < header->capacity; ++i) { - normalize_entry(bb_entry_at(base, h, i)); + normalize_entry(bb_entry_at(base, i)); } } /// Return a held entry to Free. Identity is cleared before the state /// is published so a Free entry never carries an owner for the next /// claimant to inherit. - void publish_free(BlackboardEntry* e) + void publish_free(BlackboardEntry* entry) { - clear_identity(e); - e->state.store(blackboard::Free, std::memory_order_release); + clear_identity(entry); + entry->state.store(blackboard::Free, std::memory_order_release); } /// An entry is takeable when nobody owns it or its owner is provably @@ -320,9 +323,9 @@ namespace kickmsg return pid == 0 or owner_is_dead(pid, starttime); } - bool key_matches(BlackboardEntry const* e, char const* key, std::size_t key_len) + bool key_matches(BlackboardEntry const* entry, char const* key, std::size_t key_len) { - return bb_key_equals(e->key, key, key_len); + return bb_key_equals(entry->key, key, key_len); } /// Fingerprint of the RAW (namespace, name) pair, each component @@ -332,61 +335,60 @@ namespace kickmsg uint64_t derived_identity(std::string const& kmsg_namespace, std::string const& name) { - uint64_t h = hash::fnv1a_64(std::string_view("blackboard"), + uint64_t header = hash::fnv1a_64(std::string_view("blackboard"), hash::FNV1A_64_OFFSET_BASIS); - h = hash::fnv1a_64(std::size_t{10}, h); - h = hash::fnv1a_64(std::string_view(kmsg_namespace), h); - h = hash::fnv1a_64(kmsg_namespace.size(), h); - h = hash::fnv1a_64(std::string_view(name), h); - h = hash::fnv1a_64(name.size(), h); - if (h == 0) + header = hash::fnv1a_64(std::size_t{10}, header); + header = hash::fnv1a_64(std::string_view(kmsg_namespace), header); + header = hash::fnv1a_64(kmsg_namespace.size(), header); + header = hash::fnv1a_64(std::string_view(name), header); + header = hash::fnv1a_64(name.size(), header); + if (header == 0) { - h = 1; + header = 1; } - return h; + return header; } /// key_hash == 0 is the "no key published yet" sentinel, so a real key /// must never hash to it. uint64_t key_fingerprint(char const* key, std::size_t key_len) { - uint64_t kh = hash::fnv1a_64(std::string_view(key, key_len)); - if (kh == 0) + uint64_t key_hash = hash::fnv1a_64(std::string_view(key, key_len)); + if (key_hash == 0) { - kh = 1; + key_hash = 1; } - return kh; + return key_hash; } - void notify_change(BlackboardHeader* h) + void notify_change(BlackboardHeader* header) { - h->change_seq.fetch_add(1, std::memory_order_release); + header->change_seq.fetch_add(1, std::memory_order_release); // Orders the change_seq bump before the waiters load: without it a // weakly-ordered CPU reads waiters == 0 stale and a parked reader // sleeps to its timeout. Pairs with wait()'s fence. std::atomic_thread_fence(std::memory_order_seq_cst); - if (h->waiters.load(std::memory_order_relaxed) != 0) + if (header->waiters.load(std::memory_order_relaxed) != 0) { - futex_wake_all(h->change_seq); + futex_wake_all(header->change_seq); } } } - BlackboardEntry* bb_entry_at(void* base, BlackboardHeader const* h, uint32_t idx) + BlackboardEntry* bb_entry_at(void* base, uint32_t idx) { - (void)h; auto* bytes = static_cast(base) + sizeof(BlackboardHeader); return reinterpret_cast(bytes + static_cast(idx) * sizeof(BlackboardEntry)); } - BlackboardCell* bb_cell_at(void* base, BlackboardHeader const* h, - uint32_t idx, uint64_t parity) + BlackboardCell* bb_cell_at(void* base, uint32_t idx, uint64_t parity) { + auto const* header = static_cast(base); std::size_t values = sizeof(BlackboardHeader) - + static_cast(h->capacity) * sizeof(BlackboardEntry); + + static_cast(header->capacity) * sizeof(BlackboardEntry); std::size_t cell = (static_cast(idx) * blackboard::CELLS_PER_KEY + static_cast(parity & CELL_MASK)) - * value_stride(h); + * value_stride(header); return reinterpret_cast(static_cast(base) + values + cell); } @@ -397,9 +399,9 @@ namespace kickmsg uint64_t bb_config_hash(blackboard::Config const& cfg) { - uint64_t h = hash::fnv1a_64(cfg.capacity); - h = hash::fnv1a_64(static_cast(cfg.max_value_size), h); - return h; + uint64_t header = hash::fnv1a_64(cfg.capacity); + header = hash::fnv1a_64(static_cast(cfg.max_value_size), header); + return header; } // ---- construction ---------------------------------------------------- @@ -443,18 +445,17 @@ namespace kickmsg std::size_t bytes = region_size(cfg.capacity, cfg.max_value_size); std::memset(base_, 0, bytes); - auto* h = header(); - h->version = blackboard::VERSION; - h->capacity = cfg.capacity; - h->max_value_size = cfg.max_value_size; - h->total_size = bytes; - h->config_hash = bb_config_hash(cfg); - h->identity_hash = cfg.identity; - h->creator_pid = current_pid(); - h->created_at_ns = static_cast(since_epoch().count()); + header()->version = blackboard::VERSION; + header()->capacity = cfg.capacity; + header()->max_value_size = cfg.max_value_size; + header()->total_size = bytes; + header()->config_hash = bb_config_hash(cfg); + header()->identity_hash = cfg.identity; + header()->creator_pid = current_pid(); + header()->created_at_ns = static_cast(since_epoch().count()); // MAGIC published last -- openers spin on it with acquire. - h->magic.store(blackboard::MAGIC, std::memory_order_release); + header()->magic.store(blackboard::MAGIC, std::memory_order_release); } std::optional Blackboard::spin_open(std::string const& shm, @@ -470,15 +471,15 @@ namespace kickmsg { throw std::runtime_error("Blackboard segment too small: " + shm); } - auto const* h = static_cast(mapping.address()); - if (h->magic.load(std::memory_order_acquire) == blackboard::MAGIC) + auto const* header = static_cast(mapping.address()); + if (header->magic.load(std::memory_order_acquire) == blackboard::MAGIC) { - if (h->version != blackboard::VERSION) + if (header->version != blackboard::VERSION) { throw std::runtime_error("Blackboard version mismatch on " + shm); } - if (h->total_size < sizeof(BlackboardHeader) - or h->total_size > mapping.size()) + if (header->total_size < sizeof(BlackboardHeader) + or header->total_size > mapping.size()) { throw std::runtime_error("Blackboard total_size invalid on " + shm); } @@ -486,35 +487,35 @@ namespace kickmsg // pointer computation; a corrupt value would send // snapshot()/read() off the mapping. Bound both // (division-based, so no intermediate can overflow). - std::size_t after_header = static_cast(h->total_size) + std::size_t after_header = static_cast(header->total_size) - sizeof(BlackboardHeader); - if (h->capacity == 0 or h->capacity > blackboard::MAX_CAPACITY - or h->capacity > after_header / sizeof(BlackboardEntry)) + if (header->capacity == 0 or header->capacity > blackboard::MAX_CAPACITY + or header->capacity > after_header / sizeof(BlackboardEntry)) { throw std::runtime_error("Blackboard capacity exceeds segment on " + shm); } // Bounding max_value_size (not the derived stride) is what // keeps the accepted range identical on both sides: a board // created at exactly MAX_VALUE_SIZE must stay openable. - if (h->max_value_size == 0 - or h->max_value_size > blackboard::MAX_VALUE_SIZE) + if (header->max_value_size == 0 + or header->max_value_size > blackboard::MAX_VALUE_SIZE) { throw std::runtime_error("Blackboard max_value_size invalid on " + shm); } - std::size_t entries_bytes = static_cast(h->capacity) + std::size_t entries_bytes = static_cast(header->capacity) * sizeof(BlackboardEntry); std::size_t after_entries = after_header - entries_bytes; - if (h->capacity > after_entries - / (blackboard::CELLS_PER_KEY * value_stride(h))) + if (header->capacity > after_entries + / (blackboard::CELLS_PER_KEY * value_stride(header))) { throw std::runtime_error("Blackboard value area exceeds segment on " + shm); } - if (check_config and h->config_hash != bb_config_hash(cfg)) + if (check_config and header->config_hash != bb_config_hash(cfg)) { throw std::runtime_error("Blackboard config mismatch on " + shm); } - if (cfg.identity != 0 and h->identity_hash != 0 - and h->identity_hash != cfg.identity) + if (cfg.identity != 0 and header->identity_hash != 0 + and header->identity_hash != cfg.identity) { throw std::runtime_error("Blackboard identity mismatch on " + shm); } @@ -645,9 +646,8 @@ namespace kickmsg throw std::invalid_argument("Blackboard key exceeds KEY_MAX"); } - auto* h = header(); - uint32_t cap = h->capacity; - uint64_t kh = key_fingerprint(key, key_len); + uint32_t capacity = header()->capacity; + uint64_t key_hash = key_fingerprint(key, key_len); SelfIdentity self = self_identity(); uint64_t my_pid = self.pid; @@ -657,7 +657,7 @@ namespace kickmsg // key" spans the whole board, so scanning for the key and claiming a // slot must be one indivisible step -- otherwise two claimants can each // scan, each see nothing, and each commit. - BoardGuard guard{base_, h, 4096}; + BoardGuard guard{base_, header(), 4096}; if (not guard) { throw std::runtime_error("Blackboard is busy: could not take the board lock"); @@ -666,37 +666,37 @@ namespace kickmsg // Pass 1: an entry already holds this key. Taking it over preserves // the value and the publish counter, so a restarted writer causes no // blackout for readers. - for (uint32_t i = 0; i < cap; ++i) + for (uint32_t i = 0; i < capacity; ++i) { - auto* e = bb_entry_at(base_, h, i); - if (e->state.load(std::memory_order_acquire) != blackboard::Active) + auto* entry = bb_entry_at(base_, i); + if (entry->state.load(std::memory_order_acquire) != blackboard::Active) { continue; } - if (e->key_hash.load(std::memory_order_relaxed) != kh) + if (entry->key_hash.load(std::memory_order_relaxed) != key_hash) { continue; } - if (not key_matches(e, key, key_len)) + if (not key_matches(entry, key, key_len)) { continue; } - uint64_t pid = e->owner_pid.load(std::memory_order_relaxed); - uint64_t start = e->owner_starttime.load(std::memory_order_relaxed); + uint64_t pid = entry->owner_pid.load(std::memory_order_relaxed); + uint64_t start = entry->owner_starttime.load(std::memory_order_relaxed); if (not entry_takeable(pid, start)) { throw std::runtime_error( std::string("Blackboard key already owned by a live process: ") + key); } - copy_field(e->owner_node, sizeof(e->owner_node), owner_node); - e->owner_starttime.store(my_start, std::memory_order_relaxed); - e->owner_pid.store(my_pid, std::memory_order_release); - uint64_t tenancy = e->tenancy.fetch_add(1, std::memory_order_release) + 1; - uint64_t writes = e->publish.load(std::memory_order_acquire) >> 1; + copy_field(entry->owner_node, sizeof(entry->owner_node), owner_node); + entry->owner_starttime.store(my_start, std::memory_order_relaxed); + entry->owner_pid.store(my_pid, std::memory_order_release); + uint64_t tenancy = entry->tenancy.fetch_add(1, std::memory_order_release) + 1; + uint64_t writes = entry->publish.load(std::memory_order_acquire) >> 1; - notify_change(h); + notify_change(header()); return Writer(base_, i, tenancy, writes, my_pid, std::string(key, key_len)); } @@ -705,46 +705,46 @@ namespace kickmsg // key and read the previous tenant's bytes. auto claim_free_slot = [&]() -> uint32_t { - for (uint32_t i = 0; i < cap; ++i) + for (uint32_t i = 0; i < capacity; ++i) { - auto* e = bb_entry_at(base_, h, i); - if (e->state.load(std::memory_order_acquire) != blackboard::Free) + auto* entry = bb_entry_at(base_, i); + if (entry->state.load(std::memory_order_acquire) != blackboard::Free) { continue; } // Claiming makes a death here recoverable: the next lock // holder returns the entry to Free. - e->state.store(blackboard::Claiming, std::memory_order_release); - e->publish.store(0, std::memory_order_relaxed); - copy_field(e->key, sizeof(e->key), key); - copy_field(e->owner_node, sizeof(e->owner_node), owner_node); - e->declared_at_ns.store( + entry->state.store(blackboard::Claiming, std::memory_order_release); + entry->publish.store(0, std::memory_order_relaxed); + copy_field(entry->key, sizeof(entry->key), key); + copy_field(entry->owner_node, sizeof(entry->owner_node), owner_node); + entry->declared_at_ns.store( static_cast(monotonic_ns().count()), std::memory_order_relaxed); - e->owner_starttime.store(my_start, std::memory_order_relaxed); - e->owner_pid.store(my_pid, std::memory_order_relaxed); - e->key_hash.store(kh, std::memory_order_relaxed); + entry->owner_starttime.store(my_start, std::memory_order_relaxed); + entry->owner_pid.store(my_pid, std::memory_order_relaxed); + entry->key_hash.store(key_hash, std::memory_order_relaxed); // relaxed above, release here: the store of Active is the one // fence that publishes every field to a lock-free reader. - e->tenancy.fetch_add(1, std::memory_order_release); - e->state.store(blackboard::Active, std::memory_order_release); + entry->tenancy.fetch_add(1, std::memory_order_release); + entry->state.store(blackboard::Active, std::memory_order_release); return i; } return INVALID_SLOT; }; uint32_t claimed = claim_free_slot(); - if (claimed == INVALID_SLOT and sweep_locked(h) > 0) + if (claimed == INVALID_SLOT and sweep_locked(header()) > 0) { // Crash residue can be sitting on the last free slots. claimed = claim_free_slot(); } if (claimed != INVALID_SLOT) { - auto* e = bb_entry_at(base_, h, claimed); - uint64_t tenancy = e->tenancy.load(std::memory_order_acquire); - notify_change(h); + auto* entry = bb_entry_at(base_, claimed); + uint64_t tenancy = entry->tenancy.load(std::memory_order_acquire); + notify_change(header()); return Writer(base_, claimed, tenancy, 0, my_pid, std::string(key, key_len)); } @@ -813,39 +813,43 @@ namespace kickmsg return *this; } - bool Blackboard::Writer::write(void const* data, std::size_t len) + std::error_code Blackboard::Writer::write(void const* data, std::size_t len) { if (base_ == nullptr) { - return false; + return std::make_error_code(std::errc::bad_file_descriptor); } // A Writer inherited across fork() is not a claim: the entry belongs to // the process that declared it, and tenancy alone cannot tell them apart. if (self_identity().pid != owner_pid_) { - return false; + return std::make_error_code(std::errc::operation_not_permitted); } - auto* h = static_cast(base_); - if (len > value_capacity(h) or entry_idx_ >= h->capacity) + auto* header = static_cast(base_); + if (entry_idx_ >= header->capacity) { - return false; + return std::make_error_code(std::errc::bad_file_descriptor); + } + if (len > value_capacity(header)) + { + return std::make_error_code(std::errc::message_size); } - auto* e = bb_entry_at(base_, h, entry_idx_); - if (e->state.load(std::memory_order_acquire) != blackboard::Active) + auto* entry = bb_entry_at(base_, entry_idx_); + if (entry->state.load(std::memory_order_acquire) != blackboard::Active) { - return false; + return std::make_error_code(std::errc::state_not_recoverable); } // relaxed: the acquire load of state above orders this read. - if (e->tenancy.load(std::memory_order_relaxed) != tenancy_) + if (entry->tenancy.load(std::memory_order_relaxed) != tenancy_) { - return false; + return std::make_error_code(std::errc::state_not_recoverable); } - uint64_t k = writes_ + 1; - auto* cell = bb_cell_at(base_, h, entry_idx_, k); + uint64_t writes = writes_ + 1; + auto* cell = bb_cell_at(base_, entry_idx_, writes); - e->publish.store(2 * k - 1, std::memory_order_relaxed); + entry->publish.store(2 * writes - 1, std::memory_order_relaxed); // Keeps the write-in-progress store above the payload stores. A // release RMW would NOT do this: it orders prior operations only. std::atomic_thread_fence(std::memory_order_release); @@ -861,18 +865,18 @@ namespace kickmsg static_cast(monotonic_ns().count()), std::memory_order_relaxed); std::atomic_thread_fence(std::memory_order_release); - e->publish.store(2 * k, std::memory_order_relaxed); - writes_ = k; + entry->publish.store(2 * writes, std::memory_order_relaxed); + writes_ = writes; - notify_change(h); - return true; + notify_change(header); + return {}; } - void Blackboard::Writer::release() + std::error_code Blackboard::Writer::release() { if (base_ == nullptr) { - return; + return {}; } // ~Writer runs on every exit path in a forked child too; releasing // there would hand the parent's key away behind its back. @@ -880,32 +884,38 @@ namespace kickmsg { base_ = nullptr; entry_idx_ = INVALID_SLOT; - return; + return std::make_error_code(std::errc::operation_not_permitted); } - auto* h = static_cast(base_); - if (entry_idx_ < h->capacity) + + std::error_code ec = std::make_error_code(std::errc::bad_file_descriptor); + auto* header = static_cast(base_); + if (entry_idx_ < header->capacity) { - BoardGuard guard{base_, h, RELEASE_LOCK_WAIT}; + ec = std::make_error_code(std::errc::device_or_resource_busy); + BoardGuard guard{base_, header, RELEASE_LOCK_WAIT}; if (guard) { - auto* e = bb_entry_at(base_, h, entry_idx_); - if (e->state.load(std::memory_order_acquire) == blackboard::Active - and e->tenancy.load(std::memory_order_relaxed) == tenancy_) + ec = std::make_error_code(std::errc::state_not_recoverable); + auto* entry = bb_entry_at(base_, entry_idx_); + if (entry->state.load(std::memory_order_acquire) == blackboard::Active + and entry->tenancy.load(std::memory_order_relaxed) == tenancy_) { // The entry stays Active holding its value: readers keep // seeing the last state, and a later declare() takes it over. // pid is cleared before start time so a racing liveness // probe never sees (live pid, zeroed start) and concludes // the owner is gone. - e->owner_pid.store(0, std::memory_order_relaxed); - e->owner_starttime.store(0, std::memory_order_relaxed); - e->tenancy.fetch_add(1, std::memory_order_release); - notify_change(h); + entry->owner_pid.store(0, std::memory_order_relaxed); + entry->owner_starttime.store(0, std::memory_order_relaxed); + entry->tenancy.fetch_add(1, std::memory_order_release); + notify_change(header); + ec = {}; } } } base_ = nullptr; entry_idx_ = INVALID_SLOT; + return ec; } // ---- Reader ---------------------------------------------------------- @@ -923,35 +933,35 @@ namespace kickmsg { return false; } - auto const* h = static_cast(base_); + auto const* header = static_cast(base_); - if (entry_idx_ < h->capacity) + if (entry_idx_ < header->capacity) { - auto* e = bb_entry_at(base_, h, entry_idx_); - if (e->state.load(std::memory_order_acquire) == blackboard::Active - and e->tenancy.load(std::memory_order_acquire) == tenancy_) + auto* entry = bb_entry_at(base_, entry_idx_); + if (entry->state.load(std::memory_order_acquire) == blackboard::Active + and entry->tenancy.load(std::memory_order_acquire) == tenancy_) { return true; } } - for (uint32_t i = 0; i < h->capacity; ++i) + for (uint32_t i = 0; i < header->capacity; ++i) { - auto* e = bb_entry_at(base_, h, i); - if (e->state.load(std::memory_order_acquire) != blackboard::Active) + auto* entry = bb_entry_at(base_, i); + if (entry->state.load(std::memory_order_acquire) != blackboard::Active) { continue; } // relaxed: the acquire load of state above orders it; a mismatch // just costs a skipped candidate. - if (e->key_hash.load(std::memory_order_relaxed) != key_hash_) + if (entry->key_hash.load(std::memory_order_relaxed) != key_hash_) { continue; } - uint64_t tenancy = e->tenancy.load(std::memory_order_acquire); + uint64_t tenancy = entry->tenancy.load(std::memory_order_acquire); // key_hash is only a pre-filter: it cannot survive a collision, // so the bytes must actually match. - if (not bb_key_equals(e->key, key_.data(), key_.size())) + if (not bb_key_equals(entry->key, key_.data(), key_.size())) { continue; } @@ -964,39 +974,39 @@ namespace kickmsg return false; } - blackboard::ReadOutcome Blackboard::Reader::read(void* out, std::size_t cap) const + blackboard::ReadOutcome Blackboard::Reader::read(void* out, std::size_t capacity) const { blackboard::ReadOutcome result; + result.ec = std::make_error_code(std::errc::no_such_file_or_directory); if (base_ == nullptr) { return result; } - auto* h = static_cast(base_); - std::size_t limit = value_capacity(h); + auto* header = static_cast(base_); + std::size_t limit = value_capacity(header); for (int retry = 0; retry < blackboard::READ_RETRY_BUDGET; ++retry) { if (not resolve()) { - result.status = blackboard::Missing; return result; } - auto* e = bb_entry_at(base_, h, entry_idx_); + auto* entry = bb_entry_at(base_, entry_idx_); - uint64_t t1 = e->tenancy.load(std::memory_order_acquire); - if (t1 != tenancy_) + uint64_t tenancy_before = entry->tenancy.load(std::memory_order_acquire); + if (tenancy_before != tenancy_) { continue; } - uint64_t v1 = e->publish.load(std::memory_order_acquire); - uint64_t k = v1 >> 1; - if (k == 0) + uint64_t publish_before = entry->publish.load(std::memory_order_acquire); + uint64_t writes = publish_before >> 1; + if (writes == 0) { - result.status = blackboard::Unset; + result.ec = std::make_error_code(std::errc::no_message); return result; } - auto* cell = bb_cell_at(base_, h, entry_idx_, k); + auto* cell = bb_cell_at(base_, entry_idx_, writes); // relaxed: ordered by the acquire load of publish above and // validated by the re-check below. std::size_t len = cell->value_len.load(std::memory_order_relaxed); @@ -1005,7 +1015,7 @@ namespace kickmsg { len = limit; } - bool fits = len <= cap; + bool fits = len <= capacity; if (fits and len > 0) { bb_copy_payload(out, bb_cell_payload(cell), len); @@ -1016,33 +1026,33 @@ namespace kickmsg // without this the relaxed cell reads could be satisfied after the // re-checks below (cf. read_seqretry's smp_rmb). std::atomic_thread_fence(std::memory_order_acquire); - uint64_t v2 = e->publish.load(std::memory_order_acquire); - uint64_t t2 = e->tenancy.load(std::memory_order_acquire); + uint64_t publish_after = entry->publish.load(std::memory_order_acquire); + uint64_t tenancy_after = entry->tenancy.load(std::memory_order_acquire); - if (t2 != t1) + if (tenancy_after != tenancy_before) { continue; } - // Cell k is clobbered once write k + CELLS_PER_KEY starts. The - // unsigned form is exact for both parities of v1 and cannot + // The cell is clobbered once write `writes + CELLS_PER_KEY` starts. The + // unsigned form is exact for both parities of publish_before and cannot // overflow on a corrupt publish word. - if (v2 - (v1 & ~1ULL) >= 2 * blackboard::CELLS_PER_KEY - 1) + if (publish_after - (publish_before & ~1ULL) >= 2 * blackboard::CELLS_PER_KEY - 1) { continue; } result.len = len; result.updated_at_ns = stamp; - result.update_count = k; - result.status = blackboard::Ok; + result.update_count = writes; + result.ec = {}; if (not fits) { - result.status = blackboard::Truncated; + result.ec = std::make_error_code(std::errc::message_size); } return result; } - result.status = blackboard::Busy; + result.ec = std::make_error_code(std::errc::resource_unavailable_try_again); return result; } @@ -1051,19 +1061,19 @@ namespace kickmsg for (int attempt = 0; attempt < 4; ++attempt) { auto result = read(out.data(), out.size()); - if (result.status == blackboard::Truncated) + if (result.ec == std::errc::message_size) { out.resize(result.len); continue; } - if (result.status == blackboard::Ok) + if (not result.ec) { out.resize(result.len); } return result; } blackboard::ReadOutcome result; - result.status = blackboard::Busy; + result.ec = std::make_error_code(std::errc::resource_unavailable_try_again); return result; } @@ -1073,179 +1083,254 @@ namespace kickmsg { return false; } - auto const* h = static_cast(base_); - auto* e = bb_entry_at(base_, h, entry_idx_); - uint64_t pid = e->owner_pid.load(std::memory_order_acquire); + auto* entry = bb_entry_at(base_, entry_idx_); + uint64_t pid = entry->owner_pid.load(std::memory_order_acquire); if (pid == 0) { return false; } - return not owner_is_dead(pid, e->owner_starttime.load(std::memory_order_relaxed)); + return not owner_is_dead(pid, entry->owner_starttime.load(std::memory_order_relaxed)); } // ---- wait / snapshot / sweep ----------------------------------------- - bool Blackboard::wait(uint64_t last_seen, nanoseconds timeout) + std::error_code Blackboard::wait(uint64_t last_seen, nanoseconds timeout) { require_open(base_); - auto* h = header(); - nanoseconds start = monotonic_ns(); + nanoseconds start = monotonic_ns(); - for (;;) + while (true) { - if (h->change_seq.load(std::memory_order_acquire) != last_seen) + if (header()->change_seq.load(std::memory_order_acquire) != last_seen) { - return true; + return {}; } nanoseconds elapsed = kickmsg::elapsed_time(start); if (elapsed >= timeout) { - return false; + return std::make_error_code(std::errc::timed_out); } nanoseconds remaining = timeout - elapsed; // relaxed: the seq_cst fence below is the ordering edge. - h->waiters.fetch_add(1, std::memory_order_relaxed); + header()->waiters.fetch_add(1, std::memory_order_relaxed); // Pairs with notify_change()'s fence: orders our registration // before futex_wait's kernel read of change_seq. std::atomic_thread_fence(std::memory_order_seq_cst); // A writer that bumped before our increment may have seen // waiters == 0 and skipped the wake. - if (h->change_seq.load(std::memory_order_acquire) != last_seen) + if (header()->change_seq.load(std::memory_order_acquire) != last_seen) { - h->waiters.fetch_sub(1, std::memory_order_relaxed); - return true; + header()->waiters.fetch_sub(1, std::memory_order_relaxed); + return {}; } - futex_wait(h->change_seq, last_seen, remaining); - h->waiters.fetch_sub(1, std::memory_order_relaxed); + futex_wait(header()->change_seq, last_seen, remaining); + header()->waiters.fetch_sub(1, std::memory_order_relaxed); } } std::vector Blackboard::snapshot() const { require_open(base_); - auto* h = const_cast(header()); - uint32_t cap = h->capacity; + uint32_t capacity = header()->capacity; std::vector out; std::vector starttimes; // Reserved before the lock: row allocation is the only thing under // the critical section that can throw. - out.reserve(cap); - starttimes.reserve(cap); + out.reserve(capacity); + starttimes.reserve(capacity); // Serialized: a takeover rewrites owner_node with no seqlock over // those bytes, so an unlocked listing can return torn text. { - BoardGuard guard{const_cast(base_), h, 1024}; + BoardGuard guard{const_cast(base_), + const_cast(header()), 1024}; if (not guard) { throw std::runtime_error("Blackboard is busy: could not take the board lock"); } - for (uint32_t i = 0; i < cap; ++i) + EntryRead read; + for (uint32_t i = 0; i < capacity; ++i) { - auto* e = bb_entry_at(base_, h, i); - if (e->state.load(std::memory_order_acquire) != blackboard::Active) + // The lock covers the owner fields, never the value: read_entry's + // publish re-check is what stops two writes wrapping onto the chosen + // cell and pairing one update_count with a later write's len/stamp. + std::error_code const ec = read_entry(i, read, nullptr); + if (ec and ec != std::errc::no_message) { continue; } - blackboard::KeyStatus ks{}; - bool coherent = false; + auto* entry = bb_entry_at(base_, i); + blackboard::KeyStatus status{}; + status.key = std::move(read.key); + status.value_len = read.value_len; + status.updated_at_ns = read.updated_at_ns; + status.update_count = read.update_count; + status.owner_node = std::string(entry->owner_node, + ::strnlen(entry->owner_node, + sizeof(entry->owner_node))); + status.owner_pid = entry->owner_pid.load(std::memory_order_relaxed); + + starttimes.push_back(entry->owner_starttime.load(std::memory_order_relaxed)); + out.push_back(std::move(status)); + } + } - // The lock excludes metadata, not writers. Without the same - // publish re-check read() does, two writes can wrap onto the - // chosen cell and pair update_count k with the len/stamp of k+2. - for (int retry = 0; retry < blackboard::READ_RETRY_BUDGET; ++retry) - { - uint64_t t1 = e->tenancy.load(std::memory_order_acquire); - - ks.key = read_field(e->key, sizeof(e->key)); - ks.owner_node = read_field(e->owner_node, sizeof(e->owner_node)); - ks.owner_pid = e->owner_pid.load(std::memory_order_relaxed); - - uint64_t v1 = e->publish.load(std::memory_order_acquire); - uint64_t k = v1 >> 1; - ks.update_count = k; - ks.value_len = 0; - ks.updated_at_ns = 0; - if (k > 0) - { - auto* cell = bb_cell_at(base_, h, i, k); - std::size_t len = cell->value_len.load(std::memory_order_relaxed); - if (len > value_capacity(h)) - { - len = value_capacity(h); - } - ks.value_len = len; - ks.updated_at_ns = cell->updated_at_ns.load(std::memory_order_relaxed); - } + // Probed after the lock is dropped: one /proc read per active key, + // held across a full board, is long enough for a concurrent declare() + // to burn its yield budget and report a busy board. + for (std::size_t i = 0; i < out.size(); ++i) + { + out[i].owner_alive = out[i].owner_pid != 0 + and not owner_is_dead(out[i].owner_pid, starttimes[i]); + } + return out; + } - std::atomic_thread_fence(std::memory_order_acquire); - uint64_t v2 = e->publish.load(std::memory_order_acquire); - uint64_t t2 = e->tenancy.load(std::memory_order_acquire); - uint32_t s2 = e->state.load(std::memory_order_acquire); + std::error_code Blackboard::read_entry(uint32_t i, EntryRead& out, + std::vector* value) const + { + BlackboardEntry* entry = bb_entry_at(base_, i); + std::size_t const limit = value_capacity(header()); - if (s2 != blackboard::Active or t1 != t2) - { - break; - } - if (k > 0 - and v2 - (v1 & ~1ULL) >= 2 * blackboard::CELLS_PER_KEY - 1) - { - continue; - } - coherent = true; - break; + for (int retry = 0; retry < blackboard::READ_RETRY_BUDGET; ++retry) + { + if (entry->state.load(std::memory_order_acquire) != blackboard::Active) + { + return std::make_error_code(std::errc::no_such_file_or_directory); + } + uint64_t tenancy_before = entry->tenancy.load(std::memory_order_acquire); + + out = EntryRead{bb_read_key(entry->key, sizeof(entry->key)), 0, 0, 0}; + + uint64_t publish_before = entry->publish.load(std::memory_order_acquire); + uint64_t writes = publish_before >> 1; + out.update_count = writes; + if (writes > 0) + { + auto* cell = bb_cell_at(base_, i, writes); + // relaxed: ordered by the acquire load of publish above and + // validated by the re-check below. + std::size_t len = cell->value_len.load(std::memory_order_relaxed); + // A torn or hostile length must never reach the memcpy. + if (len > limit) + { + len = limit; } - if (not coherent) + out.value_len = len; + out.updated_at_ns = cell->updated_at_ns.load(std::memory_order_relaxed); + if (value != nullptr) { - continue; + value->resize(len); + if (len > 0) + { + bb_copy_payload(value->data(), bb_cell_payload(cell), len); + } } + } + + // Load-bearing: an acquire LOAD orders only later accesses, so + // without this the relaxed cell reads could be satisfied after the + // re-checks below (cf. Reader::read). + std::atomic_thread_fence(std::memory_order_acquire); + uint64_t publish_after = entry->publish.load(std::memory_order_acquire); + uint64_t tenancy_after = entry->tenancy.load(std::memory_order_acquire); + uint32_t state_after = entry->state.load(std::memory_order_acquire); - starttimes.push_back(e->owner_starttime.load(std::memory_order_relaxed)); - out.push_back(std::move(ks)); + if (state_after != blackboard::Active or tenancy_after != tenancy_before) + { + continue; } + // The cell is clobbered once write `writes + CELLS_PER_KEY` starts. + if (writes > 0 and publish_after - (publish_before & ~1ULL) >= 2 * blackboard::CELLS_PER_KEY - 1) + { + continue; + } + if (writes == 0) + { + return std::make_error_code(std::errc::no_message); + } + return {}; } + return std::make_error_code(std::errc::resource_unavailable_try_again); + } - // Probed after the lock is dropped: one /proc read per active key, - // held across a full board, is long enough for a concurrent declare() - // to burn its yield budget and report a busy board. - for (std::size_t i = 0; i < out.size(); ++i) + std::vector Blackboard::keys(std::string_view prefix) const + { + require_open(base_); + uint32_t const capacity = header()->capacity; + + std::vector out; + EntryRead entry; + for (uint32_t i = 0; i < capacity; ++i) { - out[i].owner_alive = out[i].owner_pid != 0 - and not owner_is_dead(out[i].owner_pid, starttimes[i]); + std::error_code const ec = read_entry(i, entry, nullptr); + if (ec and ec != std::errc::no_message) + { + continue; + } + if (not std::string_view{entry.key}.starts_with(prefix)) + { + continue; + } + out.push_back(std::move(entry.key)); + } + return out; + } + + std::unordered_map> + Blackboard::read_all(std::string_view prefix) const + { + require_open(base_); + uint32_t const capacity = header()->capacity; + + std::unordered_map> out; + EntryRead entry; + std::vector value; + for (uint32_t i = 0; i < capacity; ++i) + { + if (read_entry(i, entry, &value)) + { + continue; + } + if (not std::string_view{entry.key}.starts_with(prefix)) + { + continue; + } + out.emplace(std::move(entry.key), std::move(value)); } return out; } - uint32_t Blackboard::sweep_locked(BlackboardHeader* h) + uint32_t Blackboard::sweep_locked(BlackboardHeader* header) { uint32_t reclaimed = 0; - for (uint32_t i = 0; i < h->capacity; ++i) + for (uint32_t i = 0; i < header->capacity; ++i) { - auto* e = bb_entry_at(base_, h, i); - if (normalize_entry(e)) + auto* entry = bb_entry_at(base_, i); + if (normalize_entry(entry)) { ++reclaimed; continue; } // Only a clean Free entry or an Active one with a key reaches here. - uint64_t pid = e->owner_pid.load(std::memory_order_relaxed); + uint64_t pid = entry->owner_pid.load(std::memory_order_relaxed); if (pid == 0) { // Released on purpose, still holding its value -- not residue. continue; } - if (not owner_is_dead(pid, e->owner_starttime.load(std::memory_order_relaxed))) + if (not owner_is_dead(pid, entry->owner_starttime.load(std::memory_order_relaxed))) { continue; } - e->tenancy.fetch_add(1, std::memory_order_relaxed); - publish_free(e); + entry->tenancy.fetch_add(1, std::memory_order_relaxed); + publish_free(entry); ++reclaimed; } return reclaimed; @@ -1254,18 +1339,17 @@ namespace kickmsg uint32_t Blackboard::sweep_stale() { require_open(base_); - auto* h = header(); - BoardGuard guard{base_, h, 4096}; + BoardGuard guard{base_, header(), 4096}; if (not guard) { return 0; } - uint32_t reclaimed = sweep_locked(h); + uint32_t reclaimed = sweep_locked(header()); if (reclaimed > 0) { - notify_change(h); + notify_change(header()); } return reclaimed; } diff --git a/tests/blackboard_crash_test.cc b/tests/blackboard_crash_test.cc index 119a885..7b5f6bf 100644 --- a/tests/blackboard_crash_test.cc +++ b/tests/blackboard_crash_test.cc @@ -174,7 +174,7 @@ namespace BbPayload got{}; for (int i = 0; i < 5000; ++i) { - if (r.read(got).status == blackboard::Ok) + if (r.read(got).ec == std::error_code{}) { return true; } @@ -220,11 +220,11 @@ static bool test_value_survives_owner_death() { BbPayload got{}; auto out = r.read(got); - if (out.status != blackboard::Ok) + if (out.ec) { std::fprintf(stderr, - " [FAIL] round %d: read status %u after owner death\n", - round, static_cast(out.status)); + " [FAIL] round %d: read %s after owner death\n", + round, out.ec.message().c_str()); ok = false; break; } @@ -269,7 +269,7 @@ static bool test_value_survives_owner_death() ok = false; } BbPayload got{}; - if (r.read(got).status != blackboard::Missing) + if (r.read(got).ec != std::make_error_code(std::errc::no_such_file_or_directory)) { std::fprintf(stderr, " [FAIL] round %d: key readable after sweep\n", round); ok = false; @@ -301,7 +301,7 @@ static bool test_takeover_after_owner_death() BbPayload before{}; auto out = r.read(before); - if (out.status != blackboard::Ok or not payload_valid(before)) + if (out.ec != std::error_code{} or not payload_valid(before)) { std::fprintf(stderr, " [FAIL] no valid value after owner death\n"); ok = false; @@ -322,7 +322,7 @@ static bool test_takeover_after_owner_death() BbPayload after{}; out = r.read(after); - if (out.status != blackboard::Ok or after.seq != before.seq + if (out.ec != std::error_code{} or after.seq != before.seq or out.update_count != count_before) { std::fprintf(stderr, " [FAIL] takeover did not preserve the prior value\n"); @@ -332,13 +332,13 @@ static bool test_takeover_after_owner_death() // The publish counter continues rather than rewinding. BbPayload fresh{}; fill_payload(fresh, 0xABCD); - if (not w2.write(fresh)) + if (w2.write(fresh)) { std::fprintf(stderr, " [FAIL] write after takeover failed\n"); ok = false; } out = r.read(after); - if (out.status != blackboard::Ok or after.seq != 0xABCD + if (out.ec != std::error_code{} or after.seq != 0xABCD or out.update_count != count_before + 1) { std::fprintf(stderr, " [FAIL] counter did not continue after takeover " @@ -508,7 +508,7 @@ static bool test_unreaped_owner_is_reclaimable() auto w = bb.declare(KEY, "restarted"); BbPayload fresh{}; fill_payload(fresh, 7); - if (not w.write(fresh)) + if (w.write(fresh)) { std::fprintf(stderr, " [FAIL] write after zombie takeover failed\n"); ok = false; @@ -546,7 +546,7 @@ static bool test_forked_writer_does_not_touch_the_parents_key() BbPayload mine{}; fill_payload(mine, 1); - if (not w.write(mine)) + if (w.write(mine)) { std::fprintf(stderr, " [FAIL] parent could not write\n"); return false; @@ -560,7 +560,7 @@ static bool test_forked_writer_does_not_touch_the_parents_key() BbPayload theirs{}; fill_payload(theirs, 2); int code = 0; - if (w.write(theirs)) + if (not w.write(theirs)) { code = 1; } @@ -584,14 +584,14 @@ static bool test_forked_writer_does_not_touch_the_parents_key() } BbPayload got{}; - if (r.read(got).status != blackboard::Ok or got.seq != 1) + if (r.read(got).ec != std::error_code{} or got.seq != 1) { std::fprintf(stderr, " [FAIL] the child's write reached the value\n"); ok = false; } fill_payload(mine, 3); - if (not w.write(mine)) + if (w.write(mine)) { std::fprintf(stderr, " [FAIL] parent lost its claim to the fork\n"); ok = false; diff --git a/tests/python/test_blackboard.py b/tests/python/test_blackboard.py index cd88338..fe4cc6e 100644 --- a/tests/python/test_blackboard.py +++ b/tests/python/test_blackboard.py @@ -3,6 +3,7 @@ from __future__ import annotations import datetime +import errno import os import signal import struct @@ -34,14 +35,14 @@ def board(request): def test_late_reader_sees_current_value(board): bb, name = board w = bb.declare("arm/state", "arm_driver") - assert w.write(struct.pack("(writer_id), seq); - if (w.write(value)) + if (not w.write(value)) { g_writes.fetch_add(1, std::memory_order_relaxed); } @@ -94,13 +94,13 @@ namespace for (int i = 0; i < num_writers; ++i) { auto out = readers[static_cast(i)].read(got); - if (out.status == blackboard::Busy) + if (out.ec == std::errc::resource_unavailable_try_again) { g_busy.fetch_add(1, std::memory_order_relaxed); continue; } - if (out.status == blackboard::Unset - or out.status == blackboard::Missing) + if (out.ec == std::errc::no_message + or out.ec == std::errc::no_such_file_or_directory) { continue; } diff --git a/tests/tsan.supp b/tests/tsan.supp index 68d0f8f..6cd5f3c 100644 --- a/tests/tsan.supp +++ b/tests/tsan.supp @@ -22,6 +22,12 @@ race:bb_copy_payload # another key's value. race:bb_key_equals -# Both helpers are noinline (KICKMSG_BB_NOINLINE in src/Blackboard.cc) so these +# Blackboard key bytes, copy side: snapshot() reads them under the board lock, +# but keys()/read_all() walk the entry array unlocked and hit the same +# claim_free_slot() write that bb_key_equals races. A torn copy is discarded by +# the tenancy re-check that follows it, so it can never be returned. +race:bb_read_key + +# All three helpers are noinline (KICKMSG_BB_NOINLINE in src/Blackboard.cc) so these # suppressions match one frame each: write(), read() and resolve() themselves # stay checked, as does the whole publish/receive/repair core. diff --git a/tests/unit/blackboard-t.cc b/tests/unit/blackboard-t.cc index 7ed2707..5ecbb02 100644 --- a/tests/unit/blackboard-t.cc +++ b/tests/unit/blackboard-t.cc @@ -1,4 +1,6 @@ +#include +#include #include #include @@ -24,7 +26,7 @@ namespace BlackboardEntry* entry(Blackboard& bb, uint32_t index) { auto* h = bb.header(); - return bb_entry_at(static_cast(h), h, index); + return bb_entry_at(static_cast(h), index); } } @@ -113,7 +115,7 @@ TEST_F(BlackboardTest, LateReaderSeesCurrentValue) { auto bb = open(); auto w = bb.declare("arm/state"); - ASSERT_TRUE(w.write(Sample{7, 3})); + ASSERT_FALSE(w.write(Sample{7, 3})); // The reader attaches only now and never waits for a second write. auto other = open(); @@ -121,7 +123,7 @@ TEST_F(BlackboardTest, LateReaderSeesCurrentValue) Sample got{}; auto out = reader.read(got); - EXPECT_EQ(out.status, blackboard::Ok); + EXPECT_EQ(out.ec, std::error_code{}); EXPECT_EQ(out.len, sizeof(Sample)); EXPECT_EQ(out.update_count, 1u); EXPECT_EQ(got.id, 7u); @@ -134,14 +136,14 @@ TEST_F(BlackboardTest, ObserveBeforeDeclareResolvesLazily) auto reader = bb.observe("late/key"); Sample got{}; - EXPECT_EQ(reader.read(got).status, blackboard::Missing); + EXPECT_EQ(reader.read(got).ec, std::make_error_code(std::errc::no_such_file_or_directory)); auto w = bb.declare("late/key"); - ASSERT_TRUE(w.write(Sample{1, 2})); + ASSERT_FALSE(w.write(Sample{1, 2})); // Same Reader object, no re-observe. auto out = reader.read(got); - EXPECT_EQ(out.status, blackboard::Ok); + EXPECT_EQ(out.ec, std::error_code{}); EXPECT_EQ(got.id, 1u); } @@ -153,7 +155,7 @@ TEST_F(BlackboardTest, DeclaredButUnwrittenKeyReadsUnset) Sample got{}; auto out = r.read(got); - EXPECT_EQ(out.status, blackboard::Unset); + EXPECT_EQ(out.ec, std::make_error_code(std::errc::no_message)); EXPECT_EQ(out.len, 0u); EXPECT_EQ(out.update_count, 0u); } @@ -172,7 +174,7 @@ TEST_F(BlackboardTest, RedeclareAfterWriterDestructionSucceeds) auto bb = open(); { auto w = bb.declare("cycle"); - ASSERT_TRUE(w.write(Sample{5, 5})); + ASSERT_FALSE(w.write(Sample{5, 5})); } auto w2 = bb.declare("cycle"); EXPECT_TRUE(w2.valid()); @@ -184,12 +186,12 @@ TEST_F(BlackboardTest, ReleasedKeyKeepsItsValue) auto r = bb.observe("released"); { auto w = bb.declare("released"); - ASSERT_TRUE(w.write(Sample{9, 1})); + ASSERT_FALSE(w.write(Sample{9, 1})); } Sample got{}; auto out = r.read(got); - EXPECT_EQ(out.status, blackboard::Ok); + EXPECT_EQ(out.ec, std::error_code{}); EXPECT_EQ(got.id, 9u); EXPECT_FALSE(r.owner_alive()); @@ -204,7 +206,7 @@ TEST_F(BlackboardTest, TakeoverPreservesPriorValueUntilFirstWrite) auto bb = open(); { auto w = bb.declare("arm/state"); - ASSERT_TRUE(w.write(Sample{11, 4})); + ASSERT_FALSE(w.write(Sample{11, 4})); orphan(bb, "arm/state"); } @@ -213,14 +215,14 @@ TEST_F(BlackboardTest, TakeoverPreservesPriorValueUntilFirstWrite) Sample got{}; auto out = r.read(got); - EXPECT_EQ(out.status, blackboard::Ok); + EXPECT_EQ(out.ec, std::error_code{}); EXPECT_EQ(got.id, 11u); EXPECT_EQ(out.update_count, 1u); // The counter continues rather than rewinding. - ASSERT_TRUE(w2.write(Sample{12, 5})); + ASSERT_FALSE(w2.write(Sample{12, 5})); out = r.read(got); - EXPECT_EQ(out.status, blackboard::Ok); + EXPECT_EQ(out.ec, std::error_code{}); EXPECT_EQ(got.id, 12u); EXPECT_EQ(out.update_count, 2u); } @@ -229,22 +231,141 @@ TEST_F(BlackboardTest, SweepStaleLeavesLiveOwnersAlone) { auto bb = open(); auto w = bb.declare("live"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); EXPECT_EQ(bb.sweep_stale(), 0u); EXPECT_EQ(bb.snapshot().size(), 1u); } +TEST_F(BlackboardTest, KeysListsEveryActiveKeyIncludingUnwritten) +{ + auto bb = open(); + auto a = bb.declare("arm/state"); + auto b = bb.declare("hand/0"); + ASSERT_FALSE(a.write(Sample{1, 1})); + // b is declared and never written: it exists, so it is listed. + + auto got = bb.keys(); + std::sort(got.begin(), got.end()); + EXPECT_EQ(got, (std::vector{"arm/state", "hand/0"})); +} + +TEST_F(BlackboardTest, KeysAndReadAllFilterOnPrefix) +{ + auto bb = open(); + auto h0 = bb.declare("hand/0"); + auto h1 = bb.declare("hand/1"); + auto arm = bb.declare("arm/state"); + ASSERT_FALSE(h0.write(Sample{1, 1})); + ASSERT_FALSE(h1.write(Sample{2, 2})); + ASSERT_FALSE(arm.write(Sample{3, 3})); + + auto named = bb.keys("hand/"); + std::sort(named.begin(), named.end()); + EXPECT_EQ(named, (std::vector{"hand/0", "hand/1"})); + + auto values = bb.read_all("hand/"); + ASSERT_EQ(values.size(), 2u); + EXPECT_TRUE(values.contains("hand/0")); + EXPECT_TRUE(values.contains("hand/1")); + EXPECT_FALSE(values.contains("arm/state")); +} + +TEST_F(BlackboardTest, ReadAllReturnsTheValuesReaderWouldRead) +{ + auto bb = open(); + auto w = bb.declare("hand/0"); + ASSERT_FALSE(w.write(Sample{7, 3})); + + auto values = bb.read_all(); + ASSERT_EQ(values.size(), 1u); + + auto const& bytes = values.at("hand/0"); + ASSERT_EQ(bytes.size(), sizeof(Sample)); + Sample got{}; + std::memcpy(&got, bytes.data(), sizeof(got)); + EXPECT_EQ(got.id, 7u); + EXPECT_EQ(got.state, 3u); +} + +// A declared-but-unwritten key has no value, so it is a keys() row and not a +// read_all() one: the two answer different questions about the same entry. +TEST_F(BlackboardTest, ReadAllSkipsAKeyThatHoldsNoValue) +{ + auto bb = open(); + auto w = bb.declare("hand/0"); + + EXPECT_EQ(bb.keys().size(), 1u); + EXPECT_TRUE(bb.read_all().empty()); + + ASSERT_FALSE(w.write(Sample{1, 1})); + EXPECT_EQ(bb.read_all().size(), 1u); +} + +// Releasing a key keeps its value, so both listings still report it: that is +// what lets a reader see what a writer said before it went away. +TEST_F(BlackboardTest, KeysAndReadAllOutliveTheWriter) +{ + auto bb = open(); + { + auto w = bb.declare("hand/0"); + ASSERT_FALSE(w.write(Sample{5, 2})); + } + + EXPECT_EQ(bb.keys(), (std::vector{"hand/0"})); + EXPECT_EQ(bb.read_all().size(), 1u); +} + +// The typed form skips a key whose value is not sizeof(T), the listing +// counterpart of the EBADMSG a typed read reports for one. +TEST_F(BlackboardTest, TypedReadAllSkipsAValueOfTheWrongSize) +{ + auto bb = open(); + auto right = bb.declare("hand/0"); + auto wrong = bb.declare("hand/1"); + ASSERT_FALSE(right.write(Sample{7, 3})); + ASSERT_FALSE(wrong.write(uint32_t{9})); + + auto typed = bb.read_all("hand/"); + ASSERT_EQ(typed.size(), 1u); + EXPECT_EQ(typed.at("hand/0").id, 7u); + EXPECT_EQ(typed.at("hand/0").state, 3u); + + EXPECT_EQ(bb.read_all("hand/").size(), 2u); +} + +// Each cause gets its own errno rather than one collapsed "false". +TEST_F(BlackboardTest, WriteNamesWhyItFailed) +{ + auto bb = open(); + auto w = bb.declare("hand/0"); + + std::vector const over(bb.max_value_size() + 1, 0xAB); + EXPECT_EQ(w.write(over.data(), over.size()), + std::make_error_code(std::errc::message_size)); + + Blackboard::Writer const moved_from; + Blackboard::Writer taken = std::move(w); + EXPECT_EQ(w.write(Sample{1, 1}), + std::make_error_code(std::errc::bad_file_descriptor)); + + // Sweeping the entry out from under a live claim is the false-death-verdict + // path: the handle survives, the claim does not. + orphan(bb, "hand/0"); + EXPECT_EQ(taken.write(Sample{1, 1}), + std::make_error_code(std::errc::state_not_recoverable)); +} + TEST_F(BlackboardTest, SweepStaleFreesDeadOwnerAndLeavesUnownedAlone) { auto bb = open(); { auto dead = bb.declare("dead"); - ASSERT_TRUE(dead.write(Sample{1, 1})); + ASSERT_FALSE(dead.write(Sample{1, 1})); orphan(bb, "dead"); } { auto released = bb.declare("released"); - ASSERT_TRUE(released.write(Sample{2, 2})); + ASSERT_FALSE(released.write(Sample{2, 2})); } EXPECT_EQ(bb.sweep_stale(), 1u); @@ -261,15 +382,15 @@ TEST_F(BlackboardTest, UpdateCountIncrementsAndTimestampAdvances) auto w = bb.declare("tick"); auto r = bb.observe("tick"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); Sample got{}; auto first = r.read(got); - ASSERT_EQ(first.status, blackboard::Ok); + ASSERT_EQ(first.ec, std::error_code{}); std::this_thread::sleep_for(2ms); - ASSERT_TRUE(w.write(Sample{2, 2})); + ASSERT_FALSE(w.write(Sample{2, 2})); auto second = r.read(got); - ASSERT_EQ(second.status, blackboard::Ok); + ASSERT_EQ(second.ec, std::error_code{}); EXPECT_EQ(first.update_count, 1u); EXPECT_EQ(second.update_count, 2u); @@ -281,7 +402,7 @@ TEST_F(BlackboardTest, SnapshotReportsOwnerAndSkipsFreeSlots) auto bb = open(); auto a = bb.declare("a", "node_a"); auto b = bb.declare("b", "node_b"); - ASSERT_TRUE(a.write(Sample{1, 1})); + ASSERT_FALSE(a.write(Sample{1, 1})); auto snap = bb.snapshot(); ASSERT_EQ(snap.size(), 2u); @@ -318,7 +439,7 @@ TEST_F(BlackboardTest, WaitWakesOnAnyChange) }); // Waiting on the board, not on key "a": any value change wakes us. - EXPECT_TRUE(bb.wait(seq, 5s)); + EXPECT_FALSE(bb.wait(seq, 5s)); writer.join(); EXPECT_NE(bb.change_seq(), seq); } @@ -330,7 +451,7 @@ TEST_F(BlackboardTest, WaitReturnsFalseOnTimeout) uint64_t seq = bb.change_seq(); auto start = std::chrono::steady_clock::now(); - EXPECT_FALSE(bb.wait(seq, 50ms)); + EXPECT_TRUE(bb.wait(seq, 50ms)); EXPECT_GE(std::chrono::steady_clock::now() - start, 45ms); } @@ -339,10 +460,10 @@ TEST_F(BlackboardTest, WaitReturnsImmediatelyWhenSeqAlreadyAdvanced) auto bb = open(); auto w = bb.declare("k"); uint64_t stale = bb.change_seq(); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); auto start = std::chrono::steady_clock::now(); - EXPECT_TRUE(bb.wait(stale, 5s)); + EXPECT_FALSE(bb.wait(stale, 5s)); EXPECT_LT(std::chrono::steady_clock::now() - start, 1s); } @@ -353,14 +474,14 @@ TEST_F(BlackboardTest, ValueTooLargeIsRejectedAndPreservesPrevious) auto bb = open(); auto w = bb.declare("k"); auto r = bb.observe("k"); - ASSERT_TRUE(w.write(Sample{3, 3})); + ASSERT_FALSE(w.write(Sample{3, 3})); std::vector huge(4096, 0xAB); - EXPECT_FALSE(w.write(huge.data(), huge.size())); + EXPECT_TRUE(w.write(huge.data(), huge.size())); Sample got{}; auto out = r.read(got); - EXPECT_EQ(out.status, blackboard::Ok); + EXPECT_EQ(out.ec, std::error_code{}); EXPECT_EQ(out.update_count, 1u); EXPECT_EQ(got.id, 3u); } @@ -372,11 +493,11 @@ TEST_F(BlackboardTest, ReadIntoSmallBufferReportsTruncated) auto r = bb.observe("k"); std::vector payload(32, 0x5A); - ASSERT_TRUE(w.write(payload.data(), payload.size())); + ASSERT_FALSE(w.write(payload.data(), payload.size())); uint8_t guarded[8] = {0, 0, 0, 0, 0, 0, 0, 0}; auto out = r.read(guarded, 4); - EXPECT_EQ(out.status, blackboard::Truncated); + EXPECT_EQ(out.ec, std::make_error_code(std::errc::message_size)); EXPECT_EQ(out.len, 32u); for (auto byte : guarded) { @@ -412,15 +533,15 @@ TEST_F(BlackboardTest, CorruptValueLenIsClamped) auto bb = open(); auto w = bb.declare("k"); auto r = bb.observe("k"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); auto* h = bb.header(); - auto* cell = bb_cell_at(static_cast(h), h, 0, 1); + auto* cell = bb_cell_at(static_cast(h), 0, 1); cell->value_len.store(0xFFFFFFFFu, std::memory_order_relaxed); std::vector out; auto result = r.read(out); - EXPECT_EQ(result.status, blackboard::Ok); + EXPECT_EQ(result.ec, std::error_code{}); EXPECT_LE(result.len, bb.max_value_size()); } @@ -496,19 +617,19 @@ TEST_F(BlackboardTest, EntryRetenancyInvalidatesCachedReaderIndex) auto reader = bb.observe("a"); { auto w = bb.declare("a"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); Sample got{}; - ASSERT_EQ(reader.read(got).status, blackboard::Ok); // caches the index + ASSERT_EQ(reader.read(got).ec, std::error_code{}); // caches the index orphan(bb, "a"); } ASSERT_EQ(bb.sweep_stale(), 1u); auto w2 = bb.declare("b"); - ASSERT_TRUE(w2.write(Sample{99, 99})); + ASSERT_FALSE(w2.write(Sample{99, 99})); Sample got{}; auto out = reader.read(got); - EXPECT_EQ(out.status, blackboard::Missing); + EXPECT_EQ(out.ec, std::make_error_code(std::errc::no_such_file_or_directory)); EXPECT_NE(got.id, 99u); } @@ -521,7 +642,7 @@ TEST_F(BlackboardTest, FreshClaimDoesNotResurrectPreviousTenantValue) { auto w = bb.declare("a"); - ASSERT_TRUE(w.write(Sample{42, 42})); + ASSERT_FALSE(w.write(Sample{42, 42})); orphan(bb, "a"); } ASSERT_EQ(bb.sweep_stale(), 1u); @@ -529,7 +650,7 @@ TEST_F(BlackboardTest, FreshClaimDoesNotResurrectPreviousTenantValue) auto w2 = bb.declare("b"); auto r = bb.observe("b"); Sample got{}; - EXPECT_EQ(r.read(got).status, blackboard::Unset); + EXPECT_EQ(r.read(got).ec, std::make_error_code(std::errc::no_message)); } // ---- convenience overloads ---------------------------------------------- @@ -541,17 +662,17 @@ TEST_F(BlackboardTest, VectorReadResizesAndReusesCapacity) auto r = bb.observe("k"); std::vector payload(48, 0x7E); - ASSERT_TRUE(w.write(payload.data(), payload.size())); + ASSERT_FALSE(w.write(payload.data(), payload.size())); std::vector out; auto result = r.read(out); - ASSERT_EQ(result.status, blackboard::Ok); + ASSERT_EQ(result.ec, std::error_code{}); ASSERT_EQ(out.size(), 48u); EXPECT_EQ(out, payload); std::size_t cap_before = out.capacity(); result = r.read(out); - EXPECT_EQ(result.status, blackboard::Ok); + EXPECT_EQ(result.ec, std::error_code{}); EXPECT_EQ(out.capacity(), cap_before); } @@ -563,22 +684,22 @@ TEST_F(BlackboardTest, TypedReadRejectsAValueOfADifferentSize) auto w = bb.declare("k"); auto r = bb.observe("k"); - ASSERT_TRUE(w.write(uint32_t{0xABCD1234})); + ASSERT_FALSE(w.write(uint32_t{0xABCD1234})); uint64_t wide = 0xFFFFFFFFFFFFFFFFull; auto out = r.read(wide); - EXPECT_EQ(out.status, blackboard::SizeMismatch); + EXPECT_EQ(out.ec, std::make_error_code(std::errc::bad_message)); EXPECT_EQ(out.len, 4u); EXPECT_EQ(wide, 0xFFFFFFFFFFFFFFFFull) << "out must not be touched"; // The matching type still reads. uint32_t narrow = 0; - EXPECT_EQ(r.read(narrow).status, blackboard::Ok); + EXPECT_EQ(r.read(narrow).ec, std::error_code{}); EXPECT_EQ(narrow, 0xABCD1234u); // And a value larger than T is still Truncated, not a partial fill. - ASSERT_TRUE(w.write(std::vector(32, 0x11).data(), 32)); - EXPECT_EQ(r.read(narrow).status, blackboard::Truncated); + ASSERT_FALSE(w.write(std::vector(32, 0x11).data(), 32)); + EXPECT_EQ(r.read(narrow).ec, std::make_error_code(std::errc::message_size)); EXPECT_EQ(narrow, 0xABCD1234u); } @@ -610,11 +731,11 @@ TEST_F(BlackboardTest, RawByteRoundTrip) auto r = bb.observe("k"); char const* text = "lifecycle=ACTIVE"; - ASSERT_TRUE(w.write(text, std::strlen(text))); + ASSERT_FALSE(w.write(text, std::strlen(text))); char buf[64] = {}; auto out = r.read(buf, sizeof(buf)); - ASSERT_EQ(out.status, blackboard::Ok); + ASSERT_EQ(out.ec, std::error_code{}); EXPECT_EQ(std::string(buf, out.len), text); } @@ -683,10 +804,10 @@ TEST_F(BlackboardTest, MaxValueSizeIsExactlyAsConfigured) auto w = bb.declare("k"); std::vector exact(128, 0xEE); - EXPECT_TRUE(w.write(exact.data(), exact.size())); + EXPECT_FALSE(w.write(exact.data(), exact.size())); std::vector over(129, 0xEE); - EXPECT_FALSE(w.write(over.data(), over.size())); + EXPECT_TRUE(w.write(over.data(), over.size())); } TEST_F(BlackboardTest, BoardAtMaxValueSizeCanBeReopened) @@ -728,7 +849,7 @@ TEST_F(BlackboardTest, ConcurrentReleaseAndTakeoverYieldAWorkingWriter) { auto w = bb.declare("contended"); taken.store(true, std::memory_order_release); - wrote.store(w.write(uint32_t{7}), std::memory_order_release); + wrote.store(not w.write(uint32_t{7}), std::memory_order_release); w.release(); return; } @@ -778,7 +899,7 @@ TEST_F(BlackboardTest, FreedEntryDoesNotLeaveADeadPidForTheNextClaimant) auto bb = open(); { auto w = bb.declare("gone"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); orphan(bb, "gone"); } ASSERT_EQ(bb.sweep_stale(), 1u); @@ -792,7 +913,7 @@ TEST_F(BlackboardTest, FreedEntryDoesNotLeaveADeadPidForTheNextClaimant) // A fresh claim must survive a sweep running against it. auto w2 = bb.declare("fresh"); EXPECT_EQ(bb.sweep_stale(), 0u); - EXPECT_TRUE(w2.write(Sample{2, 2})); + EXPECT_FALSE(w2.write(Sample{2, 2})); } // ---- crash-point matrix -------------------------------------------------- @@ -926,9 +1047,9 @@ TEST_F(BlackboardTest, CrashPointMatrix) // Index 0 is the victim; index 1 is an untouched neighbour. auto victim = bb.declare("victim", "owner"); - ASSERT_TRUE(victim.write(Sample{1, 1})) << point.name; + ASSERT_FALSE(victim.write(Sample{1, 1})) << point.name; auto neighbour = bb.declare("neighbour", "owner"); - ASSERT_TRUE(neighbour.write(Sample{2, 2})) << point.name; + ASSERT_FALSE(neighbour.write(Sample{2, 2})) << point.name; auto* h = bb.header(); auto* e = entry(bb, 0); @@ -963,7 +1084,7 @@ TEST_F(BlackboardTest, CrashPointMatrix) try { auto probe = bb.declare("probe"); - declarable = probe.write(Sample{3, 3}); + declarable = not probe.write(Sample{3, 3}); } catch (std::runtime_error const&) { @@ -973,7 +1094,7 @@ TEST_F(BlackboardTest, CrashPointMatrix) Sample got{}; auto out = bb.observe("neighbour").read(got); - ok = ok and out.status == blackboard::Ok and got.id == 2u; + ok = ok and not out.ec and got.id == 2u; EXPECT_TRUE(ok) << ctx << " | state=" << st @@ -1013,7 +1134,7 @@ TEST_F(BlackboardTest, SnapshotReportsBusyRatherThanReadingUnlocked) // lock exists to prevent, so it must report busy instead of returning rows. auto bb = open(); auto w = bb.declare("k", "owner"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); auto* h = bb.header(); h->lock_token.store(live_lock_token(), std::memory_order_release); @@ -1034,7 +1155,7 @@ TEST_F(BlackboardTest, LockIsNeverLeftHeldOnAThrowingPath) auto* h = bb.header(); auto w = bb.declare("owned", "owner"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); EXPECT_THROW(bb.declare("owned"), std::runtime_error); EXPECT_EQ(h->lock_token.load(std::memory_order_acquire), 0u); @@ -1053,7 +1174,7 @@ TEST_F(BlackboardTest, LockIsNeverLeftHeldOnAThrowingPath) // And the board still works. auto w2 = bb.declare("after"); - EXPECT_TRUE(w2.write(Sample{2, 2})); + EXPECT_FALSE(w2.write(Sample{2, 2})); } TEST_F(BlackboardTest, ReleaseWaitsRatherThanStrandingTheKey) @@ -1065,7 +1186,7 @@ TEST_F(BlackboardTest, ReleaseWaitsRatherThanStrandingTheKey) auto* h = bb.header(); auto w = bb.declare("stuck", "owner"); - ASSERT_TRUE(w.write(Sample{1, 1})); + ASSERT_FALSE(w.write(Sample{1, 1})); // A peer takes the board and holds it across the release. h->lock_token.store(live_lock_token(), std::memory_order_release); @@ -1085,7 +1206,7 @@ TEST_F(BlackboardTest, ReleaseWaitsRatherThanStrandingTheKey) // Ownership really was cleared, so the key is redeclarable. auto w2 = bb.declare("stuck"); - EXPECT_TRUE(w2.write(Sample{2, 2})); + EXPECT_FALSE(w2.write(Sample{2, 2})); } // A hammer, not a proof: it still passes with the publish re-check removed, diff --git a/tests/unit/node-t.cc b/tests/unit/node-t.cc index b97ff3c..cbfa7e1 100644 --- a/tests/unit/node-t.cc +++ b/tests/unit/node-t.cc @@ -543,7 +543,7 @@ TEST_F(NodeTest, BlackboardCrossNodeLateReader) kickmsg::Node writer_node("writer", "test"); auto& bb = writer_node.blackboard("state"); auto w = bb.declare("lifecycle", "writer"); - ASSERT_TRUE(w.write(value)); + ASSERT_FALSE(w.write(value)); // The reader node is constructed only now, after the single write. kickmsg::Node reader_node("reader", "test"); @@ -552,7 +552,7 @@ TEST_F(NodeTest, BlackboardCrossNodeLateReader) uint32_t got = 0; auto out = r.read(got); - EXPECT_EQ(out.status, kickmsg::blackboard::Ok); + EXPECT_FALSE(out.ec); EXPECT_EQ(got, value); } }