feat(coio): mpsc

Author: Anton Melnikov

src/coio.rs

@@ -16,7 +16,10 @@ use failure::_core::ptr::null_mut;
 
 use crate::error::{Error, TarantoolError};
 use crate::ffi::tarantool as ffi;
-use crate::fiber::unpack_callback;
+use crate::fiber::{unpack_callback, Cond};
+use std::cell::RefCell;
+use std::collections::VecDeque;
+use std::rc::Rc;
 
 const TIMEOUT_INFINITY: f64 = 365.0 * 86400.0 * 100.0;
 
@@ -286,3 +289,62 @@ pub(crate) fn write(fd: RawFd, buf: &[u8], timeout: Option<Duration>) -> Result<
         Ok(result as usize)
     }
 }
+
+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(),
+    });
+
+    (Sender(chan.clone()), Receiver(chan))
+}
+
+pub struct Sender<T>(Rc<Chan<T>>);
+
+impl<T> Sender<T> {
+    pub fn send(&self, value: T) {
+        self.0.send(value)
+    }
+}
+
+impl<T> Clone for Sender<T> {
+    fn clone(&self) -> Self {
+        Sender(self.0.clone())
+    }
+}
+
+pub struct Receiver<T>(Rc<Chan<T>>);
+
+impl<T> Receiver<T> {
+    pub fn recv(&self) -> T {
+        self.0.recv()
+    }
+}
+
+struct Chan<T> {
+    buffer: RefCell<VecDeque<T>>,
+    cond: Cond,
+}
+
+impl<T> Chan<T> {
+    fn send(&self, value: T) {
+        let was_empty = {
+            let mut buffer = self.buffer.borrow_mut();
+            let was_empty = buffer.len() == 0;
+            buffer.push_back(value);
+            was_empty
+        };
+
+        if was_empty {
+            self.cond.signal();
+        }
+    }
+
+    fn recv(&self) -> T {
+        while self.buffer.borrow().len() == 0 {
+            self.cond.wait();
+        }
+
+        self.buffer.borrow_mut().pop_front().unwrap()
+    }
+}

tests/src/lib.rs

@@ -224,6 +224,7 @@ 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_transaction::test_transaction_commit,
                 test_transaction::test_transaction_rollback,
                 test_log::test_log,

tests/src/test_coio.rs

@@ -4,7 +4,7 @@ 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,25 @@ 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);
+        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();
+        assert_eq!(value, 99);
+        0
+    });
+    fiber_b.set_joinable(true);
+    fiber_b.start(rx);
+
+    fiber_a.join();
+    fiber_b.join();
+}