Make background runner terminate correctly

This commit is contained in:
Alex 2022-05-05 10:56:44 +02:00
parent 176715c5b2
commit 7b474855e3
Signed by: lx
GPG key ID: 0E496D15096376BE

View file

@ -6,7 +6,9 @@ use std::time::Duration;
use futures::future::*; use futures::future::*;
use futures::select; use futures::select;
use tokio::sync::{mpsc, watch, Mutex}; use futures::stream::FuturesUnordered;
use futures::StreamExt;
use tokio::sync::{mpsc, mpsc::error::TryRecvError, watch, Mutex};
use crate::error::Error; use crate::error::Error;
@ -30,26 +32,31 @@ impl BackgroundRunner {
let stop_signal_2 = stop_signal.clone(); let stop_signal_2 = stop_signal.clone();
let await_all_done = tokio::spawn(async move { let await_all_done = tokio::spawn(async move {
let mut workers = FuturesUnordered::new();
let mut shutdown_timer = 0;
loop { loop {
let wkr = { let closed = match worker_out.try_recv() {
select! { Ok(wkr) => {
item = worker_out.recv().fuse() => { workers.push(wkr);
match item { false
Some(x) => x, }
None => break, Err(TryRecvError::Empty) => false,
} Err(TryRecvError::Disconnected) => true,
};
select! {
res = workers.next() => {
if let Some(Err(e)) = res {
error!("Worker exited with error: {}", e);
} }
_ = tokio::time::sleep(Duration::from_secs(5)).fuse() => { }
if *stop_signal_2.borrow() { _ = tokio::time::sleep(Duration::from_secs(1)).fuse() => {
if closed || *stop_signal_2.borrow() {
shutdown_timer += 1;
if shutdown_timer >= 10 {
break; break;
} else {
continue;
} }
} }
} }
};
if let Err(e) = wkr.await {
error!("Error while awaiting for worker: {}", e);
} }
} }
}); });