Minor async fix

src/network/tokio/demultiplexer.rs

@@ -5,7 +5,6 @@ use tokio::net::{TcpListener, TcpStream};
 #[cfg(feature = "tokio")]
 use tokio::task::JoinHandle;
 
-use anyhow::anyhow;
 use std::collections::HashMap;
 use std::net::ToSocketAddrs;
 
@@ -82,12 +81,12 @@ async fn bind_remotes<In: ExchangeData>(
     let listener = TcpListener::bind(&*address)
         .await
         .map_err(|e| {
-            anyhow!(
+            panic!(
                 "Failed to bind socket for {} at {:?}: {:?}",
                 coord,
                 address,
                 e
-            )
+            ) // TODO
         })
         .unwrap();
     let address = listener

src/operator/map_async.rs

@@ -213,7 +213,7 @@ where
             }
 
             let el = self.prev.next();
-            let kind = el.take();
+            let kind = el.variant();
 
             if let Some(b) = self.batcher.enqueue(el) {
                 self.schedule_batch(b);

src/operator/source/channel.rs

@@ -4,10 +4,10 @@ use flume::{bounded, Receiver, RecvError, Sender, TryRecvError};
 
 use crate::block::{BlockStructure, OperatorKind, OperatorStructure, Replication};
 use crate::operator::source::Source;
-use crate::operator::{Operator, StreamElement};
+use crate::operator::{replication, Operator, StreamElement};
 use crate::scheduler::ExecutionMetadata;
 
-const MAX_RETRY: u8 = 8;
+const MAX_RETRY: u8 = 16;
 
 /// Source that consumes an iterator and emits all its elements into the stream.
 ///
@@ -20,6 +20,7 @@ pub struct ChannelSource<Out: Send + 'static> {
     rx: Receiver<Out>,
     terminated: bool,
     retry_count: u8,
+    replication: Replication,
 }
 
 impl<Out: Send> Display for ChannelSource<Out> {
@@ -31,10 +32,11 @@ impl<Out: Send> Display for ChannelSource<Out> {
 impl<Out: Send + 'static> ChannelSource<Out> {
     /// Create a new source that reads the items from the iterator provided as input.
     ///
-    /// **Note**: this source is **not parallel**, the iterator will be consumed only on a single
-    /// replica, on all the others no item will be read from the iterator. If you want to achieve
-    /// parallelism you need to add an operator that shuffles the data (e.g.
-    /// [`Stream::shuffle`](crate::Stream::shuffle)).
+    /// **Note**: the replication of this source is determined by the `replication` parameter
+    /// The channel is an MPMC channel so items will be captured by one and only one of the replicas
+    /// with no specified order. Developers must take into account this replication when sending items
+    /// to the channel in order for messages to be delivered to renoir (eg. if replication is One,
+    /// only the host with id 0 should send messages to the channel)
     ///
     /// ## Example
     ///
@@ -47,12 +49,13 @@ impl<Out: Send + 'static> ChannelSource<Out> {
     /// tx_channel.send(1);
     /// tx_channel.send(2);
     /// ```
-    pub fn new(channel_size: usize) -> (Sender<Out>, Self) {
+    pub fn new(channel_size: usize, replication: Replication) -> (Sender<Out>, Self) {
         let (tx, rx) = bounded(channel_size);
         let s = Self {
             rx,
             terminated: false,
             retry_count: 0,
+            replication,
         };
 
         (tx, s)
@@ -61,7 +64,7 @@ impl<Out: Send + 'static> ChannelSource<Out> {
 // TODO: remove Debug requirement
 impl<Out: Send + core::fmt::Debug> Source for ChannelSource<Out> {
     fn replication(&self) -> Replication {
-        Replication::One
+        self.replication
     }
 }