NoraCodes · May 5, 2025 11:02 · Aug 30, 2017 · Aug 30, 2017 · Aug 30, 2017
diff --git a/work_queue.rs b/work_queue.rs
@@ -373,10 +373,10 @@ fn main() {
 
 
     // Get completed work from the channel while there's work to be done.
-    while jobs_remaining > 0 {
+    while jobs_total > 0 {
         match results_rx.recv() {
             // If the control thread successfully receives, a job was completed.
-            Ok(_) => { jobs_remaining -= 1 },
+            Ok(_) => { jobs_total -= 1 },
             // If the control thread is the one left standing, that's pretty
             // problematic.
             Err(_) => {panic!("All workers died unexpectedly.");}

diff --git a/work_queue.rs b/work_queue.rs
@@ -0,0 +1,405 @@
+// Here is an extremely simple version of work scheduling for multiple
+// processors.
+//
+// The Problem: 
+//  We have a lot of numbers that need to be math'ed. Doing this on one 
+//  CPU core is slow. We have 4 CPU cores. We would thus like to use those
+//  cores to do math, because it will be a little less slow (ideally
+//  4 times faster actually).
+//
+// The Solution:
+//  Create a queue which can be safely shared between threads from which those
+//  threads can pull work each time they finish their current work.
+//
+// We assume that we have 4 processor cores which will do work for us.
+// You could use the num_cpus crate to find this for a particular machine.
+const MAX_WORKER: usize = 4;
+
+// Here is the work we want done: a simple non-recursive Fibonacci calculation.
+fn fib(n: u64) -> u64 {
+    // Special case: the 0th Fib. number is 0.
+    if n == 0 {
+        return 0;
+    }
+    // Special case: the 1st Fib. number is 1.
+    if n == 1 {
+        return 1;
+    }
+
+    let mut iteration = 0;
+    let mut sum = 0;
+    let mut last = fib(0);
+    let mut current = fib(1);
+
+    // Loop through all the Fib. numbers until we get to
+    // the nth one.
+    while iteration < n-1 {
+        sum = last + current;
+        last = current;
+        current = sum;
+        iteration += 1;
+    }
+
+    return sum;
+}
+
+// We need a way to keep track of what work needs to be done.
+// This is a multi-source, multi-consumer queue which we call a 
+// WorkQueue.
+
+// To create this type, we wrap a mutex (std::sync::mutex) around a 
+// queue (technically a double-ended queue, std::collections::VecDeque).
+//
+// Mutex stands for MUTually EXclusive. It essentially ensures that only
+// one thread has access to a given resource at one time.
+use std::sync::Mutex;
+
+// A VecDeque is a double-ended queue, but we will only be using it in forward
+// mode; that is, we will push onto the back and pull from the front.
+use std::collections::VecDeque;
+
+// Finally we wrap the whole thing in Arc (Atomic Reference Counting) so that
+// we can safely share it with other threads. Arc (std::sync::arc) is a lot
+// like Rc (std::rc::Rc), in that it allows multiple references to some memory
+// which is freed when no references remain, except that it is atomic, making
+// it comparitively slow but able to be shared across the thread boundary.
+use std::sync::Arc;
+
+// All three of these types are wrapped around a generic type T.
+// T is required to be Send (a marker trait automatically implemented when
+// it is safe to do so) because it denotes types that are safe to move between
+// threads, which is the whole point of the WorkQueue.
+// For this implementation, T is required to be Copy as well, for simplicity.
+
+/// A generic work queue for work elements which can be trivially copied.
+/// Any producer of work can add elements and any worker can consume them.
+/// WorkQueue derives Clone so that it can be distributed among threads.
+#[derive(Clone)]
+struct WorkQueue<T: Send + Copy> {
+    inner: Arc<Mutex<VecDeque<T>>>,
+}
+
+impl<T: Send + Copy> WorkQueue<T> {
+    // Creating one of these by hand would be kind of a pain, so let's provide a
+    // convenience function.
+
+    /// Creates a new WorkQueue, ready to be used.
+    fn new() -> Self { 
+        Self { inner: Arc::new(Mutex::new(VecDeque::new())) } 
+    }
+
+    // This is the function workers will use to acquire work from the queue.
+    // They will call it in a loop, checking to see if there is any work available.
+
+    /// Blocks the current thread until work is available, then
+    /// gets the data required to perform that work.
+    ///
+    /// # Errors
+    /// Returns None if there is no more work in the queue.
+    ///
+    /// # Panics
+    /// Panics if the underlying mutex became poisoned. This is exceedingly
+    /// unlikely.
+    fn get_work(&self) -> Option<T> {
+        // Try to get a lock on the Mutex. If this fails, there is a 
+        // problem with the mutex - it's poisoned, meaning that a thread that
+        // held the mutex lock panicked before releasing it. There is no way
+        // to guarantee that all its invariants are upheld, so we need to not
+        // use it in that case.
+        let maybe_queue = self.inner.lock();
+        // A lot is going on here. self.inner is an Arc of Mutex. Arc can deref
+        // into its internal type, so we can call the methods of that inner
+        // type (Mutex) without dereferencing, so this is like 
+        //      *(self.inner).lock() 
+        // but doesn't look awful. Mutex::lock() returns a 
+        // Result<MutexGuard<VecDeque<T>>>. 
+
+        // Unwrapping with if let, we get a MutexGuard, which is an RAII guard
+        // that unlocks the Mutex when it goes out of scope.
+        if let Ok(mut queue) = maybe_queue {
+            // queue is a MutexGuard<VecDeque>, so this is like 
+            //      (*queue).pop_front()
+            // Returns Some(item) or None if there are no more items.
+            queue.pop_front()
+
+            // The function has returned, so queue goes out of scope and the
+            // mutex unlocks.
+        } else {
+            // There's a problem with the mutex.
+            panic!("WorkQueue::get_work() tried to lock a poisoned mutex");
+        }
+    }
+
+    // Both the controller (main thread) and possibly workers can use this
+    // function to add work to the queue.
+
+    /// Blocks the current thread until work can be added, then
+    /// adds that work to the end of the queue.
+    /// Returns the amount of work now in the queue.
+    ///
+    /// # Panics
+    /// Panics if the underlying mutex became poisoned. This is exceedingly
+    /// unlikely.
+    fn add_work(&self, work: T) -> usize {
+        // As above, try to get a lock on the mutex. 
+        if let Ok(mut queue) = self.inner.lock() {
+            // As above, we can use the MutexGuard<VecDeque<T>> to access
+            // the internal VecDeque.
+            queue.push_back(work);
+
+            // Now return the length of the queue.
+            queue.len()
+        } else {
+            panic!("WorkQueue::add_work() tried to lock a poisoned mutex");
+        }
+    }
+}
+
+// Now we have a way of getting work from one thread to many threads.
+// We need one more thing: a way to tell the workers when they're done.
+// We could just say that they should give up when the work queue is empty.
+// This will work in situations like this one, but not in situations where
+// new work can be created after all the initial work is complete.
+// In our case, it gives the capability to start workers before adding anything
+// to the work queue, so they can work while the controller is adding work.
+//
+// For this we'll use another Mutex-based thing, but this time it will just
+// wrap a boolean value. We'll call this a syncronization flag, or SyncFlag.
+//
+// However, we only want the controller thread to be able to tell the workers
+// that they're done, so we'll go the route of std::sync::mpsc and create a
+// Transmitter and Reciever version of the struct.
+//
+// The transmitter is where messages are sent; it has the ability to set the
+// underlying Boolean value. The receiver only has the ability to read that
+// value.
+
+/// SyncFlagTx is the transmitting (mutable) half of a Single Producer,
+/// Multiple Consumer Boolean (e.g. the opposite of std::sync::mpsc).
+/// A single controller can use this to send info to any number of worker
+/// threads, for instance.
+///
+/// SyncFlagTx is not Clone because it should only exist in one place.
+struct SyncFlagTx {
+    inner: Arc<Mutex<bool>>,
+}
+
+impl SyncFlagTx {
+    // This function will be used by the controller thread to tell the worker
+    // threads about the end of computation.
+
+    /// Sets the interior value of the SyncFlagTx which will be read by any
+    /// SyncFlagRx that exist for this SyncFlag.
+    ///
+    /// # Errors
+    /// If the underlying mutex is poisoned this may return an error.
+    fn set(&mut self, state: bool) -> Result<(), ()> {
+        if let Ok(mut v) = self.inner.lock() {
+            // The * (deref operator) means assigning to what's inside the
+            // MutexGuard, not the guard itself (which would be silly)
+            *v = state;
+            Ok(())
+        } else {
+            Err(())
+        }
+    }
+}
+
+/// SyncFlagRx is the receiving (immutable) half of a Single Producer,
+/// Multiple Consumer Boolean (e.g. the opposite of std::sync::mpsc).
+/// An number of worker threads can use this to get info from a single
+/// controller, for instance.
+///
+/// SyncFlagRx is Clone so it can be shared across threads.
+#[derive(Clone)]
+struct SyncFlagRx {
+    inner: Arc<Mutex<bool>>,
+}
+
+impl SyncFlagRx {
+    // This function will be used by the worker threads to check if they should
+    // stop looking for work to do.
+
+    /// Gets the interior state of the SyncFlagRx to whatever the corresponding
+    /// SyncFlagTx last set it to.
+    ///
+    /// # Errors
+    /// If the underlying mutex is poisoned this might return an error.
+    fn get(&self) -> Result<bool, ()> {
+        if let Ok(v) = self.inner.lock() {
+            // Deref the MutexGuard to get at the bool inside
+            Ok(*v)
+        } else {
+            Err(())
+        }
+    }
+}
+
+/// Create a new SyncFlagTx and SyncFlagRx that can be used to share a bool 
+/// across a number of threads.
+fn new_syncflag(initial_state: bool) -> (SyncFlagTx, SyncFlagRx) {
+    let state = Arc::new(Mutex::new(initial_state));
+    let tx = SyncFlagTx { inner: state.clone() };
+    let rx = SyncFlagRx { inner: state.clone() };
+
+    return (tx, rx);
+}
+
+fn main() {
+    // Create a new work queue to keep track of what work needs to be done.
+    // Note that the queue is internally mutable (or, rather, the Mutex is),
+    // but this binding doesn't need to be mutable. This isn't unsound because
+    // the Mutex ensures at runtime that no two references can be used;
+    // therefore no mutation can occur at the same time as aliasing.
+    let queue = WorkQueue::new();
+
+    // Create a MPSC (Multiple Producer, Single Consumer) channel. Every worker
+    // is a producer, the main thread is a consumer; the producers put their
+    // work into the channel when it's done.
+    use std::sync::mpsc::channel;
+    let (results_tx, results_rx) = channel();
+
+    // Create a SyncFlag to share whether or not there are more jobs to be done.
+    let (mut more_jobs_tx, more_jobs_rx) = new_syncflag(true);
+
+    // std::thread allows us to spawn threads to do work in.
+    use std::thread;
+    // This Vec will hold thread join handles to allow us to not exit while work
+    // is still being done. These handles provide a .join() method which blocks
+    // the current thread until the thread referred to by the handle exits.
+    let mut threads = Vec::new();
+
+    println!("Spawning {} workers.", MAX_WORKER);
+
+    for thread_num in 0..MAX_WORKER {
+        // Get a reference to the queue for the thread to use
+        // .clone() here doesn't clone the actual queue data, but rather the
+        // internal Arc produces a new reference for use in the new queue 
+        // instance.
+        let thread_queue = queue.clone();
+
+        // Similarly, create a new transmitter for the thread to use
+        let thread_results_tx = results_tx.clone();
+
+        // ... and a SyncFlagRx for the thread.
+        let thread_more_jobs_rx = more_jobs_rx.clone();
+
+        // thread::spawn takes a closure (an anonymous function that "closes"
+        // over its environment). The move keyword means it takes ownership of
+        // those variables, meaning they can't be used again in the main thread.
+        let handle = thread::spawn(move || {
+            // A varaible to keep track of how much work was done.
+            let mut work_done = 0;
+
+            // Loop while there's expected to be work, looking for work.
+            while thread_more_jobs_rx.get().unwrap() {
+
+                // If work is available, do that work.
+                if let Some(work) = thread_queue.get_work() {
+                    // Do some work.
+                    let result = fib(work);
+
+                    // Record that some work was done.
+                    work_done += 1;
+
+                    // Send the work and the result of that work.
+                    //
+                    // Sending could fail. If so, there's no use in
+                    // doing any more work, so abort.
+                    match thread_results_tx.send((work, result)) {
+                        Ok(_) => (),
+                        Err(_) => { break; },
+                    }
+
+                }
+
+                // Signal to the operating system that now is a good time
+                // to give another thread a chance to run.
+                //
+                // This isn't strictly necessary - the OS can preemptively 
+                // switch between threads, without asking - but it helps make
+                // sure that other threads do get a chance to get some work.
+                std::thread::yield_now();
+            }
+
+            // Report the amount of work done.
+            println!("Thread {} did {} jobs.", thread_num, work_done);
+        });
+
+        // Add the handle for the newly spawned thread to the list of handles
+        threads.push(handle);
+    }
+
+    println!("Workers successfully started.");
+
+    println!("Adding jobs to the queue.");
+    // Variables to keep track of the number of jobs we expect to do.
+    let mut jobs_remaining = 0;
+    let mut jobs_total = 0;
+
+    // Just add some numbers to the queue.
+    // These numbers will be passed into fib(), so they need to stay pretty
+    // small.
+    for work in 0..90 {
+        // Add each one several times.
+        for _ in 0..100 {
+            jobs_remaining = queue.add_work(work);
+            jobs_total += 1;
+        }
+    }
+
+
+    // Report that some jobs were inserted, and how many are left to be done.
+    // This is interesting because the workers have been taking jobs out of the queue
+    // the whole time the control thread has been putting them in!
+    //
+    // Try removing the use of std::thread::yield_now() in the thread closure.
+    // You'll probably (depending on your system) notice that the number remaining 
+    // after insertion goes way up. That's because the operating system is usually
+    // (not always, but usually) fairly conservative about interrupting a thread
+    // that is actually doing work.
+    //
+    // Similarly, if you add a call to yield_now() in the loop above, you'll see the
+    // number remaining probably drop to 1 or 2. This can also change depending on
+    // how optimized the output code is - try `cargo run --release` vs `cargo run`.
+    //
+    // This inconsistency should drive home to you that you as the programmer can't
+    // make any assumptions at all about when and in what order things will happen
+    // in parallel code unless you use thread control primatives as demonstrated
+    // in this program.
+    println!("Total of {} jobs inserted into the queue ({} remaining at this time).", 
+             jobs_total,
+             jobs_remaining);
+
+
+    // Get completed work from the channel while there's work to be done.
+    while jobs_remaining > 0 {
+        match results_rx.recv() {
+            // If the control thread successfully receives, a job was completed.
+            Ok(_) => { jobs_remaining -= 1 },
+            // If the control thread is the one left standing, that's pretty
+            // problematic.
+            Err(_) => {panic!("All workers died unexpectedly.");}
+        }
+    }
+
+
+    // When all the jobs are completed, inform the workers.
+    more_jobs_tx.set(false).unwrap();
+
+    // If we didn't do that, the workers would just look for work forever.
+    // This is useful because many applications of this technique don't 
+    // have a defined stopping point that is known in advance - that is,
+    // they will have to perform a lot of work that isn't known at the time
+    // the work queue is created.
+    //
+    // A SyncFlag can be used so that when the main thread encounters a
+    // kill condition (e.g. Ctrl+C, or perhaps a fatal error of some kind),
+    // it can gracefully shut down all of those workers at once.
+
+
+    // Just make sure that all the other threads are done.
+    for handle in threads {
+        handle.join().unwrap();
+    }
+}