From caf92d9dee20771eb07ccefa629bb4ca4c4f3f12 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 16 Jul 2026 23:10:56 +0100 Subject: [PATCH] Make Muninn OBS receiver reload-safe --- native/obs_stem_source/CMakeLists.txt | 12 ++++ .../mimir_obs_muninn_source.cpp | 63 +++++++++++++++---- native/obs_stem_source/muninn_rudp_url.h | 5 +- native/obs_stem_source/muninn_udp_socket.h | 32 ++++++++++ .../tests/muninn_rudp_url_tests.cpp | 4 +- .../tests/muninn_udp_socket_tests.cpp | 34 ++++++++++ 6 files changed, 135 insertions(+), 15 deletions(-) create mode 100644 native/obs_stem_source/muninn_udp_socket.h create mode 100644 native/obs_stem_source/tests/muninn_udp_socket_tests.cpp diff --git a/native/obs_stem_source/CMakeLists.txt b/native/obs_stem_source/CMakeLists.txt index 090a2ff..527739d 100644 --- a/native/obs_stem_source/CMakeLists.txt +++ b/native/obs_stem_source/CMakeLists.txt @@ -45,3 +45,15 @@ target_compile_features(muninn_audio_fec_tests PRIVATE cxx_std_20) target_include_directories(muninn_audio_fec_tests PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}) add_test(NAME muninn_audio_fec_tests COMMAND muninn_audio_fec_tests) + +add_executable(muninn_udp_socket_tests + tests/muninn_udp_socket_tests.cpp) +target_compile_features(muninn_udp_socket_tests PRIVATE cxx_std_20) +target_compile_definitions(muninn_udp_socket_tests PRIVATE + WIN32_LEAN_AND_MEAN + NOMINMAX + _WIN32_WINNT=0x0A00) +target_include_directories(muninn_udp_socket_tests PRIVATE + ${CMAKE_CURRENT_SOURCE_DIR}) +target_link_libraries(muninn_udp_socket_tests PRIVATE ws2_32) +add_test(NAME muninn_udp_socket_tests COMMAND muninn_udp_socket_tests) diff --git a/native/obs_stem_source/mimir_obs_muninn_source.cpp b/native/obs_stem_source/mimir_obs_muninn_source.cpp index 3600db3..9910999 100644 --- a/native/obs_stem_source/mimir_obs_muninn_source.cpp +++ b/native/obs_stem_source/mimir_obs_muninn_source.cpp @@ -12,6 +12,7 @@ #include "muninn_rudp_ack.h" #include "muninn_rudp_fragments.h" #include "muninn_rudp_url.h" +#include "muninn_udp_socket.h" #include #include @@ -171,7 +172,7 @@ struct RudpBridgeOutputs { static std::optional allocate_udp_loopback_port() { #ifdef _WIN32 - SOCKET socket = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + SOCKET socket = muninn_udp_socket::create(); if (socket == INVALID_SOCKET) { return std::nullopt; } @@ -870,22 +871,24 @@ private: void run() { #ifdef _WIN32 + SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_HIGHEST); WSADATA data; WSAStartup(MAKEWORD(2, 2), &data); - socket_ = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + socket_ = muninn_udp_socket::create(true); if (socket_ == INVALID_SOCKET) { blog(LOG_WARNING, "Muninn RUDP bridge could not create socket"); running_ = false; WSACleanup(); return; } - const int socket_buffer_bytes = 4 * 1024 * 1024; - const BOOL reuse_addr = TRUE; - setsockopt(socket_, SOL_SOCKET, SO_REUSEADDR, reinterpret_cast(&reuse_addr), - sizeof(reuse_addr)); + const int socket_buffer_bytes = 16 * 1024 * 1024; setsockopt(socket_, SOL_SOCKET, SO_RCVBUF, reinterpret_cast(&socket_buffer_bytes), sizeof(socket_buffer_bytes)); + int actual_receive_buffer_bytes = 0; + int actual_receive_buffer_size = sizeof(actual_receive_buffer_bytes); + getsockopt(socket_, SOL_SOCKET, SO_RCVBUF, + reinterpret_cast(&actual_receive_buffer_bytes), &actual_receive_buffer_size); const DWORD receive_timeout_ms = 2; setsockopt(socket_, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast(&receive_timeout_ms), sizeof(receive_timeout_ms)); @@ -902,9 +905,11 @@ private: WSACleanup(); return; } + blog(LOG_INFO, "Muninn RUDP bridge receive owner priority=highest socket_buffer_bytes=%d", + actual_receive_buffer_bytes); - const SOCKET video_socket = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); - const SOCKET audio_socket = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + const SOCKET video_socket = muninn_udp_socket::create(); + const SOCKET audio_socket = muninn_udp_socket::create(); setsockopt(video_socket, SOL_SOCKET, SO_SNDBUF, reinterpret_cast(&socket_buffer_bytes), sizeof(socket_buffer_bytes)); setsockopt(audio_socket, SOL_SOCKET, SO_SNDBUF, reinterpret_cast(&socket_buffer_bytes), @@ -920,16 +925,26 @@ private: std::vector buffer(65535); uint32_t bridge_sequence = 1; + bool logged_receive_loop = false; + bool logged_datagram = false; while (running_) { sockaddr_in remote = {}; int remote_len = sizeof(remote); const int received = recvfrom(socket_, reinterpret_cast(buffer.data()), static_cast(buffer.size()), 0, reinterpret_cast(&remote), &remote_len); if (received <= 0) { + if (!logged_receive_loop) { + logged_receive_loop = true; + blog(LOG_INFO, "Muninn RUDP bridge receive loop active udp/%u error=%d", parts_.port, WSAGetLastError()); + } expire_rudp_packet_fragments(std::chrono::steady_clock::now()); expire_video_assemblies(std::chrono::steady_clock::now(), bridge_sequence); continue; } + if (!logged_datagram) { + logged_datagram = true; + blog(LOG_INFO, "Muninn RUDP bridge received first datagram bytes=%d", received); + } handle_packet(buffer.data(), static_cast(received), remote, bridge_sequence, video_socket, video_addr, audio_socket, audio_addr); } @@ -1033,6 +1048,16 @@ private: } if (packet_type == 3) { + auto &logged_first_data = is_audio_connection ? logged_first_audio_data_ : logged_first_video_data_; + if (!logged_first_data) { + logged_first_data = true; + blog(LOG_INFO, + "Muninn RUDP bridge first %s data sequence=%u payload=%u fragment_id=%u fragment_index=%u fragment_count=%u datagram=%zu", + is_audio_connection ? "audio" : "video", sequence, payload_len, + static_cast(read_be16(packet + 24)), + static_cast(read_be16(packet + 26)), + static_cast(fragment_count), size); + } if (!media_connected_) { video_assemblies_.clear(); video_parity_cache_.clear(); @@ -1076,7 +1101,7 @@ private: const std::string channel_id( reinterpret_cast(packet + 36), reinterpret_cast(packet + 36 + channel_len)); - if (channel_id != "media") { + if (channel_id != parts_.channel_id) { const auto [ack, ack_mask] = connection_ack_tracker.state(); send_control_packet(4, 0x00, connection_id, bridge_sequence++, ack, ack_mask, remote); return; @@ -1157,6 +1182,15 @@ private: log_bridge_pressure("dropped undecodable media payload"); return; } + if (!logged_first_decoded_media_) { + logged_first_decoded_media_ = true; + blog(LOG_INFO, + "Muninn RUDP bridge decoded first media kind=%u payload=%zu frame=%llu chunk=%u/%u", + static_cast(media->kind), media->payload.size(), + static_cast(media->frame_id), + static_cast(media->chunk_index), + static_cast(media->chunk_count)); + } size_t forwarded_bytes = 0; if (media->kind == MuninnMediaPayload::Kind::Video || media->kind == MuninnMediaPayload::Kind::VideoParity) { const auto assembled = assemble_video_payload(*media, bridge_sequence); @@ -1978,7 +2012,7 @@ private: void send_media_packet(const std::vector &payload, uint32_t sequence, const sockaddr_in &remote) { - const std::string channel = "media"; + const std::string &channel = parts_.channel_id; std::vector response(36 + channel.size() + payload.size()); response[0] = 'C'; response[1] = 'N'; @@ -2060,6 +2094,9 @@ private: uint64_t last_video_expiry_scan_ms_ = 0; bool waiting_for_video_keyframe_ = true; bool media_connected_ = false; + bool logged_first_video_data_ = false; + bool logged_first_audio_data_ = false; + bool logged_first_decoded_media_ = false; std::atomic last_keyframe_request_ms_ = 0; std::string source_url_; RudpBridgeOutputs outputs_; @@ -2686,7 +2723,7 @@ static bool publish_capture_command_rudp(const std::string &target, const std::v if (WSAStartup(MAKEWORD(2, 2), &wsa) != 0) { return false; } - SOCKET socket = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + SOCKET socket = muninn_udp_socket::create(); if (socket == INVALID_SOCKET) { WSACleanup(); return false; @@ -2742,7 +2779,7 @@ static std::string expected_rudp_stream_url(const MuninnSource *ctx, const std:: { const auto reliable_expire_after_ms = std::max(1, ctx->rudp_latency_budget_ms); return "rudp://" + ctx->media_target_host + ":" + std::to_string(ctx->media_port) + "/" + stream_id + - "?channel=media&format=muninn-typed-media&connection=0x6d750001&audio_connection=0x6d750004&profile=muninn.rudp.low_latency_h264_lan.v1&sender_resend_delay_ms=5&reliable_expire_after_ms=" + + "?channel=realtime&format=muninn-typed-media&connection=0x6d750001&audio_connection=0x6d750004&profile=muninn.rudp.low_latency_h264_lan.v1&sender_resend_delay_ms=5&reliable_expire_after_ms=" + std::to_string(reliable_expire_after_ms) + "&assembly_deadline_ms=" + std::to_string(ctx->rudp_latency_budget_ms) + "&gap_wait_ms=16&audio_reorder_ms=" + std::to_string(ctx->rudp_audio_reorder_ms); } @@ -3378,7 +3415,7 @@ private: return; } #endif - socket_ = ::socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP); + socket_ = muninn_udp_socket::create(true); if (socket_ == INVALID_SOCKET) { #ifdef _WIN32 WSACleanup(); diff --git a/native/obs_stem_source/muninn_rudp_url.h b/native/obs_stem_source/muninn_rudp_url.h index ffbfbd2..c6275b9 100644 --- a/native/obs_stem_source/muninn_rudp_url.h +++ b/native/obs_stem_source/muninn_rudp_url.h @@ -18,6 +18,7 @@ inline constexpr uint32_t DefaultSenderResendDelayMs = 5; inline constexpr uint32_t MaxLatencyBudgetMs = 500; struct RudpUrlParts { + std::string channel_id = "realtime"; uint16_t port = 0; uint16_t video_local_port = 0; uint16_t audio_local_port = 0; @@ -81,7 +82,9 @@ inline std::optional parse(const std::string &url) } const auto key = item.substr(0, equals); const auto value = item.substr(equals + 1); - if (key == "connection") { + if (key == "channel" && !value.empty()) { + parts.channel_id = value; + } else if (key == "connection") { if (const auto parsed = parse_u32_hex_or_decimal(value)) { parts.connection_id = *parsed; } diff --git a/native/obs_stem_source/muninn_udp_socket.h b/native/obs_stem_source/muninn_udp_socket.h new file mode 100644 index 0000000..b49f767 --- /dev/null +++ b/native/obs_stem_source/muninn_udp_socket.h @@ -0,0 +1,32 @@ +#pragma once + +#ifdef _WIN32 +#include + +namespace muninn_udp_socket { + +inline SOCKET create(bool exclusive_listener = false) +{ + const SOCKET socket = WSASocketW( + AF_INET, + SOCK_DGRAM, + IPPROTO_UDP, + nullptr, + 0, + WSA_FLAG_NO_HANDLE_INHERIT); + if (socket == INVALID_SOCKET) { + return INVALID_SOCKET; + } + if (exclusive_listener) { + const BOOL exclusive = TRUE; + if (setsockopt(socket, SOL_SOCKET, SO_EXCLUSIVEADDRUSE, + reinterpret_cast(&exclusive), sizeof(exclusive)) != 0) { + closesocket(socket); + return INVALID_SOCKET; + } + } + return socket; +} + +} // namespace muninn_udp_socket +#endif diff --git a/native/obs_stem_source/tests/muninn_rudp_url_tests.cpp b/native/obs_stem_source/tests/muninn_rudp_url_tests.cpp index 2cdebd0..1fba928 100644 --- a/native/obs_stem_source/tests/muninn_rudp_url_tests.cpp +++ b/native/obs_stem_source/tests/muninn_rudp_url_tests.cpp @@ -19,9 +19,10 @@ void require(bool condition, const std::string &message) int main() { const auto configured = muninn_rudp_url::parse( - "rudp://192.168.1.66:5204/muninn.raven.av.rudp?channel=media&format=muninn-typed-media&connection=0x6d750001&audio_connection=0x6d750004&sender_resend_delay_ms=5&reliable_expire_after_ms=120&assembly_deadline_ms=120&gap_wait_ms=16&audio_reorder_ms=80"); + "rudp://192.168.1.66:5204/muninn.raven.av.rudp?channel=realtime&format=muninn-typed-media&connection=0x6d750001&audio_connection=0x6d750004&sender_resend_delay_ms=5&reliable_expire_after_ms=120&assembly_deadline_ms=120&gap_wait_ms=16&audio_reorder_ms=80"); require(configured.has_value(), "configured RUDP URL did not parse"); require(configured->port == 5204, "configured RUDP URL port was not parsed"); + require(configured->channel_id == "realtime", "configured RUDP realtime channel was not parsed"); require(configured->video_local_port == 15204, "configured RUDP video local port was not derived"); require(configured->audio_local_port == 15205, "configured RUDP audio local port was not derived"); require(configured->connection_id == 0x6d750001, "configured RUDP connection id was not parsed"); @@ -41,6 +42,7 @@ int main() const auto fallback = muninn_rudp_url::parse("rudp://192.168.1.66:5204/muninn.raven.av.rudp"); require(fallback.has_value(), "minimal RUDP URL did not parse"); + require(fallback->channel_id == "realtime", "minimal RUDP URL did not use the experimental realtime lane"); require(fallback->connection_id == muninn_rudp_url::DefaultMediaRudpConnectionId, "minimal RUDP URL did not keep the default connection id"); require(fallback->audio_connection_id == muninn_rudp_url::DefaultAudioRudpConnectionId, diff --git a/native/obs_stem_source/tests/muninn_udp_socket_tests.cpp b/native/obs_stem_source/tests/muninn_udp_socket_tests.cpp new file mode 100644 index 0000000..db2c15b --- /dev/null +++ b/native/obs_stem_source/tests/muninn_udp_socket_tests.cpp @@ -0,0 +1,34 @@ +#include "muninn_udp_socket.h" + +#include +#include + +int main() +{ + WSADATA data{}; + assert(WSAStartup(MAKEWORD(2, 2), &data) == 0); + + const SOCKET owner = muninn_udp_socket::create(true); + assert(owner != INVALID_SOCKET); + DWORD flags = HANDLE_FLAG_INHERIT; + assert(GetHandleInformation(reinterpret_cast(owner), &flags)); + assert((flags & HANDLE_FLAG_INHERIT) == 0); + + sockaddr_in address{}; + address.sin_family = AF_INET; + address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); + address.sin_port = 0; + assert(bind(owner, reinterpret_cast(&address), sizeof(address)) == 0); + + int address_size = sizeof(address); + assert(getsockname(owner, reinterpret_cast(&address), &address_size) == 0); + const SOCKET contender = muninn_udp_socket::create(true); + assert(contender != INVALID_SOCKET); + assert(bind(contender, reinterpret_cast(&address), sizeof(address)) != 0); + assert(WSAGetLastError() == WSAEADDRINUSE || WSAGetLastError() == WSAEACCES); + + closesocket(contender); + closesocket(owner); + WSACleanup(); + return 0; +}