c36b0fb165
This PR moves `IO.CancelToken` from `Init.System.IO` to its own file
`Init.System.CancelToken`, backed by `IO.Promise Unit` instead of
`IO.Ref Bool`. This enables non-polling cancellation propagation: the
token's underlying promise can be used directly with `IO.waitAny`, and
callbacks can be registered to fire when cancellation is requested.
The structure carries both the promise *and* a plain `IO.Ref Bool` flag,
set in lockstep by `set`. `isSet` reads the flag directly (used on hot
paths like `Core.checkInterrupted`); `task`/`onSet` go through the
promise. The avoids a ~0.4% regression that a pure-promise
representation introduced.
API additions:
- `CancelToken.task : Task (Option Unit)`. Returns the underlying
promise's `result?` task directly — the same task object on every call,
so further `Task.map`/`BaseIO.bindTask` dependencies can be safely
attached. Resolves with `some ()` when `set` is called, or `none` if the
token is dropped without ever being set.
- `CancelToken.onSet : BaseIO Unit → BaseIO Unit`. Registers a callback
that runs synchronously on the cancelling thread when `set` is called
(or immediately if the token is already set). Implemented via
`BaseIO.chainTask` on `result?`, so no fresh `Task.map` per call and no
GC hazard.
Runtime cleanup:
- Add `LEAN_TASK_STATE_{WAITING,RUNNING,FINISHED}` constants in `lean.h`
matching `IO.TaskState`.
- Factor `lean::promise_is_resolved` inline in `object.h`, replacing
three open-coded `lean_io_get_task_state_core(...) == 2` checks (in
`interrupt.cpp`, `uv/timer.cpp`, `uv/signal.cpp`).
- Drop the manual `inc_ref(g_cancel_tk)` in `check_interrupted`; the
token is owned by the enclosing `scope_cancel_tk` for the duration of
the call (documented).
- Replace the bare `lean_always_assert(g_task_manager)` in
`lean_promise_new` with an explicit `lean_internal_panic` carrying a
message that names `Promise.new`, identifies the typical trigger
(`initialize` blocks, transitively via `IO.CancelToken.new`), and
recommends lazy construction. Without this, users got an opaque "LEAN
ASSERTION VIOLATION ... Condition: g_task_manager" with no actionable
hint.
Behavioural notes documented inline:
- `new` cannot be called from `initialize` blocks (task manager not
running yet); construct lazily.
- `task` documents the dropped-promise case (`none`) and steers callers
to `onSet` for callback chaining.
A consumer of `onSet` for parent → child cancel-token propagation in
parallel tactic combinators is in #13428 (fixes #13300).
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
---------
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
103 lines
3.1 KiB
C++
103 lines
3.1 KiB
C++
/*
|
|
Copyright (c) 2013 Microsoft Corporation. All rights reserved.
|
|
Released under Apache 2.0 license as described in the file LICENSE.
|
|
|
|
Author: Leonardo de Moura
|
|
*/
|
|
#include <limits>
|
|
#include "runtime/thread.h"
|
|
#include "runtime/interrupt.h"
|
|
#include "runtime/exception.h"
|
|
#include "runtime/memory.h"
|
|
#include "runtime/object.h"
|
|
#include "lean/lean.h"
|
|
#include "util/io.h"
|
|
|
|
namespace lean {
|
|
LEAN_THREAD_VALUE(size_t, g_max_heartbeat, 0);
|
|
LEAN_THREAD_VALUE(size_t, g_heartbeat, 0);
|
|
|
|
extern "C" LEAN_EXPORT obj_res lean_internal_get_default_max_heartbeat() {
|
|
#ifdef LEAN_DEFAULT_MAX_HEARTBEAT
|
|
return lean_box(LEAN_DEFAULT_MAX_HEARTBEAT);
|
|
#else
|
|
return lean_box(0);
|
|
#endif
|
|
}
|
|
|
|
void inc_heartbeat() { g_heartbeat++; }
|
|
|
|
void reset_heartbeat() { g_heartbeat = 0; }
|
|
|
|
void set_max_heartbeat(size_t max) { g_max_heartbeat = max; }
|
|
|
|
extern "C" LEAN_EXPORT obj_res lean_internal_set_max_heartbeat(usize max) {
|
|
set_max_heartbeat(max);
|
|
return lean_box(0);
|
|
}
|
|
|
|
size_t get_max_heartbeat() { return g_max_heartbeat; }
|
|
|
|
void set_max_heartbeat_thousands(unsigned max) { g_max_heartbeat = static_cast<size_t>(max) * 1000; }
|
|
|
|
scope_heartbeat::scope_heartbeat(size_t max):flet<size_t>(g_heartbeat, max) {}
|
|
LEAN_EXPORT scope_max_heartbeat::scope_max_heartbeat(size_t max):flet<size_t>(g_max_heartbeat, max) {}
|
|
|
|
// separate definition to allow breakpoint in debugger
|
|
void throw_heartbeat_exception() {
|
|
throw heartbeat_exception();
|
|
}
|
|
|
|
void check_heartbeat() {
|
|
inc_heartbeat();
|
|
if (g_max_heartbeat > 0 && g_heartbeat > g_max_heartbeat)
|
|
throw_heartbeat_exception();
|
|
}
|
|
|
|
LEAN_THREAD_VALUE(lean_object *, g_cancel_tk, nullptr);
|
|
|
|
LEAN_EXPORT scope_cancel_tk::scope_cancel_tk(lean_object * o):flet<lean_object *>(g_cancel_tk, o) {}
|
|
|
|
// `IO.CancelToken` is `structure { promise : IO.Promise Unit; setRef : IO.Ref Bool }`. We read
|
|
// the `Bool` flag (field 1) directly: cheaper than walking the promise's task state, and this
|
|
// is on the hot `Core.checkInterrupted` path. Must stay in sync with the field order in
|
|
// `Init/System/CancelToken.lean`.
|
|
static bool cancel_tk_is_set(lean_object * tk) {
|
|
lean_object * setRef = lean_ctor_get(tk, 1);
|
|
return lean_unbox(lean_to_ref(setRef)->m_value) != 0;
|
|
}
|
|
|
|
void check_interrupted() {
|
|
if (g_cancel_tk) {
|
|
// `g_cancel_tk` is owned by the enclosing `scope_cancel_tk`, so it stays alive for the
|
|
// duration of this call without an explicit `inc_ref`.
|
|
if (cancel_tk_is_set(g_cancel_tk) &&
|
|
!std::uncaught_exceptions()) {
|
|
throw interrupted();
|
|
}
|
|
}
|
|
}
|
|
|
|
void check_system(char const * component_name, bool do_check_interrupted) {
|
|
check_stack(component_name);
|
|
check_memory(component_name);
|
|
if (do_check_interrupted) {
|
|
check_interrupted();
|
|
check_heartbeat();
|
|
}
|
|
}
|
|
|
|
void sleep_for(unsigned ms, unsigned step_ms) {
|
|
if (step_ms == 0)
|
|
step_ms = 1;
|
|
unsigned rounds = ms / step_ms;
|
|
chrono::milliseconds c(step_ms);
|
|
chrono::milliseconds r(ms % step_ms);
|
|
for (unsigned i = 0; i < rounds; i++) {
|
|
this_thread::sleep_for(c);
|
|
check_interrupted();
|
|
}
|
|
this_thread::sleep_for(r);
|
|
check_interrupted();
|
|
}
|
|
}
|