From ce4e5a16163bef75c7a553ff2dc54064ef92bd59 Mon Sep 17 00:00:00 2001 From: Philipp Oppermann Date: Tue, 13 Dec 2022 12:09:16 +0100 Subject: [PATCH] Send metadata through TCP for now --- apis/rust/node/src/daemon.rs | 9 +++++++-- apis/rust/node/src/lib.rs | 11 +++-------- binaries/daemon/src/listener.rs | 12 +++++++++--- binaries/daemon/src/main.rs | 22 ++++++++++++---------- libraries/core/src/daemon_messages.rs | 3 ++- libraries/message/src/lib.rs | 2 +- 6 files changed, 34 insertions(+), 25 deletions(-) diff --git a/apis/rust/node/src/daemon.rs b/apis/rust/node/src/daemon.rs index 7735d0b4..975a1d3c 100644 --- a/apis/rust/node/src/daemon.rs +++ b/apis/rust/node/src/daemon.rs @@ -52,11 +52,16 @@ impl ControlChannel { pub fn prepare_message( &mut self, output_id: DataId, - len: usize, + metadata: dora_message::Metadata<'static>, + data_len: usize, ) -> eyre::Result { tcp_send( &mut self.0, - &ControlRequest::PrepareOutputMessage { output_id, len }, + &ControlRequest::PrepareOutputMessage { + output_id, + metadata, + data_len, + }, ) .wrap_err("failed to send PrepareOutputMessage request to dora-daemon")?; match tcp_receive(&mut self.0) diff --git a/apis/rust/node/src/lib.rs b/apis/rust/node/src/lib.rs index 65825a49..f12e6a70 100644 --- a/apis/rust/node/src/lib.rs +++ b/apis/rust/node/src/lib.rs @@ -67,15 +67,11 @@ impl DoraNode { if !self.node_config.outputs.contains(&output_id) { eyre::bail!("unknown output"); } - let metadata = Metadata::from_parameters(self.hlc.new_timestamp(), parameters); - let serialized_metadata = metadata - .serialize() - .with_context(|| format!("failed to serialize `{}` message", output_id))?; - let full_len = serialized_metadata.len() + data_len; + let metadata = Metadata::from_parameters(self.hlc.new_timestamp(), parameters.into_owned()); let sample = self .control_channel - .prepare_message(output_id.clone(), full_len) + .prepare_message(output_id.clone(), metadata, data_len) .wrap_err("failed to prepare sample for output message")?; // map shared memory and fill in data @@ -86,8 +82,7 @@ impl DoraNode { .wrap_err("failed to open shared memory sample")?; let raw = unsafe { shared_memory.as_slice_mut() }; - raw[..serialized_metadata.len()].copy_from_slice(&serialized_metadata); - data(&mut raw[serialized_metadata.len()..]); + data(raw); } self.control_channel diff --git a/binaries/daemon/src/listener.rs b/binaries/daemon/src/listener.rs index 9dd4ba0d..de1b18a8 100644 --- a/binaries/daemon/src/listener.rs +++ b/binaries/daemon/src/listener.rs @@ -68,9 +68,15 @@ pub async fn handle_connection(mut connection: TcpStream, events_tx: mpsc::Sende } } daemon_messages::ControlRequest::Stopped => DaemonNodeEvent::Stopped, - daemon_messages::ControlRequest::PrepareOutputMessage { output_id, len } => { - DaemonNodeEvent::PrepareOutputMessage { output_id, len } - } + daemon_messages::ControlRequest::PrepareOutputMessage { + output_id, + metadata, + data_len, + } => DaemonNodeEvent::PrepareOutputMessage { + output_id, + metadata, + data_len, + }, daemon_messages::ControlRequest::SendOutMessage { id } => { DaemonNodeEvent::SendOutMessage { id } } diff --git a/binaries/daemon/src/main.rs b/binaries/daemon/src/main.rs index 77f68bd4..0fd9f4fd 100644 --- a/binaries/daemon/src/main.rs +++ b/binaries/daemon/src/main.rs @@ -3,7 +3,6 @@ use dora_core::{ daemon_messages::{self, ControlReply, DaemonCoordinatorEvent, DataflowId, SpawnDataflowNodes}, topics::DORA_COORDINATOR_PORT_DEFAULT, }; -use dora_message::{uhlc, Metadata}; use eyre::{bail, eyre, Context, ContextCompat}; use futures_concurrency::stream::Merge; use shared_memory::{Shmem, ShmemConf}; @@ -44,8 +43,7 @@ async fn run() -> eyre::Result<()> { struct Daemon { port: u16, - hlc: uhlc::HLC, - uninit_shared_memory: HashMap, + uninit_shared_memory: HashMap, Shmem)>, sent_out_shared_memory: HashMap, running: HashMap, @@ -74,7 +72,6 @@ impl Daemon { let daemon = Self { port, - hlc: uhlc::HLC::default(), uninit_shared_memory: Default::default(), sent_out_shared_memory: Default::default(), running: HashMap::new(), @@ -172,14 +169,18 @@ impl Daemon { }; let _ = reply_sender.send(ControlReply::Result(result)); } - DaemonNodeEvent::PrepareOutputMessage { output_id, len } => { + DaemonNodeEvent::PrepareOutputMessage { + output_id, + metadata, + data_len, + } => { let memory = ShmemConf::new() - .size(len) + .size(data_len) .create() .wrap_err("failed to allocate shared memory")?; let id = memory.get_os_id().to_owned(); self.uninit_shared_memory - .insert(id.clone(), (output_id, memory)); + .insert(id.clone(), (output_id, metadata, memory)); let reply = ControlReply::PreparedMessage { shared_memory_id: id.clone(), @@ -190,7 +191,7 @@ impl Daemon { } } DaemonNodeEvent::SendOutMessage { id } => { - let (output_id, memory) = self + let (output_id, metadata, memory) = self .uninit_shared_memory .remove(&id) .ok_or_else(|| eyre!("invalid shared memory id"))?; @@ -214,7 +215,7 @@ impl Daemon { if channel .send_async(daemon_messages::NodeEvent::Input { id: input_id.clone(), - metadata: Metadata::new(self.hlc.new_timestamp()), // TODO + metadata: metadata.clone(), data: unsafe { daemon_messages::InputData::new(id.clone()) }, }) .await @@ -272,7 +273,8 @@ pub enum Event { pub enum DaemonNodeEvent { PrepareOutputMessage { output_id: DataId, - len: usize, + metadata: dora_message::Metadata<'static>, + data_len: usize, }, SendOutMessage { id: MessageId, diff --git a/libraries/core/src/daemon_messages.rs b/libraries/core/src/daemon_messages.rs index 41f82977..7a1eaf4a 100644 --- a/libraries/core/src/daemon_messages.rs +++ b/libraries/core/src/daemon_messages.rs @@ -29,7 +29,8 @@ pub enum ControlRequest { }, PrepareOutputMessage { output_id: DataId, - len: usize, + metadata: Metadata<'static>, + data_len: usize, }, SendOutMessage { id: SharedMemoryId, diff --git a/libraries/message/src/lib.rs b/libraries/message/src/lib.rs index f0bc6c8c..891e854f 100644 --- a/libraries/message/src/lib.rs +++ b/libraries/message/src/lib.rs @@ -22,7 +22,7 @@ pub struct MetadataParameters<'a> { } impl MetadataParameters<'_> { - fn into_owned(self) -> MetadataParameters<'static> { + pub fn into_owned(self) -> MetadataParameters<'static> { MetadataParameters { open_telemetry_context: self.open_telemetry_context.into_owned().into(), ..self