pasts 0.1.0

Minimal and simpler alternative to the futures crate.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124

// Pasts
//
// Copyright (c) 2019-2020 Jeron Aldaron Lau
//
// Licensed under the Apache License, Version 2.0, <LICENSE-APACHE or
// https://apache.org/licenses/LICENSE-2.0>, or the Zlib License, <LICENSE-ZLIB
// or http://opensource.org/licenses/Zlib>, at your option. This file may not be
// copied, modified, or distributed except according to those terms.

use std::{
    boxed::Box,
    sync::{Arc, Condvar, Mutex},
    thread, vec,
    vec::Vec,
};

// A thread that may or may not be running a task for the async loop.
struct JoinHandle {
    mutex: Mutex<Option<()>>,
    condvar: Condvar,
}

// A thread that may or may not be running a task for the async loop.
struct Thread {
    mutex: Mutex<Option<Box<dyn FnOnce() + Send>>>,
    condvar: Condvar,
}

pub(super) struct ThreadPool {
    // Vec of available threads.
    available: Mutex<Vec<ThreadHandle>>,
}

pub(super) struct ThreadHandle {
    join: Arc<JoinHandle>,
    thread: Arc<Thread>,
    pool: Arc<ThreadPool>,
}

impl ThreadHandle {
    pub(super) fn join(self) {
        loop {
            // Lock the mutex.
            let mut guard = self.join.mutex.lock().unwrap();

            // Return if task has completed.
            if guard.take().is_some() {
                break;
            }

            // Wait until not zero (unlock mutex).
            let _guard = self.join.condvar.wait(guard).unwrap();
        }

        // Add back to the stack, as task has completed.
        let pool = self.pool.clone();
        let mut stack = pool.available.lock().unwrap();

        (*stack).push(self);
    }
}

impl ThreadPool {
    /// Create a new ThreadPool.
    pub(super) fn new() -> Arc<Self> {
        Arc::new(ThreadPool {
            available: Mutex::new(vec![]),
        })
    }

    // Take a thread off of the pool stack, and start execution.
    pub(super) fn spawn<F>(self: Arc<Self>, function: F) -> ThreadHandle
    where
        F: FnOnce() + Send + 'static,
    {
        let mut stack = self.available.lock().unwrap();

        if let Some(thread) = (*stack).pop() {
            *thread.thread.mutex.lock().unwrap() = Some(Box::new(function));
            thread.thread.condvar.notify_one();

            thread
        } else {
            let data = Arc::new(Thread {
                mutex: Mutex::new(None),
                condvar: Condvar::new(),
            });
            let thread_data = data.clone();

            let join_handle = Arc::new(JoinHandle {
                mutex: Mutex::new(None),
                condvar: Condvar::new(),
            });
            let thread_join_handle = join_handle.clone();

            *data.mutex.lock().unwrap() = Some(Box::new(function));
            data.condvar.notify_one();

            thread::spawn(move || {
                // Wait for tasks.
                loop {
                    // Lock the mutex.
                    let mut guard = thread_data.mutex.lock().unwrap();

                    // Return if task has completed.
                    if let Some(closure) = guard.take() {
                        closure();
                        *thread_join_handle.mutex.lock().unwrap() = Some(());
                        thread_join_handle.condvar.notify_one();
                    }

                    // Wait until not zero (unlock mutex).
                    let _guard = thread_data.condvar.wait(guard).unwrap();
                }
            });

            ThreadHandle {
                thread: data,
                join: join_handle,
                pool: self.clone(),
            }
        }
    }
}