timed_thread_context: destructor latency under concurrent submissions #2337

Summary

timed_thread_context::~timed_thread_context() can take orders of magnitude
longer than expected when other threads submit operations while the context is
shutting down. The close sequence in run()
(include/exec/timed_thread_scheduler.hpp, currently around lines 195–205)
spins on n_submissions_in_flight_ with STDEXEC::__spin_loop_pause() until
it sees 0, and a submitter that is mid-schedule() keeps that counter
non-zero, so shutdown does not complete until every in-flight submission has
been released.

I measured this on macOS (Apple clang 21, arm64, -O2):
with 4 threads submitting schedule_after(300ms) operations in a tight loop
and then stopping, the destructor blocked for about 300 ms while roughly
1.3 million operations were started and stopped during the spin. In a variant
with 50 ms timers the shutdown still cost ~236 ms, i.e. several times the
remaining timer deadline.

This matches the concern raised during the review of #2231, where it was noted
that the shutdown loop "seems like it could spin for an unbounded amount of
time if other threads keep queuing new items onto it". #2231 merged with a
plan to file follow-up issues for the review comments; as far as I can tell
none of the others have been filed yet (approx. 2 months later), so I decided
to write this one up.

One contributing factor worth mentioning in the same discussion: a submitter
that observes n >= 0 right after the close CAS succeeded will still push its
command into command_queue_ (the fetch_add in schedule() is
unconditional), even though run() has already committed to exiting. That
command never gets processed; its set_stopped()/set_value() only happens
because the heap pass and the caller-provided stop callbacks race it. So the
shutdown path isn't just slow under contention, it also leaves a window where
a queued command is handled by nobody.

What I propose to investigate

Two of the alternatives suggested in the #2231 review seem reasonable:

1. Keep the current "wait for in-flight submissions to drain" design, but
   bound the wait: make the counter CAS into a state where no new
   submissions are accepted, not just into a negative sentinel the next
   fetch_add happens to race past. That guarantees shutdown terminates
   within at most (number of schedule() calls currently in progress) steps.
2. Document a stronger precondition on the destructor — all submissions have
   completed before destruction begins — and simplify the shutdown loop to
   assume an empty queue. This is the bigger refactor also suggested in the
   Fix missing enqueue notifications in timed_thread_context. #2231 review (n_submissions_in_flight_ could be deleted entirely).

Option 2 changes public documentation, so it likely needs a maintainer
decision; option 1 is a smaller local change. Happy to put up a PR for
whichever direction you prefer.

Notes

• The reproducer and measurement scripts are straightforward ~60-line
  programs; I can attach or paste them if useful.
• I ran this against main at 5fc945f (the unless_stop_requested consteval
  port merge), on commit 30bb116 directly after the Fix missing enqueue notifications in timed_thread_context. #2231 merge the same
  failure mode was already present.
• timed_thread_context: an operation stopped just before the context is destroyed never completes #2334 and Complete commands queued when timed_thread_context shuts down #2335 cover a related but distinct failure mode (a stop command
  queued during shutdown is never drained, leaving a receiver permanently
  incomplete). This issue is about the latency / termination bound of the
  same close sequence, plus the queued-command race, not about the lost stop
  command's refcount.

#include <exec/timed_thread_scheduler.hpp>
#include <exec/start_detached.hpp>
#include <stdexec/execution.hpp>

#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdlib>
#include <thread>
#include <vector>

// Measures how long the timed_thread_context destructor takes while other
// threads submit operations concurrently with shutdown.
namespace ex = STDEXEC;

int main(int argc, char** argv) {
  const int submitters = argc > 1 ? std::atoi(argv[1]) : 4;
  const int millis = argc > 2 ? std::atoi(argv[2]) : 300;
  exec::timed_thread_context ctx;
  std::atomic<bool> keep_going{true};
  std::atomic<long long> completions{0};

  std::vector<std::thread> threads;
  for (int t = 0; t < submitters; ++t) {
    threads.emplace_back([&] {
      while (keep_going.load(std::memory_order_relaxed)) {
        exec::start_detached(
          exec::schedule_after(ctx.get_scheduler(), std::chrono::milliseconds(millis))
          | ex::upon_stopped([&completions] {
              completions.fetch_add(1, std::memory_order_relaxed);
            }));
      }
    });
  }

  std::this_thread::sleep_for(std::chrono::milliseconds(millis));
  keep_going.store(false, std::memory_order_relaxed);
  for (auto& thread : threads) {
    thread.join();
  }

  auto t0 = std::chrono::steady_clock::now();
  {
    // run the destructor manually, then reconstruct the object so the
    // automatic destructor at scope exit is not a double destruction
    exec::timed_thread_context* p = &ctx;
    p->~timed_thread_context();  // request_stop() + join()
    new (p) exec::timed_thread_context();
  }
  auto t1 = std::chrono::steady_clock::now();
  auto us = std::chrono::duration_cast<std::chrono::microseconds>(t1 - t0).count();

  std::printf(
    "submitters=%d completions=%lld destructor-latency=%lldus\n",
    submitters,
    completions.load(),
    static_cast<long long>(us));
  return 0;
}