hayodea/salmanoff
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Nursery: Initial integration

Browse Source 
StimulusProducer: syncAwaitAllSettlements should pump caller io_context
This commit is contained in:
Hayodea Hekol
2026-06-09 11:19:42 -04:00
parent 5b81ea893c
commit 91fc655b25
15 changed files with 326 additions and 383 deletions
Download Patch File Download Diff File Expand all files Collapse all files
commonLibs/attachmentSupport/stimulusProducer.cpp
+125 -104
View File
@@ -3,10 +3,9 @@
#include <chrono> #include <chrono>
#include <algorithm> #include <algorithm>
#include <boost/asio/io_context.hpp> #include <boost/asio/io_context.hpp>
#include <boost/asio/deadline_timer.hpp>
#include <boost/system/error_code.hpp>
#include <opts.h> #include <opts.h>
#include <componentThread.h> #include <componentThread.h>
#include <adapters/boostAsio/deadlineTimerAReq.h>
#include <spinscale/spinLock.h> #include <spinscale/spinLock.h>
#include <user/stimulusProducer.h> #include <user/stimulusProducer.h>
#include <user/stimulusBuffer.h> #include <user/stimulusBuffer.h>
@@ -88,115 +87,137 @@ void StimulusProducer::destroyAttachedStimulusBuffer(
} }
} }
sscl::co::NonViralNonPostingInvoker StimulusProducer::productionCDaemon(
std::exception_ptr &, std::function<void()>,
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 StimulusProducer::stop()
{ {
(void)stimulusProducerCanceler.requestStop();
// Cancel timer immediately // 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 " std::cout << __func__ << ": Stopped stimulus producer for device "
<< deviceAttachmentSpec->deviceSelector << std::endl; << 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 stim_buff
} // namespace smo } // namespace smo
commonLibs/livoxProto1/udpCommandDemuxer.cpp
+5 -5
View File
@@ -577,11 +577,11 @@ UdpCommandDemuxer::waitForCommandResponseCReq(
* we will consider the command to have failed. * we will consider the command to have failed.
*/ */
boost::asio::io_context &ioContext = componentThread->getIoContext(); boost::asio::io_context &ioContext = componentThread->getIoContext();
std::optional<std::shared_ptr<boost::asio::deadline_timer>> raceTimer; boost::asio::deadline_timer raceTimer(ioContext);
auto timerAwaiter = adapters::boostAsio::getDeadlineTimerAReqAwaiter( auto timerAwaiter = adapters::boostAsio::getDeadlineTimerAReqAwaiter(
ioContext, ioContext,
boost::posix_time::milliseconds(timeoutMs), raceTimer,
raceTimer); boost::posix_time::milliseconds(timeoutMs));
auto responseInvoker = waitForCommandResponseCReq(cmdSet, cmdId, deviceIp); auto responseInvoker = waitForCommandResponseCReq(cmdSet, cmdId, deviceIp);
static constexpr int timerMemberSettlementIndex = 0; static constexpr int timerMemberSettlementIndex = 0;
@@ -598,8 +598,8 @@ UdpCommandDemuxer::waitForCommandResponseCReq(
if (timerWonFirst) { if (timerWonFirst) {
cancelPendingCommandWait(cmdSet, cmdId, deviceIp); cancelPendingCommandWait(cmdSet, cmdId, deviceIp);
} else if (raceTimer) { } else {
(*raceTimer)->cancel(); raceTimer.cancel();
} }
/** Group member adapter coros are fire-and-forget; keep group alive until /** Group member adapter coros are fire-and-forget; keep group alive until
include/adapters/boostAsio/deadlineTimerAReq.h
+11 -22
View File
@@ -5,7 +5,6 @@
#include <atomic> #include <atomic>
#include <coroutine> #include <coroutine>
#include <memory> #include <memory>
#include <optional>
#include <boost/asio/deadline_timer.hpp> #include <boost/asio/deadline_timer.hpp>
#include <boost/asio/io_context.hpp> #include <boost/asio/io_context.hpp>
@@ -15,7 +14,10 @@
namespace adapters::boostAsio { 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 class DeadlineTimerAReq
{ {
public: public:
@@ -24,22 +26,17 @@ public:
std::atomic<bool> settled{false}; std::atomic<bool> settled{false};
bool timerExpiredNormally = false; bool timerExpiredNormally = false;
std::coroutine_handle<> callerSchedHandle; std::coroutine_handle<> callerSchedHandle;
std::shared_ptr<boost::asio::deadline_timer> timer;
}; };
DeadlineTimerAReq( DeadlineTimerAReq(
boost::asio::io_context &resumeIoContext, boost::asio::io_context &resumeIoContext,
const boost::posix_time::milliseconds delay, boost::asio::deadline_timer &timer,
std::optional<std::shared_ptr<boost::asio::deadline_timer>> &timerOut) const boost::posix_time::milliseconds delay)
: asyncState(std::make_shared<AsyncState>()), : asyncState(std::make_shared<AsyncState>()),
resumeIoContext(resumeIoContext) resumeIoContext(resumeIoContext)
{ {
asyncState->timer = timer.expires_from_now(delay);
std::make_shared<boost::asio::deadline_timer>(resumeIoContext); timer.async_wait(
timerOut = asyncState->timer;
asyncState->timer->expires_from_now(delay);
asyncState->timer->async_wait(
[this](const boost::system::error_code &error) [this](const boost::system::error_code &error)
{ {
onTimer(error); onTimer(error);
@@ -92,19 +89,11 @@ private:
}; };
inline auto getDeadlineTimerAReqAwaiter( inline auto getDeadlineTimerAReqAwaiter(
boost::asio::io_context &ioContext, boost::asio::io_context &resumeIoContext,
boost::asio::deadline_timer &timer,
const boost::posix_time::milliseconds delay) const boost::posix_time::milliseconds delay)
{ {
std::optional<std::shared_ptr<boost::asio::deadline_timer>> timerOut; return DeadlineTimerAReq(resumeIoContext, timer, delay);
return DeadlineTimerAReq(ioContext, delay, timerOut);
}
inline auto getDeadlineTimerAReqAwaiter(
boost::asio::io_context &ioContext,
const boost::posix_time::milliseconds delay,
std::optional<std::shared_ptr<boost::asio::deadline_timer>> &timerOut)
{
return DeadlineTimerAReq(ioContext, delay, timerOut);
} }
} // namespace adapters::boostAsio } // namespace adapters::boostAsio
include/user/stimulusProducer.h
+17 -27
View File
@@ -4,14 +4,14 @@
#include <vector> #include <vector>
#include <memory> #include <memory>
#include <cstdint> #include <cstdint>
#include <atomic>
#include <mutex>
#include <functional> #include <functional>
#include <iostream> #include <iostream>
#include <chrono> #include <chrono>
#include <config.h> #include <config.h>
#include <boost/asio/io_context.hpp> #include <boost/asio/io_context.hpp>
#include <boost/asio/deadline_timer.hpp> #include <boost/asio/deadline_timer.hpp>
#include <spinscale/co/invokers.h>
#include <spinscale/co/nonViralTaskNursery.h>
#include <spinscale/spinLock.h> #include <spinscale/spinLock.h>
#include <spinscale/syncCancelerForAsyncWork.h> #include <spinscale/syncCancelerForAsyncWork.h>
#include "deviceAttachmentSpec.h" #include "deviceAttachmentSpec.h"
@@ -39,9 +39,8 @@ public:
const std::shared_ptr<device::DeviceAttachmentSpec> const std::shared_ptr<device::DeviceAttachmentSpec>
&deviceAttachmentSpec, &deviceAttachmentSpec,
boost::asio::io_context& ioContext_) boost::asio::io_context& ioContext_)
: deviceAttachmentSpec(deviceAttachmentSpec), : daemonTimer(ioContext_), nDeferrals(0),
ioContext(ioContext_), deviceAttachmentSpec(deviceAttachmentSpec), ioContext(ioContext_)
timer(ioContext), nDeferrals(0)
{} {}
virtual ~StimulusProducer() = default; virtual ~StimulusProducer() = default;
@@ -52,17 +51,7 @@ public:
StimulusProducer(StimulusProducer&&) = default; StimulusProducer(StimulusProducer&&) = default;
StimulusProducer& operator=(StimulusProducer&&) = default; StimulusProducer& operator=(StimulusProducer&&) = default;
// Control methods virtual void start();
virtual void start()
{
std::cout << __func__ << ": Starting stimulus producer for device "
<< deviceAttachmentSpec->deviceSelector << std::endl;
stimulusProducerCanceler.startAcceptingWork();
nDeferrals = 0;
scheduleNextTimeout();
}
virtual void stop(); virtual void stop();
void allowNextStimulusFrame() void allowNextStimulusFrame()
@@ -96,10 +85,19 @@ protected:
return CONFIG_STIMBUFF_FRAME_PERIOD_MS; return CONFIG_STIMBUFF_FRAME_PERIOD_MS;
} }
virtual void stimFrameProductionTimesliceInd() = 0; virtual sscl::co::ViralNonPostingInvoker<void>
stimFrameProductionTimesliceCInd(
sscl::SyncCancelerForAsyncWork &canceler) = 0;
private: sscl::co::NonViralNonPostingInvoker productionCDaemon(
void onTimeout(const boost::system::error_code& error); std::exception_ptr &exceptionPtr,
std::function<void()> 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: public:
std::shared_ptr<device::DeviceAttachmentSpec> deviceAttachmentSpec; std::shared_ptr<device::DeviceAttachmentSpec> deviceAttachmentSpec;
@@ -107,14 +105,6 @@ public:
private: private:
boost::asio::io_context& ioContext; 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 } // namespace stim_buff
libspinscale
+1 -1
Submodule libspinscale updated: b04b0db155...656aae37c8
smocore/deviceManager/deviceReattacher.cpp
+49 -122
View File
@@ -1,161 +1,88 @@
#include <config.h> #include <config.h>
#include <chrono>
#include <iostream> #include <iostream>
#include <functional> #include <functional>
#include <componentThread.h> #include <componentThread.h>
#include <adapters/boostAsio/deadlineTimerAReq.h>
#include <deviceManager/deviceReattacher.h> #include <deviceManager/deviceReattacher.h>
#include <deviceManager/deviceManager.h> #include <deviceManager/deviceManager.h>
#include <marionette/marionetteThread.h>
#include <spinscale/co/nonViralCompletion.h> #include <spinscale/co/nonViralCompletion.h>
namespace smo { namespace smo {
namespace device { namespace device {
namespace {
constexpr unsigned int reattachInFlightStaleThresholdMultiplier = 4;
} // namespace
DeviceReattacher::DeviceReattacher( DeviceReattacher::DeviceReattacher(
DeviceManager& parent, std::shared_ptr<sscl::ComponentThread> ioThread) DeviceManager& parent, std::shared_ptr<sscl::ComponentThread> ioThread)
: parent(parent), ioThread(ioThread), timer(ioThread->getIoContext()) : parent(parent),
ioThread(ioThread), daemonTimer(ioThread->getIoContext())
{ {
/** EXPLANATION: /** EXPLANATION:
* The thread on which DeviceReattacher runs is whichever thread executes * deviceReattacherCDaemon is a dynamic posting non-viral coroutine: start()
* the io_context that owns deadline_timer. Timer async_wait handlers * passes ExplicitPostTarget{ioThread->getIoContext()} so the daemon body
* (onTimeout, holdReattachCReq, reattachKnownListCReq) are dispatched on * always runs on ioThread. daemonTimer is reused each loop iteration.
* that thread. ioThread selects that io_context here; start() only arms
* the timer on it.
*/ */
} }
mrntt::MrnttNonViralNonPostingInvoker DeviceReattacher::reattachKnownListCReq( sscl::co::DynamicNonViralPostingInvoker
DeviceReattacher::deviceReattacherCDaemon(
[[maybe_unused]] sscl::co::ExplicitPostTarget postTarget,
[[maybe_unused]] std::exception_ptr &exceptionPtr, [[maybe_unused]] std::exception_ptr &exceptionPtr,
[[maybe_unused]] std::function<void()> callback) [[maybe_unused]] std::function<void()> callback,
sscl::SyncCancelerForAsyncWork &canceler)
{ {
/** EXPLANATION: boost::asio::io_context &timerIoContext =
* Non-posting: invoked from holdReattachCReq on the timer callback thread sscl::ComponentThread::getSelf()->getIoContext();
* (see ctor). Nested DeviceManager attach APIs still post to MRNTT as const auto periodMs = boost::posix_time::milliseconds(
* needed via their own viral posting invokers. CONFIG_MRNTT_DEVMGR_REATTACHER_PERIOD_MS);
*/
co_await parent.attachAllUnattachedDevicesFromKnownListCReq(); while (!canceler.isCancellationRequested())
{
const bool expiredNormally = co_await
adapters::boostAsio::getDeadlineTimerAReqAwaiter(
timerIoContext, daemonTimer, periodMs);
if (!expiredNormally) {
break;
}
co_await parent.attachAllUnattachedDevicesFromKnownListCReq();
}
co_return; co_return;
} }
void DeviceReattacher::start() void DeviceReattacher::start()
{ {
deviceReattacherCanceler.startAcceptingWork(); taskNursery.openAdmission();
scheduleNextTimeout(); taskNursery.launch(
} [this](sscl::co::NonViralTaskNursery::Slot::Lease &lease)
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]()
{ {
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()) if (nvc.hasException())
{ {
try { try {
nvc.checkAndRethrowException(); nvc.checkAndRethrowException();
} catch (const std::exception &e) { } catch (const std::exception &e) {
std::cerr << "DeviceReattacher: " << e.what() std::cerr << "DeviceReattacher: "
<< std::endl; << 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 daemonTimer.cancel();
if (error == boost::asio::error::operation_aborted) { taskNursery.requestCancelOnAll();
return; taskNursery.closeAdmission();
} taskNursery.syncAwaitAllSettlements(ioThread->getIoContext());
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<std::chrono::milliseconds>(
elapsedSinceLastReattachReq).count()
<< "ms (threshold "
<< staleThreshold.count()
<< "ms); forcing a new reattach request."
<< std::endl;
holdReattachCReq();
}
}
// Schedule the next timeout
scheduleNextTimeout();
} }
} // namespace device } // namespace device
smocore/include/deviceManager/deviceReattacher.h
+10 -18
View File
@@ -1,14 +1,12 @@
#ifndef DEVICEREATTACHER_H #ifndef DEVICEREATTACHER_H
#define DEVICEREATTACHER_H #define DEVICEREATTACHER_H
#include <atomic>
#include <chrono>
#include <exception>
#include <functional> #include <functional>
#include <memory> #include <memory>
#include <optional>
#include <boost/asio/deadline_timer.hpp> #include <boost/asio/deadline_timer.hpp>
#include <marionette/marionetteThread.h> #include <spinscale/co/dynamicPostingInvoker.h>
#include <spinscale/co/nonViralTaskNursery.h>
#include <spinscale/co/postTarget.h>
#include <spinscale/syncCancelerForAsyncWork.h> #include <spinscale/syncCancelerForAsyncWork.h>
namespace smo { namespace smo {
@@ -32,23 +30,17 @@ public:
void stop(); void stop();
private: private:
void scheduleNextTimeout(); sscl::co::DynamicNonViralPostingInvoker deviceReattacherCDaemon(
void onTimeout(const boost::system::error_code& error); sscl::co::ExplicitPostTarget postTarget,
void holdReattachCReq();
mrntt::MrnttNonViralNonPostingInvoker reattachKnownListCReq(
std::exception_ptr &exceptionPtr, std::exception_ptr &exceptionPtr,
std::function<void()> callback); std::function<void()> callback,
sscl::SyncCancelerForAsyncWork &canceler);
DeviceManager &parent; DeviceManager &parent;
// io_context thread for timer and non-posting reattach shell (see ctor). // io_context thread for timer (see ctor).
std::shared_ptr<sscl::ComponentThread> ioThread; std::shared_ptr<sscl::ComponentThread> ioThread;
sscl::SyncCancelerForAsyncWork deviceReattacherCanceler; sscl::co::NonViralTaskNursery taskNursery;
boost::asio::deadline_timer timer; boost::asio::deadline_timer daemonTimer;
std::exception_ptr reattachLifetimeExceptionPtr;
std::optional<mrntt::MrnttNonViralNonPostingInvoker> reattachCReqInvoker;
bool reattachOpInFlight = false;
std::chrono::steady_clock::time_point lastReattachReqTimestamp{};
}; };
} // namespace device } // namespace device
smocore/include/marionette/marionette.h
+6 -8
View File
@@ -9,6 +9,7 @@
#include <memory> #include <memory>
#include <optional> #include <optional>
#include <spinscale/component.h> #include <spinscale/component.h>
#include <spinscale/co/nonViralTaskNursery.h>
#include <marionette/marionetteThread.h> #include <marionette/marionetteThread.h>
namespace sscl { namespace sscl {
@@ -29,8 +30,10 @@ public:
{} {}
~MarionetteComponent() = default; ~MarionetteComponent() = default;
void holdInitializeCReq(std::function<void()> completion); void holdInitializeCReq(
void holdFinalizeCReq(std::function<void()> completion); std::function<void(std::exception_ptr &exceptionPtr)> completion);
void holdFinalizeCReq(
std::function<void(std::exception_ptr &exceptionPtr)> completion);
MrnttNonViralPostingInvoker initializeCReq( MrnttNonViralPostingInvoker initializeCReq(
std::exception_ptr &exceptionPtr, std::exception_ptr &exceptionPtr,
@@ -58,12 +61,7 @@ protected:
private: private:
std::unique_ptr<boost::asio::signal_set> signals; std::unique_ptr<boost::asio::signal_set> signals;
bool callShutdownSalmanoff = false; bool callShutdownSalmanoff = false;
std::optional<MrnttNonViralPostingInvoker> initializeCReqInvoker; sscl::co::NonViralTaskNursery taskNursery;
std::optional<MrnttNonViralPostingInvoker> finalizeCReqInvoker;
public:
std::exception_ptr initializeLifetimeExceptionPtr;
std::exception_ptr finalizeLifetimeExceptionPtr;
}; };
extern std::shared_ptr<sscl::PuppeteerThread> thread; extern std::shared_ptr<sscl::PuppeteerThread> thread;
smocore/marionette/lifetime.cpp
+22 -21
View File
@@ -30,33 +30,29 @@ void assertMarionetteThread()
} // namespace } // namespace
void MarionetteComponent::holdInitializeCReq( void MarionetteComponent::holdInitializeCReq(
std::function<void()> completion) std::function<void(std::exception_ptr &exceptionPtr)> completion)
{ {
initializeLifetimeExceptionPtr = nullptr; taskNursery.launch(
initializeCReqInvoker.emplace(initializeCReq( [this](sscl::co::NonViralTaskNursery::Slot::Lease &lease)
initializeLifetimeExceptionPtr,
[completion = std::move(completion)]()
{ {
sscl::co::NonViralCompletion nvc( return initializeCReq(
mrntt.initializeLifetimeExceptionPtr); lease.getExceptionStorage(),
nvc.checkAndRethrowException(); lease.getCallerLambda());
completion(); },
})); std::move(completion));
} }
void MarionetteComponent::holdFinalizeCReq( void MarionetteComponent::holdFinalizeCReq(
std::function<void()> completion) std::function<void(std::exception_ptr &exceptionPtr)> completion)
{ {
finalizeLifetimeExceptionPtr = nullptr; taskNursery.launch(
finalizeCReqInvoker.emplace(finalizeCReq( [this](sscl::co::NonViralTaskNursery::Slot::Lease &lease)
finalizeLifetimeExceptionPtr,
[completion = std::move(completion)]()
{ {
sscl::co::NonViralCompletion nvc( return finalizeCReq(
mrntt.finalizeLifetimeExceptionPtr); lease.getExceptionStorage(),
nvc.checkAndRethrowException(); lease.getCallerLambda());
completion(); },
})); std::move(completion));
} }
MrnttNonViralPostingInvoker MarionetteComponent::initializeCReq( MrnttNonViralPostingInvoker MarionetteComponent::initializeCReq(
@@ -107,7 +103,12 @@ void MarionetteComponent::exceptionInd()
[] []
{ {
mrntt.holdFinalizeCReq( mrntt.holdFinalizeCReq(
[]() { marionetteFinalizeReqCb(true); }); [](std::exception_ptr &exceptionPtr)
{
sscl::co::NonViralCompletion nvc(exceptionPtr);
nvc.checkAndRethrowException();
marionetteFinalizeReqCb(true);
});
}); });
} }
smocore/marionette/main.cpp
+24 -3
View File
@@ -5,6 +5,7 @@
#include <spinscale/component.h> #include <spinscale/component.h>
#include <spinscale/componentThread.h> #include <spinscale/componentThread.h>
#include <marionette/marionette.h> #include <marionette/marionette.h>
#include <spinscale/co/nonViralCompletion.h>
#include <spinscale/runtime.h> #include <spinscale/runtime.h>
#include <componentThread.h> #include <componentThread.h>
#include <mindManager/mindManager.h> #include <mindManager/mindManager.h>
@@ -33,7 +34,12 @@ void marionetteInitializeReqCb(bool success)
<< '\n'; << '\n';
mrntt.holdFinalizeCReq( mrntt.holdFinalizeCReq(
[]() { marionetteFinalizeReqCb(true); }); [](std::exception_ptr &exceptionPtr)
{
sscl::co::NonViralCompletion nvc(exceptionPtr);
nvc.checkAndRethrowException();
marionetteFinalizeReqCb(true);
});
} }
void marionetteFinalizeReqCb(bool success) void marionetteFinalizeReqCb(bool success)
@@ -82,7 +88,12 @@ void MarionetteComponent::postJoltHook()
break; break;
} }
mrntt.holdFinalizeCReq( 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(); smo::initializeSalmanoff();
callShutdownSalmanoff = true; callShutdownSalmanoff = true;
taskNursery.openAdmission();
holdInitializeCReq( holdInitializeCReq(
[] { marionetteInitializeReqCb(true); }); [](std::exception_ptr &exceptionPtr)
{
sscl::co::NonViralCompletion nvc(exceptionPtr);
nvc.checkAndRethrowException();
marionetteInitializeReqCb(true);
});
std::cout << "PuppeteerThread::main: Entering event loop" << "\n"; std::cout << "PuppeteerThread::main: Entering event loop" << "\n";
} }
void MarionetteComponent::postLoopHook() void MarionetteComponent::postLoopHook()
{ {
taskNursery.requestCancelOnAll();
taskNursery.closeAdmission();
std::cout << "PuppeteerThread::main: Exited event loop" << "\n"; std::cout << "PuppeteerThread::main: Exited event loop" << "\n";
} }
stimBuffApis/livoxGen1/livoxGen1.cpp
+8 -2
View File
@@ -355,8 +355,11 @@ attachByCreatingProducer(
* may not yet be ready for another command. * may not yet be ready for another command.
*/ */
// Initialize timer with LivoxGen1 metadata io_context // Initialize timer with LivoxGen1 metadata io_context
boost::asio::deadline_timer commandDelayTimer(
componentThread->getIoContext());
const bool delayOk = co_await adapters::boostAsio::getDeadlineTimerAReqAwaiter( const bool delayOk = co_await adapters::boostAsio::getDeadlineTimerAReqAwaiter(
componentThread->getIoContext(), componentThread->getIoContext(),
commandDelayTimer,
boost::posix_time::milliseconds(LIVOX_GEN1_DEVICE_COMMAND_DELAY_MS)); boost::posix_time::milliseconds(LIVOX_GEN1_DEVICE_COMMAND_DELAY_MS));
if (!delayOk) if (!delayOk)
@@ -540,10 +543,13 @@ livoxGen1_detachDeviceCReq(
// Add 5ms delay before destroying device // Add 5ms delay before destroying device
// Helper method to delay and then call destroyDeviceReq // Helper method to delay and then call destroyDeviceReq
// Initialize timer with LivoxGen1 metadata io_context // Initialize timer with LivoxGen1 metadata io_context
boost::asio::deadline_timer commandDelayTimer(
requestComponentThread->getIoContext());
co_await adapters::boostAsio::getDeadlineTimerAReqAwaiter( co_await adapters::boostAsio::getDeadlineTimerAReqAwaiter(
requestComponentThread->getIoContext(), requestComponentThread->getIoContext(),
commandDelayTimer,
boost::posix_time::milliseconds(LIVOX_GEN1_DEVICE_COMMAND_DELAY_MS)); boost::posix_time::milliseconds(LIVOX_GEN1_DEVICE_COMMAND_DELAY_MS));
// No other buffers - stop and remove StimProducer // No other buffers - stop and remove StimProducer
stimBuffApis/livoxGen1/pcloudStimulusProducer.cpp
+16 -32
View File
@@ -428,11 +428,6 @@ PcloudStimulusProducer::getOrCreateAttachedStimulusBuffer(
} }
} }
void PcloudStimulusProducer::stimFrameProductionTimesliceInd()
{
holdProduceFrameCReq();
}
struct AllowNextStimulusFrameGuard struct AllowNextStimulusFrameGuard
{ {
PcloudStimulusProducer& producer; PcloudStimulusProducer& producer;
@@ -445,22 +440,14 @@ struct AllowNextStimulusFrameGuard
{ producer.allowNextStimulusFrame(); } { producer.allowNextStimulusFrame(); }
}; };
void PcloudStimulusProducer::holdProduceFrameCReq() /** EXPLANATION:
{ * productionCDaemon co_awaits this viral timeslice and passes its
/** EXPLANATION: * nursery-slot canceler. Cooperative shutdown uses that canceler between
* We shouldn't acquire stimulusProducerCanceler.s.lock here because * co_await steps; do not use execUncancelableSegmentOrAbort here.
* this function is called from */
* StimulusProducer::stimFrameProductionTimesliceInd(), which is already sscl::co::ViralNonPostingInvoker<void>
* holding the lock. PcloudStimulusProducer::stimFrameProductionTimesliceCInd(
*/ sscl::SyncCancelerForAsyncWork &canceler)
activeProduceFrameInvoker.emplace(produceFrameCReq(
produceFrameExceptionPtr,
[this]() { activeProduceFrameInvoker.reset(); }));
}
sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq(
[[maybe_unused]] std::exception_ptr& exceptionPtr,
[[maybe_unused]] std::function<void()> completion)
{ {
AllowNextStimulusFrameGuard allowNextGuard(*this); AllowNextStimulusFrameGuard allowNextGuard(*this);
StimulusFrame& stimulusFrame = tempStimulusFrame; StimulusFrame& stimulusFrame = tempStimulusFrame;
@@ -473,14 +460,11 @@ sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq(
// produceFrameReq1_doAssemble_posted // produceFrameReq1_doAssemble_posted
/** EXPLANATION: /** EXPLANATION:
* stimFrameProductionTimesliceInd() is entered with * productionCDaemon co_awaits this timeslice outside its uncancelable
* stimulusProducerCanceler.s.lock held; do not re-acquire here. * rate-limiter segment; use isCancellationRequested() here, not
* * execUncancelableSegmentOrAbort.
* This function is called from
* StimulusProducer::stimFrameProductionTimesliceInd(), whose caller is
* already holding the lock.
*/ */
if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) if (canceler.isCancellationRequested())
{ co_return; } { co_return; }
cpsBoundary::AssembleFrameResult assembleResult = co_await cpsBoundary::AssembleFrameResult assembleResult = co_await
@@ -491,8 +475,8 @@ sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq(
std::optional<AmbienceProductionDesc> lightAmbienceProductionDescDesc; std::optional<AmbienceProductionDesc> lightAmbienceProductionDescDesc;
std::optional<AmbienceProductionDesc> darkAmbienceProductionDescDesc; std::optional<AmbienceProductionDesc> darkAmbienceProductionDescDesc;
{ {
sscl::SpinLock::Guard guard(stimulusProducerCanceler.s.lock); sscl::SpinLock::Guard guard(canceler.s.lock);
if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) if (canceler.isCancellationRequestedUnlocked())
{ co_return; } { co_return; }
if (!assembleResult.success) if (!assembleResult.success)
@@ -612,8 +596,8 @@ sscl::co::NonViralNonPostingInvoker PcloudStimulusProducer::produceFrameCReq(
} }
{ {
sscl::SpinLock::Guard guard(stimulusProducerCanceler.s.lock); sscl::SpinLock::Guard guard(canceler.s.lock);
if (stimulusProducerCanceler.isCancellationRequestedUnlocked()) if (canceler.isCancellationRequestedUnlocked())
{ co_return; } { co_return; }
if (!compactCollateSuccess) { if (!compactCollateSuccess) {
stimBuffApis/livoxGen1/pcloudStimulusProducer.h
+3 -10
View File
@@ -82,13 +82,9 @@ public:
const std::shared_ptr<StimulusBuffer>& buffer) override; const std::shared_ptr<StimulusBuffer>& buffer) override;
protected: protected:
void stimFrameProductionTimesliceInd() override; sscl::co::ViralNonPostingInvoker<void>
stimFrameProductionTimesliceCInd(
void holdProduceFrameCReq(); sscl::SyncCancelerForAsyncWork &canceler) override;
sscl::co::NonViralNonPostingInvoker produceFrameCReq(
std::exception_ptr& exceptionPtr,
std::function<void()> completion);
public: public:
size_t nDgramsPerStagingBufferFrame; size_t nDgramsPerStagingBufferFrame;
@@ -112,9 +108,6 @@ public:
lightAmbienceStimulusBuffer; lightAmbienceStimulusBuffer;
std::atomic<std::shared_ptr<PcloudDarkAmbienceStimulusBuffer>> std::atomic<std::shared_ptr<PcloudDarkAmbienceStimulusBuffer>>
darkAmbienceStimulusBuffer; darkAmbienceStimulusBuffer;
std::optional<sscl::co::NonViralNonPostingInvoker> activeProduceFrameInvoker;
std::exception_ptr produceFrameExceptionPtr;
}; };
} // namespace stim_buff } // namespace stim_buff
tests/libspinscale/nonViralTaskNursery_tests.cpp
+28 -8
View File
@@ -139,7 +139,9 @@ TEST_F(NonViralTaskNurseryTest, SetOnSettledHookRejectsAfterFillSlot)
lease.getCallerLambda()); lease.getCallerLambda());
}); });
EXPECT_THROW(lease.setOnSettledHook([]() {}), std::runtime_error); EXPECT_THROW(
lease.setOnSettledHook([](std::exception_ptr &) {}),
std::runtime_error);
lease.commit(); lease.commit();
} }
@@ -322,6 +324,26 @@ TEST_F(NonViralTaskNurseryTest, LaunchSugar)
EXPECT_TRUE(nursery.allSettled()); EXPECT_TRUE(nursery.allSettled());
} }
TEST_F(NonViralTaskNurseryTest, LaunchWithOnSettledHook)
{
std::atomic<bool> 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) TEST_F(NonViralTaskNurseryTest, HandleStability)
{ {
auto handle = nursery.launch( auto handle = nursery.launch(
@@ -520,7 +542,7 @@ TEST_F(NonViralTaskNurseryTest, OnSettledHookRunsAtRetirement)
auto lease = nursery.getNewSlotLease(); auto lease = nursery.getNewSlotLease();
lease.setOnSettledHook( lease.setOnSettledHook(
[&hookRan]() [&hookRan](std::exception_ptr &)
{ {
hookRan.store(true, std::memory_order_release); hookRan.store(true, std::memory_order_release);
}); });
@@ -536,13 +558,14 @@ TEST_F(NonViralTaskNurseryTest, OnSettledHookRunsAtRetirement)
EXPECT_TRUE(hookRan.load(std::memory_order_acquire)); EXPECT_TRUE(hookRan.load(std::memory_order_acquire));
} }
TEST_F(NonViralTaskNurseryTest, OnSettledHookExceptionStillRetiresSlot) TEST_F(NonViralTaskNurseryTest, OnSettledHookSeesRetiredSlot)
{ {
auto lease = nursery.getNewSlotLease(); auto lease = nursery.getNewSlotLease();
lease.setOnSettledHook( 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.fillSlot(
[&lease]() [&lease]()
@@ -552,9 +575,6 @@ TEST_F(NonViralTaskNurseryTest, OnSettledHookExceptionStillRetiresSlot)
lease.getCallerLambda()); lease.getCallerLambda());
}); });
lease.commit(); lease.commit();
EXPECT_TRUE(nursery.allSettled());
EXPECT_EQ(nursery.unsettledCount(), 0U);
} }
TEST_F(NonViralTaskNurseryTest, DuplicateRetireThrows) TEST_F(NonViralTaskNurseryTest, DuplicateRetireThrows)
tests/smocore/qutex_tests.cpp
+1
View File
@@ -25,6 +25,7 @@ public:
} }
void awaken(bool forceAwaken = false) override { void awaken(bool forceAwaken = false) override {
(void)forceAwaken;
awakened = true; awakened = true;
} }
}; };