Cancelling Work and Shutting Down
Shutdown is complete when work no longer needs callbacks and no running thread can access the resources you release. Requesting cancellation is often only the first step. This page explains how to stop producing work, finish or abandon outstanding operations, keep required completion paths running, and finally end and join their threads.
The ownership rules in Keeping Callbacks and Their Owners Alive are the foundation. For the examples, follow either the application-part design in Using Event Loops in Application Parts or the worker design in Processing Work on Event Threads; you do not need both architectures in one application.
Resource Completion Comes Before Loop Termination
Consider a server that has stopped accepting new work but still has committed responses to send. Its close operation needs the owner’s loop to process network events and deliver a final notification. Ending that loop first would remove the mechanism that finishes the close.
A useful shutdown sequence therefore follows the resource’s dependencies:
Stop accepting application requests and stop producers that could create more operations.
Ask active operations to finish or cancel, keeping their required reply targets running.
Observe the required completions, then release sources, timers, and subscriptions on their owner loops.
End the loops whose work is complete, join their threads from another thread, and release the remaining owners.
An application-level stopping state is useful during the first steps. It prevents new requests while leaving loop admission open for close commands and final callbacks. Closing the loop’s queue is a later decision. With multiple resources, count the terminal operations explicitly; the first final callback does not prove that every resource has finished.
The file-server application demonstrates this sequence through an application part.
Its _server member retains the HTTP source for the entire close operation.
During initialization, it installs the error and final handlers before starting that source.
When shutdown begins, the framework calls stopping() on the part’s own loop.
The excerpts below show the retained member, the end of the handler chain, and the shutdown hooks.
event/FileServer/ServerPart.hpp
el::HttpServerPtr _server; ///< Event source owned by this part's loop.
.onError([](const el::NetworkErrorContext &error) -> void { throw el::NetworkError{error}; })
.onFinal([this]() -> void {
if (state() == el::ApplicationPartState::Stopping) {
completeStopping();
}
});
void stopping() override {
// Allow committed responses to drain before ending this part's event loop.
if (_server == nullptr || _server->state() == el::NetworkSourceState::Closed ||
_server->state() == el::NetworkSourceState::Failed) {
completeStopping();
} else {
_server->close();
}
}
void cleanup() noexcept override {
if (_server != nullptr) {
_server->abort();
_server.reset();
}
}
HttpServer::close() begins graceful closure.
Returning from stopping() lets the loop continue dispatching.
Only the final handler calls completeStopping() after
asynchronous shutdown reaches its terminal notification.
When the source is already absent, closed, or failed, the hook completes immediately because no close work remains.
cleanup() runs after dispatch ends.
It performs unconditional release and does not begin another operation that needs callbacks.
The fallback abort() also makes cleanup safe after partial initialization or failure.
For this source’s close and abort guarantees, see Using HTTP Servers.
For registration, readiness, and the complete lifecycle example, see Using Event Loops in Application Parts.
Cancellation Belongs to the Operation
There is no universal meaning of “cancel all event work”. A timer, observer registration, network source, loop, and application each have a different scope:
Action |
Purpose |
|---|---|
Operation cancellation |
Abandon one operation according to its source’s contract; completion notifications may differ by API. |
Graceful source closure |
Finish the source’s accepted work and observe its required terminal notification while its loop still runs. |
Loop |
Close queue admission and drain already accepted queued work. |
Loop |
End the current dispatch run without draining its waiting queue. |
Application |
Request application-wide shutdown, including the managed part and thread lifecycle. |
Stopping a retained timer prevents further scheduled activity according to the timer’s contract. Cancelling a subscription prevents future observer invocations, but a callback already executing may still finish. Neither action proves that another thread has stopped accessing a borrowed recipient. See Keeping Callbacks and Their Owners Alive for that distinction and for the lookup example whose cancellation suppresses its result and error callbacks while still allowing a final notification.
A workflow can also become logically cancelled before its underlying work ends. Mark it terminal, stop its retry and deadline timers, and reject obsolete replies by request identity. That prevents an abandoned result from changing current state; it does not interrupt a running computation. Communicating Between Event Loops applies identity checks across thread boundaries.
For lengthy computation, cancellation must be cooperative. Divide the work into bounded steps so the worker can process a queued cancellation command between steps, or use an explicitly synchronized cancellation signal checked by the computation. A cancellation command on the same loop cannot run while the current callback is still calculating or blocked in I/O. The I/O operation also needs its own timeout or cancellation facility if it can block indefinitely.
What Graceful Loop Quit Drains
EventLoop::quit() closes admission and lets already accepted queued
callbacks finish.
It stops backend polling; it does not wait for every timer deadline or native completion that might eventually produce
an event.
A callback that runs during draining cannot rely on posting another accepted callback to that loop.
This small demo queues two satellite observations, requests quit, and then tries to queue a third. It runs the loop only after all three submissions so the admission boundary is unambiguous:
event/EventShutdown/DrainAcceptedWork.cpp
/// Finish accepted satellite observations before ending dispatch; reject later work.
void drainAcceptedWork() {
const auto loop = el::EventLoop::create(el::EventLoopOptions{}.setErrorPolicy(el::EventLoopErrorPolicy::Quit));
auto observations = el::StringList{};
loop->invoke([&observations]() -> void { observations.append("Molntäcke"_el); });
loop->invoke([&observations]() -> void { observations.append("Havsis"_el); });
loop->quit();
loop->invoke([&observations]() -> void { observations.append("Snödjup"_el); });
loop->run();
if (const auto error = loop->takeError()) {
std::rethrow_exception(error);
}
if (observations != el::StringList{"Molntäcke"_el, "Havsis"_el}) {
throw el::LogicFailure{"Quit did not preserve exactly the accepted observations."_el};
}
for (const auto &observation : observations) {
el::io::printLine("Recorded: "_el, observation);
}
}
$ event/event_shutdown --demo DrainAcceptedWork
Recorded: Molntäcke
Recorded: Havsis
The verified result contains exactly the first two observations, in order.
The later invocation is discarded; returning from invoke() is not an acknowledgement that a callback will execute.
The captured vector is safe here because all dispatch happens on the calling thread and run() returns before the
vector leaves scope.
This local loop isolates queue admission from application and resource shutdown.
In a request/reply exchange, quitting the receiver as soon as the last command is sent is too early.
The workers have not necessarily posted their replies yet.
Keep the receiver open until required replies are accounted for, then finish the workers and receiver in an order that
preserves any remaining dependencies.
The preview demo waits for all of its replies before asking the application to quit.
Likewise, the server above calls completeStopping() after source closure, not immediately after requesting it.
An immediate stop() ends dispatch without draining queued callbacks.
It does not interrupt the callback already running, and it does not mark waiting application tasks as cancelled.
If those callbacks would have reported completion, your shutdown logic must account for their abandonment separately.
For a directly controlled loop, stopping a run is also distinct from permanently closing admission with quit(); it
is not a resource-cleanup protocol.
Failure and Shutdown Deadlines
A graceful close can fail or take too long. Decide how the owner transitions to failure and releases resources even when the expected successful completion never arrives. Keep any deadline on a loop that remains able to dispatch until that decision is made. Once a deadline wins, mark the shutdown failed, use the source’s abort or release path, and do not continue waiting for callbacks that path suppresses. The application may then end the affected loops and join their threads.
The file server’s error handler throws so a source failure reaches the part manager.
The default manager policy stops the part graph; cleanup() still releases the server safely.
The manager also bounds graceful stopping, with a default timeout of 30 seconds per part.
A timeout records a failure and requests loop termination.
These lifecycle policies and their configuration are covered in Designing Application Parts.
A deadline does not forcibly interrupt C++ code. A timer on a blocked loop cannot dispatch at all, and a timeout detected on another loop still cannot make an unbounded callback return. Therefore callbacks, lifecycle hooks, and blocking operations need their own bounded or cooperative behavior. A worker join can otherwise remain blocked even after shutdown has been requested.
Ending and Joining Worker Threads
A worker’s quit() is an asynchronous, idempotent request to end its
loop gracefully.
join() waits for thread completion; it does not ask the thread to stop.
Request termination only after the worker’s required asynchronous operations are complete, then join from an owner
thread whose dispatch is no longer needed by that worker.
For a worker whose only remaining work is already queued and needs no further callbacks, the final owner-side sequence can be as short as:
worker->quit();
worker->join();
worker.reset();
The demo in Processing Work on Event Threads retains its application-managed workers until application shutdown
requests their termination and joins them.
Their registration does not replace that retained owner.
An UnmanagedEventThread has no application registration: its owner
must arrange the whole shutdown sequence explicitly.
The same quit-before-join and callback-lifetime rules apply when you own execution directly.
Joining a worker requires another thread: a callback on the worker cannot wait for its own thread to finish. Destroying the final owning thread pointer also requests termination and joins a started worker, so release that final pointer on another thread. A worker callback that captures the only remaining owning pointer to its own thread can otherwise cause destruction on that worker as the callback is released. Keeping an owner-side handle through the join avoids that trap.
Event threads are one-shot objects.
After start() succeeds, they cannot be restarted, even after joining.
isStarted() stays true after completion; isRunning() is only a snapshot of current execution.
Neither accessor is a startup or completion synchronization barrier.
Use a ready message for application readiness and a join for completed thread execution.
Quit may be requested before startup; if started afterward, the loop drains previously accepted work and ends.
Joining before startup or after a previous join is a no-op.
The precise lifecycle contract is in Event System.
A join does not transport worker exceptions.
Choose an error policy and a failure completion path before starting work; the managed-worker tutorial explains
FailApplication and its application diagnostic boundary.
An exception escaping an event thread terminates the process rather than becoming a result of join().
Application-Wide Shutdown
Use Application::quit() when the application should end, after the
resources you manage directly no longer need its loops.
For application parts, the framework provides the earlier coordination point through stopping() and
completeStopping() so sources can close before those loops end.
Keep dependency ordering and service shutdown in that framework instead of manually quitting each part’s loop.
Designing Application Parts explains dependency-aware stopping and cleanup. Creating Services and Foreground Daemons connects the same lifecycle to platform stop requests. These are ways to enter the shutdown sequence; they do not change which callbacks your resources need to finish.
Required Completion After an Optional View Closes
The pi calculator in Simplifying Event Code with Bindings keeps the computation session alive independently of its display. S cancels computation and leaves the stopped display open for browsing. Q, Escape, and a positive timeout release the view and request cancellation, but main keeps dispatching until required completion arrives:
event/PiCalculator/src/PiCalculatorApp.cpp
void PiCalculatorApp::requestExit() {
_exitRequested = true;
// Release optional UI immediately, but keep main open for strongly bound required completion.
_view.reset();
if (_session->isFinished()) {
quit();
} else {
_session->requestStop();
}
}
Releasing the weak notification recipient does not cancel computation or satisfy required completion. The worker observes a separate cancellation flag between arithmetic advancements, and a waiting worker is woken without requiring a queued command on its thread. It flushes a nonempty partial batch before its final callback. That callback is strongly bound to the session on main, where completion joins the worker before notifying the application:
event/PiCalculator/src/PiSession.cpp
void PiSession::onFinished(std::exception_ptr error) {
_worker.join();
_finished = true;
if (_onUpdated) {
_onUpdated();
}
// Retire target-retaining callbacks once the operation is terminal, including on exceptional shutdown.
_onUpdated = {};
_events.reset();
if (auto completion = std::move(_onFinished)) {
completion(std::move(error));
}
}
The application still owns the session here. After joining, the session retires its retained target and callbacks; the application can then quit main safely. When S initiated the stop, completion instead leaves the history and view available. The independent timeout timer remains active while the user browses.
An exception can end main dispatch before this normal sequence completes. The inherited application loop propagates callback failures. Application cleanup cancels and joins the calculation unconditionally, without depending on another delivery:
event/PiCalculator/src/PiSession.cpp
void PiSession::shutdown() {
// Exception cleanup can run after dispatch has ended; joining must not depend on a pending main-loop callback.
_stopping = true;
_worker.requestStop();
_worker.join();
// A stopped loop may still own callbacks retaining this session. Break the return ownership path.
_onUpdated = {};
_onFinished = {};
_events.reset();
}
Cancellation can stop between advancements even while the next digit is being determined. A single growing large-integer operation cannot be interrupted, so this design promises eventual cooperative cleanup, not a fixed bound on joining time. The retained history and arithmetic values grow with the computation.
Complete demo sources
event/FileServer/ServerPart.hpp
// Copyright (c) 2026 Tobias Erbsland - https://erbsland.dev
// SPDX-License-Identifier: Apache-2.0
#pragma once
#include <erbsland/all_core.hpp>
#include <erbsland/all_event.hpp>
#include <erbsland/all_mem.hpp>
#include <erbsland/all_network.hpp>
#include <erbsland/all_network_http.hpp>
#include <erbsland/all_network_http_server.hpp>
#include <erbsland/all_network_source.hpp>
#include <erbsland/all_options.hpp>
#include <erbsland/all_path.hpp>
#include <erbsland/all_stream.hpp>
#include <erbsland/all_text.hpp>
#include <erbsland/all_time.hpp>
#include <erbsland/ApplicationPart.hpp>
#include <erbsland/ApplicationPartIdentifier.hpp>
#include <erbsland/ApplicationPartState.hpp>
#include <erbsland/ByteBlock.hpp>
#include <erbsland/HttpFieldType.hpp>
#include <erbsland/HttpHeaders.hpp>
#include <erbsland/HttpMethod.hpp>
#include <erbsland/HttpMethodType.hpp>
#include <erbsland/HttpResponseHead.hpp>
#include <erbsland/HttpServer.hpp>
#include <erbsland/HttpServerRequest.hpp>
#include <erbsland/HttpServerSession.hpp>
#include <erbsland/HttpStaticFileHandler.hpp>
#include <erbsland/HttpStatus.hpp>
#include <erbsland/HttpVersion.hpp>
#include <erbsland/IpAddress.hpp>
#include <erbsland/IpEndpoint.hpp>
#include <erbsland/Network.hpp>
#include <erbsland/NetworkError.hpp>
#include <erbsland/NetworkErrorContext.hpp>
#include <erbsland/NetworkSourceState.hpp>
#include <erbsland/Options.hpp>
#include <erbsland/OptionValues.hpp>
#include <erbsland/Path.hpp>
#include <erbsland/Port.hpp>
#include <erbsland/StandardStreams.hpp>
#include <erbsland/String.hpp>
#include <erbsland/StringBomMode.hpp>
#include <erbsland/StringEncoder.hpp>
#include <erbsland/StringEncoding.hpp>
#include <erbsland/TimeDelta.hpp>
#include <memory>
namespace demo {
using namespace el::text::literals;
/// Own an HTTP server on an application part's dedicated event thread.
/// Command-line hooks configure the part before startup; lifecycle hooks create and close its event sources.
/// @notest{Verified by running the file_server demo.}
class ServerPart final : public el::ApplicationPart {
public: // registration
/// Get the stable identifier used to register this part.
[[nodiscard]] static auto partIdentifier() -> el::ApplicationPartIdentifierPtr {
static const auto identifier = el::ApplicationPartIdentifier::create("dev.erbsland.demo.file-server"_el);
return identifier;
}
/// Construct this part when the application prepares its registered parts.
/// @return The new server part.
[[nodiscard]] static auto create() -> std::shared_ptr<ServerPart> { return std::make_shared<ServerPart>(); }
public: // implement ApplicationPart
void registerCommandLineOptions(const el::OptionsPtr &options) override;
void parseCommandLine(const el::OptionValuesPtr &values) override;
protected: // implement ApplicationPart
void initialize() override {
// The network source belongs to this part's event loop.
_server = events()->get<el::Network>().createHttpServer();
_server->addStaticContentHandler(el::HttpStaticFileHandler::create(_root, "/files"_el));
_server->events()
.onRequest(
el::HttpMethod{el::HttpMethodType::Get},
"/"_el,
[this](el::HttpServerSessionPtr, el::HttpServerRequestPtr request, el::ByteBlock) -> void {
auto headers = el::HttpHeaders{};
headers.setField(el::HttpFieldType::ContentType, "text/html; charset=utf-8"_el);
const auto body = createIndex();
request->sendResponse(
el::HttpResponseHead{el::HttpVersion::Http11, el::HttpStatus::Ok, "OK"_el, headers},
el::StringEncoder{body}.encode(el::StringEncoding::Utf8, el::StringBomMode::Reject));
})
.onListening([this]() -> void {
el::io::printLine("Browse http://"_el, _server->localEndpoint()->toString(), "/"_el);
if (_timeout.isPositive()) {
events()->invokeAfter(_timeout, []() -> void { el::application().quit(); });
}
})
.onError([](const el::NetworkErrorContext &error) -> void { throw el::NetworkError{error}; })
.onFinal([this]() -> void {
if (state() == el::ApplicationPartState::Stopping) {
completeStopping();
}
});
// Bind to loopback so the demo exposes files only on the local machine.
_server->start(el::IpEndpoint{el::IpAddress::fromStringOrThrow("127.0.0.1"_el), _port});
}
void stopping() override {
// Allow committed responses to drain before ending this part's event loop.
if (_server == nullptr || _server->state() == el::NetworkSourceState::Closed ||
_server->state() == el::NetworkSourceState::Failed) {
completeStopping();
} else {
_server->close();
}
}
void cleanup() noexcept override {
if (_server != nullptr) {
_server->abort();
_server.reset();
}
}
private:
/// Build the HTML listing for the configured content root.
/// @return The HTML document with links to regular files.
auto createIndex() const -> el::String;
private:
el::Path _root; ///< Configured, physically resolved content root.
el::Port _port; ///< Configured loopback listening port.
el::TimeDelta _timeout; ///< Optional demo run duration.
el::HttpServerPtr _server; ///< Event source owned by this part's loop.
};
}
event/EventShutdown/DrainAcceptedWork.cpp
// Copyright (c) 2026 Tobias Erbsland - https://erbsland.dev
// SPDX-License-Identifier: Apache-2.0
#include <DemoCommon.hpp>
#include <erbsland/EventLoop.hpp>
#include <erbsland/EventLoopErrorPolicy.hpp>
#include <erbsland/EventLoopOptions.hpp>
#include <erbsland/LogicFailure.hpp>
#include <erbsland/StandardStreams.hpp>
#include <erbsland/String.hpp>
#include <erbsland/StringList.hpp>
#include <exception>
#include <vector>
namespace demo {
/// Finish accepted satellite observations before ending dispatch; reject later work.
void drainAcceptedWork() {
const auto loop = el::EventLoop::create(el::EventLoopOptions{}.setErrorPolicy(el::EventLoopErrorPolicy::Quit));
auto observations = el::StringList{};
loop->invoke([&observations]() -> void { observations.append("Molntäcke"_el); });
loop->invoke([&observations]() -> void { observations.append("Havsis"_el); });
loop->quit();
loop->invoke([&observations]() -> void { observations.append("Snödjup"_el); });
loop->run();
if (const auto error = loop->takeError()) {
std::rethrow_exception(error);
}
if (observations != el::StringList{"Molntäcke"_el, "Havsis"_el}) {
throw el::LogicFailure{"Quit did not preserve exactly the accepted observations."_el};
}
for (const auto &observation : observations) {
el::io::printLine("Recorded: "_el, observation);
}
}
}
event/PiCalculator/src/PiCalculatorApp.cpp
// Copyright (c) 2026 Tobias Erbsland - https://erbsland.dev
// SPDX-License-Identifier: Apache-2.0
#include "PiCalculatorApp.hpp"
#include <DemoTerminal.hpp>
#include <erbsland/ApplicationError.hpp>
#include <erbsland/BindEvent.hpp>
#include <erbsland/CpCount.hpp>
#include <erbsland/CpIndex.hpp>
#include <erbsland/cterm/Input.hpp>
#include <erbsland/CurrentEvents.hpp>
#include <erbsland/EventLoop.hpp>
#include <erbsland/OptionError.hpp>
#include <erbsland/Options.hpp>
#include <erbsland/OptionType.hpp>
#include <erbsland/OptionValues.hpp>
#include <erbsland/TimeAmounts.hpp>
#include <exception>
#include <utility>
namespace demo {
using namespace el::text::literals;
void PiCalculatorApp::initialize() {
DemoTerminal::enable(*this);
info().setApplicationName("Pi Calculator"_el);
}
void PiCalculatorApp::registerCommandLineOptions(const el::OptionsPtr &options) {
options->setHelpDescription("Calculate exact decimal digits of pi while browsing a live terminal display."_el);
options->addOption({"--digits"_el, "digits"_el})
.setType(el::OptionType::Integer)
.setHelp("Calculate this many fractional digits and exit; omitted means unlimited."_el);
options->addOption({"--timeout"_el, "timeout"_el})
.setType(el::OptionType::Integer)
.setDefaultValue(0)
.setHelp("Stop and exit after this many seconds; zero means unlimited."_el);
options->addOption({"--plain"_el, "plain"_el})
.setHelp(
"Stream digits without screen controls. Interactive keys: S stop, Q/Esc quit, F follow, arrows scroll."_el);
}
void PiCalculatorApp::handleCommandLineOptions(const el::OptionValuesPtr &values) {
const auto timeout = values->getInteger("timeout"_el);
if (timeout < 0) {
throw el::OptionError{
"The timeout must be non-negative."_el,
values->value("timeout"_el),
};
}
_timeout = el::Seconds{timeout};
if (values->value("digits"_el)) {
const auto digits = values->getInteger("digits"_el);
if (!std::in_range<std::size_t>(digits)) {
throw el::OptionError{
"The fractional digit count must be a non-negative size."_el,
values->value("digits"_el),
};
}
_limit = el::CpCount::fromSizeTOrThrow(static_cast<std::size_t>(digits));
}
_plain = values->getFlag("plain"_el) || !terminal()->isInteractive();
// Start the application as soon the main loop has started.
events()->invoke([this]() -> void { begin(); });
}
void PiCalculatorApp::begin() {
_history = std::make_shared<DigitHistory>();
PiSession::OnUpdated updated;
if (_plain) {
terminal()->print("3."_el);
updated = [this]() -> void { printNewDigits(); };
} else {
terminal()->input().setMode(el::cterm::Input::Mode::Key);
_view = std::make_shared<PiView>(terminal(), _history);
// Optional presentation must not keep the view alive after Q or timeout releases it.
updated = el::bindWeakEvent(std::weak_ptr<PiView>{_view}, &PiView::onUpdated);
}
_session = std::make_shared<PiSession>(
el::currentEvents(), _history, std::move(updated), [this](std::exception_ptr error) -> void {
onCalculationFinished(std::move(error));
});
if (_view) {
_view->setSession(_session);
}
_session->start(_limit);
_refresh = events()->createTimer([this]() -> void { tick(); });
_refresh->startFixedDelay(el::Milliseconds{33});
if (_timeout.isPositive()) {
_deadline = events()->createTimer([this]() -> void { requestExit(); });
_deadline->startOnce(_timeout);
}
}
void PiCalculatorApp::tick() {
if (!_view) {
return;
}
const auto action = _view->pollKeys();
if (action == PiView::Action::Quit) {
requestExit();
return;
}
if (action == PiView::Action::Stop) {
_session->requestStop();
}
_view->render();
}
void PiCalculatorApp::printNewDigits() {
const auto count = _history->count();
terminal()->print(_history->range({el::CpIndex::end(_printed), el::CpCount::infinite()}));
terminal()->flush();
_printed = count;
}
void PiCalculatorApp::requestExit() {
_exitRequested = true;
// Release optional UI immediately, but keep main open for strongly bound required completion.
_view.reset();
if (_session->isFinished()) {
quit();
} else {
_session->requestStop();
}
}
void PiCalculatorApp::onCalculationFinished(std::exception_ptr error) {
if (error) {
std::rethrow_exception(error);
}
if (_plain) {
printNewDigits();
terminal()->printLine();
}
// S leaves the stopped view available even with --digits. Only normal finite completion exits automatically.
const auto finiteCompletion = _limit.has_value() && !_session->isStopping();
if (_exitRequested || finiteCompletion || _plain) {
quit();
}
}
void PiCalculatorApp::cleanup() noexcept {
_view.reset();
if (_refresh) {
_refresh->stop();
_refresh.reset();
}
if (_deadline) {
_deadline->stop();
_deadline.reset();
}
if (_session) {
_session->shutdown();
_session.reset();
}
}
}
event/PiCalculator/src/PiSession.cpp
// Copyright (c) 2026 Tobias Erbsland - https://erbsland.dev
// SPDX-License-Identifier: Apache-2.0
#include "PiSession.hpp"
#include <erbsland/BindEvent.hpp>
#include <erbsland/CpCount.hpp>
#include <erbsland/Events.hpp>
#include <erbsland/LogicFailure.hpp>
#include <erbsland/ParameterFailure.hpp>
#include <erbsland/String.hpp>
#include <utility>
namespace demo {
using namespace el::text::literals;
PiSession::PiSession(
el::EventsPtr events,
std::shared_ptr<DigitHistory> history,
OnUpdated onUpdated,
OnFinished onFinished,
PiWorker::Advance advance) :
_events{std::move(events)},
_history{std::move(history)},
_onUpdated{std::move(onUpdated)},
_onFinished{std::move(onFinished)},
_worker{std::move(advance)} {
if (!_events || !_history || !_onFinished) {
throw el::ParameterFailure{"A pi session needs a target, history, and completion callback."_el, "session"_el};
}
}
void PiSession::start(const std::optional<el::CpCount> limit) {
if (_started) {
throw el::LogicFailure{"A pi calculation session starts once."_el};
}
// Required results retain the session and always return to its owner loop.
_worker.start(
limit,
el::bindEvent(_events, shared_from_this(), &PiSession::onDigits),
el::bindEvent(_events, shared_from_this(), &PiSession::onFinished));
_started = true;
}
void PiSession::onDigits(el::String digits) {
_history->append(std::move(digits));
if (_onUpdated) {
_onUpdated();
}
// Main accepts the result before asking for more. Optional display delivery never releases a worker slot.
if (!_stopping) {
_worker.requestNextBatch();
}
}
void PiSession::requestStop() {
if (_finished || _stopping) {
return;
}
_stopping = true;
_worker.requestStop();
if (_onUpdated) {
_onUpdated();
}
}
void PiSession::onFinished(std::exception_ptr error) {
_worker.join();
_finished = true;
if (_onUpdated) {
_onUpdated();
}
// Retire target-retaining callbacks once the operation is terminal, including on exceptional shutdown.
_onUpdated = {};
_events.reset();
if (auto completion = std::move(_onFinished)) {
completion(std::move(error));
}
}
void PiSession::shutdown() {
// Exception cleanup can run after dispatch has ended; joining must not depend on a pending main-loop callback.
_stopping = true;
_worker.requestStop();
_worker.join();
// A stopped loop may still own callbacks retaining this session. Break the return ownership path.
_onUpdated = {};
_onFinished = {};
_events.reset();
}
}