diff --git a/pj_base/include/pj_base/data_source_protocol.h b/pj_base/include/pj_base/data_source_protocol.h index a75851c6..7a79dfb8 100644 --- a/pj_base/include/pj_base/data_source_protocol.h +++ b/pj_base/include/pj_base/data_source_protocol.h @@ -8,7 +8,9 @@ * See pj_base/plugin_data_api.h. append_arrow_ipc is removed. * - Every vtable slot is PJ_NOEXCEPT. Trampolines that drop exceptions * through the ABI boundary are now a compile-time error in C++. - * - Every slot carries a thread-class tag (// [main-thread], etc.). + * - Every slot carries a thread-class tag (// [main-thread], etc.), except + * start/stop, whose rule none of the tags expresses; see the threading + * table on PJ_data_source_vtable_t. * * The host obtains the plugin's vtable via `PJ_get_data_source_vtable()` * and drives the plugin through: create -> bind(registry) -> load_config @@ -63,13 +65,15 @@ extern "C" { #endif /** - * Plugin lifecycle state machine. + * Plugin lifecycle states, as reported through notify_state / current_state. * - * Valid transitions: - * idle -> configuring -> starting -> running -> stopping -> stopped - * running -> paused -> running (if SUPPORTS_PAUSE capability set) - * any -> failed (terminal) - * stopped is terminal — create a new instance to restart. + * ADVISORY: a host may ignore notify_state and never call current_state (the + * reference host does both), so a plugin must not depend on either for + * correctness. What a host acts on is the start() return value, request_stop, + * is_stop_requested and complete_ingest. A plugin is not required to visit + * every state; the SDK bases report starting, then running (stream) or + * stopped/failed (file). stopped and failed are terminal for the instance: + * the host never calls start() twice and creates a new instance to restart. */ typedef enum { PJ_DATA_SOURCE_STATE_IDLE = 0, @@ -255,7 +259,8 @@ typedef struct PJ_ingest_completion_t { * raw message payloads. The per-slot thread tag governs: the host may drive * start()/stop() on one thread and poll() (with its [stream-thread] * callbacks) on another, so a host must treat [stream-thread] callbacks as - * arriving off its UI thread and marshal accordingly. + * arriving off its UI thread and marshal accordingly. When a plugin may call + * the [stream-thread] slots is narrowed in PJ_data_source_vtable_t. * * Fallible calls take a `PJ_error_t* out_error` which the callee populates * on failure. Callers may pass NULL if they don't need the detail. @@ -287,7 +292,8 @@ typedef struct PJ_data_source_runtime_host_vtable_t { /** [thread-safe] Returns true if the host has requested the plugin to stop. */ bool (*is_stop_requested)(void* ctx) PJ_NOEXCEPT; - /** [thread-safe] Inform the host that the plugin has transitioned to @p state. */ + /** [thread-safe] Inform the host that the plugin has transitioned to @p state. + * Advisory: a host may ignore it (see PJ_data_source_state_t). */ void (*notify_state)(void* ctx, PJ_data_source_state_t state) PJ_NOEXCEPT; /** @@ -308,7 +314,7 @@ typedef struct PJ_data_source_runtime_host_vtable_t { PJ_error_t* out_error) PJ_NOEXCEPT; /** - * [main-thread] Display a modal message box to the user and wait for + * [thread-safe] Display a modal message box to the user and wait for * their response. BLOCKS until the user closes the dialog. The host is * responsible for showing the dialog on the UI thread in a thread-safe * manner; the plugin may call from any thread and the host will marshal. @@ -499,6 +505,43 @@ typedef struct { * create -> bind(registry) -> load_config (optional) * -> start -> poll* -> stop -> destroy * + * Threading (normative). The host never runs two of create, bind, + * load_config, save_config, get_dialog, start, poll, stop and destroy + * concurrently on one instance, and each happens-after the previous returned; + * consecutive calls may come from different threads. + * + * create, destroy, capabilities, GUI thread, never while start/poll/stop + * bind, load_config, save_config, runs. load_config may be called several + * get_dialog times before start; save_config also + * after start returned. + * start Once. Any host thread: a FINITE_IMPORT + * start may run on a worker and block for + * the whole import (the host calls no + * poll); a CONTINUOUS_STREAM start may run + * on the GUI thread and must return + * promptly. + * poll CONTINUOUS_STREAM only, after start + * returned true; repeatedly, from one host + * thread that need not be start's. + * stop Possibly on poll's thread, after the + * last poll returned. May be skipped: a + * host may destroy after a failed start or + * a finished import without calling stop, + * so release resources in destroy as well. + * get_plugin_extension Any thread, concurrently with anything. + * set_active_topics (extension) GUI thread, concurrently with poll and + * with the plugin's own threads. + * pause, resume, current_state Advisory; a host may never call them. + * If called: GUI thread, never while + * start/poll/stop runs. + * + * Plugin -> host: [stream-thread] slots of the write hosts and of the runtime + * host may be called only while a start, poll or stop call on this instance + * is in progress, from one thread at a time (a plugin thread that the call + * waits on counts; a thread still producing after the call returned does not: + * between polls the host touches the same host objects). [thread-safe] slots + * may be called from any thread while the instance exists. + * * Fallible slots take a PJ_error_t* out-param which the callee populates * on failure. Callers may pass NULL to discard error detail. Every slot * is PJ_NOEXCEPT; exceptions from the implementation must be caught @@ -556,23 +599,60 @@ typedef struct PJ_data_source_vtable_t { * On success, returns true and writes to @p out_json a view over a * plugin-owned string that remains valid until the next call to this * function on the same ctx. + * + * The output must be a UTF-8 JSON object that load_config accepts; a host + * persists it (e.g. in a layout) and later passes it back verbatim apart + * from the host-written keys listed under load_config. It may carry + * "filepath" and "_parser_config" back (a file layout replays this output, + * so that is how its parser config survives), never "__pj_saved_filepath". */ bool (*save_config)(void* ctx, PJ_string_view_t* out_json, PJ_error_t* out_error) PJ_NOEXCEPT; - /** [main-thread] Restore plugin configuration from JSON. */ + /** + * [main-thread] Restore plugin configuration from JSON. + * + * @p config_json is either empty, meaning "no saved config" (load_config + * is still called, e.g. on a stream's first run, and must keep defaults), + * or a UTF-8 JSON object: the output of this plugin's save_config or of its + * embedded dialog's save_config (after an accepted dialog), with host keys + * added or rewritten, or host keys alone. It may have been written by an + * OLDER version of the plugin, which load_config must still accept. Top-level + * keys starting with "_" belong to the host; the plugin must not define + * its own. Keys the reference host writes: + * "filepath" — file sources: the file to import, rewritten + * on layout replay to the file actually opened. + * "_parser_config" — string: the embedded parser dialog's config; + * forward it as parser_config_json in + * ensure_parser_binding (the host does not). + * "__pj_saved_filepath" — layout replay only: the absolute path that + * "filepath" replaced, to rebase sidecar files. + * "__pj_fanout" is host-defined but plugin-written: an array of complete + * config strings in an accepted file-source config, asking the host to + * import each with its own instance. May be called several times before + * start. + */ bool (*load_config)(void* ctx, PJ_string_view_t config_json, PJ_error_t* out_error) PJ_NOEXCEPT; - /** [main-thread] Begin data acquisition. May spawn stream threads internally. */ + /** Begin data acquisition. Thread: see the table above (not necessarily the + * GUI thread). May spawn plugin threads. If it fails, release what it + * acquired: the host may destroy the instance without calling stop. */ bool (*start)(void* ctx, PJ_error_t* out_error) PJ_NOEXCEPT; - /** [main-thread] Stop data acquisition. Must be idempotent. Failures are not reportable. */ + /** Stop data acquisition. Thread: see the table above (may be poll's + * thread). Must be idempotent. Failures are not reportable. */ void (*stop)(void* ctx) PJ_NOEXCEPT; - /** [main-thread] Pause a running source. Returns false + error if unsupported. */ + /** [main-thread] Advisory: a host may never call it and pauses ingest + * host-side instead. Pause a running source. Returns false + error if + * unsupported. */ bool (*pause)(void* ctx, PJ_error_t* out_error) PJ_NOEXCEPT; - /** [main-thread] Resume a paused source. Returns false + error if unsupported. */ + /** [main-thread] Advisory, as pause. Resume a paused source. Returns + * false + error if unsupported. */ bool (*resume)(void* ctx, PJ_error_t* out_error) PJ_NOEXCEPT; - /** [stream-thread] Called periodically by the host while running. */ + /** [stream-thread] Called periodically by the host while running, from one + * host thread (see the table above). Must not block. */ bool (*poll)(void* ctx, PJ_error_t* out_error) PJ_NOEXCEPT; - /** [thread-safe] Return the plugin's current lifecycle state. */ + /** [main-thread] Advisory: a host may never call it. Return the plugin's + * current lifecycle state. Not [thread-safe]: the SDK bases keep the state + * unsynchronised, so never call it while start/poll/stop runs. */ PJ_data_source_state_t (*current_state)(void* ctx) PJ_NOEXCEPT; /** @@ -634,8 +714,10 @@ typedef struct PJ_data_source_vtable_t { typedef struct { uint32_t struct_size; /**< sizeof(PJ_topic_subscription_v1_t) — for forward-compatible growth. */ /** - * [main-thread] Set the full active-topic set. Entered on the host thread; - * the plugin marshals to its own stream thread. Returns false + populates + * [main-thread] Set the full active-topic set. Entered on the GUI thread, + * concurrently with poll (also once right after start, before the first + * poll); the plugin stores the set in a mutex-protected latest-wins slot + * that its stream thread drains, never a command queue. Returns false + populates * out_error on failure. Names the source does not recognize are ignored * (not an error). */ diff --git a/pj_base/include/pj_base/plugin_data_api.h b/pj_base/include/pj_base/plugin_data_api.h index 768c14ef..48be68dc 100644 --- a/pj_base/include/pj_base/plugin_data_api.h +++ b/pj_base/include/pj_base/plugin_data_api.h @@ -442,7 +442,9 @@ typedef struct { * object carries a SHARED error buffer whose pointer is * handed back through PJ_error_t*, so two threads must not * be inside slots of the SAME host object concurrently. - * Serialise with your own mutex if you fan out. + * Serialise with your own mutex if you fan out. A + * DataSource narrows this further: only while a host + * start/poll/stop call is in progress (data_source_protocol.h). * [thread-safe] Any thread, concurrently. * ========================================================================== */ @@ -611,9 +613,10 @@ typedef struct { * for exactly this). * * The two ABI-appended tail slots (register_object_topic_on_dataset, - * set_object_topic_retention) deliberately keep a conservative [main-thread] - * tag: no host implements them yet — every known vtable leaves them NULL — so - * there is no implementation to audit. Re-tag when the first one lands. + * set_object_topic_retention) follow the same rule. In the reference host the + * first takes the engine lock only for its dataset lookup; both otherwise go + * through ObjectStore's locks, like register_object_topic, and share its + * error buffer. * * read_series_arrow: caller zero-initialises both out structs. Host fills * them (allocates buffers, sets release callbacks). On success the caller @@ -686,7 +689,7 @@ typedef struct PJ_toolbox_host_vtable_t { void* ctx, PJ_object_topic_handle_t topic, int64_t timestamp_ns, const uint8_t* data, uint64_t size, PJ_error_t* out_error) PJ_NOEXCEPT; - /* [main-thread] Register an object topic on an EXISTING dataset, addressed by + /* [stream-thread] Register an object topic on an EXISTING dataset, addressed by * its DatasetId, rather than under a data source the toolbox created. This is * the ANNOTATION path: a toolbox attaches objects (e.g. plot markers) to data * loaded by another source. Idempotent — if a topic with this name already @@ -701,7 +704,7 @@ typedef struct PJ_toolbox_host_vtable_t { void* ctx, uint32_t dataset_id, PJ_string_view_t topic_name, PJ_string_view_t metadata_json, PJ_object_topic_handle_t* out_handle, PJ_error_t* out_error) PJ_NOEXCEPT; - /* [main-thread] Bound an object topic's retention to keep at most `max_entries` + /* [stream-thread] Bound an object topic's retention to keep at most `max_entries` * of the most recent snapshots (0 = unlimited). Generic capability: a producer * that republishes a whole set under one topic at a sentinel timestamp (e.g. * plot markers) sets max_entries=1 so superseded snapshots are evicted instead diff --git a/pj_base/include/pj_base/sdk/data_source_patterns.hpp b/pj_base/include/pj_base/sdk/data_source_patterns.hpp index e2d1c13c..0e5a0682 100644 --- a/pj_base/include/pj_base/sdk/data_source_patterns.hpp +++ b/pj_base/include/pj_base/sdk/data_source_patterns.hpp @@ -10,8 +10,9 @@ * - **StreamSourceBase** — long-lived streaming sources. * Override: extraCapabilities(), onStart(), onPoll(), onStop(). * - * Both manage the lifecycle state machine automatically — derived classes - * only implement the domain-specific work. + * Both report lifecycle state automatically (advisory; see + * PJ_data_source_state_t) — derived classes only implement the + * domain-specific work. * * A StreamSourceBase with its own receive thread should NOT hand-roll the * thread -> onPoll() buffer: pj_plugins/sdk/streaming_source.hpp ships @@ -55,6 +56,7 @@ namespace PJ { * * Manages the full lifecycle state machine. The derived class implements * importData() — all file I/O, parsing, and write-host calls happen there. + * It runs inside start(), which a host may call on a worker thread. * * ## Config convention * @@ -121,18 +123,23 @@ class StreamSourceBase : public DataSourcePluginBase { /// Return additional capability flags (e.g. kCapabilityDirectIngest). virtual uint64_t extraCapabilities() const = 0; - /// Called from start(). Open connections, allocate resources. + /// Called from start(), possibly on the GUI thread. Open connections, + /// allocate resources. On failure, release what it acquired: onStop() is + /// not called and the host may destroy the instance without stop(). virtual Status onStart() = 0; /// Called periodically by the host's polling thread. /// MUST NOT BLOCK — drain buffered data and return immediately. /// Do not call recv(), read(), or any syscall that may wait. /// If your source has a receive thread, swap-drain a buffer here. - /// Host methods (appendRecord, pushMessage) may only be called from this method. + /// [stream-thread] host methods (appendRecord, pushMessage, ...) may only be + /// called from onStart/onPoll/onStop, never from a thread you spawned; + /// [thread-safe] ones (reportMessage, requestStop, isStopRequested) from any. virtual Status onPoll() = 0; - /// Called from stop(). Close connections, free resources. - /// Must be idempotent. + /// Called from stop(), possibly on the polling thread rather than the one + /// that ran onStart(). Close connections, join your threads, free + /// resources. Must be idempotent. virtual void onStop() = 0; Status start() final; diff --git a/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp b/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp index 1e4bf31b..72620a07 100644 --- a/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp +++ b/pj_base/include/pj_base/sdk/toolbox_plugin_base.hpp @@ -159,7 +159,8 @@ class ToolboxPluginBase { return PJ_borrowed_dialog_t{nullptr, nullptr}; } - /// React to data appended to the datastore. Default is no-op. + /// React to data appended to the datastore. Default is no-op. Advisory: a + /// host may never call it, so do not depend on it for correctness. virtual void onDataChanged() {} /// Return a pointer to a static plugin-exposed extension for @p id, or diff --git a/pj_base/include/pj_base/toolbox_protocol.h b/pj_base/include/pj_base/toolbox_protocol.h index 8db4a17d..cedb1dfa 100644 --- a/pj_base/include/pj_base/toolbox_protocol.h +++ b/pj_base/include/pj_base/toolbox_protocol.h @@ -89,7 +89,9 @@ typedef struct PJ_toolbox_runtime_host_vtable_t { * by the toolbox write host's create_data_source (== the dataset id). * On success fills `out_host` with a standard data-source runtime host * fat pointer: ensure_parser_binding / push_message on it behave exactly - * as they do for file/stream DataSource plugins. The context stays valid + * as they do for file/stream DataSource plugins. Its [stream-thread] slots + * may be called from any thread (typically a toolbox worker), one thread at + * a time per context; there is no start/poll/stop window. The context stays valid * until release_parser_ingest or host teardown. * Returns false (with out_error populated) if `data_source_id` is not a live * data source or parser ingest is not configured on this host; `out_host` is @@ -124,6 +126,23 @@ typedef struct { * Typical lifecycle: create -> bind(registry) -> load_config (optional) * -> [user interacts] -> save_config -> destroy. * Every slot is PJ_NOEXCEPT. + * + * Threading (normative). Host -> plugin: every slot except + * get_plugin_extension is called on the GUI thread, one at a time per + * instance; get_plugin_extension may be called from any thread. + * on_data_changed is advisory: a host may never call it. + * + * Plugin -> host, by service: + * pj.toolbox_runtime.v1 report_message, notify_data_changed and the + * parser-ingest create/release/discard slots: + * any thread, concurrently. The ingest context + * they return: one thread at a time. + * pj.toolbox_write.v1 [stream-thread]: any thread, typically the + * toolbox's own worker, one thread at a time per + * host object (see plugin_data_api.h). + * every other service [main-thread] unless its slot says otherwise: + * call from inside a host->plugin call or a + * dialog callback, not from a plugin thread. */ typedef struct PJ_toolbox_vtable_t { uint32_t protocol_version; @@ -168,7 +187,9 @@ typedef struct PJ_toolbox_vtable_t { PJ_borrowed_dialog_t (*get_dialog)(void* ctx) PJ_NOEXCEPT; /** [main-thread] Notify the plugin that new records have been appended - * to the datastore. */ + * to the datastore. Advisory: a host is not required to call it (the + * reference host currently does not), so a plugin must not depend on it + * for correctness; re-read the catalog when it needs fresh data. */ void (*on_data_changed)(void* ctx) PJ_NOEXCEPT; /** [thread-safe] Query a plugin-exposed extension by reverse-DNS id. diff --git a/pj_plugins/docs/ARCHITECTURE.md b/pj_plugins/docs/ARCHITECTURE.md index 6fb26bce..56bad5a3 100644 --- a/pj_plugins/docs/ARCHITECTURE.md +++ b/pj_plugins/docs/ARCHITECTURE.md @@ -218,7 +218,10 @@ previously-circulated pre-v4 design included): enforced at compile time. Trampolines catch and translate internally. - **Thread-class tags on every slot.** Every function-pointer field in the ABI headers carries a `[main-thread]` / `[stream-thread]` / - `[thread-safe]` comment. Host-side runtime checking is optional + `[thread-safe]` comment (DataSource `start`/`stop` state their rule in + words). Each family's plugin vtable also carries a normative per-slot + threading table (`data_source_protocol.h`, `toolbox_protocol.h`). + Host-side runtime checking is optional (reserved for a future `"pj.thread_check.v1"` service). - **Embedded-manifest plugin discovery.** Each DSO exports a family-specific protocol vtable with embedded metadata (`manifest_json` diff --git a/pj_plugins/docs/REQUIREMENTS.md b/pj_plugins/docs/REQUIREMENTS.md index 6e233577..dc0aeaf2 100644 --- a/pj_plugins/docs/REQUIREMENTS.md +++ b/pj_plugins/docs/REQUIREMENTS.md @@ -100,8 +100,9 @@ finite imports (CSV, Parquet, MCAP) and continuous streams (MQTT, ZMQ, UDP). - May decode data directly (direct ingest) or hand raw payloads to a MessageParser (delegated ingest). - Topics may appear lazily at runtime. -- Lifecycle: idle → configuring → starting → running → stopping → stopped. - Optional pause/resume. +- Lifecycle: create → bind → load_config → start → poll* → stop → destroy, + signalled by the `start`/`poll` results, `request_stop`, `is_stop_requested` + and `complete_ingest`. Reported states and pause/resume are advisory. ### MessageParser @@ -234,7 +235,7 @@ DataSource plugins receive a runtime host providing: - Message reporting (info/warning/error to host UI log). - Progress bar (start/update/finish with optional cancellation). -- State notifications and plugin-initiated stop. +- Advisory state notifications and plugin-initiated stop. - Delegated parser binding and raw message dispatch. Toolbox plugins receive a simpler runtime host: @@ -270,9 +271,11 @@ controls inside a DataSource dialog via the `pj_parser_slot` placeholder. ### Threading -All plugin callbacks are called on the host's thread. The host guarantees -single-threaded access per instance. Plugins needing internal threads must -synchronize and call host methods only from callback context. +Each slot's thread rule is its tag; the per-family tables on +`PJ_data_source_vtable_t` and `PJ_toolbox_vtable_t` are normative. A DataSource +`start` may run on a worker and `stop` on the polling thread; per-topic +`set_active_topics` runs concurrently with `poll`. Plugins needing internal +threads must synchronize and call host methods only where those tables allow. ## 6. Capability System diff --git a/pj_plugins/docs/data-source-guide.md b/pj_plugins/docs/data-source-guide.md index d4996f51..a447583d 100644 --- a/pj_plugins/docs/data-source-guide.md +++ b/pj_plugins/docs/data-source-guide.md @@ -58,11 +58,12 @@ prevent runtime failures and confusing host behaviour. **MUST** - Return `PJ::okStatus()` on success and `PJ::unexpected("reason")` on failure from every fallible method. -- Call `runtimeHost().notifyState(...)` on every state transition you trigger. -- Call host methods (`writeHost()`, `runtimeHost()`) only from the host's - callback thread — the same thread that invoked `onStart`/`onPoll`/`onStop`/ - `importData`/`pause`/`resume`. -- Make `onStop()` idempotent. +- Call `[stream-thread]` host methods (`writeHost()`, and `runtimeHost()` + except `reportMessage`/`requestStop`/`isStopRequested`/`notifyState`) only + while the host is inside `start()`/`poll()`/`stop()` — i.e. from + `onStart`/`onPoll`/`onStop`/`importData`, one thread at a time. +- Make `onStop()` idempotent, and release what a failed `onStart()` acquired: + the host may destroy the instance without calling `stop()`. - Persist all state needed for headless restart in `saveConfig()`. The host must be able to `loadConfig(saved) → start()` without showing a dialog. @@ -70,8 +71,9 @@ prevent runtime failures and confusing host behaviour. - Throw exceptions across virtual overrides — the SDK trampolines catch them, but the host receives a generic error and the plugin loses the chance to report a useful reason. -- Call host methods from a background thread you spawned. Buffer in plugin - memory and flush from `onPoll()`. `PJ::sdk::DrainQueue` / `LatestValueSlot` +- Call `[stream-thread]` host methods from a thread you spawned, unless the + host's `start()`/`poll()`/`stop()` call is blocked waiting for that thread. + Buffer in plugin memory and flush from `onPoll()`. `PJ::sdk::DrainQueue` / `LatestValueSlot` in `pj_plugins/sdk/streaming_source.hpp` provide swap-drain containers for this handoff. `DelegatedIngestCache` in the same header wraps `ensureParserBinding` + `pushMessage` for delegated-ingest sources. @@ -351,7 +353,7 @@ class UdpReceiver : public PJ::StreamSourceBase { PJ::Status onPoll() override { // Flush buffered data into the host. Called periodically from the - // host's thread — this is the only place we may call host methods. + // host's polling thread — call host methods here, never from recvLoop(). auto batch = buffer_.drain(); while (!batch.empty()) { const auto& msg = batch.front(); @@ -425,18 +427,20 @@ changes, only a different config value. **Threading and `onPoll()` semantics.** The host calls `onPoll()` periodically on its thread to flush buffered data. Asynchronous I/O sources (sockets, hardware) must manage their own receive thread and hand off buffered data in -`onPoll()`. Host methods (`pushMessage`, `appendRecord`, etc.) must only be called -from `onPoll()` / the host's thread, never from the background thread. +`onPoll()`. `[stream-thread]` host methods (`pushMessage`, `appendRecord`, etc.) +must only be called from `onStart()`/`onPoll()`/`onStop()`, never from the +background thread: between polls the host flushes the same host objects on its +polling thread. Key traits of `StreamSourceBase`: - `capabilities()` automatically includes `kCapabilityContinuousStream`; you provide additional flags via `extraCapabilities()`. - `onStart()` opens connections and creates topics or parser bindings. - `onPoll()` flushes buffered data into the host — drain what your plugin - has accumulated and return immediately. Must not block. Host methods - (`pushMessage`, `appendRecord`, etc.) may only be called from this - callback. -- `onStop()` tears down connections. Must be idempotent. + has accumulated and return immediately. Must not block. +- `onStop()` tears down connections. Must be idempotent. It may run on the + polling thread, not the thread that ran `onStart()`, and is skipped when + `onStart()` fails. - `stop()`, `start()`, `poll()`, and `currentState()` are managed by the base class. - **Pause/resume** are NOT wired by `StreamSourceBase`. To support pause, @@ -470,7 +474,7 @@ Access via `runtimeHost()`. Use this for lifecycle coordination and diagnostics. | `progressStart(label, total, cancellable)` | Begin a progress bar. | | `progressUpdate(step)` | Advance progress. Returns false if cancelled. | | `progressFinish()` | End the progress sequence. | -| `notifyState(state)` | Tell the host your state changed. | +| `notifyState(state)` | Tell the host your state changed. Advisory: a host may ignore it. | | `requestStop(terminal_state, reason)` | Ask the host to stop you (self-terminate). | | `isStopRequested()` | Check if the host wants you to stop. | | `ensureParserBinding(request)` | Bind a parser for delegated ingest (see below). | @@ -483,8 +487,10 @@ Access via `runtimeHost()`. Use this for lifecycle coordination and diagnostics. ### Pause and resume -Override `pause()` and `resume()`, and declare `kCapabilitySupportsPause` in -your capabilities: +`pause()`/`resume()` are advisory: a host may never call them (the reference +host pauses a stream host-side, by redirecting its writes), so a source must +work without them. To offer them anyway, override `pause()` and `resume()`, and +declare `kCapabilitySupportsPause` in your capabilities: ```cpp uint64_t extraCapabilities() const override { @@ -545,9 +551,9 @@ const void* pluginExtension(std::string_view id) override { } // topic_subscription_ext_ is a PJ_topic_subscription_v1_t{sizeof(...), &setActiveTopicsThunk}; // setActiveTopicsThunk receives the PLUGIN INSTANCE as ctx (not the extension -// pointer). It is entered on the host thread — marshal the desired set to your -// stream thread (e.g. a command queue drained in onPoll), do not touch your -// subscription state directly here. +// pointer). It is entered on the GUI thread, concurrently with onPoll — store +// the desired set in a PJ::sdk::LatestValueSlot drained in onPoll (latest +// wins; not a command queue), do not touch your subscription state here. ``` ### Periodic polling @@ -568,6 +574,27 @@ PJ::Status loadConfig(std::string_view json) override { } ``` +The config is a UTF-8 JSON object. The rules (normative text on +`save_config`/`load_config` in `data_source_protocol.h`): + +- `loadConfig("")` means "no saved config" and is still called (a stream's + first run does). Keep your defaults; do not fail. +- `loadConfig()` may be called several times before `start()`. After an + accepted dialog the host passes your **embedded dialog's** `saveConfig()` + output (plus the host keys below), so source and dialog must share one format. +- Accept configs written by older versions of your plugin: layouts and saved + settings are replayed verbatim, apart from the host keys. +- Top-level keys starting with `_` belong to the host; do not define your own. + The reference host writes `"filepath"` (file sources: the file to import), + `"_parser_config"` (a string: the embedded parser dialog's config; pass it as + `parser_config_json` to `ensureParserBinding`, read it with + `PJ::sdk::parserConfigOverride`) and `"__pj_saved_filepath"` (layout replay + only, see below). `"__pj_fanout"` is the one host-defined key a plugin + writes, to ask for one instance per child config. +- `saveConfig()` may return `"filepath"` and `"_parser_config"` (a file layout + replays `saveConfig()` output, so that is how its parser config survives), + never `"__pj_saved_filepath"` (the reference host strips it). + ### Progress reporting Report progress during long operations (e.g. file imports): @@ -617,22 +644,26 @@ Both dialogs share a window but persist config independently: `ConfigEnvelope.source_config` for the source and `ConfigEnvelope.parser_binding` for the parser. See `pj_plugins/docs/message-parser-guide.md` § "Dialog integration". -## State Machine +## Lifecycle Signals -``` -idle --> configuring --> starting --> running --> stopping --> stopped - | ^ - pause | | resume - v | - paused +The host acts on these signals only: -Any state --> failed -``` +- **`start()` return.** For a finite import, returning from `start()` ends the + import; an error fails it. For a stream, an error aborts the launch. +- **`poll()` return.** An error ends the stream. +- **`runtimeHost().requestStop(kStopped | kFailed, reason)`** ends the stream + from the plugin side. `kStopped` is an orderly end; any other state reports + `reason` as the failure. +- **`runtimeHost().isStopRequested()`** tells you the host (user stop, cancel) + wants you to stop. Check it during long operations. +- **`completeIngest()`** reports a finite ingest's terminal outcome for + caching (see *Ingest completion and rollback* in ARCHITECTURE.md). -- **stopped** and **failed** are terminal — create a new instance to restart. -- Always call `runtimeHost().notifyState()` when you transition. -- Use `runtimeHost().requestStop(kStopped, reason)` to self-terminate. -- Check `runtimeHost().isStopRequested()` during long operations. +`notifyState()`, `currentState()` and the `PJ_data_source_state_t` states +(`idle`, `starting`, `running`, `paused`, `stopped`, `failed`, ...) are +advisory: a host may ignore them, so report them if you like but do not +depend on them. An instance is started at most once — the host creates a new +instance to restart. ## Capability Flags Reference @@ -741,18 +772,26 @@ remains an ABI escape hatch. The rvalue-reference form is the documented default ## Threading Model -All plugin callbacks — `start()`, `stop()`, `poll()`, `pause()`, `resume()`, -`loadConfig()`, `saveConfig()` — are called **on the host's thread**. The host -guarantees single-threaded access per plugin instance: no two callbacks will -overlap for the same instance. - -Write host and runtime host methods (`appendRecord()`, `ensureParserBinding()`, -etc.) must be called from the same thread that invoked the callback. Do not -cache host views and call them from a background thread. +The normative per-slot table is on `PJ_data_source_vtable_t` in +`pj_base/include/pj_base/data_source_protocol.h`. In short: + +- `create`, `bind`, `loadConfig()`, `saveConfig()`, `getDialog()`, + `capabilities()` and `destroy` run on the GUI thread, never while + `start()`/`poll()`/`stop()` runs. +- `start()` may run on a worker (a finite import's `importData()` blocks there + for the whole import) or on the GUI thread (a stream's `onStart()`, which + must return promptly). `poll()` runs on one host polling thread; `stop()` may + run on that polling thread. These never overlap each other. +- `setActiveTopics` (`pj.topic_subscription.v1`) runs on the GUI thread + **concurrently with `poll()`**; `getPluginExtension` may run on any thread. +- `[stream-thread]` host methods (`appendRecord()`, `ensureParserBinding()`, + `pushMessage()`, progress, ...) only while a `start()`/`poll()`/`stop()` call + is in progress, one thread at a time. `[thread-safe]` ones + (`reportMessage()`, `requestStop()`, `isStopRequested()`, `notifyState()`) + from any thread. If your plugin needs internal threading (e.g. a network receive thread), you -must synchronize access to shared state yourself and only call host methods -from the callback thread. +must synchronize access to shared state yourself and hand data to `onPoll()`. ## Lifecycle Invariants @@ -761,16 +800,17 @@ The host guarantees the following call ordering: 1. `create()` — always first. 2. `bind(registry)` — before `start()`. The SDK default bind resolves `"pj.source_write.v1"` and `"pj.runtime.v1"` from the service registry. -3. `load_config()` — before `start()`, may be called multiple times. -4. `start()` — transitions from idle/configuring to starting. -5. `poll()` — only while running (after `start()` returns success). -6. `pause()`/`resume()` — only while running/paused. -7. `stop()` — may be called at any time after `start()`. +3. `load_config()` — before `start()`, may be called multiple times, possibly + with an empty string. +4. `start()` — at most once. +5. `poll()` — continuous streams only, after `start()` returns success. +6. `pause()`/`resume()` — advisory; a host may never call them. +7. `stop()` — after `start()` and the last `poll()` have returned; may be + skipped when `start()` failed or a finite import finished. 8. `destroy()` — always last. -The host will never call `poll()` before `start()`, nor `start()` after -`stop()`. Terminal states (`stopped`, `failed`) are final — the host creates -a new instance to restart. +The host will never call `poll()` before `start()`, nor `start()` twice. To +restart, the host creates a new instance. ## Error Handling diff --git a/pj_plugins/docs/toolbox-guide.md b/pj_plugins/docs/toolbox-guide.md index cbea0e1c..382185d1 100644 --- a/pj_plugins/docs/toolbox-guide.md +++ b/pj_plugins/docs/toolbox-guide.md @@ -67,11 +67,13 @@ the contract is broader than the other families. `ArrowArrayHolder` so the release callbacks fire on scope exit. - Persist tool state in `saveConfig()` so a layout reload restores the same view. The host has no ambient persistence. -- Only call host methods from the host's callback thread. Background work - must marshal back through the host thread to write data. +- Call each host method only from the threads its tag allows (see + [Threading Model](#threading-model)). **MUST NOT** - Throw exceptions across virtual overrides. +- Depend on `onDataChanged()` for correctness: it is advisory and a host may + never call it. - Hold an `ArrowSchema*` / `ArrowArray*` past the scope of the holders that own them — the host may reuse the underlying buffers. - Treat `catalogSnapshot()` as live data. Snapshots are immutable views at @@ -500,13 +502,21 @@ No exceptions cross the C ABI boundary. ## Threading Model -All plugin callbacks — `bind()`, -`loadConfig()`, `saveConfig()`, `getDialog()` — are called **on the host's -thread**. The host guarantees single-threaded access per plugin instance. - -Toolbox host and runtime host methods must be called from the same thread that -invoked the callback. If your plugin uses internal threading, synchronize -access and only call host methods from the host's thread. +The normative table is on `PJ_toolbox_vtable_t` in +`pj_base/include/pj_base/toolbox_protocol.h`. In short: + +- Every plugin callback (`bind()`, `loadConfig()`, `saveConfig()`, + `getDialog()`, `onDataChanged()`, dialog events) runs on the GUI thread, one + at a time per instance. `onDataChanged()` is advisory. +- `toolboxHost()` (`pj.toolbox_write.v1`) is `[stream-thread]`: call it from + any thread, typically your own worker, but one thread at a time per host + object — serialize with your own mutex if you fan out. +- `runtimeHost().reportMessage()`, `notifyDataChanged()` and the + `create/release/discardParserIngest` (dataset-ingest) calls are + `[thread-safe]`. The ingest view they return is driven one thread at a time. +- Every other service (object reads, playback, plot tabs, viewport, data + processors, colormaps, ...) is `[main-thread]` unless its slot says + otherwise: call it from a callback, not from a thread you spawned. ## Testing @@ -608,6 +618,6 @@ to `source` for one release; new code uses `pj_base/sdk/source/`. | `read_series_arrow` succeeds but later code crashes accessing the data | `ArrowSchema` / `ArrowArray` released early, or held by raw pointer past holder scope | Use `PJ::sdk::ArrowSchemaHolder` / `ArrowArrayHolder` and keep them alive while the data is read | | Catalog reads stale data immediately after writing | `catalogSnapshot()` was acquired before the write | Reacquire the snapshot after `notifyDataChanged()` | | Duplicate source names appear in the UI | `createDataSource(name)` always creates a new dataset, even when the display name already exists | Check the catalog first; pick a unique derived-data name or surface a confirmation in the dialog | -| Plugin works in tests but crashes in the host | Host method called from a thread the toolbox spawned | Marshal back to the host thread (use the dialog's `onTick` or a host-thread queue) | +| Plugin works in tests but crashes in the host | A `[main-thread]` service called from a thread the toolbox spawned, or two threads inside one `[stream-thread]` host at once | Marshal `[main-thread]` calls back to the GUI thread (the dialog's `onTick` or a host-thread queue); serialize `[stream-thread]` calls with a mutex | | Bulk transform output is one row at a time | Output written record-by-record instead of via Arrow | Build an `ArrowArrayStream` and use `appendArrowStream()` for the output | | Plugin restarts but the tool's view is empty | Config not persisted | Round-trip every UI-relevant field through `saveConfig()` / `loadConfig()` | diff --git a/pj_plugins/include/pj_plugins/host/data_source_handle.hpp b/pj_plugins/include/pj_plugins/host/data_source_handle.hpp index 348902c6..c7a514df 100644 --- a/pj_plugins/include/pj_plugins/host/data_source_handle.hpp +++ b/pj_plugins/include/pj_plugins/host/data_source_handle.hpp @@ -11,7 +11,7 @@ * if (auto s = handle.bind(registry.view()); !s) { ... } * if (auto s = handle.loadConfig(json); !s) { ... } * if (auto s = handle.start(); !s) { ... } - * while (handle.currentState() == PJ::DataSourceState::kRunning) { + * while (!stop_requested) { // the host's own flag, set via request_stop * if (auto s = handle.poll(); !s) { ... } * } * handle.stop();