From d26d4050c85aaf82142bb1d42a939d3af4e6ecad Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Tue, 17 Mar 2026 16:33:47 -0700 Subject: [PATCH] simp --- rust-core/src/engine.rs | 56 +++++++++++++++++++++++++++++++++ rust-core/src/lib_c.rs | 70 +++-------------------------------------- 2 files changed, 61 insertions(+), 65 deletions(-) diff --git a/rust-core/src/engine.rs b/rust-core/src/engine.rs index 7b0b48297..f5da54a2c 100644 --- a/rust-core/src/engine.rs +++ b/rust-core/src/engine.rs @@ -62,6 +62,31 @@ pub enum NodeOutcome { Suspended { wait: WaitRequest }, } +#[derive(Debug, Deserialize)] +struct CallbackSendPayloadJson { + node: String, + #[serde(default)] + arg: Value, +} + +#[derive(Debug, Deserialize)] +struct CallbackNodeExecResultJsonWire { + update: Option, + #[serde(default)] + sends: Vec, +} + +#[derive(Debug, Deserialize)] +struct CallbackEnvelopeIn { + ok: bool, + #[serde(default)] + payload: Option, + #[serde(default)] + suspend: Option, + #[serde(default)] + error: Option, +} + type Task = Box; fn debug_enabled() -> bool { @@ -649,3 +674,34 @@ fn wrap_resume_arg(arg: Value, event: WaitEvent) -> Value { "__lg_resume_event__": event, }) } + +pub fn parse_callback_envelope_json( + raw: &str, + node_name: &str, +) -> Result, String> { + let parsed: CallbackEnvelopeIn = serde_json::from_str(raw) + .map_err(|e| format!("decode callback envelope for `{node_name}` failed: {e}"))?; + if !parsed.ok { + return Err(parsed + .error + .unwrap_or_else(|| format!("callback reported error for `{node_name}`"))); + } + if let Some(wait) = parsed.suspend { + return Ok(NodeOutcome::Suspended { wait }); + } + let payload = parsed + .payload + .ok_or_else(|| format!("callback payload missing for `{node_name}`"))?; + let sends = payload + .sends + .into_iter() + .map(|s| SendPayload { + node: s.node, + arg: s.arg, + }) + .collect(); + Ok(NodeOutcome::Completed(NodeExecResult { + update: payload.update, + sends, + })) +} diff --git a/rust-core/src/lib_c.rs b/rust-core/src/lib_c.rs index 42597cfe4..52e1d12d5 100644 --- a/rust-core/src/lib_c.rs +++ b/rust-core/src/lib_c.rs @@ -1,38 +1,12 @@ use crate::engine::{ - run_graph_json_with_callback, run_loop_block_on, run_loop_spawn, AnyOfCondition, Engine, - NodeExecResult, NodeOutcome, SendPayload, WaitRequest, + parse_callback_envelope_json, run_graph_json_with_callback, run_loop_block_on, run_loop_spawn, + AnyOfCondition, Engine, NodeOutcome, }; -use serde::Deserialize; use serde_json::Value; use std::ffi::{CStr, CString}; use std::os::raw::c_char; use std::sync::mpsc; -#[derive(Debug, Deserialize)] -struct SendPayloadJson { - node: String, - #[serde(default)] - arg: Value, -} - -#[derive(Debug, Deserialize)] -struct NodeExecResultJsonWire { - update: Option, - #[serde(default)] - sends: Vec, -} - -#[derive(Debug, Deserialize)] -struct CallbackEnvelopeIn { - ok: bool, - #[serde(default)] - payload: Option, - #[serde(default)] - suspend: Option, - #[serde(default)] - error: Option, -} - type CNodeCallback = unsafe extern "C" fn( user_data: libc::c_ulong, node: *mut c_char, @@ -40,9 +14,6 @@ type CNodeCallback = unsafe extern "C" fn( state_json: *mut c_char, ) -> *mut c_char; -#[derive(Clone, Copy)] -struct CUserData(libc::c_ulong); - fn cstr_to_str<'a>(ptr: *const c_char) -> Result<&'a str, String> { if ptr.is_null() { return Err("Received null pointer".to_string()); @@ -61,37 +32,6 @@ fn into_c_ptr(s: String) -> *mut c_char { } } -fn parse_c_callback_result( - raw: String, - node_name: &str, -) -> Result, String> { - let parsed: CallbackEnvelopeIn = serde_json::from_str(&raw) - .map_err(|e| format!("decode callback envelope for `{node_name}` failed: {e}"))?; - if !parsed.ok { - return Err(parsed - .error - .unwrap_or_else(|| format!("callback reported error for `{node_name}`"))); - } - if let Some(wait) = parsed.suspend { - return Ok(NodeOutcome::Suspended { wait }); - } - let payload = parsed - .payload - .ok_or_else(|| format!("callback payload missing for `{node_name}`"))?; - let sends = payload - .sends - .into_iter() - .map(|s| SendPayload { - node: s.node, - arg: s.arg, - }) - .collect(); - Ok(NodeOutcome::Completed(NodeExecResult { - update: payload.update, - sends, - })) -} - #[no_mangle] pub extern "C" fn rc_engine_new() -> *mut Engine { Box::into_raw(Box::new(Engine::new())) @@ -351,7 +291,7 @@ pub unsafe extern "C" fn rc_run_graph_json( } let (tx, rx) = mpsc::channel::>(); - let user_data = CUserData(user_data); + let user_data_bits = user_data; let run_engine = (*ptr).clone(); let submit = run_loop_spawn(async move { let callback_wrapper = move |node: String, @@ -370,7 +310,7 @@ pub unsafe extern "C" fn rc_run_graph_json( CString::new(state_json).map_err(|e| format!("invalid state JSON bytes: {e}"))?; let out_ptr = unsafe { callback( - user_data.0, + user_data_bits, node_c.as_ptr() as *mut c_char, arg_c.as_ptr() as *mut c_char, state_c.as_ptr() as *mut c_char, @@ -385,7 +325,7 @@ pub unsafe extern "C" fn rc_run_graph_json( unsafe { libc::free(out_ptr.cast()); } - parse_c_callback_result(out_raw, &node) + parse_callback_envelope_json(&out_raw, &node) }; let out = run_graph_json_with_callback( entry_point,