diff --git a/newsfragments/382.fixed.md b/newsfragments/382.fixed.md new file mode 100644 index 000000000..cd0d7237f --- /dev/null +++ b/newsfragments/382.fixed.md @@ -0,0 +1 @@ +Remove blocking async from overlay events loop. diff --git a/trin-history/src/events.rs b/trin-history/src/events.rs index cb7d229a5..3e99d32e9 100644 --- a/trin-history/src/events.rs +++ b/trin-history/src/events.rs @@ -16,7 +16,7 @@ impl HistoryEvents { loop { tokio::select! { Some(talk_request) = self.event_rx.recv() => { - self.handle_history_talk_request(talk_request).await; + self.handle_history_talk_request(talk_request); } Some(event) = self.utp_listener_rx.recv() => { if let Err(err) = self.network.overlay.process_utp_event(event) { @@ -28,25 +28,27 @@ impl HistoryEvents { } /// Handle history network TalkRequest event - async fn handle_history_talk_request(&self, talk_request: TalkRequest) { - let reply = match self - .network - .overlay - .process_one_request(&talk_request) - .instrument(tracing::info_span!("history_network")) - .await - { - Ok(response) => { - debug!("Sending reply: {:?}", response); - Message::from(response).into() - } - Err(error) => { - error!("Failed to process portal history event: {error}"); - error.to_string().into_bytes() + fn handle_history_talk_request(&self, talk_request: TalkRequest) { + let network = Arc::clone(&self.network); + tokio::spawn(async move { + let reply = match network + .overlay + .process_one_request(&talk_request) + .instrument(tracing::info_span!("history_network")) + .await + { + Ok(response) => { + debug!("Sending reply: {:?}", response); + Message::from(response).into() + } + Err(error) => { + error!("Failed to process portal history event: {error}"); + error.to_string().into_bytes() + } + }; + if let Err(error) = talk_request.respond(reply) { + warn!("Failed to send reply: {error}"); } - }; - if let Err(error) = talk_request.respond(reply) { - warn!("Failed to send reply: {error}"); - } + }); } } diff --git a/trin-state/src/events.rs b/trin-state/src/events.rs index 13462ffb5..c382bb53b 100644 --- a/trin-state/src/events.rs +++ b/trin-state/src/events.rs @@ -16,7 +16,7 @@ impl StateEvents { loop { tokio::select! { Some(talk_request) = self.event_rx.recv() => { - self.handle_state_talk_request(talk_request).await; + self.handle_state_talk_request(talk_request); } Some(event) = self.utp_listener_rx.recv() => { if let Err(err) = self.network.overlay.process_utp_event(event) { @@ -28,25 +28,27 @@ impl StateEvents { } /// Handle state network TalkRequest event - async fn handle_state_talk_request(&self, talk_request: TalkRequest) { - let reply = match self - .network - .overlay - .process_one_request(&talk_request) - .instrument(tracing::info_span!("state_network")) - .await - { - Ok(response) => { - debug!("Sending reply: {:?}", response); - Message::from(response).into() - } - Err(error) => { - error!("Failed to process portal state event: {error}"); - error.to_string().into_bytes() + fn handle_state_talk_request(&self, talk_request: TalkRequest) { + let network = Arc::clone(&self.network); + tokio::spawn(async move { + let reply = match network + .overlay + .process_one_request(&talk_request) + .instrument(tracing::info_span!("state_network")) + .await + { + Ok(response) => { + debug!("Sending reply: {:?}", response); + Message::from(response).into() + } + Err(error) => { + error!("Failed to process portal state event: {error}"); + error.to_string().into_bytes() + } + }; + if let Err(error) = talk_request.respond(reply) { + warn!("Failed to send reply: {error}"); } - }; - if let Err(error) = talk_request.respond(reply) { - warn!("Failed to send reply: {error}"); - } + }); } }