Bidirectional streaming fails unless every message is responded to.

[valarauca]

What I Expect

When building a Bidi streaming I could return Ok(Async::NotReady) from with my stream sent to StreamingResponse and still get the next message.

What I Got

When I return Ok(Async::NotReady) within the constructed StreamingResponse my poll never gets called again.

I can see messages leaving my client, then arriving at the server, then being ack'd, but poll is never called.

Workaround

If the server can send an EmptyReply it seems to handle this.

---

Working to build small project to reproduce.

---

Root Cause I think

It seems like httpbis::Service is always expecting to return a response before it can receive a new request. This assumption doesn't generally hold up in grpc.

[stepancheg]

When I return Ok(Async::NotReady) within the constructed StreamingResponse my poll never gets called again.

Are you sure you notify task after you return NotReady?

Quoting documentation: If NotReady is returned then this stream's next value is not ready yet and implementations will ensure that the current task will be notified when the next value may be ready.

[valarauca]

I'm not sure I understand. How do I notify when the next value is ready?

Isn't that the duty of rust-http2/grpc?

[stepancheg]

How do I notify when the next value is ready?

It depends on stream implementation.

For example, if stream is a receiving part of a channel, then notify happens when you send a message to a channel.

If stream is for example created as a wrapper (combinator) around another stream, then notify happens in that underlying stream implementation.

If stream if for example, backed by networked io (like stream for http connection), then notify happens by the event loop.

Anyway, implementing stream yourself is hard. You usually need to provide a stream using one of building blocks.

If you are interested in low level mechanics, have a look at some stream implementation in future-rs. Basically, inside stream implementation you need to call Task::current() to get a pointer to current task, and when you have new data in the stream, you call task.notify(). But I repeat, it is low level stuff.

[valarauca]

For example, if stream is a receiving part of a channel, then notify happens when you send a message to a channel.

This is the very reason I opened this issue.

GRPC allows for the receiving end to just receive messages in a BIDI stream. You don't have to reply to every single one to receive the next.

Yet in a BIDI stream (with the library) you have to send a message for ever received one.

[stepancheg]

Yet in a BIDI stream (with the library) you have to send a message for ever received one.

That's strange. It is not how it must work.

Could you please describe, what stream do you return from method look like? (And of course full test case would be perfect, but not always possible).

[valarauca]

I've attempted to replicate it here.

This is roughly what I'm attempting to do. Where the server is responding only to every 3rd remote message.

But in this example about 50% of the time the client will panic in a separate thread, but continue functioning. The only issue is it sends no network communication.

[stepancheg]

I think I've spotted a bug.

    fn poll(&mut self) -> Result<Async<Option<Self::Item>>,Self::Error> {
        match self.arg.poll() {
            Ok(Async::Ready(Some(hellorequest))) => {
                    ....
                    Ok(Async::NotReady) // <!-- here
                }
             }
             ...
         }
         ...
}

So, when NotReady is returned, underlying stream may not notify task about incoming data, because it returned Ready. Stream is required to notify only after it returned NotReady.

Proper fix would be calling underlying poll again.

[valarauca]

So, when NotReady is returned, underlying stream may not notify task about incoming data, because it returned Ready. Stream is required to notify only after it returned NotReady.

This is the very issue I've been attempting to describe.

It seems a lot nicer, for the library to handle NotReady then having to write loop in poll(&mut self) or is that the expectation?

[stepancheg]

That's the convention of futures-rs library, grpc-rust just follows that convention.

BTW, that convention seems to be reasonable to me. So you suggest that stream must be immediately repolled after NotReady, right? And what if NotReady is returned again, should it repoll again? Probably not. So the rule you (I guess) propose is that after NotReady poll must be called again, but only once? That's too complicated IMHO to enforce that rule for all possible call sites of poll.

Anyway, that's not my decision.