diff --git a/modules/Language/XML/FlowRunner.mpp b/modules/Language/XML/FlowRunner.mpp index 55174666..b0c8388a 100644 --- a/modules/Language/XML/FlowRunner.mpp +++ b/modules/Language/XML/FlowRunner.mpp @@ -6,6 +6,7 @@ import CppUtils.String; import CppUtils.Container.Tree; import CppUtils.FileSystem; import CppUtils.Thread.ThreadPool; +import CppUtils.Thread.SharedLocker; import CppUtils.Logger; import CppUtils.Chrono.Chronometer; import CppUtils.Container.MultiKeyMap; @@ -100,9 +101,10 @@ export namespace CppUtils::Language::Xml public: auto loadFlows() -> void { - for (const auto& entry : m_flows) + auto flowsAccessor = m_flows.uniqueAccess(); + for (const auto& entry : flowsAccessor.value()) entry.value->stopWatching(); - m_flows.clear(); + flowsAccessor.value().clear(); for (const auto& filePath : discoverFlowConfigs()) try @@ -111,7 +113,7 @@ export namespace CppUtils::Language::Xml const auto& flowNode = findFlowNode(rootNode); const auto flowName = extractFlowName(flowNode, filePath); const auto& factory = resolveFactory(flowNode, filePath); - m_flows.emplace(filePath, flowName, std::shared_ptr{factory(flowNode)}); + flowsAccessor.value().emplace(filePath, flowName, std::shared_ptr{factory(flowNode)}); } catch (const std::exception& exception) { @@ -120,21 +122,23 @@ export namespace CppUtils::Language::Xml } } - [[nodiscard]] auto flows() const -> const FlowContainer& + [[nodiscard]] auto flows() const -> decltype(auto) { - return m_flows; + return m_flows.sharedAccess(); } [[nodiscard]] auto flow(const Type::Exact auto& filePath) const -> std::shared_ptr { - const auto iterator = m_flows.find(filePath); - return iterator != std::ranges::end(m_flows) ? iterator->value : nullptr; + auto flowsAccessor = m_flows.sharedAccess(); + const auto iterator = flowsAccessor.value().find(filePath); + return iterator != std::ranges::end(flowsAccessor.value()) ? iterator->value : nullptr; } [[nodiscard]] auto flow(std::string_view flowName) const -> std::shared_ptr { - const auto iterator = m_flows.template findAt<1>(flowName); - return iterator != std::ranges::end(m_flows) ? iterator->value : nullptr; + auto flowsAccessor = m_flows.sharedAccess(); + const auto iterator = flowsAccessor.value().template findAt<1>(flowName); + return iterator != std::ranges::end(flowsAccessor.value()) ? iterator->value : nullptr; } [[nodiscard]] auto discoverFlowConfigs() const -> std::vector @@ -151,20 +155,28 @@ export namespace CppUtils::Language::Xml auto execute() -> Stats { - if (std::ranges::empty(m_flows)) + auto flowsToExecute = std::vector>{}; { - Logger<"Flow">::template print<"warning">( - "FlowRunner: 0 active flows found to execute (did you call loadFlows()?)"); - return Stats{}; + auto flowsAccessor = m_flows.sharedAccess(); + if (std::ranges::empty(flowsAccessor.value())) + { + Logger<"Flow">::template print<"warning">( + "FlowRunner: 0 active flows found to execute (did you call loadFlows()?)"); + return Stats{}; + } + + flowsToExecute.reserve(std::ranges::size(flowsAccessor.value())); + for (const auto& entry : flowsAccessor.value()) + flowsToExecute.push_back(entry.value); } auto futures = std::vector>{}; - futures.reserve(std::ranges::size(m_flows)); + futures.reserve(std::ranges::size(flowsToExecute)); auto chronometer = Chrono::Chronometer{}; - for (const auto& entry : m_flows) - futures.push_back(std::async(std::launch::async, [flow = entry.value] { - return flow->execute(); + for (auto&& flowInstance : flowsToExecute) + futures.push_back(std::async(std::launch::async, [flowInstance = std::move(flowInstance)] { + return flowInstance->execute(); })); auto processedFilesCount = 0uz; @@ -189,11 +201,14 @@ export namespace CppUtils::Language::Xml return; } - if (std::ranges::empty(m_flows)) + if (std::ranges::empty(m_flows.sharedAccess().value())) loadFlows(); - for (const auto& entry : m_flows) - entry.value->startWatching(); + { + auto flowsAccessor = m_flows.sharedAccess(); + for (const auto& entry : flowsAccessor.value()) + entry.value->startWatching(); + } m_flowsWatcher.reset(); m_flowsWatcher.onEvent([this](FileSystem::Event event, const std::filesystem::path& filePath) { @@ -202,11 +217,13 @@ export namespace CppUtils::Language::Xml if (event & (FileSystem::Event::Deleted | FileSystem::Event::MovedFrom)) { - if (m_flows.contains(filePath)) + if (auto flowsAccessor = m_flows.uniqueAccess(); + flowsAccessor.value().contains(filePath)) { - if (auto existingFlow = flow(filePath)) - existingFlow->stopWatching(); - m_flows.erase(filePath); + const auto iterator = flowsAccessor.value().find(filePath); + if (iterator != std::ranges::end(flowsAccessor.value()) and iterator->value) + iterator->value->stopWatching(); + flowsAccessor.value().erase(filePath); Logger<"Flow">::template print<"info">( "Workflow '{}' removed from active flows", filePath.filename().string()); } @@ -221,17 +238,19 @@ export namespace CppUtils::Language::Xml const auto& factory = resolveFactory(flowNode, filePath); auto newFlow = std::shared_ptr{factory(flowNode)}; - const auto isUpdate = m_flows.contains(filePath); + auto flowsAccessor = m_flows.uniqueAccess(); + const auto isUpdate = flowsAccessor.value().contains(filePath); if (isUpdate) { - if (auto existingFlow = flow(filePath)) - existingFlow->stopWatching(); - m_flows.erase(filePath); + if (const auto iterator = flowsAccessor.value().find(filePath); + iterator != std::ranges::end(flowsAccessor.value()) and iterator->value) + iterator->value->stopWatching(); + flowsAccessor.value().erase(filePath); } newFlow->startWatching(); const auto flowName = extractFlowName(flowNode, filePath); - m_flows.emplace(filePath, flowName, std::move(newFlow)); + flowsAccessor.value().emplace(filePath, flowName, std::move(newFlow)); if (isUpdate) Logger<"Flow">::template print<"info">( @@ -252,14 +271,15 @@ export namespace CppUtils::Language::Xml auto stopWatching() -> void { m_flowsWatcher.reset(); - for (const auto& entry : m_flows) + auto flowsAccessor = m_flows.sharedAccess(); + for (const auto& entry : flowsAccessor.value()) entry.value->stopWatching(); } private: Thread::ThreadPool& m_threadPool; std::filesystem::path m_flowsDirectory; - FlowContainer m_flows; + Thread::SharedLocker m_flows; std::unordered_map m_factories; FileSystem::Watcher m_flowsWatcher; }; diff --git a/tests/Language/XML/FlowRunner.mpp b/tests/Language/XML/FlowRunner.mpp index f27e4c09..90e92406 100644 --- a/tests/Language/XML/FlowRunner.mpp +++ b/tests/Language/XML/FlowRunner.mpp @@ -89,9 +89,9 @@ namespace CppUtils::UnitTest::Language::Xml::FlowRunner runner.registerMapping("alpha"); runner.registerMapping("beta"); - suite.expectEqual(std::ranges::size(runner.flows()), 0uz); + suite.expectEqual(std::ranges::size(runner.flows().value()), 0uz); runner.loadFlows(); - suite.expectEqual(std::ranges::size(runner.flows()), 2uz); + suite.expectEqual(std::ranges::size(runner.flows().value()), 2uz); const auto configs = runner.discoverFlowConfigs(); suite.expectEqual(std::ranges::size(configs), 2uz); @@ -123,5 +123,42 @@ namespace CppUtils::UnitTest::Language::Xml::FlowRunner suite.expect(not out2Content.contains("rec3")); }}; }); + + suite.addTest("Concurrent flow lookups and reloading", [&] { + CppUtils::FileSystem::TemporaryDirectory{[&suite](const auto& temporaryDirectory) -> void { + std::filesystem::create_directories(temporaryDirectory / "flows"); + const auto flowXml = R"( + + + + )"; + CppUtils::FileSystem::String::write(temporaryDirectory / "flows/alpha.xml", flowXml); + + auto threadPool = CppUtils::Thread::ThreadPool{2}; + auto runner = CppUtils::Language::Xml::FlowRunner{threadPool, temporaryDirectory / "flows"}; + runner.registerMapping("alpha"); + runner.loadFlows(); + + auto running = std::atomic{true}; + auto readerThreads = std::vector{}; + for (const auto _ : std::views::iota(0uz, 4uz)) + readerThreads.emplace_back([&runner, &running] { + while (running.load(std::memory_order_relaxed)) + { + auto _ = runner.flow("FlowAlpha"); + auto allFlows = runner.flows(); + volatile auto _ = std::ranges::size(allFlows.value()); + } + }); + + for (const auto _ : std::views::iota(0uz, 100uz)) + runner.loadFlows(); + + running.store(false, std::memory_order_relaxed); + readerThreads.clear(); + + suite.expect(runner.flow("FlowAlpha") != nullptr); + }}; + }); }}; }