sync, coop: apply cooperative scheduling to sync::watch

[tglane]

Motivation

#6839 showed that sync::watch::Receiver::changed will not cooperatively yield control back to the runtime which led to it blocking the thread it was running on.
Edit: This is also true for sync::watch::Receiver::wait_for and sync::watch::Sender::closed

Solution

Added cooperative scheduling to sync::watch::Receiver::changed by utilizing the runtime::coop module.
More specifically I added a new future type that wraps the implementation of sync::watch::Receiver::changed and checks if the current task has exceeded its budget before polling the wrapped future.

Closes #6839.

[Darksonn]

Can you add a test for this?

[tglane]

Sure! It has to be a loom test, right?

[Darksonn]

That's not necessary. You can do something like this:

let my_fut = async { loop { chan.changed().await } };
tokio::select! {
    biased;
    _ = my_fut => {},
    _ = core::future::ready(()) => {},
}

Without coop, this test would run forever.

[tglane]

I added tests to tokio::tests::sync_watch for changed, wait_for and closed.

[Darksonn]

This doesn't make wait_for cooperative, as it may exit before it reaches this call without touching the budget. You need to wrap the entire function.

Also, you may want to make a wait_for_inner method instead of wrapping the entire body in cooperative(async { ... }) to avoid extra indentation.

[tglane]

Oh yea I totally missed that! Thanks. Done in d5a6b3b.

[Darksonn]

Suggested change

	let _ = rx.wait_for(|val| *val == 1).await;
	assert!(rx.wait_for(|val| *val == 1).await.is_err());

[tglane]

Done in d5a6b3b.

[Darksonn]

Thank you.