Struct workerpool::Pool
source · pub struct Pool<T>where
T: Worker,{ /* private fields */ }
Expand description
Abstraction of a thread pool for basic parallelism.
Implementations§
source§impl<T: Worker> Pool<T>
impl<T: Worker> Pool<T>
sourcepub fn with_name(name: String, num_threads: usize) -> Pool<T>
pub fn with_name(name: String, num_threads: usize) -> Pool<T>
Creates a new thread pool capable of executing num_threads
number of jobs concurrently.
Each thread will have the name name
.
Panics
This function will panic if num_threads
is 0.
Examples
use std::thread;
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
let pool: Pool<ThunkWorker<()>> = Pool::with_name("worker".into(), 2);
for _ in 0..2 {
pool.execute(Thunk::of(|| {
assert_eq!(
thread::current().name(),
Some("worker")
);
}));
}
pool.join();
sourcepub fn execute(&self, inp: T::Input)
pub fn execute(&self, inp: T::Input)
Executes with the input on a worker in the pool. Non-blocking and disregards output of the worker’s execution.
Examples
Execute four jobs on a thread pool that can run two jobs concurrently:
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
let pool: Pool<ThunkWorker<()>> = Pool::new(2);
pool.execute(Thunk::of(|| println!("hello")));
pool.execute(Thunk::of(|| println!("world")));
pool.execute(Thunk::of(|| println!("foo")));
pool.execute(Thunk::of(|| println!("bar")));
pool.join();
sourcepub fn execute_to(&self, tx: Sender<T::Output>, inp: T::Input)
pub fn execute_to(&self, tx: Sender<T::Output>, inp: T::Input)
Executes with the input on a worker in the pool.
Non-blocking and sends output of the worker’s execution to the given sender.
If you want to use crossbeam_channel::Sender
instead of
std::sync::mpsc::Sender
, enable the crossbeam
feature for this library.
Examples
Execute four jobs on a thread pool that can run two jobs concurrently:
use std::sync::mpsc::channel;
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
let pool: Pool<ThunkWorker<i32>> = Pool::new(2);
let (tx, rx) = channel();
pool.execute_to(tx.clone(), Thunk::of(|| 1));
pool.execute_to(tx.clone(), Thunk::of(|| 2));
pool.execute_to(tx.clone(), Thunk::of(|| 3));
pool.execute_to(tx.clone(), Thunk::of(|| 4));
assert_eq!(10, rx.iter().take(4).sum());
sourcepub fn queued_count(&self) -> usize
pub fn queued_count(&self) -> usize
Returns the number of jobs waiting to executed in the pool.
Examples
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
use std::time::Duration;
use std::thread::sleep;
let pool: Pool<ThunkWorker<()>> = Pool::new(2);
for _ in 0..10 {
pool.execute(Thunk::of(|| {
sleep(Duration::from_secs(100));
}));
}
sleep(Duration::from_secs(1)); // wait for threads to start
assert_eq!(8, pool.queued_count());
sourcepub fn active_count(&self) -> usize
pub fn active_count(&self) -> usize
Returns the number of currently active threads.
Examples
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
use std::time::Duration;
use std::thread::sleep;
let pool = Pool::<ThunkWorker<()>>::new(4);
for _ in 0..10 {
pool.execute(Thunk::of(|| {
sleep(Duration::from_secs(100));
}))
}
sleep(Duration::from_secs(1)); // wait for threads to start
assert_eq!(4, pool.active_count());
sourcepub fn max_count(&self) -> usize
pub fn max_count(&self) -> usize
Returns the maximum number of threads the pool will execute concurrently.
Examples
use workerpool::Pool;
let mut pool: Pool<MyWorker> = Pool::new(4);
assert_eq!(4, pool.max_count());
pool.set_num_threads(8);
assert_eq!(8, pool.max_count());
sourcepub fn panic_count(&self) -> usize
pub fn panic_count(&self) -> usize
Returns the number of panicked threads over the lifetime of the pool.
Examples
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
let pool: Pool<ThunkWorker<()>> = Pool::new(4);
for n in 0..10 {
pool.execute(Thunk::of(move || {
// simulate a panic
if n % 2 == 0 {
panic!()
}
}));
}
pool.join();
assert_eq!(5, pool.panic_count());
sourcepub fn set_num_threads(&mut self, num_threads: usize)
pub fn set_num_threads(&mut self, num_threads: usize)
Sets the number of worker-threads to use as num_threads
.
Can be used to change the workerpool size during runtime.
Will not abort already running or waiting threads.
Panics
This function will panic if num_threads
is 0.
Examples
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
use std::time::Duration;
use std::thread::sleep;
let mut pool: Pool<ThunkWorker<()>> = Pool::new(4);
for _ in 0..10 {
pool.execute(Thunk::of(|| {
sleep(Duration::from_secs(100));
}));
}
sleep(Duration::from_secs(1)); // wait for threads to start
assert_eq!(4, pool.active_count());
assert_eq!(6, pool.queued_count());
// Increase thread capacity of the pool
pool.set_num_threads(8);
sleep(Duration::from_secs(1)); // wait for new threads to start
assert_eq!(8, pool.active_count());
assert_eq!(2, pool.queued_count());
// Decrease thread capacity of the pool
// No active threads are killed
pool.set_num_threads(4);
assert_eq!(8, pool.active_count());
assert_eq!(2, pool.queued_count());
sourcepub fn join(&self)
pub fn join(&self)
Block the current thread until all jobs in the pool have been executed.
Calling join
on an empty pool will cause an immediate return.
join
may be called from multiple threads concurrently.
A join
is an atomic point in time. All threads joining before the join
event will exit together even if the pool is processing new jobs by the
time they get scheduled.
Calling join
from a thread within the pool will cause a deadlock. This
behavior is considered safe.
Examples
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};
let pool: Pool<ThunkWorker<()>> = Pool::new(8);
let test_count = Arc::new(AtomicUsize::new(0));
for _ in 0..42 {
let test_count = test_count.clone();
pool.execute(Thunk::of(move || {
test_count.fetch_add(1, Ordering::Relaxed);
}))
}
pool.join();
assert_eq!(42, test_count.load(Ordering::Relaxed));
Trait Implementations§
source§impl<T: Worker> Clone for Pool<T>
impl<T: Worker> Clone for Pool<T>
source§fn clone(&self) -> Pool<T>
fn clone(&self) -> Pool<T>
Cloning a pool will create a new handle to the pool. The behavior is similar to Arc.
We could for example submit jobs from multiple threads concurrently.
use workerpool::Pool;
use workerpool::thunk::{Thunk, ThunkWorker};
use std::thread;
use std::sync::mpsc::channel;
let pool: Pool<ThunkWorker<()>> = Pool::with_name("clone example".into(), 2);
let results = (0..2)
.map(|i| {
let pool = pool.clone();
thread::spawn(move || {
let (tx, rx) = channel();
for i in 1..12 {
let tx = tx.clone();
pool.execute(Thunk::of(move || {
tx.send(i).expect("channel will be waiting");
}));
}
drop(tx);
if i == 0 {
rx.iter().fold(0, |accumulator, element| accumulator + element)
} else {
rx.iter().fold(1, |accumulator, element| accumulator * element)
}
})
})
.map(|join_handle| join_handle.join().expect("collect results from threads"))
.collect::<Vec<usize>>();
assert_eq!(vec![66, 39916800], results);
1.0.0 · source§fn clone_from(&mut self, source: &Self)
fn clone_from(&mut self, source: &Self)
source
. Read moresource§impl<T: Worker> Default for Pool<T>
impl<T: Worker> Default for Pool<T>
Create a thread pool with one thread per CPU. On machines with hyperthreading, this will create one thread per hyperthread.