sangfroid
A load balanced thread pool.
How does it work?
We maintain a binary heap of worker threads. Worker threads are ordered by the number of pending tasks associated with it. Whenever a new job is scheduled, we pick up the least loaded worker and dispatch the job to it. We update its pending task count and restore the heap accordingly.
When a worker is done executing a job, it sends its Uid to the done channel. A background balancer thread continuously receives on this channel until it receives a None value. When the balancer thread receives an uid, it looks up the associated worker from the worker pool heap, decrements it's load and restores the heap accordingly.
All communication to and from threads is done via channels. We use locks as sparingly as possible. More specifically, we only lock on the binary heap in the thread pool with a Mutex since it is also used by the balancer thread. We stay true to the concept:
Do not communicate by sharing memory; instead, share memory by communicating.
//! example 0: basic api usage
let thread_pool = ThreadPool::<u8, u8>::new(1);
let (job, result_src) = Job::with_result_sink(|x| x * 2, 2);
thread_pool.schedule(job).expect("job not scheduled");
assert_eq!(result_src.recv().unwrap(), 4);
//! example 1: with shared resource
let shared_hashmap = Arc::new(RwLock::new(HashMap::<u8, u8>::new()));
const PLACE_HOLDER: u8 = 5;
assert_eq!(shared_hashmap.read().unwrap().get(&1), None);
let thread_pool = ThreadPool::<((u8, u8), Arc<RwLock<HashMap<u8, u8>>>), ()>::new(2);
let job = Job::new(
|((key, value), shared_map)| {
shared_map.write().unwrap().insert(key, value);
},
((1, 2), Arc::clone(&shared_hashmap)),
);
thread_pool.schedule(job).expect("job not scheduled");
let job = Job::new(
|((key, _), shared_map)| {
shared_map.read().unwrap().get(&key);
},
((1, PLACE_HOLDER), Arc::clone(&shared_hashmap)),
);
thread_pool.schedule(job).expect("job not scheduled");
// wait some time for writers to finish
thread::sleep(Duration::from_secs(1));
assert_eq!(shared_hashmap.read().unwrap().get(&1), Some(&2));
Refer to API documentation for more details.
Why did you make this?
This crate was inspired from the load balancer described in the excellent "Concurrency is not Parallelism" talk by Rob Pike.
Well, did you learn anything?
Thread management and synchorization in Rust. I was also able to improve my ownership concepts. This was a great project for applying the knowledge I assimilated from the aforementioned talk and improve my ability to express my ideas in Rust.
Dependencies
This does not depend on any third party crates. The only dependency is the bheap crate, which I wrote for managing the workers.
Usage
This is a library crate. You may include it in your Cargo.toml as follows:
[dependencies]
sangfroid = { git = "https://github.com/arindas/sangfroid" }
Why the weird name?
sangfroid /sɒ̃ˈfrwɑː/ noun; composure or coolness shown in danger or under trying circumstances.
I found it fitting.
LICENSE
sangfroid is licensed under the MIT License. See LICENSE for the full license text.
1.2k Jan 7, 2023
22 Jan 2, 2023
34 Nov 7, 2022
55 Jan 8, 2023
289 Jan 1, 2023
17 Jan 5, 2023
7 Jan 25, 2023
6 Jan 10, 2023
11 Jan 28, 2023
112 Apr 7, 2023
67 Apr 16, 2023
5 May 15, 2023
30 Jun 16, 2023
11 Aug 7, 2023
4 Apr 9, 2024
129 Jul 20, 2024
3 Nov 22, 2022
1.5k Dec 28, 2022
100 Dec 27, 2022
939 Jan 5, 2023
73 Dec 27, 2022
90 Jan 3, 2023
22 Jan 2, 2023
9 Aug 4, 2022
2 Aug 23, 2022