FuturesUnordered: Do not poll the same future twice per iteration

[taiki-e]

Instead of using a constant, use the length of FuturesUnordered to ensures that each future is polled only once at most per iteration. This does not solve the fundamental problem, but should be able to improve the current situation somewhat.

This fixes at least one of the problems reported in tokio-rs/tokio#3493. See that issue for details of the bug.

r? @cramertj @jonhoo
cc #2053

[jonhoo]

I think my only concern here is whether the futures are guaranteed to not have their order changed during a single call to FuturesUnordered::poll If that were the case, then we might end up calling (worst case) a single future .len() times if it keeps being moved to the front of the poll list. This is just from memory, but I seem to recall there being a linked list in the implementation with an insert on .wake(), which is where this fear is coming from. If that's not the case, then +1 from me.

It doesn't solve the underlying issue, but does make things at least a bit more sane.

[taiki-e]

I think my only concern here is whether the futures are guaranteed to not have their order changed during a single call to FuturesUnordered::poll

Good point.
IIUC, ReadyToRunQueue::dequeue is the only function that can change the order of futures within the range (0..self.len) that acquired at the first of poll_next. And AFAIK this is only called in Stream::poll_next and Drop::drop that take mutable reference. (enqueue adds a task at the last of the queue, so the order doesn't change.)

futures-rs/futures-util/src/stream/futures_unordered/ready_to_run_queue.rs

Lines 45 to 49 in ddbf522

	/// The dequeue function from the 1024cores intrusive MPSC queue algorithm
	///
	/// Note that this is unsafe as it required mutual exclusion (only one
	/// thread can call this) to be guaranteed elsewhere.
	pub(super) unsafe fn dequeue(&self) -> Dequeue<Fut> {

futures-rs/futures-util/src/stream/futures_unordered/mod.rs

Lines 425 to 427 in ddbf522

	// Safety: &mut self guarantees the mutual exclusion `dequeue`
	// expects
	let task = match unsafe { self.ready_to_run_queue.dequeue() } {

So, my understanding is that the order doesn't change. But I might have missed something.