Skip to content

Commit 66ee171

Browse files
committed
fix(runtime): repair Wasm socket and signal integration
1 parent c6d8147 commit 66ee171

13 files changed

Lines changed: 636 additions & 622 deletions

File tree

crates/bridge/bridge-contract.json

Lines changed: 18 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -189,12 +189,13 @@
189189
"_childProcessSpawnStart",
190190
"_childProcessPoll",
191191
"_childProcessStdinWrite",
192-
"_childProcessStdinClose",
193-
"_childProcessKill",
194-
"_processKill",
195-
"_processSignalState"
196-
]
197-
},
192+
"_childProcessStdinClose",
193+
"_childProcessKill",
194+
"_processKill",
195+
"_processSignalState",
196+
"_processTakeSignal"
197+
]
198+
},
198199
{
199200
"convention": "syncPromise",
200201
"argumentTypes": [
@@ -945,13 +946,17 @@
945946
"method": "process.kill",
946947
"translateArgs": false
947948
},
948-
"_processSignalState": {
949-
"method": "process.signal_state",
950-
"translateArgs": false
951-
},
952-
"_ptySetRawMode": {
953-
"method": "__pty_set_raw_mode",
954-
"translateArgs": false
949+
"_processSignalState": {
950+
"method": "process.signal_state",
951+
"translateArgs": false
952+
},
953+
"_processTakeSignal": {
954+
"method": "process.take_signal",
955+
"translateArgs": false
956+
},
957+
"_ptySetRawMode": {
958+
"method": "__pty_set_raw_mode",
959+
"translateArgs": false
955960
},
956961
"_resolveModule": {
957962
"method": "__resolve_module",

crates/execution/assets/runners/wasm-runner.mjs

Lines changed: 40 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -2652,6 +2652,8 @@ function readSyncRpcLine() {
26522652
}
26532653
}
26542654

2655+
const pendingWasmSignals = [];
2656+
26552657
function callSyncRpc(method, args = []) {
26562658
if (
26572659
globalThis.__agentOSSyncRpc &&
@@ -3040,6 +3042,7 @@ const hostNetImport = {
30403042
process.env.AGENTOS_SANDBOX_ROOT.length > 0);
30413043
try {
30423044
while (true) {
3045+
dispatchPendingWasmSignals();
30433046
const view = new DataView(instanceMemory.buffer);
30443047
let ready = 0;
30453048
// fds the kernel owns (PTY/pipe stdio in sidecar-managed mode): their readiness
@@ -3144,6 +3147,7 @@ const hostNetImport = {
31443147
}
31453148

31463149
if (ready > 0 || t === 0 || (deadline != null && Date.now() >= deadline)) {
3150+
dispatchPendingWasmSignals();
31473151
new DataView(instanceMemory.buffer).setUint32(Number(retReadyPtr) >>> 0, ready >>> 0, true);
31483152
return 0;
31493153
}
@@ -4159,9 +4163,12 @@ const hostProcessImport = {
41594163
const deadline = Date.now() + (Number(milliseconds) >>> 0);
41604164
while (Date.now() < deadline) {
41614165
// Keep guest sleeps interruptible by V8 termination during SIGTERM,
4162-
// SIGKILL, and VM disposal.
4166+
// SIGKILL, and VM disposal. Also drain handled Wasm signals at
4167+
// syscall boundaries so cooperative handlers run during sleeps.
4168+
dispatchPendingWasmSignals();
41634169
Atomics.wait(waitArray, 0, 0, Math.max(1, Math.min(10, deadline - Date.now())));
41644170
}
4171+
dispatchPendingWasmSignals();
41654172
return WASI_ERRNO_SUCCESS;
41664173
} catch {
41674174
return WASI_ERRNO_FAULT;
@@ -4180,11 +4187,12 @@ const hostProcessImport = {
41804187
mask: decodeSignalMask(maskLo, maskHi),
41814188
flags: Number(flags) >>> 0,
41824189
};
4183-
emitControlMessage({
4184-
type: 'signal_state',
4185-
signal: Number(signal) >>> 0,
4186-
registration,
4187-
});
4190+
callSyncRpc('process.signal_state', [
4191+
Number(signal) >>> 0,
4192+
registration.action,
4193+
JSON.stringify(registration.mask),
4194+
registration.flags,
4195+
]);
41884196
return WASI_ERRNO_SUCCESS;
41894197
} catch {
41904198
return WASI_ERRNO_FAULT;
@@ -5820,6 +5828,7 @@ wasiImport.poll_oneoff = (inPtr, outPtr, nsubscriptions, neventsPtr) => {
58205828
}
58215829

58225830
while (readyEvents.length === 0) {
5831+
dispatchPendingWasmSignals();
58235832
for (const subscription of subscriptions) {
58245833
if (subscription.error != null) {
58255834
readyEvents.push({
@@ -5903,7 +5912,7 @@ wasiImport.poll_oneoff = (inPtr, outPtr, nsubscriptions, neventsPtr) => {
59035912
);
59045913
}
59055914

5906-
if (readyEvents.length === 0 && hasClockSubscription) {
5915+
if (readyEvents.length === 0 && hasClockSubscription) {
59075916
const clockSubscription = subscriptions.find((subscription) => subscription.kind === 'clock');
59085917
readyEvents.push({
59095918
userdata: clockSubscription.userdata,
@@ -6044,6 +6053,27 @@ function dispatchWasmSignal(signal) {
60446053
}
60456054
}
60466055

6056+
function dispatchPendingWasmSignals() {
6057+
while (pendingWasmSignals.length > 0) {
6058+
dispatchWasmSignal(pendingWasmSignals.shift());
6059+
}
6060+
while (true) {
6061+
let signal;
6062+
try {
6063+
signal = callSyncRpc('process.take_signal', []);
6064+
} catch (error) {
6065+
if (error?.code === 'ERR_AGENTOS_WASM_SYNC_RPC_UNAVAILABLE') {
6066+
return;
6067+
}
6068+
throw error;
6069+
}
6070+
if (typeof signal !== 'number') {
6071+
return;
6072+
}
6073+
dispatchWasmSignal(signal);
6074+
}
6075+
}
6076+
60476077
Object.defineProperty(globalThis, '__secureExecWasmSignalDispatch', {
60486078
configurable: true,
60496079
writable: true,
@@ -6052,7 +6082,9 @@ Object.defineProperty(globalThis, '__secureExecWasmSignalDispatch', {
60526082
typeof payload?.number === 'number'
60536083
? payload.number
60546084
: signalNumberFromName(payload?.signal);
6055-
dispatchWasmSignal(signal);
6085+
if (signal > 0) {
6086+
pendingWasmSignals.push(signal);
6087+
}
60566088
},
60576089
});
60586090

crates/execution/src/node_import_cache.rs

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,7 +15,7 @@ const NODE_IMPORT_CACHE_PATH_ENV: &str = "AGENTOS_NODE_IMPORT_CACHE_PATH";
1515
const NODE_IMPORT_CACHE_LOADER_PATH_ENV: &str = "AGENTOS_NODE_IMPORT_CACHE_LOADER_PATH";
1616
const NODE_IMPORT_CACHE_SCHEMA_VERSION: &str = "1";
1717
const NODE_IMPORT_CACHE_LOADER_VERSION: &str = "8";
18-
const NODE_IMPORT_CACHE_ASSET_VERSION: &str = "86";
18+
const NODE_IMPORT_CACHE_ASSET_VERSION: &str = "93";
1919
const NODE_IMPORT_CACHE_DIR_PREFIX: &str = "agentos-node-import-cache";
2020
const DEFAULT_NODE_IMPORT_CACHE_MATERIALIZE_TIMEOUT: Duration = Duration::from_secs(30);
2121
const PYODIDE_DIST_DIR: &str = "pyodide-dist";

crates/execution/src/wasm.rs

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2085,7 +2085,11 @@ fn translate_wasm_signal_state_sync_rpc_request(
20852085
let flags = request
20862086
.args
20872087
.get(3)
2088-
.and_then(Value::as_u64)
2088+
.and_then(|value| {
2089+
value
2090+
.as_u64()
2091+
.or_else(|| value.as_i64().map(|signed| signed as u64))
2092+
})
20892093
.unwrap_or_default() as u32;
20902094

20912095
execution
@@ -2843,6 +2847,11 @@ if (typeof globalThis !== "undefined") {{
28432847
flags,
28442848
]);
28452849
}}
2850+
case "process.take_signal":
2851+
if (typeof _processTakeSignal === "undefined") {{
2852+
throw new Error("secure-exec WASM signal-drain bridge is unavailable");
2853+
}}
2854+
return _processTakeSignal.applySync(void 0, args);
28462855
default:
28472856
throw new Error(`secure-exec WASM sync RPC method not implemented in V8 runtime: ${{method}}`);
28482857
}}

crates/native-sidecar/src/execution.rs

Lines changed: 121 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -483,6 +483,7 @@ impl ActiveProcess {
483483
next_mapped_host_fd: MAPPED_HOST_FD_START,
484484
pending_execution_events: VecDeque::new(),
485485
pending_self_signal_exit: None,
486+
pending_wasm_signals: VecDeque::new(),
486487
child_processes: BTreeMap::new(),
487488
next_child_process_id: 0,
488489
http_servers: BTreeMap::new(),
@@ -4523,6 +4524,8 @@ where
45234524
let (connection_id, session_id, vm_id) = self.vm_scope_for(&request.ownership)?;
45244525
self.require_owned_vm(&connection_id, &session_id, &vm_id)?;
45254526

4527+
self.drain_root_signal_state_events(&vm_id, &payload.process_id)?;
4528+
45264529
let handlers = self
45274530
.vms
45284531
.get(&vm_id)
@@ -4536,6 +4539,77 @@ where
45364539
})
45374540
}
45384541

4542+
fn drain_root_signal_state_events(
4543+
&mut self,
4544+
vm_id: &str,
4545+
process_id: &str,
4546+
) -> Result<(), SidecarError> {
4547+
let mut deferred = VecDeque::new();
4548+
4549+
loop {
4550+
let event = {
4551+
let Some(vm) = self.vms.get_mut(vm_id) else {
4552+
break;
4553+
};
4554+
let Some(process) = vm.active_processes.get_mut(process_id) else {
4555+
break;
4556+
};
4557+
if let Some(event) = process.pending_execution_events.pop_front() {
4558+
Some(event)
4559+
} else {
4560+
match process.execution.poll_event_blocking(Duration::ZERO) {
4561+
Ok(event) => event,
4562+
Err(SidecarError::Execution(message))
4563+
if (process.runtime == GuestRuntimeKind::JavaScript
4564+
&& closed_javascript_event_channel(&message))
4565+
|| (process.runtime == GuestRuntimeKind::Python
4566+
&& closed_python_event_channel(&message))
4567+
|| (process.runtime == GuestRuntimeKind::WebAssembly
4568+
&& closed_wasm_event_channel(&message)) =>
4569+
{
4570+
None
4571+
}
4572+
Err(error) => return Err(error),
4573+
}
4574+
}
4575+
};
4576+
4577+
let Some(event) = event else {
4578+
break;
4579+
};
4580+
4581+
match event {
4582+
ActiveExecutionEvent::SignalState {
4583+
signal,
4584+
registration,
4585+
} => {
4586+
if let Some(vm) = self.vms.get_mut(vm_id) {
4587+
apply_process_signal_state_update(
4588+
&mut vm.signal_states,
4589+
process_id,
4590+
signal,
4591+
registration,
4592+
);
4593+
}
4594+
}
4595+
other => deferred.push_back(other),
4596+
}
4597+
}
4598+
4599+
if let Some(vm) = self.vms.get_mut(vm_id) {
4600+
if let Some(process) = vm.active_processes.get_mut(process_id) {
4601+
for event in deferred.into_iter().rev() {
4602+
if process.pending_execution_events.len() >= MAX_PROCESS_EVENT_QUEUE {
4603+
return Err(process_event_queue_overflow_error());
4604+
}
4605+
process.pending_execution_events.push_front(event);
4606+
}
4607+
}
4608+
}
4609+
4610+
Ok(())
4611+
}
4612+
45394613
pub(crate) async fn get_zombie_timer_count(
45404614
&mut self,
45414615
request: &RequestFrame,
@@ -4627,7 +4701,7 @@ where
46274701
KillBehavior::SharedV8DispatchOrTerminate
46284702
}
46294703
ActiveExecution::Wasm(execution) if execution.uses_shared_v8_runtime() => {
4630-
KillBehavior::SharedV8Terminate
4704+
KillBehavior::SharedV8DispatchOrTerminate
46314705
}
46324706
ActiveExecution::Python(execution) if execution.uses_shared_v8_runtime() => {
46334707
KillBehavior::SharedV8Terminate
@@ -4680,8 +4754,37 @@ where
46804754
}
46814755
}
46824756
KillBehavior::SharedV8DispatchOrTerminate => {
4683-
if signal != 0 && !dispatch_v8_process_signal(process, signal)? {
4684-
process.execution.terminate()?;
4757+
if signal != 0 {
4758+
let is_shared_wasm = matches!(
4759+
&process.execution,
4760+
ActiveExecution::Wasm(execution) if execution.uses_shared_v8_runtime()
4761+
);
4762+
if is_shared_wasm {
4763+
let action = vm
4764+
.signal_states
4765+
.get(process_id)
4766+
.and_then(|handlers| handlers.get(&(signal as u32)))
4767+
.map(|registration| &registration.action);
4768+
match action {
4769+
Some(SignalDispositionAction::Ignore) => {}
4770+
Some(SignalDispositionAction::User) => {
4771+
process.pending_wasm_signals.push_back(signal);
4772+
}
4773+
Some(SignalDispositionAction::Default) | None => {
4774+
if !matches!(
4775+
canonical_signal_name(signal),
4776+
Some("SIGWINCH" | "SIGCHLD" | "SIGURG")
4777+
) {
4778+
process.execution.terminate()?;
4779+
process.queue_pending_execution_event(
4780+
ActiveExecutionEvent::Exited(128 + signal),
4781+
)?;
4782+
}
4783+
}
4784+
}
4785+
} else if !dispatch_v8_process_signal(process, signal)? {
4786+
process.execution.terminate()?;
4787+
}
46854788
}
46864789
}
46874790
KillBehavior::Noop => {}
@@ -8747,9 +8850,10 @@ where
87478850
"ESRCH: unknown process pid {target_pid}"
87488851
)));
87498852
};
8750-
if let Some(session) = target.execution.javascript_v8_session_handle().filter(
8751-
|_| matches!(&target.execution, ActiveExecution::Javascript(execution) if execution.uses_shared_v8_runtime())
8752-
|| matches!(&target.execution, ActiveExecution::Wasm(execution) if execution.uses_shared_v8_runtime()),
8853+
if matches!(&target.execution, ActiveExecution::Wasm(execution) if execution.uses_shared_v8_runtime()) {
8854+
target.pending_wasm_signals.push_back(signal);
8855+
} else if let Some(session) = target.execution.javascript_v8_session_handle().filter(
8856+
|_| matches!(&target.execution, ActiveExecution::Javascript(execution) if execution.uses_shared_v8_runtime()),
87538857
) {
87548858
dispatch_v8_session_signal_async(session, signal);
87558859
} else if !dispatch_v8_process_signal(target, signal)? {
@@ -16754,6 +16858,10 @@ where
1675416858
"action": "default",
1675516859
}))
1675616860
}
16861+
"process.take_signal" => {
16862+
let signal = process.pending_wasm_signals.pop_front();
16863+
Ok(signal.map(Value::from).unwrap_or(Value::Null).into())
16864+
}
1675716865
"process.umask" => {
1675816866
let new_mask = javascript_sync_rpc_arg_u32_optional(&request.args, 0, "process umask")?;
1675916867
kernel
@@ -24141,6 +24249,13 @@ fn dispatch_v8_process_signal(process: &ActiveProcess, signal: i32) -> Result<bo
2414124249
let Some(signal_name) = signal_name_for_stream_event(signal) else {
2414224250
return Ok(false);
2414324251
};
24252+
if matches!(&process.execution, ActiveExecution::Wasm(execution) if execution.uses_shared_v8_runtime())
24253+
{
24254+
if let Some(session) = process.execution.javascript_v8_session_handle() {
24255+
dispatch_v8_session_signal_async(session, signal);
24256+
return Ok(true);
24257+
}
24258+
}
2414424259
process.execution.send_javascript_stream_event(
2414524260
"signal",
2414624261
json!({

crates/native-sidecar/src/state.rs

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -470,6 +470,7 @@ pub(crate) struct ActiveProcess {
470470
pub(crate) next_mapped_host_fd: u32,
471471
pub(crate) pending_execution_events: VecDeque<ActiveExecutionEvent>,
472472
pub(crate) pending_self_signal_exit: Option<i32>,
473+
pub(crate) pending_wasm_signals: VecDeque<i32>,
473474
pub(crate) child_processes: BTreeMap<String, ActiveProcess>,
474475
pub(crate) next_child_process_id: usize,
475476
pub(crate) http_servers: BTreeMap<u64, ActiveHttpServer>,

0 commit comments

Comments
 (0)