Compare commits

...
2 Commits
Author SHA1 Message Date
latentprionandCursor 21fa125d52 Summarize and merge MultiOperationResultSetWithException from settled Groups.
Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-14 05:06:18 -04:00
latentprionandCursor 3156652257 Document closeAdmission-before-cancel nursery stop ordering.
Explain the admit-after-cancel race and that daemons should disconnect at
the protocol layer before sealing admission (or handle closed-admission on enqueue).

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-07-13 19:02:52 -04:00
5 changed files with 95 additions and 7 deletions
+1
View File
@@ -87,6 +87,7 @@ add_library(spinscale SHARED
src/puppetApplication.cpp
src/runtime.cpp
src/callableTracer.cpp
src/multiOperationResultSet.cpp
)
set_target_properties(spinscale PROPERTIES
+6 -4
View File
@@ -202,17 +202,19 @@ nursery.launch(
nvc.checkAndRethrowException();
});
nursery.requestCancelOnAll();
nursery.closeAdmission();
nursery.requestCancelOnAll();
nursery.syncAwaitAllSettlements(
sscl::ComponentThread::getSelf()->getIoContext());
```
Each slot owns a `SyncCancelerForAsyncWork`. `requestCancelOnAll()` only signals
cooperative stop; it does not destroy invokers. Invokers are retired when their
completion callbacks run. Call `closeAdmission()` explicitly before
`asyncAwaitAllSettlements()` or `syncAwaitAllSettlements()`; those APIs wait until
all slots have retired naturally and throw if admission is still open.
completion callbacks run. Call `closeAdmission()` before `requestCancelOnAll()`
so no new work can be admitted after cancel begins, and call `closeAdmission()`
explicitly before `asyncAwaitAllSettlements()` or `syncAwaitAllSettlements()`;
those APIs wait until all slots have retired naturally and throw if admission is
still open.
`syncAwaitAllSettlements()` runs a nested `io_context` loop on the **calling
thread** (it blocks in `run_one()` until every slot has retired). Pass the
+30 -3
View File
@@ -55,9 +55,6 @@ struct MemberInvoker : MemberInvokerBase
* nursery member. The external submitter should add the complete flow to the
* nursery and then return; the nursery owns that flow until the flow settles.
*
* Call closeAdmission() explicitly before asyncAwaitAllSettlements() or
* syncAwaitAllSettlements().
*
* syncAwaitAllSettlements() runs a nested io_context loop on the calling
* thread (AsynchronousBridge). Pass the calling thread's io_context —
* typically
@@ -240,6 +237,36 @@ public:
s.rsrc.admissionOpen = true;
}
/** EXPLANATION:
* Stopping a nursery: always closeAdmission() before
* requestCancelOnAll(). requestCancelOnAll() only marks currently
* ACTIVE_UNSETTLED slots; it does not refuse new leases. If cancel
* runs while admission is still open, a concurrent submitter can still
* getNewSlotLease() / launch() after cancel has fanned out, and that
* newly admitted work will not have been cancelled — it races past the
* stop wave and keeps the drain from reaching "all settled" until it
* finishes on its own (or a later cancel). Closing admission first
* seals the nursery so cancel applies to a fixed membership set, then
* drain with asyncAwaitAllSettlements() / syncAwaitAllSettlements()
* (those APIs also require admission already closed).
*
* Preferred stop stack for a daemon/service that enqueues request
* coroutines into the nursery: keep protocol "stop listening /
* disconnect / refuse new connections and requests" separate from
* protocol state destruction. Stop accepting at the protocol level
* first, then nursery closeAdmission(), then requestCancelOnAll(),
* then cancel any awaited I/O owned outside the cancelers, then drain,
* then destroy protocol state. That ordering stops new work at the
* source before admission is sealed.
*
* If the daemon/service cannot disconnect/stop listening separately
* from destruction, the spinscale-using embedding project must handle
* closed-admission failures when it tries to enqueue. For example,
* catch the "admission closed" throw around nursery.launch() (or in
* the factory that calls it) and emit a protocol-specific failure such
* as "connection failed" or "request timed out" instead of letting the
* exception escape the accept/request path unbounded.
*/
void closeAdmission()
{
sscl::SpinLock::Guard guard(s.lock);
@@ -4,6 +4,9 @@
#include <exception>
namespace sscl {
namespace co {
struct Group;
} // namespace co
/** Plain aggregate for fan-out / fan-in results returned from coroutines. */
struct MultiOperationResultSet
@@ -38,9 +41,29 @@ struct MultiOperationResultSetWithException
memberFailureException(memberFailureExceptionIn)
{}
/** Summarize a settled Group into counts + aggregated member failure. */
explicit MultiOperationResultSetWithException(const co::Group &group);
bool hasMemberFailure() const
{ return memberFailureException != nullptr; }
/** Combine this result set with another phase's counts and exception. */
MultiOperationResultSetWithException mergeWith(
const MultiOperationResultSetWithException &other) const
{
std::exception_ptr memberFailure = memberFailureException;
if (!memberFailure && other.hasMemberFailure()) {
memberFailure = other.memberFailureException;
}
return MultiOperationResultSetWithException(
MultiOperationResultSet(
results.nTotal + other.results.nTotal,
results.nSucceeded + other.results.nSucceeded,
results.nFailed + other.results.nFailed),
memberFailure);
}
MultiOperationResultSet results;
std::exception_ptr memberFailureException = nullptr;
};
+35
View File
@@ -0,0 +1,35 @@
#include <boostAsioLinkageFix.h>
#include <spinscale/multiOperationResultSet.h>
#include <spinscale/co/group.h>
namespace sscl {
MultiOperationResultSetWithException::MultiOperationResultSetWithException(
const co::Group &group)
{
unsigned int nSucceeded = 0;
unsigned int nFailed = 0;
using SettlementType = co::Group::SettlementDescriptor::TypeE;
for (const auto &desc : group.s.rsrc.settlements)
{
if (desc.type == SettlementType::EXCEPTION_THROWN) {
nFailed++;
}
else {
nSucceeded++;
}
}
results = MultiOperationResultSet(
static_cast<unsigned int>(group.s.rsrc.settlements.size()),
nSucceeded,
nFailed);
if (nFailed > 0) {
memberFailureException = group.captureAggregatedGroupExceptions();
}
}
} // namespace sscl