astatin3/unshell
1
0
Fork 0
mirror of https://github.com/Astatin3/unshell.git synced 2026-06-08 22:38:01 -06:00
Code Issues Packages Projects Releases Wiki Activity

Avoid cloning call payloads in runtimes

Browse Source
This commit is contained in:
Michael Mikovsky
2026-04-25 21:14:02 -06:00
parent 31a0bd39b0
commit 80611a4d74
2 changed files with 171 additions and 167 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/protocol/tree/call.rs
+59 -49
View File
@@ -256,60 +256,70 @@ where
let mut runtime = RuntimeOutcome::default(); let mut runtime = RuntimeOutcome::default();
match event { match event {
LocalEvent::Call { header, message } => { LocalEvent::Call { header, message } => {
let incoming = IncomingCall { let CallMessage {
header, procedure_id,
message: message.clone(), data,
}; response_hook,
match self.leaf.dispatch_call(incoming) { } = message;
Ok(CallReply::Reply(bytes)) => { let fault_hook = response_hook.as_ref();
if let Some(hook) = message.response_hook { let incoming = IncomingCall {
runtime.frames.extend(self.send_reply_data( header,
hook, message: CallMessage {
message.procedure_id, procedure_id: procedure_id.clone(),
bytes, data,
true, response_hook: response_hook.clone(),
)?); },
};
match self.leaf.dispatch_call(incoming) {
Ok(CallReply::Reply(bytes)) => {
if let Some(hook) = response_hook {
runtime.frames.extend(self.send_reply_data(
hook,
procedure_id,
bytes,
true,
)?);
}
}
Ok(CallReply::NoReply) => {}
Err(error) => {
runtime
.frames
.extend(self.emit_internal_fault_if_possible(fault_hook)?);
return Err(LeafRuntimeError::Dispatch(error));
}
} }
} }
Ok(CallReply::NoReply) => {} LocalEvent::Data {
Err(error) => {
runtime
.frames
.extend(self.emit_internal_fault_if_possible(&message)?);
return Err(LeafRuntimeError::Dispatch(error));
}
}
}
LocalEvent::Data {
header,
message,
hook_key,
} => {
let outgoing = self
.leaf
.on_data(IncomingData {
header, header,
message, message,
hook_key, hook_key,
}) } => {
.map_err(LeafRuntimeError::Leaf)?; let outgoing = self
runtime.frames.extend(self.emit_outgoing(outgoing)?.frames); .leaf
} .on_data(IncomingData {
LocalEvent::Fault { header,
header, message,
message, hook_key,
hook_key, })
} => { .map_err(LeafRuntimeError::Leaf)?;
self.leaf runtime.frames.extend(self.emit_outgoing(outgoing)?.frames);
.on_fault(IncomingFault { }
LocalEvent::Fault {
header, header,
fault: message, message,
hook_key, hook_key,
}) } => {
.map_err(LeafRuntimeError::Leaf)?; self.leaf
} .on_fault(IncomingFault {
} header,
fault: message,
hook_key,
})
.map_err(LeafRuntimeError::Leaf)?;
}
}
Ok(runtime) Ok(runtime)
} }
@@ -355,9 +365,9 @@ where
fn emit_internal_fault_if_possible( fn emit_internal_fault_if_possible(
&mut self, &mut self,
message: &CallMessage, hook: Option<&HookTarget>,
) -> Result<Vec<FrameBytes>, LeafRuntimeError<<L as CallLeaf>::Error>> { ) -> Result<Vec<FrameBytes>, LeafRuntimeError<<L as CallLeaf>::Error>> {
let Some(hook) = message.response_hook.as_ref() else { let Some(hook) = hook else {
return Ok(Vec::new()); return Ok(Vec::new());
}; };
let key = HookKey::new(hook.return_path.clone(), hook.hook_id); let key = HookKey::new(hook.return_path.clone(), hook.hook_id);
src/protocol/tree/procedure.rs
+112 -118
View File
@@ -23,8 +23,8 @@ use rkyv::{Archive, rancor::Error};
use crate::protocol::{CallMessage, FrameBytes, HookTarget, ProtocolFault}; use crate::protocol::{CallMessage, FrameBytes, HookTarget, ProtocolFault};
use super::{ use super::{
DispatchError, Endpoint, EndpointError, HookKey, IncomingCall, IncomingData, IncomingFault, DispatchError, Endpoint, EndpointError, HookKey, IncomingData, IncomingFault, Ingress,
Ingress, LocalEvent, OutgoingData, ProtocolEndpoint, ProtocolLeaf, decode_call_input, LocalEvent, OutgoingData, ProtocolEndpoint, ProtocolLeaf, decode_call_input,
}; };
/// Generated metadata for one stateful procedure bound to one leaf type. /// Generated metadata for one stateful procedure bound to one leaf type.
@@ -305,7 +305,7 @@ where
Ok(effect) => self.ensure_terminal_packet(&key, effect), Ok(effect) => self.ensure_terminal_packet(&key, effect),
Err(error) => { Err(error) => {
let _ = P::close(&mut self.leaf, session); let _ = P::close(&mut self.leaf, session);
frames.extend(self.emit_internal_fault(&key)?); frames.extend(self.emit_internal_fault(Some(key.clone()))?);
let _ = error; let _ = error;
continue; continue;
} }
@@ -357,130 +357,129 @@ where
let mut runtime = ProcedureRuntimeOutcome::default(); let mut runtime = ProcedureRuntimeOutcome::default();
match event { match event {
LocalEvent::Call { header, message } => { LocalEvent::Call { header, message } => {
if message.procedure_id != P::procedure_id() { if message.procedure_id != P::procedure_id() {
runtime runtime.frames.extend(
.frames self.emit_internal_fault_if_possible(message.response_hook.as_ref())?,
.extend(self.emit_internal_fault_if_possible(&message)?); );
return Ok(runtime); return Ok(runtime);
} }
if message.response_hook.is_none() { let Some(hook) = message.response_hook.as_ref() else {
return Ok(runtime); return Ok(runtime);
} };
let hook_key = HookKey::new(hook.return_path.clone(), hook.hook_id);
let session = match self.open_session(IncomingCall { let session = match self.open_session(header, message) {
header, Ok(session) => session,
message: message.clone(), Err(error) => {
}) { runtime.frames.extend(
Ok(session) => session, self.emit_internal_fault(Some(hook_key.clone()))?,
Err(error) => { );
runtime let _ = error;
.frames return Ok(runtime);
.extend(self.emit_internal_fault_if_possible(&message)?); }
let _ = error; };
return Ok(runtime);
self.leaf.procedure_sessions().insert(hook_key, session);
} }
}; LocalEvent::Data {
if let Some(hook) = message.response_hook {
self.leaf
.procedure_sessions()
.insert(HookKey::new(hook.return_path, hook.hook_id), session);
}
}
LocalEvent::Data {
header,
message,
hook_key,
} => {
let Some(mut session) = self.leaf.procedure_sessions().remove(&hook_key) else {
return Ok(runtime);
};
let effect = match P::on_data(
&mut self.leaf,
&mut session,
IncomingData {
header, header,
message, message,
hook_key: hook_key.clone(), hook_key,
}, } => {
) { let Some(mut session) = self.leaf.procedure_sessions().remove(&hook_key) else {
Ok(effect) => self.ensure_terminal_packet(&hook_key, effect), return Ok(runtime);
Err(error) => { };
let _ = P::close(&mut self.leaf, session); let effect = match P::on_data(
runtime.frames.extend(self.emit_internal_fault(&hook_key)?); &mut self.leaf,
let _ = error; &mut session,
return Ok(runtime); IncomingData {
} header,
}; message,
match self.emit_outgoing(effect.outgoing) { hook_key: hook_key.clone(),
Ok(outgoing) => runtime.frames.extend(outgoing.frames), },
Err(error) => { ) {
Ok(effect) => self.ensure_terminal_packet(&hook_key, effect),
Err(error) => {
let _ = P::close(&mut self.leaf, session);
runtime.frames.extend(self.emit_internal_fault(Some(hook_key.clone()))?);
let _ = error;
return Ok(runtime);
}
};
match self.emit_outgoing(effect.outgoing) {
Ok(outgoing) => runtime.frames.extend(outgoing.frames),
Err(error) => {
if !effect.close_session {
self.leaf.procedure_sessions().insert(hook_key, session);
} else {
let _ = P::close(&mut self.leaf, session);
}
return Err(error);
}
}
if !effect.close_session { if !effect.close_session {
self.leaf.procedure_sessions().insert(hook_key, session); self.leaf.procedure_sessions().insert(hook_key, session);
} else { } else {
let _ = P::close(&mut self.leaf, session); let _ = P::close(&mut self.leaf, session);
} }
return Err(error); }
LocalEvent::Fault {
header,
message,
hook_key,
} => {
let Some(mut session) = self.leaf.procedure_sessions().remove(&hook_key) else {
return Ok(runtime);
};
let on_fault_result = P::on_fault(
&mut self.leaf,
&mut session,
IncomingFault {
header,
fault: message,
hook_key: hook_key.clone(),
},
);
let close_result = P::close(&mut self.leaf, session);
if let Err(error) = on_fault_result {
let _ = close_result;
runtime.frames.extend(self.emit_internal_fault(Some(hook_key.clone()))?);
let _ = error;
return Ok(runtime);
}
if let Err(error) = close_result {
runtime.frames.extend(self.emit_internal_fault(Some(hook_key))?);
let _ = error;
return Ok(runtime);
}
} }
} }
if !effect.close_session {
self.leaf.procedure_sessions().insert(hook_key, session);
} else {
let _ = P::close(&mut self.leaf, session);
}
}
LocalEvent::Fault {
header,
message,
hook_key,
} => {
let Some(mut session) = self.leaf.procedure_sessions().remove(&hook_key) else {
return Ok(runtime);
};
let on_fault_result = P::on_fault(
&mut self.leaf,
&mut session,
IncomingFault {
header,
fault: message,
hook_key: hook_key.clone(),
},
);
let close_result = P::close(&mut self.leaf, session);
if let Err(error) = on_fault_result {
let _ = close_result;
runtime.frames.extend(self.emit_internal_fault(&hook_key)?);
let _ = error;
return Ok(runtime);
}
if let Err(error) = close_result {
runtime.frames.extend(self.emit_internal_fault(&hook_key)?);
let _ = error;
return Ok(runtime);
}
}
}
Ok(runtime) Ok(runtime)
} }
} }
} }
fn open_session(&mut self, call: IncomingCall) -> Result<P, DispatchError<P::Error>> { fn open_session(
let input = decode_call_input::<P::Input>(call.message.data.as_slice()) &mut self,
.map_err(DispatchError::Decode)?; header: crate::protocol::PacketHeader,
message: CallMessage,
) -> Result<P, DispatchError<P::Error>> {
let CallMessage {
procedure_id,
data,
response_hook,
} = message;
let input = decode_call_input::<P::Input>(data.as_slice()).map_err(DispatchError::Decode)?;
P::open( P::open(
&mut self.leaf, &mut self.leaf,
super::Call { super::Call {
input, input,
caller_path: call.header.src_path, caller_path: header.src_path,
procedure_id: call.message.procedure_id, procedure_id,
dst_leaf: call.header.dst_leaf, dst_leaf: header.dst_leaf,
response_hook: call response_hook: response_hook.map(|hook| HookKey::new(hook.return_path, hook.hook_id)),
.message
.response_hook
.map(|hook| HookKey::new(hook.return_path, hook.hook_id)),
}, },
) )
.map_err(DispatchError::Handler) .map_err(DispatchError::Handler)
@@ -510,29 +509,24 @@ where
/// declared a response hook. /// declared a response hook.
pub fn emit_internal_fault_if_possible( pub fn emit_internal_fault_if_possible(
&mut self, &mut self,
message: &CallMessage, hook: Option<&HookTarget>,
) -> Result<Vec<FrameBytes>, ProcedureRuntimeError<P::Error>> { ) -> Result<Vec<FrameBytes>, ProcedureRuntimeError<P::Error>> {
let Some(HookTarget { let Some(HookTarget { return_path, hook_id }) = hook else {
return_path,
hook_id,
}) = message.response_hook.as_ref()
else {
return Ok(Vec::new()); return Ok(Vec::new());
}; };
let outcome = self.endpoint.emit_fault_if_possible( let outcome = self
Some(HookKey::new(return_path.clone(), *hook_id)), .endpoint
ProtocolFault::INTERNAL_ERROR, .emit_fault_if_possible(Some(HookKey::new(return_path.clone(), *hook_id)), ProtocolFault::INTERNAL_ERROR)?;
)?;
Ok(self.process_endpoint_outcome(outcome)?.frames) Ok(self.process_endpoint_outcome(outcome)?.frames)
} }
fn emit_internal_fault( fn emit_internal_fault(
&mut self, &mut self,
hook_key: &HookKey, hook_key: Option<HookKey>,
) -> Result<Vec<FrameBytes>, ProcedureRuntimeError<P::Error>> { ) -> Result<Vec<FrameBytes>, ProcedureRuntimeError<P::Error>> {
let outcome = self let outcome = self
.endpoint .endpoint
.emit_fault_if_possible(Some(hook_key.clone()), ProtocolFault::INTERNAL_ERROR)?; .emit_fault_if_possible(hook_key, ProtocolFault::INTERNAL_ERROR)?;
Ok(self.process_endpoint_outcome(outcome)?.frames) Ok(self.process_endpoint_outcome(outcome)?.frames)
} }