aboutsummaryrefslogtreecommitdiff
path: root/src/util/background/mod.rs
blob: 619f506858ccd5e3e283fedc4b016afcef421399 (plain) (blame)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
//! 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 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>>>,
}

#[derive(Clone, Serialize, Deserialize, Debug)]
pub struct WorkerInfo {
	pub name: String,
	pub info: Option<String>,
	pub state: WorkerState,
	pub errors: usize,
	pub consecutive_errors: usize,
	pub last_error: Option<(String, u64)>,
}

impl BackgroundRunner {
	/// Create a new BackgroundRunner
	pub fn new(
		n_runners: usize,
		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()));
		let mut worker_processor =
			WorkerProcessor::new(worker_out, stop_signal, worker_info.clone());

		let await_all_done = tokio::spawn(async move {
			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,
		});
		(bgrunner, await_all_done)
	}

	pub fn get_worker_info(&self) -> HashMap<usize, WorkerInfo> {
		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,
	{
		self.send_worker
			.send(Box::new(worker))
			.ok()
			.expect("Could not put worker in queue");
	}
}