RustThreadPool

Documentation

Concept

RustThreadPool is a facility for processing a stream of messages with a function object concurrently on a specified number of threads, using a thread-safe blocking queue. The function object uses messages posted to RustBlockingQueue.

Design

There is one struct, ThreadPool<M>, with three methods:

#[derive(Debug)]
pub struct ThreadPool<M> {
    sbq: Arc<BlockingQueue<M>>,
    thrd: Vec<Option<JoinHandle<()>>>
}

pub fn new<F>(nt: u8, f: F) -> ThreadPool<M>
where F: FnOnce(&BlockingQueue<M>) -> () + Send + 'static + Copy

pub fn wait(&mut self)

pub fn post_message(&mut self, _msg: M)
where M: Debug + Clone

Sharing between threads is only possible, due to rules of the Rust language, if shared items are Mutexes, Condvars, or aggregates of those. BlockingQueue<T> is shareable because its two fields are a Mutex<VecDeque<T>> and a Condvar.

Public API

ThreadPool<M> is parameterized over message type M: Send + 'static.

Function	Signature	Description
new	fn new<F>(nt: u8, f: F) -> ThreadPool<M>	Spawn nt threads, each running closure f against the shared queue
post_message	fn post_message(&mut self, msg: M)	Enqueue a work item for any available worker
get_message	fn get_message(&mut self) -> M	Stub - returns M::default()
wait	fn wait(&mut self)	Block until all worker threads finish

The closure bound F: FnOnce(&BlockingQueue<M>) + Send + 'static + Copy means workers share a Copyable function but each receive their own reference to the shared queue. Closures that capture Arc or other non-Copy values will not satisfy this bound; use non-capturing closures or bare function pointers instead.

Shutdown Pattern

The library does not provide a built-in shutdown mechanism. The standard approach is to post one poison-pill message per worker thread, where the worker breaks its loop on receiving it:

for _ in 0..num_threads {
    pool.post_message(WorkItem::default());
}
pool.wait();

Interface Demos

interface_demos/ contains a runnable binary crate generated by interface_demo_agent.py:

File	Description
src/rust_thread_pool_demos.rs	Four demos covering all public API functions
rust_thread_pool_interface.json	Extracted interface description used by the agent
generation_summary.json	Agent run metadata

Build and run the demos:

cd interface_demos
cargo run

Demos:

1. demo_basic_pool - new, post_message, wait with 4 threads and 10 work items
2. demo_get_message - get_message stub returning M::default()
3. demo_string_pool - ThreadPool<StringMsg> showing generic type parameter
4. demo_stress_test - 8 threads, 200 messages

Script Invocation

interface_demo_agent.py was run from the project root:

python interface_demo_agent.py c:\github\JimFawcett\RustThreadPool

It extracted the public interface from src/lib.rs, wrote rust_thread_pool_interface.json and rust_thread_pool_demo.rs into interface_demos/, and recorded generation_summary.json.

Demo Output

*****************************************************************
  rust_thread_pool - Public API Demonstration
*****************************************************************

=================================================================
Demo 1: new, post_message, wait
=================================================================

[Demo 1] posting 10 work items
[Demo 1] posting 4 shutdown signals
[Demo 1] waiting ...
  [Worker ThreadId(3)] id=3 payload="task-3"
  [Worker ThreadId(4)] id=2 payload="task-2"
  [Worker ThreadId(2)] id=1 payload="task-1"
  [Worker ThreadId(5)] id=4 payload="task-4"
  [Worker ThreadId(4)] id=6 payload="task-6"
  [Worker ThreadId(5)] id=8 payload="task-8"
  [Worker ThreadId(3)] id=5 payload="task-5"
  [Worker ThreadId(3)] shutdown
  [Worker ThreadId(2)] id=7 payload="task-7"
  [Worker ThreadId(2)] shutdown
  [Worker ThreadId(4)] id=9 payload="task-9"
  [Worker ThreadId(5)] id=10 payload="task-10"
  [Worker ThreadId(5)] shutdown
  [Worker ThreadId(4)] shutdown
[Demo 1] done

=================================================================
Demo 2: get_message stub
=================================================================

[Demo 2] get_message returned: WorkItem { id: 0, payload: "" }
[Demo 2] done

=================================================================
Demo 3: ThreadPool<StringMsg>
=================================================================

[Demo 3] posting 6 messages
  [Worker ThreadId(7)] "Hello from Demo 3"
[Demo 3] posting 3 STOP signals
  [Worker ThreadId(8)] "Rust thread pools are generic"
  [Worker ThreadId(9)] "BlockingQueue handles synchronization"
  [Worker ThreadId(9)] "Workers share the queue automatically"
  [Worker ThreadId(8)] "No manual locking required by the caller"
  [Worker ThreadId(7)] "Each message handled by one worker"
  [Worker ThreadId(8)] STOP
  [Worker ThreadId(7)] STOP
  [Worker ThreadId(9)] STOP
[Demo 3] done

=================================================================
Demo 4: stress test - 8 threads, 200 messages
=================================================================

[Demo 4] posting 200 items
[Demo 4] posting 8 shutdown signals
[Demo 4] waiting ...
[Demo 4] done

*****************************************************************
  All demonstrations completed.
*****************************************************************

Operation

Operation is illustrated by src/rust_thread_pool_demos.rs in interface_demos/ and by test1.rs in examples/.

Build

Download and, in a command prompt, use one of:

cargo build
cargo test
cargo run --example test1

Status

• new, post_message, and wait are complete and functional
• get_message is a stub pending an output queue implementation
• ThreadPool has been used in several projects - see RustCommExperiments