jtp 0.1.0

A simple implementation of a thread pool.
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

use std::{
    collections::HashSet,
    sync::{Arc, Mutex},
    thread,
    time::Duration,
};

use crate::{RejectedTaskHandler, ThreadPoolBuilder};

#[test]
fn test_builder() {
    let thread_pool = ThreadPoolBuilder::default()
        .set_max_pool_size(7)
        .set_channel_capacity(3)
        .build();
    assert!(thread_pool.is_err());

    let thread_pool = ThreadPoolBuilder::default()
        .set_max_pool_size(6)
        .set_core_pool_size(7)
        .build();
    assert!(thread_pool.is_err());

    ThreadPoolBuilder::default().build().unwrap();
}

#[test]
fn test_thread_pool() {
    let mut thread_pool = ThreadPoolBuilder::default()
        .set_core_pool_size(3)
        .set_max_pool_size(7)
        .set_channel_capacity(10)
        .set_keep_alive_time(Duration::from_millis(500))
        .set_rejected_handler(RejectedTaskHandler::Discard)
        .build()
        .unwrap();

    let parent_id = thread::current().id();
    for _ in 0..20 {
        thread_pool
            .execute(move || {
                let cur_id = thread::current().id();
                assert!(parent_id != cur_id);
                // Simulate expensive task.
                thread::sleep(Duration::from_millis(20));
            })
            .unwrap();
    }

    assert_eq!(3, thread_pool.core_workers.len());
    for ref ele in thread_pool.core_workers {
        assert!(ele.is_core);
        assert!(!ele.is_finished());
    }

    // Even too many expensive tasks are executed, the number of
    // workers in the thread pool must be less then or equal to a
    // specified number(7 max_pool_size - 3 core_pool_size).
    assert!(thread_pool.workers.len() <= 4);
    for ref ele in thread_pool.workers.iter() {
        assert!(!ele.is_core);
        assert!(!ele.is_finished());
    }

    thread::sleep(Duration::from_millis(700));
    // 700 ms later, all non-core workers are idle.
    for ref ele in thread_pool.workers.iter() {
        assert!(ele.is_finished());
    }
}

#[test]
fn test_lisenters() {
    let map0 = Arc::new(Mutex::new(HashSet::new()));
    let map1 = map0.clone();
    let map2 = map0.clone();

    let mut thread_pool = ThreadPoolBuilder::default()
        .set_lisenter_before_execute(move |id| {
            let mut map = map0.lock().unwrap();
            map.insert(id);
        })
        .set_lisenter_after_execute(move |id| {
            assert!(map1.lock().unwrap().contains(&id));
        })
        .set_channel_capacity(50)
        .build()
        .unwrap();

    for _ in 0..50 {
        thread_pool
            .execute(|| {
                thread::sleep(Duration::from_millis(20));
            })
            .unwrap();
    }

    thread_pool.close_channel();
    thread_pool.wait().unwrap();
    assert_eq!(50, map2.lock().unwrap().len());
}