ppl 0.1.6

A structured parallel programming library for Rust.
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
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167

use super::err::{ReceiverError, SenderError};
use crate::core::configuration::WaitPolicy;

#[cfg(feature = "ff")]
use super::channel_ff as backend;

#[cfg(feature = "crossbeam")]
use super::channel_cb as backend;

#[cfg(feature = "kanal")]
use super::channel_kanal as backend;

#[cfg(feature = "flume")]
use super::channel_flume as backend;

/// Trait defining a channel receiver.
pub trait Receiver<T> {
    /// Receive a message from the channel.
    fn receive(&self) -> Result<Option<T>, ReceiverError>;
    /// Check if the channel is empty.
    fn is_empty(&self) -> bool;
}

/// Trait defining a channel sender.
pub trait Sender<T> {
    /// Send a message to the channel.
    fn send(&self, msg: T) -> Result<(), SenderError>;
}

/// Struct defining the receiver side of a channel.
/// This struct is implemented by the channel backend.
/// The channel backend is selected at compile time by the feature flag.
pub struct ReceiverChannel<T> {
    rx: Box<dyn Receiver<T> + Sync + Send>,
    blocking: bool,
}
impl<T> ReceiverChannel<T>
where
    T: Send,
{
    /// Receive a message from the channel.
    pub fn receive(&self) -> Result<Option<T>, ReceiverError> {
        self.rx.receive()
    }

    /// Receive all messages from the channel, if any.
    /// This method does not block.
    pub fn try_receive_all(&self) -> Vec<T> {
        let mut res = Vec::new();
        // if we're in blocking mode and the queue is empty, then we return immediately to avoid blocking
        while !self.is_empty() {
            match self.receive() {
                Ok(Some(msg)) => res.push(msg),
                Ok(None) => break, // The channel is empty, so we break
                Err(_e) => break,  // The channel is disconnected, so we break
            }
        }

        res
    }

    /// Check if the channel is in blocking mode.
    pub fn is_blocking(&self) -> bool {
        self.blocking
    }
    /// Check if the channel is empty.
    pub fn is_empty(&self) -> bool {
        self.rx.is_empty()
    }
}

/// Struct defining the sender side of a channel.
/// This struct is implemented by the channel backend.
/// The channel backend is selected at compile time by the feature flag.
pub struct SenderChannel<T> {
    tx: Box<dyn Sender<T> + Sync + Send>,
}
impl<T> SenderChannel<T>
where
    T: Send,
{
    /// Send a message to the channel.
    pub fn send(&self, msg: T) -> Result<(), SenderError> {
        self.tx.send(msg)
    }
}

/// Channel factory.
/// This struct is used to create a channel.
/// The channel backend is selected at compile time by the feature flag.
pub struct Channel;

impl Channel {
    /// Create a new channel.
    pub fn channel<T: Send + 'static>(
        wait_policy: WaitPolicy,
    ) -> (ReceiverChannel<T>, SenderChannel<T>) {
        let blocking = match wait_policy {
            WaitPolicy::Active => false,
            WaitPolicy::Passive => true,
        };

        let (rx, tx) = backend::Channel::channel(blocking);
        (ReceiverChannel { rx, blocking }, SenderChannel { tx })
    }
}

#[cfg(test)]
mod tests {
    use serial_test::parallel;

    use super::Channel;

    #[test]
    #[parallel]
    fn test_non_blocking() {
        let (rx, tx) = Channel::channel(crate::core::configuration::WaitPolicy::Active);
        let mut check = true;
        for i in 0..1000 {
            let _ = tx.send(i);
        }

        for i in 0..1000 {
            match rx.receive() {
                Ok(Some(msg)) => {
                    if msg != i {
                        check = false;
                    }
                }
                Ok(None) => {}
                Err(_) => check = false,
            }
        }

        assert!(check)
    }

    #[test]
    #[parallel]
    fn test_blocking() {
        let (rx, tx) = Channel::channel(crate::core::configuration::WaitPolicy::Passive);
        let mut check = true;
        for i in 0..1000 {
            let _ = tx.send(i);
        }

        drop(tx);

        for i in 0..1000 {
            match rx.receive() {
                Ok(Some(msg)) => {
                    if msg != i {
                        check = false;
                    }
                }
                Ok(None) => {}
                Err(_) => check = false,
            }
        }

        if rx.receive().is_ok() {
            check = false;
        }

        assert!(check)
    }
}