Allows queueing of multiple messages on the TCP socket, which considerably improves throughput.tags/v0.2.0-candidate
| @@ -151,19 +151,18 @@ impl ControlChannel { | |||||
| metadata: Metadata<'static>, | metadata: Metadata<'static>, | ||||
| data: Vec<u8>, | data: Vec<u8>, | ||||
| ) -> eyre::Result<()> { | ) -> eyre::Result<()> { | ||||
| let request = DaemonRequest::SendMessage { | |||||
| output_id, | |||||
| metadata, | |||||
| data, | |||||
| }; | |||||
| let reply = self | let reply = self | ||||
| .channel | .channel | ||||
| .request(&DaemonRequest::SendMessage { | |||||
| output_id, | |||||
| metadata, | |||||
| data, | |||||
| }) | |||||
| .wrap_err("failed to send SendEmptyMessage request to dora-daemon")?; | |||||
| .request(&request) | |||||
| .wrap_err("failed to send SendMessage request to dora-daemon")?; | |||||
| match reply { | match reply { | ||||
| dora_core::daemon_messages::DaemonReply::Result(result) => { | |||||
| result.map_err(|err| eyre!(err)) | |||||
| } | |||||
| other => bail!("unexpected SendEmptyMessage reply: {other:?}"), | |||||
| dora_core::daemon_messages::DaemonReply::Empty => Ok(()), | |||||
| other => bail!("unexpected SendMessage reply: {other:?}"), | |||||
| } | } | ||||
| } | } | ||||
| } | } | ||||
| @@ -7,8 +7,12 @@ use std::{ | |||||
| pub fn request(connection: &mut TcpStream, request: &DaemonRequest) -> eyre::Result<DaemonReply> { | pub fn request(connection: &mut TcpStream, request: &DaemonRequest) -> eyre::Result<DaemonReply> { | ||||
| send_message(connection, request)?; | send_message(connection, request)?; | ||||
| receive_reply(connection) | |||||
| .and_then(|reply| reply.ok_or_else(|| eyre!("server disconnected unexpectedly"))) | |||||
| if request.expects_tcp_reply() { | |||||
| receive_reply(connection) | |||||
| .and_then(|reply| reply.ok_or_else(|| eyre!("server disconnected unexpectedly"))) | |||||
| } else { | |||||
| Ok(DaemonReply::Empty) | |||||
| } | |||||
| } | } | ||||
| fn send_message(connection: &mut TcpStream, message: &DaemonRequest) -> eyre::Result<()> { | fn send_message(connection: &mut TcpStream, message: &DaemonRequest) -> eyre::Result<()> { | ||||
| @@ -316,7 +316,10 @@ where | |||||
| .send_daemon_event(event) | .send_daemon_event(event) | ||||
| .await | .await | ||||
| .map_err(|_| "failed to receive send_empty_message reply".to_owned()); | .map_err(|_| "failed to receive send_empty_message reply".to_owned()); | ||||
| self.send_reply(DaemonReply::Result(result)) | |||||
| if let Err(err) = result { | |||||
| tracing::warn!("{err:?}"); | |||||
| } | |||||
| self.send_reply(DaemonReply::Empty) | |||||
| .await | .await | ||||
| .wrap_err("failed to send SendEmptyMessage reply")?; | .wrap_err("failed to send SendEmptyMessage reply")?; | ||||
| } | } | ||||
| @@ -75,6 +75,10 @@ impl super::Connection for TcpConnection { | |||||
| } | } | ||||
| async fn send_reply(&mut self, message: DaemonReply) -> eyre::Result<()> { | async fn send_reply(&mut self, message: DaemonReply) -> eyre::Result<()> { | ||||
| if matches!(message, DaemonReply::Empty) { | |||||
| // don't send empty replies | |||||
| return Ok(()); | |||||
| } | |||||
| let serialized = | let serialized = | ||||
| bincode::serialize(&message).wrap_err("failed to serialize DaemonReply")?; | bincode::serialize(&message).wrap_err("failed to serialize DaemonReply")?; | ||||
| tcp_send(&mut self.0, &serialized) | tcp_send(&mut self.0, &serialized) | ||||
| @@ -59,6 +59,16 @@ pub enum DaemonRequest { | |||||
| }, | }, | ||||
| } | } | ||||
| impl DaemonRequest { | |||||
| pub fn expects_tcp_reply(&self) -> bool { | |||||
| #[allow(clippy::match_like_matches_macro)] | |||||
| match self { | |||||
| DaemonRequest::SendMessage { .. } => false, | |||||
| _ => true, | |||||
| } | |||||
| } | |||||
| } | |||||
| type SharedMemoryId = String; | type SharedMemoryId = String; | ||||
| #[derive(Debug, serde::Serialize, serde::Deserialize)] | #[derive(Debug, serde::Serialize, serde::Deserialize)] | ||||
| @@ -66,6 +76,7 @@ pub enum DaemonReply { | |||||
| Result(Result<(), String>), | Result(Result<(), String>), | ||||
| PreparedMessage { shared_memory_id: SharedMemoryId }, | PreparedMessage { shared_memory_id: SharedMemoryId }, | ||||
| NextEvents(Vec<NodeEvent>), | NextEvents(Vec<NodeEvent>), | ||||
| Empty, | |||||
| } | } | ||||
| #[derive(Debug, serde::Serialize, serde::Deserialize)] | #[derive(Debug, serde::Serialize, serde::Deserialize)] | ||||