port of my c++ timer queue to rust

Question

This is part of a larger c++ project that I want to migrate to rust (a pdp11 emulator). This is one of the gnarliest bits of the c++ code so I thought I would start with it - its also self contained and relatively small.

Its purpose. It maintains a queue of tasks to be run at some time in the future (typically less than a second away). The tasks are run on a dedicated thread. Typically there are 1 or 2 in the queue at any time. The tasks are typically fast (think read 100 bytes from a file). It is used to simulate the async IO subsystem of the pdp. A task is a closure

This is my first bit of complicated rust code. I dont like two things about it and would welcome suggestions.

• I dont like that the client has to Box the closure. I would prefer to pass the closure in directly.
• I would like to return a 'handle' to my caller so they can tell if there task has finished. The c++ code returns a std::future. Cant work out the equivalent for rust.

Here is what I currently have. The sleep at the end is just to keep the app alive long enough for all the tasks to run

use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};
use log::{trace};

type TQIFunc = Box<dyn Fn() -> () + Send>;


struct TimerQueueItem {
    when: Instant,  // when it should run
    name: String,   // for trace only
    what: TQIFunc,  // what to run
}

// internal implementation
struct _TimerQueue {
    stop: bool,             // tells the queue to do an orderly shutdown
    condvar: Arc<Condvar>,
    queue: Vec<TimerQueueItem>,
}

// the public interface
pub struct TimerQueue {
    inner: Arc<Mutex<_TimerQueue>>,
    thread: Option<JoinHandle<()>>, // if queue is running this is its thread handle
}

impl TimerQueue {
    pub fn start(&mut self) {
        let inner = self.inner.clone();
        let jh = thread::spawn(move || {
            // outer loop runs forever till stop is set.
            loop {
                // inner loop exists once a runnable is ready
                // this 2 loop system is to get the run outside the scope of the queue lock
                let tqi_to_run:TimerQueueItem;
                loop {

                    let mut tqimp = inner.lock().unwrap();
                    let cv = tqimp.condvar.clone();

                    // here we wait on an empty queue for some work to do.
                    while !tqimp.stop && tqimp.queue.is_empty() {
                        tqimp = cv.wait(tqimp).unwrap();
                    }

                    if tqimp.stop {
                        return;
                    }

                    //look at the first item - queue is sorted by time due
                    let now = Instant::now();
                    let tqi = &tqimp.queue[0];
                    let due = tqi.when;
                    // is to due now?
                    if due > now {

                        // no - sleep until it is or until something new arrives
                        let wait = due - now;
                        let (mut tqimp, _) = cv.wait_timeout(tqimp, wait).unwrap();

                        // inspect the fron of the queue again (things may have changed while we were asleep)
                        let tqi = &tqimp.queue[0];
                        let now = Instant::now();
                        let due = tqi.when;
                        if due <= now {
                            // ok its ready to run, pick it off the queue and drop out of inner loop
                            tqi_to_run = tqimp.queue.remove(0);
                            break;
                        }
                        // otherwise loop round again
                    } else {
                        // item is due now, run it. Pick of queue and leave inner loop
                        tqi_to_run = tqimp.queue.remove(0);
                        break;
                    }
                }
                trace!(target:"TimerQueue", "run {0}", tqi_to_run.name);
                // run it - outside queue lock
                (tqi_to_run.what)();
            }
        });

        self.thread = Some(jh);
    }

    pub fn queue(&self, f: TQIFunc, n: String, when: Instant) {
        trace!(target:"TimerQueue", "queued {0}", &n);
        let qi = TimerQueueItem {
            what: f,
            name: n,
            when: when,
        };

        let mut inner = self.inner.lock().unwrap();
        inner.queue.push(qi);
        inner.queue.sort_by_key(|k| k.when);
        inner.condvar.notify_one();
    }

    pub fn new() -> TimerQueue {
        let inner = Arc::new(Mutex::new(_TimerQueue {
            queue: Vec::new(),
            stop: false,
            condvar: Arc::new(Condvar::new()),
        }));
        TimerQueue {
            inner: inner,
            thread: None,
        }
    }
}

impl Drop for TimerQueue {
    fn drop(&mut self) {
        {
            let mut tqimp = self.inner.lock().unwrap();
            tqimp.stop = true;
            tqimp.condvar.notify_one();
        }

        match self.thread.take(){
            Some(jh) =>jh.join().unwrap(),
            None =>{}
        }
    }
}
fn main() {
    env_logger::init();
    let x = || {
        println!("hello");
    };
    let y = || {
        println!("hello2");
    };

    let mut tq = TimerQueue::new();


    tq.queue(
        Box::new(|| {
            println!("first in")
        }),
        String::from("should say first in"),
        Instant::now() + Duration::from_millis(5000),
    );


    tq.start();

    tq.queue(
        Box::new(x),
        String::from("should say hello"),
        Instant::now() + Duration::from_millis(1000),
    );
    tq.queue(
        Box::new(y),
        String::from("should say hello 2"),
        Instant::now() + Duration::from_millis(3000),
    );

    thread::sleep(Duration::from_millis(10000));
}

Answer 1

I'll try to answer your two questions first, then I'll add some general comments.

• I dont like that the client has to Box the closure. I would prefer to pass the closure in directly.

This is actually fairly easy, you just need to change TQIFunc to

type TQIFunc = fn() -> ();

and fix whatever fails to compile.

The original version is specifying a Box to a type object, but since type objects have no size at compilation time it has to be held in some kind of reference object. However, fn() -> () is a function pointer which does have a known size, and thus you can hold it directly in TimerQueueItem.

• I would like to return a 'handle' to my caller so they can tell if there task has finished. The c++ code returns a std::future. Cant work out the equivalent for rust.

Without opening the async can of worms, we can come up with a relatively simple first draft of a solution by using a mutex and condition variable:

struct TimerQueueItem {
    when: Instant,  // when it should run
    name: String,   // for trace only
    what: TQIFunc,  // what to run
    handle: TimerQueueHandle,
}

#[derive(Clone)]
pub struct TimerQueueHandle {
    handle: Arc<(Mutex<bool>, Condvar)>,
}

impl TimerQueueHandle {
    fn new() -> Self {
        Self {
            handle: Arc::new((Mutex::new(false), Condvar::new())),
        }
    }

    pub fn join(&self) {
        let (lock, cv) = &*self.handle;
        let mut finished = lock.lock().unwrap();
        while !*finished {
            finished = cv.wait(finished).unwrap();
        }
    }
    
    fn kill(&self) {
        let (lock, cv) = &*self.handle;
        let mut finished = lock.lock().unwrap();
        *finished = true;
        cv.notify_all();
    }
}

Now we can update the queue function

pub fn queue(&self, f: TQIFunc, n: String, when: Instant) -> TimerQueueHandle {
    trace!(target:"TimerQueue", "queued {0}", &n);
    let handle = TimerQueueHandle::new();
    let qi = TimerQueueItem {
        what: f,
        name: n,
        when: when,
        handle: handle.clone(),
    };

    // <snip>...
    handle
}

After this, you just need to call kill and join in the appropriate places. I'll talk about how this can be improved below.

Additional thoughts

• Inserting into a vector and then sorting will work fine as long as the queue size actually is small, but it's not going to scale well. On the plus side you don't actually need vector semantics here, but rather priority queue semantics. Take a look at std::collections::binary_heap.
• Consider using sentinel values instead of a boolean flag for the queue status, e.g.

enum QueueInstruction {
    Do(TimerQueueItem),
    Stop,
}

with queue: Vec<QueueInstruction>. I find that unifying control flow in this way is much easier to reason about asynchronously. By specializing PartialOrd and adding data fields to the Stop variant, you can also achieve pretty fine-grained control of queue shutdown. Speaking of asynchronous reasoning though...

• Don't use Arc and Mutex at all. Use a std::sync::mpsc channel to communicate with enqueueing threads. Bear in mind that mpsc is a FIFO queue so you'll have to treat it as an intermediate structure and flush it into your priority queue every iteration.
• You can also use an mpsc as the basis for your task handles, instead of the mutex/cv approach. The code for this is left as an exercise, but I think it'll be much friendlier than what I wrote above.