mirror of
https://github.com/openai/codex.git
synced 2026-10-19 16:17:49 +00:00
Cancel code mode timer tasks when cleared or the cell finishes (#45399)
## Why Each `setTimeout` spawned a sleeping thread that remained alive until its delay elapsed, even after `clearTimeout` or cell completion. ## What changed Replace per-timer threads with Tokio sleep tasks held by `AbortOnDropHandle`, so removing a timeout or dropping the isolate cancels its task. Enter the caller's Tokio runtime on the code mode runtime thread to support scheduling these timers. ## Testing Add a regression test using virtual time to verify that cleared timers release their tasks, an awaited timer completes, and cell completion cancels remaining timers without emitting their output. GitOrigin-RevId: 59c7e57ed226cb32633e2d4c25b778cfab2af423
This commit is contained in:
@@ -103,7 +103,9 @@ pub(crate) fn spawn_runtime(
|
||||
stored_values,
|
||||
};
|
||||
|
||||
let runtime_handle = tokio::runtime::Handle::current();
|
||||
spawn_supervised_runtime_thread(event_tx.clone(), task_failure_handler, move || {
|
||||
let _runtime_guard = runtime_handle.enter();
|
||||
run_runtime(
|
||||
config,
|
||||
event_tx,
|
||||
@@ -512,4 +514,70 @@ await new Promise(() => {});
|
||||
.send(RuntimeControlCommand::Terminate)
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn timers_release_tasks_when_cleared_or_the_cell_finishes() {
|
||||
let (event_tx, mut event_rx) = mpsc::unbounded_channel();
|
||||
let (_runtime_tx, _runtime_control_tx, _runtime_terminate_handle) = spawn_runtime(
|
||||
HashMap::new(),
|
||||
execute_request(
|
||||
r#"
|
||||
clearTimeout(setTimeout(() => text("cancelled"), 3_600_000));
|
||||
await new Promise((resolve) => setTimeout(resolve, 3_600_000));
|
||||
text("done");
|
||||
setTimeout(() => text("late"), 3_600_000);
|
||||
"#,
|
||||
),
|
||||
event_tx,
|
||||
PendingRuntimeMode::Continue,
|
||||
/*task_failure_handler*/ None,
|
||||
)
|
||||
.unwrap();
|
||||
|
||||
loop {
|
||||
match tokio::time::timeout(Duration::from_secs(1), event_rx.recv())
|
||||
.await
|
||||
.unwrap()
|
||||
{
|
||||
Some(RuntimeEvent::Pending) => break,
|
||||
Some(_) => {}
|
||||
None => panic!("runtime closed before the timer was pending"),
|
||||
}
|
||||
}
|
||||
tokio::task::yield_now().await;
|
||||
assert_eq!(
|
||||
tokio::runtime::Handle::current()
|
||||
.metrics()
|
||||
.num_alive_tasks(),
|
||||
1
|
||||
);
|
||||
tokio::time::pause();
|
||||
tokio::time::advance(Duration::from_secs(3_600)).await;
|
||||
tokio::time::resume();
|
||||
|
||||
let mut output = Vec::new();
|
||||
while let Some(event) = tokio::time::timeout(Duration::from_secs(1), event_rx.recv())
|
||||
.await
|
||||
.expect("timer runtime should finish")
|
||||
{
|
||||
match event {
|
||||
RuntimeEvent::ContentItem(item) => output.push(item),
|
||||
RuntimeEvent::Result { error_text, .. } => assert_eq!(error_text, None),
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
assert_eq!(
|
||||
output,
|
||||
vec![FunctionCallOutputContentItem::InputText {
|
||||
text: "done".to_string()
|
||||
}]
|
||||
);
|
||||
tokio::task::yield_now().await;
|
||||
assert_eq!(
|
||||
tokio::runtime::Handle::current()
|
||||
.metrics()
|
||||
.num_alive_tasks(),
|
||||
0
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,12 +1,15 @@
|
||||
use std::thread;
|
||||
use std::time::Duration;
|
||||
|
||||
use tokio_util::task::AbortOnDropHandle;
|
||||
|
||||
use super::RuntimeCommand;
|
||||
use super::RuntimeState;
|
||||
use super::value::value_to_error_text;
|
||||
|
||||
pub(super) struct ScheduledTimeout {
|
||||
callback: v8::Global<v8::Function>,
|
||||
// Clearing the timeout or dropping the isolate also cancels its sleep.
|
||||
_task: AbortOnDropHandle<()>,
|
||||
}
|
||||
|
||||
pub(super) fn schedule_timeout(
|
||||
@@ -33,13 +36,18 @@ pub(super) fn schedule_timeout(
|
||||
let timeout_id = state.next_timeout_id;
|
||||
state.next_timeout_id = state.next_timeout_id.saturating_add(1);
|
||||
let runtime_command_tx = state.runtime_command_tx.clone();
|
||||
state
|
||||
.pending_timeouts
|
||||
.insert(timeout_id, ScheduledTimeout { callback });
|
||||
thread::spawn(move || {
|
||||
thread::sleep(Duration::from_millis(delay_ms));
|
||||
let sleep = tokio::time::sleep(Duration::from_millis(delay_ms));
|
||||
let task = tokio::spawn(async move {
|
||||
sleep.await;
|
||||
let _ = runtime_command_tx.send(RuntimeCommand::TimeoutFired { id: timeout_id });
|
||||
});
|
||||
state.pending_timeouts.insert(
|
||||
timeout_id,
|
||||
ScheduledTimeout {
|
||||
callback,
|
||||
_task: AbortOnDropHandle::new(task),
|
||||
},
|
||||
);
|
||||
|
||||
Ok(timeout_id)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user