Add macro yield_now for futures_util::async_await

[DCjanus]

A useful 'helper macro', the asynchronous version of std::thread::yield_now().

Reference: https://users.rust-lang.org/t/what-is-the-asynchronous-version-of-yield-now/24367/2

PS: Thanks for being patient with this. I'm not a native speaker, my English is very bad. If there is something wrong, whether it is code or grammar, feel free to point it out.

[cramertj]

Can you say more about your usecase? How do you intend to wake up this task when it should be polled again?

[DCjanus]

I'm writing a static file server for tide with hyper. In order to read very large files, I need to use the channel to read and send data to user block by block.

And hyper requires that I use poll_ready to check if I can send it, each time I send data through the sender.

Which means, my code like this:

fn work(self, mut sender: hyper::body::Sender, mut file: File) -> FutureObj<'static, Result<(), ()>> {
        let mut chunk:Option<Chunk> = None;
        FutureObj::new(Box::new(async move {
            loop {
                // read 4K data from file to 'chunk', if it is 'None'

                match sender.poll_ready() {
                    Err(_) => {
                        // Some code
                    }
                    Ok(x) => {
                        if x.is_not_ready() {
                            yield_now!();
                            continue;
                        }
                    }
                }

                // send data and clean 'chunk'
            }
        }))
    }

Without yield_now!(), this task will always occupy the thread until the sender is ready, which is unacceptable for the web server.

[cramertj]

Is there a reason you're not using the send method?

[DCjanus]

Document of hyper::body::Sender::send_data say: "This should be called after poll_ready indicated the channel could accept another Chunk."

In fact, I don't want to look at the specific scene. The point is, there is currently no simple way to polling(not Future::poll) something in the async block.

I believe that most of the time there will be more efficient implementations(implement Future and handle the wake up thing), but there are not simple enough.

[DCjanus]

I think yield_now!() is a bad name, but I haven't got a better yet.

[laizy]

maybe one usecase is doing some cpu-intensive work gradually in single thread executor :

async {
	for i in 0..N {
		// do some cpu-intensive work...
		if i % 1000 == 0 {
			yield_now!(); 
		}
	}
}

[cramertj]

@laizy That would be incorrect, since nothing would cause that task to resume. That's why I'm trying to clarify here, as this is the wrong function for most purposes I can think of.

[cramertj]

@DCjanus That poll_ready function is taking an invisible thread-local wakeup handle, so calling it like that wouldn't work anyway. In order to set this up properly, you'd have to use something like the in_notify function in futures-util/src/compat/compat01as03.rs. I'd be okay exposing that function or something like it publicly for compat purposes, but really I'm confused why hyper::body::Sender doesn't implement the Sink trait, whose signature it matches, and which is already supported by Compat.

[DCjanus]

I found that my implementation doesn't always work, I will close this pull request.

Maybe, tokio_timer::Delay is better for this.

[cramertj]

@DCjanus I don't think you want to delay for some arbitrary time period-- that poll_ready function in hyper will call the currently-set wakeup handler in TLS once it should be polled again. Unfortunately, that argument isn't transparent to users so it's easy to miss (this has since been fixed in 0.2 and 0.3 iterations of futures).

[DCjanus]

@cramertj You are right, I am not trying to delay, but i can replace 'yield_now!()' with 'await(Delay::new( now() ))'

[laizy]

@cramertj Does't the inner YieldNow future share the same LocalWaker with the outer one when calling poll? and when YieldNow::poll called, it call LocalWaker::wake, which makes the whole task be scheduled to repoll again. Am I missing something?

    let mut pool = executor::LocalPool::new();
    let mut spawner = pool.spawner();

    let a = async {
        for i in 0..10u32 {
            println!("task a: {}", i);
            if i % 4 == 3 {
                println!("task a begin yield at {}", i);
                yield_now!();
                println!("task a end yield at {}", i);
            }
        }
    };
    let b = async {
        for i in 0..10u32 {
            println!("task b: {}", i);
            if i % 4 == 3 {
                println!("task b begin yield at {}", i);
                yield_now!();
                println!("task b end yield at {}", i);
            }
        }
    };
    spawner.spawn(a);
    spawner.spawn(b);
    pool.run();

output:

task a: 0
task a: 1
task a: 2
task a: 3
task a begin yield at 3
task b: 0
task b: 1
task b: 2
task b: 3
task b begin yield at 3
task a end yield at 3
task a: 4
task a: 5
task a: 6
task a: 7
task a begin yield at 7
task b end yield at 3
task b: 4
task b: 5
task b: 6
task b: 7
task b begin yield at 7
task a end yield at 7
task a: 8
task a: 9
task b end yield at 7
task b: 8
task b: 9

[cramertj]

@laizy Immediately waking will just result in spin-looping.

[laizy]

@cramertj em,I agree this macro is not suitable for @DCjanus 's usecase. But for cpu-intensive task, one option is setup an dedicated cpu-pool. but if i want run it in the current executor, periodically calling yield_now!(); can avoid blocking the thread too long, and let other tasks to make progress.

[cramertj]

@laizy Yeah, that could be true depending on the scheduling model of the executor.

[laizy]

@cramertj I see what you mean now, thanks.