tokio-sync mpsc poll_ready irrevocably reserves slot

[jonhoo]

Version

tokio-sync 0.1.1

Description

I have a sharded "write handle" that contains multiple mpsc::Senders. When a user wants to send data, they first poll_ready on the write handle, which internally does poll_ready on all its Senders, and when they get Ready they provide the write handle with the data they wish to send. The write handle shards that data, and only sends to the shards for which there is data in the request. This means that if the user only provides data that goes to a few of the shards, every remaining shard will now have a slot permanently taken.

This can be represented by the following test, which fails as tx2.poll_ready() returns NotReady:

#[test]
fn send_recv_clone_ready() {
    let mut task = MockTask::new();
    let (mut tx, mut rx) = mpsc::channel::<i32>(1);
    let mut tx2 = tx.clone();

    assert_ready!(task.enter(|| tx.poll_ready()));
    assert_ready!(task.enter(|| tx2.poll_ready()));
    tx.try_send(1).unwrap();
    assert_not_ready!(task.enter(|| tx.poll_ready()));
    assert_not_ready!(task.enter(|| tx2.poll_ready()));
    let val = assert_ready!(rx.poll());
    assert_eq!(val, Some(1));
    assert_ready!(task.enter(|| tx.poll_ready()));
    assert_ready!(task.enter(|| tx2.poll_ready()));
}

This behavior is pretty unfortunate, and it's not clear how to fix it. My write handle implements tower::Service, so poll_ready doesn't know what data is going to be sent, so it can't only poll_ready the necessary Senders. And there is no way for call to "un-poll" the slots that it has already reserved. As far as I can tell, the only way to work around this is to clone the sender and then drop the old one to force it to let go of its slot? Can we do better?

[jonhoo]

In fact, it's even a little worse than that I think? Is this code okay?

impl Handle {
    fn poll_ready(&mut self) -> Poll<(), Error> {
        for s in &mut self.shards {
            try_ready!(s.poll_ready());
        }
        Ok(Async::Ready(()))
    }
}

Imagine that the second shard returns NotReady the first time Handle::poll_ready is called. Does the Sender in self.shards[0] remember that it has already reserved a slot when Handle::poll_ready is called again, or will it reserve a new slot on the second call to poll_ready?

[carllerche]

This is intentional.

A work around is to add a wrapper around the buffered Sender that has a single buffer slot. Then poll_ready is updated to return Ready when that slot is empty and poll_complete flushes the slot to the inner sink.

[jonhoo]

@carllerche can you speak to whether calling poll_ready twice in a row is okay?

Also, I still think it'd be useful to "return" a reserved slot if you decide not to send after all.

[carllerche]

@jonhoo calling poll_ready twice in a row is OK. I would be OK adding a fn to revoke the capacity.

[carllerche]

I don't know the best way to expose a fn that explicitly revokes capacity w/o dropping the handle. Thoughts?

[jonhoo]

fn disarm(&mut self)?

[mitsuhiko]

I was thinking about this recently and I think having a disarm function would be a good idea. Since the default can be a noop it should be relatively safe to add. There are already some generic wrappers involving poll_ready (for instance the load shed middleware) and I think some of the rules about how they work should probably be better explained in the process.

In particular Clone currently on these services drops the registration. If we add a disarm it would be reasonable to document that a Clone implementation should always do this.