Repository navigation
timed_thread_context: an operation stopped just before the context is destroyed never completes #2334
Description
Summary
When timed_thread_context::run() sees stop_requested_, it closes submissions, stops the operations in the heap and exits, without draining command_queue_ again:
while (true) {
while (command_type* op = command_queue_.pop_front()) { ... } // the only drain
...
cv_.wait_until(lock, deadline, [this] { return ready_ || stop_requested_; });
if (stop_requested) {
// close submissions (CAS n_submissions_in_flight_ to context_closed)
// set_stopped_() on every operation in heap_
break; // commands pushed since the drain above are never processed
}
}A stop request from another thread raises the operation's ref_count_ from 1 to 2 and queues a stop command. If the context is destroyed before its thread drains that command, the heap pass only drops the count to 1. The stop command, which would drop it to 0 and call set_stopped, is never processed. So the receiver gets neither set_value nor set_stopped, and anything waiting on it (sync_wait, async_scope::on_empty(), ...) hangs.
A schedule command queued after the last drain is lost the same way: it is never inserted into the heap, so the heap pass doesn't stop it.
This breaks the context's own shutdown guarantee: the destructor stops the operations in the heap, and submissions made after close complete inline with set_stopped.
It seems that the current behavior violates https://eel.is/c++draft/exec.async.ops#8: “After its start operation has begun executing, exactly one completion operation shall execute.”
Reproducer
#include <exec/timed_thread_scheduler.hpp>
#include <stdexec/execution.hpp>
#include <atomic>
#include <chrono>
#include <cstdio>
#include <optional>
#include <thread>
namespace ex = STDEXEC;
struct env {
ex::inplace_stop_token token;
auto query(ex::get_stop_token_t) const noexcept { return token; }
};
struct receiver {
using receiver_concept = ex::receiver_tag;
std::atomic<int>* completed;
ex::inplace_stop_token token;
void set_value() noexcept { completed->store(1); }
void set_stopped() noexcept { completed->store(2); }
env get_env() const noexcept { return {token}; }
};
int main() {
int lost = 0;
const int runs = 1000;
for (int i = 0; i < runs; ++i) {
std::atomic<int> completed{0};
ex::inplace_stop_source source;
std::optional<exec::timed_thread_context> ctx{std::in_place};
auto op = ex::connect(exec::schedule_after(ctx->get_scheduler(), std::chrono::hours(1)),
receiver{&completed, source.get_token()});
ex::start(op);
std::this_thread::sleep_for(std::chrono::microseconds(200)); // the operation reaches the heap
source.request_stop(); // queues a stop command
ctx.reset(); // request_stop() + join, usually before the thread drains the command
if (completed.load() == 0) { // the thread has exited, so nothing will ever complete it
++lost;
}
}
std::printf("receiver never completed in %d of %d runs\n", lost, runs);
return lost == 0 ? 0 : 1;
}The output will be: receiver never completed in 1000 of 1000 runs. It should be 0 of 1000.
Suggested fix: #2335
Activity