Struct ThreadPool

Source

pub struct ThreadPool { /* private fields */ }

Expand description

Threadpool abstraction to keep some state

Implementations§

Source §

impl ThreadPool

Source

pub fn send_job(&self, job: impl FnOnce() + Send + UnwindSafe + 'static)

The send_job function takes in a function or closure without any arguments and sends it to the threadpool to be executed. Jobs will be taken from the job queue in order of them being sent, but that in no way guarantees they will be executed in order.

jobs must implement Send in order to be safely sent across threads and UnwindSafe to allow catching panics when executing the jobs. Both of these traits are auto implemented.

§Examples

Sending a function or closure to the threadpool

use easy_threadpool::ThreadPoolBuilder;

fn job() {
    println!("Hello world from a function!");
}

let builder = ThreadPoolBuilder::with_max_threads()?;
let pool = builder.build()?;

pool.send_job(job);

pool.send_job(|| println!("Hello world from a closure!"));

Source

pub fn wait_until_finished(&self) -> Result<(), JobHasPanicedError>

This function will wait until all jobs have finished sending. Additionally it will return early if any job panics.

Be careful though, returning early DOES NOT mean that the sent jobs are cancelled. They will remain running. Cancelling jobs that are queued is not a feature provided by this crate as of now.

§Errors

This function will error if any job sent to the threadpool has errored. This includes any errors since either the threadpool was created or since the state was reset.

§Examples

use easy_threadpool::ThreadPoolBuilder;

let builder = ThreadPoolBuilder::with_max_threads()?;
let pool = builder.build()?;

for _ in 0..10 {
    pool.send_job(|| println!("Hello world"));
}

assert!(pool.wait_until_finished().is_ok());
assert!(pool.is_finished());

pool.send_job(|| panic!("Test panic"));

assert!(pool.wait_until_finished().is_err());
assert!(pool.has_paniced());

Source

pub fn wait_until_job_done(&self) -> Result<(), JobHasPanicedError>

This function will wait until one job finished after calling the function. Additionally, if the threadpool is finished this function will also return. Additionally it will return early if any job panics.

Be careful though, returning early DOES NOT mean that the sent jobs are cancelled. They will remain running. Cancelling jobs that are queued is not a feature provided by this crate as of now.

§Errors

This function will error if any job sent to the threadpool has errored. This includes any errors since either the threadpool was created or since the state was reset.

§Examples

use easy_threadpool::ThreadPoolBuilder;

let builder = ThreadPoolBuilder::with_max_threads()?;
let pool = builder.build()?;

assert!(pool.wait_until_job_done().is_ok());
assert!(pool.is_finished());

pool.send_job(|| panic!("Test panic"));

assert!(pool.wait_until_job_done().is_err());
assert!(pool.has_paniced());

Source

pub fn wait_until_finished_unchecked(&self)

This function will wait until all jobs have finished sending. It will continue waiting if a job panics in the thread pool.

I highly doubt this has much of a performance improvement, but it’s very useful if you know that for whatever reason your jobs might panic and that would be fine.

§Examples

use easy_threadpool::ThreadPoolBuilder;

let builder = ThreadPoolBuilder::with_max_threads()?;
let pool = builder.build()?;

for _ in 0..10 {
    pool.send_job(|| println!("Hello world"));
}

pool.wait_until_finished_unchecked();
assert!(pool.is_finished());

pool.send_job(|| panic!("Test panic"));

pool.wait_until_finished_unchecked();
assert!(pool.has_paniced());

Source

pub fn wait_until_job_done_unchecked(&self)

This function will wait until one job finished after calling the function. Additionally, if the threadpool is finished this function will also return.

Be careful though, returning early DOES NOT mean that the sent jobs are cancelled. They will remain running. Cancelling jobs that are queued is not a feature provided by this crate as of now.

§Examples

use easy_threadpool::ThreadPoolBuilder;

let builder = ThreadPoolBuilder::with_max_threads()?;
let pool = builder.build()?;

assert!(pool.wait_until_job_done().is_ok());
assert!(pool.is_finished());

pool.send_job(|| panic!("Test panic"));

assert!(pool.wait_until_job_done().is_err());
assert!(pool.has_paniced());

Source

pub fn reset_state(&mut self)

This function will reset the state of this instance of the threadpool.

When resetting the state you lose all information about previously sent jobs. If a job you previously sent panics, you will not be notified, nor can you wait until your previously sent jobs are done running. HOWEVER they will still be running. Be very careful to not see this as a “stop” button.

§Examples

use easy_threadpool::ThreadPoolBuilder;

let builder = ThreadPoolBuilder::with_max_threads()?;
let mut pool = builder.build()?;

pool.send_job(|| panic!("Test panic"));

assert!(pool.wait_until_finished().is_err());
assert!(pool.has_paniced());

pool.reset_state();

assert!(pool.wait_until_finished().is_ok());
assert!(!pool.has_paniced());

Source

pub fn clone_with_new_state(&self) -> Self

This function will clone the threadpool and then reset its state. This makes it so you can have 2 different states operate on the same threads, effectively sharing the threads.

Note however that there is no mechanism to give different instances equal CPU time, jobs are executed on a first come first server basis.

§Examples

use easy_threadpool::ThreadPoolBuilder;

let builder = ThreadPoolBuilder::with_max_threads()?;
let pool = builder.build()?;

let pool_clone = pool.clone_with_new_state();

pool.send_job(|| panic!("Test panic"));

assert!(pool.wait_until_finished().is_err());
assert!(pool.has_paniced());

assert!(pool_clone.wait_until_finished().is_ok());
assert!(!pool_clone.has_paniced());

Source

pub fn jobs_running(&self) -> usize

Returns the amount of jobs currently being ran by this instance of the thread pool. If muliple different instances of this threadpool (see [clone_with_new_state]) this number might be lower than the max amount of threads, even if there are still jobs queued

§Examples

use easy_threadpool::ThreadPoolBuilder;
use std::{
    num::NonZeroUsize,
    sync::{Arc, Barrier},
};
let threads = 16;
let tasks = threads * 10;

let num = NonZeroUsize::try_from(threads)?;
let pool = ThreadPoolBuilder::with_thread_amount(num).build()?;

let b0 = Arc::new(Barrier::new(threads + 1));
let b1 = Arc::new(Barrier::new(threads + 1));

for i in 0..tasks {
    let b0_copy = b0.clone();
    let b1_copy = b1.clone();

    pool.send_job(move || {
        if i < threads {
            b0_copy.wait();
            b1_copy.wait();
        }
    });
}

b0.wait();
assert_eq!(pool.jobs_running(), threads);

Source

pub fn jobs_queued(&self) -> usize

Returns the amount of jobs currently queued by this threadpool instance. There might be more jobs queued that we don’t know about if there are other instances of this threadpool (see [clone_with_new_state]).

§Examples

use easy_threadpool::ThreadPoolBuilder;
use std::{
    num::NonZeroUsize,
    sync::{Arc, Barrier},
};
let threads = 16;
let tasks = 100;

let num = NonZeroUsize::try_from(threads)?;
let pool = ThreadPoolBuilder::with_thread_amount(num).build()?;

let b0 = Arc::new(Barrier::new(threads + 1));
let b1 = Arc::new(Barrier::new(threads + 1));

for i in 0..tasks {
    let b0_copy = b0.clone();
    let b1_copy = b1.clone();

    pool.send_job(move || {
        if i < threads {
            b0_copy.wait();
            b1_copy.wait();
        }
    });
}

b0.wait();
assert_eq!(pool.jobs_queued(), tasks - threads);

Source

pub fn jobs_paniced(&self) -> usize

Returns the amount of jobs that were sent by this instance of the threadpool and that paniced.

§Examples

use easy_threadpool::ThreadPoolBuilder;

let pool = ThreadPoolBuilder::with_max_threads()?.build()?;

for i in 0..10 {
    pool.send_job(|| panic!("Test panic"));
}

pool.wait_until_finished_unchecked();

assert_eq!(pool.jobs_paniced(), 10);

Source

pub fn has_paniced(&self) -> bool

Returns whether a thread has had any jobs panic at all

§Examples

use easy_threadpool::ThreadPoolBuilder;

let pool = ThreadPoolBuilder::with_max_threads()?.build()?;

pool.send_job(|| panic!("Test panic"));

pool.wait_until_finished_unchecked();

assert!(pool.has_paniced());

Source

pub fn is_finished(&self) -> bool

Returns whether a threadpool instance has no jobs running and no jobs queued, in other words if it’s finished.

§Examples

use easy_threadpool::ThreadPoolBuilder;
use std::{
    num::NonZeroUsize,
    sync::{Arc, Barrier},
};
let pool = ThreadPoolBuilder::with_max_threads()?.build()?;

let b = Arc::new(Barrier::new(2));

assert!(pool.is_finished());

let b_clone = b.clone();
pool.send_job(move || { b_clone.wait(); });

assert!(!pool.is_finished());

Source

pub const fn threads(&self) -> NonZeroUsize

This function returns the amount of threads used to create the threadpool

§Examples

use easy_threadpool::ThreadPoolBuilder;
use std::num::NonZeroUsize;

let threads = 10;

let num = NonZeroUsize::try_from(threads)?;
let pool = ThreadPoolBuilder::with_thread_amount(num).build()?;

assert_eq!(pool.threads().get(), threads);