vector-stream Batcher doubts

[jonasrichard]

I need to chase a concurrency issue and for that I need to understand what timeout and sleep exactly do in Batcher poll_next function. Here it is the code

My doubt is: why do we set a timeout if there is anything in the batch in line 79.

So what I think happens:

• Poll::Ready(None) (line 62) is quite clear: if we reached the end of the stream, depending on what we collected return None or the remaining batch
• Poll::Pending (line 90) is also okay: if we have a timer set, let us wait 'timeout' duration and give back the batch. If there is nothing in the batch, timer is set to None in line 76, so we don't wait. Also there is a debug assert to cover that inconsistency. If we don't have timer here, we just return Pending
• Poll::Ready(Some(item)) line 71: once we get an item, let us check if it fits in the batch.
  • If it fits and makes the batch full, set the timer no None and return the batch (so we have this.state.len() == 0)
  • If Item fits but the batch is not full and we have a single Item in the batch (?), set the timer, but here we don't return with that single item, but going to the top of the loop and poll the stream. If there isn't anything more in the stream this single item is waiting here in the batch and maybe only after a significant time we can return that from this function (this is the doubtful part).
  • If Item doesn't fit, take the batch return that and save the item to the new batch. Here also we set the timer, so the next pending part we again wait for timeout quantity time.

Not sure if there is a role of the timeout (quality of service) but I think here we wait at least two times without we need to.

What do others think? Am I confused or have spotted something?

[jonasrichard]

After investigating a bit more I have more doubts about that timer. Let us see what is the context and an execution path where that time can be polled.

So the batcher stream is used in the driver here. This line wraps the batcher to a ReadyChunks futures which in the futures crate. It basically collects items from the underlying stream and collects them into a Vec. Once it gets a Pending, it returns with the collected chunk.

So let us suppose that this batcher reads an item and sets the timer. Then in the next iteration it goes to the Pending branch and polls the timer. This macro returns if the timer is in Pending. So the control goes back to ReadyChunk and it results in Pending, which is correct. In the next calls the ready! works as a gate, it doesn't allow through the control until the timer times out. Once it happens, it sets an empty timer (None) and it returns with the batch (?) we collected. Why do we wait here? On the other hand this makes the future non-cancel safe, since it makes an await point but there is an internal state, so these items can be lost - in case this future is dropped.

Am I right?