diff --git a/src/plugins/plugin_utils/hand_injector.cpp b/src/plugins/plugin_utils/hand_injector.cpp index ea8b84a7bf..73060606af 100644 --- a/src/plugins/plugin_utils/hand_injector.cpp +++ b/src/plugins/plugin_utils/hand_injector.cpp @@ -10,17 +10,34 @@ #include #include #include +#include namespace plugin_utils { namespace { +//! Callers that cannot link plugin_utils recognize it by name() and read code().value(). +class XrResultCategory final : public std::error_category +{ +public: + const char* name() const noexcept override + { + return "XrResult"; + } + std::string message(int value) const override + { + return std::to_string(value); + } +}; + void CheckXrResult(XrResult result, const char* message) { + static const XrResultCategory category; if (XR_FAILED(result)) { - throw std::runtime_error(std::string(message) + " failed with XrResult: " + std::to_string(result)); + // what() reads " failed with XrResult: ". + throw std::system_error(result, category, std::string(message) + " failed with XrResult"); } } } // namespace diff --git a/src/plugins/remote_device_bridge/AGENTS.md b/src/plugins/remote_device_bridge/AGENTS.md index 7239a9c741..5713475589 100644 --- a/src/plugins/remote_device_bridge/AGENTS.md +++ b/src/plugins/remote_device_bridge/AGENTS.md @@ -10,6 +10,9 @@ SPDX-License-Identifier: Apache-2.0 - `session/protocol_frame.*` and `session/protocol_payloads.*` are the only serialization points. Every frame is encoded and decoded there and nowhere else; changing the wire format means changing them and `tests/cpp/plugins/remote_device_bridge/test_wire_codec.cpp` together. Sessions hold an `ITransport` (`session/inc/remote_plugin_session/transport.hpp`) and only call `send/receive/shutdown` on complete frames; TCP framing lives in `TcpTransport`. - WebSocket is a second transport, not a second protocol: `WebSocketTransport` carries exactly one wire frame per binary message (no TLS; terminate `wss://` in a proxy). Clients pick it with `--remote=ws://host:port[/path]` via `parse_remote`/`connect_transport`; the bridge serves it on a separate `--ws-listen` port sharing the fork-per-client loop and `--max-clients`. Never sniff the protocol on the TCP port. The server-side HTTP upgrade runs in the first `receive()` so the accept loop never blocks on a client. - Never add `OpenXR::openxr_loader`, `Teleop::plugin_utils` or DeviceIO session libraries to `Teleop::remote_plugin_session`: a plugin using `--remote` must run without an OpenXR runtime. -- ERROR on channel 0 ends the session; ERROR on a channel id refuses only that OPEN and the session and its other channels continue. Refuse per channel whenever the session itself is still sound. +- `Teleop::remote_device_bridge_server` (`server/`) gets each client's session from a `SessionFactory`; only `app/main.cpp` builds the OpenXR session, with `plugin_utils::make_openxr_plugin_session_factory`. Keep the OpenXR loader and `plugin_utils` out of `server/` so `tests/cpp/plugins/remote_device_bridge_server` can run the real fork-per-client bridge against a fake session. +- ERROR on channel 0 ends the session; ERROR on a channel id refuses only that OPEN (schema above `--max-schema-bytes`, a hand someone else owns) and the session and its other channels continue. Refuse per channel whenever the session itself is still sound. - `RemotePluginSession` never throws on the sample path for transport failures (`push`, `push_buffer`, `query`, `update`): after one they become no-ops; only channel creation and the constructor throw. Caller bugs (null joints or buffer) still throw `std::invalid_argument`. -- Timestamps cross the wire in the sender's `CLOCK_MONOTONIC`; the receiver maps them with `ClockOffsetEstimator`, which prefers PING/PONG exchanges (minimum RTT) over the one-way minimum of `local_recv - remote_send`; the one-way minimum hides the minimum network delay. Do not push raw remote timestamps into OpenXR. +- The bridge (`server/`) talks to OpenXR from one thread per client (the thread running that `ClientSession`). Socket reading happens on a helper thread that only parses frames into a queue. Keep it that way; `OpenXRPluginSession` and its channels are not thread-safe. +- Timestamps cross the wire in the sender's `CLOCK_MONOTONIC`; the receiver maps them with `ClockOffsetEstimator`. The bridge's offset comes from PING/PONG (the minimum-RTT recent exchange, `--clock-source=two-way`); the one-way minimum of `local_recv - remote_send` hides the minimum network delay and is only the fallback (no PONG in time) or the `--clock-source=one-way` A/B baseline. Do not push raw remote timestamps into OpenXR. +- One bridge process serves N clients. The accept loop never creates threads or OpenXR objects so it may `fork()` a child per client (the isolation of N local plugin processes). Do not serve clients on threads: the OpenXR loader allows one `XrInstance` per process (a second one fails with -10). State the children must share (hand ownership) lives in a `MAP_SHARED` page mapped before the first fork; the parent clears a dead child's entries when it reaps it. Sample paths are lossy by design: per-channel latest-wins queues on the client, coalescing and a staleness guard on the bridge; control frames are never dropped or reordered. diff --git a/src/plugins/remote_device_bridge/CMakeLists.txt b/src/plugins/remote_device_bridge/CMakeLists.txt index 0e516622d0..40f20ed96b 100644 --- a/src/plugins/remote_device_bridge/CMakeLists.txt +++ b/src/plugins/remote_device_bridge/CMakeLists.txt @@ -9,3 +9,9 @@ if(WIN32) endif() add_subdirectory(session) +add_subdirectory(server) +add_subdirectory(app) + +install(FILES plugin.yaml + DESTINATION plugins/remote_device_bridge +) diff --git a/src/plugins/remote_device_bridge/app/CMakeLists.txt b/src/plugins/remote_device_bridge/app/CMakeLists.txt new file mode 100644 index 0000000000..e6f95e2845 --- /dev/null +++ b/src/plugins/remote_device_bridge/app/CMakeLists.txt @@ -0,0 +1,18 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +add_executable(remote_device_bridge + main.cpp +) + +target_link_libraries(remote_device_bridge + PRIVATE + Teleop::plugin_utils + Teleop::remote_device_bridge_server + Teleop::remote_plugin_session + log_bridge::log_bridge_core +) + +install(TARGETS remote_device_bridge + RUNTIME DESTINATION plugins/remote_device_bridge +) diff --git a/src/plugins/remote_device_bridge/app/main.cpp b/src/plugins/remote_device_bridge/app/main.cpp new file mode 100644 index 0000000000..6fa6ce3f24 --- /dev/null +++ b/src/plugins/remote_device_bridge/app/main.cpp @@ -0,0 +1,188 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace +{ + +static_assert(ATOMIC_BOOL_LOCK_FREE == 2, "lock-free atomic bool is required for signal safety"); + +std::atomic g_stop_requested{ false }; + +void signal_handler(int signal) +{ + if (signal == SIGINT || signal == SIGTERM) + { + g_stop_requested.store(true, std::memory_order_relaxed); + } +} + +std::optional flag_value(std::string_view arg, std::string_view name) +{ + if (arg.starts_with(name) && arg.size() > name.size() && arg[name.size()] == '=') + { + return std::string(arg.substr(name.size() + 1)); + } + return std::nullopt; +} + +//! Whole-string integer in [min, max of T]: std::stoi accepts trailing junk and std::stoul wraps "-1". +template +T parse_integer(std::string_view flag, const std::string& value, T min) +{ + size_t used = 0; + long long parsed = 0; + try + { + parsed = std::stoll(value, &used); + } + catch (const std::exception&) + { + used = 0; + } + if (used == 0 || used != value.size() || parsed < static_cast(min) || + static_cast(parsed) > static_cast(std::numeric_limits::max())) + { + throw std::invalid_argument(std::string(flag) + " must be an integer >= " + std::to_string(min) + ", got '" + + value + "'"); + } + return static_cast(parsed); +} + +void print_usage() +{ + const remote_device_bridge::BridgeOptions defaults; + std::cout + << "Usage: remote_device_bridge [--listen=] [--ws-listen=] [--app-name=]\n" + " [--wrist-rate-hz=] [--max-clients=] [--max-sample-age-ms=]\n" + " [--max-schema-bytes=] [--clock-source=two-way|one-way]\n" + " [--stats-csv=]\n" + << " --listen bind address (default " << defaults.listen.host << ":" << defaults.listen.port << ")\n" + << " --ws-listen also accept WebSocket clients (--remote=ws://...) here; no TLS (default off,\n" + << " port " << remote_plugin_session::kDefaultBridgeWsPort << " when omitted)\n" + << " --app-name OpenXR app name; defaults to the name the client sends in HELLO\n" + << " --wrist-rate-hz wrist sample rate towards the client (default " << defaults.wrist_rate_hz << ")\n" + << " --max-clients concurrent clients; further ones get HELLO_ACK 'bridge full' (default " + << defaults.max_clients << ")\n" + << " --max-sample-age-ms drop samples older than this instead of pushing them; 0 = off (default " + << defaults.max_sample_age->count() << ")\n" + << " --max-schema-bytes refuse schema channels whose sample size exceeds this (default " + << defaults.max_schema_bytes << ", the\n" + << " largest the runtime accepts)\n" + << " --clock-source two-way: PING/PONG min-RTT clock offset, one-way if the client never answers\n" + << " in time; one-way: frame-header minimum, no PINGs (default two-way)\n" + << " --stats-csv write one row per pushed hand sample to this file (truncated at start)\n"; +} + +remote_device_bridge::BridgeOptions parse_args(int argc, char** argv) +{ + remote_device_bridge::BridgeOptions options; + for (int i = 1; i < argc; ++i) + { + const std::string arg = argv[i]; + if (auto value = flag_value(arg, "--listen")) + { + options.listen = remote_plugin_session::parse_endpoint(*value, remote_plugin_session::kDefaultBridgePort); + } + else if (auto value = flag_value(arg, "--ws-listen")) + { + options.ws_listen = + remote_plugin_session::parse_endpoint(*value, remote_plugin_session::kDefaultBridgeWsPort); + } + else if (auto value = flag_value(arg, "--app-name")) + { + if (value->empty()) + { + throw std::invalid_argument("--app-name must not be empty"); + } + options.app_name = *value; + } + else if (auto value = flag_value(arg, "--wrist-rate-hz")) + { + options.wrist_rate_hz = parse_integer("--wrist-rate-hz", *value, 1); + } + else if (auto value = flag_value(arg, "--max-clients")) + { + options.max_clients = parse_integer("--max-clients", *value, 1); + } + else if (auto value = flag_value(arg, "--max-sample-age-ms")) + { + const int ms = parse_integer("--max-sample-age-ms", *value, 0); + options.max_sample_age = + ms == 0 ? std::nullopt : std::optional(std::chrono::milliseconds(ms)); + } + else if (auto value = flag_value(arg, "--max-schema-bytes")) + { + options.max_schema_bytes = parse_integer("--max-schema-bytes", *value, 1); + } + else if (auto value = flag_value(arg, "--clock-source")) + { + if (*value != "two-way" && *value != "one-way") + { + throw std::invalid_argument("--clock-source must be two-way or one-way"); + } + options.clock_source = *value == "two-way" ? remote_device_bridge::ClockSource::TwoWay : + remote_device_bridge::ClockSource::OneWay; + } + else if (auto value = flag_value(arg, "--stats-csv")) + { + options.stats_csv = *value; + } + else if (flag_value(arg, "--plugin-root-id")) + { + // Injected by the PluginManager; not used by the bridge. + } + else if (arg == "--help" || arg == "-h") + { + print_usage(); + std::exit(0); + } + else + { + throw std::invalid_argument("unknown argument '" + arg + "'"); + } + } + return options; +} + +} // namespace + +int main(int argc, char** argv) +try +{ + const remote_device_bridge::BridgeOptions options = parse_args(argc, argv); + + std::signal(SIGINT, signal_handler); + std::signal(SIGTERM, signal_handler); + + remote_device_bridge::BridgeServer server(options, plugin_utils::make_openxr_plugin_session_factory); + server.run(g_stop_requested); + return 0; +} +catch (const std::exception& e) +{ + auto logger = isaaccapture::Logger::get("isaaccapture.plugins.remote_device_bridge.main"); + logger->error("{}: {}", argv[0], e.what()); + return 1; +} +catch (...) +{ + auto logger = isaaccapture::Logger::get("isaaccapture.plugins.remote_device_bridge.main"); + logger->error("{}: Unknown error occurred", argv[0]); + return 1; +} diff --git a/src/plugins/remote_device_bridge/plugin.yaml b/src/plugins/remote_device_bridge/plugin.yaml new file mode 100644 index 0000000000..f794bfc528 --- /dev/null +++ b/src/plugins/remote_device_bridge/plugin.yaml @@ -0,0 +1,14 @@ +# SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +name: remote_device_bridge +description: "Mirrors a remote plugin session (hand tracking, schema push, wrist pull) onto the local OpenXR runtime" +command: "./remote_device_bridge" +version: "0.1.0" +devices: + - path: "/hand/left" + type: "hand_tracker" + description: "Left hand from a remote plugin" + - path: "/hand/right" + type: "hand_tracker" + description: "Right hand from a remote plugin" diff --git a/src/plugins/remote_device_bridge/server/CMakeLists.txt b/src/plugins/remote_device_bridge/server/CMakeLists.txt new file mode 100644 index 0000000000..443a5f2404 --- /dev/null +++ b/src/plugins/remote_device_bridge/server/CMakeLists.txt @@ -0,0 +1,26 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +find_package(Threads REQUIRED) + +add_library(remote_device_bridge_server STATIC + bridge_server.cpp + inc/remote_device_bridge/bridge_server.hpp + latency_stats.cpp + inc/remote_device_bridge/latency_stats.hpp +) + +target_include_directories(remote_device_bridge_server + INTERFACE + $ +) + +target_link_libraries(remote_device_bridge_server + PUBLIC + Teleop::remote_plugin_session + PRIVATE + log_bridge::log_bridge_core + Threads::Threads +) + +add_library(Teleop::remote_device_bridge_server ALIAS remote_device_bridge_server) diff --git a/src/plugins/remote_device_bridge/server/bridge_server.cpp b/src/plugins/remote_device_bridge/server/bridge_server.cpp new file mode 100644 index 0000000000..96818543e4 --- /dev/null +++ b/src/plugins/remote_device_bridge/server/bridge_server.cpp @@ -0,0 +1,1131 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "inc/remote_device_bridge/bridge_server.hpp" + +#include "inc/remote_device_bridge/latency_stats.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace remote_device_bridge +{ + +namespace +{ + +using remote_plugin_session::ClockOffsetEstimator; +using remote_plugin_session::ITransport; +using remote_plugin_session::PingScheduler; +using remote_plugin_session::ReceiveResult; +namespace protocol = remote_plugin_protocol; + +std::shared_ptr logger() +{ + return isaaccapture::Logger::get("isaaccapture.plugins.remote_device_bridge.BridgeServer"); +} + +constexpr std::chrono::milliseconds kAcceptPoll{ 200 }; +constexpr std::chrono::milliseconds kHandshakeTimeout{ 3000 }; +constexpr std::chrono::milliseconds kReceivePoll{ 100 }; +constexpr std::chrono::nanoseconds kHeartbeatPeriod{ 500'000'000 }; +constexpr std::chrono::nanoseconds kPeerTimeout{ 3'000'000'000 }; +//! After a failed send, how long the reader gets to see the hang-up and apply the client's last frames before the +//! session ends anyway (a stalled link). +constexpr std::chrono::milliseconds kSendFailureGrace{ 500 }; +// A refused client gets "bridge full" as soon as its transport is up (ws://: after the upgrade, waited for this +// long), then the connection lingers this long so the ACK is delivered and a late HELLO is read, not reset. +constexpr std::chrono::milliseconds kLingerWait{ 1000 }; +// Refusal children alive at once; past this a refused client gets a best-effort ACK and an immediate close. +constexpr size_t kMaxRefusing = 16; +constexpr std::chrono::milliseconds kReapPoll{ 100 }; +constexpr std::chrono::seconds kStopGrace{ 3 }; + +bool is_sample(protocol::FrameType type) +{ + return type == protocol::FrameType::HandSample || type == protocol::FrameType::SchemaSample; +} + +struct Received +{ + protocol::Frame frame; + int64_t t_recv_ns; +}; + +//! HandOwners::owner index of a hand. +size_t hand_index(XrHandEXT hand) +{ + return hand == XR_HAND_LEFT_EXT ? 0 : 1; +} + +const char* hand_side(size_t hand_index) +{ + return hand_index == 0 ? "left" : "right"; +} + +} // namespace + +//! Bridge client id owning each hand's push device (0 = free). Lives in a MAP_SHARED page created before any +//! fork, so every child sees the others' claims; the parent frees a crashed child's claims when it reaps it. +struct detail::HandOwners +{ + std::array, 2> owner{}; + + //! Empty when `client_id` now owns the hand, else the current owner. + std::optional claim(size_t hand, uint32_t client_id) + { + uint32_t expected = 0; + if (owner[hand].compare_exchange_strong(expected, client_id)) + { + return std::nullopt; + } + return expected; + } + + void release(size_t hand, uint32_t client_id) + { + uint32_t expected = client_id; + owner[hand].compare_exchange_strong(expected, 0); + } +}; +static_assert(std::atomic::is_always_lock_free, "hand owners are shared across processes"); + +void detail::HandOwnersUnmap::operator()(HandOwners* owners) const noexcept +{ + munmap(owners, sizeof(HandOwners)); +} + +namespace +{ + +//! Refuses one OPEN_*: answered with ERROR on that channel id, the session and its other channels continue. +class ChannelRefused : public std::runtime_error +{ +public: + using std::runtime_error::runtime_error; +}; + +//! The handshake did not produce a session; the reason is already logged. +class HandshakeRefused : public std::runtime_error +{ +public: + using std::runtime_error::runtime_error; +}; + +//! An owner outside this bridge (a local plugin) makes xrCreatePushDeviceNV fail with XR_ERROR_RUNTIME_FAILURE (-2, +//! seen 2026-09-29; XR_NVX1_device_interface has no dedicated code). plugin_utils throws it as a std::system_error +//! in the "XrResult" category. Any other result stays a session error. +bool push_device_owned_elsewhere(const std::exception& e) +{ + const auto* error = dynamic_cast(&e); + return error != nullptr && std::strcmp(error->code().category().name(), "XrResult") == 0 && + error->code().value() == XR_ERROR_RUNTIME_FAILURE && + std::string(error->what()).rfind("xrCreatePushDeviceNV(", 0) == 0; +} + +//! One accepted connection: owns the mirrored plugin session and its channels. +class ClientSession +{ +public: + //! Reads HELLO, creates the session and answers HELLO_ACK; throws HandshakeRefused when the client is refused. + ClientSession(std::unique_ptr transport, + uint32_t client_id, + const BridgeOptions& options, + const SessionFactory& session_factory, + detail::HandOwners& hand_owners, + const SampleCsv* csv) + : m_transport(std::move(transport)), + m_client_id(client_id), + m_tag("[bridge c" + std::to_string(client_id) + "]"), + m_options(options), + m_hand_owners(hand_owners), + m_csv(csv) + { + // The reader starts last, so a throw here leaves only members whose destructors close the transport. + handshake(session_factory); + } + + ~ClientSession() + { + m_done = true; + m_transport->shutdown(); + m_reader.join(); + + // Push channels close before the session, which owns the pull channel and wrist source (m_session is + // declared before the channel maps, so it is destroyed last). + while (!m_hand_channels.empty()) + { + close_channel(m_hand_channels.begin()->first); + } + while (!m_schema_channels.empty()) + { + close_channel(m_schema_channels.begin()->first); + } + } + + void run(const std::atomic& stop) + { + const auto period = std::chrono::nanoseconds(1'000'000'000 / m_options.wrist_rate_hz); + auto next_tick = std::chrono::steady_clock::now() + period; + auto next_stats = std::chrono::steady_clock::now() + m_options.stats_period; + std::optional send_failed_deadline; + + while (!stop.load() && !m_done.load()) + { + { + std::unique_lock lock(m_queue_mutex); + m_queue_cv.wait_until(lock, next_tick, [this] { return !m_queue.empty() || m_reader_closed.load(); }); + } + if (!apply_queued()) + { + return; + } + + if (m_reader_closed.load()) + { + // The reader never pushes after raising the flag, so one more pass picks up the client's last + // frames (typically CLOSE for the pull channel) that landed between the swap above and the check. + if (!apply_queued()) + { + return; + } + if (m_reader_error.empty()) + { + logger()->info("{} client disconnected", m_tag); + } + else + { + fail(m_reader_error); + } + return; + } + + const int64_t now_ns = core::os_monotonic_now_ns(); + if (now_ns - m_last_recv_ns > kPeerTimeout.count()) + { + logger()->warn("{} no frames from the client within the peer timeout; closing", m_tag); + return; + } + + const auto now = std::chrono::steady_clock::now(); + if (m_send_failed) + { + if (!send_failed_deadline) + { + send_failed_deadline = now + kSendFailureGrace; + } + else if (now >= *send_failed_deadline) + { + logger()->warn("{} sending to the client failed; closing", m_tag); + return; + } + } + if (now >= next_tick) + { + tick(now_ns); + next_tick += period; + if (next_tick < now) + { + next_tick = now + period; + } + } + poll_pings(now_ns); + if (now_ns - m_last_send_ns > kHeartbeatPeriod.count()) + { + send(protocol::FrameType::Heartbeat, protocol::kSessionChannelId, {}); + } + if (now >= next_stats) + { + log_stats(); + next_stats += m_options.stats_period; + } + } + } + +private: + struct HandChannel + { + std::unique_ptr channel; + size_t hand = 0; //!< HandOwners index + uint64_t pushes = 0; + }; + + struct SchemaChannel + { + std::unique_ptr channel; + std::string collection_id; + uint64_t pushes = 0; + }; + + struct Counters + { + uint64_t rx_hand = 0; + uint64_t rx_schema = 0; + uint64_t applied = 0; + uint64_t coalesced = 0; + uint64_t stale = 0; + uint64_t tx_wrist = 0; + }; + + //! The client gets the logged reason as a HELLO_ACK refusal before the connection closes. + [[noreturn]] void refuse_handshake(const std::string& reason) + { + logger()->warn("{} {}; dropping connection", m_tag, reason); + send(protocol::FrameType::HelloAck, protocol::kSessionChannelId, + protocol::encode_hello_ack(protocol::HelloAck{ false, reason })); + throw HandshakeRefused(reason); + } + + void handshake(const SessionFactory& session_factory) + { + std::vector bytes; + if (m_transport->receive(bytes, kHandshakeTimeout) != ReceiveResult::Ok) + { + refuse_handshake("bad HELLO (no frame)"); + } + protocol::Frame frame; + protocol::Hello hello; + try + { + frame = protocol::split_frame(std::move(bytes)); + if (frame.header.type != protocol::FrameType::Hello) + { + refuse_handshake("bad HELLO (expected HELLO, got frame type " + + std::to_string(static_cast(frame.header.type)) + ")"); + } + hello = protocol::decode_hello(frame.payload); + } + catch (const protocol::DecodeError& e) + { + refuse_handshake(std::string("bad HELLO (") + e.what() + ")"); + } + note_received(frame.header, core::os_monotonic_now_ns()); + + const std::string app_name = + m_options.app_name.value_or(!hello.app_name.empty() ? hello.app_name : std::string("remote_device_bridge")); + logger()->info( + "{} HELLO from '{}' (schema_push={:d} hand_tracking_push={:d} wrist_tracking_pull={:d}); app name '{}'", + m_tag, hello.app_name, hello.requirements.schema_push, hello.requirements.hand_tracking_push, + hello.requirements.wrist_tracking_pull.has_value(), app_name); + + logger()->info("{} creating plugin session", m_tag); + try + { + m_session = session_factory(app_name)(hello.requirements, {}); + } + catch (const std::exception& e) + { + logger()->error("{} plugin session creation failed: {}", m_tag, e.what()); + send(protocol::FrameType::HelloAck, protocol::kSessionChannelId, + protocol::encode_hello_ack( + protocol::HelloAck{ false, std::string("plugin session creation failed: ") + e.what() })); + throw HandshakeRefused(e.what()); + } + logger()->info("{} plugin session ready", m_tag); + m_wrist_declared = hello.requirements.wrist_tracking_pull.has_value(); + + send(protocol::FrameType::HelloAck, protocol::kSessionChannelId, + protocol::encode_hello_ack(protocol::HelloAck{ true, "" })); + if (m_send_failed) + { + refuse_handshake("HELLO_ACK not delivered"); + } + if (m_options.clock_source == ClockSource::TwoWay) + { + m_pings.emplace(core::os_monotonic_now_ns()); + } + // Last: nothing may throw once the reader runs. + m_reader = std::thread(&ClientSession::reader_loop, this); + } + + void reader_loop() + { + while (!m_done.load()) + { + std::vector bytes; + const ReceiveResult result = m_transport->receive(bytes, kReceivePoll); + if (result == ReceiveResult::Timeout) + { + continue; + } + if (result == ReceiveResult::Malformed) + { + m_reader_error = "malformed message from the transport"; + break; + } + if (result != ReceiveResult::Ok) + { + break; + } + Received received{ {}, core::os_monotonic_now_ns() }; + try + { + received.frame = protocol::split_frame(std::move(bytes)); + } + catch (const protocol::DecodeError& e) + { + m_reader_error = std::string("bad frame: ") + e.what(); + break; + } + { + std::lock_guard lock(m_queue_mutex); + m_queue.push_back(std::move(received)); + } + m_queue_cv.notify_one(); + } + m_reader_closed.store(true); + m_queue_cv.notify_all(); + } + + //! Swaps out the queue and applies it: control frames in order, samples coalesced to the newest per channel + //! (a slow OpenXR push must not turn into a growing backlog). False once the session has failed. + bool apply_queued() + { + std::deque batch; + { + std::lock_guard lock(m_queue_mutex); + batch.swap(m_queue); + } + std::map newest; + for (size_t i = 0; i < batch.size(); ++i) + { + if (is_sample(batch[i].frame.header.type)) + { + newest[batch[i].frame.header.channel_id] = i; + } + } + for (size_t i = 0; i < batch.size(); ++i) + { + const protocol::FrameHeader& header = batch[i].frame.header; + if (is_sample(header.type)) + { + ++(header.type == protocol::FrameType::HandSample ? m_counts.rx_hand : m_counts.rx_schema); + if (newest[header.channel_id] != i) + { + ++m_counts.coalesced; + note_received(header, batch[i].t_recv_ns); + continue; + } + } + apply(batch[i]); + if (m_done.load()) + { + return false; + } + } + return true; + } + + void note_received(const protocol::FrameHeader& header, int64_t t_recv_ns) + { + m_clock.add_sample(header.t_send_monotonic_ns, t_recv_ns); + m_last_recv_ns = t_recv_ns; + } + + //! Set while OPEN_PULL is open and HELLO declared a wrist source; owned by m_session. + core::IWristTrackingSource* wrist_source() const + { + return m_pull && m_wrist_declared ? &m_pull->channel->wrist_tracking_source() : nullptr; + } + + void apply(const Received& received) + { + const protocol::FrameHeader& header = received.frame.header; + const std::vector& payload = received.frame.payload; + note_received(header, received.t_recv_ns); + + try + { + switch (header.type) + { + case protocol::FrameType::OpenHand: + { + const XrHandEXT open_hand = protocol::decode_open_hand(payload); + require_free_channel(header.channel_id); + const size_t hand = hand_index(open_hand); + const char* const side = hand_side(hand); + // The runtime allows one push device per hand; refuse a second owner before it gets there. + if (const std::optional owner = m_hand_owners.claim(hand, m_client_id)) + { + throw ChannelRefused(std::string("hand ") + side + " is already owned by bridge client c" + + std::to_string(*owner) + "; channel refused"); + } + try + { + auto channel = m_session->create_hand_tracking_push_channel(open_hand); + m_hand_channels.emplace(header.channel_id, HandChannel{ std::move(channel), hand }); + } + catch (const std::exception& e) + { + m_hand_owners.release(hand, m_client_id); + if (!push_device_owned_elsewhere(e)) + { + throw; + } + logger()->warn("{} {}", m_tag, e.what()); + throw ChannelRefused(std::string("hand ") + side + + " is already owned by another OpenXR session (runtime refused the push " + "device); channel refused"); + } + logger()->info("{} channel {}: hand {} opened", m_tag, header.channel_id, side); + break; + } + case protocol::FrameType::OpenSchema: + { + const core::SchemaPusherConfig config = protocol::decode_open_schema(payload); + require_free_channel(header.channel_id); + // The runtime tensor is max_flatbuffer_size bytes whatever the sample holds. + if (const std::optional error = remote_plugin_session::schema_cap_error( + config.collection_id, config.max_flatbuffer_size, m_options.max_schema_bytes)) + { + throw ChannelRefused(*error); + } + auto channel = m_session->create_schema_push_channel(config); + m_schema_channels.emplace(header.channel_id, SchemaChannel{ std::move(channel), config.collection_id }); + logger()->info("{} channel {}: schema '{}' opened", m_tag, header.channel_id, config.collection_id); + break; + } + case protocol::FrameType::OpenPull: + { + if (m_pull) + { + throw std::runtime_error("only one pull channel per session"); + } + require_free_channel(header.channel_id); + // The session built the pull channel, and the wrist source HELLO declared, when it was created. + m_pull = PullChannel{ &m_session->pull_channel(), header.channel_id }; + logger()->info("{} channel {}: pull channel opened{}", m_tag, header.channel_id, + wrist_source() != nullptr ? " with wrist source" : ""); + break; + } + case protocol::FrameType::Close: + close_channel(header.channel_id); + break; + case protocol::FrameType::HandSample: + { + const auto it = m_hand_channels.find(header.channel_id); + if (it == m_hand_channels.end()) + { + if (m_refused_channels.count(header.channel_id) != 0) + { + break; // sent before the client saw the refusal + } + throw std::runtime_error("HAND_SAMPLE on unknown channel " + std::to_string(header.channel_id)); + } + const protocol::HandSample sample = protocol::decode_hand_sample(payload); + const int64_t t_ns = m_clock.remote_to_local(sample.sample_time_ns); + const int64_t now_ns = core::os_monotonic_now_ns(); + if (stale(t_ns, now_ns)) + { + break; + } + it->second.channel->push(sample.joints.data(), t_ns); + ++it->second.pushes; + ++m_counts.applied; + if (m_csv && + !m_csv->write({ core::os_monotonic_now_ns(), received.t_recv_ns, sample.sample_time_ns, t_ns, + now_ns - t_ns, hand_side(it->second.hand), header.seq, + index_curl_m(sample.joints.data()) }) && + !m_csv_failed) + { + m_csv_failed = true; + logger()->warn( + "{} stats CSV row not written ({}); later rows may be missing", m_tag, std::strerror(errno)); + } + break; + } + case protocol::FrameType::SchemaSample: + { + const auto it = m_schema_channels.find(header.channel_id); + if (it == m_schema_channels.end()) + { + if (m_refused_channels.count(header.channel_id) != 0) + { + break; + } + throw std::runtime_error("SCHEMA_SAMPLE on unknown channel " + std::to_string(header.channel_id)); + } + const protocol::SchemaSample sample = protocol::decode_schema_sample(payload); + const int64_t t_ns = m_clock.remote_to_local(sample.sample_time_local_common_clock_ns); + if (stale(t_ns, core::os_monotonic_now_ns())) + { + break; + } + it->second.channel->push_buffer( + sample.bytes.data(), sample.bytes.size(), t_ns, sample.sample_time_raw_device_clock_ns); + ++it->second.pushes; + ++m_counts.applied; + break; + } + case protocol::FrameType::Pong: + { + const protocol::Pong pong = protocol::decode_pong(payload); + if (!m_pings || !m_pings->active()) + { + break; + } + m_pings->on_pong(); + m_clock.add_two_way_sample( + pong.ping_sent_ns, pong.ping_received_ns, pong.pong_sent_ns, received.t_recv_ns); + if (!std::exchange(m_logged_pong, true)) + { + logger()->info( + "{} clock: first PONG, rtt_ms={:g}; using two-way offset", m_tag, m_clock.rtt_last_ns() / 1e6); + } + break; + } + case protocol::FrameType::Heartbeat: + break; + case protocol::FrameType::Hello: + throw std::runtime_error("HELLO after the handshake"); + default: + throw std::runtime_error("unexpected frame type " + std::to_string(static_cast(header.type))); + } + } + catch (const ChannelRefused& e) + { + refuse_channel(header.channel_id, e.what()); + } + catch (const std::exception& e) + { + fail(e.what()); + } + } + + //! True (and counted) when the mapped sample time is older than max_sample_age. Records the age either way. + bool stale(int64_t mapped_t_ns, int64_t now_ns) + { + m_ages_ns.push_back(now_ns - mapped_t_ns); + if (!m_options.max_sample_age || + !remote_plugin_session::should_drop_stale( + mapped_t_ns, now_ns, std::chrono::nanoseconds(*m_options.max_sample_age).count())) + { + return false; + } + ++m_counts.stale; + return true; + } + + void require_free_channel(uint16_t channel_id) const + { + if (channel_id == protocol::kSessionChannelId || m_hand_channels.count(channel_id) != 0 || + m_schema_channels.count(channel_id) != 0 || m_refused_channels.count(channel_id) != 0 || + (m_pull && channel_id == m_pull->id)) + { + throw std::runtime_error("channel id " + std::to_string(channel_id) + " is already in use"); + } + } + + //! Logs each push channel's push count as it closes. + void close_channel(uint16_t channel_id) + { + if (const auto hand = m_hand_channels.find(channel_id); hand != m_hand_channels.end()) + { + const uint64_t pushes = hand->second.pushes; + const size_t hand_index = hand->second.hand; + m_hand_channels.erase(hand); + m_hand_owners.release(hand_index, m_client_id); + logger()->info("{} hand channel {} closed after {} pushes", m_tag, channel_id, pushes); + return; + } + if (const auto schema = m_schema_channels.find(channel_id); schema != m_schema_channels.end()) + { + const uint64_t pushes = schema->second.pushes; + const std::string collection_id = schema->second.collection_id; + m_schema_channels.erase(schema); + logger()->info("{} schema channel {} ({}) closed after {} pushes", m_tag, channel_id, collection_id, pushes); + return; + } + if (m_pull && channel_id == m_pull->id) + { + m_pull.reset(); + logger()->info("{} channel {} (pull) closed", m_tag, channel_id); + return; + } + if (m_refused_channels.erase(channel_id) != 0) + { + return; + } + logger()->warn("{} CLOSE for unknown channel {}", m_tag, channel_id); + } + + void poll_pings(int64_t now_ns) + { + if (!m_pings) + { + return; + } + switch (m_pings->poll(now_ns)) + { + case PingScheduler::Action::Ping: + send(protocol::FrameType::Ping, protocol::kSessionChannelId, + protocol::encode_ping(core::os_monotonic_now_ns())); + break; + case PingScheduler::Action::GiveUp: + logger()->warn("{} clock: no PONG, falling back to one-way", m_tag); + break; + case PingScheduler::Action::None: + break; + } + } + + void tick(int64_t now_ns) + { + if (!m_pull) + { + return; + } + try + { + m_pull->channel->update(); + if (core::IWristTrackingSource* const wrist = wrist_source()) + { + for (const XrHandEXT hand : { XR_HAND_LEFT_EXT, XR_HAND_RIGHT_EXT }) + { + protocol::WristSample sample; + sample.sample_time_ns = now_ns; + sample.hand = hand; + sample.sample = wrist->query(hand, now_ns); + send(protocol::FrameType::WristSample, m_pull->id, protocol::encode_wrist_sample(sample)); + if (m_send_failed) + { + return; + } + ++m_counts.tx_wrist; + } + } + } + catch (const std::exception& e) + { + fail(e.what()); + } + } + + //! After a failed send the rest are dropped; run() ends the session once the reader has seen the hang-up (so the + //! client's last frames still apply) or kSendFailureGrace has passed. + void send(protocol::FrameType type, uint16_t channel_id, const std::vector& payload) + { + if (m_send_failed) + { + return; + } + std::vector frame = protocol::encode_frame(type, channel_id, payload); + m_last_send_ns = core::os_monotonic_now_ns(); + protocol::stamp_header(frame, m_next_seq++, m_last_send_ns); + if (!m_transport->send(frame.data(), frame.size())) + { + m_send_failed = true; + } + } + + void refuse_channel(uint16_t channel_id, const std::string& message) + { + logger()->warn("{} channel {} refused: {}", m_tag, channel_id, message); + m_refused_channels.insert(channel_id); + send(protocol::FrameType::Error, channel_id, protocol::encode_error(message)); + } + + void fail(const std::string& message) + { + logger()->error("{} error: {}; closing connection", m_tag, message); + send(protocol::FrameType::Error, protocol::kSessionChannelId, protocol::encode_error(message)); + m_done.store(true); + } + + //! Deltas since the previous line; all counters live on the run thread. + void log_stats() + { + logger()->info("{} 5s: rx hand={} schema={} | applied={} coalesced={} stale={} | tx wrist={} | offset_ns={}{}", + m_tag, m_counts.rx_hand - m_logged.rx_hand, m_counts.rx_schema - m_logged.rx_schema, + m_counts.applied - m_logged.applied, m_counts.coalesced - m_logged.coalesced, + m_counts.stale - m_logged.stale, m_counts.tx_wrist - m_logged.tx_wrist, m_clock.offset_ns(), + format_clock_stats(std::move(m_ages_ns), m_clock)); + m_logged = m_counts; + m_ages_ns.clear(); + } + + std::unique_ptr m_transport; + const uint32_t m_client_id; + const std::string m_tag; + const BridgeOptions& m_options; + detail::HandOwners& m_hand_owners; + + std::unique_ptr m_session; + std::map m_hand_channels; + std::map m_schema_channels; + struct PullChannel + { + core::IPluginPullChannel* channel; //!< owned by m_session + uint16_t id; + }; + std::optional m_pull; //!< set while OPEN_PULL is open + bool m_wrist_declared = false; + std::set m_refused_channels; //!< until the client's CLOSE + + ClockOffsetEstimator m_clock; + std::optional m_pings; //!< set for --clock-source=two-way + bool m_logged_pong = false; + std::vector m_ages_ns; //!< since the last status line + bool m_csv_failed = false; //!< warned once + const SampleCsv* m_csv = nullptr; //!< the parent's --stats-csv file, inherited across fork(); null without it + uint32_t m_next_seq = 0; + int64_t m_last_send_ns = 0; + int64_t m_last_recv_ns = 0; + Counters m_counts; + Counters m_logged; //!< m_counts at the previous status line + + std::thread m_reader; + std::string m_reader_error; //!< why the reader stopped, if not a clean close; written before m_reader_closed + std::mutex m_queue_mutex; + std::condition_variable m_queue_cv; + std::deque m_queue; + std::atomic m_reader_closed{ false }; + std::atomic m_done{ false }; + bool m_send_failed = false; //!< run thread only (the handshake runs on it too) +}; + +//! Runs one client to completion in the forked child; false after a fatal error. +bool serve_client(std::unique_ptr transport, + uint32_t client_id, + const BridgeOptions& options, + const SessionFactory& session_factory, + detail::HandOwners& hand_owners, + const SampleCsv* csv, + const std::atomic& stop) +{ + bool ok = true; + try + { + ClientSession session(std::move(transport), client_id, options, session_factory, hand_owners, csv); + session.run(stop); + } + catch (const HandshakeRefused&) + { + } + catch (const std::exception& e) + { + logger()->error("[bridge c{}] fatal: {}", client_id, e.what()); + ok = false; + } + logger()->info("[bridge c{}] session closed", client_id); + return ok; +} + +std::vector bridge_full_ack() +{ + return protocol::encode_frame(protocol::FrameType::HelloAck, protocol::kSessionChannelId, + protocol::encode_hello_ack(protocol::HelloAck{ false, "bridge full" })); +} + +//! Runs in a refusal child: HELLO_ACK first, then read until the client hangs up (it does once it reads the +//! ACK) or kLingerWait passes, so the close is a FIN even when the HELLO arrives late. +void send_bridge_full(ITransport& client, std::vector ack) +{ + if (client.wait_ready(kLingerWait) != ReceiveResult::Ok) + { + return; + } + protocol::stamp_header(ack, 0, core::os_monotonic_now_ns()); + if (!client.send(ack.data(), ack.size())) + { + return; + } + const auto deadline = std::chrono::steady_clock::now() + kLingerWait; + for (std::vector bytes;;) + { + const auto left = + std::chrono::duration_cast(deadline - std::chrono::steady_clock::now()); + if (left <= std::chrono::milliseconds{ 0 }) + { + return; + } + const ReceiveResult result = client.receive(bytes, left); + if (result == ReceiveResult::Closed || result == ReceiveResult::Malformed) + { + return; + } + } +} + +double uptime_s(std::chrono::steady_clock::time_point started) +{ + return std::chrono::duration(std::chrono::steady_clock::now() - started).count(); +} + +} // namespace + +struct BridgeServer::ClientRecord +{ + uint32_t id = 0; + std::string peer; + std::chrono::steady_clock::time_point started; + pid_t pid = 0; + bool refusal = false; //!< only sends "bridge full"; not counted against --max-clients +}; + +BridgeServer::BridgeServer(BridgeOptions options, SessionFactory session_factory) + : m_options(std::move(options)), m_session_factory(std::move(session_factory)) +{ + if (!m_session_factory) + { + throw std::invalid_argument("BridgeServer requires a session factory"); + } + if (m_options.wrist_rate_hz <= 0 || m_options.max_clients <= 0 || + (m_options.max_sample_age && m_options.max_sample_age->count() <= 0)) + { + throw std::invalid_argument("BridgeOptions: wrist_rate_hz, max_clients and max_sample_age must be positive"); + } + if (m_options.max_schema_bytes == 0) + { + throw std::invalid_argument("BridgeOptions: max_schema_bytes must be positive"); + } + m_listener = remote_plugin_session::TcpSocket::listen(m_options.listen, m_options.max_clients); + if (m_options.ws_listen) + { + m_ws_listener = remote_plugin_session::TcpSocket::listen(*m_options.ws_listen, m_options.max_clients); + logger()->info( + "remote_device_bridge listening for ws:// on {}:{}", m_options.ws_listen->host, m_ws_listener.local_port()); + } + if (m_options.stats_csv) + { + m_csv = std::make_unique(*m_options.stats_csv); + } + logger()->info( + "remote_device_bridge listening on {}:{} (wrist {} Hz, process per client, max {}, max sample age {}, max " + "schema {} bytes)", + m_options.listen.host, m_listener.local_port(), m_options.wrist_rate_hz, m_options.max_clients, + m_options.max_sample_age ? std::to_string(m_options.max_sample_age->count()) + " ms" : std::string("off"), + m_options.max_schema_bytes); + logger()->info("remote_device_bridge clock={}{}", + m_options.clock_source == ClockSource::TwoWay ? "two-way" : "one-way", + m_options.stats_csv ? " stats-csv=" + *m_options.stats_csv : std::string()); + void* const shared = + mmap(nullptr, sizeof(detail::HandOwners), PROT_READ | PROT_WRITE, MAP_SHARED | MAP_ANONYMOUS, -1, 0); + if (shared == MAP_FAILED) + { + throw std::runtime_error(std::string("mmap(hand owners): ") + std::strerror(errno)); + } + m_hand_owners.reset(new (shared) detail::HandOwners()); +} + +BridgeServer::~BridgeServer() = default; + +uint16_t BridgeServer::port() const +{ + return m_listener.local_port(); +} + +std::optional BridgeServer::ws_port() const +{ + if (!m_ws_listener.valid()) + { + return std::nullopt; + } + return m_ws_listener.local_port(); +} + +void BridgeServer::release_hands(const ClientRecord& client) +{ + for (size_t hand = 0; hand < m_hand_owners->owner.size(); ++hand) + { + m_hand_owners->release(hand, client.id); + } +} + +void BridgeServer::on_reaped(const ClientRecord& client, int status) +{ + if (client.refusal) + { + return; + } + release_hands(client); + logger()->info("[bridge] client pid {} exited ({}) after {:g} s", client.pid, + WIFEXITED(status) ? "status " + std::to_string(WEXITSTATUS(status)) : + "signal " + std::to_string(WTERMSIG(status)), + uptime_s(client.started)); +} + +void BridgeServer::reap_finished() +{ + for (auto it = m_clients.begin(); it != m_clients.end();) + { + int status = 0; + const pid_t reaped = waitpid(it->pid, &status, WNOHANG); + if (reaped == 0) + { + ++it; + continue; + } + if (reaped < 0) + { + // The child is gone without a status (ECHILD); drop it so it stops counting against max_clients. + logger()->warn("[bridge] waitpid(client pid {}) failed: {}", it->pid, std::strerror(errno)); + release_hands(*it); + } + else + { + on_reaped(*it, status); + } + it = m_clients.erase(it); + } +} + +void BridgeServer::stop_and_reap_all() +{ + for (const ClientRecord& client : m_clients) + { + kill(client.pid, SIGTERM); + } + // A child still inside the session factory (the OpenXR one waits for a headset) never sees the stop flag, so + // the grace period ends in SIGKILL. + const auto grace_end = std::chrono::steady_clock::now() + kStopGrace; + for (const ClientRecord& client : m_clients) + { + int status = 0; + pid_t reaped = waitpid(client.pid, &status, WNOHANG); + while (reaped == 0) + { + if (std::chrono::steady_clock::now() >= grace_end) + { + logger()->warn("[bridge] client pid {} killed after {} s grace", client.pid, kStopGrace.count()); + kill(client.pid, SIGKILL); + reaped = waitpid(client.pid, &status, 0); + break; + } + std::this_thread::sleep_for(kReapPoll); + reaped = waitpid(client.pid, &status, WNOHANG); + } + if (reaped == client.pid) + { + on_reaped(client, status); + } + } + m_clients.clear(); +} + +void BridgeServer::admit_client(std::unique_ptr client, const std::string& peer, const std::atomic& stop) +{ + const uint32_t id = ++m_accepted; + const auto refusing = static_cast( + std::count_if(m_clients.begin(), m_clients.end(), [](const ClientRecord& c) { return c.refusal; })); + const size_t active = m_clients.size() - refusing; + if (active >= static_cast(m_options.max_clients)) + { + logger()->warn("[bridge] client {} refused: bridge full ({})", peer, active); + if (refusing >= kMaxRefusing) + { + std::vector ack = bridge_full_ack(); + protocol::stamp_header(ack, 0, core::os_monotonic_now_ns()); + client->send(ack.data(), ack.size()); + return; + } + // A child answers so the accept loop never waits on a slow client (the ws:// upgrade, the linger). + const pid_t pid = fork(); + if (pid == 0) + { + m_listener.close(); + m_ws_listener.close(); + send_bridge_full(*client, bridge_full_ack()); + _exit(0); + } + if (pid < 0) + { + logger()->error("[bridge] fork failed ({}); closing refused client {}", std::strerror(errno), peer); + return; + } + m_clients.push_back(ClientRecord{ id, peer, std::chrono::steady_clock::now(), pid, true }); + return; + } + + const pid_t pid = fork(); + if (pid < 0) + { + logger()->error("[bridge] fork failed ({}); dropping client {}", std::strerror(errno), peer); + return; + } + if (pid == 0) + { + m_listener.close(); + m_ws_listener.close(); + const bool ok = + serve_client(std::move(client), id, m_options, m_session_factory, *m_hand_owners, m_csv.get(), stop); + // _exit skips stdio teardown: flush the logger's file sink and stdout first. + logger()->flush(); + std::fflush(nullptr); + _exit(ok ? 0 : 1); + } + logger()->info("[bridge] client {} started (pid {}, c{})", peer, pid, id); + m_clients.push_back(ClientRecord{ id, peer, std::chrono::steady_clock::now(), pid }); +} + +void BridgeServer::run(const std::atomic& stop) +{ + while (!stop.load()) + { + reap_finished(); + std::array listeners{ pollfd{ m_listener.fd(), POLLIN, 0 }, pollfd{ m_ws_listener.fd(), POLLIN, 0 } }; + const nfds_t count = m_ws_listener.valid() ? 2 : 1; + const int ready = ::poll(listeners.data(), count, static_cast(kAcceptPoll.count())); + if (ready < 0 && errno != EINTR) + { + throw std::runtime_error(std::string("poll(accept): ") + std::strerror(errno)); + } + if (ready <= 0) + { + continue; + } + for (nfds_t i = 0; i < count; ++i) + { + if ((listeners[i].revents & POLLIN) == 0) + { + continue; + } + std::string peer; + std::unique_ptr client = + i == 0 ? remote_plugin_session::TcpTransport::accept(m_listener, std::chrono::milliseconds{ 0 }, &peer) : + remote_plugin_session::WebSocketTransport::accept( + m_ws_listener, std::chrono::milliseconds{ 0 }, &peer); + if (client) + { + admit_client(std::move(client), i == 0 ? peer : "ws://" + peer, stop); + } + } + } + stop_and_reap_all(); +} + +} // namespace remote_device_bridge diff --git a/src/plugins/remote_device_bridge/server/inc/remote_device_bridge/bridge_server.hpp b/src/plugins/remote_device_bridge/server/inc/remote_device_bridge/bridge_server.hpp new file mode 100644 index 0000000000..c844eb8c73 --- /dev/null +++ b/src/plugins/remote_device_bridge/server/inc/remote_device_bridge/bridge_server.hpp @@ -0,0 +1,108 @@ +// SPDX-FileCopyrightText: Copyright (c) 2025-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace remote_device_bridge +{ + +class SampleCsv; + +namespace detail +{ +//! Hand push-device owners shared with the forked children (defined in the .cpp). +struct HandOwners; +struct HandOwnersUnmap +{ + void operator()(HandOwners* owners) const noexcept; +}; +} // namespace detail + +enum class ClockSource +{ + TwoWay, //!< PING/PONG min-RTT offset; one-way only when no PONG answers + OneWay, //!< frame-header minimum only, no PINGs (A/B baseline) +}; + +struct BridgeOptions +{ + remote_plugin_session::Endpoint listen{ "0.0.0.0", remote_plugin_session::kDefaultBridgePort }; + //! Second listener speaking WebSocket (same frames, same --max-clients budget). + std::optional ws_listen; + //! Overrides the HELLO app name. + std::optional app_name; + int wrist_rate_hz = 90; + int max_clients = 8; + //! Samples whose mapped time is older than this are dropped instead of pushed. + std::optional max_sample_age{ 250 }; + //! OPEN_SCHEMA with a larger max_flatbuffer_size is refused before it reaches the runtime. + size_t max_schema_bytes = remote_plugin_session::kDefaultMaxSchemaBytes; + ClockSource clock_source = ClockSource::TwoWay; + //! One CSV row per pushed hand sample (SampleCsv). + std::optional stats_csv; + //! Period of each client's status line; not a flag (tests shorten it). + std::chrono::milliseconds stats_period{ 5000 }; +}; + +//! Returns the factory for a client's session, given the app name; called in the forked child once HELLO arrived. +using SessionFactory = std::function; + +/*! + * @brief Accepts remote plugin sessions and mirrors each onto a session from the SessionFactory. + * + * Every client is served by its own forked child process (see this plugin's AGENTS.md for why). + */ +class BridgeServer +{ +public: + BridgeServer(BridgeOptions options, SessionFactory session_factory); + ~BridgeServer(); + + //! Bound ports (useful with port 0). + uint16_t port() const; + std::optional ws_port() const; + + //! Accept loop; returns once stop becomes true and every client has been reaped. + void run(const std::atomic& stop); + +private: + struct ClientRecord; + + //! Logs and forgets clients that have exited, without blocking. + void reap_finished(); + //! Asks every child to stop and blocks until all are reaped (SIGKILL after a grace period). + void stop_and_reap_all(); + void on_reaped(const ClientRecord& client, int status); + //! Frees every hand `client` claimed. + void release_hands(const ClientRecord& client); + void admit_client(std::unique_ptr client, + const std::string& peer, + const std::atomic& stop); + + BridgeOptions m_options; + SessionFactory m_session_factory; + remote_plugin_session::TcpSocket m_listener; + remote_plugin_session::TcpSocket m_ws_listener; + std::unique_ptr m_hand_owners; + std::unique_ptr m_csv; //!< --stats-csv, opened before the first fork + std::list m_clients; + uint32_t m_accepted = 0; +}; + +} // namespace remote_device_bridge diff --git a/src/plugins/remote_device_bridge/server/inc/remote_device_bridge/latency_stats.hpp b/src/plugins/remote_device_bridge/server/inc/remote_device_bridge/latency_stats.hpp new file mode 100644 index 0000000000..96d3394402 --- /dev/null +++ b/src/plugins/remote_device_bridge/server/inc/remote_device_bridge/latency_stats.hpp @@ -0,0 +1,63 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include + +#include +#include +#include +#include + +namespace remote_device_bridge +{ + +//! Index-tip to wrist distance in metres: the step-test signal of the latency tools. +float index_curl_m(const XrHandJointLocationEXT* joints); + +//! " | age_ms p50=.. p99=.. max=.. | rtt_ms min=.. last=.. | offset_ms=.. clock=.." for the bridge's status line; +//! "-" marks a value with no data. +std::string format_clock_stats(std::vector ages_ns, const remote_plugin_session::ClockOffsetEstimator& clock); + +//! One pushed hand sample; all times are bridge CLOCK_MONOTONIC ns except capture_sender_ns. +struct SampleCsvRow +{ + int64_t push_mono_ns = 0; //!< after the OpenXR push returned + int64_t recv_mono_ns = 0; //!< frame read off the transport + int64_t capture_sender_ns = 0; + int64_t capture_mapped_ns = 0; + int64_t age_ns = 0; //!< staleness check time - capture_mapped_ns + const char* side = ""; + uint32_t seq = 0; + float curl_m = 0.0f; +}; + +/*! + * @brief --stats-csv writer. + * + * The bridge parent opens it once; forked clients inherit the descriptor and append each row with one write(2) + * (O_APPEND), so rows from concurrent clients never split. + */ +class SampleCsv +{ +public: + static constexpr const char* kHeader = + "push_mono_ns,recv_mono_ns,capture_sender_ns,capture_mapped_ns,age_ms,side,seq,curl\n"; + + //! Truncates `path`, writes the header and keeps it open for appending; throws std::runtime_error. + explicit SampleCsv(const std::string& path); + ~SampleCsv(); + + SampleCsv(const SampleCsv&) = delete; + SampleCsv& operator=(const SampleCsv&) = delete; + + //! False when the row was not written whole; errno says why. + bool write(const SampleCsvRow& row) const; + +private: + int m_fd; +}; + +} // namespace remote_device_bridge diff --git a/src/plugins/remote_device_bridge/server/latency_stats.cpp b/src/plugins/remote_device_bridge/server/latency_stats.cpp new file mode 100644 index 0000000000..a7c4a041f7 --- /dev/null +++ b/src/plugins/remote_device_bridge/server/latency_stats.cpp @@ -0,0 +1,116 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include "inc/remote_device_bridge/latency_stats.hpp" + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace remote_device_bridge +{ + +namespace +{ + +std::runtime_error file_error(const char* what, const std::string& path) +{ + return std::runtime_error(std::string(what) + " stats CSV '" + path + "': " + std::strerror(errno)); +} + +//! Formats one row (newline included) without allocating; returns the snprintf result. +int format_row(char* out, size_t size, const SampleCsvRow& row) +{ + return std::snprintf(out, size, "%lld,%lld,%lld,%lld,%.3f,%s,%u,%.5f\n", static_cast(row.push_mono_ns), + static_cast(row.recv_mono_ns), static_cast(row.capture_sender_ns), + static_cast(row.capture_mapped_ns), static_cast(row.age_ns) / 1e6, row.side, + static_cast(row.seq), static_cast(row.curl_m)); +} + +std::string ms(int64_t ns) +{ + char text[32]; + std::snprintf(text, sizeof(text), "%.1f", static_cast(ns) / 1e6); + return text; +} + +//! Nearest-rank percentile; reorders `values`. +int64_t percentile(std::vector& values, double p) +{ + const size_t rank = static_cast(std::ceil(p * static_cast(values.size()))); + const auto nth = values.begin() + static_cast(std::max(rank, 1) - 1); + std::nth_element(values.begin(), nth, values.end()); + return *nth; +} + +} // namespace + +float index_curl_m(const XrHandJointLocationEXT* joints) +{ + const XrVector3f& tip = joints[XR_HAND_JOINT_INDEX_TIP_EXT].pose.position; + const XrVector3f& wrist = joints[XR_HAND_JOINT_WRIST_EXT].pose.position; + const float dx = tip.x - wrist.x; + const float dy = tip.y - wrist.y; + const float dz = tip.z - wrist.z; + return std::sqrt(dx * dx + dy * dy + dz * dz); +} + +std::string format_clock_stats(std::vector ages_ns, const remote_plugin_session::ClockOffsetEstimator& clock) +{ + const bool two_way = clock.offset_source() == remote_plugin_session::ClockOffsetEstimator::Source::TwoWay; + std::string out = " | age_ms"; + if (ages_ns.empty()) + { + out += " p50=- p99=- max=-"; + } + else + { + out += " p50=" + ms(percentile(ages_ns, 0.50)) + " p99=" + ms(percentile(ages_ns, 0.99)) + + " max=" + ms(*std::max_element(ages_ns.begin(), ages_ns.end())); + } + out += two_way ? " | rtt_ms min=" + ms(clock.rtt_min_ns()) + " last=" + ms(clock.rtt_last_ns()) : + std::string(" | rtt_ms min=- last=-"); + char offset[48]; + std::snprintf(offset, sizeof(offset), "%.3f", static_cast(clock.offset_ns()) / 1e6); + return out + " | offset_ms=" + offset + " clock=" + (two_way ? "two-way" : "one-way"); +} + +SampleCsv::SampleCsv(const std::string& path) + : m_fd(::open(path.c_str(), O_WRONLY | O_CREAT | O_TRUNC | O_APPEND | O_CLOEXEC, 0644)) +{ + if (m_fd < 0) + { + throw file_error("cannot open", path); + } + const size_t size = std::strlen(kHeader); + if (::write(m_fd, kHeader, size) != static_cast(size)) + { + const std::runtime_error error = file_error("cannot write", path); + ::close(m_fd); + throw error; + } +} + +SampleCsv::~SampleCsv() +{ + ::close(m_fd); +} + +bool SampleCsv::write(const SampleCsvRow& row) const +{ + char line[256]; + const int size = format_row(line, sizeof(line), row); + if (size <= 0 || static_cast(size) >= sizeof(line)) + { + return false; + } + // One write(2) per row, so rows from concurrent clients never interleave. + return ::write(m_fd, line, static_cast(size)) == size; +} + +} // namespace remote_device_bridge diff --git a/src/plugins/remote_device_bridge/session/CMakeLists.txt b/src/plugins/remote_device_bridge/session/CMakeLists.txt index 60f7ff9a50..da87d216ba 100644 --- a/src/plugins/remote_device_bridge/session/CMakeLists.txt +++ b/src/plugins/remote_device_bridge/session/CMakeLists.txt @@ -6,6 +6,7 @@ cmake_minimum_required(VERSION 3.20) find_package(Threads REQUIRED) add_library(remote_plugin_session STATIC + inc/remote_plugin_session/bridge_limits.hpp clock_offset_estimator.cpp inc/remote_plugin_session/clock_offset_estimator.hpp protocol_frame.cpp diff --git a/src/plugins/remote_device_bridge/session/inc/remote_plugin_session/bridge_limits.hpp b/src/plugins/remote_device_bridge/session/inc/remote_plugin_session/bridge_limits.hpp new file mode 100644 index 0000000000..ebba894e5f --- /dev/null +++ b/src/plugins/remote_device_bridge/session/inc/remote_plugin_session/bridge_limits.hpp @@ -0,0 +1,28 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#pragma once + +#include +#include +#include + +namespace remote_plugin_session +{ + +//! Largest schema sample the runtime takes: 32 KiB pushed fine up to 720 Hz; >= 40 KiB fails with -2 after a few +//! pushes and the runtime then drops the client's IPC. +constexpr size_t kDefaultMaxSchemaBytes = 32 * 1024; + +//! The refusal text for the client when a schema channel with this sample size exceeds the bridge cap. +inline std::optional schema_cap_error(const std::string& collection_id, size_t sample_bytes, size_t cap_bytes) +{ + if (sample_bytes <= cap_bytes) + { + return std::nullopt; + } + return "schema '" + collection_id + "' sample size " + std::to_string(sample_bytes) + " bytes exceeds bridge cap " + + std::to_string(cap_bytes) + " bytes; channel refused"; +} + +} // namespace remote_plugin_session diff --git a/src/plugins/remote_device_bridge/session/inc/remote_plugin_session/remote_address.hpp b/src/plugins/remote_device_bridge/session/inc/remote_plugin_session/remote_address.hpp index a02ddbd54d..a01e43b763 100644 --- a/src/plugins/remote_device_bridge/session/inc/remote_plugin_session/remote_address.hpp +++ b/src/plugins/remote_device_bridge/session/inc/remote_plugin_session/remote_address.hpp @@ -15,6 +15,10 @@ namespace remote_plugin_session { +//! Ports the bridge listens on (TCP and ws://) and clients dial when the address names none. +constexpr uint16_t kDefaultBridgePort = 47500; +constexpr uint16_t kDefaultBridgeWsPort = 47501; + //! Parses "host[:port]" or "[v6-host][:port]"; a missing port falls back to default_port. Throws //! std::invalid_argument. Endpoint parse_endpoint(const std::string& text, uint16_t default_port); @@ -28,6 +32,8 @@ struct RemoteAddress //! Throws std::invalid_argument; wss:// is refused (terminate TLS in a proxy in front of the bridge). RemoteAddress parse_remote(const std::string& text, uint16_t default_port); +//! parse_remote with the bridge's default port for the address's scheme. +RemoteAddress parse_bridge_remote(const std::string& text); //! Connects (and for ws:// completes the upgrade) within `timeout`; throws std::runtime_error. std::unique_ptr connect_transport(const RemoteAddress& remote, std::chrono::milliseconds timeout = kConnectTimeout); diff --git a/src/plugins/remote_device_bridge/session/remote_address.cpp b/src/plugins/remote_device_bridge/session/remote_address.cpp index aefc49e827..1a52c87843 100644 --- a/src/plugins/remote_device_bridge/session/remote_address.cpp +++ b/src/plugins/remote_device_bridge/session/remote_address.cpp @@ -99,6 +99,11 @@ RemoteAddress parse_remote(const std::string& text, uint16_t default_port) return remote; } +RemoteAddress parse_bridge_remote(const std::string& text) +{ + return parse_remote(text, text.starts_with("ws://") ? kDefaultBridgeWsPort : kDefaultBridgePort); +} + std::unique_ptr connect_transport(const RemoteAddress& remote, std::chrono::milliseconds timeout) { TcpSocket socket = TcpSocket::connect(remote.endpoint, timeout); diff --git a/tests/cpp/plugins/CMakeLists.txt b/tests/cpp/plugins/CMakeLists.txt index 0ce371daea..40ec574700 100644 --- a/tests/cpp/plugins/CMakeLists.txt +++ b/tests/cpp/plugins/CMakeLists.txt @@ -15,4 +15,5 @@ endif() if(TARGET Teleop::remote_plugin_session) add_subdirectory(remote_device_bridge) + add_subdirectory(remote_device_bridge_server) endif() diff --git a/tests/cpp/plugins/remote_device_bridge/CMakeLists.txt b/tests/cpp/plugins/remote_device_bridge/CMakeLists.txt index a7f9589058..a681d6b5a9 100644 --- a/tests/cpp/plugins/remote_device_bridge/CMakeLists.txt +++ b/tests/cpp/plugins/remote_device_bridge/CMakeLists.txt @@ -4,6 +4,7 @@ find_package(Threads REQUIRED) add_executable(remote_plugin_session_tests + test_bridge_limits.cpp test_clock_offset_estimator.cpp test_remote_address.cpp test_remote_plugin_session.cpp diff --git a/tests/cpp/plugins/remote_device_bridge/test_bridge_limits.cpp b/tests/cpp/plugins/remote_device_bridge/test_bridge_limits.cpp new file mode 100644 index 0000000000..a398b9e1d3 --- /dev/null +++ b/tests/cpp/plugins/remote_device_bridge/test_bridge_limits.cpp @@ -0,0 +1,22 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include +#include + +using remote_plugin_session::kDefaultMaxSchemaBytes; +using remote_plugin_session::schema_cap_error; + +TEST_CASE("schema_cap_error accepts a schema sample at the cap", "[remote_device_bridge][unit]") +{ + REQUIRE_FALSE(schema_cap_error("tactile", kDefaultMaxSchemaBytes, kDefaultMaxSchemaBytes).has_value()); + REQUIRE_FALSE(schema_cap_error("tactile", 4096, kDefaultMaxSchemaBytes).has_value()); +} + +TEST_CASE("schema_cap_error refuses a schema sample above the cap", "[remote_device_bridge][unit]") +{ + REQUIRE(schema_cap_error("tactile", kDefaultMaxSchemaBytes + 1, kDefaultMaxSchemaBytes) == + "schema 'tactile' sample size 32769 bytes exceeds bridge cap 32768 bytes; channel refused"); + REQUIRE(schema_cap_error("tactile", 40960, kDefaultMaxSchemaBytes).has_value()); + REQUIRE_FALSE(schema_cap_error("tactile", 40960, 65536).has_value()); // --max-schema-bytes raises the cap +} diff --git a/tests/cpp/plugins/remote_device_bridge/test_remote_address.cpp b/tests/cpp/plugins/remote_device_bridge/test_remote_address.cpp index e6be1f3ad1..8359070f17 100644 --- a/tests/cpp/plugins/remote_device_bridge/test_remote_address.cpp +++ b/tests/cpp/plugins/remote_device_bridge/test_remote_address.cpp @@ -44,3 +44,10 @@ TEST_CASE("parse_endpoint validates ports and accepts IPv6 hosts", "[remote_devi REQUIRE(bare_v6.port == 47500); REQUIRE(parse_endpoint("[::1]", 47500).port == 47500); } + +TEST_CASE("parse_bridge_remote defaults to the bridge port of the scheme", "[remote_device_bridge][unit]") +{ + REQUIRE(parse_bridge_remote("10.0.0.2").endpoint.port == kDefaultBridgePort); + REQUIRE(parse_bridge_remote("ws://10.0.0.2").endpoint.port == kDefaultBridgeWsPort); + REQUIRE(parse_bridge_remote("ws://10.0.0.2:9000/bridge").endpoint.port == 9000); +} diff --git a/tests/cpp/plugins/remote_device_bridge_server/CMakeLists.txt b/tests/cpp/plugins/remote_device_bridge_server/CMakeLists.txt new file mode 100644 index 0000000000..b1a9c9ab97 --- /dev/null +++ b/tests/cpp/plugins/remote_device_bridge_server/CMakeLists.txt @@ -0,0 +1,15 @@ +# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +# SPDX-License-Identifier: Apache-2.0 + +add_executable(remote_device_bridge_server_tests + test_bridge_server.cpp + test_latency_stats.cpp +) + +target_link_libraries(remote_device_bridge_server_tests PRIVATE + Teleop::remote_device_bridge_server + Teleop::remote_plugin_session + Catch2::Catch2WithMain +) + +catch_discover_tests(remote_device_bridge_server_tests ADD_TAGS_AS_LABELS) diff --git a/tests/cpp/plugins/remote_device_bridge_server/test_bridge_server.cpp b/tests/cpp/plugins/remote_device_bridge_server/test_bridge_server.cpp new file mode 100644 index 0000000000..cedfc44a1b --- /dev/null +++ b/tests/cpp/plugins/remote_device_bridge_server/test_bridge_server.cpp @@ -0,0 +1,1013 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +// End-to-end tests of the real BridgeServer (fork per client) over loopback TCP. The bridge runs in a forked +// process so the test's client threads never exist in a process that forks; its plugin session is a fake that +// logs every call, so no OpenXR runtime is involved. Bridge and fake output share one log file. + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace remote_plugin_session; +using remote_device_bridge::BridgeOptions; +using remote_device_bridge::BridgeServer; +namespace protocol = remote_plugin_protocol; + +namespace +{ + +using Ms = std::chrono::milliseconds; +constexpr Ms kWait{ 5000 }; +constexpr int64_t kMs = 1'000'000; + +//! Read by the fake session in the bridge process; set before BridgeProcess forks. +struct FakeBehavior +{ + int first_push_delay_ms = 0; + XrResult hand_create_result = XR_SUCCESS; //!< xrCreatePushDeviceNV result the fake reports + int pull_update_delay_ms = 0; //!< stands in for the OpenXR work of a real pull update +}; +FakeBehavior g_fake; + +std::atomic g_bridge_stop{ false }; + +void on_sigterm(int) +{ + g_bridge_stop.store(true); +} + +//! Same category name and what() text as plugin_utils' CheckXrResult. +class XrResultCategory final : public std::error_category +{ +public: + const char* name() const noexcept override + { + return "XrResult"; + } + std::string message(int value) const override + { + return std::to_string(value); + } +}; + +void log_event(const std::string& line) +{ + std::cout << "[fake pid=" << getpid() << "] " << line << std::endl; +} + +const char* side(XrHandEXT hand) +{ + return hand == XR_HAND_LEFT_EXT ? "left" : "right"; +} + +class FakeHandChannel final : public core::IHandTrackingPushChannel +{ +public: + explicit FakeHandChannel(XrHandEXT hand) : m_hand(hand) + { + log_event(std::string("open hand ") + side(hand)); + } + ~FakeHandChannel() override + { + log_event(std::string("close hand ") + side(m_hand)); + } + void push(const XrHandJointLocationEXT*, int64_t sample_time_ns) override + { + if (m_pushes++ == 0 && g_fake.first_push_delay_ms > 0) + { + std::this_thread::sleep_for(Ms(g_fake.first_push_delay_ms)); + } + log_event(std::string("push hand ") + side(m_hand) + " t=" + std::to_string(sample_time_ns)); + } + +private: + XrHandEXT m_hand; + uint64_t m_pushes = 0; +}; + +class FakeSchemaChannel final : public core::ISchemaPushChannel +{ +public: + explicit FakeSchemaChannel(core::SchemaPusherConfig config) : m_config(std::move(config)) + { + log_event("open schema " + m_config.collection_id); + } + ~FakeSchemaChannel() override + { + log_event("close schema " + m_config.collection_id); + } + void push_buffer(const uint8_t*, size_t size, int64_t, int64_t) override + { + log_event("push schema " + m_config.collection_id + " size=" + std::to_string(size)); + } + +private: + core::SchemaPusherConfig m_config; +}; + +class FakeWristSource final : public core::IWristTrackingSource +{ +public: + core::WristTrackingSample query(XrHandEXT, int64_t) override + { + return {}; + } +}; + +//! Created with the session, like the OpenXR one; the wrist source exists only when HELLO declared it. +class FakePullChannel final : public core::IPluginPullChannel +{ +public: + explicit FakePullChannel(bool has_wrist) : m_has_wrist(has_wrist) + { + } + void update() override + { + std::this_thread::sleep_for(Ms(g_fake.pull_update_delay_ms)); + } + const core::ITrackerImpl& get_tracker_impl(const core::ITracker&) const override + { + throw std::runtime_error("no trackers"); + } + core::IWristTrackingSource& wrist_tracking_source() override + { + if (!m_has_wrist) + { + throw std::logic_error("wrist tracking pull was not declared"); + } + return m_wrist; + } + +private: + bool m_has_wrist; + FakeWristSource m_wrist; +}; + +class FakeSession final : public core::IPluginSession +{ +public: + FakeSession(const std::string& app_name, const core::PluginSessionRequirements& requirements) + : m_pull(requirements.wrist_tracking_pull.has_value()) + { + log_event("session open app=" + app_name); + } + ~FakeSession() override + { + log_event("session closed"); + } + core::IPluginPullChannel& pull_channel() override + { + return m_pull; + } + std::unique_ptr create_schema_push_channel(const core::SchemaPusherConfig& config) override + { + return std::make_unique(config); + } + std::unique_ptr create_hand_tracking_push_channel(XrHandEXT hand) override + { + if (XR_FAILED(g_fake.hand_create_result)) + { + static const XrResultCategory category; + throw std::system_error(g_fake.hand_create_result, category, + std::string("xrCreatePushDeviceNV(") + side(hand) + ") failed with XrResult"); + } + return std::make_unique(hand); + } + +private: + FakePullChannel m_pull; +}; + +bool wait_until(const std::function& predicate, Ms timeout = kWait) +{ + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (!predicate()) + { + if (std::chrono::steady_clock::now() > deadline) + { + return false; + } + std::this_thread::sleep_for(Ms(10)); + } + return true; +} + +size_t count(const std::string& text, const std::string& needle) +{ + size_t n = 0; + for (size_t pos = text.find(needle); pos != std::string::npos; pos = text.find(needle, pos + needle.size())) + { + ++n; + } + return n; +} + +//! Sum of every "=" on the bridge's status lines. +uint64_t status_sum(const std::string& log, const std::string& key) +{ + const std::regex pattern(" 5s: (?:.*[ |])?" + key + "=([0-9]+)"); + uint64_t sum = 0; + std::istringstream lines(log); + for (std::string line; std::getline(lines, line);) + { + std::smatch match; + if (std::regex_search(line, match, pattern)) + { + sum += std::stoull(match[1]); + } + } + return sum; +} + +//! Sample times the fake pushed for one hand, in push order. +std::vector pushed_times(const std::string& log, const char* hand_side) +{ + const std::regex pattern(std::string("push hand ") + hand_side + " t=(-?[0-9]+)"); + std::vector times; + for (std::sregex_iterator it(log.begin(), log.end(), pattern), end; it != end; ++it) + { + times.push_back(std::stoll((*it)[1])); + } + return times; +} + +//! The bridge in its own process, logging to a temp file. The destructor stops it like Ctrl-C would. +class BridgeProcess +{ +public: + explicit BridgeProcess(BridgeOptions options = {}) + { + options.listen = Endpoint{ "127.0.0.1", 0 }; + std::string path = (std::filesystem::temp_directory_path() / "remote_device_bridge_test_XXXXXX").string(); + const int log_fd = mkstemp(path.data()); + REQUIRE(log_fd >= 0); + close(log_fd); + m_log_path = path; + + int ready[2]; + REQUIRE(pipe(ready) == 0); + std::cout.flush(); + std::cerr.flush(); + std::fflush(nullptr); + m_pid = fork(); + REQUIRE(m_pid >= 0); + if (m_pid == 0) + { + close(ready[0]); + run_bridge(options, ready[1]); + } + close(ready[1]); + pollfd fd{ ready[0], POLLIN, 0 }; + std::array ports{}; + if (poll(&fd, 1, static_cast(kWait.count())) == 1 && + read(ready[0], ports.data(), sizeof(ports)) == sizeof(ports)) + { + m_port = ports[0]; + m_ws_port = ports[1]; + } + close(ready[0]); + REQUIRE(m_port != 0); + } + + ~BridgeProcess() + { + kill(m_pid, SIGTERM); + int status = 0; + const bool exited = wait_until([&] { return waitpid(m_pid, &status, WNOHANG) == m_pid; }, Ms(10000)); + if (!exited) + { + kill(m_pid, SIGKILL); + waitpid(m_pid, &status, 0); + } + if (std::uncaught_exceptions() > 0 || !exited) + { + std::cerr << "---- bridge log ----\n" << log() << "--------------------" << std::endl; + } + std::filesystem::remove(m_log_path); + } + + BridgeProcess(const BridgeProcess&) = delete; + BridgeProcess& operator=(const BridgeProcess&) = delete; + + uint16_t port() const + { + return m_port; + } + + //! 0 unless the options set ws_listen. + uint16_t ws_port() const + { + return m_ws_port; + } + + std::string log() const + { + std::ifstream file(m_log_path); + std::stringstream text; + text << file.rdbuf(); + return text.str(); + } + + size_t count(const std::string& needle) const + { + return ::count(log(), needle); + } + + bool wait_log(const std::string& needle, size_t times = 1, Ms timeout = kWait) const + { + return wait_until([&] { return count(needle) >= times; }, timeout); + } + + //! pid of the bridge child serving the client that sent HELLO with `app_name`. + pid_t client_pid(const std::string& app_name) const + { + const std::regex pattern("\\[fake pid=([0-9]+)\\] session open app=" + app_name + "\n"); + const std::string text = log(); + std::smatch match; + return std::regex_search(text, match, pattern) ? static_cast(std::stoi(match[1])) : 0; + } + +private: + [[noreturn]] void run_bridge(const BridgeOptions& options, int ready_fd) + { + const int log_fd = open(m_log_path.c_str(), O_WRONLY | O_APPEND); + dup2(log_fd, STDOUT_FILENO); + dup2(log_fd, STDERR_FILENO); + close(log_fd); + std::signal(SIGTERM, on_sigterm); + std::array ports{}; + try + { + BridgeServer server(options, + [](const std::string& app_name) -> core::PluginSessionFactory + { + return [app_name](const core::PluginSessionRequirements& requirements, + std::vector>) + { return std::make_unique(app_name, requirements); }; + }); + ports = { server.port(), server.ws_port().value_or(0) }; + (void)!write(ready_fd, ports.data(), sizeof(ports)); + close(ready_fd); + server.run(g_bridge_stop); + } + catch (const std::exception& e) + { + std::cerr << "bridge failed: " << e.what() << std::endl; + (void)!write(ready_fd, ports.data(), sizeof(ports)); + } + std::cout.flush(); + std::cerr.flush(); + _exit(0); + } + + std::string m_log_path; + pid_t m_pid = -1; + uint16_t m_port = 0; + uint16_t m_ws_port = 0; +}; + +//! Speaks the wire protocol by hand: optionally no HELLO, any capture time, no PONGs. Heartbeats keep the bridge's +//! 3 s peer timeout away. +class RawClient +{ +public: + explicit RawClient(uint16_t port, bool send_hello = true) + : m_transport(std::make_unique(TcpSocket::connect(Endpoint{ "127.0.0.1", port }, kWait))) + { + if (!send_hello) + { + return; + } + protocol::Hello hello; + hello.app_name = "raw"; + hello.requirements = { .schema_push = true, + .hand_tracking_push = true, + .wrist_tracking_pull = core::WristTrackingSourceConfig{} }; + send(protocol::FrameType::Hello, protocol::kSessionChannelId, protocol::encode_hello(hello)); + m_heartbeat = std::thread( + [this] + { + while (!m_stop.load()) + { + send(protocol::FrameType::Heartbeat, protocol::kSessionChannelId, {}); + std::this_thread::sleep_for(Ms(300)); + } + }); + } + + ~RawClient() + { + disconnect(); + } + + bool send(protocol::FrameType type, uint16_t channel_id, const std::vector& payload) + { + return send_frame(protocol::encode_frame(type, channel_id, payload)); + } + + bool send_hand_sample(uint16_t channel_id, int64_t capture_ns) + { + const std::array joints{}; + return send_frame(protocol::encode_hand_sample_frame(channel_id, joints.data(), capture_ns)); + } + + //! Next frame of `type`; frames of other types are counted and skipped. + std::optional wait_for(protocol::FrameType type, Ms timeout = kWait) + { + const auto deadline = std::chrono::steady_clock::now() + timeout; + while (std::chrono::steady_clock::now() < deadline) + { + std::vector bytes; + const auto left = std::chrono::duration_cast(deadline - std::chrono::steady_clock::now()); + const ReceiveResult result = m_transport->receive(bytes, std::max(left, Ms(1))); + if (result == ReceiveResult::Timeout) + { + continue; + } + if (result != ReceiveResult::Ok) + { + return std::nullopt; + } + protocol::Frame frame = protocol::split_frame(std::move(bytes)); + m_receive_ns[frame.header.type].push_back(core::os_monotonic_now_ns()); + if (frame.header.type == type) + { + return frame; + } + } + return std::nullopt; + } + + //! Receive times of every frame of `type` seen so far. + const std::vector& received(protocol::FrameType type) + { + return m_receive_ns[type]; + } + + void disconnect() + { + m_stop.store(true); + if (m_heartbeat.joinable()) + { + m_heartbeat.join(); + } + m_transport->shutdown(); + } + +private: + bool send_frame(std::vector frame) + { + std::lock_guard lock(m_send_mutex); + protocol::stamp_header(frame, m_seq++, core::os_monotonic_now_ns()); + return m_transport->send(frame.data(), frame.size()); + } + + std::unique_ptr m_transport; + std::mutex m_send_mutex; + uint32_t m_seq = 0; + std::map> m_receive_ns; + std::atomic m_stop{ false }; + std::thread m_heartbeat; +}; + +bool accepted(RawClient& client) +{ + const auto ack = client.wait_for(protocol::FrameType::HelloAck); + return ack.has_value() && protocol::decode_hello_ack(ack->payload).accepted; +} + +core::SchemaPusherConfig schema_config(const std::string& collection_id, size_t max_flatbuffer_size) +{ + return { .collection_id = collection_id, + .max_flatbuffer_size = max_flatbuffer_size, + .tensor_identifier = "tensor", + .localized_name = collection_id, + .app_name = "test" }; +} + +std::unique_ptr connect(uint16_t port, const std::string& app_name) +{ + return std::make_unique( + std::make_unique(TcpSocket::connect(Endpoint{ "127.0.0.1", port }, kWait)), app_name, + core::PluginSessionRequirements{ + .schema_push = true, .hand_tracking_push = true, .wrist_tracking_pull = core::WristTrackingSourceConfig{} }); +} + +const std::array kJoints{}; + +//! Pushes to `channel` every 10 ms on a thread until destroyed. +class Streamer +{ +public: + explicit Streamer(core::IHandTrackingPushChannel& channel) + : m_thread( + [this, &channel] + { + while (!m_stop.load()) + { + channel.push(kJoints.data(), core::os_monotonic_now_ns()); + std::this_thread::sleep_for(Ms(10)); + } + }) + { + } + ~Streamer() + { + m_stop.store(true); + m_thread.join(); + } + +private: + std::atomic m_stop{ false }; + std::thread m_thread; +}; + +} // namespace + +TEST_CASE("Bridge opens plugin-session channels only on OPEN and releases them on CLOSE and disconnect", + "[remote_device_bridge][unit]") +{ + BridgeProcess bridge; + RawClient client(bridge.port()); + REQUIRE(accepted(client)); + REQUIRE(bridge.wait_log("session open app=raw")); + + std::this_thread::sleep_for(Ms(300)); + REQUIRE(bridge.count("] open ") == 0); + + REQUIRE(client.send(protocol::FrameType::OpenHand, 1, protocol::encode_open_hand(XR_HAND_LEFT_EXT))); + REQUIRE(bridge.wait_log("open hand left")); + REQUIRE(client.send(protocol::FrameType::Close, 1, {})); + REQUIRE(bridge.wait_log("close hand left")); + + REQUIRE(client.send(protocol::FrameType::OpenHand, 2, protocol::encode_open_hand(XR_HAND_RIGHT_EXT))); + REQUIRE(client.send(protocol::FrameType::OpenSchema, 3, protocol::encode_open_schema(schema_config("sensors", 256)))); + REQUIRE(client.send(protocol::FrameType::OpenPull, 4, {})); + REQUIRE(bridge.wait_log("channel 4: pull channel opened with wrist source")); + + // Disconnect without CLOSE: the bridge releases everything, push channels before the session (which owns the + // pull channel). + client.disconnect(); + REQUIRE(bridge.wait_log("session closed")); + const std::string log = bridge.log(); + REQUIRE(count(log, "close hand right") == 1); + REQUIRE(count(log, "close schema sensors") == 1); + REQUIRE(log.find("close hand right") < log.find("session closed")); + REQUIRE(log.find("close schema sensors") < log.find("session closed")); +} + +TEST_CASE("Bridge maps the sender's capture time by the clock offset only", "[remote_device_bridge][unit]") +{ + BridgeProcess bridge; + RawClient client(bridge.port()); + REQUIRE(accepted(client)); + REQUIRE(client.send(protocol::FrameType::OpenHand, 1, protocol::encode_open_hand(XR_HAND_LEFT_EXT))); + REQUIRE(bridge.wait_log("open hand left")); + + // Same host, same CLOCK_MONOTONIC: the offset is ~0, so the pushed time is the capture time give or take + // the loopback delay. + const int64_t capture_ns = core::os_monotonic_now_ns() - 100 * kMs; + REQUIRE(client.send_hand_sample(1, capture_ns)); + REQUIRE(bridge.wait_log("push hand left")); + const std::vector pushed = pushed_times(bridge.log(), "left"); + REQUIRE(pushed.size() == 1); + INFO("pushed - capture = " << pushed[0] - capture_ns << " ns"); + REQUIRE(std::abs(pushed[0] - capture_ns) < 5 * kMs); +} + +TEST_CASE("Killing one client process leaves the bridge and the other clients running", "[remote_device_bridge][unit]") +{ + BridgeProcess bridge; + auto a = connect(bridge.port(), "client_a"); + auto b = connect(bridge.port(), "client_b"); + auto a_left = a->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + auto b_right = b->create_hand_tracking_push_channel(XR_HAND_RIGHT_EXT); + REQUIRE(bridge.wait_log("open hand left")); + REQUIRE(bridge.wait_log("open hand right")); + const pid_t a_pid = bridge.client_pid("client_a"); + REQUIRE(a_pid > 0); + REQUIRE(bridge.client_pid("client_b") != a_pid); + + REQUIRE(kill(a_pid, SIGKILL) == 0); + REQUIRE(bridge.wait_log("client pid " + std::to_string(a_pid) + " exited (signal 9)")); + REQUIRE(wait_until([&] { return !a->connected(); })); + + const size_t before = bridge.count("push hand right"); + { + Streamer streaming(*b_right); + REQUIRE(bridge.wait_log("push hand right", before + 10)); + } + REQUIRE(b->connected()); + + // The parent is still accepting, and the dead child's hand is free again. + auto c = connect(bridge.port(), "client_c"); + auto c_left = c->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + REQUIRE(bridge.wait_log("open hand left", 2)); + REQUIRE(c->stats().refused_channels.empty()); +} + +TEST_CASE("A second client opening an owned hand is refused and the owner keeps streaming", "[remote_device_bridge][unit]") +{ + BridgeProcess bridge; + auto a = connect(bridge.port(), "client_a"); + auto b = connect(bridge.port(), "client_b"); + auto a_left = a->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + REQUIRE(bridge.wait_log("open hand left")); + Streamer streaming(*a_left); + REQUIRE(bridge.wait_log("push hand left", 5)); + + auto b_left = b->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + REQUIRE(wait_until([&] { return b->stats().refused_channels.size() == 1; })); + const std::string reason = b->stats().refused_channels.begin()->second; + REQUIRE(reason == "hand left is already owned by bridge client c1; channel refused"); + REQUIRE(bridge.wait_log("refused: hand left is already owned by bridge client c1; channel refused")); + REQUIRE(b->connected()); + + // The refused client's session and other channels work; the owner never noticed. + auto b_right = b->create_hand_tracking_push_channel(XR_HAND_RIGHT_EXT); + b_left->push(kJoints.data(), core::os_monotonic_now_ns()); + b_right->push(kJoints.data(), core::os_monotonic_now_ns()); + REQUIRE(bridge.wait_log("push hand right")); + const size_t before = bridge.count("push hand left"); + REQUIRE(bridge.wait_log("push hand left", before + 10)); + REQUIRE(a->connected()); + REQUIRE(a->stats().refused_channels.empty()); + REQUIRE(bridge.count("open hand left") == 1); + REQUIRE(bridge.count("closing connection") == 0); +} + +TEST_CASE("A hand the runtime refuses is a channel refusal, not a session error", "[remote_device_bridge][unit]") +{ + g_fake.hand_create_result = XR_ERROR_RUNTIME_FAILURE; + BridgeProcess bridge; + g_fake.hand_create_result = XR_SUCCESS; + auto client = connect(bridge.port(), "client"); + auto left = client->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + REQUIRE(wait_until([&] { return client->stats().refused_channels.size() == 1; })); + REQUIRE(client->stats().refused_channels.begin()->second == + "hand left is already owned by another OpenXR session (runtime refused the push device); channel refused"); + auto schema = client->create_schema_push_channel(schema_config("after", 64)); + REQUIRE(bridge.wait_log("open schema after")); + REQUIRE(client->connected()); +} + +TEST_CASE("Any other runtime failure opening a hand ends the session", "[remote_device_bridge][unit]") +{ + XrResult result = XR_SUCCESS; + SECTION("XR_ERROR_VALIDATION_FAILURE") + { + result = XR_ERROR_VALIDATION_FAILURE; + } + SECTION("XR_ERROR_SESSION_LOST") + { + result = XR_ERROR_SESSION_LOST; + } + g_fake.hand_create_result = result; + BridgeProcess bridge; + g_fake.hand_create_result = XR_SUCCESS; + RawClient client(bridge.port()); + REQUIRE(accepted(client)); + REQUIRE(client.send(protocol::FrameType::OpenHand, 1, protocol::encode_open_hand(XR_HAND_LEFT_EXT))); + + const std::string message = "xrCreatePushDeviceNV(left) failed with XrResult: " + std::to_string(result); + const auto error = client.wait_for(protocol::FrameType::Error); + REQUIRE(error.has_value()); + REQUIRE(error->header.channel_id == protocol::kSessionChannelId); + REQUIRE(protocol::decode_error(error->payload) == message); + REQUIRE(bridge.wait_log("[bridge c1] error: " + message + "; closing connection")); + REQUIRE(bridge.wait_log("session closed")); + REQUIRE(bridge.count("refused") == 0); +} + +TEST_CASE("A schema channel above 32768 bytes is refused at open and the session keeps working", + "[remote_device_bridge][unit]") +{ + BridgeOptions options; + options.max_schema_bytes = 32768; + BridgeProcess bridge(options); + auto client = connect(bridge.port(), "client"); + auto too_big = client->create_schema_push_channel(schema_config("tactile", 32769)); + auto at_cap = client->create_schema_push_channel(schema_config("at_cap", 32768)); + auto left = client->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + + REQUIRE(wait_until([&] { return client->stats().refused_channels.size() == 1; })); + const std::string reason = client->stats().refused_channels.begin()->second; + REQUIRE(reason == "schema 'tactile' sample size 32769 bytes exceeds bridge cap 32768 bytes; channel refused"); + REQUIRE(bridge.wait_log("refused: " + reason)); + REQUIRE(bridge.wait_log("open schema at_cap")); + REQUIRE(bridge.wait_log("open hand left")); + REQUIRE(bridge.count("open schema tactile") == 0); + + const std::vector bytes(100, 0x11); + too_big->push_buffer(bytes.data(), bytes.size(), core::os_monotonic_now_ns(), 0); + at_cap->push_buffer(bytes.data(), bytes.size(), core::os_monotonic_now_ns(), 0); + left->push(kJoints.data(), core::os_monotonic_now_ns()); + REQUIRE(bridge.wait_log("push schema at_cap size=100")); + REQUIRE(bridge.wait_log("push hand left")); + + too_big.reset(); // CLOSE for the refused id is accepted quietly + at_cap->push_buffer(bytes.data(), bytes.size(), core::os_monotonic_now_ns(), 0); + REQUIRE(bridge.wait_log("push schema at_cap size=100", 2)); + REQUIRE(client->connected()); + REQUIRE(bridge.count("closing connection") == 0); + REQUIRE(bridge.count("CLOSE for unknown channel") == 0); +} + +TEST_CASE("The 9th client gets HELLO_ACK 'bridge full' within 500 ms", "[remote_device_bridge][unit]") +{ + BridgeOptions options; + options.max_clients = 8; + BridgeProcess bridge(options); + std::vector> clients; + for (int i = 0; i < 8; ++i) + { + clients.push_back(connect(bridge.port(), "client" + std::to_string(i))); + } + REQUIRE(bridge.wait_log("session open app=", 8)); + + SECTION("a client that sends HELLO") + { + const auto start = std::chrono::steady_clock::now(); + try + { + connect(bridge.port(), "ninth"); + FAIL("expected the bridge to refuse the 9th client"); + } + catch (const std::runtime_error& e) + { + const auto took = std::chrono::steady_clock::now() - start; + REQUIRE(took < Ms(500)); + REQUIRE(std::string(e.what()) == "RemotePluginSession: bridge rejected the session: bridge full"); + } + } + SECTION("a client that never sends HELLO") + { + const auto start = std::chrono::steady_clock::now(); + RawClient silent(bridge.port(), false); + const auto ack = silent.wait_for(protocol::FrameType::HelloAck, Ms(1000)); + REQUIRE(ack.has_value()); + const auto took = std::chrono::steady_clock::now() - start; + REQUIRE(took < Ms(500)); + const protocol::HelloAck decoded = protocol::decode_hello_ack(ack->payload); + REQUIRE_FALSE(decoded.accepted); + REQUIRE(decoded.reason == "bridge full"); + } + REQUIRE(bridge.wait_log("refused: bridge full (8)")); + for (const auto& client : clients) + { + REQUIRE(client->connected()); + } +} + +TEST_CASE("A refused client whose HELLO is 400 ms late still gets 'bridge full'", "[remote_device_bridge][unit]") +{ + BridgeOptions options; + options.max_clients = 1; + options.ws_listen = Endpoint{ "127.0.0.1", 0 }; + BridgeProcess bridge(options); + auto first = connect(bridge.port(), "first"); + REQUIRE(bridge.wait_log("session open app=first")); + + protocol::Hello hello; + hello.app_name = "late"; + std::vector hello_frame = + protocol::encode_frame(protocol::FrameType::Hello, protocol::kSessionChannelId, protocol::encode_hello(hello)); + protocol::stamp_header(hello_frame, 0, core::os_monotonic_now_ns()); + + std::unique_ptr late; + std::vector bytes; + SECTION("tcp: the ACK is there before the HELLO is sent") + { + TcpSocket socket = TcpSocket::connect(Endpoint{ "127.0.0.1", bridge.port() }, kWait); + std::this_thread::sleep_for(Ms(400)); + late = std::make_unique(std::move(socket)); + REQUIRE(late->receive(bytes, Ms(0)) == ReceiveResult::Ok); + } + SECTION("ws: the upgrade itself arrives 400 ms late") + { + REQUIRE(bridge.ws_port() != 0); + TcpSocket socket = TcpSocket::connect(Endpoint{ "127.0.0.1", bridge.ws_port() }, kWait); + std::this_thread::sleep_for(Ms(400)); + late = std::make_unique( + std::move(socket), "127.0.0.1:" + std::to_string(bridge.ws_port()), "/", kWait); + } + REQUIRE(late->send(hello_frame.data(), hello_frame.size())); + if (bytes.empty()) + { + REQUIRE(late->receive(bytes, kWait) == ReceiveResult::Ok); + } + const protocol::Frame frame = protocol::split_frame(std::move(bytes)); + REQUIRE(frame.header.type == protocol::FrameType::HelloAck); + const protocol::HelloAck ack = protocol::decode_hello_ack(frame.payload); + REQUIRE_FALSE(ack.accepted); + REQUIRE(ack.reason == "bridge full"); + // The refusal child read the late HELLO and closed; nothing else follows the ACK. + REQUIRE(late->receive(bytes, Ms(2000)) == ReceiveResult::Closed); + REQUIRE(bridge.wait_log("refused: bridge full (1)")); + REQUIRE(first->connected()); +} + +TEST_CASE("A HELLO of another version gets HELLO_ACK(false) with the logged reason, then the close", + "[remote_device_bridge][unit]") +{ + BridgeOptions options; + options.ws_listen = Endpoint{ "127.0.0.1", 0 }; + BridgeProcess bridge(options); + + protocol::Hello hello; + hello.app_name = "v1"; + std::vector payload = protocol::encode_hello(hello); + payload[0] = 1; // little-endian u16 version, as a HELLO v1 client sends it + payload[1] = 0; + std::vector hello_frame = + protocol::encode_frame(protocol::FrameType::Hello, protocol::kSessionChannelId, payload); + protocol::stamp_header(hello_frame, 0, core::os_monotonic_now_ns()); + + std::unique_ptr old_client; + SECTION("tcp") + { + old_client = std::make_unique(TcpSocket::connect(Endpoint{ "127.0.0.1", bridge.port() }, kWait)); + } + SECTION("ws") + { + old_client = + std::make_unique(TcpSocket::connect(Endpoint{ "127.0.0.1", bridge.ws_port() }, kWait), + "127.0.0.1:" + std::to_string(bridge.ws_port()), "/", kWait); + } + REQUIRE(old_client->send(hello_frame.data(), hello_frame.size())); + std::vector bytes; + REQUIRE(old_client->receive(bytes, kWait) == ReceiveResult::Ok); + const protocol::Frame frame = protocol::split_frame(std::move(bytes)); + REQUIRE(frame.header.type == protocol::FrameType::HelloAck); + const protocol::HelloAck ack = protocol::decode_hello_ack(frame.payload); + REQUIRE_FALSE(ack.accepted); + REQUIRE(ack.reason == "bad HELLO (protocol: unsupported HELLO version 1)"); + REQUIRE(old_client->receive(bytes, Ms(2000)) == ReceiveResult::Closed); + REQUIRE(bridge.wait_log("] " + ack.reason + "; dropping connection")); + + // The bridge keeps serving. + auto current = connect(bridge.port(), "current"); + REQUIRE(current->connected()); +} + +TEST_CASE("With wrist pull, a hang-up still applies the client's CLOSEs and logs the disconnect", + "[remote_device_bridge][unit]") +{ + // A slow pull update keeps the run thread inside tick(), so its next wrist send is the first to see the hang-up. + g_fake.pull_update_delay_ms = 20; + BridgeProcess bridge; + g_fake.pull_update_delay_ms = 0; + { + auto client = connect(bridge.port(), "wrist"); + auto left = client->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + left->push(kJoints.data(), core::os_monotonic_now_ns()); + REQUIRE(bridge.wait_log("push hand left")); + REQUIRE(wait_until([&] { return client->stats().wrist_rx > 100; })); + } + REQUIRE(bridge.wait_log("session closed")); + const std::string log = bridge.log(); + REQUIRE(count(log, "hand channel 2 closed") == 1); + REQUIRE(count(log, "(pull) closed") == 1); + REQUIRE(count(log, "client disconnected") == 1); +} + +TEST_CASE("Bridge pushes the newest sample per pass, drops stale ones and applies every control frame", + "[remote_device_bridge][unit]") +{ + BridgeOptions options; + options.max_sample_age = Ms(250); + options.stats_period = Ms(500); + g_fake.first_push_delay_ms = 200; // holds the bridge's run thread so the burst below queues up + BridgeProcess bridge(options); + g_fake.first_push_delay_ms = 0; + + RawClient client(bridge.port()); + REQUIRE(accepted(client)); + REQUIRE(client.send(protocol::FrameType::OpenHand, 1, protocol::encode_open_hand(XR_HAND_LEFT_EXT))); + REQUIRE(bridge.wait_log("open hand left")); + + REQUIRE(client.send_hand_sample(1, core::os_monotonic_now_ns())); + std::this_thread::sleep_for(Ms(50)); + // 10 samples 10 ms apart with two control frames in the middle. The newest is applied ~150 ms after its + // capture, inside the 250 ms staleness limit. + const int64_t newest_ns = core::os_monotonic_now_ns(); + for (int i = 0; i < 10; ++i) + { + REQUIRE(client.send_hand_sample(1, newest_ns - (9 - i) * 10 * kMs)); + if (i == 4) + { + REQUIRE(client.send( + protocol::FrameType::OpenSchema, 2, protocol::encode_open_schema(schema_config("control", 64)))); + REQUIRE(client.send(protocol::FrameType::OpenPull, 3, {})); + } + } + + REQUIRE(bridge.wait_log("push hand left", 2)); + REQUIRE(bridge.wait_log("open schema control")); + REQUIRE(bridge.wait_log("pull channel opened")); + REQUIRE(wait_until([&] { return status_sum(bridge.log(), "coalesced") == 9; })); + const std::vector pushed = pushed_times(bridge.log(), "left"); + REQUIRE(pushed.size() == 2); + INFO("pushed - newest = " << pushed[1] - newest_ns << " ns"); + REQUIRE(std::abs(pushed[1] - newest_ns) < 5 * kMs); // the newest, not one 10 ms older + + REQUIRE(client.send_hand_sample(1, core::os_monotonic_now_ns() - 1000 * kMs)); + REQUIRE(wait_until([&] { return status_sum(bridge.log(), "stale") == 1; })); + REQUIRE(pushed_times(bridge.log(), "left").size() == 2); + REQUIRE(status_sum(bridge.log(), "rx hand") == 12); + REQUIRE(status_sum(bridge.log(), "applied") == 2); +} + +TEST_CASE("With --clock-source=one-way the bridge never PINGs and streams on the one-way offset", + "[remote_device_bridge][unit]") +{ + BridgeOptions options; + options.clock_source = remote_device_bridge::ClockSource::OneWay; + BridgeProcess bridge(options); + RawClient client(bridge.port()); + REQUIRE(accepted(client)); + REQUIRE(client.send(protocol::FrameType::OpenHand, 1, protocol::encode_open_hand(XR_HAND_RIGHT_EXT))); + REQUIRE(client.send_hand_sample(1, core::os_monotonic_now_ns())); + REQUIRE(bridge.wait_log("push hand right")); + + // Past the PING burst and the 2 s PONG timeout. + REQUIRE_FALSE(client.wait_for(protocol::FrameType::Ping, Ms(2500)).has_value()); + REQUIRE(client.received(protocol::FrameType::Ping).empty()); + REQUIRE(bridge.count("clock: no PONG") == 0); +} + +TEST_CASE("A client that never answers PING gets 10 PINGs at 50 ms, one at 1 Hz, then one-way", + "[remote_device_bridge][unit]") +{ + BridgeProcess bridge; + RawClient client(bridge.port()); + REQUIRE(accepted(client)); + REQUIRE_FALSE(client.wait_for(protocol::FrameType::Error, Ms(3000)).has_value()); + + const std::vector& pings = client.received(protocol::FrameType::Ping); + REQUIRE(pings.size() == 11); + for (size_t i = 1; i < 10; ++i) + { + INFO("gap " << i << ": " << (pings[i] - pings[i - 1]) / kMs << " ms"); + REQUIRE(pings[i] - pings[i - 1] > 30 * kMs); + REQUIRE(pings[i] - pings[i - 1] < 90 * kMs); + } + REQUIRE(pings[10] - pings[9] > 900 * kMs); + REQUIRE(pings[10] - pings[9] < 1100 * kMs); + REQUIRE(bridge.count("clock: no PONG, falling back to one-way") == 1); +} + +TEST_CASE("The 5 s status line carries age, RTT, offset and clock", "[remote_device_bridge][unit]") +{ + BridgeProcess bridge; + auto client = connect(bridge.port(), "status"); + auto left = client->create_hand_tracking_push_channel(XR_HAND_LEFT_EXT); + { + Streamer streaming(*left); + REQUIRE(bridge.wait_log(" 5s: ", 1, Ms(8000))); + } + const std::string log = bridge.log(); + const std::regex pattern( + "\\[bridge c1\\] 5s: rx hand=[0-9]+ schema=0 \\| applied=[0-9]+ coalesced=[0-9]+ " + "stale=[0-9]+ \\| tx wrist=[1-9][0-9]* \\| offset_ns=-?[0-9]+ \\| age_ms p50=-?[0-9.]+ " + "p99=-?[0-9.]+ max=-?[0-9.]+ \\| rtt_ms min=[0-9.]+ last=[0-9.]+ \\| " + "offset_ms=-?[0-9.]+ clock=two-way\n"); + std::smatch match; + REQUIRE(std::regex_search(log, match, pattern)); +} diff --git a/tests/cpp/plugins/remote_device_bridge_server/test_latency_stats.cpp b/tests/cpp/plugins/remote_device_bridge_server/test_latency_stats.cpp new file mode 100644 index 0000000000..581b74b7a0 --- /dev/null +++ b/tests/cpp/plugins/remote_device_bridge_server/test_latency_stats.cpp @@ -0,0 +1,93 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +#include +#include +#include +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace remote_device_bridge; +using remote_plugin_session::ClockOffsetEstimator; + +namespace +{ + +constexpr int64_t kMs = 1'000'000; + +std::string read_file(const std::string& path) +{ + std::ifstream in(path); + std::stringstream text; + text << in.rdbuf(); + return text.str(); +} + +} // namespace + +TEST_CASE("index_curl_m is the index-tip to wrist distance", "[remote_device_bridge][unit]") +{ + std::array joints{}; + joints[XR_HAND_JOINT_WRIST_EXT].pose.position = { 1.0f, 2.0f, 3.0f }; + joints[XR_HAND_JOINT_INDEX_TIP_EXT].pose.position = { 1.03f, 2.04f, 3.0f }; + joints[XR_HAND_JOINT_PALM_EXT].pose.position = { 9.0f, 9.0f, 9.0f }; // ignored + REQUIRE(index_curl_m(joints.data()) == Catch::Approx(0.05f).epsilon(1e-4)); +} + +TEST_CASE("format_clock_stats renders the status line suffix", "[remote_device_bridge][unit]") +{ + ClockOffsetEstimator clock; + std::vector ages; + REQUIRE(format_clock_stats(ages, clock) == + " | age_ms p50=- p99=- max=- | rtt_ms min=- last=- | offset_ms=0.000 clock=one-way"); + + // Offset 5 s, 70 ms each way, then a slower 151 ms exchange. + clock.add_two_way_sample(0, 70 * kMs - 5'000 * kMs, 71 * kMs - 5'000 * kMs, 141 * kMs); + clock.add_two_way_sample(1'000 * kMs, 1'080 * kMs - 5'000 * kMs, 1'081 * kMs - 5'000 * kMs, 1'152 * kMs); + for (int i = 1; i <= 100; ++i) + { + ages.push_back(i * kMs / 2); // 0.5 .. 50 ms + } + const std::string line = format_clock_stats(ages, clock); + INFO(line); + REQUIRE(line == + " | age_ms p50=25.0 p99=49.5 max=50.0 | rtt_ms min=140.0 last=151.0 | offset_ms=5000.000 clock=two-way"); +} + +TEST_CASE("SampleCsv writes the header once and forked clients append rows", "[remote_device_bridge][unit]") +{ + const SampleCsvRow row{ 1'000'000'123, 1'000'000'100, 42'000, 999'930'000, 70'123'456, "right", 17, 0.08765f }; + const std::string line = "1000000123,1000000100,42000,999930000,70.123,right,17,0.08765\n"; + + const std::string path = + (std::filesystem::temp_directory_path() / ("remote_device_bridge_csv_" + std::to_string(::getpid()) + ".csv")) + .string(); + { + const SampleCsv csv(path); + csv.write(row); + const pid_t pid = fork(); + if (pid == 0) + { + csv.write(row); // a forked client appends through the inherited descriptor + _exit(0); + } + REQUIRE(pid > 0); + int status = 0; + REQUIRE(waitpid(pid, &status, 0) == pid); + } + REQUIRE(read_file(path) == std::string(SampleCsv::kHeader) + line + line); + REQUIRE(std::string(SampleCsv::kHeader) == + "push_mono_ns,recv_mono_ns,capture_sender_ns,capture_mapped_ns,age_ms,side,seq,curl\n"); + std::remove(path.c_str()); + + REQUIRE_THROWS_AS(SampleCsv("/nonexistent_dir/x.csv"), std::runtime_error); +}