SHM: obtain_shm_provider(true) before the provider is Ready leaves the implicit provider permanently Initializing (1.10.1)

[ciamsm1]

Describe the bug

In 1.10.1, calling obtain_shm_provider(true) (the new blocking argument) before the session's implicit SHM provider has finished initializing returns a Ready provider, but leaves the session's provider state permanently Initializing:

• every later non-blocking obtain_shm_provider(false) returns Initializing for the life of the session
• a second blocking call returns Error

Calling the blocking variant after the provider is already Ready is fine. Only a blocking call made while the state is still Initializing triggers this.

The implicit provider is documented as lazily initialized and available once initialization completes, so a state that never leaves Initializing does not look like the intended contract.

To reproduce

Self-contained, no router needed (zenoh-cpp over zenoh-c):

repro.cpp

// Minimal reproducer: obtain_shm_provider(true) before the provider is Ready leaves the
// session's implicit SHM provider permanently "Initializing".
#include <chrono>
#include <cstdio>
#include <string>
#include <thread>
#include <zenoh.hxx>

static const char* state(const std::variant<zenoh::SharedShmProvider, zenoh::ShmProviderNotReadyState>& r) {
    if (std::holds_alternative<zenoh::SharedShmProvider>(r)) return "Ready";
    switch (std::get<zenoh::ShmProviderNotReadyState>(r)) {
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_DISABLED: return "Disabled";
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_INITIALIZING: return "Initializing";
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_ERROR: return "Error";
    }
    return "?";
}

static zenoh::Session open_session() {
    auto c = zenoh::Config::create_default();
    c.insert_json5("scouting/multicast/enabled", "false");
    c.insert_json5("transport/shared_memory/enabled", "true");
    c.insert_json5("transport/shared_memory/transport_optimization/enabled", "true");
    return zenoh::Session::open(std::move(c));
}

int main(int argc, char** argv) {
    const std::string order = argc > 1 ? argv[1] : "blocking-first";
    auto s = open_session();
    std::printf("order: %s\n", order.c_str());
    if (order == "blocking-first") {
        auto first = s.obtain_shm_provider(true);
        std::printf("  blocking obtain                 -> %s\n", state(first));
        int ready = 0, init = 0, err = 0;
        for (int i = 0; i < 200; ++i) {
            std::string st = state(s.obtain_shm_provider(false));
            ready += st == "Ready"; init += st == "Initializing"; err += st == "Error";
            std::this_thread::sleep_for(std::chrono::milliseconds(10));
        }
        std::printf("  non-blocking x200 over 2 s      -> Ready %d, Initializing %d, Error %d\n", ready, init, err);
        std::printf("  blocking obtain again           -> %s\n", state(s.obtain_shm_provider(true)));
    } else {
        int polls = 0; std::string st;
        do { st = state(s.obtain_shm_provider(false)); ++polls; if (st != "Ready") std::this_thread::sleep_for(std::chrono::milliseconds(1)); }
        while (st != "Ready" && polls < 3000);
        std::printf("  non-blocking polling            -> %s after %d polls\n", st.c_str(), polls);
        int ready = 0;
        for (int i = 0; i < 200; ++i) ready += std::string(state(s.obtain_shm_provider(false))) == "Ready";
        std::printf("  non-blocking x200               -> Ready %d / 200\n", ready);
        std::printf("  blocking obtain after Ready     -> %s\n", state(s.obtain_shm_provider(true)));
    }
    return 0;
}

g++ -std=c++17 -DZENOHCXX_ZENOHC -I<prefix>/include repro.cpp -L<prefix>/lib -lzenohc -o repro
./repro blocking-first
./repro nonblocking-first

Output (identical over 3 runs):

order: blocking-first
  blocking obtain                 -> Ready
  non-blocking x200 over 2 s      -> Ready 0, Initializing 200, Error 0
  blocking obtain again           -> Error
order: nonblocking-first
  non-blocking polling            -> Ready after 6 polls
  non-blocking x200               -> Ready 200 / 200
  blocking obtain after Ready     -> Ready

The nonblocking-first run is the control: the same session reaches Ready after a few polls and stays Ready, and a blocking call made after that also returns Ready.

Likely cause

LazyShmProvider::try_get_provider creates a flume::bounded(1) channel, the init task sends the provider on it once, and the state moves to Ready only when a non-blocking caller's try_recv() succeeds:

• https://github.com/eclipse-zenoh/zenoh/blob/1.10.1/io/zenoh-transport/src/common/shm/interop.rs#L108-L148

The blocking path in zenoh-c receives that single message through the cloned receiver and returns it to the caller, but nothing moves the shared state to Ready:

• https://github.com/eclipse-zenoh/zenoh-c/blob/1.10.1/src/session.rs#L148-L175

After that the channel is empty and the sender is dropped, so every later try_recv() hits the Err(_) arm (interop.rs#L148) and returns Initializing again. A second blocking call gets None from the drained channel and reports Error.

Impact

Observed. An application that uses the implicit provider when it is Ready and otherwise sends a regular (non-SHM) payload, and that did one early blocking obtain right after opening the session, never saw the provider become Ready through its non-blocking path. Measured with a 6 MiB payload at 30 Hz for 12 s, 3 runs: 360 / 360 frames were received as non-SHM in every run (vs. 1 cold frame and the rest SHM without the early blocking call). No warning or error was logged. Publish p99 rose from about 0.9 ms to about 5.5 ms.

Expected, not directly tested.

• An application that treats "no Ready provider" as a failure would fail on every publish for the rest of the session.
• Transport-level implicit SHM for large regular payloads appears to be affected as well: in the run above the payloads were well above message_size_threshold with transport_optimization enabled, yet the receiver saw no SHM samples, and the transport path uses the same try_get_provider. We did not isolate this case separately.

System info

• zenoh / zenoh-c / zenoh-cpp 1.10.1, rustc 1.97.1
• zenoh-c built with shared-memory and unstable
• Linux x86_64, kernel 5.15.0
• 1.9.0 is not affected only because obtain_shm_provider had no blocking argument there

[PushpakAg]

It seems like calling obtain_shm_provider(true) too early causes a permanent Initializing state. Have you checked if this behavior is consistent across different sessions or just a one-time occurrence?

[ciamsm1]

Yes. I added a same-process, multi-session test to check this specifically.

The behavior is reproducible across sessions, and in this reproducer it appears to be local to the affected session.

I ran the test twice. Both runs printed:

s1 blocking obtain             -> Ready
s1 non-blocking poll (30 s)    -> Initializing after 30002 ms

s2 non-blocking poll (10 s)    -> Ready after 10 ms   [while s1 is still open]
s1 non-blocking (after s2)     -> Initializing

s3 blocking obtain             -> Ready
s3 non-blocking poll (10 s)    -> Initializing after 10001 ms

5 more blocking-first sessions -> stuck 5 / 5

So far:

• blocking-first reproduced in 7/7 sessions per process, in 2 independent runs
• the original minimal reproducer also reproduced in 3/3 fresh-process runs
• an affected session did not recover during 30 seconds of polling
• opening another session in the same process does not poison the new session: if its first access is non-blocking, it reaches Ready normally
• opening another session also does not recover the already affected session

This seems consistent with the analysis in the issue body: the blocking path consumes the one-shot flume::bounded(1) initialization result, but the cached lazy-provider state is not transitioned to Ready, so later non-blocking accesses for that session continue to observe Initializing.

Multi-session reproducer

// Follow-up for #2800: is the stuck state per-session, and does it persist over time?
#include <chrono>
#include <cstdio>
#include <string>
#include <thread>
#include <vector>
#include <zenoh.hxx>
using namespace std::chrono;
static const char* state(const std::variant<zenoh::SharedShmProvider, zenoh::ShmProviderNotReadyState>& r) {
    if (std::holds_alternative<zenoh::SharedShmProvider>(r)) return "Ready";
    switch (std::get<zenoh::ShmProviderNotReadyState>(r)) {
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_DISABLED: return "Disabled";
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_INITIALIZING: return "Initializing";
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_ERROR: return "Error";
    }
    return "?";
}
static zenoh::Session open_session() {
    auto c = zenoh::Config::create_default();
    c.insert_json5("scouting/multicast/enabled", "false");
    c.insert_json5("transport/shared_memory/enabled", "true");
    return zenoh::Session::open(std::move(c));
}
static std::string poll_until_ready(zenoh::Session& s, int max_ms) {
    auto t0 = steady_clock::now(); std::string st;
    while (true) { st = state(s.obtain_shm_provider(false)); if (st == "Ready") break;
        if (duration_cast<milliseconds>(steady_clock::now() - t0).count() > max_ms) break;
        std::this_thread::sleep_for(milliseconds(5)); }
    return st + " after " + std::to_string(duration_cast<milliseconds>(steady_clock::now() - t0).count()) + " ms";
}
int main() {
    // A: blocking-first on session 1, then check session 1 for 30 s
    auto s1 = open_session();
    std::printf("s1 blocking obtain            -> %s\n", state(s1.obtain_shm_provider(true)));
    std::printf("s1 non-blocking poll (30 s)   -> %s\n", poll_until_ready(s1, 30000).c_str());
    // B: new session 2 in the same process (s1 still open), non-blocking first
    auto s2 = open_session();
    std::printf("s2 non-blocking poll (10 s)   -> %s   [s1 still open]\n", poll_until_ready(s2, 10000).c_str());
    std::printf("s1 non-blocking (after s2)    -> %s\n", state(s1.obtain_shm_provider(false)));
    // C: session 3, blocking-first again
    auto s3 = open_session();
    std::printf("s3 blocking obtain            -> %s\n", state(s3.obtain_shm_provider(true)));
    std::printf("s3 non-blocking poll (10 s)   -> %s\n", poll_until_ready(s3, 10000).c_str());
    // D: 5 sessions opened in sequence, each blocking-first
    int stuck = 0;
    for (int i = 0; i < 5; ++i) { auto s = open_session(); s.obtain_shm_provider(true);
        stuck += poll_until_ready(s, 2000).rfind("Initializing", 0) == 0; }
    std::printf("5 more sessions blocking-first -> stuck %d / 5 (2 s poll each)\n", stuck);
    return 0;
}

[ciamsm1]

I tightened the multi-session reproducer and reran it three times. Changes:

• non-blocking polling now stops on any terminal state (Ready / Error / Disabled), not only on Ready
• a control session (non-blocking first access) runs first and is destroyed before the other cases
• case D is labelled precisely: additional sessions created while s1/s2/s3 are still alive
• each run is a separate process under an external timeout 120, so a blocking obtain that never returned would show up as exit code 124

Three runs gave identical results (exit code 0 each, so no blocking obtain hung):

control: non-blocking first (10 s poll)    -> Ready after 10 ms
A  s1 blocking obtain                       -> Ready
A  s1 non-blocking poll (30 s)              -> Initializing after 30003 ms
B  s2 non-blocking first (10 s poll)        -> Ready after 10 ms   [s1 still open]
B  s1 non-blocking again (after s2 Ready)   -> Initializing
C  s3 blocking obtain                       -> Ready
C  s3 non-blocking poll (10 s)              -> Initializing after 10002 ms
D  5 more blocking-first sessions           -> still Initializing after 2 s: 5 / 5

Reading it in order:

• first-access mode: with non-blocking first access the provider reaches Ready in about 10 ms (control). With blocking first access, the blocking call itself returns Ready, but later non-blocking accesses on that session report Initializing (A).
• session state: the affected session did not leave Initializing during 30 s of polling (A).
• propagation: while s1 is affected, a new session in the same process with non-blocking first access still reaches Ready normally (B). That session reaching Ready does not change s1.
  • So in this reproducer the condition does not look like a single process-global state. It looks like each session that takes the blocking-first path is affected on its own.
• repeatability: blocking-first reproduced in every such session (C, and 5/5 in D).

Reproducer (v2)

// Follow-up for #2800: first-access mode -> session state -> propagation to other sessions -> repeatability.
// Run under an external timeout (e.g. `timeout 120 ./multi2`) in case a blocking obtain itself never returns.
#include <chrono>
#include <cstdio>
#include <string>
#include <thread>
#include <zenoh.hxx>
using namespace std::chrono;

static const char* state(const std::variant<zenoh::SharedShmProvider, zenoh::ShmProviderNotReadyState>& r) {
    if (std::holds_alternative<zenoh::SharedShmProvider>(r)) return "Ready";
    switch (std::get<zenoh::ShmProviderNotReadyState>(r)) {
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_DISABLED: return "Disabled";
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_INITIALIZING: return "Initializing";
        case zenoh::ShmProviderNotReadyState::SHM_PROVIDER_ERROR: return "Error";
    }
    return "?";
}

static zenoh::Session open_session() {
    auto c = zenoh::Config::create_default();
    c.insert_json5("scouting/multicast/enabled", "false");
    c.insert_json5("transport/shared_memory/enabled", "true");
    return zenoh::Session::open(std::move(c));
}

// Non-blocking polling. Stops on a terminal state (Ready / Error / Disabled) or when max_ms has elapsed.
static std::string poll(zenoh::Session& s, int max_ms) {
    const auto t0 = steady_clock::now();
    std::string st;
    while (true) {
        st = state(s.obtain_shm_provider(false));
        if (st == "Ready" || st == "Error" || st == "Disabled") break;
        if (duration_cast<milliseconds>(steady_clock::now() - t0).count() >= max_ms) break;
        std::this_thread::sleep_for(milliseconds(5));
    }
    const auto ms = duration_cast<milliseconds>(steady_clock::now() - t0).count();
    return st + " after " + std::to_string(ms) + " ms";
}

int main() {
    // Control: fresh session, non-blocking first access. Destroyed before the other cases.
    {
        auto s = open_session();
        std::printf("control: non-blocking first (10 s poll)    -> %s\n", poll(s, 10000).c_str());
    }

    // A: fresh session, blocking first access, then non-blocking polling on the same session.
    auto s1 = open_session();
    std::printf("A  s1 blocking obtain                       -> %s\n", state(s1.obtain_shm_provider(true)));
    std::printf("A  s1 non-blocking poll (30 s)              -> %s\n", poll(s1, 30000).c_str());

    // B: new session in the same process while s1 is still open, non-blocking first access.
    auto s2 = open_session();
    std::printf("B  s2 non-blocking first (10 s poll)        -> %s   [s1 still open]\n", poll(s2, 10000).c_str());
    std::printf("B  s1 non-blocking again (after s2 Ready)   -> %s\n", state(s1.obtain_shm_provider(false)));

    // C: another new session (s1, s2 still open), blocking first access again.
    auto s3 = open_session();
    std::printf("C  s3 blocking obtain                       -> %s\n", state(s3.obtain_shm_provider(true)));
    std::printf("C  s3 non-blocking poll (10 s)              -> %s\n", poll(s3, 10000).c_str());

    // D: 5 additional sessions in the same process, while s1/s2/s3 are still alive.
    //    Each new session uses blocking-first, is polled for 2 s, then destroyed.
    int stuck = 0;
    for (int i = 0; i < 5; ++i) {
        auto s = open_session();
        s.obtain_shm_provider(true);
        const std::string result = poll(s, 2000);
        if (result.rfind("Initializing", 0) == 0) {
            ++stuck;
        }
    }
    std::printf("D  5 more blocking-first sessions           -> still Initializing after 2 s: %d / 5\n", stuck);
    return 0;
}