Current 0.3 Sink interface inflicts complexity on implementers

[dekellum]

In my existing futures 0.1 Sink implementations, the single start_send method was able to access the SinkItem (it needs the item buffer's length), calling tokio_threadpool::blocking if necessary and depending on that result, consuming the item buffer and returning Ready, or giving the item back via NotReady:

https://github.com/dekellum/body-image/blob/316596356fd72042cba2c6c4dc00b584ea99576d/body-image-futio/src/uni_sink.rs#L68

I'm now trying to port these Sinks to futures 0.3-alpha. With the current Sink interface, it is necessary to signal any potential for Poll::Pending (for Sink meaning lack of readiness to receive) in the trait method poll_ready, before the Item buffer is given.

https://github.com/dekellum/body-image/blob/316596356fd72042cba2c6c4dc00b584ea99576d/body-image-futio/src/uni_sink.rs#L134

The futures compatibility layer already deals with this interface difference, in the Sink01As03 shim (#1364), by maintaining an additional buffer of one Item, and deferring the actual consumption of that Item to subsequent calls to poll_ready and poll_flush.

I couldn't find an original rationale for this change to the Sink interface? Intuitively I would think it should simplify aspects of the Forward combinator, but at the expense of pushing the issue to most Sink implementers, where the complexity risks bugs. As we are still alpha, is there any willingness to reconsider this interface change? Alternatively, would it be reasonable to introduce here some alternative trait, e.g. TardySink:

/// A sink which doesn't know it's readyiness until it tries to consume an Item
trait TardySink<Item> {
    type SinkError;

    fn poll_send(self: Pin<&mut Self>, cx: &mut Context, item: Item)
        -> (Poll<Result<(), Self::SinkError>>, Option<Item>);

    fn poll_flush(self: Pin<&mut Self>, cx: &mut Context)
        -> Poll<Result<(), Self::SinkError>>;

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context)
        -> Poll<Result<(), Self::SinkError>>;
}

...and impl Sink<Item> for TardySink<Item> using the same 1-item buffering strategy as in Sink01As03, but not depending on anything from futures 0.1? Note, the tuple return type of poll_send above is awkward. An alternative would be a TardyPoll<Item> enum with TardyPoll::Pending(Item) variant and some easy conversion to Poll.

[LucioFranco]

So I think the goal around splitting poll_ready and send is similar to the goal for why we have poll_ready in tower_service::Service. It allows you to send items off task and poll for readiness separately from having to submit an item. I think for your usecase you could just have a sink inbetween that buffers and does the calculations then submits similar to the compat type.

[dekellum]

The counter to that goal, is that there is at least a subset of Sink implementations, where readiness is either (minimally) dependent on the Item, or worse as in my case, were readiness can't be determined until we make a sometimes costly attempt to consume the Item.

To your last point, yes I agree, my proposed support would also need a TardySinkWrapper<T: TardySink> struct to hold the 1-Item buffer. It can't be a blanket impl Sink<Item> for TardySink<Item>. Bare with me—so far I don't fully understand your impl. in #1364, Compat01As03Sink. How much simpler does that get if something like TardySink exists which uses Pin, etc, and is otherwise more consistent with 0.3?

[dekellum]

I've made some progress implementing 0.3 Sinks for both my original application (early links for which are above) as well as a draft attempt at independent reusable generic helper types for the same. Basic tests are passing, including with Pending scenario's. Here is a more complete generic TardySink trait and BufferedSink wrapper type:

https://github.com/dekellum/tardy-sink/blob/f5afecca4/src/lib.rs#L51

Unfortunately there is still a lot about Compat01As03Sink I don't understand, which is worrisome, and I suspect I won't be the last Sink (0.3) implementer to have these questions. Most notably:

• Why does it bother spawning the inner 0.1 Sink on a (potentially Independent?) 0.1 executor and using the returned Spawn::poll_fn_notify, while my simpler implementation and tests, doesn't appear to need this?

• Unlike Compat01As03Sink I'm currently imposing some Unpin bounds on the TardySink impl and its Item. I'm not sure if that makes mine too restrictive, or how hard it would be to remove those bounds?

So I would appreciate some feedback on this with either of the following goals in a new PR:

1. To include this in futures 0.3 for reuse and living documentation, not relegated to the compat feature.

2. Offering this as part of the documentation and as an example (though, I'm not sure which crate and examples directory it fits in).

[Nemo157]

Why does it bother spawning the inner 0.1 Sink on a (potentially Independent?) 0.1 executor and using the returned Spawn::poll_fn_notify, while my simpler implementation and tests, doesn't appear to need this?

That's the compatibility layer, we need to map any 0.1 task::current().notify() to the 0.3 cx.waker().wake(), this is done by sort of running the current poll within a slim executor that does that mapping. If your implementation is 0.3 native it doesn't need this.

Unlike Compat01As03Sink I'm currently imposing some Unpin bounds on the TardySink impl and its Item. I'm not sure if that makes mine too restrictive, or how hard it would be to remove those bounds?

Those bounds can be removed by adding some pin-projection in to the BufferedSink implementation, I would recommend dropping the where Self: Unpin, Item: Unpin from TardySink directly and just have those bounds on the BufferedSink to make removing them in the future easier (hopefully at some point we will have some good documentation on how to do pin-projection...)

(Whether this should be added to futures, I'm not sure, I don't know what all the tradeoffs between the two APIs are, and it seems bad to offer both as interfaces that could be used by libraries, but having a way to easily implement whatever the Sink API is no matter what your underlying data model supports seems useful).

[dekellum]

That's the compatibility layer, we need to map any 0.1 task::current().notify() to the 0.3 cx.waker().wake() [...] If your implementation is 0.3 native it doesn't need this.

That's a relief, thank you. I thought I might be missing something fundamental.

seems bad to offer both as interfaces that could be used by libraries, but having a way to easily implement whatever the Sink API is no matter what your underlying data model supports seems useful.

Fair enough. So it does come back to the need for clarification as to the intention/goals of the 0.3 Sink interface change, and to calibrate against the induced complexity, how frequently Sink implementations have readiness dependent on the Item.

Who might be able to authoritatively answer?

[dekellum]

Those bounds can be removed by adding some pin-projection in to the BufferedSink implementation

I managed to implement the pin-projections and removed the Unpin bounds:

https://github.com/dekellum/tardy-sink/blob/master/src/lib.rs

[dekellum]

RFC: Any positive or negative opinion on at least the problem statement of this issue? @cramertj? @taiki-e? @seanmonstar? (Just picking folks I see contributing/commenting frequently on the project, who haven't yet commented. Apologies if this is unwelcome.)

If something like TardySink is only rarely needed, then I guess I could package this as a small standalone crate. If it will be a common enough problem, asked about here repeatedly and needing a documented solution, then I could offer a PR to merge this here. In futures-sink crate, specifically?

[LucioFranco]

@dekellum I'm sorry if i may be missing something but at a high level im assuming what you want is something like this? https://github.com/libra/libra/blob/master/network/src/sink/buffered_send.rs

[seanmonstar]

I haven't commented because I thought others did a fair job.

Any positive or negative opinion on at least the problem statement of this issue?

I read through the original issue, and the replies, and now I don't know what exactly the problem statement is.

[dekellum]

@LucioFranco At a quick glance, your BufferedSend link looks like about half of whats needed to fix the problem described here. It does the one-buffering (e.g. Option<Item>, but it doesn't offer a single method allowing underlying readiness to be dependent on the Item (length in my case). Could you describe how this is used in libra or by libra users? I'm interested, if you are interested, in seeing if this libra use case could be solved with the same TardySink or similar here.

[dekellum]

@seanmonstar Fair enough, my initial report was less informed/articulate and the thread spends more time on solution implementation details.

Reformulated Problem Statement

There is at least a subset of Sink implementations, where readiness is either (minimally) dependent on the Item, or worse as in my case, were readiness can't be determined until we make a sometimes costly attempt to consume the Item.

In futures 0.1, the Sink interface directly supported this need (allowing the SinkItem to be returned in the single start_send method. In futures 0.3 (as of alpha 16) this interface has been changed, such that readiness must be indicated (via poll_ready) before the Item is seen. The way to deal with these conflicting needs is to buffer one Item, but implementing this may be overly challenging and bug prone for every user that needs it.

---

Thus I'm offering and suggesting merging such a secondary utility as TardySink (yes, naming is hard) as in https://github.com/dekellum/tardy-sink/blob/master/src/lib.rs here.

[taiki-e]

@dekellum

It does the one-buffering (e.g. Option<Item>, but it doesn't offer a single method allowing underlying readiness to be dependent on the Item (length in my case).

How about SinkExt::buffer?

[dekellum]

I saw SinkExt::buffer, @taiki-e, an its implementation informed TardySink. I assume (but have not profiled) the Option<Item> of TardySink has some perf advantage over VecDeque<Item> of capacity 1, but more importantly, Buffer just wraps over the same futures 0.3 Sink interface, so it doesn't satisfy the need of seeing Item to decide readiness, as in TardySink::poll_send.

[cramertj]

I think this is safe to close now that there are workarounds (although they may not have exactly the same performance characteristics of those based on the previous Sink trait). The current shape of the Sink trait was informed by quite a bit of feedback, and I don't personally think there's enough evidence here to suggest that we should go back to the original signatures.

[dekellum]

Part of what I was looking for, in opening this issue 5 months ago, was the details of "quite a bit of feedback" and the design justification for the new Sink trait. I'm sure there must have been reasons for it, but I haven't to date been able to find it. Yes I found a workaround in the course of this. I'm not sure how easy it will be for anyone else to find, left as is, or via this closed issue.