Merge branch 'feature/mpsc' into 'master'

MPSC (channels)

See merge request picodata/brod/tarantool-module!39

Author: volt0

src/coio.rs

@@ -2,21 +2,24 @@
 //!
 //! See also:
 //! - [C API reference: Module coio](https://www.tarantool.io/en/doc/latest/dev_guide/reference_capi/coio/)
+use std::cell::{Cell, RefCell};
+use std::collections::VecDeque;
 use std::convert::TryFrom;
 use std::ffi::c_void;
-use std::io;
-use std::io::{Read, Write};
+use std::io::{self, Read, Write};
 use std::mem::forget;
 use std::net::{SocketAddr, TcpListener, TcpStream, ToSocketAddrs};
 use std::os::raw::c_char;
 use std::os::unix::io::{AsRawFd, IntoRawFd, RawFd};
+use std::rc::Rc;
 use std::time::Duration;
 
 use failure::_core::ptr::null_mut;
+use num_traits::Zero;
 
 use crate::error::{Error, TarantoolError};
 use crate::ffi::tarantool as ffi;
-use crate::fiber::unpack_callback;
+use crate::fiber::{unpack_callback, Cond};
 
 const TIMEOUT_INFINITY: f64 = 365.0 * 86400.0 * 100.0;
 
@@ -286,3 +289,91 @@ pub(crate) fn write(fd: RawFd, buf: &[u8], timeout: Option<Duration>) -> Result<
         Ok(result as usize)
     }
 }
+
+/// Creates a new asynchronous channel, returning the sender/receiver halves.
+///
+/// All data sent on the Sender will become available on the [Receiver] in the same order as it was sent,
+/// and no `send` will block the calling fiber, `recv` will block until a message is available.
+pub fn channel<T>(capacity: usize) -> (Sender<T>, Receiver<T>) {
+    let chan = Rc::new(Chan {
+        buffer: RefCell::new(VecDeque::with_capacity(capacity)),
+        cond: Cond::new(),
+        tx_count: Cell::new(1),
+        rx_is_active: Cell::new(true),
+    });
+
+    (Sender(chan.clone()), Receiver(chan))
+}
+
+/// The sending-half of channel.
+///
+/// Messages can be sent through this channel with `send`. Can be cloned.
+pub struct Sender<T>(Rc<Chan<T>>);
+
+impl<T> Sender<T> {
+    /// Attempts to send a value on this channel, returning it back if it could not be sent.
+    /// This method will never block.
+    pub fn send(&self, value: T) -> Result<(), io::Error> {
+        if !self.0.rx_is_active.get() {
+            return Err(io::ErrorKind::NotConnected.into());
+        }
+
+        let was_empty = {
+            let mut buffer = self.0.buffer.borrow_mut();
+            let was_empty = buffer.len() == 0;
+            buffer.push_back(value);
+            was_empty
+        };
+
+        if was_empty {
+            self.0.cond.signal();
+        }
+
+        Ok(())
+    }
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        self.0.tx_count.set(self.0.tx_count.get() + 1);
+        Sender(self.0.clone())
+    }
+}
+
+impl<T> Drop for Sender<T> {
+    fn drop(&mut self) {
+        self.0.tx_count.set(self.0.tx_count.get() - 1);
+        self.0.cond.signal();
+    }
+}
+
+/// The receiving half of channel.
+pub struct Receiver<T>(Rc<Chan<T>>);
+
+impl<T> Receiver<T> {
+    /// Attempts to wait for a value on this receiver, returning `None` if the corresponding channel has hung up.
+    pub fn recv(&self) -> Option<T> {
+        if self.0.buffer.borrow().len() == 0 {
+            if self.0.tx_count.get().is_zero() {
+                return None;
+            }
+
+            self.0.cond.wait();
+        }
+
+        self.0.buffer.borrow_mut().pop_front()
+    }
+}
+
+impl<T> Drop for Receiver<T> {
+    fn drop(&mut self) {
+        self.0.rx_is_active.set(false);
+    }
+}
+
+struct Chan<T> {
+    buffer: RefCell<VecDeque<T>>,
+    cond: Cond,
+    tx_count: Cell<usize>,
+    rx_is_active: Cell<bool>,
+}

tests/src/lib.rs

@@ -224,6 +224,9 @@ fn run_tests(cfg: TestConfig) -> Result<bool, io::Error> {
                 test_coio::test_coio_accept,
                 test_coio::test_coio_read_write,
                 test_coio::test_coio_call,
+                test_coio::test_channel,
+                test_coio::test_channel_rx_closed,
+                test_coio::test_channel_tx_closed,
                 test_transaction::test_transaction_commit,
                 test_transaction::test_transaction_rollback,
                 test_log::test_log,

tests/src/test_coio.rs

@@ -1,10 +1,10 @@
 use std::convert::TryInto;
-use std::io::{Read, Write};
+use std::io::{self, Read, Write};
 use std::net::{TcpListener, TcpStream};
 use std::os::unix::io::{AsRawFd, FromRawFd};
 use std::os::unix::net::UnixStream;
 
-use tarantool::coio::{coio_call, CoIOListener, CoIOStream};
+use tarantool::coio::{channel, coio_call, CoIOListener, CoIOStream, Receiver, Sender};
 use tarantool::fiber::{sleep, Fiber};
 
 pub fn test_coio_accept() {
@@ -59,3 +59,49 @@ pub fn test_coio_call() {
     );
     assert_eq!(res, 100)
 }
+
+pub fn test_channel() {
+    let (tx, rx) = channel::<i32>(1);
+
+    let mut fiber_a = Fiber::new("test_fiber_a", &mut |tx: Box<Sender<i32>>| {
+        tx.send(99).unwrap();
+        0
+    });
+    fiber_a.set_joinable(true);
+    fiber_a.start(tx);
+
+    let mut fiber_b = Fiber::new("test_fiber_b", &mut |rx: Box<Receiver<i32>>| {
+        let value = rx.recv().unwrap();
+        assert_eq!(value, 99);
+        0
+    });
+    fiber_b.set_joinable(true);
+    fiber_b.start(rx);
+
+    fiber_a.join();
+    fiber_b.join();
+}
+
+pub fn test_channel_rx_closed() {
+    let (tx, _) = channel::<i32>(1);
+
+    let mut fiber = Fiber::new("test_fiber", &mut |tx: Box<Sender<i32>>| {
+        assert!(tx.send(99).is_err());
+        0
+    });
+    fiber.set_joinable(true);
+    fiber.start(tx);
+    fiber.join();
+}
+
+pub fn test_channel_tx_closed() {
+    let (_, rx) = channel::<i32>(1);
+
+    let mut fiber = Fiber::new("test_fiber", &mut |rx: Box<Receiver<i32>>| {
+        assert!(rx.recv().is_none());
+        0
+    });
+    fiber.set_joinable(true);
+    fiber.start(rx);
+    fiber.join();
+}