5
\$\begingroup\$

Purpose. Provides a background task queue so that code can ask for a function to be run at a specified point in the future. They indicate what to run (a closure) and when to run it (also a name just used for tracing). My use case is typically 2-3 items queued at any time, typical items are very fast (think read 100 bytes from a disk). Typically the executions are between 10 and 500ms in the future. It is used for the async IO subsystem in my PDP11 emulator I am porting from c++

The client receives a handle that they can query to see if their task is finished. They can also wait for it to finish.

I posted my first version and got an excellent review and have almost totally recoded it and stolen the notification handle mechanism from the reviewer too. Original Review port of my c++ timer queue to rust

Here is the code

use log::trace;
use std::cmp::Ordering;
use std::collections::BinaryHeap;
use std::sync::mpsc;
use std::sync::mpsc::RecvTimeoutError::{Disconnected, Timeout};
use std::sync::mpsc::{Receiver, Sender};
use std::sync::{Arc, Condvar, Mutex};
use std::thread;
use std::thread::JoinHandle;
use std::time::{Duration, Instant};

pub struct TimerQueue {
    jh: Option<JoinHandle<()>>,
    tx: Sender<QueueInstruction>,
}


type TQIFunc = fn() -> ();
#[derive(Debug)]
struct TimerQueueItem {
    when: Instant, // when it should run
    name: String,  // for trace only
    what: TQIFunc, // what to run
    handle: TimerQueueHandle,
}

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

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

    pub fn wait(&self) {
        let (lock, cv) = &*self.handle;
        let mut finished = lock.lock().unwrap();
        while !*finished {
            finished = cv.wait(finished).unwrap();
        }
    }
    pub fn finished(&self) -> bool {
        let (lock, _cv) = &*self.handle;
        let finished = lock.lock().unwrap();
        *finished
    }
    fn signal(&self) {
        let (lock, cv) = &*self.handle;
        let mut finished = lock.lock().unwrap();
        *finished = true;
        cv.notify_all();
    }
}

enum QueueInstruction {
    Do(TimerQueueItem),
    Stop,
}

// all these Trait impls are required so that binaryheap can sort on due time
// ====================================================
impl Ord for TimerQueueItem {
    fn cmp(&self, other: &TimerQueueItem) -> Ordering {
        other.when.cmp(&self.when)
    }
}
impl PartialOrd for TimerQueueItem {
    fn partial_cmp(&self, other: &TimerQueueItem) -> Option<Ordering> {
        Some(self.cmp(other))
    }
}
impl PartialEq for TimerQueueItem {
    fn eq(&self, other: &Self) -> bool {
        self.when == other.when
    }
}
impl Eq for TimerQueueItem {}
// ====================================================

enum GetItemResult {
    Timeout,
    Stop,
    NewItem(TimerQueueItem),
}
// either we got a new tqi, were told to stop or timed out
// timeout indicates that we need to (probably ) run the top of the queue
// called in one of 2 ways - duration == 0 , wait forever, duration != 0, wait that long
// maybe should be an option
fn get_item(rx: &Receiver<QueueInstruction>, wait: Duration) -> GetItemResult {
    let mut stop = false;
    let mut timeout = false;
    let mut qinst = QueueInstruction::Stop; // init to something
    if wait == Duration::from_secs(0) {
        match rx.recv() {
            Ok(qinstx) => qinst = qinstx,
            Err(_) => stop = true,
        };
    } else {
        match rx.recv_timeout(wait) {
            Ok(qinstx) => qinst = qinstx,
            Err(e) => match e {
                Timeout => timeout = true,
                Disconnected => stop = true,
            },
        };
    }
    if stop {
        GetItemResult::Stop
    } else if timeout {
        GetItemResult::Timeout
    } else {
        match qinst {
            QueueInstruction::Do(tqi) => {
                trace!(target:"TimerQueue","got item {}", tqi.name);
                GetItemResult::NewItem(tqi)
            }
            QueueInstruction::Stop => {
                trace!("got stop request");
                GetItemResult::Stop
            }
        }
    }
}

impl TimerQueue {
    pub fn new() -> TimerQueue {
        let (tx, rx): (Sender<QueueInstruction>, Receiver<QueueInstruction>) = mpsc::channel();
        let jh = thread::spawn(move || {
            let mut queue: BinaryHeap<TimerQueueItem> = BinaryHeap::new();
            let mut stop = false;
            loop {
                if queue.len() == 0 {
                    if stop {
                        break;
                    }
                    match get_item(&rx, Duration::from_secs(0)) {
                        GetItemResult::Stop => {
                            break; // no work and a request to stop - so stop
                        }
                        GetItemResult::NewItem(i) => {
                            queue.push(i);
                        }
                        GetItemResult::Timeout => panic!(), // cannot actually happen here
                    }
                } else {
                    let now = Instant::now();
                    let tqi = queue.peek().expect("oops");
                    let due = tqi.when;
                    if due > now {
                        let wait = due - now;
                        trace!(target:"TimerQueue","sleep for {0}ms", wait.as_millis());
                        match get_item(&rx, wait) {
                            GetItemResult::Stop => {
                                stop = true; // drain the queue first
                            }
                            GetItemResult::NewItem(i) => {
                                queue.push(i);
                            }
                            GetItemResult::Timeout => {
                                continue;
                            }
                        }
                    } else {
                        trace!(target:"TimerQueue","running {0}", tqi.name);
                        (tqi.what)();
                        tqi.handle.signal();
                        queue.pop().unwrap();
                    }
                }
            }
            trace!(target:"TimerQueue", "thread completed");
        });

        TimerQueue { jh: Some(jh), tx }
    }

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

impl Drop for TimerQueue {
    fn drop(&mut self) {
        self.tx.send(QueueInstruction::Stop).unwrap();
        match self.jh.take() {
            Some(jh) => jh.join().unwrap(),
            None => {}
        }
    }
}
fn main() {
    env_logger::init();

    let tq = TimerQueue::new();
    tq.queue(
        || {
            println!("1 sec");
        },
        String::from("1 sec"),
        Instant::now() + Duration::from_millis(1000),
    );

    let h5 = tq.queue(
        || println!("5 sec"),
        String::from("5 sec"),
        Instant::now() + Duration::from_millis(5000),
    );

    tq.queue(
        || {
            println!("1 sec");
        },
        String::from("1 sec"),
        Instant::now() + Duration::from_millis(1000),
    );

    let h3 = tq.queue(
        || {
            println!("3 sec");
        },
        String::from("3 sec"),
        Instant::now() + Duration::from_millis(3000),
    );

    h3.wait();
    println!("h3 done");
    println!("h5 is finised {}", h5.finished());
    h5.wait();
    println!("h5 is finised {}", h5.finished());
}

sorry it fairly long

\$\endgroup\$
2
  • \$\begingroup\$ Hello, to better clarify scope of your project you could add a reference link to the previous post your actual question is a follow up. \$\endgroup\$ Aug 5, 2020 at 5:20
  • \$\begingroup\$ @dariosicily here u go codereview.stackexchange.com/questions/247327/… \$\endgroup\$
    – pm100
    Aug 5, 2020 at 6:08

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.