diff --git a/obs-studio-server/source/osn-scene.cpp b/obs-studio-server/source/osn-scene.cpp index 5058fd0d4..86b515906 100644 --- a/obs-studio-server/source/osn-scene.cpp +++ b/obs-studio-server/source/osn-scene.cpp @@ -96,13 +96,16 @@ void osn::Scene::CreatePrivate(void *data, const int64_t id, const std::vector &args, std::vector &rval) { - // Atomically find and acquire a strong reference under the manager lock, - // preventing the source from being destroyed between find() and obs_source_get_ref(). + // Promote the manager's retained weak reference while its registration is + // locked, so the source remains alive for this call or is rejected as expired. OBSSourceAutoRelease src = osn::Source::Manager::GetInstance().findAndRef(args[0].value_union.ui64); if (!src) { PRETTY_ERROR_RETURN(ErrorCode::InvalidReference, "Source reference is not valid."); diff --git a/obs-studio-server/source/osn-source.hpp b/obs-studio-server/source/osn-source.hpp index 0f2b4f99c..39b1820f1 100644 --- a/obs-studio-server/source/osn-source.hpp +++ b/obs-studio-server/source/osn-source.hpp @@ -20,6 +20,7 @@ #include #include #include +#include #include "utility.hpp" #undef strtoll #include "nlohmann/json.hpp" @@ -32,7 +33,10 @@ class Source { protected: Manager() {} - ~Manager() {} + ~Manager() { clear(); } + + private: + std::map weak_sources; public: Manager(Manager const &) = delete; @@ -41,16 +45,73 @@ class Source { public: static Manager &GetInstance(); - // Atomically finds the source and acquires a strong reference under the - // manager lock, preventing destruction between find() and obs_source_get_ref(). - // Returns null (as OBSSourceAutoRelease) if not found or already destroyed. + utility::unique_id::id_t allocate(obs_source_t *source) + { + std::lock_guard lock(internal_mutex); + OBSWeakSourceAutoRelease weakSource(obs_source_get_weak_source(source)); + + // A destroyed source's address can be reused while a stale manager + // entry still retains its expired weak control block. Only deduplicate + // when both the source address and weak control block identify the same + // live source; otherwise discard the stale registration. + for (auto iter = object_map.begin(); iter != object_map.end();) { + if (iter->second != source) { + ++iter; + continue; + } + + const auto weakIter = weak_sources.find(iter->first); + if (weakIter != weak_sources.end() && weakIter->second.Get() == weakSource.Get()) + return iter->first; + + weak_sources.erase(iter->first); + iter = object_map.erase(iter); + } + + const auto uid = utility::unique_object_manager::allocate(source); + if (uid != std::numeric_limits::max()) { + try { + weak_sources.emplace(uid, std::move(weakSource)); + } catch (...) { + utility::unique_object_manager::free(uid); + throw; + } + } + return uid; + } + + utility::unique_id::id_t free(obs_source_t *source) + { + std::lock_guard lock(internal_mutex); + const auto uid = utility::unique_object_manager::free(source); + weak_sources.erase(uid); + return uid; + } + + obs_source_t *free(utility::unique_id::id_t uid) + { + std::lock_guard lock(internal_mutex); + obs_source_t *source = utility::unique_object_manager::free(uid); + weak_sources.erase(uid); + return source; + } + + void clear() + { + std::lock_guard lock(internal_mutex); + object_map.clear(); + weak_sources.clear(); + } + + // Promote the retained weak reference instead of deriving a reference from + // the raw source pointer, whose control block may already be destroyed. OBSSourceAutoRelease findAndRef(utility::unique_id::id_t id) { std::lock_guard lock(internal_mutex); - auto iter = object_map.find(id); - if (iter == object_map.end()) + auto iter = weak_sources.find(id); + if (iter == weak_sources.end()) return nullptr; - return obs_source_get_ref(iter->second); + return obs_weak_source_get_source(iter->second.Get()); } }; diff --git a/obs-studio-server/tests/test-osn-source.cpp b/obs-studio-server/tests/test-osn-source.cpp index f0d01630f..4823e3b04 100644 --- a/obs-studio-server/tests/test-osn-source.cpp +++ b/obs-studio-server/tests/test-osn-source.cpp @@ -6,43 +6,130 @@ #include "osn-source.hpp" #include #include "shared.hpp" -#include +#include #include #include +#include #include -#include "obs-setup.hpp" -#include -#include #include -// Since we do not use C++ 20 (std::jthread), defining a scoped thread. -struct joining_thread { - std::thread t; - explicit joining_thread(std::thread t_) : t(std::move(t_)) {} - joining_thread(joining_thread &&) = default; - joining_thread &operator=(joining_thread &&) = default; - joining_thread(const joining_thread &) = delete; - joining_thread &operator=(const joining_thread &) = delete; - ~joining_thread() +namespace { + +// Keep the test source independent of OBS plugins so its behavior depends only +// on the source manager and libobs reference lifecycle. +constexpr char TEST_SOURCE_ID[] = "source_manager_lifetime_test_source"; + +const char *testSourceGetName(void *) +{ + return "Source Manager Lifetime Test Source"; +} + +void *testSourceCreate(obs_data_t *, obs_source_t *source) +{ + return source; +} + +void testSourceDestroy(void *) {} + +obs_properties_t *testSourceGetProperties(void *) +{ + obs_properties_t *properties = obs_properties_create(); + obs_properties_add_bool(properties, "enabled", "Enabled"); + return properties; +} + +obs_source_info makeTestSourceInfo() +{ + obs_source_info info{}; + info.id = TEST_SOURCE_ID; + info.type = OBS_SOURCE_TYPE_INPUT; + info.get_name = testSourceGetName; + info.create = testSourceCreate; + info.destroy = testSourceDestroy; + info.get_properties = testSourceGetProperties; + return info; +} + +// These tests only need the OBS core and its deferred-destruction queue. +class ObsCoreSetup { +public: + ObsCoreSetup() { REQUIRE(obs_startup("en-US", nullptr, nullptr)); } + ~ObsCoreSetup() { - if (t.joinable()) - t.join(); + // Do not let a queued source destruction outlive the OBS core. + obs_wait_for_destroy_queue(); + obs_shutdown(); } + + ObsCoreSetup(const ObsCoreSetup &) = delete; + ObsCoreSetup &operator=(const ObsCoreSetup &) = delete; }; -static bool wait_for_source_manager_size(std::size_t expectedSize) -{ - for (int i = 0; i < 100; i++) { - obs_wait_for_destroy_queue(); +// Source destruction runs on OBS_TASK_DESTROY. Blocking that queue creates the +// precise window where the last strong reference is gone but the manager's +// destroy callback has not removed the source registration yet. +class DestroyQueueGate { +public: + DestroyQueueGate() + { + obs_queue_task(OBS_TASK_DESTROY, waitForRelease, this, false); + // Wait until the gate is running so later destruction is guaranteed to + // queue behind it. + std::unique_lock lock(mutex); + condition.wait(lock, [this] { return entered; }); + } - if (osn::Source::Manager::GetInstance().size() == expectedSize) - return true; + ~DestroyQueueGate() { releaseAndWait(); } - std::this_thread::sleep_for(std::chrono::milliseconds(10)); + void releaseAndWait() + { + { + std::lock_guard lock(mutex); + released = true; + } + condition.notify_all(); + + if (!drained) { + // Draining also runs the source's destroy signal, which unregisters + // it from Source::Manager. + obs_wait_for_destroy_queue(); + drained = true; + } } - return false; -} + DestroyQueueGate(const DestroyQueueGate &) = delete; + DestroyQueueGate &operator=(const DestroyQueueGate &) = delete; + +private: + static void waitForRelease(void *data) + { + auto *gate = static_cast(data); + std::unique_lock lock(gate->mutex); + gate->entered = true; + gate->condition.notify_all(); + gate->condition.wait(lock, [gate] { return gate->released; }); + } + + std::mutex mutex; + std::condition_variable condition; + bool entered = false; + bool released = false; + bool drained = false; +}; + +// The production manager deliberately hides its storage. This test-only +// subclass exposes one narrow operation so a stale raw-pointer entry can be +// made to look as if its address were reused by a different OBS source. +class AddressReuseSourceManager : public osn::Source::Manager { +public: + void simulateAddressReuse(uint64_t sourceId, obs_source_t *replacement) + { + std::lock_guard lock(internal_mutex); + object_map.at(sourceId) = replacement; + } +}; + +} // namespace TEST_CASE("Scene AddSource rejects malformed argument counts") { @@ -60,59 +147,164 @@ TEST_CASE("Scene AddSource rejects malformed argument counts") } } -TEST_CASE("Run osn::source tests") +TEST_CASE("Private scene registrations are removed after release") +{ + ObsCoreSetup setupOBS; + auto &manager = osn::Source::Manager::GetInstance(); + const auto sourceCount = manager.size(); + + auto createPrivateScene = [](const char *name) { + std::vector response; + osn::Scene::CreatePrivate(nullptr, 0, {ipc::value(name)}, response); + + REQUIRE(response.size() >= 2); + REQUIRE((ErrorCode)response[0].value_union.ui64 == ErrorCode::Ok); + return response[1].value_union.ui64; + }; + auto checkProperties = [](uint64_t sourceId) { + std::vector response; + osn::Source::GetProperties(nullptr, 0, {ipc::value(sourceId)}, response); + + REQUIRE(!response.empty()); + CHECK((ErrorCode)response[0].value_union.ui64 == ErrorCode::Ok); + }; + auto releasePrivateScene = [](uint64_t sourceId) { + std::vector response; + osn::Scene::Release(nullptr, 0, {ipc::value(sourceId)}, response); + + REQUIRE(!response.empty()); + CHECK((ErrorCode)response[0].value_union.ui64 == ErrorCode::Ok); + // Source destruction may run on OBS_TASK_DESTROY. Wait for its destroy + // signal to remove the manager registration before inspecting the map. + obs_wait_for_destroy_queue(); + }; + + // Exercise the same handlers used by SceneFactory.createPrivate and the + // scene properties accessor. Private sources skip OBS's global source-create + // signal, so CreatePrivate itself must attach the destroy callback. + const uint64_t firstSceneId = createPrivateScene("first private scene"); + CHECK(manager.size() == sourceCount + 1); + checkProperties(firstSceneId); + releasePrivateScene(firstSceneId); + CHECK(manager.size() == sourceCount); + + // Recreating the scene verifies that no stale registration can capture the + // new source if OBS's allocator gives it a recently released address. + const uint64_t secondSceneId = createPrivateScene("second private scene"); + CHECK(manager.size() == sourceCount + 1); + checkProperties(secondSceneId); + releasePrivateScene(secondSceneId); + CHECK(manager.size() == sourceCount); +} + +TEST_CASE("Source manager rejects an expired identity after address reuse") { - osn::tests::ObsSetup setupOBS; + ObsCoreSetup setupOBS; + static const obs_source_info testSourceInfo = makeTestSourceInfo(); + obs_register_source(&testSourceInfo); + AddressReuseSourceManager manager; + + OBSSourceAutoRelease firstSource = obs_source_create_private(TEST_SOURCE_ID, "first source", nullptr); + REQUIRE(firstSource != nullptr); + const uint64_t firstSourceId = manager.allocate(firstSource); + REQUIRE(firstSourceId != UINT64_MAX); + + // Leave the raw pointer and weak reference registered while destroying the + // source, matching the stale state that existed without a destroy callback. + firstSource = nullptr; + obs_wait_for_destroy_queue(); + CHECK(!manager.findAndRef(firstSourceId)); + + OBSSourceAutoRelease secondSource = obs_source_create_private(TEST_SOURCE_ID, "second source", nullptr); + REQUIRE(secondSource != nullptr); + // ASan and different allocators do not reliably reuse an address on demand. + // Rewrite only the stored raw pointer to model that reuse deterministically; + // the entry's weak control block still belongs to the expired first source. + manager.simulateAddressReuse(firstSourceId, secondSource); + + const uint64_t secondSourceId = manager.allocate(secondSource); + REQUIRE(secondSourceId != UINT64_MAX); + CHECK(secondSourceId != firstSourceId); + CHECK(manager.size() == 1); + + // allocate() must replace the stale identity with the second source's weak + // reference, so production lookups retain and return the live source. + OBSSourceAutoRelease retainedSource = manager.findAndRef(secondSourceId); + REQUIRE(retainedSource != nullptr); + CHECK(retainedSource.Get() == secondSource.Get()); +} + +TEST_CASE("Source manager safely promotes references during deferred destruction") +{ + ObsCoreSetup setupOBS; + static const obs_source_info testSourceInfo = makeTestSourceInfo(); + obs_register_source(&testSourceInfo); + auto &manager = osn::Source::Manager::GetInstance(); + const auto sourceCount = manager.size(); - SECTION("Get properties of browser source while releasing concurrently does not crash") { - auto sourceCount = osn::Source::Manager::GetInstance().size(); - const int iterations = 20; - std::vector workers; - std::vector releaseOk(iterations, 0); - std::vector getPropertiesCode(iterations, ErrorCode::Error); - - for (int i = 0; i < iterations; i++) { - const std::string sourceName = "test-input-" + std::to_string(i); - std::vector args = {ipc::value("browser_source"), ipc::value(sourceName)}; - std::vector response; - - osn::Input::Create(nullptr, 0, args, response); - REQUIRE(response.size() >= 2); - ErrorCode error = (ErrorCode)response[0].value_union.ui64; - REQUIRE(error == ErrorCode::Ok); - - uint64_t sourceId = response[1].value_union.ui64; - - workers.push_back(joining_thread(std::thread([sourceId, i, &getPropertiesCode]() { - std::vector propArgs = {ipc::value(sourceId)}; - std::vector propResponse; - osn::Source::GetProperties(nullptr, 0, propArgs, propResponse); - if (propResponse.size() >= 1) { - getPropertiesCode[i] = (ErrorCode)propResponse[0].value_union.ui64; - } - }))); - - workers.push_back(joining_thread(std::thread([sourceId, i, &releaseOk]() { - std::vector propArgs = {ipc::value(sourceId)}; - std::vector propResponse; - osn::Source::Release(nullptr, 0, propArgs, propResponse); - // Capture result for checking on the main thread after join. - if (propResponse.size() >= 1) { - releaseOk[i] = ((ErrorCode)propResponse[0].value_union.ui64 == ErrorCode::Ok); - } - }))); - } + INFO("The final source release wins the race"); + DestroyQueueGate destroyQueue; + OBSSourceAutoRelease source = obs_source_create_private(TEST_SOURCE_ID, "released source", nullptr); + REQUIRE(source != nullptr); - workers.clear(); - // Check release results on the main thread where Catch2 is safe to use. - for (int i = 0; i < iterations; i++) { - CHECK(releaseOk[i]); - // ErrorCode::InvalidReference is possible if the source was deleted before we could acquire the source - bool expectedErrorCode = getPropertiesCode[i] == ErrorCode::Ok || getPropertiesCode[i] == ErrorCode::InvalidReference; - CHECK(expectedErrorCode); - } + // Private sources do not emit the global source-create signal, so mirror + // production registration here. Repeated allocation must keep one ID and + // one retained weak reference for the source. + const uint64_t sourceId = manager.allocate(source); + osn::Source::attach_source_signals(source); + REQUIRE(sourceId != UINT64_MAX); + CHECK(manager.allocate(source) == sourceId); + CHECK(manager.size() == sourceCount + 1); + + // The gate keeps deferred destruction from unregistering the source. + // Dropping the only strong reference therefore leaves an expired source + // ID in the manager, reproducing the original race window. + source = nullptr; + + CHECK(manager.size() == sourceCount + 1); + std::vector args = {ipc::value(sourceId)}; + std::vector response; + // Promotion of the retained weak reference must fail cleanly instead of + // dereferencing the stale raw pointer stored by the old implementation. + osn::Source::GetProperties(nullptr, 0, args, response); + REQUIRE(!response.empty()); + CHECK((ErrorCode)response[0].value_union.ui64 == ErrorCode::InvalidReference); + + // Let OBS finish destruction and deliver the manager's destroy callback. + destroyQueue.releaseAndWait(); + CHECK(manager.size() == sourceCount); + } + + { + INFO("The source lookup wins the race"); + DestroyQueueGate destroyQueue; + OBSSourceAutoRelease source = obs_source_create_private(TEST_SOURCE_ID, "retained source", nullptr); + REQUIRE(source != nullptr); + + const uint64_t sourceId = manager.allocate(source); + osn::Source::attach_source_signals(source); + REQUIRE(sourceId != UINT64_MAX); + // This time lookup wins: promotion happens while the original strong + // reference is still alive and must keep the source usable. + OBSSourceAutoRelease retainedSource = manager.findAndRef(sourceId); + REQUIRE(retainedSource != nullptr); + + source = nullptr; + CHECK(std::string(obs_source_get_name(retainedSource)) == "retained source"); + OBSSourceAutoRelease secondReference = manager.findAndRef(sourceId); + CHECK(secondReference != nullptr); + + // Releasing every promoted reference expires the source. Its registration + // remains until the blocked destroy callback runs, but another promotion + // must already report that the source is gone. + secondReference = nullptr; + retainedSource = nullptr; + CHECK(manager.size() == sourceCount + 1); + CHECK(!manager.findAndRef(sourceId)); - CHECK(wait_for_source_manager_size(sourceCount)); // Check to see if all objects released. + // Draining the queue completes destruction and removes the registration. + destroyQueue.releaseAndWait(); + CHECK(manager.size() == sourceCount); } }