<commit_before>/* <NAME>
 *
 *
 * Ferrous Threads
 *
 */
use std::mem;
use std::boxed::FnBox;
use std::thread::{self, spawn, JoinHandle};
use canal::mpmc::{Sender, Receiver, mpmc_channel};

const QUEUE_SIZE: usize = ((0 - 1) as u8) as usize;


/// The unit of work for a TaskRunner.
pub enum Task<'a> {
    Data(TaskData<'a>),
    Stop,
}

pub struct TaskData<'a> {
    task_func: Box<FnBox() + Send + 'a>,
}

impl<'a> TaskData<'a> {
    fn run(self) {
        self.task_func.call_box(())
    }
}

impl<'a> Task<'a> {
    fn new<F>(func: F) -> Task<'a> where F: FnOnce() + Send + 'a {
        Task::Data(TaskData { task_func: box func })
    }

    fn run(self) {
        match self {
            Task::Data(task) => task.run(),
            Task::Stop => (),
        }
    }
}

/// A TaskRunner is used to run short-lived tasks in parallel without having to
/// spin up a new thread each and every time.
///
/// The TaskRunner will immediately spin up the number of threads that was passed in
/// on creation.
///
/// Spins up a number of threads and distbutes the enqueued tasks through a
/// multi-producer/multi-consumer queue. This allows every worker to draw from the same queue,
/// ensuring that work will be efficiently distributed across the threads.
///
/// # Panics
/// The failure case of a task panicking and destroying the worker is not handled.
///
/// # Examples
/// ```
/// use ferrous_threads::task_runner::TaskRunner;
///
/// use std::sync::mpsc::channel;
///
/// let (sn, rc) = channel();
/// let taskpool = TaskRunner::new(1);
/// taskpool.enqueue(move || { sn.send(9u8).unwrap();}).ok().expect("Task not enqueued");
/// assert!(rc.recv().unwrap() == 9u8);
/// ```
pub struct TaskRunner<'a> {
    queue: Sender<Task<'a>>,
    workers: Vec<JoinHandle<()>>,
}

impl<'a> TaskRunner<'a> {
    pub fn new(num_threads: u8) -> TaskRunner<'a> {
        let (sn, rc): (Sender<Task<'a>>, Receiver<Task<'a>>) = mpmc_channel::<Task>(QUEUE_SIZE);
        let mut guards = Vec::new();
        for _i in 0..num_threads {
            // spawned threads cannot guarantee lifetimes, but we explicitly join on Drop.
            let rc: Receiver<Task<'static>> = unsafe { mem::transmute(rc.clone()) };
            let thr = spawn(move || { TaskRunner::worker(rc) });
            guards.push(thr);
        }
        TaskRunner { queue: sn, workers: guards }
    }

    /// Places the enqueued function on the worker queue.
    pub fn enqueue<F>(&self, func: F) -> Result<(), Task<'a>> where F: 'a + FnOnce() + Send {
        let task = Task::new(func);
        self.queue.send(task)
    }

    fn worker(rc: Receiver<Task>) {
        loop {
            let msg = rc.recv();
            match msg {
                Some(Task::Data(task)) => task.run(),
                Some(Task::Stop) => break,
                None  => thread::yield_now(),
            }
        }
    }<commit_msg>Fix canal 0.1.2 issue<commit_after>/* <NAME>
 *
 *
 * Ferrous Threads
 *
 */
use std::mem;
use std::boxed::FnBox;
use std::thread::{self, spawn, JoinHandle};
use canal::mpmc::{Sender, Receiver, mpmc_channel};

const QUEUE_SIZE: usize = ((0 - 1) as u8) as usize;


/// The unit of work for a TaskRunner.
pub enum Task<'a> {
    Data(TaskData<'a>),
    Stop,
}

pub struct TaskData<'a> {
    task_func: Box<FnBox() + Send + 'a>,
}

impl<'a> TaskData<'a> {
    fn run(self) {
        self.task_func.call_box(())
    }
}

impl<'a> Task<'a> {
    fn new<F>(func: F) -> Task<'a> where F: FnOnce() + Send + 'a {
        Task::Data(TaskData { task_func: box func })
    }

    fn run(self) {
        match self {
            Task::Data(task) => task.run(),
            Task::Stop => (),
        }
    }
}

/// A TaskRunner is used to run short-lived tasks in parallel without having to
/// spin up a new thread each and every time.
///
/// The TaskRunner will immediately spin up the number of threads that was passed in
/// on creation.
///
/// Spins up a number of threads and distbutes the enqueued tasks through a
/// multi-producer/multi-consumer queue. This allows every worker to draw from the same queue,
/// ensuring that work will be efficiently distributed across the threads.
///
/// # Panics
/// The failure case of a task panicking and destroying the worker is not handled.
///
/// # Examples
/// ```
/// use ferrous_threads::task_runner::TaskRunner;
///
/// use std::sync::mpsc::channel;
///
/// let (sn, rc) = channel();
/// let taskpool = TaskRunner::new(1);
/// taskpool.enqueue(move || { sn.send(9u8).unwrap();}).ok().expect("Task not enqueued");
/// assert!(rc.recv().unwrap() == 9u8);
/// ```
pub struct TaskRunner<'a> {
    queue: Sender<Task<'a>>,
    workers: Vec<JoinHandle<()>>,
}

impl<'a> TaskRunner<'a> {
    pub fn new(num_threads: u8) -> TaskRunner<'a> {
        let (sn, rc): (Sender<Task<'a>>, Receiver<Task<'a>>) = mpmc_channel::<Task>(QUEUE_SIZE);
        let mut guards = Vec::new();
        for _i in 0..num_threads {
            // spawned threads cannot guarantee lifetimes, but we explicitly join on Drop.
            let rc: Receiver<Task<'static>> = unsafe { mem::transmute(rc.clone()) };
            let thr = spawn(move || { TaskRunner::worker(rc) });
            guards.push(thr);
        }
        TaskRunner { queue: sn, workers: guards }
    }

    /// Places the enqueued function on the worker queue.
    pub fn enqueue<F>(&self, func: F) -> Result<(), Task<'a>> where F: 'a + FnOnce() + Send {
        let task = Task::new(func);
        self.queue.send(task)
    }

    fn worker(rc: Receiver<Task>) {
        loop {
            let msg = rc.recv();
            match msg {
                Ok(Task::Data(task)) => task.run(),
                Ok(Task::Stop) => break,
                Err(e)  => panic!(e),
            }
        }
    }
{'fraction_non_alphanumeric': 0.17274305555555555, 'fraction_numerical': 0.003993055555555555, 'mean_word_length': 2.425089179548157, 'pattern_counts': {'":': 0, '<': 63, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4509717', 'n_tokens_mistral': 1883, 'n_tokens_neox': 1901, 'n_words': 545}