taba_cli/tui/worker/
mod.rs

1//! The workers (doc 34, section 4): `taba-client` is blocking, so the calls `update` asks for run on a few threads and
2//! come back as a `Msg::Reply` marked with the request and the generation they were asked in.
3
4use std::sync::mpsc::{Receiver, Sender};
5use std::sync::{Arc, Mutex};
6use std::thread;
7
8use taba_client::{Call, Client, Error, Reply};
9
10use crate::tui::api::ApiCall;
11use crate::tui::msg::Msg;
12
13/// What a worker calls. The real one is the `Client`; the tests give a scripted one.
14pub trait Backend: Send + Sync {
15    fn call(&self, call: &Call) -> Result<Reply, Error>;
16}
17
18impl Backend for Client {
19    fn call(&self, call: &Call) -> Result<Reply, Error> {
20        Client::call(self, call)
21    }
22}
23
24/// A backend that can be replaced while the workers run: choosing another profile points every later call at
25/// the new server.
26pub struct SwitchBackend {
27    current: std::sync::RwLock<Arc<dyn Backend>>,
28}
29
30impl SwitchBackend {
31    pub fn new(first: Arc<dyn Backend>) -> SwitchBackend {
32        SwitchBackend {
33            current: std::sync::RwLock::new(first),
34        }
35    }
36
37    pub fn switch(&self, next: Arc<dyn Backend>) {
38        *self.current.write().unwrap_or_else(|e| e.into_inner()) = next;
39    }
40}
41
42impl Backend for SwitchBackend {
43    fn call(&self, call: &Call) -> Result<Reply, Error> {
44        let backend = Arc::clone(&self.current.read().unwrap_or_else(|e| e.into_inner()));
45        backend.call(call)
46    }
47}
48
49/// How many calls can be in flight at once.
50pub const THREADS: usize = 4;
51
52struct Job {
53    id: u64,
54    generation: u64,
55    call: ApiCall,
56}
57
58pub struct Workers {
59    jobs: Sender<Job>,
60}
61
62impl Workers {
63    /// Starts the threads; they end when the `Workers` is dropped.
64    pub fn start(backend: Arc<dyn Backend>, results: Sender<Msg>, threads: usize) -> Workers {
65        let (jobs, queue) = std::sync::mpsc::channel::<Job>();
66        let queue: Arc<Mutex<Receiver<Job>>> = Arc::new(Mutex::new(queue));
67        for _ in 0..threads.max(1) {
68            let queue = Arc::clone(&queue);
69            let backend = Arc::clone(&backend);
70            let results = results.clone();
71            thread::spawn(move || loop {
72                // The lock is held only to take the next job, never while the call runs.
73                let job = match queue.lock() {
74                    Ok(guard) => guard.recv(),
75                    Err(_) => return,
76                };
77                let Ok(job) = job else { return };
78                let result = backend.call(&job.call.to_call());
79                let reply = Msg::Reply {
80                    id: job.id,
81                    generation: job.generation,
82                    call: job.call,
83                    result,
84                };
85                if results.send(reply).is_err() {
86                    return;
87                }
88            });
89        }
90        Workers { jobs }
91    }
92
93    pub fn submit(&self, id: u64, generation: u64, call: ApiCall) {
94        let _ = self.jobs.send(Job {
95            id,
96            generation,
97            call,
98        });
99    }
100}
101
102#[cfg(test)]
103mod tests;
taba_cli/tui/worker/mod.rs

taba_cli/tui/worker/
mod.rs
