From d56c472712df7c064387429a5af73d3bc0eb438d Mon Sep 17 00:00:00 2001
From: Alex Auvolat <alex@adnab.me>
Date: Wed, 14 Dec 2022 12:51:16 +0100
Subject: Refactor background runner and get rid of job worker

---
 src/util/background/job_worker.rs | 48 -----------------------------
 src/util/background/mod.rs        | 64 ++++++++-------------------------------
 2 files changed, 13 insertions(+), 99 deletions(-)
 delete mode 100644 src/util/background/job_worker.rs

(limited to 'src/util/background')

diff --git a/src/util/background/job_worker.rs b/src/util/background/job_worker.rs
deleted file mode 100644
index 2568ea11..00000000
--- a/src/util/background/job_worker.rs
+++ /dev/null
@@ -1,48 +0,0 @@
-//! Job worker: a generic worker that just processes incoming
-//! jobs one by one
-
-use std::sync::Arc;
-
-use async_trait::async_trait;
-use tokio::sync::{mpsc, Mutex};
-
-use crate::background::worker::*;
-use crate::background::*;
-
-pub(crate) struct JobWorker {
-	pub(crate) index: usize,
-	pub(crate) job_chan: Arc<Mutex<mpsc::UnboundedReceiver<(Job, bool)>>>,
-	pub(crate) next_job: Option<Job>,
-}
-
-#[async_trait]
-impl Worker for JobWorker {
-	fn name(&self) -> String {
-		format!("Job worker #{}", self.index)
-	}
-
-	async fn work(&mut self, _must_exit: &mut watch::Receiver<bool>) -> Result<WorkerState, Error> {
-		match self.next_job.take() {
-			None => return Ok(WorkerState::Idle),
-			Some(job) => {
-				job.await?;
-				Ok(WorkerState::Busy)
-			}
-		}
-	}
-
-	async fn wait_for_work(&mut self, must_exit: &watch::Receiver<bool>) -> WorkerState {
-		loop {
-			match self.job_chan.lock().await.recv().await {
-				Some((job, cancellable)) => {
-					if cancellable && *must_exit.borrow() {
-						continue;
-					}
-					self.next_job = Some(job);
-					return WorkerState::Busy;
-				}
-				None => return WorkerState::Done,
-			}
-		}
-	}
-}
diff --git a/src/util/background/mod.rs b/src/util/background/mod.rs
index fd9258b8..0bb4fb67 100644
--- a/src/util/background/mod.rs
+++ b/src/util/background/mod.rs
@@ -1,27 +1,23 @@
 //! Job runner for futures and async functions
 
-pub mod job_worker;
 pub mod worker;
 
 use core::future::Future;
 
 use std::collections::HashMap;
-use std::pin::Pin;
 use std::sync::Arc;
 
 use serde::{Deserialize, Serialize};
-use tokio::sync::{mpsc, watch, Mutex};
+use tokio::sync::{mpsc, watch};
 
 use crate::error::Error;
 use worker::WorkerProcessor;
 pub use worker::{Worker, WorkerState};
 
 pub(crate) type JobOutput = Result<(), Error>;
-pub(crate) type Job = Pin<Box<dyn Future<Output = JobOutput> + Send>>;
 
 /// Job runner for futures and async functions
 pub struct BackgroundRunner {
-	send_job: mpsc::UnboundedSender<(Job, bool)>,
 	send_worker: mpsc::UnboundedSender<Box<dyn Worker>>,
 	worker_info: Arc<std::sync::Mutex<HashMap<usize, WorkerInfo>>>,
 }
@@ -49,10 +45,7 @@ pub struct WorkerStatus {
 
 impl BackgroundRunner {
 	/// Create a new BackgroundRunner
-	pub fn new(
-		n_runners: usize,
-		stop_signal: watch::Receiver<bool>,
-	) -> (Arc<Self>, tokio::task::JoinHandle<()>) {
+	pub fn new(stop_signal: watch::Receiver<bool>) -> (Arc<Self>, tokio::task::JoinHandle<()>) {
 		let (send_worker, worker_out) = mpsc::unbounded_channel::<Box<dyn Worker>>();
 
 		let worker_info = Arc::new(std::sync::Mutex::new(HashMap::new()));
@@ -63,24 +56,7 @@ impl BackgroundRunner {
 			worker_processor.run().await;
 		});
 
-		let (send_job, queue_out) = mpsc::unbounded_channel();
-		let queue_out = Arc::new(Mutex::new(queue_out));
-
-		for i in 0..n_runners {
-			let queue_out = queue_out.clone();
-
-			send_worker
-				.send(Box::new(job_worker::JobWorker {
-					index: i,
-					job_chan: queue_out.clone(),
-					next_job: None,
-				}))
-				.ok()
-				.unwrap();
-		}
-
 		let bgrunner = Arc::new(Self {
-			send_job,
 			send_worker,
 			worker_info,
 		});
@@ -91,31 +67,6 @@ impl BackgroundRunner {
 		self.worker_info.lock().unwrap().clone()
 	}
 
-	/// Spawn a task to be run in background
-	pub fn spawn<T>(&self, job: T)
-	where
-		T: Future<Output = JobOutput> + Send + 'static,
-	{
-		let boxed: Job = Box::pin(job);
-		self.send_job
-			.send((boxed, false))
-			.ok()
-			.expect("Could not put job in queue");
-	}
-
-	/// Spawn a task to be run in background. It may get discarded before running if spawned while
-	/// the runner is stopping
-	pub fn spawn_cancellable<T>(&self, job: T)
-	where
-		T: Future<Output = JobOutput> + Send + 'static,
-	{
-		let boxed: Job = Box::pin(job);
-		self.send_job
-			.send((boxed, true))
-			.ok()
-			.expect("Could not put job in queue");
-	}
-
 	pub fn spawn_worker<W>(&self, worker: W)
 	where
 		W: Worker + 'static,
@@ -126,3 +77,14 @@ impl BackgroundRunner {
 			.expect("Could not put worker in queue");
 	}
 }
+
+pub fn spawn<T>(job: T)
+where
+	T: Future<Output = JobOutput> + Send + 'static,
+{
+	tokio::spawn(async move {
+		if let Err(e) = job.await {
+			error!("{}", e);
+		}
+	});
+}
-- 
cgit v1.2.3