Rethinking CallAll::poll_next order of operations.

[geofmureithi]

Description

In CallAll, once its determined that its not end of a stream, we have the following operations:

1. Fetch the next item in the stream
2. Wait for service to be ready
3. Push the future from .call into the queue.

This can be observed here:

tower/tower/src/util/call_all/common.rs

Lines 112 to 138 in 6283f3a

	// If not done, and we don't have a stored request, gather the next request from the
	// stream (if there is one), or return `Pending` if the stream is not ready.
	if this.curr_req.is_none() {
	*this.curr_req = match ready!(this.stream.as_mut().poll_next(cx)) {
	Some(next_req) => Some(next_req),
	None => {
	// Mark that there will be no more requests.
	*this.eof = true;
	continue;
	}
	};
	}
	// Then, see that the service is ready for another request
	let svc = this
	.service
	.as_mut()
	.expect("Using CallAll after extracting inner Service");
	if let Err(e) = ready!(svc.poll_ready(cx)) {
	// Set eof to prevent the service from being called again after a `poll_ready` error
	*this.eof = true;
	return Poll::Ready(Some(Err(e)));
	}
	// Unwrap: The check above always sets `this.curr_req` if none.
	this.queue.push(svc.call(this.curr_req.take().unwrap()));

Motivation

In apalis, we use tower::CallAll to process jobs and in a scenario where someone has a layer blocking service readiness such as ConcurrencyLayer then it means a Request is polled from the Stream then held until the service becomes ready while there might be other consumers that may be available to handle the request.

• Is it worth considering changing the logic to first check if service is ready then fetch the next item in stream?