Fix test yield_calls_park_before_scheduling_again

[Darksonn]

If you run the following command from the tokio subdirectory:

RUSTFLAGS="-Dwarnings --cfg loom  -C debug_assertions" LOOM_MAX_PREEMPTIONS=1 LOOM_MAX_BRANCHES=10000 cargo test --lib --release --features full -- --test-threads=1 --nocapture loom_multi_thread::yield_now::yield_calls_park_before_scheduling_again

then this test will fail:

tokio/tokio/src/runtime/tests/loom_multi_thread/yield_now.rs

Lines 5 to 30 in f6eb1ee

	#[test]
	fn yield_calls_park_before_scheduling_again() {
	// Don't need to check all permutations
	let mut loom = loom::model::Builder::default();
	loom.max_permutations = Some(1);
	loom.check(|| {
	let rt = mk_runtime(2);
	let (tx, rx) = oneshot::channel::<()>();
	rt.spawn(async {
	let tid = loom::thread::current().id();
	let park_count = park::current_thread_park_count();
	crate::task::yield_now().await;
	if tid == loom::thread::current().id() {
	let new_park_count = park::current_thread_park_count();
	assert_eq!(park_count + 1, new_park_count);
	}
	tx.send(());
	});
	rx.recv();
	});
	}

then the assert_eq!(park_count + 1, new_park_count) assertion will fail like this:

running 1 test
test runtime::tests::loom_multi_thread::yield_now::yield_calls_park_before_scheduling_again ...
thread 'runtime::tests::loom_multi_thread::yield_now::yield_calls_park_before_scheduling_again' panicked at tokio/src/runtime/tests/loom_multi_thread/yield_now.rs:22:17:
assertion `left == right` failed
  left: 1
 right: 0
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
// and some more stuff that's not important

This test should be fixed.

---

Details on why the test fails:

The yield_now function uses a special defer queue to make sure that the task that is using yield_now doesn't get polled again until we have checked the IO/timer drivers for new IO or timer events. This ensures that IO and timer events are given higher priority than the yield_now task. Behind the scenes, the IO and timer drivers are checked by calling the park_timeout function with a timeout of zero. The test in question verifies that the IO/timer drivers are actually checked before the task gets scheduled again.

The test is able to detect whether parking has happened using this counter:

tokio/tokio/src/runtime/park.rs

Lines 35 to 39 in f6eb1ee

	// Bit of a hack, but it is only for loom
	#[cfg(loom)]
	tokio_thread_local! {
	static CURRENT_THREAD_PARK_COUNT: AtomicUsize = AtomicUsize::new(0);
	}

However, there is one situation where the runtime attempts to park, but is unable to actually do so:

tokio/tokio/src/runtime/scheduler/multi_thread/park.rs

Lines 70 to 77 in f6eb1ee

	pub(crate) fn park_timeout(&mut self, handle: &driver::Handle, duration: Duration) {
	// Only parking with zero is supported...
	assert_eq!(duration, Duration::from_millis(0));
	if let Some(mut driver) = self.inner.shared.driver.try_lock() {
	driver.park_timeout(handle, duration);
	}
	}

Basically, the call to try_lock fails if a different worker thread is also calling park_timeout at the same time. In that case, the current thread doesn't actually perform the park operation, since there is another thread that will take care of IO/timer events for us.

To fix this we, need to increment the CURRENT_THREAD_PARK_COUNT counter even if try_lock fails.

[evanrittenhouse]

Feel free to assign this to me, I can pick it up