Repository: alexcrichton/futures-rs
Pull request #632 (2017-11-05 09:29:25)

Implement Sink::fanout
Implements a sink combinator that is able to fan-out writes to multiple underlying sinks.

In my use-case, I'm dealing with an incoming stream of bytes, and my goal is to both forward this stream to an upstream destination, as well as do various stats/aggregation computations in-process, such as calculating a hash digest over the bitstream.

Being able to fan-out to multiple sinks allows me to implement each consumer of the stream as a sink, and then forward the incoming stream to the combined sink.

I'm unsure about the name in particular, maybe you have other suggestions/ideas?

Let me know what you think!

carllerche: Sink fanout is tricky in that it requires handling lagging target sinks. A single blocked destination sink could stall the entire fan out unless it is explicitly dealt with.

There are a few strategies.

a) Allow lagging sinks to stall dispatching.
b) Add buffering (infinite?)
c) Drop messages to stalled sinks.

Could you describe a bit how you handled slow sinks. Also, given that there is no "one size fits all" strategy here, I would rather not add a function on the `Stream` trait.
srijs: Hi @carllerche!

I agree with you on the challenges behind a fan-out mechanism.

The sink combinator that is implemented in this PR will propagate backpressure from the underlying sinks upwards (your option A). I can call this out more explicitly in the docs later.

My reasoning for this is that the backpressure-propagating version is able to function as a building block for the two other versions that are resilient to backpressure.

Given this basic combinator, backpressure resilience can be added in a modular way to the downstream sinks, for instance via the `Buffer` combinator:

```
left_sink.buffer(100).fanout(right_sink.buffer(100))
```

One could also imagine a combinator that discard items when observing backpressure from down stream, which would be composed in a similar way.

Decomposing it like this gives us two main advantages IMO:

1. New ways of coping with backpressure can be easily composed with a single, well tested fan-out implementation
2. It is possible to pick different coping mechanisms for different downstream sinks, depending on the resilience and consistency requirements

Curious to hear your thoughts on this!
carllerche: So, while it isn't the best possible strategy for all cases, I think I agree w/ your argument and it is an acceptable strategy for most cases.

I'm OK w/ this conceptually if @alexcrichton wants to pull it in.
alexcrichton: Thanks for the PR @srijs! This all sounds great to me. I agree that this strategy of "wait for everything to be ready" seems like the lowest common denominator. Buffering can be added (like you've mentioned with the `buffer` combinator) and I imagine we could add an "always ready" combinator which is always ready by dropping message that can't be sent to a lower layer.

@carllerche did you want to request changes to this? Or were you ok with this as-is?

Patch:
diff --git a/src/sink/fanout.rs b/src/sink/fanout.rs
new file mode 100644
index 00000000..8d2456e7
--- /dev/null
+++ b/src/sink/fanout.rs
@@ -0,0 +1,135 @@
+use core::fmt::{Debug, Formatter, Result as FmtResult};
+use core::mem::replace;
+
+use {Async, AsyncSink, Poll, Sink, StartSend};
+
+/// Sink that clones incoming items and forwards them to two sinks at the same time.
+///
+/// Backpressure from any downstream sink propagates up, which means that this sink
+/// can only process items as fast as its _slowest_ downstream sink.
+pub struct Fanout<A: Sink, B: Sink> {
+    left: Downstream<A>,
+    right: Downstream<B>
+}
+
+impl<A: Sink, B: Sink> Fanout<A, B> {
+    /// Consumes this combinator, returning the underlying sinks.
+    ///
+    /// Note that this may discard intermediate state of this combinator,
+    /// so care should be taken to avoid losing resources when this is called.
+    pub fn into_inner(self) -> (A, B) {
+        (self.left.sink, self.right.sink)
+    }
+}
+
+impl<A: Sink + Debug, B: Sink + Debug> Debug for Fanout<A, B>
+    where A::SinkItem: Debug,
+          B::SinkItem: Debug
+{
+    fn fmt(&self, f: &mut Formatter) -> FmtResult {
+        f.debug_struct("Fanout")
+            .field("left", &self.left)
+            .field("right", &self.right)
+            .finish()
+    }
+}
+
+pub fn new<A: Sink, B: Sink>(left: A, right: B) -> Fanout<A, B> {
+    Fanout {
+        left: Downstream::new(left),
+        right: Downstream::new(right)
+    }
+}
+
+impl<A, B> Sink for Fanout<A, B>
+    where A: Sink,
+          A::SinkItem: Clone,
+          B: Sink<SinkItem=A::SinkItem, SinkError=A::SinkError>
+{
+    type SinkItem = A::SinkItem;
+    type SinkError = A::SinkError;
+
+    fn start_send(
+        &mut self, 
+        item: Self::SinkItem
+    ) -> StartSend<Self::SinkItem, Self::SinkError> {
+        // Attempt to complete processing any outstanding requests.
+        self.left.keep_flushing()?;
+        self.right.keep_flushing()?;
+        // Only if both downstream sinks are ready, start sending the next item.
+        if self.left.is_ready() && self.right.is_ready() {
+            self.left.state = self.left.sink.start_send(item.clone())?;
+            self.right.state = self.right.sink.start_send(item)?;
+            Ok(AsyncSink::Ready)
+        } else {
+            Ok(AsyncSink::NotReady(item))
+        }
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), Self::SinkError> {
+        let left_async = self.left.poll_complete()?;
+        let right_async = self.right.poll_complete()?;
+        // Only if both downstream sinks are ready, signal readiness.
+        if left_async.is_ready() && right_async.is_ready() {
+            Ok(Async::Ready(()))
+        } else {
+            Ok(Async::NotReady)
+        }
+    }
+
+    fn close(&mut self) -> Poll<(), Self::SinkError> {
+        let left_async = self.left.close()?;
+        let right_async = self.right.close()?;
+        // Only if both downstream sinks are ready, signal readiness.
+        if left_async.is_ready() && right_async.is_ready() {
+            Ok(Async::Ready(()))
+        } else {
+            Ok(Async::NotReady)
+        } 
+    }
+}
+
+#[derive(Debug)]
+struct Downstream<S: Sink> {
+    sink: S,
+    state: AsyncSink<S::SinkItem>
+}
+
+impl<S: Sink> Downstream<S> {
+    fn new(sink: S) -> Self {
+        Downstream { sink: sink, state: AsyncSink::Ready }
+    }
+
+    fn is_ready(&self) -> bool {
+        self.state.is_ready()
+    }
+
+    fn keep_flushing(&mut self) -> Result<(), S::SinkError> {
+        if let AsyncSink::NotReady(item) = replace(&mut self.state, AsyncSink::Ready) {
+            self.state = self.sink.start_send(item)?;
+        }
+        Ok(())
+    }
+
+    fn poll_complete(&mut self) -> Poll<(), S::SinkError> {
+        self.keep_flushing()?;
+        let async = self.sink.poll_complete()?;
+        // Only if all values have been sent _and_ the underlying
+        // sink is completely flushed, signal readiness.
+        if self.state.is_ready() && async.is_ready() {
+            Ok(Async::Ready(()))
+        } else {
+            Ok(Async::NotReady)
+        }
+    }
+
+    fn close(&mut self) -> Poll<(), S::SinkError> {
+        self.keep_flushing()?;
+        // If all items have been flushed, initiate close.
+        if self.state.is_ready() {
+            self.sink.close()
+        } else {
+            Ok(Async::NotReady)
+        }
+    }
+}
diff --git a/src/sink/mod.rs b/src/sink/mod.rs
index 9dece12b..0d103421 100644
--- a/src/sink/mod.rs
+++ b/src/sink/mod.rs
@@ -21,6 +21,7 @@ mod from_err;
 mod send;
 mod send_all;
 mod map_err;
+mod fanout;
 
 if_std! {
     mod buffer;
@@ -80,6 +81,7 @@ pub use self::send::Send;
 pub use self::send_all::SendAll;
 pub use self::map_err::SinkMapErr;
 pub use self::from_err::SinkFromErr;
+pub use self::fanout::Fanout;
 
 /// A `Sink` is a value into which other values can be sent, asynchronously.
 ///
@@ -407,6 +409,18 @@ pub trait Sink {
         buffer::new(self, amt)
     }
 
+    /// Fanout items to multiple sinks.
+    ///
+    /// This adapter clones each incoming item and forwards it to both this as well as
+    /// the other sink at the same time.
+    fn fanout<S>(self, other: S) -> Fanout<Self, S>
+        where Self: Sized,
+              Self::SinkItem: Clone,
+              S: Sink<SinkItem=Self::SinkItem, SinkError=Self::SinkError>
+    {
+        fanout::new(self, other)
+    }
+
     /// A future that completes when the sink has finished processing all
     /// pending requests.
     ///


Test patch:
diff --git a/tests/sink.rs b/tests/sink.rs
index 6d940934..cb2fdcf2 100644
--- a/tests/sink.rs
+++ b/tests/sink.rs
@@ -352,6 +352,56 @@ fn buffer() {
     }
 }
 
+#[test]
+fn fanout_smoke() {
+    let sink1 = Vec::new();
+    let sink2 = Vec::new();
+    let sink = sink1.fanout(sink2);
+    let stream = futures::stream::iter_ok(vec![1,2,3]);
+    let (sink, _) = sink.send_all(stream).wait().unwrap();
+    let (sink1, sink2) = sink.into_inner();
+    assert_eq!(sink1, vec![1,2,3]);
+    assert_eq!(sink2, vec![1,2,3]);
+}
+
+#[test]
+fn fanout_backpressure() {
+    let (left_send, left_recv) = mpsc::channel(0);
+    let (right_send, right_recv) = mpsc::channel(0);
+    let sink = left_send.fanout(right_send);
+
+    let sink = StartSendFut::new(sink, 0).wait().unwrap();
+    let sink = StartSendFut::new(sink, 1).wait().unwrap();
+ 
+    let flag = Flag::new();
+    let mut task = executor::spawn(sink.send(2));
+    assert!(!flag.get());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
+    let (item, left_recv) = left_recv.into_future().wait().unwrap();
+    assert_eq!(item, Some(0));
+    assert!(flag.get());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
+    let (item, right_recv) = right_recv.into_future().wait().unwrap();
+    assert_eq!(item, Some(0));
+    assert!(flag.get());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
+    let (item, left_recv) = left_recv.into_future().wait().unwrap();
+    assert_eq!(item, Some(1));
+    assert!(flag.get());
+    assert!(task.poll_future_notify(&flag, 0).unwrap().is_not_ready());
+    let (item, right_recv) = right_recv.into_future().wait().unwrap();
+    assert_eq!(item, Some(1));
+    assert!(flag.get());
+    match task.poll_future_notify(&flag, 0).unwrap() {
+        Async::Ready(_) => {
+        },
+        _ => panic!()
+    };
+    // make sure receivers live until end of test to prevent send errors
+    drop(left_recv);
+    drop(right_recv);
+}
+
 #[test]
 fn map_err() {
     {