oxide_core/builtins/
agent.rs

1//! `$262.agent`: Test262's multi-agent host, as QuickJS's `run-test262`
2//! implements it. Each agent is a thread with its own runtime running a script
3//! through the embedder's [`ScriptRunner`]; a broadcast hands every agent the
4//! shared bytes of a `SharedArrayBuffer` (an `Arc`, so no heap crosses threads)
5//! and waits until each has taken them; reports are strings in a shared queue.
6
7use std::collections::VecDeque;
8use std::sync::{Arc, Condvar, Mutex, MutexGuard};
9use std::thread::JoinHandle;
10use std::time::{Duration, Instant};
11
12use super::{Native, define, plain_object};
13use crate::heap::HeapError;
14use crate::heap::buffer::{Buffer, SharedBytes, Store};
15use crate::heap::gc::HeapLimits;
16use crate::heap::object::{InternalData, InternalTag, ObjectKind};
17use crate::runtime::{Intrinsic, Runtime};
18use crate::shape::PropertyFlags;
19use crate::value::Value;
20use crate::vm::native::CallSite;
21use crate::vm::{Cx, Throw, error, ops, typed};
22
23/// Runs a script's source in a runtime (the embedder compiles it), then
24/// its jobs; the error is the uncaught exception's text.
25pub type ScriptRunner = fn(&mut Runtime, &str) -> Result<(), String>;
26
27/// A broadcast: the shared bytes, the buffer's maximum length, the value.
28#[derive(Clone)]
29struct Message {
30    bytes: Arc<SharedBytes>,
31    max: Option<usize>,
32    value: f64,
33}
34
35#[derive(Default)]
36struct HubState {
37    /// Each agent's pending broadcast.
38    pending: Vec<Option<Message>>,
39    reports: VecDeque<String>,
40    threads: Vec<JoinHandle<()>>,
41}
42
43/// What the main agent and its agents share.
44#[derive(Default)]
45pub struct AgentHub {
46    state: Mutex<HubState>,
47    changed: Condvar,
48}
49
50impl AgentHub {
51    fn lock(&self) -> MutexGuard<'_, HubState> {
52        self.state
53            .lock()
54            .unwrap_or_else(std::sync::PoisonError::into_inner)
55    }
56
57    /// Joins every agent thread.
58    pub fn join(&self) {
59        let threads = std::mem::take(&mut self.lock().threads);
60        for thread in threads {
61            let _ = thread.join();
62        }
63    }
64}
65
66/// The agent side of a runtime: its hub and position.
67pub struct AgentSelf {
68    hub: Arc<AgentHub>,
69    /// The `receiveBroadcast` callback (a fixed root), until called.
70    callback: Option<Value>,
71}
72
73static START: std::sync::OnceLock<Instant> = std::sync::OnceLock::new();
74
75pub(super) fn monotonic_ms() -> f64 {
76    START.get_or_init(Instant::now).elapsed().as_secs_f64() * 1000.0
77}
78
79/// Defines `$262.agent` on `host` for the main agent (`main`) or an agent.
80pub(crate) fn install(cx: &mut Cx<'_>, host: crate::ObjectId, main: bool) -> Result<(), HeapError> {
81    let prototype = cx.realm.intrinsic(Intrinsic::ObjectPrototype);
82    let agent = plain_object(cx, Some(prototype), 8)?;
83    let all = PropertyFlags::DEFAULT;
84    let natives: &[Native] = if main {
85        &[
86            Native::AgentStart,
87            Native::AgentBroadcast,
88            Native::AgentGetReport,
89            Native::AgentSleep,
90            Native::AgentMonotonicNow,
91        ]
92    } else {
93        &[
94            Native::AgentReceiveBroadcast,
95            Native::AgentReport,
96            Native::AgentLeaving,
97            Native::AgentSleep,
98            Native::AgentMonotonicNow,
99        ]
100    };
101    for &native in natives {
102        let function = super::native_function(cx, native)?;
103        define(cx, agent, native.js_name(), Value::object(function), all)?;
104    }
105    define(cx, host, "agent", Value::object(agent), all)
106}
107
108fn hub(cx: &mut Cx<'_>) -> Arc<AgentHub> {
109    cx.host
110        .agent_hub
111        .get_or_insert_with(Default::default)
112        .clone()
113}
114
115/// `$262.agent.start(source)`.
116pub(super) fn start(cx: &mut Cx<'_>, site: CallSite) -> Result<Value, Throw> {
117    let source = site.arg(cx, 0);
118    let source = ops::to_string(cx, source)?;
119    let source = cx.heap.string_text(source.as_string().unwrap());
120    let Some(runner) = cx.host.script_runner else {
121        return Err(error::type_error(
122            cx,
123            "agents are not supported by this host",
124        ));
125    };
126    let hub = hub(cx);
127    let index = {
128        let mut state = hub.lock();
129        state.pending.push(None);
130        state.pending.len() - 1
131    };
132    let agent_hub = hub.clone();
133    let compile = cx.host.compile;
134    let thread = std::thread::spawn(move || run_agent(agent_hub, index, &source, runner, compile));
135    hub.lock().threads.push(thread);
136    Ok(Value::UNDEFINED)
137}
138
139/// An agent thread: the script, its jobs, then one broadcast if it asked
140/// for one (QuickJS's `agent_start`).
141fn run_agent(
142    hub: Arc<AgentHub>,
143    index: usize,
144    source: &str,
145    runner: ScriptRunner,
146    compile: Option<crate::vm::compile::CompileFn>,
147) {
148    let Ok(mut runtime) = Runtime::new(HeapLimits::default()) else {
149        return;
150    };
151    runtime.host.compile = compile;
152    if runtime.install_print().is_err() || runtime.install_agent_host(hub.clone(), index).is_err() {
153        return;
154    }
155    if let Err(message) = runner(&mut runtime, source) {
156        eprintln!("{message}");
157        return;
158    }
159    loop {
160        if runtime.run_jobs().is_err() {
161            return;
162        }
163        let Some(callback) = runtime.take_agent_callback() else {
164            return;
165        };
166        let message = {
167            let mut state = hub.lock();
168            loop {
169                if let Some(message) = state.pending[index].take() {
170                    hub.changed.notify_all();
171                    break message;
172                }
173                state = hub
174                    .changed
175                    .wait(state)
176                    .unwrap_or_else(std::sync::PoisonError::into_inner);
177            }
178        };
179        if runtime
180            .call_agent_callback(callback, message.bytes, message.max, message.value)
181            .is_err()
182        {
183            return;
184        }
185    }
186}
187
188/// `$262.agent.broadcast(sab, value)`: blocks until every agent took it.
189pub(super) fn broadcast(cx: &mut Cx<'_>, site: CallSite) -> Result<Value, Throw> {
190    let buffer = site.arg(cx, 0);
191    let shared = match buffer.as_object() {
192        Some(object) if cx.heap.internal_tag(object) == InternalTag::SharedArrayBuffer => {
193            let record = typed::buffer(cx.heap, object);
194            match &record.store {
195                Store::Shared(bytes) => (bytes.clone(), record.max),
196                Store::Owned(_) => unreachable!("a shared buffer"),
197            }
198        }
199        _ => return Err(error::object_expected(cx, "ArrayBuffer")),
200    };
201    let value = site.arg(cx, 1);
202    let value = ops::to_number(cx, value)?;
203    let value = f64::from(crate::vm::number::to_int32(value));
204    let hub = hub(cx);
205    let mut state = hub.lock();
206    let message = Message {
207        bytes: shared.0,
208        max: shared.1,
209        value,
210    };
211    for slot in &mut state.pending {
212        *slot = Some(message.clone());
213    }
214    hub.changed.notify_all();
215    while state.pending.iter().any(Option::is_some) {
216        state = hub
217            .changed
218            .wait(state)
219            .unwrap_or_else(std::sync::PoisonError::into_inner);
220    }
221    Ok(Value::UNDEFINED)
222}
223
224/// `$262.agent.getReport()`: the oldest report, or `null`.
225pub(super) fn get_report(cx: &mut Cx<'_>, _site: CallSite) -> Result<Value, Throw> {
226    let hub = hub(cx);
227    let report = hub.lock().reports.pop_front();
228    match report {
229        Some(text) => super::string_text(cx, &text),
230        None => Ok(Value::NULL),
231    }
232}
233
234pub(super) fn sleep(cx: &mut Cx<'_>, site: CallSite) -> Result<Value, Throw> {
235    let time = site.arg(cx, 0);
236    let time = ops::to_number(cx, time)?;
237    if time > 0.0 {
238        std::thread::sleep(Duration::from_secs_f64(time / 1000.0));
239    }
240    Ok(Value::UNDEFINED)
241}
242
243pub(super) fn monotonic_now(_cx: &mut Cx<'_>, _site: CallSite) -> Result<Value, Throw> {
244    Ok(Value::number(monotonic_ms().floor()))
245}
246
247/// `$262.agent.receiveBroadcast(callback)` in an agent.
248pub(super) fn receive_broadcast(cx: &mut Cx<'_>, site: CallSite) -> Result<Value, Throw> {
249    let callback = site.arg(cx, 0);
250    if !ops::is_callable(cx.heap, callback) {
251        return Err(error::type_error(cx, "expecting function"));
252    }
253    cx.heap.add_fixed_root(callback);
254    if let Some(agent) = &mut cx.host.agent {
255        agent.callback = Some(callback);
256    }
257    Ok(Value::UNDEFINED)
258}
259
260/// `$262.agent.report(value)` in an agent.
261pub(super) fn report(cx: &mut Cx<'_>, site: CallSite) -> Result<Value, Throw> {
262    let value = site.arg(cx, 0);
263    let text = ops::to_string(cx, value)?;
264    let text = cx.heap.string_text(text.as_string().unwrap());
265    if let Some(agent) = &cx.host.agent {
266        agent.hub.lock().reports.push_back(text);
267    }
268    Ok(Value::UNDEFINED)
269}
270
271pub(super) fn leaving(_cx: &mut Cx<'_>, _site: CallSite) -> Result<Value, Throw> {
272    Ok(Value::UNDEFINED)
273}
274
275impl Runtime {
276    /// Lets `$262.agent.start` run agents' scripts through `runner`.
277    pub fn set_script_runner(&mut self, runner: ScriptRunner) {
278        self.host.script_runner = Some(runner);
279    }
280
281    /// Waits for every agent this runtime started.
282    pub fn join_agents(&mut self) {
283        if let Some(hub) = self.host.agent_hub.clone() {
284            hub.join();
285        }
286    }
287
288    /// Defines an agent's `$262` (only `agent`).
289    fn install_agent_host(&mut self, hub: Arc<AgentHub>, index: usize) -> Result<(), HeapError> {
290        let _ = index;
291        self.host.agent = Some(AgentSelf {
292            hub,
293            callback: None,
294        });
295        let mut cx = self.cx();
296        let mark = cx.heap.roots().mark();
297        let result = (|| {
298            let prototype = cx.realm.intrinsic(Intrinsic::ObjectPrototype);
299            let host = plain_object(&mut cx, Some(prototype), 2)?;
300            install(&mut cx, host, false)?;
301            let global = cx.realm.global();
302            define(
303                &mut cx,
304                global,
305                "$262",
306                Value::object(host),
307                PropertyFlags::DEFAULT,
308            )
309        })();
310        cx.heap.roots().truncate(mark);
311        result
312    }
313
314    fn take_agent_callback(&mut self) -> Option<Value> {
315        self.host
316            .agent
317            .as_mut()
318            .and_then(|agent| agent.callback.take())
319    }
320
321    fn call_agent_callback(
322        &mut self,
323        callback: Value,
324        bytes: Arc<SharedBytes>,
325        max: Option<usize>,
326        value: f64,
327    ) -> Result<(), Throw> {
328        let cx = self.cx();
329        let prototype = cx.realm.intrinsic(Intrinsic::SharedArrayBufferPrototype);
330        let buffer = Buffer {
331            store: Store::Shared(bytes),
332            detached: false,
333            max,
334        };
335        let object = cx
336            .heap
337            .alloc_internal(
338                ObjectKind::Builtin,
339                InternalTag::SharedArrayBuffer,
340                Some(prototype),
341                0,
342                &[],
343                InternalData::Buffer(Box::new(buffer)),
344                0,
345            )
346            .map_err(|error| error::heap_error(&cx, error))?;
347        let handle = self.hold(Value::object(object));
348        let result = self.call(
349            callback,
350            Value::UNDEFINED,
351            &[Value::object(object), Value::number(value)],
352        );
353        self.release(handle);
354        result.map(drop)
355    }
356}
oxide_core/builtins/agent.rs

oxide_core/builtins/
agent.rs
