GameCult/Mimir
Watch
9
0
Fork
You've already forked Mimir
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Make Muninn OBS receiver reload-safe

Browse source
This commit is contained in:
Meta Vens 2026-07-16 23:10:56 +01:00
commit caf92d9dee
6 changed files with 135 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

12
native/obs_stem_source/CMakeLists.txt
View file

@ -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)

63
native/obs_stem_source/mimir_obs_muninn_source.cpp
View file

@ -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 <algorithm>
#include <array>
@ -171,7 +172,7 @@ struct RudpBridgeOutputs {
static std::optional<uint16_t> 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<const char *>(&reuse_addr),
sizeof(reuse_addr));
const int socket_buffer_bytes = 16 * 1024 * 1024;
setsockopt(socket_, SOL_SOCKET, SO_RCVBUF, reinterpret_cast<const char *>(&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<char *>(&actual_receive_buffer_bytes), &actual_receive_buffer_size);
const DWORD receive_timeout_ms = 2;
setsockopt(socket_, SOL_SOCKET, SO_RCVTIMEO, reinterpret_cast<const char *>(&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<const char *>(&socket_buffer_bytes),
sizeof(socket_buffer_bytes));
setsockopt(audio_socket, SOL_SOCKET, SO_SNDBUF, reinterpret_cast<const char *>(&socket_buffer_bytes),
@ -920,16 +925,26 @@ private:
std::vector<uint8_t> 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<char *>(buffer.data()), static_cast<int>(buffer.size()), 0,
reinterpret_cast<sockaddr *>(&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<size_t>(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<unsigned int>(read_be16(packet + 24)),
static_cast<unsigned int>(read_be16(packet + 26)),
static_cast<unsigned int>(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<const char *>(packet + 36),
reinterpret_cast<const char *>(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<unsigned int>(media->kind), media->payload.size(),
static_cast<unsigned long long>(media->frame_id),
static_cast<unsigned int>(media->chunk_index),
static_cast<unsigned int>(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<uint8_t> &payload, uint32_t sequence, const sockaddr_in &remote)
{
const std::string channel = "media";
const std::string &channel = parts_.channel_id;
std::vector<uint8_t> 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<uint64_t> 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<uint32_t>(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();

5
native/obs_stem_source/muninn_rudp_url.h
View file

@ -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<RudpUrlParts> 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;
}

32
native/obs_stem_source/muninn_udp_socket.h Normal file
View file

@ -0,0 +1,32 @@
#pragma once
#ifdef _WIN32
#include <winsock2.h>
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<const char *>(&exclusive), sizeof(exclusive)) != 0) {
closesocket(socket);
return INVALID_SOCKET;
}
}
return socket;
}
} // namespace muninn_udp_socket
#endif

4
native/obs_stem_source/tests/muninn_rudp_url_tests.cpp
View file

@ -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,

34
native/obs_stem_source/tests/muninn_udp_socket_tests.cpp Normal file
View file

@ -0,0 +1,34 @@
#include "muninn_udp_socket.h"
#include <cassert>
#include <windows.h>
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<HANDLE>(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<sockaddr *>(&address), sizeof(address)) == 0);
int address_size = sizeof(address);
assert(getsockname(owner, reinterpret_cast<sockaddr *>(&address), &address_size) == 0);
const SOCKET contender = muninn_udp_socket::create(true);
assert(contender != INVALID_SOCKET);
assert(bind(contender, reinterpret_cast<sockaddr *>(&address), sizeof(address)) != 0);
assert(WSAGetLastError() == WSAEADDRINUSE || WSAGetLastError() == WSAEACCES);
closesocket(contender);
closesocket(owner);
WSACleanup();
return 0;
}