diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 3c8308f0..2a43f38d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -562,6 +562,16 @@ jobs: working-directory: stress-tests/zig/lifecycle_churn run: ./run.py --scenario reentrant --threads 4 --iterations 25 --tsan --timeout 300 + # listener + cft each surfaced a real data race that now has a fix + # (see stress-tests/README.md "Findings"); a TSan lane pins them. + - name: lifecycle_churn/listener under ThreadSanitizer + working-directory: stress-tests/zig/lifecycle_churn + run: ./run.py --scenario listener --threads 4 --duration 20 --tsan --timeout 300 + + - name: lifecycle_churn/cft under ThreadSanitizer + working-directory: stress-tests/zig/lifecycle_churn + run: ./run.py --scenario cft --threads 6 --duration 20 --tsan --timeout 300 + # Weekly full-vendor sweep trigger only: heavier, longer, + --large. - name: Nightly heavy matrix if: github.event_name == 'schedule' diff --git a/CHANGELOG.md b/CHANGELOG.md index 5080eb2f..87ca1e58 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,88 @@ see [`docs/implementation_status.md`](docs/implementation_status.md); for planne Dated entries (no release tags past `v0.2.1-zig.0.16.0`; `build.zig.zon` is `0.2.1-zig.0.16.0-dev`). +## 2026-09-02 + +- **Pinned `zidl` v0.3.12-zig.0.16.0** (`build.zig.zon`) — the selective-parse family + (`deserialize_selected` / `KEY_FIELD_MASK` / `skipPrimitives`) that rewires + `get_key_value` and `get_field_from_cdr` off the key-only deserializer. The + `lifecycle_churn` `instance` scenario now **asserts** `get_key_value`'s returned + `subject_id` (non-leading `@key`) on both the writer and reader sides — it was + call-only-not-asserted while the fix was unreleased. Verified: 8 threads × 6 s, plus + ThreadSanitizer, clean; full stress suite green. +- **Fix — keyed writers now always send inline `PID_KEY_HASH`; a present all-zero hash is + honoured.** Two connected key-hash bugs the `instance` stress scenario exposed for a + zero-valued key (`subject_id == 0`): + 1. `writer_sm.zig` suppressed the inline `PID_KEY_HASH` parameter whenever the computed + hash was all-zero bytes — indistinguishable from "keyless topic" but also true for a + legitimate zero key. A subscriber then had to reconstruct the per-instance hash from + the payload. + 2. `resolveKeyHash` treated a *received* all-zero hash as "absent, recompute" rather + than as the value the writer sent. + Now: `TypeSupport` gains `has_key`; `pubCreateProtoWriter` marks the RTPS `StatefulWriter` + keyed (via new `setKeyed` / adapter passthrough); a keyed writer emits `PID_KEY_HASH` on + every DATA/DATA_FRAG including an all-zero key (`writer_sm.zig` `inlineKeyHash` helper). + `decodeKeyHash` returns `?[16]u8` and `resolveKeyHash` returns a *present* hash verbatim, + even all-zero. The C-ABI `zzdds_register_type_support{,_ctx}` infer `has_key` from a + non-NULL `compute_key_hash_fn` (per their documented contract); Zig-native callers set it + explicitly (stress harness updated). Regression: `test/dcps/type_support_test.zig` — a + keyed writer's zero-key sample routes by the wire hash and bypasses `key_hash_fn`, with a + control test showing the unchanged non-`has_key` fallback. Verified by re-breaking. +- **Known follow-up.** `TypeSupport.compute_key_hash`'s contract is "full CDR payload in, + hash out", but zidl's generated `computeKeyHashFromCdr` runs the key-only deserializer, + so a non-`has_key` fallback still misreads a non-leading `@key` from a non-zzdds peer that + omits the inline hash. Completion — route the fallback through + `deserialize_selected(KEY_FIELD_MASK)` on a full payload, K-flag-gated — needs a + `TypeSupport.compute_key_hash` signature change (`is_key_only: bool`) rippling to the + C ABI mirror and a further zidl release, so it is a follow-up beyond the v0.3.12 bump. + Tracked in `docs/roadmap.md` "Selective CDR parse — deferred follow-ups". + +## 2026-08-30 + +- **Stress tests — five new `lifecycle_churn` churn scenarios.** On top of + `entities`/`reentrant`: `waitset` (threads attach/detach Read/QueryConditions on a + shared WaitSet while a waiter is in `wait()` and a waker flips a GuardCondition; + includes a deliberate delete-while-attached), `listener` (participant + publisher + + subscriber listeners installed; per-iteration `set_listener` swaps incl. `null` racing + entity teardown and event delivery), `cft` (a shared long-lived ContentFilteredTopic + hammered with `set_expression_parameters` while its reader is drained, alongside + unique-name CFT + reader lifecycle churn), `participants` (N threads each churning a + whole participant on one shared domain, listeners at every level, writer/reader + fan-in/fan-out — the participant level of the §2.2.4.1.5 fallback under teardown), and + `instance` (per-thread writer; `register_instance` / `write` / `dispose` / + `unregister_instance` / `get_key_value` / `lookup_instance` churn with a shared reader + fan-in). All gating in the `stress` CI job; `listener` and `cft` also run under + ThreadSanitizer. See `stress-tests/README.md`. The `instance` scenario surfaced two + pre-existing bugs: the concurrent-`write()` race (next entry, fixed here) and + `get_key_value` decoding the wrong key for a non-leading `@key` member — fixed in zidl + (a *selective-parse family*: `deserialize_selected(KEY_FIELD_MASK)` decodes just the + `@key` members and skips the rest, all four backends), landing here with the zidl + v0.3.12 `build.zig.zon` bump; the scenario's `get_key_value` value assertion is staged + for that PR (`zz-dev/zidl-v0.3.12-pin-bump-followups.md`). +- **Fix — concurrent `write()` on a single `DataWriter` was unsynchronised.** + `DataWriterImpl.writeRaw` updated `last_sn` and the `get_key_value` key registry (a + `HashMapUnmanaged`) with no lock, so two application threads calling `write()` / + `dispose()` / `unregister_instance()` on the same writer — spec-legal — raced on the + map's grow/insert and could abort on its `SafetyLock` (found by `instance` under + ThreadSanitizer). The RTPS layer under `proto_writer` was already internally locked; now + `last_sn` is a `std.atomic.Value` and the key registry is guarded by a dedicated + `key_registry_mu`. `docs/design/thread-model.md` documents the guarantee. Regression: + `test/dcps/writer_vtable_test.zig` — 6 threads × 40 keyed `write_raw` calls on one + writer plus a concurrent `getKeyValueRaw` poller; also runs in the `test-tsan` lane. +- **Fix — unsynchronised `listener_mask` (data race).** `listener_mask` was a plain + `u32` written unlocked by `set_listener` and read unlocked by the discovery/timer + dispatch path (`listener_mu` only ever covered the `ListenerBox` swap beside it). + Every *runtime* access in `src/dcps/{writer,reader,publisher,subscriber,participant, + topic}.zig` is now `@atomicLoad`/`@atomicStore` `.monotonic`; struct-literal + initialisers stay plain. Found by the `listener` scenario under TSan. +- **Fix — CFT `set_expression_parameters` use-after-free.** `ContentFilteredTopicImpl` + had no synchronisation on `expr_params`: `set_expression_parameters` frees the old + parameter strings + backing array and swaps in the new list while the receive thread's + `matchSample` is mid-`filter_mod.eval` holding those strings by reference (SEGV in + `parseFloat`). New `params_lock: Mutex` on the impl, held across `matchSample`'s eval + and around the swap in `set_expression_parameters` / the read in + `get_expression_parameters`. Found by the `cft` scenario (~40% repro at 12 threads). + ## 2026-08-29 - **CI flake fix — unique DDS domain per test binary.** `zig build test` runs the ~29 diff --git a/build.zig.zon b/build.zig.zon index 27668351..d284b45f 100644 --- a/build.zig.zon +++ b/build.zig.zon @@ -5,8 +5,8 @@ .minimum_zig_version = "0.16.0", .dependencies = .{ .zidl = .{ - .url = "https://github.com/zz-iot/zidl/archive/refs/tags/v0.3.11-zig.0.16.0.tar.gz", - .hash = "zidl-0.3.11-zig.0.16.0-H6NwLiWeKwBt69SjWREwhsbM6fQEFZTdRIfdrzz9TKKy", + .url = "https://github.com/zz-iot/zidl/archive/refs/tags/v0.3.12-zig.0.16.0.tar.gz", + .hash = "zidl-0.3.12-zig.0.16.0-H6NwLvgwLADRpeAy5HYLSMMIV9u0PO-S4ek4IyXEUlCF", }, }, .paths = .{ diff --git a/docs/binding-release-plan.md b/docs/binding-release-plan.md index b7aeeab8..96ae50cc 100644 --- a/docs/binding-release-plan.md +++ b/docs/binding-release-plan.md @@ -9,7 +9,7 @@ zzdds client surfaces. *(done — shipped in `zidl v0.3.11-zig.0.16.0`.)* - Update `build.zig.zon` from the local `../zidl` path to the released zidl package URL/hash after that tag exists. - *(done — `build.zig.zon` pins `zidl v0.3.11-zig.0.16.0` by URL + hash.)* + *(done — `build.zig.zon` pins `zidl v0.3.12-zig.0.16.0` by URL + hash.)* - Keep `idl/dcps.idl` normative and put zzdds-specific configuration and extension interfaces in `idl/zzdds.idl`. - Keep `include/zzdds_c.h` as a small support ABI for generated wrappers; prefer diff --git a/docs/design/dcps-api-coverage-audit.md b/docs/design/dcps-api-coverage-audit.md index 73e06b37..a181aaa2 100644 --- a/docs/design/dcps-api-coverage-audit.md +++ b/docs/design/dcps-api-coverage-audit.md @@ -48,7 +48,7 @@ For each API found, classified as: | Rejection/loss | `on_sample_rejected`/`on_sample_lost`, `get_sample_rejected_status`/`get_sample_lost_status` — neither listener nor polling form, anywhere | | Historical data | `wait_for_historical_data` — confirmed zero across every harness | | Timestamped/explicit instance ops | `register_instance` (explicit), `register_instance_w_timestamp`, `write_w_timestamp`, `dispose_w_timestamp`, `unregister_instance_w_timestamp` | -| Instance introspection | `get_key_value`, `lookup_instance` | +| Instance introspection | `lookup_instance`; `get_key_value` now exercised by the stress `instance` scenario, which found it returns the wrong key for non-leading-key types (zidl codegen, all backends — see below) | | Loans | `return_loan_raw`, any loaned-read (`take_raw`/`read_raw` in loan mode) or write-loan (`loan_raw`/`publish_loan_raw`) path — no `zzdds-examples` port exercises these yet (internal test-suite coverage exists, see the loan-lifecycle entry below) | | Entity admin, post-creation | `set_qos`/`get_qos` round-trip, `get_listener` read-back, `enable()`, `get_status_changes()`, `contains_entity()` | | Discovery/ignore | `ignore_participant`/`ignore_topic`/`ignore_publication`/`ignore_subscription`, `get_discovered_participants`/`get_discovered_topics` + `_data` variants | @@ -128,8 +128,10 @@ concurrency/lifecycle-under-load, `OpenDDS EntityLifecycleStress`-shaped). needs 2 real processes to mean anything. - **`set_expression_parameters` at runtime** (CFT dynamic reconfiguration) — does changing parameters without recreating the CFT actually re-filter subsequent samples? - Real spec-mandated behavior, currently fully untested, and CFT has an established bug - history in this project (missing null-checks found in this same audit). + Real spec-mandated behavior; the *behavioural* question is still untested (the + stress-tier `cft` scenario now covers its *concurrency safety* and fixed a UAF there). + CFT has an established bug history in this project (missing null-checks found in this + same audit). - **Coherent/ordered access grouping correctness** — build a real coherent set across multiple writers, verify atomic delivery. High value given past CoherentSets flakiness investigations. @@ -156,23 +158,32 @@ concurrency/lifecycle-under-load, `OpenDDS EntityLifecycleStress`-shaped). project has repeatedly found real bugs specifically in teardown-cascade edge cases. ### → Stress tests (new, in-repo) -- **Generalized reentrant-listener/entity-lifecycle churn** — the existing - `participant_vtable_test` "reentrant delete_participant from a timer-driven listener" - unit test is a single hand-built scenario; a stress test hammers this pattern with many - concurrent writers/readers/waitsets/listeners simultaneously creating/deleting/firing. -- **WaitSet/Condition churn under load** — many threads attaching/detaching conditions to - a shared WaitSet while `wait()` is blocked elsewhere, concurrent GuardCondition - set/reset — directly exercises the `CachedCAbiHandle`/`EntityQuiesce` machinery under - real pressure, not just narrow unit tests. -- **Listener-fallback chain under load** — concurrent reader/subscriber/participant - deletion racing concurrent `set_listener()` replacement and event delivery (this - project's most recent feature, see `docs/decisions.md` "Listener hierarchy fallback"). -- **Many-writer/many-reader fan-in/fan-out discovery** — SPDP/SEDP under N participants - joining/leaving concurrently, matching OpenDDS Bench's discovery/fan-in/fan-out - scenario shapes at a small scale (targeted tests, not a full framework, to start). -- **Rapid DataWriter/DataReader create/delete during active SEDP matching** — plausible - source of use-after-free/leak bugs; matches OpenDDS `EntityLifecycleStress` directly - in spirit. +Landed in `stress-tests/` (`lifecycle_churn` scenarios + `entity_lifecycle_stress`): +- ~~**Generalized reentrant-listener/entity-lifecycle churn**~~ — `--scenario reentrant`. +- ~~**WaitSet/Condition churn under load**~~ — `--scenario waitset` (threads + attach/detach ReadConditions/QueryConditions on a shared WaitSet while a waiter is in + `wait()` and a waker flips a GuardCondition; includes delete-while-attached). +- ~~**Listener-fallback chain under load**~~ — `--scenario listener` (participant + + publisher + subscriber listeners; per-iteration `set_listener` swaps incl. `null` + racing entity teardown and event delivery). Found the unsynchronised `listener_mask` + race, now fixed + TSan-gated. +- ~~**Rapid DataWriter/DataReader create/delete during active SEDP matching**~~ — + `--scenario entities`. Found the discovery-driven listener-dispatch UAF, now fixed. +- **Runtime `set_expression_parameters` reconfiguration** — `--scenario cft`. Found a + UAF between the reconfigure and receive-thread filter eval, now fixed + TSan-gated. +- ~~**Many-writer/many-reader fan-in/fan-out discovery**~~ + ~~**participant-churning + fallback**~~ — `--scenario participants` (N threads each churning a whole participant on + one shared domain, listeners at every level, W/R mix for fan-in/fan-out). Clean. +- ~~**`instance` churn**~~ — `--scenario instance`. Clean for the instance-map / + reader-tracking / register-dispose-unregister paths, but surfaced two pre-existing bugs + it deliberately doesn't gate on: `get_key_value` parses the stored *full* sample with the + *key-only* deserializer in all four zidl backends (wrong key for any type whose key + member isn't first — see `stress-tests/README.md`), and concurrent `write()` on one + `DataWriter` is unsynchronised (`writeRaw` takes no lock). Each needs its own PR. + +Still open: +- A scenario that churns the reader-side WaitSet/condition graph *and* the participant at + once (the closest current pair is `waitset` + `participants` run separately). ### Not prioritized / low value - Condition introspection getters (`get_query_expression`, `get_sample_state_mask`, etc.) diff --git a/docs/design/thread-model.md b/docs/design/thread-model.md index 4e8d86ac..15b1c716 100644 --- a/docs/design/thread-model.md +++ b/docs/design/thread-model.md @@ -25,6 +25,13 @@ would invert locks held by the receive path. `WaitSet.wait()` blocks only the calling thread. +Concurrent `DataWriter.write()` (and the `dispose`/`unregister_instance`/`write_raw` +family) from multiple application threads on a *single* writer is safe: the RTPS +`StatefulWriter`/`StatelessWriter` layer is internally locked, and `DataWriterImpl`'s own +per-write state (`last_sn`, the `get_key_value` key registry) is guarded +(`src/dcps/writer.zig` — atomic `last_sn`, `key_registry_mu`). Ordering between concurrent +writes is not defined — they interleave — but no write is lost or corrupted. + ## Single-Threaded Direction An embedded/single-threaded API such as `DomainParticipant.drive(timeout)` is not diff --git a/docs/roadmap.md b/docs/roadmap.md index e80fe797..882f8460 100644 --- a/docs/roadmap.md +++ b/docs/roadmap.md @@ -89,6 +89,53 @@ Forward-looking only: known gaps, planned features, and open design questions. long-term fix is XTypes TypeLookup (below); a nearer-term option is to require `registerTypeSupport` for keyed topics and error rather than silently degrade. See `implementation_status.md` "Known Limitations" and `design/history-cache.md`. + - *Mitigated for zzdds→zzdds:* a writer whose `TypeSupport.has_key` is set now emits an + inline `PID_KEY_HASH` on **every** sample (RTPS §8.7.9), including a zero-valued key — + previously suppressed as all-zeros — so the subscriber routes by the wire hash and + never reconstructs one from the payload. `resolveKeyHash` also now honours a *present* + all-zero `PID_KEY_HASH` instead of treating it as "recompute". Residual: a non-zzdds + peer that omits the inline hash for an alive keyed sample still falls to + `key_hash_fn`; see the selective-CDR-parse follow-ups below. +- **`key_hash_fn` reconstructs a non-leading `@key` incorrectly** — `resolveKeyHash`'s + fallback (`TypeSupport.compute_key_hash`, whose contract is "full CDR wire payload in, + 16-byte hash out") is served by zidl's generated `computeKeyHashFromCdr`, which runs the + **key-only** deserializer and so only reads a leading, contiguous key correctly. A + non-leading `@key` member (e.g. `Message.subject_id`, member 3) is misread. The + zzdds→zzdds case is fixed (keyed writers now always send inline `PID_KEY_HASH`, so the + fallback isn't reached); the remaining case is a non-zzdds peer that omits the inline + hash for an alive keyed sample. Fix: route the fallback through + `deserialize_selected(KEY_FIELD_MASK)` on a full payload while keeping the key-only path + for genuine DISPOSE/UNREGISTER payloads (distinguish via the DATA submessage K flag). + The selective parser is available (zidl v0.3.12, pinned), but wiring `key_hash_fn` onto + it needs a `TypeSupport.compute_key_hash` signature change (`is_key_only: bool`) + C-ABI + mirror + a further zidl release. + +### Selective CDR parse (`deserialize_selected`) — deferred follow-ups + +zidl v0.3.12 adds a mask-driven selective parser (`deserialize_selected(want)` / +`KEY_FIELD_MASK` / `field_index`, `skipPrimitives` fast path) across all four backends, and +rewires `get_key_value` and `get_field_from_cdr` onto it. Three refinements +were deliberately left out of that change; none is a regression (each is a new capability +or an optimisation on an already-improved path): + +- **Batched one-pass multi-field filter reads** — a CFT/QueryCondition referencing N fields + currently does N `deserialize_selected` walks (one per field reference). Resolve the + filter AST's whole referenced-field set to one `FieldMask` at condition-creation and do a + single walk per sample. Needs a `filter_mod` `FieldAccessor` API change (pull API → cache + the parse, or pre-resolve fields in `eval`). Impact of deferring: a redundant struct walk + per extra referenced field per sample — pure CPU, proportional to filter complexity × rate. +- **Nested field paths (`a.b.c`) in filter expressions** — the DDS filter grammar (DDS 1.4 + Annex A) allows dotted member navigation and `[n]` subscripts; zzdds only supports + top-level simple members (always has — `get_field_from_cdr` / `classifyFilterFieldKind`). + Needs grammar/parser/AST/evaluator changes in `filter_mod` plus a `Spec`-tree form of + `deserialize_selected` for selective descent (the `u64` mask primitive was built to + accept this later without rework). ~1 day with full-decode of wanted nested structs, + ~2 days fully selective. Impact of deferring: a nested field reference silently never + matches (unknown field → `eval` passes the sample) — a pre-existing gap. +- **`computeKeyHashFromCdr` full-payload path** — see the `key_hash_fn` bullet above. The + zzdds→zzdds mitigation has landed; routing the fallback itself through + `deserialize_selected` (for non-zzdds peers) still needs the `compute_key_hash` + signature change + a further zidl release. - **`on_inconsistent_topic` and `on_data_on_readers` have zero firing sites** — the underlying status detection is not wired up. - **`SampleInfo` `sample_rank` / `generation_rank` / `absolute_generation_rank`** stay at @@ -227,11 +274,22 @@ model: replay; `ignore_*` across two processes; runtime `set_expression_parameters` CFT reconfiguration; coherent/ordered grouping atomicity across multiple writers; `_w_timestamp` source-timestamp propagation; `delete_contained_entities` across the C-ABI. -- **Stress tier** — reentrant-listener / entity-lifecycle churn; WaitSet/Condition churn - under load; listener-fallback chain under load; many-participant SPDP/SEDP fan-in/out; - rapid DataWriter/DataReader create/delete during SEDP matching. Plus a loaned-read example - (loan lifecycle has zero C/C++ coverage today — nothing stops a C/C++ caller reading a - returned loan). +- **Stress tier** — landed in `stress-tests/` as seven `lifecycle_churn` scenarios + (`reentrant`, `entities`, `waitset`, `listener`, `cft`, `participants`, `instance`) plus + the `entity_lifecycle_stress` multi-process port. Found + fixed three concurrency bugs + (discovery/teardown UAF, unsynchronised `listener_mask`, CFT param UAF) and + unsynchronised concurrent single-writer `write()`. The `instance` scenario also surfaced + two key-hash correctness bugs: (1) `get_key_value` decoding a full stored sample with the + key-only deserializer (zidl codegen, all four backends) — fixed in zidl via the + selective-parse family; landed here with the zidl v0.3.12 pin, the scenario now asserts + the returned key value; (2) `resolveKeyHash` misroute + for a zero-valued key — mitigated (keyed writers now always send inline `PID_KEY_HASH`; + present all-zero hash honoured), with the `key_hash_fn` full-payload path itself tracked + as a selective-parse follow-up. See `stress-tests/README.md`. Remaining stress ideas: a + scenario that also churns the reader-side WaitSet/condition graph under participant + churn; a Bench-style discovery-latency measurement (explicitly out of scope for this + tier). Plus a loaned-read example (loan lifecycle has zero C/C++ coverage today — + nothing stops a C/C++ caller reading a returned loan). Harness is Python, reusing the examples' `_common.py` pattern. diff --git a/src/c_abi/typesupport.zig b/src/c_abi/typesupport.zig index addc9f86..33515b5b 100644 --- a/src/c_abi/typesupport.zig +++ b/src/c_abi/typesupport.zig @@ -144,6 +144,9 @@ pub export fn zzdds_register_type_support( if (!impl.registerTypeSupport(name, TypeSupport{ .ctx = adapter, .compute_key_hash = CTypeSupportAdapter.computeKeyHash, + // Per this entry point's contract, a keyed type passes its generated + // `_compute_key_hash_from_cdr`; a keyless type passes NULL. + .has_key = compute_key_hash_fn != null, .get_field = if (get_field_fn != null) CTypeSupportAdapter.getField else null, .deinit = CTypeSupportAdapter.deinitAdapter, })) { @@ -259,6 +262,7 @@ pub export fn zzdds_register_type_support_ctx( if (!impl.registerTypeSupport(name, TypeSupport{ .ctx = adapter, .compute_key_hash = CtxTypeSupportAdapter.computeKeyHash, + .has_key = compute_key_hash_fn != null, .get_field = if (get_field_fn != null) CtxTypeSupportAdapter.getField else null, .deinit = CtxTypeSupportAdapter.deinitAdapter, })) { diff --git a/src/dcps/participant.zig b/src/dcps/participant.zig index 027e49a0..2a6cd679 100644 --- a/src/dcps/participant.zig +++ b/src/dcps/participant.zig @@ -691,6 +691,14 @@ pub const TypeSupport = struct { /// `payload` includes the 4-byte encapsulation header (as received from /// the wire). Return `zeroes([16]u8)` for keyless types. compute_key_hash: *const fn (ctx: *anyopaque, payload: []const u8) [16]u8, + /// True when the type has one or more `@key` members. Consulted by + /// `pubCreateProtoWriter` so a keyed writer emits inline `PID_KEY_HASH` on + /// every sample (RTPS §8.7.9), including a zero-valued key — otherwise a + /// subscriber would have to reconstruct the per-instance hash from the + /// payload (see `resolveKeyHash`). Must be registered before the writer is + /// created to take effect, same ordering constraint as `key_hash_fn` on + /// readers. Defaults false; keyless types may leave it unset. + has_key: bool = false, /// Optional: extract a named field value from a raw CDR payload. /// Used to evaluate ContentFilteredTopic expressions at delivery time. /// null = CFT evaluation deferred to the typed DataReader layer. @@ -786,6 +794,10 @@ pub const DomainParticipantImpl = struct { /// Guards `listener_box` swaps/acquires only — never held across a /// dispatch or any other call (see listener_box.zig). listener_mu: Mutex = .{}, + /// See `writer.zig`'s matching field: `dispatchFallback` (the bottom + /// of the s2.2.4.1.5 chain, run from discovery/timer threads) reads it + /// while an application `set_listener` may write it, so both use + /// `@atomicLoad`/`@atomicStore` `.monotonic`; initialisers stay plain. listener_mask: DDS.StatusMask, instance_handle: DDS.InstanceHandle_t, status_changes: DDS.StatusMask, @@ -1514,6 +1526,18 @@ pub const DomainParticipantImpl = struct { .nanosec = qos.lifespan.duration.nanosec, })); + // Keyed writers must send inline PID_KEY_HASH on every sample so the + // subscriber never reconstructs a per-instance hash from the payload + // (RTPS §8.7.9; see resolveKeyHash). Requires the type's TypeSupport to + // be registered before this writer is created — the same ordering + // constraint readers already have for key_hash_fn. + { + self.mu.lock(); + const is_keyed = if (self.type_support_registry.get(type_name)) |ts| ts.has_key else false; + self.mu.unlock(); + adapter.setKeyed(is_keyed); + } + const pw = adapter.toProtocolWriter(); // qos.data_representation.value._buffer, as received here, borrows @@ -1800,7 +1824,11 @@ pub const DomainParticipantImpl = struct { return .alive; } - fn decodeKeyHash(iq: ?submsg_mod.InlineQos) [16]u8 { + /// Decode the inline `PID_KEY_HASH` (RTPS §9.6.4.8), if the writer sent one. + /// Returns `null` when the parameter is absent (or malformed) — distinct from + /// a *present* all-zero hash, which `resolveKeyHash` must honour verbatim + /// rather than treating as "recompute from payload". + fn decodeKeyHash(iq: ?submsg_mod.InlineQos) ?[16]u8 { if (iq) |q| { if (q.get(.key_hash)) |kh| { if (kh.len >= 16) { @@ -1810,7 +1838,7 @@ pub const DomainParticipantImpl = struct { } } } - return std.mem.zeroes([16]u8); + return null; } fn decodeCoherentSetSn(iq: ?submsg_mod.InlineQos, little_endian: bool) ?history_mod.SequenceNumber { @@ -1870,10 +1898,29 @@ pub const DomainParticipantImpl = struct { return null; } - fn resolveKeyHash(kh: [16]u8, ar: *ActiveReader, payload: []const u8) [16]u8 { - if (!std.mem.eql(u8, &kh, &std.mem.zeroes([16]u8))) return kh; + /// Resolve the effective 16-byte key hash for a received change. + /// + /// A *present* inline `PID_KEY_HASH` (`maybe_kh != null`) always wins — even + /// if it is all-zeros. An all-zero hash is a legitimate value (a zero-valued + /// key, or a keyless topic), distinct from the parameter being *absent*; + /// treating "present, all-zero" as "recompute" (the old behaviour) discarded + /// a hash the writer deliberately sent. This is the common path: RTPS §8.7.9 + /// says a keyed writer SHOULD send it, and zzdds's own writer does whenever + /// the computed hash is non-zero. + /// + /// With no inline hash we fall back to the type's `key_hash_fn` + /// (`TypeSupport.compute_key_hash`), whose contract is "full CDR wire + /// payload in, 16-byte hash out". NOTE: zidl's *generated* + /// `computeKeyHashFromCdr` currently honours that contract only for a + /// leading, contiguous key — it runs the key-only deserializer, so a + /// non-leading `@key` member is misread. That is tracked as the + /// selective-parse `key_hash_fn` rework (see `docs/roadmap.md` and the + /// v0.3.12 pin-bump follow-up); it is a zidl-side fix, not fixable here + /// without breaking contract-conforming hand-written TypeSupports. + fn resolveKeyHash(maybe_kh: ?[16]u8, ar: *ActiveReader, payload: []const u8) [16]u8 { + if (maybe_kh) |kh| return kh; if (ar.key_hash_fn) |f| return f(ar.key_hash_ctx, payload); - return kh; + return std.mem.zeroes([16]u8); } fn dispatchDirectedWrite( @@ -1883,7 +1930,7 @@ pub const DomainParticipantImpl = struct { writer_guid: Guid, sn: anytype, ts: time_mod.RtpsTimestamp, - key_hash: [16]u8, + key_hash: ?[16]u8, payload: []const u8, kind: history_mod.ChangeKind, coherent_set_sn: ?history_mod.SequenceNumber, @@ -3745,7 +3792,7 @@ pub const DomainParticipantImpl = struct { ) DDS.ReturnCode_t { const self = cast(ctx); self.swapListener(if (a_listener) |l| l.* else DDS.noop_DomainParticipantListener); - self.listener_mask = mask; + @atomicStore(DDS.StatusMask, &self.listener_mask, mask, .monotonic); return DDS.RETCODE_OK; } @@ -3800,7 +3847,8 @@ pub const DomainParticipantImpl = struct { pub fn dispatchFallback(self: *Self, comptime field: []const u8, bit: DDS.StatusMask, handle: anytype, args: anytype) bool { const box = self.acquireListener(); defer box.releaseRef(self.alloc); - return listener_fallback.tryDispatch(field, self.listener_mask, bit, box.listener, handle, args); + const mask = @atomicLoad(DDS.StatusMask, &self.listener_mask, .monotonic); + return listener_fallback.tryDispatch(field, mask, bit, box.listener, handle, args); } fn vtIgnoreParticipant(ctx: *anyopaque, handle: DDS.InstanceHandle_t) DDS.ReturnCode_t { diff --git a/src/dcps/publisher.zig b/src/dcps/publisher.zig index 4fb70947..c591849f 100644 --- a/src/dcps/publisher.zig +++ b/src/dcps/publisher.zig @@ -119,6 +119,10 @@ pub const PublisherImpl = struct { /// Guards `listener_box` swaps/acquires only — never held across a /// dispatch or any other call (see listener_box.zig). listener_mu: Mutex = .{}, + /// See `writer.zig`'s matching field: `dispatchWriterFallback` (called + /// from a discovery/timer thread) reads it while an application + /// `set_listener` may be writing it, so both go through + /// `@atomicLoad`/`@atomicStore` `.monotonic`; initialisers stay plain. listener_mask: DDS.StatusMask, instance_handle: DDS.InstanceHandle_t, status_changes: DDS.StatusMask, @@ -500,7 +504,7 @@ pub const PublisherImpl = struct { fn vtSetListener(ctx: *anyopaque, a_listener: ?*const DDS.PublisherListener, mask: DDS.StatusMask) DDS.ReturnCode_t { const self = cast(ctx); self.swapListener(if (a_listener) |l| l.* else DDS.noop_PublisherListener); - self.listener_mask = mask; + @atomicStore(DDS.StatusMask, &self.listener_mask, mask, .monotonic); return DDS.RETCODE_OK; } @@ -557,7 +561,8 @@ pub const PublisherImpl = struct { defer self.quiesce.release(self, reallyDeinit); const box = self.acquireListener(); defer box.releaseRef(self.alloc); - if (listener_fallback.tryDispatch(field, self.listener_mask, bit, box.listener, handle, args)) return true; + const mask = @atomicLoad(DDS.StatusMask, &self.listener_mask, .monotonic); + if (listener_fallback.tryDispatch(field, mask, bit, box.listener, handle, args)) return true; if (nil.isNil(self.participant)) return false; const p: *participant_mod.DomainParticipantImpl = @ptrCast(@alignCast(self.participant.ptr)); return p.dispatchFallback(field, bit, handle, args); diff --git a/src/dcps/reader.zig b/src/dcps/reader.zig index 27c9f869..611f4a13 100644 --- a/src/dcps/reader.zig +++ b/src/dcps/reader.zig @@ -232,6 +232,11 @@ pub const DataReaderImpl = struct { /// callback (RTPS receive, timer, discovery) racing `deinit()` — see /// entity_quiesce.zig. quiesce: EntityQuiesce = .{}, + /// See the matching field in `writer.zig`: runtime reads (in + /// `dispatchListener`, and cross-entity in `subscriber.zig`'s + /// `on_data_available` fan-out) and writes (`vtSetListener`) go + /// through `@atomicLoad`/`@atomicStore` `.monotonic`; struct-literal + /// initialisers stay plain. listener_mask: DDS.StatusMask, instance_handle: DDS.InstanceHandle_t, guid: proto.Guid = std.mem.zeroes(proto.Guid), @@ -2958,7 +2963,7 @@ pub const DataReaderImpl = struct { fn vtSetListener(ctx: *anyopaque, a_listener: ?*const DDS.DataReaderListener, mask: DDS.StatusMask) DDS.ReturnCode_t { const self = cast(ctx); self.swapListener(if (a_listener) |l| l.* else DDS.noop_DataReaderListener); - self.listener_mask = mask; + @atomicStore(DDS.StatusMask, &self.listener_mask, mask, .monotonic); // DDS status conditions/listeners are level-triggered, not // edge-triggered: enabling a listener for a status that's *already* @@ -3045,7 +3050,8 @@ pub const DataReaderImpl = struct { pub fn dispatchListener(self: *Self, comptime field: []const u8, bit: DDS.StatusMask, handle: *anyopaque, args: anytype) bool { const box = self.acquireListener(); defer box.releaseRef(self.alloc); - if (listener_fallback.tryDispatch(field, self.listener_mask, bit, box.listener, handle, args)) return true; + const mask = @atomicLoad(DDS.StatusMask, &self.listener_mask, .monotonic); + if (listener_fallback.tryDispatch(field, mask, bit, box.listener, handle, args)) return true; if (nil.isNil(self.subscriber)) return false; const sub: *subscriber_mod.SubscriberImpl = @ptrCast(@alignCast(self.subscriber.ptr)); return sub.dispatchReaderFallback(field, bit, handle, args); diff --git a/src/dcps/subscriber.zig b/src/dcps/subscriber.zig index 9aa0bb06..9c4a3aaa 100644 --- a/src/dcps/subscriber.zig +++ b/src/dcps/subscriber.zig @@ -132,6 +132,12 @@ pub const SubscriberImpl = struct { /// Guards `listener_box` swaps/acquires only — never held across a /// dispatch or any other call (see listener_box.zig). listener_mu: Mutex = .{}, + /// See `writer.zig`'s matching field. Read from discovery/timer + /// threads in `dispatchReaderFallback` and (cross-entity, for the + /// reader's and participant's masks too) in + /// `resolveDataAvailableFallback`, concurrently with an application + /// `set_listener`; all runtime reads/writes are + /// `@atomicLoad`/`@atomicStore` `.monotonic`. Initialisers stay plain. listener_mask: DDS.StatusMask, instance_handle: DDS.InstanceHandle_t, status_changes: DDS.StatusMask, @@ -551,7 +557,7 @@ pub const SubscriberImpl = struct { fn vtSetListener(ctx: *anyopaque, a_listener: ?*const DDS.SubscriberListener, mask: DDS.StatusMask) DDS.ReturnCode_t { const self = cast(ctx); self.swapListener(if (a_listener) |l| l.* else DDS.noop_SubscriberListener); - self.listener_mask = mask; + @atomicStore(DDS.StatusMask, &self.listener_mask, mask, .monotonic); return DDS.RETCODE_OK; } @@ -601,7 +607,8 @@ pub const SubscriberImpl = struct { defer self.quiesce.release(self, reallyDeinit); const box = self.acquireListener(); defer box.releaseRef(self.alloc); - if (listener_fallback.tryDispatch(field, self.listener_mask, bit, box.listener, handle, args)) return true; + const mask = @atomicLoad(DDS.StatusMask, &self.listener_mask, .monotonic); + if (listener_fallback.tryDispatch(field, mask, bit, box.listener, handle, args)) return true; if (nil.isNil(self.participant)) return false; const p: *participant_mod.DomainParticipantImpl = @ptrCast(@alignCast(self.participant.ptr)); return p.dispatchFallback(field, bit, handle, args); @@ -649,13 +656,13 @@ pub const SubscriberImpl = struct { /// `dispatchWriterFallback` doc comment). fn resolveDataAvailableFallback(self: *Self, r: *reader_mod.DataReaderImpl) DataAvailableResolution { const rbox = r.acquireListener(); - if (listener_fallback.peek("on_data_available", r.listener_mask, DDS.DATA_AVAILABLE_STATUS, rbox.listener)) |cb| { + if (listener_fallback.peek("on_data_available", @atomicLoad(DDS.StatusMask, &r.listener_mask, .monotonic), DDS.DATA_AVAILABLE_STATUS, rbox.listener)) |cb| { return .{ .cb = cb, .listener_data = rbox.listener.listener_data, .owner = .{ .reader = .{ .box = rbox, .alloc = r.alloc } } }; } rbox.releaseRef(r.alloc); const sbox = self.acquireListener(); - if (listener_fallback.peek("on_data_available", self.listener_mask, DDS.DATA_AVAILABLE_STATUS, sbox.listener)) |cb| { + if (listener_fallback.peek("on_data_available", @atomicLoad(DDS.StatusMask, &self.listener_mask, .monotonic), DDS.DATA_AVAILABLE_STATUS, sbox.listener)) |cb| { return .{ .cb = cb, .listener_data = sbox.listener.listener_data, .owner = .{ .subscriber = .{ .box = sbox, .alloc = self.alloc } } }; } sbox.releaseRef(self.alloc); @@ -663,7 +670,7 @@ pub const SubscriberImpl = struct { if (nil.isNil(self.participant)) return .{ .cb = null, .listener_data = null, .owner = .none }; const p: *participant_mod.DomainParticipantImpl = @ptrCast(@alignCast(self.participant.ptr)); const pbox = p.acquireListener(); - if (listener_fallback.peek("on_data_available", p.listener_mask, DDS.DATA_AVAILABLE_STATUS, pbox.listener)) |cb| { + if (listener_fallback.peek("on_data_available", @atomicLoad(DDS.StatusMask, &p.listener_mask, .monotonic), DDS.DATA_AVAILABLE_STATUS, pbox.listener)) |cb| { return .{ .cb = cb, .listener_data = pbox.listener.listener_data, .owner = .{ .participant = .{ .box = pbox, .alloc = p.alloc } } }; } pbox.releaseRef(p.alloc); diff --git a/src/dcps/topic.zig b/src/dcps/topic.zig index 4f19d1ca..39c2de67 100644 --- a/src/dcps/topic.zig +++ b/src/dcps/topic.zig @@ -37,6 +37,12 @@ pub const TopicImpl = struct { /// locked mutator, confirmed via TSan there; TopicImpl had no /// general-purpose mutex at all before this). mu: Mutex = .{}, + /// No topic-listener dispatch path reads this today, but it is the same + /// concept as the writer/reader/publisher/subscriber/participant + /// `listener_mask` (all made `@atomicStore`/`@atomicLoad` `.monotonic` + /// after a stress-test TSan finding) -- kept atomic here too so a + /// future `on_inconsistent_topic` fallback path can't reintroduce the + /// race. Initialisers stay plain. listener_mask: DDS.StatusMask, instance_handle: DDS.InstanceHandle_t, status_changes: DDS.StatusMask, @@ -212,7 +218,7 @@ pub const TopicImpl = struct { fn vtSetListener(ctx: *anyopaque, a_listener: ?*const DDS.TopicListener, mask: DDS.StatusMask) DDS.ReturnCode_t { const self = cast(ctx); self.swapListener(if (a_listener) |l| l.* else DDS.noop_TopicListener); - self.listener_mask = mask; + @atomicStore(DDS.StatusMask, &self.listener_mask, mask, .monotonic); return DDS.RETCODE_OK; } @@ -323,6 +329,18 @@ pub const ContentFilteredTopicImpl = struct { name: [:0]u8, // owned, null-terminated for C API filter_expr: [:0]u8, // owned, null-terminated for C API expr_params: std.ArrayListUnmanaged([]u8), // owned copies + /// Guards `expr_params` only. `set_expression_parameters` runs on an + /// application thread and frees + replaces the whole list; `matchSample` + /// runs on the RTPS receive thread that delivers a sample to a reader + /// built on this CFT and reads the list by reference for the duration of + /// `filter_mod.eval`. Without this, a concurrent reconfigure frees the + /// strings mid-evaluation (found by the `lifecycle_churn --scenario cft` + /// stress test: SEGV in `parseFloat` on a freed parameter). Evaluation + /// for one CFT is already serialised by the single receive thread, so a + /// plain mutex (vs. an rwlock) costs nothing in practice; only the rare + /// reconfigure writer ever blocks. `filter_expr`/`parsed_expr` are set + /// once at init and need no guard. + params_lock: Mutex = .{}, related: DDS.Topic, participant: DDS.DomainParticipant, /// Parsed AST of `filter_expr`; null when expression is empty or the @@ -396,9 +414,13 @@ pub const ContentFilteredTopicImpl = struct { /// Returns true if the sample passes the filter (should be delivered). /// An empty expression or disabled profile always returns true. pub fn matchSample( - self: *const Self, + self: *Self, accessor: filter_mod.FieldAccessor, ) bool { + // Held across the whole eval: the AST holds `params_slice` entries + // by reference (see `params_lock`). + self.params_lock.lock(); + defer self.params_lock.unlock(); const params_slice: []const []const u8 = @ptrCast(self.expr_params.items); return filter_mod.eval(self.parsed_expr, accessor, params_slice); } @@ -490,6 +512,8 @@ pub const ContentFilteredTopicImpl = struct { fn cftGetParams(ctx: *anyopaque, out: ?*DDS.StringSeq) DDS.ReturnCode_t { const seq = out orelse return DDS.RETCODE_BAD_PARAMETER; const self = cast(ctx); + self.params_lock.lock(); + defer self.params_lock.unlock(); if (seq._release) { if (seq._buffer) |b| { for (b[0..seq._length]) |s| self.alloc.free(std.mem.span(s)); @@ -519,6 +543,8 @@ pub const ContentFilteredTopicImpl = struct { // Build into a temporary list first so the old params survive any OOM. var tmp: std.ArrayListUnmanaged([]u8) = .empty; const seq = params orelse { + self.params_lock.lock(); + defer self.params_lock.unlock(); for (self.expr_params.items) |p| self.alloc.free(p); self.expr_params.clearRetainingCapacity(); return DDS.RETCODE_OK; @@ -538,7 +564,10 @@ pub const ContentFilteredTopicImpl = struct { }; } } - // All copies succeeded — swap in and free old. + // All copies succeeded — swap in and free old under the exclusive + // lock so no receive thread is mid-eval on the outgoing strings. + self.params_lock.lock(); + defer self.params_lock.unlock(); for (self.expr_params.items) |p| self.alloc.free(p); self.expr_params.deinit(self.alloc); self.expr_params = tmp; diff --git a/src/dcps/writer.zig b/src/dcps/writer.zig index 5e5e9317..a39ebe53 100644 --- a/src/dcps/writer.zig +++ b/src/dcps/writer.zig @@ -100,6 +100,14 @@ pub const DataWriterImpl = struct { /// refcount -- but it's still atomic since publish/cancel isn't /// guaranteed to happen under any particular lock. outstanding_loans: std.atomic.Value(usize) = .init(0), + /// Set once at init, then only via `vtSetListener`/`setListenerEx`. + /// A discovery- or timer-thread dispatch can read it (in + /// `dispatchListener`) concurrently with an application `set_listener` + /// on another thread, so every *runtime* read/write goes through + /// `@atomicLoad`/`@atomicStore` (`.monotonic` -- the box it gates is + /// separately synchronised by `listener_mu` + the ListenerBox + /// refcount). Struct-literal initialisers touch it single-threaded, + /// before publication, and stay plain. listener_mask: DDS.StatusMask, instance_handle: DDS.InstanceHandle_t, guid: proto.Guid = std.mem.zeroes(proto.Guid), @@ -107,7 +115,12 @@ pub const DataWriterImpl = struct { status_cond: ?*waitset.StatusConditionImpl, /// Sequence number of the most recently written sample; 0 = nothing written. - last_sn: proto.SequenceNumber = 0, + /// `writeRaw` (application thread, possibly several -- concurrent `write()` + /// on one DataWriter is spec-legal) publishes it; `wait_for_acknowledgments` + /// and `waitForAcks` read it from other threads. Plain `.monotonic` -- + /// there is no ordering dependency on other writer state, only tearing to + /// avoid. + last_sn: std.atomic.Value(proto.SequenceNumber) = .init(0), /// Cumulative count of incompatible-QoS events. incompat_total_change/ /// incompat_last_policy are now guarded by `mu` (see its doc comment -- @@ -172,6 +185,17 @@ pub const DataWriterImpl = struct { /// Maps instance_handle → last alive CDR payload for get_key_value support. /// Populated on the first alive write per instance; never overwritten. key_registry: std.AutoHashMapUnmanaged(DDS.InstanceHandle_t, []u8) = .empty, + /// Guards `key_registry` only. `writeRaw` mutates it (insert-only) from + /// the application thread(s) -- concurrent `write()` on one DataWriter is + /// spec-legal, and the RTPS layer (`proto_writer`) is already internally + /// locked, so this map plus `last_sn` were the last unsynchronised writer + /// state on the hot path (found by the stress `instance` scenario under + /// TSan: a `HashMapUnmanaged` grow racing a concurrent insert). A + /// dedicated leaf mutex, not `mu` -- `mu`'s contract is "plain counter + /// reads/writes, never held across a callback", and `writeRaw` takes no + /// other lock. Entries are inserted once and never replaced, so a `[]u8` + /// value handed out by `getKeyValueRaw` stays valid after unlock. + key_registry_mu: Mutex = .{}, /// One box for the whole object, shared across every interface view /// (DataWriter, Entity, and ZZDDS.DataWriter — see src/c_abi/extensions.zig) @@ -308,9 +332,11 @@ pub const DataWriterImpl = struct { } const sn = try self.proto_writer.write(kind, source_timestamp, instance_handle, key_hash, data); - self.last_sn = sn; + self.last_sn.store(sn, .monotonic); if (kind == .alive) { const ih = keyHashToHandle(key_hash); + self.key_registry_mu.lock(); + defer self.key_registry_mu.unlock(); if (!self.key_registry.contains(ih)) { const stored = self.alloc.dupe(u8, data) catch null; if (stored) |s| { @@ -398,7 +424,7 @@ pub const DataWriterImpl = struct { /// Returns true when all RELIABLE matched readers have acked up to last_sn. pub fn allAcked(self: *Self) bool { - return self.proto_writer.allAcked(self.last_sn); + return self.proto_writer.allAcked(self.last_sn.load(.monotonic)); } pub fn matchedReaderCount(self: *Self) usize { @@ -413,8 +439,13 @@ pub const DataWriterImpl = struct { /// Return the stored CDR payload for the given instance handle, or null if /// no alive write has been made for this instance. - /// The returned slice is valid until the next write to this writer. + /// The returned slice stays valid for the life of the writer: entries are + /// inserted once and never replaced (see `key_registry_mu`). The lock only + /// guards against a concurrent `writeRaw` insert reallocating the map's + /// index while we look up. pub fn getKeyValueRaw(self: *Self, handle: DDS.InstanceHandle_t) ?[]u8 { + self.key_registry_mu.lock(); + defer self.key_registry_mu.unlock(); return self.key_registry.get(handle); } @@ -606,7 +637,7 @@ pub const DataWriterImpl = struct { /// Backs `zzdds::DataWriter::set_listener_ex` (see src/c_abi/extensions.zig). pub fn setListenerEx(self: *Self, listener_ex: ZZDDS.DataWriterListenerEx, mask: DDS.StatusMask) void { self.swapListenerEx(listener_ex); - self.listener_mask = mask; + @atomicStore(DDS.StatusMask, &self.listener_mask, mask, .monotonic); } /// Installs `new_listener_ex`, releasing whatever it replaces. Safe @@ -652,7 +683,8 @@ pub const DataWriterImpl = struct { fn dispatchListener(self: *Self, comptime field: []const u8, bit: DDS.StatusMask, handle: *anyopaque, args: anytype) bool { const box = self.acquireListenerEx(); defer box.releaseRef(self.alloc); - if (listener_fallback.tryDispatch(field, self.listener_mask, bit, box.listener, handle, args)) return true; + const mask = @atomicLoad(DDS.StatusMask, &self.listener_mask, .monotonic); + if (listener_fallback.tryDispatch(field, mask, bit, box.listener, handle, args)) return true; if (nil.isNil(self.publisher)) return false; const pub_: *publisher_mod.PublisherImpl = @ptrCast(@alignCast(self.publisher.ptr)); return pub_.dispatchWriterFallback(field, bit, handle, args); @@ -841,7 +873,7 @@ pub const DataWriterImpl = struct { fn vtSetListener(ctx: *anyopaque, a_listener: ?*const DDS.DataWriterListener, mask: DDS.StatusMask) DDS.ReturnCode_t { const self = cast(ctx); self.swapListenerEx(listenerExFromBase(if (a_listener) |l| l.* else DDS.noop_DataWriterListener)); - self.listener_mask = mask; + @atomicStore(DDS.StatusMask, &self.listener_mask, mask, .monotonic); return DDS.RETCODE_OK; } @@ -863,7 +895,8 @@ pub const DataWriterImpl = struct { fn vtWaitForAck(ctx: *anyopaque, timeout: *const DDS.Duration_t) DDS.ReturnCode_t { const self = cast(ctx); if (self.qos.reliability.kind == .BEST_EFFORT_RELIABILITY_QOS) return DDS.RETCODE_OK; - if (self.last_sn == 0) return DDS.RETCODE_OK; + const last_sn = self.last_sn.load(.monotonic); + if (last_sn == 0) return DDS.RETCODE_OK; const deadline_ns: ?i64 = if (timeout.sec == DDS.DURATION_INFINITE_SEC and timeout.nanosec == DDS.DURATION_INFINITE_NSEC) null @@ -873,7 +906,7 @@ pub const DataWriterImpl = struct { @as(i64, timeout.sec) * std.time.ns_per_s + @as(i64, @intCast(timeout.nanosec)); }; - return if (self.proto_writer.waitAllAcked(self.last_sn, deadline_ns)) + return if (self.proto_writer.waitAllAcked(last_sn, deadline_ns)) DDS.RETCODE_OK else DDS.RETCODE_TIMEOUT; diff --git a/src/rtps/protocol_adapters.zig b/src/rtps/protocol_adapters.zig index 14ed73e7..b957bd18 100644 --- a/src/rtps/protocol_adapters.zig +++ b/src/rtps/protocol_adapters.zig @@ -96,6 +96,10 @@ pub const RtpsProtocolWriter = struct { self.writer.setLifespan(ls); } + pub fn setKeyed(self: *Self, keyed: bool) void { + self.writer.setKeyed(keyed); + } + pub fn toProtocolWriter(self: *Self) ProtocolWriter { return .{ .ctx = self, .vtable = &vtable }; } diff --git a/src/rtps/writer_sm.zig b/src/rtps/writer_sm.zig index 89cab7c9..eafaa2d1 100644 --- a/src/rtps/writer_sm.zig +++ b/src/rtps/writer_sm.zig @@ -49,6 +49,17 @@ fn statusInfoFromKind(kind: ChangeKind) ?u32 { }; } +/// The inline `PID_KEY_HASH` value to place on a DATA / DATA_FRAG (fragment 1), +/// or `null` to omit the parameter. A keyed writer always sends it — RTPS §8.7.9 +/// SHOULD, and it spares the subscriber from reconstructing a per-instance hash +/// from the payload — even when the hash is all-zeros (a zero-valued key). A +/// non-keyed writer only ever carries a non-zero hash (from a built-in endpoint +/// keyed on its GUID); a plain keyless user topic omits it. +fn inlineKeyHash(keyed: bool, kh: [16]u8) ?[16]u8 { + if (keyed or !std.mem.eql(u8, &kh, &std.mem.zeroes([16]u8))) return kh; + return null; +} + // ── Message send helper ─────────────────────────────────────────────────────── /// Maximum flat buffer size for a single send. Each DATA_FRAG fragment is @@ -446,6 +457,14 @@ pub const StatefulWriter = struct { /// via SEDP writer announcement) — some readers only apply lifespan-based expiry /// to samples that carry it inline. null = no lifespan configured. lifespan: ?time_mod.RtpsDuration, + /// True when this writer's topic type is keyed. Keyed writers emit inline + /// PID_KEY_HASH on *every* DATA/DATA_FRAG (RTPS §8.7.9 SHOULD) — including + /// when the hash is all-zeros (a zero-valued key), so a subscriber never has + /// to reconstruct a per-instance hash from the payload. Non-keyed writers + /// leave it false and only ever send a non-zero hash (they never have one). + /// Set once at creation via `setKeyed`, before any reader match; read + /// without the lock, like `guid`. + keyed: bool, const Self = @This(); @@ -489,10 +508,17 @@ pub const StatefulWriter = struct { .protocol_ready_fn = null, .protocol_ready_ctx = null, .lifespan = null, + .keyed = false, }; return self; } + /// Mark this writer's topic type as keyed. Call once, right after `init`, + /// before any reader proxy is added. + pub fn setKeyed(self: *Self, keyed: bool) void { + self.keyed = keyed; + } + pub fn setLifespan(self: *Self, ls: ?time_mod.RtpsDuration) void { self.mu.lock(); defer self.mu.unlock(); @@ -870,10 +896,7 @@ pub const StatefulWriter = struct { .reader_entity_id = rp.guid.entity_id, .writer_entity_id = self.guid.entity_id, .writer_sn = ch.sequence_number, - .key_hash = if (!std.mem.eql(u8, &ch.key_hash, &std.mem.zeroes([16]u8))) - ch.key_hash - else - null, + .key_hash = inlineKeyHash(self.keyed, ch.key_hash), .is_key = ch.kind != .alive, .status_info = statusInfoFromKind(ch.kind), .lifespan = if (ch.kind == .alive) self.lifespan else null, @@ -1013,6 +1036,7 @@ pub const StatefulWriter = struct { const self_guid = self.guid; const self_lifespan = self.lifespan; const self_transport = self.transport; + const self_keyed = self.keyed; self.mu.unlock(); defer self.mu.lock(); @@ -1026,10 +1050,7 @@ pub const StatefulWriter = struct { .reader_entity_id = rp_guid.entity_id, .writer_entity_id = self_guid.entity_id, .writer_sn = c.sn, - .key_hash = if (!std.mem.eql(u8, &c.key_hash, &std.mem.zeroes([16]u8))) - c.key_hash - else - null, + .key_hash = inlineKeyHash(self_keyed, c.key_hash), .is_key = c.kind != .alive, .status_info = statusInfoFromKind(c.kind), .lifespan = if (c.kind == .alive) self_lifespan else null, @@ -1478,10 +1499,7 @@ pub const StatefulWriter = struct { .reader_entity_id = proxy.guid.entity_id, .writer_entity_id = w.guid.entity_id, .writer_sn = ch.sequence_number, - .key_hash = if (!std.mem.eql(u8, &ch.key_hash, &std.mem.zeroes([16]u8))) - ch.key_hash - else - null, + .key_hash = inlineKeyHash(w.keyed, ch.key_hash), .is_key = ch.kind != .alive, .status_info = statusInfoFromKind(ch.kind), .coherent_set_sn = ch.coherent_set_sn, @@ -1737,6 +1755,7 @@ pub const StatefulWriter = struct { const self_guid = self.guid; const self_lifespan = self.lifespan; const self_transport = self.transport; + const self_keyed = self.keyed; self.mu.unlock(); @@ -1749,10 +1768,7 @@ pub const StatefulWriter = struct { .reader_entity_id = s.guid.entity_id, .writer_entity_id = self_guid.entity_id, .writer_sn = ch_sn, - .key_hash = if (!std.mem.eql(u8, &ch_key_hash, &std.mem.zeroes([16]u8))) - ch_key_hash - else - null, + .key_hash = inlineKeyHash(self_keyed, ch_key_hash), .is_key = ch_kind != .alive, .status_info = statusInfoFromKind(ch_kind), .coherent_set_sn = ch_coherent_set_sn, @@ -1888,10 +1904,7 @@ pub const StatefulWriter = struct { .fragments_in_submessage = 1, .fragment_size = @intCast(frag_size), .data_size = data_size, - .key_hash = if (frag_num == 1 and !std.mem.eql(u8, &ch.key_hash, &std.mem.zeroes([16]u8))) - ch.key_hash - else - null, + .key_hash = if (frag_num == 1) inlineKeyHash(self.keyed, ch.key_hash) else null, .status_info = if (frag_num == 1) statusInfoFromKind(ch.kind) else null, .coherent_set_sn = if (frag_num == 1) ch.coherent_set_sn else null, .group_seq_num = if (frag_num == 1) ch.group_seq_num else null, @@ -1940,10 +1953,7 @@ pub const StatefulWriter = struct { .fragments_in_submessage = 1, .fragment_size = @intCast(frag_size), .data_size = @intCast(ch.data.len), - .key_hash = if (!std.mem.eql(u8, &ch.key_hash, &std.mem.zeroes([16]u8))) - ch.key_hash - else - null, + .key_hash = inlineKeyHash(self.keyed, ch.key_hash), .status_info = statusInfoFromKind(ch.kind), .coherent_set_sn = ch.coherent_set_sn, .group_seq_num = ch.group_seq_num, @@ -2013,10 +2023,7 @@ pub const StatefulWriter = struct { .fragments_in_submessage = 1, .fragment_size = @intCast(frag_size), .data_size = data_size, - .key_hash = if (frag_num == 1 and !std.mem.eql(u8, &ch.key_hash, &std.mem.zeroes([16]u8))) - ch.key_hash - else - null, + .key_hash = if (frag_num == 1) inlineKeyHash(self.keyed, ch.key_hash) else null, .status_info = if (frag_num == 1) statusInfoFromKind(ch.kind) else null, .coherent_set_sn = if (frag_num == 1) ch.coherent_set_sn else null, .group_seq_num = if (frag_num == 1) ch.group_seq_num else null, diff --git a/stress-tests/README.md b/stress-tests/README.md index 28c41240..a874733e 100644 --- a/stress-tests/README.md +++ b/stress-tests/README.md @@ -107,13 +107,32 @@ timer-driven listener"). `--scenario`: - **`reentrant`** — the seed pattern run N-wide: a DEADLINE listener, fired from the participant's own timer thread, reentrantly deletes its whole entity graph including the participant. (audit: *"generalized reentrant-listener / entity-lifecycle churn"*) - -Planned follow-on scenarios (frame is built for them; not in this first pass): -`waitset` (threads attach/detach conditions on a shared WaitSet while another blocks in -`wait()`), `listener` (concurrent `set_listener()` replacement racing entity deletion and -event delivery), `instance` (`register_instance` / `unregister_instance` / `dispose` / -`get_key_value` / `lookup_instance` churn), `cft` (ContentFilteredTopic / QueryCondition / -`set_expression_parameters` churn). +- **`waitset`** — one shared reader + WaitSet: a waiter thread parked in `wait()`, a + waker thread flipping a shared `GuardCondition`, and N threads creating / attaching / + detaching / deleting `ReadCondition`s and `QueryCondition`s on that WaitSet — including + a deliberate delete-while-still-attached each 16th cycle. (audit: *"WaitSet/Condition + churn under load"*) +- **`listener`** — listeners installed at participant + publisher + subscriber level (the + full DDS 1.4 §2.2.4.1.5 fallback chain); N threads create a `DataWriter` + `DataReader` + with their own listeners, swap those listeners (including to `null`), write, then delete + both while matched / removed events are still in flight. (audit: *"listener-fallback + chain under load"*) +- **`cft`** — a writer streams samples across a range of a numeric field; N threads churn + `ContentFilteredTopic` + reader lifecycle (unique names) while also hammering + `set_expression_parameters()` on one shared long-lived CFT whose reader is being drained + concurrently. (audit: *"runtime `set_expression_parameters` CFT reconfiguration … fully + untested"*) +- **`participants`** — N threads each run a whole participant lifecycle (factory → + participant → pub+writer or sub+reader, listeners at every level → sample exchange → + `delete_participant`) on one shared domain, so ~N participants are always concurrently + joining / matching / leaving with writer/reader fan-in/fan-out. Deleting a participant + mid-match drives events onto a graph whose participant is also tearing down. (audit: + *"many-participant SPDP/SEDP fan-in/fan-out"* + *"participant-churning listener + fallback"* — covers both) +- **`instance`** — each thread owns a Publisher + DataWriter; all churn + `register_instance` / `write` / `dispose` / `unregister_instance` / `get_key_value` / + `lookup_instance` across a small keyspace while one shared reader + drainer fans in. + (audit: *"instance introspection / lifecycle churn"*) --- @@ -123,7 +142,8 @@ event delivery), `instance` (`register_instance` / `unregister_instance` / `disp native + `-Ddebug-allocator`, then `stress-tests/run_all.py --strict` with **CI-sized** parameters (small N, short durations) and a hard `timeout-minutes`. The weekly `schedule` trigger runs a heavier matrix (larger N, longer runs, `--large`). Following -`examples-tsan`, a ThreadSanitizer variant of `lifecycle_churn` runs in that lane. +`examples-tsan`, ThreadSanitizer variants of `lifecycle_churn` run in that lane — +`reentrant`, plus `listener` and `cft` (each pins a data-race fix, see Findings). ## Findings @@ -169,6 +189,78 @@ Participant-level fallback has the same shape but the `entities` scenario keeps participant stable, so it is untested here — a `--scenario` that also churns participants would be the way to exercise it. +### `lifecycle_churn --scenario listener` — unsynchronised `listener_mask` (found + fixed 2026-08-30) + +Clean under `-Ddebug-allocator` from the first run, but TSan flagged a data race in +`DataWriterImpl.dispatchListener` / `DataReaderImpl.dispatchListener`: `listener_mask` is +a plain `u32` written unlocked by `set_listener` (application thread) and read unlocked by +the discovery/timer-thread dispatch path. `listener_mu` guards the `ListenerBox` swap but +never covered the mask word beside it. The same shape was present in all five entities +that carry a listener (`writer`, `reader`, `publisher`, `subscriber`, `participant`) plus +the currently-dormant one on `topic`. + +**Fix** (`src/dcps/{writer,reader,publisher,subscriber,participant,topic}.zig`): every +*runtime* read/write of `listener_mask` goes through `@atomicLoad` / `@atomicStore` +`.monotonic` (the box it gates stays separately synchronised by `listener_mu` + the +ListenerBox refcount; struct-literal initialisers are single-threaded and stay plain). +Mirrors the earlier `incompat_total` atomic fix. + +### `lifecycle_churn --scenario cft` — UAF in `set_expression_parameters` vs. filter eval (found + fixed 2026-08-30) + +~40% repro at 12 threads: SEGV in `std.fmt.parseFloat` on a freed string, reached from +the UDP receive thread's `ContentFilteredTopicImpl.matchSample` → +`filter_mod.eval(…, params_slice)`. `ContentFilteredTopicImpl` had **no synchronisation** +on `expr_params`: `set_expression_parameters` (application thread) frees every old +parameter string and the backing array, then swaps in the new list, while `matchSample` +(receive thread) is mid-`eval` holding those same strings by reference. This is the +runtime CFT reconfiguration path the API audit flagged as fully untested. + +**Fix** (`src/dcps/topic.zig`): a `params_lock: Mutex` on `ContentFilteredTopicImpl` +guarding `expr_params` — held (shared-style, but a plain mutex: eval for one CFT is +already serialised by the single receive thread) across the whole of `matchSample`'s +`eval`, and exclusively around the free-old / swap-in step of `set_expression_parameters` +and the read in `get_expression_parameters`. + +### `lifecycle_churn --scenario instance` — three pre-existing bugs surfaced + +The `instance` scenario gives each churn thread its **own** writer and round-trips +`register_instance` / `write` / `dispose` / `unregister_instance` / `get_key_value` / +`lookup_instance` against a shared fan-in reader. Building it turned up three bugs bigger +than a stress-suite fix: + +1. **`get_key_value` decoded the wrong key for a non-leading `@key` member — FIXED in + zidl (v0.3.12).** `zzdds_get_key_value_{writer,reader}` returns the stored *full* + last-alive sample payload (`key_registry` / `key_cdr` both `dupe` the whole `data`), but + every backend's generated `get_key_value` parsed it with the *key-only* deserializer + (`{Type}_deserialize_key` / `deserializeKeyInto`), which expects a stream that starts at + the key member. For `Message` (key `subject_id` is the 3rd field) it read a preceding + field's length prefix as the key. Fixed in zidl by a **selective-parse family** — `{Type} + .deserialize_selected(reader, KEY_FIELD_MASK, out)` decodes just the `@key` members and + skips the rest — across all four backends, plus a `skipPrimitives` fast path so a large + non-key member is stepped over rather than decoded (zidl PR #47). Pinned here via + `build.zig.zon` → `zidl v0.3.12-zig.0.16.0`; the scenario now asserts `get_key_value`'s + returned `subject_id` on both the writer and reader sides (10 threads, 6 s, plus TSan — + clean). +2. **`resolveKeyHash` misrouted a zero-valued key — FIXED.** When a write carries no + inline `PID_KEY_HASH` the reader falls back to the type's `key_hash_fn`, and zzdds's + writer omitted the inline hash exactly when it was all-zero bytes — i.e. a legitimate + `subject_id == 0`. The fallback (`computeKeyHashFromCdr`) has the same + key-only-on-a-full-sample shape, so a zero-valued non-leading key was misrouted. Fixed + (`CHANGELOG.md` 2026-09-02): a keyed writer (`TypeSupport.has_key`) now sends the inline + `PID_KEY_HASH` for every sample including an all-zero key, and `resolveKeyHash` honours a + present all-zero hash. The scenario's keyspace includes `0`. The `key_hash_fn` + full-payload path itself (for a non-zzdds peer that omits the inline hash for an alive + keyed sample) is still key-only-shaped — tracked in `docs/roadmap.md` "Selective CDR + parse — deferred follow-ups"; it needs a `TypeSupport.compute_key_hash` signature change + + a new zidl release, so it is not part of the v0.3.12 bump. +3. **Concurrent `write()` on one `DataWriter` was unsynchronised — FIXED** (see the + "concurrent DataWriter.write" entry in `CHANGELOG.md`). `DataWriterImpl.writeRaw` mutated + `last_sn` and the `key_registry` `HashMapUnmanaged` with no lock; N threads on one writer + raced the map's grow/insert (TSan) and could abort on its `SafetyLock`. Now `last_sn` is + a `std.atomic.Value` and the registry is guarded by a dedicated `key_registry_mu`. + Regression: `test/dcps/writer_vtable_test.zig`. The scenario still uses a writer per + thread (that shared-writer path is covered by the unit regression now). + ## Non-goals Not a Bench-style configurable framework. Not a perf / throughput / latency benchmark. diff --git a/stress-tests/run_all.py b/stress-tests/run_all.py index 0b0e505c..e9462125 100755 --- a/stress-tests/run_all.py +++ b/stress-tests/run_all.py @@ -47,6 +47,36 @@ ["--scenario", "entities", "--threads", "8", "--duration", "8"], False, ), + ( + "lifecycle_churn/waitset", + SCRIPT_DIR / "zig" / "lifecycle_churn" / "run.py", + ["--scenario", "waitset", "--threads", "8", "--duration", "8"], + False, + ), + ( + "lifecycle_churn/listener", + SCRIPT_DIR / "zig" / "lifecycle_churn" / "run.py", + ["--scenario", "listener", "--threads", "8", "--duration", "8"], + False, + ), + ( + "lifecycle_churn/cft", + SCRIPT_DIR / "zig" / "lifecycle_churn" / "run.py", + ["--scenario", "cft", "--threads", "10", "--duration", "8"], + False, + ), + ( + "lifecycle_churn/participants", + SCRIPT_DIR / "zig" / "lifecycle_churn" / "run.py", + ["--scenario", "participants", "--threads", "8", "--duration", "8"], + False, + ), + ( + "lifecycle_churn/instance", + SCRIPT_DIR / "zig" / "lifecycle_churn" / "run.py", + ["--scenario", "instance", "--threads", "8", "--duration", "8"], + False, + ), ] diff --git a/stress-tests/zig/entity_lifecycle_stress/main.zig b/stress-tests/zig/entity_lifecycle_stress/main.zig index 3a1f8a8a..d4afe8d3 100644 --- a/stress-tests/zig/entity_lifecycle_stress/main.zig +++ b/stress-tests/zig/entity_lifecycle_stress/main.zig @@ -193,6 +193,7 @@ fn commonSetup( if (!zzdds.registerTypeSupport(dp, TYPE_NAME, .{ .ctx = @ptrCast(ts_alloc), .compute_key_hash = gen.Message.computeKeyHashFromCdr, + .has_key = gen.Message.has_key, })) { std.debug.print("FAIL: registerTypeSupport() failed\n", .{}); std.process.exit(1); diff --git a/stress-tests/zig/lifecycle_churn/main.zig b/stress-tests/zig/lifecycle_churn/main.zig index 8d51312c..42e2209a 100644 --- a/stress-tests/zig/lifecycle_churn/main.zig +++ b/stress-tests/zig/lifecycle_churn/main.zig @@ -16,6 +16,35 @@ //! publisher + datawriter whose DEADLINE listener (fired //! from the participant's own timer thread) reentrantly //! deletes the whole graph including the participant. +//! waitset -- one shared reader + WaitSet; a waiter thread parked in +//! wait(), a waker thread flipping a GuardCondition, and N +//! threads creating/attaching/detaching/deleting Read- and +//! QueryConditions on that WaitSet for --duration seconds. +//! listener -- listeners installed at participant + publisher + +//! subscriber level (the full DDS 1.4 s2.2.4.1.5 fallback +//! chain); N threads create a DataWriter+DataReader with +//! their own listeners, swap those listeners (incl. to +//! null), write, then delete both while matched/removed +//! events are still in flight -- the load regression for +//! the #77 discovery/teardown UAF fix. +//! cft -- a writer streams samples across a range of `count`; N +//! threads churn ContentFilteredTopic + reader lifecycle +//! (unique names) while also hammering +//! set_expression_parameters() on one shared long-lived +//! CFT whose reader is being drained concurrently -- the +//! runtime CFT-reconfiguration path the API audit flags as +//! untested. +//! participants -- N threads each stand up a whole participant (factory -> +//! participant -> pub+writer or sub+reader, listeners at +//! every level), exchange a few samples on one shared +//! domain, then delete the participant while peers are +//! still (un)matching. Concurrent SPDP/SEDP join/leave with +//! writer/reader fan-in/fan-out, and the participant level +//! of the s2.2.4.1.5 fallback chain under teardown. +//! instance -- one shared writer + reader; N threads churn +//! register_instance / write / dispose / unregister_instance +//! / get_key_value / lookup_instance across a small keyspace +//! while a drainer empties the reader. //! //! Ends with `SUMMARY: OK scenario= ops=` or a `FAIL:`/panic + exit 1. @@ -33,7 +62,19 @@ fn monoNs(io: std.Io) i64 { return @intCast(std.Io.Clock.awake.now(io).nanoseconds); } -const Scenario = enum { entities, reentrant }; +const Scenario = enum { entities, reentrant, waitset, listener, cft, participants, instance }; + +// zidl's Zig backend emits only the `as_{Base}` vtable slot, not a +// convenience wrapper method (see examples/zig/waitset/subscriber.zig). +fn gcAsCond(gc: DDS.GuardCondition) DDS.Condition { + return gc.vtable.as_Condition(gc.ptr); +} +fn rcAsCond(rc: DDS.ReadCondition) DDS.Condition { + return rc.vtable.as_Condition(rc.ptr); +} +fn qcAsRc(qc: DDS.QueryCondition) DDS.ReadCondition { + return qc.vtable.as_ReadCondition(qc.ptr); +} const Config = struct { scenario: Scenario, @@ -46,10 +87,10 @@ const Config = struct { fn usage() noreturn { std.debug.print( - \\usage: churn_stress --scenario entities|reentrant [options] + \\usage: churn_stress --scenario entities|reentrant|waitset|listener|cft|participants|instance [options] \\ --threads N concurrent churn threads (default 6) \\ --iterations N reentrant: cycles per thread (default 40) - \\ --duration N entities: seconds to churn (default 8) + \\ --duration N entities/waitset/listener/cft/participants/instance: seconds (default 8) \\ --domain N DDS domain id (default 71) \\ --seed N RNG seed \\ @@ -65,7 +106,7 @@ fn parseArgs(raw: std.process.Args) Config { while (it.next()) |a| { if (std.mem.eql(u8, a, "--scenario")) { const v = it.next() orelse usage(); - scenario = if (std.mem.eql(u8, v, "entities")) .entities else if (std.mem.eql(u8, v, "reentrant")) .reentrant else usage(); + scenario = if (std.mem.eql(u8, v, "entities")) .entities else if (std.mem.eql(u8, v, "reentrant")) .reentrant else if (std.mem.eql(u8, v, "waitset")) .waitset else if (std.mem.eql(u8, v, "listener")) .listener else if (std.mem.eql(u8, v, "cft")) .cft else if (std.mem.eql(u8, v, "participants")) .participants else if (std.mem.eql(u8, v, "instance")) .instance else usage(); } else if (std.mem.eql(u8, a, "--threads")) { cfg.threads = std.fmt.parseInt(u32, it.next() orelse usage(), 10) catch usage(); } else if (std.mem.eql(u8, a, "--iterations")) { @@ -114,6 +155,11 @@ pub fn main(init: std.process.Init) !void { switch (cfg.scenario) { .entities => try runEntities(io, cfg), .reentrant => try runReentrant(io, cfg), + .waitset => try runWaitset(io, cfg), + .listener => try runListener(io, cfg), + .cft => try runCft(io, cfg), + .participants => try runParticipants(io, cfg), + .instance => try runInstance(io, cfg), } if (g_fail.load(.acquire)) std.process.exit(1); @@ -186,6 +232,7 @@ fn runEntities(io: std.Io, cfg: Config) !void { if (!zzdds.registerTypeSupport(dp, TYPE_NAME, .{ .ctx = @ptrCast(&g_ts_alloc), .compute_key_hash = gen.Message.computeKeyHashFromCdr, + .has_key = gen.Message.has_key, })) { fail("registerTypeSupport"); return; @@ -299,3 +346,875 @@ fn runReentrant(io: std.Io, cfg: Config) !void { for (threads) |*t| t.* = try std.Thread.spawn(.{}, reentrantThread, .{ io, cfg }); for (threads) |t| t.join(); } + +// ── scenario: waitset ────────────────────────────────────────────────────── +// +// One shared reader + WaitSet + a long-lived GuardCondition. A `waiter` +// thread is parked in ws.wait() for the whole run; a `waker` thread flips +// the GuardCondition's trigger to keep pulling it out of wait(); N churn +// threads create ReadCondition / QueryCondition on the shared reader, then +// attach_condition -> (brief) -> detach_condition -> delete_readcondition, +// hammering the WaitSet's attached-condition list concurrently with +// wait() and get_conditions(). Every 16th cycle a churn thread deletes a +// condition *without* detaching it first -- the dangling-attached-condition +// hazard the zig/waitset example's publisher comment calls out; the WaitSet +// must not later dereference the freed condition from wait() or deinit(). + +const WaitsetCtx = struct { + io: std.Io, + ws: DDS.WaitSet, + dr: DDS.DataReader, + guard: DDS.GuardCondition, + deadline_ns: i64, + idx: u32, +}; + +fn freeCondSeq(seq: *DDS.ConditionSeq) void { + if (seq._release) { + if (seq._buffer) |b| g_alloc.free(b[0..seq._maximum]); + } +} + +fn waitsetWaiter(ctx: WaitsetCtx) void { + const step: DDS.Duration_t = .{ .sec = 0, .nanosec = 50 * std.time.ns_per_ms }; + var local: u64 = 0; + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + var active = DDS.ConditionSeq{}; + const wr = ctx.ws.wait(&active, step); + freeCondSeq(&active); + if (wr != DDS.RETCODE_OK and wr != DDS.RETCODE_TIMEOUT) { + fail("waitset: wait() returned an unexpected retcode"); + return; + } + local += 1; + } + _ = g_ops.fetchAdd(local, .monotonic); +} + +fn waitsetWaker(ctx: WaitsetCtx) void { + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + _ = ctx.guard.set_trigger_value(true); + sleepMs(ctx.io, 2); + _ = ctx.guard.set_trigger_value(false); + sleepMs(ctx.io, 2); + } +} + +fn waitsetChurn(ctx: WaitsetCtx) void { + var local: u64 = 0; + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + // Alternate ReadCondition / QueryCondition. `rc_handle` is what + // delete_readcondition takes for both (a QueryCondition is deleted + // via its ReadCondition view). + var rc_handle: DDS.ReadCondition = undefined; + if (local % 2 == 0) { + var params = [_][*:0]const u8{"5"}; + var params_seq = DDS.StringSeq{ ._buffer = ¶ms, ._length = 1, ._maximum = 1, ._release = false }; + const qc = ctx.dr.create_querycondition( + DDS.ANY_SAMPLE_STATE, + DDS.ANY_VIEW_STATE, + DDS.ANY_INSTANCE_STATE, + "count > %0", + ¶ms_seq, + ); + if (qc.ptr == zzdds.dcps.NIL_PTR) { + fail("waitset: create_querycondition returned nil under churn"); + return; + } + rc_handle = qcAsRc(qc); + } else { + const rc = ctx.dr.create_readcondition(DDS.ANY_SAMPLE_STATE, DDS.ANY_VIEW_STATE, DDS.ANY_INSTANCE_STATE); + if (rc.ptr == zzdds.dcps.NIL_PTR) { + fail("waitset: create_readcondition returned nil under churn"); + return; + } + rc_handle = rc; + } + const cond = rcAsCond(rc_handle); + + _ = ctx.ws.attach_condition(cond); + + // Snapshot the attached set concurrently with everyone else's + // attach/detach -- exercises get_conditions()'s locking, not its + // contents (which are racing by construction). + if (local % 8 == 0) { + var attached = DDS.ConditionSeq{}; + _ = ctx.ws.get_conditions(&attached); + freeCondSeq(&attached); + } + + sleepMs(ctx.io, 1); + + // 1-in-16: delete while still attached (see the scenario comment). + if (local % 16 != 15) _ = ctx.ws.detach_condition(cond); + _ = ctx.dr.delete_readcondition(rc_handle); + local += 1; + } + _ = g_ops.fetchAdd(local, .monotonic); +} + +fn runWaitset(io: std.Io, cfg: Config) !void { + var factory = zzdds.createFactory() catch { + fail("createFactory"); + return; + }; + defer factory.deinit(); + const dpf = factory.toDDSFactory(); + + const dp = dpf.create_participant(cfg.domain, .{}, null, 0); + if (dp.ptr == zzdds.dcps.NIL_PTR) { + fail("create_participant"); + return; + } + defer _ = dpf.delete_participant(dp); + + if (!zzdds.registerTypeSupport(dp, TYPE_NAME, .{ + .ctx = @ptrCast(&g_ts_alloc), + .compute_key_hash = gen.Message.computeKeyHashFromCdr, + .has_key = gen.Message.has_key, + .get_field = gen.Message.getFieldFromCdr, + })) { + fail("registerTypeSupport"); + return; + } + + const topic = dp.create_topic("WsChurnTopic", TYPE_NAME, .{}, null, 0); + if (topic.ptr == zzdds.dcps.NIL_PTR) { + fail("create_topic"); + return; + } + defer _ = dp.delete_topic(topic); + const td = dp.lookup_topicdescription("WsChurnTopic"); + + const sub = dp.create_subscriber(.{}, null, 0); + if (sub.ptr == zzdds.dcps.NIL_PTR) { + fail("create_subscriber"); + return; + } + defer _ = dp.delete_subscriber(sub); + + const dr = sub.create_datareader(td, .{}, null, 0); + if (dr.ptr == zzdds.dcps.NIL_PTR) { + fail("create_datareader"); + return; + } + defer _ = sub.delete_datareader(dr); + + const ws = zzdds.createWaitSet(g_alloc) catch { + fail("createWaitSet"); + return; + }; + defer ws.deinit(); + + const guard = zzdds.createGuardCondition(g_alloc) catch { + fail("createGuardCondition"); + return; + }; + defer guard.deinit(); + _ = ws.attach_condition(gcAsCond(guard)); + + const deadline_ns = monoNs(io) + @as(i64, cfg.duration_s) * std.time.ns_per_s; + const base = WaitsetCtx{ .io = io, .ws = ws, .dr = dr, .guard = guard, .deadline_ns = deadline_ns, .idx = 0 }; + + // [0] waiter, [1] waker, [2..] churn. + const threads = try g_alloc.alloc(std.Thread, cfg.threads + 2); + defer g_alloc.free(threads); + threads[0] = try std.Thread.spawn(.{}, waitsetWaiter, .{base}); + threads[1] = try std.Thread.spawn(.{}, waitsetWaker, .{base}); + for (threads[2..], 0..) |*t, i| { + var c = base; + c.idx = @intCast(i); + t.* = try std.Thread.spawn(.{}, waitsetChurn, .{c}); + } + for (threads) |t| t.join(); + + _ = ws.detach_condition(gcAsCond(guard)); +} + +// ── scenario: listener ───────────────────────────────────────────────────── +// +// The load regression for the #77 discovery/teardown UAF fix. Listeners are +// installed at every level of the DDS 1.4 s2.2.4.1.5 fallback chain +// (participant, publisher, subscriber) and left there for the whole run, so +// every matched/removed/data event has a 2-3 level fallback to walk. N churn +// threads each: create a DataWriter (on the shared publisher) + DataReader +// (on the shared subscriber), each with its own listener; swap those +// listeners a couple of times including to `null` (the ListenerBox +// replacement path); write a few samples; then delete both entities while +// SEDP is still (un)matching them -- so `on_publication_matched` / +// `on_subscription_matched` / `on_data_available` fire from the participant's +// receive/timer threads against an entity another thread is tearing down, +// falling through to a parent that is deliberately kept alive. +// +// Callbacks only bump global atomics -- no per-entity ctx to dangle. The +// point here is the fallback walk + ListenerBox refcount under contention, +// not ctx lifetime (that is `entities`). + +var g_cb_pub_matched = std.atomic.Value(u64).init(0); +var g_cb_sub_matched = std.atomic.Value(u64).init(0); +var g_cb_data = std.atomic.Value(u64).init(0); + +fn cbPubMatched(_: *anyopaque, _: *const DDS.PublicationMatchedStatus, _: ?*anyopaque) callconv(.c) void { + _ = g_cb_pub_matched.fetchAdd(1, .monotonic); +} +fn cbSubMatched(_: *anyopaque, _: *const DDS.SubscriptionMatchedStatus, _: ?*anyopaque) callconv(.c) void { + _ = g_cb_sub_matched.fetchAdd(1, .monotonic); +} +fn cbDataAvail(_: *anyopaque, _: ?*anyopaque) callconv(.c) void { + _ = g_cb_data.fetchAdd(1, .monotonic); +} + +const ListenerCtx = struct { + io: std.Io, + pub_: DDS.Publisher, + sub: DDS.Subscriber, + topic: DDS.Topic, + td: DDS.TopicDescription, + deadline_ns: i64, + idx: u32, +}; + +fn dwListener() DDS.DataWriterListener { + return .{ .on_publication_matched = cbPubMatched }; +} +fn drListener() DDS.DataReaderListener { + return .{ .on_subscription_matched = cbSubMatched, .on_data_available = cbDataAvail }; +} + +fn listenerChurn(ctx: ListenerCtx) void { + var local: u64 = 0; + const DW_MASK = DDS.PUBLICATION_MATCHED_STATUS; + const DR_MASK = DDS.SUBSCRIPTION_MATCHED_STATUS | DDS.DATA_AVAILABLE_STATUS; + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + var w_qos = DDS.DataWriterQos{}; + w_qos.reliability.kind = .RELIABLE_RELIABILITY_QOS; + const w = ctx.pub_.create_datawriter(ctx.topic, w_qos, dwListener(), DW_MASK); + if (w.ptr == zzdds.dcps.NIL_PTR) { + fail("listener: create_datawriter returned nil under churn"); + return; + } + var r_qos = DDS.DataReaderQos{}; + r_qos.reliability.kind = .RELIABLE_RELIABILITY_QOS; + const r = ctx.sub.create_datareader(ctx.td, r_qos, drListener(), DR_MASK); + if (r.ptr == zzdds.dcps.NIL_PTR) { + fail("listener: create_datareader returned nil under churn"); + _ = ctx.pub_.delete_datawriter(w); + return; + } + + // ListenerBox replacement: install -> replace -> clear, on both + // sides, while matched events for this pair are in flight. + _ = w.set_listener(dwListener(), DW_MASK); + _ = r.set_listener(drListener(), DR_MASK); + _ = w.set_listener(null, 0); + _ = r.set_listener(null, 0); + + // A few writes so on_data_available has something to deliver + // (races teardown -- most will land after the reader is gone). + const writer = gen.MessageDataWriter.init(w, g_alloc); + var msg = gen.Message{ .subject_id = @intCast(ctx.idx + 1), .count = 0 }; + const handle = writer.register_instance(msg); + var s: u32 = 0; + while (s < 3) : (s += 1) { + msg.count += 1; + writer.write(msg, handle) catch break; + } + + // Alternate teardown order so both fallback directions are hit: + // reader-first leaves the writer to take on_publication_matched(-1) + // (writer -> publisher -> participant); writer-first is the mirror. + if (local % 2 == 0) { + _ = ctx.sub.delete_datareader(r); + _ = ctx.pub_.delete_datawriter(w); + } else { + _ = ctx.pub_.delete_datawriter(w); + _ = ctx.sub.delete_datareader(r); + } + local += 1; + } + _ = g_ops.fetchAdd(local, .monotonic); +} + +fn runListener(io: std.Io, cfg: Config) !void { + var factory = zzdds.createFactory() catch { + fail("createFactory"); + return; + }; + defer factory.deinit(); + const dpf = factory.toDDSFactory(); + + const dp = dpf.create_participant(cfg.domain, .{}, null, 0); + if (dp.ptr == zzdds.dcps.NIL_PTR) { + fail("create_participant"); + return; + } + defer _ = dpf.delete_participant(dp); + + // Participant-level listener: the bottom of the fallback chain, always + // present. + _ = dp.set_listener(DDS.DomainParticipantListener{ + .on_publication_matched = cbPubMatched, + .on_subscription_matched = cbSubMatched, + .on_data_available = cbDataAvail, + }, DDS.PUBLICATION_MATCHED_STATUS | DDS.SUBSCRIPTION_MATCHED_STATUS | DDS.DATA_AVAILABLE_STATUS); + + if (!zzdds.registerTypeSupport(dp, TYPE_NAME, .{ + .ctx = @ptrCast(&g_ts_alloc), + .compute_key_hash = gen.Message.computeKeyHashFromCdr, + .has_key = gen.Message.has_key, + })) { + fail("registerTypeSupport"); + return; + } + + const topic = dp.create_topic("LsnChurnTopic", TYPE_NAME, .{}, null, 0); + if (topic.ptr == zzdds.dcps.NIL_PTR) { + fail("create_topic"); + return; + } + defer _ = dp.delete_topic(topic); + const td = dp.lookup_topicdescription("LsnChurnTopic"); + + // Mid-chain listeners: publisher and subscriber, also always present. + const pub_ = dp.create_publisher(.{}, DDS.PublisherListener{ + .on_publication_matched = cbPubMatched, + }, DDS.PUBLICATION_MATCHED_STATUS); + if (pub_.ptr == zzdds.dcps.NIL_PTR) { + fail("create_publisher"); + return; + } + defer _ = dp.delete_publisher(pub_); + + const sub = dp.create_subscriber(.{}, DDS.SubscriberListener{ + .on_subscription_matched = cbSubMatched, + .on_data_available = cbDataAvail, + }, DDS.SUBSCRIPTION_MATCHED_STATUS | DDS.DATA_AVAILABLE_STATUS); + if (sub.ptr == zzdds.dcps.NIL_PTR) { + fail("create_subscriber"); + return; + } + defer _ = dp.delete_subscriber(sub); + + const deadline_ns = monoNs(io) + @as(i64, cfg.duration_s) * std.time.ns_per_s; + const base = ListenerCtx{ .io = io, .pub_ = pub_, .sub = sub, .topic = topic, .td = td, .deadline_ns = deadline_ns, .idx = 0 }; + + const threads = try g_alloc.alloc(std.Thread, cfg.threads); + defer g_alloc.free(threads); + for (threads, 0..) |*t, i| { + var c = base; + c.idx = @intCast(i); + t.* = try std.Thread.spawn(.{}, listenerChurn, .{c}); + } + for (threads) |t| t.join(); + + // Sanity: if not a single matched callback fired, the scenario built the + // graph but never actually exercised event delivery -- treat that as a + // failure, not a silent pass. + if (g_cb_pub_matched.load(.monotonic) + g_cb_sub_matched.load(.monotonic) == 0) { + fail("listener: no matched callbacks ever fired -- delivery not exercised"); + } +} + +// ── scenario: cft ────────────────────────────────────────────────────────── +// +// The API audit calls runtime `set_expression_parameters` "fully untested, +// and CFT has an established bug". Here: a writer streams `Message` samples +// with `count` sweeping 0..99 so a "count > N" filter genuinely re-evaluates +// as N moves. One shared long-lived CFT (`g_cft`) + reader is drained by a +// dedicated thread while N churn threads both (a) rotate `g_cft`'s parameter +// through a small set via set_expression_parameters, and (b) create/use/ +// delete their own uniquely-named CFT + reader. DebugAllocator catches any +// leak/UAF in the param-vector swap or the CFT/reader teardown cascade. + +const CFT_THRESHOLDS = [_][*:0]const u8{ "0", "25", "50", "75", "90" }; + +const CftCtx = struct { + io: std.Io, + dp: DDS.DomainParticipant, + sub: DDS.Subscriber, + topic: DDS.Topic, + g_cft: DDS.ContentFilteredTopic, + g_dr: DDS.DataReader, + deadline_ns: i64, + idx: u32, +}; + +fn cftWriter(ctx: CftCtx, dw: DDS.DataWriter) void { + const writer = gen.MessageDataWriter.init(dw, g_alloc); + var msg = gen.Message{ .subject_id = 1, .count = 0 }; + const handle = writer.register_instance(msg); + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + msg.count = @mod(msg.count + 1, 100); + writer.write(msg, handle) catch {}; + sleepMs(ctx.io, 2); + } +} + +fn cftDrainer(ctx: CftCtx) void { + var reader = gen.MessageDataReader.init(ctx.g_dr, g_alloc); + var local: u64 = 0; + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + while (true) { + var value: gen.Message = .{}; + var info: DDS.SampleInfo = .{}; + const got = reader.take_next_sample(&value, &info) catch break; + if (!got) break; + local += 1; + } + sleepMs(ctx.io, 1); + } + _ = g_ops.fetchAdd(local, .monotonic); +} + +fn cftChurn(ctx: CftCtx) void { + var local: u64 = 0; + var name_buf: [48]u8 = undefined; + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + // (a) rotate the shared CFT's parameter -- the runtime + // reconfiguration path. + { + var p = [_][*:0]const u8{CFT_THRESHOLDS[local % CFT_THRESHOLDS.len]}; + var seq = DDS.StringSeq{ ._buffer = &p, ._length = 1, ._maximum = 1, ._release = false }; + _ = ctx.g_cft.set_expression_parameters(&seq); + } + + // (b) full CFT + reader lifecycle, uniquely named. + const name = std.fmt.bufPrintZ(&name_buf, "cft_{d}_{d}", .{ ctx.idx, local }) catch { + fail("cft: name format"); + return; + }; + var p0 = [_][*:0]const u8{CFT_THRESHOLDS[(local + 1) % CFT_THRESHOLDS.len]}; + var seq0 = DDS.StringSeq{ ._buffer = &p0, ._length = 1, ._maximum = 1, ._release = false }; + const cft = ctx.dp.create_contentfilteredtopic(name, ctx.topic, "count > %0", &seq0); + if (cft.ptr == zzdds.dcps.NIL_PTR) { + fail("cft: create_contentfilteredtopic returned nil under churn"); + return; + } + const dr = ctx.sub.create_datareader(cft.as_TopicDescription(), .{}, null, 0); + if (dr.ptr != zzdds.dcps.NIL_PTR) { + var reader = gen.MessageDataReader.init(dr, g_alloc); + var drained: u32 = 0; + while (drained < 8) : (drained += 1) { + var value: gen.Message = .{}; + var info: DDS.SampleInfo = .{}; + const got = reader.take_next_sample(&value, &info) catch break; + if (!got) break; + } + _ = ctx.sub.delete_datareader(dr); + } + _ = ctx.dp.delete_contentfilteredtopic(cft); + local += 1; + } + _ = g_ops.fetchAdd(local, .monotonic); +} + +fn runCft(io: std.Io, cfg: Config) !void { + var factory = zzdds.createFactory() catch { + fail("createFactory"); + return; + }; + defer factory.deinit(); + const dpf = factory.toDDSFactory(); + + const dp = dpf.create_participant(cfg.domain, .{}, null, 0); + if (dp.ptr == zzdds.dcps.NIL_PTR) { + fail("create_participant"); + return; + } + defer _ = dpf.delete_participant(dp); + + if (!zzdds.registerTypeSupport(dp, TYPE_NAME, .{ + .ctx = @ptrCast(&g_ts_alloc), + .compute_key_hash = gen.Message.computeKeyHashFromCdr, + .has_key = gen.Message.has_key, + .get_field = gen.Message.getFieldFromCdr, + })) { + fail("registerTypeSupport"); + return; + } + + const topic = dp.create_topic("CftChurnTopic", TYPE_NAME, .{}, null, 0); + if (topic.ptr == zzdds.dcps.NIL_PTR) { + fail("create_topic"); + return; + } + defer _ = dp.delete_topic(topic); + + const pub_ = dp.create_publisher(.{}, null, 0); + if (pub_.ptr == zzdds.dcps.NIL_PTR) { + fail("create_publisher"); + return; + } + defer _ = dp.delete_publisher(pub_); + var w_qos = DDS.DataWriterQos{}; + w_qos.reliability.kind = .RELIABLE_RELIABILITY_QOS; + const dw = pub_.create_datawriter(topic, w_qos, null, 0); + if (dw.ptr == zzdds.dcps.NIL_PTR) { + fail("create_datawriter"); + return; + } + defer _ = pub_.delete_datawriter(dw); + + const sub = dp.create_subscriber(.{}, null, 0); + if (sub.ptr == zzdds.dcps.NIL_PTR) { + fail("create_subscriber"); + return; + } + defer _ = dp.delete_subscriber(sub); + + // The shared, long-lived CFT + reader the churn threads reconfigure and + // the drainer thread empties. + var gp = [_][*:0]const u8{"50"}; + var gseq = DDS.StringSeq{ ._buffer = &gp, ._length = 1, ._maximum = 1, ._release = false }; + const g_cft = dp.create_contentfilteredtopic("CftShared", topic, "count > %0", &gseq); + if (g_cft.ptr == zzdds.dcps.NIL_PTR) { + fail("create_contentfilteredtopic (shared)"); + return; + } + const g_dr = sub.create_datareader(g_cft.as_TopicDescription(), .{}, null, 0); + if (g_dr.ptr == zzdds.dcps.NIL_PTR) { + fail("create_datareader (shared CFT)"); + _ = dp.delete_contentfilteredtopic(g_cft); + return; + } + + const deadline_ns = monoNs(io) + @as(i64, cfg.duration_s) * std.time.ns_per_s; + const base = CftCtx{ + .io = io, + .dp = dp, + .sub = sub, + .topic = topic, + .g_cft = g_cft, + .g_dr = g_dr, + .deadline_ns = deadline_ns, + .idx = 0, + }; + + // [0] writer, [1] drainer, [2..] churn. + const threads = try g_alloc.alloc(std.Thread, cfg.threads + 2); + defer g_alloc.free(threads); + threads[0] = try std.Thread.spawn(.{}, cftWriter, .{ base, dw }); + threads[1] = try std.Thread.spawn(.{}, cftDrainer, .{base}); + for (threads[2..], 0..) |*t, i| { + var c = base; + c.idx = @intCast(i); + t.* = try std.Thread.spawn(.{}, cftChurn, .{c}); + } + for (threads) |t| t.join(); + + // Ordered teardown of the shared pair (reader before its CFT). + _ = sub.delete_datareader(g_dr); + _ = dp.delete_contentfilteredtopic(g_cft); +} + +// ── scenario: participants ───────────────────────────────────────────────── +// +// Covers two API-audit stress items at once: "many-participant SPDP/SEDP +// fan-in/fan-out" and "participant-churning listener-fallback". N threads +// each run a full participant lifecycle on one shared domain -- factory -> +// participant (with a participant-level listener) -> registerTypeSupport -> +// topic -> either publisher+writer or subscriber+reader (listeners at those +// levels too) -> a few sample exchanges -> delete_participant -> +// factory.deinit(). Because every thread is on the same domain, at any +// instant there are ~N participants concurrently joining, matching and +// leaving, with writers fanning out to multiple readers and readers fanning +// in from multiple writers. Deleting a participant while its peers are still +// matched drives matched/removed events onto entities whose whole graph -- +// participant included -- is tearing down, exercising the participant level +// of the s2.2.4.1.5 fallback chain that the `listener` scenario (stable +// participant) can't reach. + +const ParticipantsCtx = struct { + io: std.Io, + domain: u32, + deadline_ns: i64, + idx: u32, +}; + +fn participantsThread(ctx: ParticipantsCtx) void { + var local: u64 = 0; + const is_writer = ctx.idx % 2 == 0; + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + var factory = zzdds.createFactory() catch { + fail("participants: createFactory"); + return; + }; + const dpf = factory.toDDSFactory(); + const dp = dpf.create_participant(ctx.domain, .{}, null, 0); + if (dp.ptr == zzdds.dcps.NIL_PTR) { + fail("participants: create_participant returned nil under churn"); + factory.deinit(); + return; + } + + _ = dp.set_listener(DDS.DomainParticipantListener{ + .on_publication_matched = cbPubMatched, + .on_subscription_matched = cbSubMatched, + .on_data_available = cbDataAvail, + }, DDS.PUBLICATION_MATCHED_STATUS | DDS.SUBSCRIPTION_MATCHED_STATUS | DDS.DATA_AVAILABLE_STATUS); + + if (!zzdds.registerTypeSupport(dp, TYPE_NAME, .{ + .ctx = @ptrCast(&g_ts_alloc), + .compute_key_hash = gen.Message.computeKeyHashFromCdr, + .has_key = gen.Message.has_key, + })) { + fail("participants: registerTypeSupport"); + _ = dpf.delete_participant(dp); + factory.deinit(); + return; + } + + const topic = dp.create_topic("PartChurnTopic", TYPE_NAME, .{}, null, 0); + if (topic.ptr == zzdds.dcps.NIL_PTR) { + fail("participants: create_topic"); + _ = dpf.delete_participant(dp); + factory.deinit(); + return; + } + + if (is_writer) { + const p = dp.create_publisher(.{}, DDS.PublisherListener{ .on_publication_matched = cbPubMatched }, DDS.PUBLICATION_MATCHED_STATUS); + if (p.ptr != zzdds.dcps.NIL_PTR) { + var qos = DDS.DataWriterQos{}; + qos.reliability.kind = .RELIABLE_RELIABILITY_QOS; + const w = p.create_datawriter(topic, qos, DDS.DataWriterListener{ .on_publication_matched = cbPubMatched }, DDS.PUBLICATION_MATCHED_STATUS); + if (w.ptr != zzdds.dcps.NIL_PTR) { + const writer = gen.MessageDataWriter.init(w, g_alloc); + var msg = gen.Message{ .subject_id = @intCast(ctx.idx + 1), .count = 0 }; + const handle = writer.register_instance(msg); + var s: u32 = 0; + while (s < 5) : (s += 1) { + msg.count += 1; + writer.write(msg, handle) catch break; + sleepMs(ctx.io, 2); + } + } + } + } else { + const s = dp.create_subscriber(.{}, DDS.SubscriberListener{ + .on_subscription_matched = cbSubMatched, + .on_data_available = cbDataAvail, + }, DDS.SUBSCRIPTION_MATCHED_STATUS | DDS.DATA_AVAILABLE_STATUS); + if (s.ptr != zzdds.dcps.NIL_PTR) { + const td = dp.lookup_topicdescription("PartChurnTopic"); + var qos = DDS.DataReaderQos{}; + qos.reliability.kind = .RELIABLE_RELIABILITY_QOS; + const r = s.create_datareader(td, qos, DDS.DataReaderListener{ + .on_subscription_matched = cbSubMatched, + .on_data_available = cbDataAvail, + }, DDS.SUBSCRIPTION_MATCHED_STATUS | DDS.DATA_AVAILABLE_STATUS); + if (r.ptr != zzdds.dcps.NIL_PTR) { + var reader = gen.MessageDataReader.init(r, g_alloc); + var polls: u32 = 0; + while (polls < 10) : (polls += 1) { + var value: gen.Message = .{}; + var info: DDS.SampleInfo = .{}; + _ = reader.take_next_sample(&value, &info) catch break; + sleepMs(ctx.io, 3); + } + } + } + } + + // Tear the whole graph down via the participant -- no explicit child + // deletes -- so delete_contained_entities runs while peers on the + // domain are still matched to these endpoints. + _ = dpf.delete_participant(dp); + factory.deinit(); + local += 1; + } + _ = g_ops.fetchAdd(local, .monotonic); +} + +fn runParticipants(io: std.Io, cfg: Config) !void { + const deadline_ns = monoNs(io) + @as(i64, cfg.duration_s) * std.time.ns_per_s; + const threads = try g_alloc.alloc(std.Thread, cfg.threads); + defer g_alloc.free(threads); + for (threads, 0..) |*t, i| { + t.* = try std.Thread.spawn(.{}, participantsThread, .{ParticipantsCtx{ + .io = io, + .domain = cfg.domain, + .deadline_ns = deadline_ns, + .idx = @intCast(i), + }}); + } + for (threads) |t| t.join(); + + if (g_cb_pub_matched.load(.monotonic) + g_cb_sub_matched.load(.monotonic) == 0) { + fail("participants: no matched callbacks ever fired -- discovery not exercised"); + } +} + +// ── scenario: instance ──────────────────────────────────────────────────── +// +// Each churn thread owns its own Publisher + DataWriter on a shared +// participant/topic (deliberately *not* one shared writer -- concurrent +// write() on a single DataWriter races on unprotected writer state, a +// separate zzdds gap noted in stress-tests/README.md). Every thread then +// hammers the instance-lifecycle API across a small keyspace: +// register_instance -> write x2 -> lookup_instance check -> get_key_value +// (value asserted: round-trips `Message`'s non-leading @key `subject_id`) -> +// periodically dispose / unregister_instance. +// One shared reader + drainer fans in from every writer and round-trips +// take_next_sample / get_key_value / lookup_instance on the read side. +// DebugAllocator + TSan cover the writer-side key registry, the reader-side +// per-instance tracking (view/generation state, dispose transitions), and +// the register/dispose/unregister lifecycle bookkeeping. + +const INSTANCE_KEYS: i32 = 12; + +const InstanceCtx = struct { + io: std.Io, + dp: DDS.DomainParticipant, + topic: DDS.Topic, + r: DDS.DataReader, + deadline_ns: i64, + idx: u32, +}; + +fn instanceChurn(ctx: InstanceCtx) void { + const p = ctx.dp.create_publisher(.{}, null, 0); + if (p.ptr == zzdds.dcps.NIL_PTR) { + fail("instance: create_publisher under churn"); + return; + } + defer _ = ctx.dp.delete_publisher(p); + var w_qos = DDS.DataWriterQos{}; + w_qos.history.kind = .KEEP_LAST_HISTORY_QOS; + const w = p.create_datawriter(ctx.topic, w_qos, null, 0); + if (w.ptr == zzdds.dcps.NIL_PTR) { + fail("instance: create_datawriter under churn"); + return; + } + defer _ = p.delete_datawriter(w); + const writer = gen.MessageDataWriter.init(w, g_alloc); + + var local: u64 = 0; + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + const key: i32 = @mod(@as(i32, @intCast(ctx.idx)) + @as(i32, @intCast(local)), INSTANCE_KEYS); + var msg = gen.Message{ .subject_id = key, .count = @intCast(local & 0x7fff) }; + + const h = writer.register_instance(msg); + writer.write(msg, h) catch {}; + msg.count +%= 1; + writer.write(msg, h) catch {}; + + // lookup_instance is a pure hash→handle function; it must agree with + // register_instance for the same key. + if (writer.lookup_instance(msg) != h) { + fail("instance: writer.lookup_instance != register_instance handle"); + return; + } + // get_key_value must round-trip `Message`'s @key `subject_id` -- its + // 3rd member, the non-leading-key case that zidl's selective parse + // (v0.3.12) fixed. The instance was just registered + written twice + // above, so a failure here is a real bug, not an expected + // "unregistered" error. + var kh: gen.Message = .{}; + writer.get_key_value(&kh, h) catch { + fail("instance: writer.get_key_value returned an error for a live instance"); + return; + }; + if (kh.subject_id != key) { + fail("instance: writer.get_key_value subject_id mismatch"); + return; + } + + switch (local % 4) { + 0 => writer.dispose(msg, h) catch {}, + 1 => writer.unregister_instance(msg, h) catch {}, + else => {}, // leave it registered/alive + } + local += 1; + } + _ = g_ops.fetchAdd(local, .monotonic); +} + +fn instanceDrainer(ctx: InstanceCtx) void { + var reader = gen.MessageDataReader.init(ctx.r, g_alloc); + while (monoNs(ctx.io) < ctx.deadline_ns and !g_fail.load(.acquire)) { + while (true) { + var value: gen.Message = .{}; + var info: DDS.SampleInfo = .{}; + const got = reader.take_next_sample(&value, &info) catch break; + if (!got) break; + // Gate on valid_data: an ALIVE sample has a populated `value` and a + // live instance to look up. A dispose/unregister notification + // (valid_data=false) may reference an instance the reader has + // already purged, where get_key_value legitimately errors. + if (info.valid_data and info.instance_handle != DDS.HANDLE_NIL) { + var kh: gen.Message = .{}; + reader.get_key_value(&kh, info.instance_handle) catch { + fail("instance: reader.get_key_value returned an error for a live instance"); + return; + }; + if (kh.subject_id != value.subject_id) { + fail("instance: reader.get_key_value subject_id mismatch"); + return; + } + } + _ = reader.lookup_instance(value); + } + sleepMs(ctx.io, 1); + } +} + +fn runInstance(io: std.Io, cfg: Config) !void { + var factory = zzdds.createFactory() catch { + fail("createFactory"); + return; + }; + defer factory.deinit(); + const dpf = factory.toDDSFactory(); + + const dp = dpf.create_participant(cfg.domain, .{}, null, 0); + if (dp.ptr == zzdds.dcps.NIL_PTR) { + fail("create_participant"); + return; + } + defer _ = dpf.delete_participant(dp); + + if (!zzdds.registerTypeSupport(dp, TYPE_NAME, .{ + .ctx = @ptrCast(&g_ts_alloc), + .compute_key_hash = gen.Message.computeKeyHashFromCdr, + .has_key = gen.Message.has_key, + })) { + fail("registerTypeSupport"); + return; + } + + const topic = dp.create_topic("InstChurnTopic", TYPE_NAME, .{}, null, 0); + if (topic.ptr == zzdds.dcps.NIL_PTR) { + fail("create_topic"); + return; + } + defer _ = dp.delete_topic(topic); + const td = dp.lookup_topicdescription("InstChurnTopic"); + + const sub = dp.create_subscriber(.{}, null, 0); + if (sub.ptr == zzdds.dcps.NIL_PTR) { + fail("create_subscriber"); + return; + } + defer _ = dp.delete_subscriber(sub); + const r = sub.create_datareader(td, .{}, null, 0); + if (r.ptr == zzdds.dcps.NIL_PTR) { + fail("create_datareader"); + return; + } + defer _ = sub.delete_datareader(r); + + const deadline_ns = monoNs(io) + @as(i64, cfg.duration_s) * std.time.ns_per_s; + const base = InstanceCtx{ .io = io, .dp = dp, .topic = topic, .r = r, .deadline_ns = deadline_ns, .idx = 0 }; + + // [0] drainer, [1..] churn. + const threads = try g_alloc.alloc(std.Thread, cfg.threads + 1); + defer g_alloc.free(threads); + threads[0] = try std.Thread.spawn(.{}, instanceDrainer, .{base}); + for (threads[1..], 0..) |*t, i| { + var c = base; + c.idx = @intCast(i); + t.* = try std.Thread.spawn(.{}, instanceChurn, .{c}); + } + for (threads) |t| t.join(); +} diff --git a/stress-tests/zig/lifecycle_churn/run.py b/stress-tests/zig/lifecycle_churn/run.py index b946515a..e45f3665 100755 --- a/stress-tests/zig/lifecycle_churn/run.py +++ b/stress-tests/zig/lifecycle_churn/run.py @@ -22,6 +22,7 @@ sys.path.insert(0, str(APP_DIR.parents[2] / "examples")) import _common # noqa: E402 +SCENARIOS = ["entities", "reentrant", "waitset", "listener", "cft", "participants", "instance"] SUMMARY_RE = re.compile(r"^SUMMARY: OK scenario=(\w+) ops=(\d+)", re.M) BAD = ("FAIL:", "panic", "Segmentation fault", "General protection", "ThreadSanitizer", "data race", "leaked") @@ -40,7 +41,7 @@ def build(tsan: bool) -> Path: def main() -> int: ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) - ap.add_argument("--scenario", required=True, choices=["entities", "reentrant"]) + ap.add_argument("--scenario", required=True, choices=SCENARIOS) ap.add_argument("--threads", type=int, default=6) ap.add_argument("--iterations", type=int, default=40) ap.add_argument("--duration", type=int, default=8) diff --git a/test/dcps/type_support_test.zig b/test/dcps/type_support_test.zig index 50d8c742..8742d009 100644 --- a/test/dcps/type_support_test.zig +++ b/test/dcps/type_support_test.zig @@ -217,6 +217,12 @@ fn exclusiveDrQos() DDS.DataReaderQos { return q; } +fn keepAllDrQos() DDS.DataReaderQos { + var q = DDS.DataReaderQos{}; + q.history.kind = .KEEP_ALL_HISTORY_QOS; + return q; +} + // ── Tests ───────────────────────────────────────────────────────────────────── test "TypeSupport: registerTypeSupport stores callback" { @@ -335,3 +341,70 @@ test "TypeSupport: non-nil inline key_hash takes precedence over TypeSupport" { // 0xBB (A, from inline) ≠ 0xAA (B, from TypeSupport) → two instances. try testing.expectEqual(@as(usize, 2), pendingCount(dr)); } + +test "TypeSupport: has_key writer sends inline PID_KEY_HASH for a zero-valued key, bypassing key_hash_fn" { + // A keyed writer (TypeSupport.has_key = true) must put an inline + // PID_KEY_HASH on every alive sample per RTPS §8.7.9 — including when the + // hash is all-zeros (a zero-valued key). The subscriber then routes by that + // *present* hash and never calls key_hash_fn to reconstruct one from the + // payload (which, for a non-leading @key, would misread it). + // + // Here both writers emit a zero key hash for different payloads. testKeyHash + // (reads payload[4]) *would* derive distinct hashes 0xAA / 0xBB if it were + // consulted — but it must NOT be, because the wire now carries an explicit + // all-zero PID_KEY_HASH. So both samples land on one instance. + const alloc = testing.allocator; + var fx = try Fixture.init(alloc); + defer fx.deinit(); + + // has_key must be registered on the *writer* participants, before the + // writers are created, for pubCreateProtoWriter to mark them keyed. + inline for (.{ fx.dp_a, fx.dp_b, fx.dp_r }) |dp| { + _ = fx.dpImpl(dp).registerTypeSupport("TSType", .{ + .ctx = undefined, + .compute_key_hash = testKeyHash, + .has_key = true, + }); + } + + const dw_a = fx.makeWriterA(.{}); + const dw_b = fx.makeWriterB(.{}); + const dr = fx.makeReader(keepAllDrQos()); + + try writeNilKey(dw_a, &PAYLOAD_A); // payload[4] = 0xAA — ignored + try writeNilKey(dw_b, &PAYLOAD_B); // payload[4] = 0xBB — ignored + + // Both carry an inline all-zero key hash → same instance → both retained + // (KEEP_ALL) under one instance handle. + try testing.expectEqual(@as(usize, 2), pendingCount(dr)); + dr.mu.lock(); + const ih0 = dr.pending.items[0].info.instance_handle; + const ih1 = dr.pending.items[1].info.instance_handle; + dr.mu.unlock(); + try testing.expectEqual(ih0, ih1); +} + +test "TypeSupport: without has_key, a zero-valued key still falls back to key_hash_fn (unchanged)" { + // Control: same inputs as above but has_key defaults false. The writer omits + // the inline hash for an all-zero key, so the reader reconstructs distinct + // hashes via key_hash_fn (testKeyHash → 0xAA / 0xBB) → two instances. + const alloc = testing.allocator; + var fx = try Fixture.init(alloc); + defer fx.deinit(); + + _ = fx.dpImpl(fx.dp_r).registerTypeSupport("TSType", .{ .ctx = undefined, .compute_key_hash = testKeyHash }); + + const dw_a = fx.makeWriterA(.{}); + const dw_b = fx.makeWriterB(.{}); + const dr = fx.makeReader(keepAllDrQos()); + + try writeNilKey(dw_a, &PAYLOAD_A); + try writeNilKey(dw_b, &PAYLOAD_B); + + try testing.expectEqual(@as(usize, 2), pendingCount(dr)); + dr.mu.lock(); + const ih0 = dr.pending.items[0].info.instance_handle; + const ih1 = dr.pending.items[1].info.instance_handle; + dr.mu.unlock(); + try testing.expect(ih0 != ih1); +} diff --git a/test/dcps/writer_vtable_test.zig b/test/dcps/writer_vtable_test.zig index 32102816..05fc2958 100644 --- a/test/dcps/writer_vtable_test.zig +++ b/test/dcps/writer_vtable_test.zig @@ -1030,3 +1030,83 @@ test "publish_loan_raw: an explicit instance_handle that doesn't match the key h // hit PRECONDITION_NOT_MET. try testing.expectEqual(DDS.RETCODE_OK, dw.vtable.return_loan_raw(dw.ptr, &cdr_payload)); } + +// ── Regression: concurrent write() on one DataWriter ────────────────────────── + +test "concurrent write_raw on one DataWriter keeps key_registry and last_sn consistent" { + // Regression for the stress `instance` scenario finding: DataWriterImpl.writeRaw + // mutated `key_registry` (a HashMapUnmanaged) and `last_sn` with no lock, so + // concurrent write() on one DataWriter -- spec-legal -- raced on the map's + // grow/insert (TSan) and could abort on its SafetyLock. Now guarded by + // `key_registry_mu` + an atomic `last_sn` (the RTPS layer under `proto_writer` + // was already internally locked). Consistency check here; the data-race half + // is caught by the `-Dsanitize-thread` lane running this same file. + const N_THREADS = 6; + const PER_THREAD = 40; // 240 distinct keys -> several map grows + + var fx = try SingleFixture.init(alloc); + defer fx.deinit(); + const dw = fx.makeWriter(.{}, null, 0); + defer _ = fx.pub_.vtable.delete_datawriter(fx.pub_.ptr, dw); + const impl: *DataWriterImpl = @ptrCast(@alignCast(dw.ptr)); + + const Writer = struct { + dw: DDS.DataWriter, + tid: u32, + fn run(c: @This()) void { + var payload = [_]u8{ 0x00, 0x01, 0x00, 0x00, 0, 0, 0, 0 }; + var i: u32 = 0; + while (i < PER_THREAD) : (i += 1) { + var kh = std.mem.zeroes([16]u8); + std.mem.writeInt(u32, kh[0..4], c.tid, .little); + std.mem.writeInt(u32, kh[4..8], i, .little); + var kh_seq = DDS.OctetSeq{ ._buffer = &kh, ._length = 16, ._maximum = 16, ._release = false }; + std.mem.writeInt(u32, payload[4..8], c.tid *% 1000 +% i, .little); + var pl_seq = DDS.OctetSeq{ ._buffer = &payload, ._length = payload.len, ._maximum = payload.len, ._release = false }; + const ts = DDS.Time_t{ .sec = DDS.TIME_INVALID_SEC, .nanosec = DDS.TIME_INVALID_NSEC }; + _ = c.dw.vtable.write_raw(c.dw.ptr, &kh_seq, DDS.HANDLE_NIL, &pl_seq, .ALIVE_WRITE_KIND, &ts); + } + } + }; + + // Reader thread: the HashMap.get vs grow half of the race. + var stop = std.atomic.Value(bool).init(false); + const Poller = struct { + impl: *DataWriterImpl, + stop: *std.atomic.Value(bool), + fn run(c: @This()) void { + while (!c.stop.load(.acquire)) { + const kh = std.mem.zeroes([16]u8); + _ = c.impl.getKeyValueRaw(DataWriterImpl.registerInstanceRaw(kh)); + } + } + }; + + var threads: [N_THREADS]std.Thread = undefined; + const poller = try std.Thread.spawn(.{}, Poller.run, .{Poller{ .impl = impl, .stop = &stop }}); + for (&threads, 0..) |*t, i| { + t.* = try std.Thread.spawn(.{}, Writer.run, .{Writer{ .dw = dw, .tid = @intCast(i) }}); + } + for (threads) |t| t.join(); + stop.store(true, .release); + poller.join(); + + // Every distinct key must have landed. + var found: usize = 0; + var tid: u32 = 0; + while (tid < N_THREADS) : (tid += 1) { + var i: u32 = 0; + while (i < PER_THREAD) : (i += 1) { + var kh = std.mem.zeroes([16]u8); + std.mem.writeInt(u32, kh[0..4], tid, .little); + std.mem.writeInt(u32, kh[4..8], i, .little); + if (impl.getKeyValueRaw(DataWriterImpl.registerInstanceRaw(kh)) != null) found += 1; + } + } + try testing.expectEqual(@as(usize, N_THREADS * PER_THREAD), found); + + // proto_writer hands out sequential SNs under its own lock; last_sn just has + // to be a real one that was written (no tearing), 1..=total. + const last = impl.last_sn.load(.monotonic); + try testing.expect(last >= 1 and last <= @as(i64, N_THREADS * PER_THREAD)); +}