From 91fc655b25652db3ce2148d7141f1b475698913c Mon Sep 17 00:00:00 2001 From: Hayodea Hekol Date: Tue, 9 Jun 2026 11:19:42 -0400 Subject: [PATCH] Nursery: Initial integration StimulusProducer: syncAwaitAllSettlements should pump caller io_context --- .../attachmentSupport/stimulusProducer.cpp | 229 ++++++++++-------- commonLibs/livoxProto1/udpCommandDemuxer.cpp | 10 +- .../adapters/boostAsio/deadlineTimerAReq.h | 33 +-- include/user/stimulusProducer.h | 44 ++-- libspinscale | 2 +- smocore/deviceManager/deviceReattacher.cpp | 171 ++++--------- .../include/deviceManager/deviceReattacher.h | 28 +-- smocore/include/marionette/marionette.h | 14 +- smocore/marionette/lifetime.cpp | 43 ++-- smocore/marionette/main.cpp | 27 ++- stimBuffApis/livoxGen1/livoxGen1.cpp | 10 +- .../livoxGen1/pcloudStimulusProducer.cpp | 48 ++-- .../livoxGen1/pcloudStimulusProducer.h | 13 +- .../nonViralTaskNursery_tests.cpp | 36 ++- tests/smocore/qutex_tests.cpp | 1 + 15 files changed, 326 insertions(+), 383 deletions(-) diff --git a/commonLibs/attachmentSupport/stimulusProducer.cpp b/commonLibs/attachmentSupport/stimulusProducer.cpp index 551d591..130e40b 100644 --- a/commonLibs/attachmentSupport/stimulusProducer.cpp +++ b/commonLibs/attachmentSupport/stimulusProducer.cpp @@ -3,10 +3,9 @@ #include #include #include -#include -#include #include #include +#include #include #include #include @@ -88,115 +87,137 @@ void StimulusProducer::destroyAttachedStimulusBuffer( } } +sscl::co::NonViralNonPostingInvoker StimulusProducer::productionCDaemon( + std::exception_ptr &, std::function, + sscl::SyncCancelerForAsyncWork &canceler) +{ + int nextDelayMs = CONFIG_STIMBUFF_FRAME_PERIOD_MS; + + do + { + bool shouldProduceFrame = false; + + if (!canceler.execUncancelableSegmentOrAbort([&]() + { + /** EXPLANATION: + * We need to ensure that there's only ever one stimframe being + * produced during any CONFIG_STIMBUFF_FRAME_PERIOD_MS period. To + * guarantee this, we use a spinlock. + * + * When a new frame is to be produced, the async producer will + * first acquire the frameAssemblyLimiter spinlock. This ensures + * that only one stimframe is produced during any + * CONFIG_STIMBUFF_FRAME_PERIOD_MS interval. When the next + * timeout fires, it checks if the previous stimframe has + * finished production. If the previous stimframe is still + * being produced, we will sleep for + * CONFIG_STIMBUFF_FRAME_RETRY_DELAY_MS ms before retrying. + */ + if (frameAssemblyRateLimiter.tryAcquire()) + { + nextDelayMs = CONFIG_STIMBUFF_FRAME_PERIOD_MS; + + // Check if we're ending a deferral period + if (nDeferrals > 0) + { + auto deferralEndTime = + std::chrono::high_resolution_clock::now(); + auto duration = deferralEndTime - deferralStartTime; + auto durationMs = std::chrono::duration_cast< + std::chrono::milliseconds>(duration); + + std::cout << "productionCDaemon: Deferral period ended. " + << "Total deferrals: " << nDeferrals + << ", Duration: " << durationMs.count() + << "ms" << std::endl; + + nDeferrals = 0; + } + + shouldProduceFrame = true; + } + else + { + nextDelayMs = CONFIG_STIMBUFF_FRAME_RETRY_DELAY_MS; + + ++nDeferrals; + // If this is first deferral, capture start stamp and print message + if (nDeferrals == 1) + { + deferralStartTime = + std::chrono::high_resolution_clock::now(); + std::cerr << "productionCDaemon: Deferral period " + "beginning. Configured deferral period: " + << nextDelayMs << "ms" << std::endl; + } + } + })) + { break; } + + if (shouldProduceFrame) + { + /** EXPLANATION: + * Call the derived class's frame production handler + * Note: The derived class's frame production handler (aka + * its implementation of stimFrameProductionTimesliceCInd()) must + * release the lock when frame production completes + */ + co_await stimFrameProductionTimesliceCInd(canceler); + } + else if (OptionParser::getOptions().verbose) + { + std::cerr << "productionCDaemon: Deferring frame by " + << nextDelayMs << "ms due to rate limit." << std::endl; + } + + // Schedule the next timeout using the provided delay + const bool expiredNormally = co_await + adapters::boostAsio::getDeadlineTimerAReqAwaiter( + ioContext, + daemonTimer, + boost::posix_time::milliseconds(nextDelayMs)); + + if (!expiredNormally) { + // Timer was cancelled, which is expected when stopping + break; + } + + // FIXME: We should be able to release the start/stop lock at this point. + + } while (!canceler.isCancellationRequested()); + + co_return; +} + +void StimulusProducer::start() +{ + std::cout << __func__ << ": Starting stimulus producer for device " + << deviceAttachmentSpec->deviceSelector << std::endl; + + nDeferrals = 0; + taskNursery.openAdmission(); + taskNursery.launch( + [this](sscl::co::NonViralTaskNursery::Slot::Lease &lease) + { + return productionCDaemon( + lease.getExceptionStorage(), + lease.getCallerLambda(), + lease.getSyncCanceler()); + }); +} + void StimulusProducer::stop() { - (void)stimulusProducerCanceler.requestStop(); - // Cancel timer immediately - timer.cancel(); + daemonTimer.cancel(); + taskNursery.requestCancelOnAll(); + taskNursery.closeAdmission(); + taskNursery.syncAwaitAllSettlements( + sscl::ComponentThread::getSelf()->getIoContext()); std::cout << __func__ << ": Stopped stimulus producer for device " << deviceAttachmentSpec->deviceSelector << std::endl; } -void StimulusProducer::scheduleNextTimeout(int delayMs) -{ - if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) - { return; } - - // Schedule the next timeout using the provided delay - timer.expires_from_now( - boost::posix_time::milliseconds(delayMs)); - - timer.async_wait( - std::bind( - &StimulusProducer::onTimeout, this, std::placeholders::_1)); -} - -void StimulusProducer::onTimeout(const boost::system::error_code& error) -{ - // Timer was cancelled, which is expected when stopping - if (error == boost::asio::error::operation_aborted) { - return; - } - - if (error) - { - std::cerr << "StimulusProducer: Timer error: " << error.message() - << std::endl; - return; - } - - sscl::SpinLock::Guard guard(stimulusProducerCanceler.s.lock); - if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) - { return; } - - /** EXPLANATION: - * We need to ensure that there's only ever one stimframe being produced - * during any CONFIG_STIMBUFF_FRAME_PERIOD_MS period. To guarantee this, we - * use a spinlock. - * - * When a new frame is to be produced, the async producer will first acquire - * the frameAssemblyLimiter spinlock. This way, when the next timeout is - * fired it can check whether its predecessor stimframe has finished being - * produced. If the preceding stimframe is still being produced, then we'll - * sleep for CONFIG_STIMBUFF_FRAME_RETRY_DELAY_MS ms before trying again. - */ - int nextWakeupDelayMs; - bool deferred = false; - if (frameAssemblyRateLimiter.tryAcquire()) - { - nextWakeupDelayMs = CONFIG_STIMBUFF_FRAME_PERIOD_MS; - - // Check if we're ending a deferral period - if (nDeferrals > 0) - { - auto deferralEndTime = std::chrono::high_resolution_clock::now(); - auto duration = deferralEndTime - deferralStartTime; - auto durationMs = std::chrono::duration_cast< - std::chrono::milliseconds>(duration); - - std::cout << __func__ << ": Deferral period ended. " - << "Total deferrals: " << nDeferrals - << ", Duration: " << durationMs.count() << "ms" << std::endl; - - nDeferrals = 0; - } - - /** EXPLANATION: - * Call the derived class's frame production handler - * Note: The derived class's frame production handler (aka - * its implementation of stimFrameProductionTimesliceInd()) must - * release the lock when frame production completes - */ - stimFrameProductionTimesliceInd(); - } - else - { - nextWakeupDelayMs = CONFIG_STIMBUFF_FRAME_RETRY_DELAY_MS; - deferred = true; - - ++nDeferrals; - // If this is first deferral, capture start stamp and print message - if (nDeferrals == 1) - { - deferralStartTime = std::chrono::high_resolution_clock::now(); - std::cerr << __func__ << ": Deferral period beginning. " - "Configured deferral period: " << nextWakeupDelayMs << "ms" - << std::endl; - } - } - - scheduleNextTimeout(nextWakeupDelayMs); - - // FIXME: We should be able to release the start/stop lock at this point. - - if (deferred && OptionParser::getOptions().verbose) - { - std::cerr << __func__ << ": Deferring frame by " << nextWakeupDelayMs - << "ms due to rate limit." << std::endl; - } -} - } // namespace stim_buff } // namespace smo diff --git a/commonLibs/livoxProto1/udpCommandDemuxer.cpp b/commonLibs/livoxProto1/udpCommandDemuxer.cpp index 064564e..b437f93 100644 --- a/commonLibs/livoxProto1/udpCommandDemuxer.cpp +++ b/commonLibs/livoxProto1/udpCommandDemuxer.cpp @@ -577,11 +577,11 @@ UdpCommandDemuxer::waitForCommandResponseCReq( * we will consider the command to have failed. */ boost::asio::io_context &ioContext = componentThread->getIoContext(); - std::optional> raceTimer; + boost::asio::deadline_timer raceTimer(ioContext); auto timerAwaiter = adapters::boostAsio::getDeadlineTimerAReqAwaiter( ioContext, - boost::posix_time::milliseconds(timeoutMs), - raceTimer); + raceTimer, + boost::posix_time::milliseconds(timeoutMs)); auto responseInvoker = waitForCommandResponseCReq(cmdSet, cmdId, deviceIp); static constexpr int timerMemberSettlementIndex = 0; @@ -598,8 +598,8 @@ UdpCommandDemuxer::waitForCommandResponseCReq( if (timerWonFirst) { cancelPendingCommandWait(cmdSet, cmdId, deviceIp); - } else if (raceTimer) { - (*raceTimer)->cancel(); + } else { + raceTimer.cancel(); } /** Group member adapter coros are fire-and-forget; keep group alive until diff --git a/include/adapters/boostAsio/deadlineTimerAReq.h b/include/adapters/boostAsio/deadlineTimerAReq.h index af05680..7c77309 100644 --- a/include/adapters/boostAsio/deadlineTimerAReq.h +++ b/include/adapters/boostAsio/deadlineTimerAReq.h @@ -5,7 +5,6 @@ #include #include #include -#include #include #include @@ -15,7 +14,10 @@ namespace adapters::boostAsio { -/** Coroutine awaiter: true if the delay elapsed, false if cancelled/aborted. */ +/** Coroutine awaiter: true if the delay elapsed, false if cancelled/aborted. + * + * Reuses the caller-supplied deadline_timer; does not allocate a new timer. + */ class DeadlineTimerAReq { public: @@ -24,22 +26,17 @@ public: std::atomic settled{false}; bool timerExpiredNormally = false; std::coroutine_handle<> callerSchedHandle; - std::shared_ptr timer; }; DeadlineTimerAReq( boost::asio::io_context &resumeIoContext, - const boost::posix_time::milliseconds delay, - std::optional> &timerOut) + boost::asio::deadline_timer &timer, + const boost::posix_time::milliseconds delay) : asyncState(std::make_shared()), resumeIoContext(resumeIoContext) { - asyncState->timer = - std::make_shared(resumeIoContext); - timerOut = asyncState->timer; - - asyncState->timer->expires_from_now(delay); - asyncState->timer->async_wait( + timer.expires_from_now(delay); + timer.async_wait( [this](const boost::system::error_code &error) { onTimer(error); @@ -92,19 +89,11 @@ private: }; inline auto getDeadlineTimerAReqAwaiter( - boost::asio::io_context &ioContext, + boost::asio::io_context &resumeIoContext, + boost::asio::deadline_timer &timer, const boost::posix_time::milliseconds delay) { - std::optional> timerOut; - return DeadlineTimerAReq(ioContext, delay, timerOut); -} - -inline auto getDeadlineTimerAReqAwaiter( - boost::asio::io_context &ioContext, - const boost::posix_time::milliseconds delay, - std::optional> &timerOut) -{ - return DeadlineTimerAReq(ioContext, delay, timerOut); + return DeadlineTimerAReq(resumeIoContext, timer, delay); } } // namespace adapters::boostAsio diff --git a/include/user/stimulusProducer.h b/include/user/stimulusProducer.h index a91af61..9b1078b 100644 --- a/include/user/stimulusProducer.h +++ b/include/user/stimulusProducer.h @@ -4,14 +4,14 @@ #include #include #include -#include -#include #include #include #include #include #include #include +#include +#include #include #include #include "deviceAttachmentSpec.h" @@ -39,9 +39,8 @@ public: const std::shared_ptr &deviceAttachmentSpec, boost::asio::io_context& ioContext_) - : deviceAttachmentSpec(deviceAttachmentSpec), - ioContext(ioContext_), - timer(ioContext), nDeferrals(0) + : daemonTimer(ioContext_), nDeferrals(0), + deviceAttachmentSpec(deviceAttachmentSpec), ioContext(ioContext_) {} virtual ~StimulusProducer() = default; @@ -52,17 +51,7 @@ public: StimulusProducer(StimulusProducer&&) = default; StimulusProducer& operator=(StimulusProducer&&) = default; - // Control methods - virtual void start() - { - std::cout << __func__ << ": Starting stimulus producer for device " - << deviceAttachmentSpec->deviceSelector << std::endl; - - stimulusProducerCanceler.startAcceptingWork(); - nDeferrals = 0; - scheduleNextTimeout(); - } - + virtual void start(); virtual void stop(); void allowNextStimulusFrame() @@ -96,10 +85,19 @@ protected: return CONFIG_STIMBUFF_FRAME_PERIOD_MS; } - virtual void stimFrameProductionTimesliceInd() = 0; + virtual sscl::co::ViralNonPostingInvoker + stimFrameProductionTimesliceCInd( + sscl::SyncCancelerForAsyncWork &canceler) = 0; -private: - void onTimeout(const boost::system::error_code& error); + sscl::co::NonViralNonPostingInvoker productionCDaemon( + std::exception_ptr &exceptionPtr, + std::function callback, + sscl::SyncCancelerForAsyncWork &canceler); + + sscl::co::NonViralTaskNursery taskNursery; + boost::asio::deadline_timer daemonTimer; + size_t nDeferrals; + std::chrono::high_resolution_clock::time_point deferralStartTime; public: std::shared_ptr deviceAttachmentSpec; @@ -107,14 +105,6 @@ public: private: boost::asio::io_context& ioContext; -protected: - sscl::SyncCancelerForAsyncWork stimulusProducerCanceler; -private: - boost::asio::deadline_timer timer; - size_t nDeferrals; - std::chrono::high_resolution_clock::time_point deferralStartTime; - - void scheduleNextTimeout(int delayMs = CONFIG_STIMBUFF_FRAME_PERIOD_MS); }; } // namespace stim_buff diff --git a/libspinscale b/libspinscale index b04b0db..656aae3 160000 --- a/libspinscale +++ b/libspinscale @@ -1 +1 @@ -Subproject commit b04b0db1557832b5b952bd2902f2bf3169142be6 +Subproject commit 656aae37c81c52a274f309ad60e374cabdc6da8b diff --git a/smocore/deviceManager/deviceReattacher.cpp b/smocore/deviceManager/deviceReattacher.cpp index 549e83f..ea2a7be 100644 --- a/smocore/deviceManager/deviceReattacher.cpp +++ b/smocore/deviceManager/deviceReattacher.cpp @@ -1,161 +1,88 @@ #include -#include #include #include #include +#include #include #include -#include #include namespace smo { namespace device { -namespace { - -constexpr unsigned int reattachInFlightStaleThresholdMultiplier = 4; - -} // namespace - DeviceReattacher::DeviceReattacher( DeviceManager& parent, std::shared_ptr ioThread) -: parent(parent), ioThread(ioThread), timer(ioThread->getIoContext()) +: parent(parent), +ioThread(ioThread), daemonTimer(ioThread->getIoContext()) { /** EXPLANATION: - * The thread on which DeviceReattacher runs is whichever thread executes - * the io_context that owns deadline_timer. Timer async_wait handlers - * (onTimeout, holdReattachCReq, reattachKnownListCReq) are dispatched on - * that thread. ioThread selects that io_context here; start() only arms - * the timer on it. + * deviceReattacherCDaemon is a dynamic posting non-viral coroutine: start() + * passes ExplicitPostTarget{ioThread->getIoContext()} so the daemon body + * always runs on ioThread. daemonTimer is reused each loop iteration. */ } -mrntt::MrnttNonViralNonPostingInvoker DeviceReattacher::reattachKnownListCReq( +sscl::co::DynamicNonViralPostingInvoker +DeviceReattacher::deviceReattacherCDaemon( + [[maybe_unused]] sscl::co::ExplicitPostTarget postTarget, [[maybe_unused]] std::exception_ptr &exceptionPtr, - [[maybe_unused]] std::function callback) + [[maybe_unused]] std::function callback, + sscl::SyncCancelerForAsyncWork &canceler) { - /** EXPLANATION: - * Non-posting: invoked from holdReattachCReq on the timer callback thread - * (see ctor). Nested DeviceManager attach APIs still post to MRNTT as - * needed via their own viral posting invokers. - */ - co_await parent.attachAllUnattachedDevicesFromKnownListCReq(); + boost::asio::io_context &timerIoContext = + sscl::ComponentThread::getSelf()->getIoContext(); + const auto periodMs = boost::posix_time::milliseconds( + CONFIG_MRNTT_DEVMGR_REATTACHER_PERIOD_MS); + + while (!canceler.isCancellationRequested()) + { + const bool expiredNormally = co_await + adapters::boostAsio::getDeadlineTimerAReqAwaiter( + timerIoContext, daemonTimer, periodMs); + + if (!expiredNormally) { + break; + } + + co_await parent.attachAllUnattachedDevicesFromKnownListCReq(); + } + co_return; } void DeviceReattacher::start() { - deviceReattacherCanceler.startAcceptingWork(); - scheduleNextTimeout(); -} - -void DeviceReattacher::stop() -{ - { - sscl::SpinLock::Guard guard(deviceReattacherCanceler.s.lock); - reattachOpInFlight = false; - deviceReattacherCanceler.s.rsrc.shouldContinue = false; - /** EXPLANATION: - * Do not call reattachCReqInvoker.reset() here. Forcibly destroying - * the invoker would tear down an in-flight reattach coroutine frame - * mid-operation. During normal program teardown the optional (and - * its invoker) are destroyed with the rest of the binary anyway; leave - * a running reattach time to finish if shutdown races with it. - */ - } - - timer.cancel(); -} - -void DeviceReattacher::scheduleNextTimeout() -{ - if (deviceReattacherCanceler.isCancellationRequestedUnlocked()) { - return; - } - - // Schedule the next timeout using the configured period - timer.expires_from_now( - boost::posix_time::milliseconds( - CONFIG_MRNTT_DEVMGR_REATTACHER_PERIOD_MS)); - - timer.async_wait( - std::bind(&DeviceReattacher::onTimeout, this, std::placeholders::_1)); -} - -void DeviceReattacher::holdReattachCReq() -{ - reattachOpInFlight = true; - lastReattachReqTimestamp = std::chrono::steady_clock::now(); - - reattachCReqInvoker.reset(); - reattachCReqInvoker.emplace(reattachKnownListCReq( - reattachLifetimeExceptionPtr, - [this]() + taskNursery.openAdmission(); + taskNursery.launch( + [this](sscl::co::NonViralTaskNursery::Slot::Lease &lease) { - sscl::co::NonViralCompletion nvc(reattachLifetimeExceptionPtr); + return deviceReattacherCDaemon( + sscl::co::ExplicitPostTarget{ioThread->getIoContext()}, + lease.getExceptionStorage(), + lease.getCallerLambda(), + lease.getSyncCanceler()); + }, + [](std::exception_ptr &exceptionPtr) + { + sscl::co::NonViralCompletion nvc(exceptionPtr); if (nvc.hasException()) { try { nvc.checkAndRethrowException(); } catch (const std::exception &e) { - std::cerr << "DeviceReattacher: " << e.what() - << std::endl; + std::cerr << "DeviceReattacher: " + << e.what() << std::endl; } } - - sscl::SpinLock::Guard guard(deviceReattacherCanceler.s.lock); - reattachOpInFlight = false; - })); + }); } -void DeviceReattacher::onTimeout(const boost::system::error_code& error) +void DeviceReattacher::stop() { - // Timer was cancelled, which is expected when stopping - if (error == boost::asio::error::operation_aborted) { - return; - } - - if (error) - { - std::cerr << "DeviceReattacher: Timer error: " << error.message() - << std::endl; - return; - } - - sscl::SpinLock::Guard guard(deviceReattacherCanceler.s.lock); - if (deviceReattacherCanceler.isCancellationRequestedUnlocked()) { - return; - } - - const auto staleThreshold = std::chrono::milliseconds( - reattachInFlightStaleThresholdMultiplier - * CONFIG_MRNTT_DEVMGR_REATTACHER_PERIOD_MS); - - // Attempt to reattach all unattached devices from the known list - if (!reattachOpInFlight) - { - holdReattachCReq(); - } - else - { - const auto elapsedSinceLastReattachReq = - std::chrono::steady_clock::now() - lastReattachReqTimestamp; - - if (elapsedSinceLastReattachReq >= staleThreshold) - { - std::cerr << "DeviceReattacher: Reattach op still in flight after " - << std::chrono::duration_cast( - elapsedSinceLastReattachReq).count() - << "ms (threshold " - << staleThreshold.count() - << "ms); forcing a new reattach request." - << std::endl; - holdReattachCReq(); - } - } - - // Schedule the next timeout - scheduleNextTimeout(); + daemonTimer.cancel(); + taskNursery.requestCancelOnAll(); + taskNursery.closeAdmission(); + taskNursery.syncAwaitAllSettlements(ioThread->getIoContext()); } } // namespace device diff --git a/smocore/include/deviceManager/deviceReattacher.h b/smocore/include/deviceManager/deviceReattacher.h index 7ebc8ed..e7d9e7d 100644 --- a/smocore/include/deviceManager/deviceReattacher.h +++ b/smocore/include/deviceManager/deviceReattacher.h @@ -1,14 +1,12 @@ #ifndef DEVICEREATTACHER_H #define DEVICEREATTACHER_H -#include -#include -#include #include #include -#include #include -#include +#include +#include +#include #include namespace smo { @@ -32,23 +30,17 @@ public: void stop(); private: - void scheduleNextTimeout(); - void onTimeout(const boost::system::error_code& error); - void holdReattachCReq(); - - mrntt::MrnttNonViralNonPostingInvoker reattachKnownListCReq( + sscl::co::DynamicNonViralPostingInvoker deviceReattacherCDaemon( + sscl::co::ExplicitPostTarget postTarget, std::exception_ptr &exceptionPtr, - std::function callback); + std::function callback, + sscl::SyncCancelerForAsyncWork &canceler); DeviceManager &parent; - // io_context thread for timer and non-posting reattach shell (see ctor). + // io_context thread for timer (see ctor). std::shared_ptr ioThread; - sscl::SyncCancelerForAsyncWork deviceReattacherCanceler; - boost::asio::deadline_timer timer; - std::exception_ptr reattachLifetimeExceptionPtr; - std::optional reattachCReqInvoker; - bool reattachOpInFlight = false; - std::chrono::steady_clock::time_point lastReattachReqTimestamp{}; + sscl::co::NonViralTaskNursery taskNursery; + boost::asio::deadline_timer daemonTimer; }; } // namespace device diff --git a/smocore/include/marionette/marionette.h b/smocore/include/marionette/marionette.h index 5d98292..e2042d2 100644 --- a/smocore/include/marionette/marionette.h +++ b/smocore/include/marionette/marionette.h @@ -9,6 +9,7 @@ #include #include #include +#include #include namespace sscl { @@ -29,8 +30,10 @@ public: {} ~MarionetteComponent() = default; - void holdInitializeCReq(std::function completion); - void holdFinalizeCReq(std::function completion); + void holdInitializeCReq( + std::function completion); + void holdFinalizeCReq( + std::function completion); MrnttNonViralPostingInvoker initializeCReq( std::exception_ptr &exceptionPtr, @@ -58,12 +61,7 @@ protected: private: std::unique_ptr signals; bool callShutdownSalmanoff = false; - std::optional initializeCReqInvoker; - std::optional finalizeCReqInvoker; - -public: - std::exception_ptr initializeLifetimeExceptionPtr; - std::exception_ptr finalizeLifetimeExceptionPtr; + sscl::co::NonViralTaskNursery taskNursery; }; extern std::shared_ptr thread; diff --git a/smocore/marionette/lifetime.cpp b/smocore/marionette/lifetime.cpp index d3a2078..485cf30 100644 --- a/smocore/marionette/lifetime.cpp +++ b/smocore/marionette/lifetime.cpp @@ -30,33 +30,29 @@ void assertMarionetteThread() } // namespace void MarionetteComponent::holdInitializeCReq( - std::function completion) + std::function completion) { - initializeLifetimeExceptionPtr = nullptr; - initializeCReqInvoker.emplace(initializeCReq( - initializeLifetimeExceptionPtr, - [completion = std::move(completion)]() + taskNursery.launch( + [this](sscl::co::NonViralTaskNursery::Slot::Lease &lease) { - sscl::co::NonViralCompletion nvc( - mrntt.initializeLifetimeExceptionPtr); - nvc.checkAndRethrowException(); - completion(); - })); + return initializeCReq( + lease.getExceptionStorage(), + lease.getCallerLambda()); + }, + std::move(completion)); } void MarionetteComponent::holdFinalizeCReq( - std::function completion) + std::function completion) { - finalizeLifetimeExceptionPtr = nullptr; - finalizeCReqInvoker.emplace(finalizeCReq( - finalizeLifetimeExceptionPtr, - [completion = std::move(completion)]() + taskNursery.launch( + [this](sscl::co::NonViralTaskNursery::Slot::Lease &lease) { - sscl::co::NonViralCompletion nvc( - mrntt.finalizeLifetimeExceptionPtr); - nvc.checkAndRethrowException(); - completion(); - })); + return finalizeCReq( + lease.getExceptionStorage(), + lease.getCallerLambda()); + }, + std::move(completion)); } MrnttNonViralPostingInvoker MarionetteComponent::initializeCReq( @@ -107,7 +103,12 @@ void MarionetteComponent::exceptionInd() [] { mrntt.holdFinalizeCReq( - []() { marionetteFinalizeReqCb(true); }); + [](std::exception_ptr &exceptionPtr) + { + sscl::co::NonViralCompletion nvc(exceptionPtr); + nvc.checkAndRethrowException(); + marionetteFinalizeReqCb(true); + }); }); } diff --git a/smocore/marionette/main.cpp b/smocore/marionette/main.cpp index c664fcf..04ae979 100644 --- a/smocore/marionette/main.cpp +++ b/smocore/marionette/main.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -33,7 +34,12 @@ void marionetteInitializeReqCb(bool success) << '\n'; mrntt.holdFinalizeCReq( - []() { marionetteFinalizeReqCb(true); }); + [](std::exception_ptr &exceptionPtr) + { + sscl::co::NonViralCompletion nvc(exceptionPtr); + nvc.checkAndRethrowException(); + marionetteFinalizeReqCb(true); + }); } void marionetteFinalizeReqCb(bool success) @@ -82,7 +88,12 @@ void MarionetteComponent::postJoltHook() break; } mrntt.holdFinalizeCReq( - []() { marionetteFinalizeReqCb(true); }); + [](std::exception_ptr &exceptionPtr) + { + sscl::co::NonViralCompletion nvc(exceptionPtr); + nvc.checkAndRethrowException(); + marionetteFinalizeReqCb(true); + }); }); } @@ -128,14 +139,24 @@ void MarionetteComponent::preLoopHook() smo::initializeSalmanoff(); callShutdownSalmanoff = true; + taskNursery.openAdmission(); + holdInitializeCReq( - [] { marionetteInitializeReqCb(true); }); + [](std::exception_ptr &exceptionPtr) + { + sscl::co::NonViralCompletion nvc(exceptionPtr); + nvc.checkAndRethrowException(); + marionetteInitializeReqCb(true); + }); std::cout << "PuppeteerThread::main: Entering event loop" << "\n"; } void MarionetteComponent::postLoopHook() { + taskNursery.requestCancelOnAll(); + taskNursery.closeAdmission(); + std::cout << "PuppeteerThread::main: Exited event loop" << "\n"; } diff --git a/stimBuffApis/livoxGen1/livoxGen1.cpp b/stimBuffApis/livoxGen1/livoxGen1.cpp index ea472ba..e869142 100644 --- a/stimBuffApis/livoxGen1/livoxGen1.cpp +++ b/stimBuffApis/livoxGen1/livoxGen1.cpp @@ -355,8 +355,11 @@ attachByCreatingProducer( * may not yet be ready for another command. */ // Initialize timer with LivoxGen1 metadata io_context + boost::asio::deadline_timer commandDelayTimer( + componentThread->getIoContext()); const bool delayOk = co_await adapters::boostAsio::getDeadlineTimerAReqAwaiter( componentThread->getIoContext(), + commandDelayTimer, boost::posix_time::milliseconds(LIVOX_GEN1_DEVICE_COMMAND_DELAY_MS)); if (!delayOk) @@ -540,10 +543,13 @@ livoxGen1_detachDeviceCReq( // Add 5ms delay before destroying device - // Helper method to delay and then call destroyDeviceReq - // Initialize timer with LivoxGen1 metadata io_context + // Helper method to delay and then call destroyDeviceReq + // Initialize timer with LivoxGen1 metadata io_context + boost::asio::deadline_timer commandDelayTimer( + requestComponentThread->getIoContext()); co_await adapters::boostAsio::getDeadlineTimerAReqAwaiter( requestComponentThread->getIoContext(), + commandDelayTimer, boost::posix_time::milliseconds(LIVOX_GEN1_DEVICE_COMMAND_DELAY_MS)); // No other buffers - stop and remove StimProducer diff --git a/stimBuffApis/livoxGen1/pcloudStimulusProducer.cpp b/stimBuffApis/livoxGen1/pcloudStimulusProducer.cpp index 5d41179..b86fd44 100644 --- a/stimBuffApis/livoxGen1/pcloudStimulusProducer.cpp +++ b/stimBuffApis/livoxGen1/pcloudStimulusProducer.cpp @@ -428,11 +428,6 @@ PcloudStimulusProducer::getOrCreateAttachedStimulusBuffer( } } -void PcloudStimulusProducer::stimFrameProductionTimesliceInd() -{ - holdProduceFrameCReq(); -} - struct AllowNextStimulusFrameGuard { PcloudStimulusProducer& producer; @@ -445,22 +440,14 @@ struct AllowNextStimulusFrameGuard { producer.allowNextStimulusFrame(); } }; -void PcloudStimulusProducer::holdProduceFrameCReq() -{ - /** EXPLANATION: - * We shouldn't acquire stimulusProducerCanceler.s.lock here because - * this function is called from - * StimulusProducer::stimFrameProductionTimesliceInd(), which is already - * holding the lock. - */ - activeProduceFrameInvoker.emplace(produceFrameCReq( - produceFrameExceptionPtr, - [this]() { activeProduceFrameInvoker.reset(); })); -} - -sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq( - [[maybe_unused]] std::exception_ptr& exceptionPtr, - [[maybe_unused]] std::function completion) +/** EXPLANATION: + * productionCDaemon co_awaits this viral timeslice and passes its + * nursery-slot canceler. Cooperative shutdown uses that canceler between + * co_await steps; do not use execUncancelableSegmentOrAbort here. + */ +sscl::co::ViralNonPostingInvoker +PcloudStimulusProducer::stimFrameProductionTimesliceCInd( + sscl::SyncCancelerForAsyncWork &canceler) { AllowNextStimulusFrameGuard allowNextGuard(*this); StimulusFrame& stimulusFrame = tempStimulusFrame; @@ -473,14 +460,11 @@ sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq( // produceFrameReq1_doAssemble_posted /** EXPLANATION: - * stimFrameProductionTimesliceInd() is entered with - * stimulusProducerCanceler.s.lock held; do not re-acquire here. - * - * This function is called from - * StimulusProducer::stimFrameProductionTimesliceInd(), whose caller is - * already holding the lock. + * productionCDaemon co_awaits this timeslice outside its uncancelable + * rate-limiter segment; use isCancellationRequested() here, not + * execUncancelableSegmentOrAbort. */ - if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) + if (canceler.isCancellationRequested()) { co_return; } cpsBoundary::AssembleFrameResult assembleResult = co_await @@ -491,8 +475,8 @@ sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq( std::optional lightAmbienceProductionDescDesc; std::optional darkAmbienceProductionDescDesc; { - sscl::SpinLock::Guard guard(stimulusProducerCanceler.s.lock); - if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) + sscl::SpinLock::Guard guard(canceler.s.lock); + if (canceler.isCancellationRequestedUnlocked()) { co_return; } if (!assembleResult.success) @@ -612,8 +596,8 @@ sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq( } { - sscl::SpinLock::Guard guard(stimulusProducerCanceler.s.lock); - if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) + sscl::SpinLock::Guard guard(canceler.s.lock); + if (canceler.isCancellationRequestedUnlocked()) { co_return; } if (!compactCollateSuccess) { diff --git a/stimBuffApis/livoxGen1/pcloudStimulusProducer.h b/stimBuffApis/livoxGen1/pcloudStimulusProducer.h index b8ab308..98682d3 100644 --- a/stimBuffApis/livoxGen1/pcloudStimulusProducer.h +++ b/stimBuffApis/livoxGen1/pcloudStimulusProducer.h @@ -82,13 +82,9 @@ public: const std::shared_ptr& buffer) override; protected: - void stimFrameProductionTimesliceInd() override; - - void holdProduceFrameCReq(); - - sscl::co::NonViralNonPostingInvoker produceFrameCReq( - std::exception_ptr& exceptionPtr, - std::function completion); + sscl::co::ViralNonPostingInvoker + stimFrameProductionTimesliceCInd( + sscl::SyncCancelerForAsyncWork &canceler) override; public: size_t nDgramsPerStagingBufferFrame; @@ -112,9 +108,6 @@ public: lightAmbienceStimulusBuffer; std::atomic> darkAmbienceStimulusBuffer; - - std::optional activeProduceFrameInvoker; - std::exception_ptr produceFrameExceptionPtr; }; } // namespace stim_buff diff --git a/tests/libspinscale/nonViralTaskNursery_tests.cpp b/tests/libspinscale/nonViralTaskNursery_tests.cpp index 078a442..4fdf7bb 100644 --- a/tests/libspinscale/nonViralTaskNursery_tests.cpp +++ b/tests/libspinscale/nonViralTaskNursery_tests.cpp @@ -139,7 +139,9 @@ TEST_F(NonViralTaskNurseryTest, SetOnSettledHookRejectsAfterFillSlot) lease.getCallerLambda()); }); - EXPECT_THROW(lease.setOnSettledHook([]() {}), std::runtime_error); + EXPECT_THROW( + lease.setOnSettledHook([](std::exception_ptr &) {}), + std::runtime_error); lease.commit(); } @@ -322,6 +324,26 @@ TEST_F(NonViralTaskNurseryTest, LaunchSugar) EXPECT_TRUE(nursery.allSettled()); } +TEST_F(NonViralTaskNurseryTest, LaunchWithOnSettledHook) +{ + std::atomic hookRan{false}; + + nursery.launch( + [](sscl::co::NonViralTaskNursery::Slot::Lease &lease) + { + return immediateCompleteCReq( + lease.getExceptionStorage(), + lease.getCallerLambda()); + }, + [&hookRan](std::exception_ptr &) + { + hookRan.store(true, std::memory_order_release); + }); + + EXPECT_TRUE(hookRan.load(std::memory_order_acquire)); + EXPECT_TRUE(nursery.allSettled()); +} + TEST_F(NonViralTaskNurseryTest, HandleStability) { auto handle = nursery.launch( @@ -520,7 +542,7 @@ TEST_F(NonViralTaskNurseryTest, OnSettledHookRunsAtRetirement) auto lease = nursery.getNewSlotLease(); lease.setOnSettledHook( - [&hookRan]() + [&hookRan](std::exception_ptr &) { hookRan.store(true, std::memory_order_release); }); @@ -536,13 +558,14 @@ TEST_F(NonViralTaskNurseryTest, OnSettledHookRunsAtRetirement) EXPECT_TRUE(hookRan.load(std::memory_order_acquire)); } -TEST_F(NonViralTaskNurseryTest, OnSettledHookExceptionStillRetiresSlot) +TEST_F(NonViralTaskNurseryTest, OnSettledHookSeesRetiredSlot) { auto lease = nursery.getNewSlotLease(); lease.setOnSettledHook( - []() + [this](std::exception_ptr &) { - throw std::runtime_error("onSettled hook failure"); + EXPECT_TRUE(nursery.allSettled()); + EXPECT_EQ(nursery.unsettledCount(), 0U); }); lease.fillSlot( [&lease]() @@ -552,9 +575,6 @@ TEST_F(NonViralTaskNurseryTest, OnSettledHookExceptionStillRetiresSlot) lease.getCallerLambda()); }); lease.commit(); - - EXPECT_TRUE(nursery.allSettled()); - EXPECT_EQ(nursery.unsettledCount(), 0U); } TEST_F(NonViralTaskNurseryTest, DuplicateRetireThrows) diff --git a/tests/smocore/qutex_tests.cpp b/tests/smocore/qutex_tests.cpp index b66dd2c..dfaf01d 100644 --- a/tests/smocore/qutex_tests.cpp +++ b/tests/smocore/qutex_tests.cpp @@ -25,6 +25,7 @@ public: } void awaken(bool forceAwaken = false) override { + (void)forceAwaken; awakened = true; } };