Miroirs/pinnacle
1
0
Fork 0
mirror of https://github.com/pinnacle-comp/pinnacle.git synced 2024-12-26 21:58:10 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge main deadlock fix

Browse source
This commit is contained in:
Ottatop 2024-02-23 16:30:59 -06:00
commit 5db8b1c54d
3 changed files with 18 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
api/rust/src/input.rs
View file

@ -178,6 +178,7 @@ impl Input {
while let Some(Ok(_response)) = stream.next().await { while let Some(Ok(_response)) = stream.next().await {
action(); action();
tokio::task::yield_now().await;
} }
} }
.boxed(), .boxed(),
@ -233,6 +234,7 @@ impl Input {
while let Some(Ok(_response)) = stream.next().await { while let Some(Ok(_response)) = stream.next().await {
action(); action();
tokio::task::yield_now().await;
} }
} }
.boxed(), .boxed(),

40
api/rust/src/lib.rs
View file

@ -81,7 +81,11 @@
use std::sync::OnceLock; use std::sync::OnceLock;
use futures::{future::BoxFuture, stream::FuturesUnordered, Future, StreamExt}; use futures::{
future::{BoxFuture, Either},
stream::FuturesUnordered,
Future, StreamExt,
};
use input::Input; use input::Input;
use output::Output; use output::Output;
use pinnacle::Pinnacle; use pinnacle::Pinnacle;
@ -186,32 +190,18 @@ pub async fn connect(
/// This function is inserted at the end of your config through the [`config`] macro. /// This function is inserted at the end of your config through the [`config`] macro.
/// You should use the macro instead of this function directly. /// You should use the macro instead of this function directly.
pub async fn listen(fut_recv: UnboundedReceiver<BoxFuture<'static, ()>>) { pub async fn listen(fut_recv: UnboundedReceiver<BoxFuture<'static, ()>>) {
let fut_recv = UnboundedReceiverStream::new(fut_recv); let mut future_set = FuturesUnordered::<BoxFuture<()>>::new();
let mut future_set = FuturesUnordered::< let mut fut_recv = UnboundedReceiverStream::new(fut_recv);
BoxFuture<(
Option<BoxFuture<()>>,
Option<UnboundedReceiverStream<BoxFuture<()>>>,
)>,
>::new();
future_set.push(Box::pin(async move { loop {
let (fut, stream) = fut_recv.into_future().await; match futures::future::select(fut_recv.next(), future_set.next()).await {
(fut, Some(stream)) Either::Left((fut, _)) => {
})); if let Some(fut) = fut {
future_set.push(fut);
while let Some((fut, stream)) = future_set.next().await { }
if let Some(fut) = fut { }
future_set.push(Box::pin(async move { Either::Right(_) => (),
fut.await;
(None, None)
}));
}
if let Some(stream) = stream {
future_set.push(Box::pin(async move {
let (fut, stream) = stream.into_future().await;
(fut, Some(stream))
}))
} }
} }
} }

1
api/rust/src/process.rs
View file

@ -154,6 +154,7 @@ impl Process {
exit(response.exit_code, exit_msg); exit(response.exit_code, exit_msg);
} }
} }
tokio::task::yield_now().await;
} }
} }
.boxed(), .boxed(),