Test worker_steal_count hangs sometimes

[Darksonn]

I tried fixing this in the past with #6327, but it seems like that attempt was not enough to fix it, because it's still happening.

tokio/tokio/tests/rt_metrics.rs

Lines 177 to 213 in 431b7c5

	#[test]
	fn worker_steal_count() {
	// This metric only applies to the multi-threaded runtime.
	//
	// We use a blocking channel to backup one worker thread.
	use std::sync::mpsc::channel;
	let rt = threaded_no_lifo();
	let metrics = rt.metrics();
	rt.block_on(async {
	let (tx, rx) = channel();
	// Move to the runtime.
	tokio::spawn(async move {
	// Spawn the task that sends to the channel
	//
	// Since the lifo slot is disabled, this task is stealable.
	tokio::spawn(async move {
	tx.send(()).unwrap();
	});
	// Blocking receive on the channel.
	rx.recv().unwrap();
	})
	.await
	.unwrap();
	});
	drop(rt);
	let n: u64 = (0..metrics.num_workers())
	.map(|i| metrics.worker_steal_count(i))
	.sum();
	assert_eq!(1, n);
	}

For now, I've ignored the test in #6471, but it should be fixed.

[ineol]

I couldn't figure it out, but I have a few observations that could hopefully be useful.

1. The usual cause of non-termination is that the inner task is never scheduled, and one worker gets parked nonethelesss
2. If one adds a call to std::thread::sleep in runtime::scheduler::multi_thread::worker::Context::run after trying to sleep and before parking. (this line)
3. Always notifying in this function called just before parking a core seems to improve the situation: without the thread::sleep mentioned above, I cannot reproduce the bug anymore (in 10^6 runs)
   

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

   Lines 898 to 903 in 6fcd9c0

   	self.is_searching = false;
   	if is_last_searcher {
   	worker.handle.notify_if_work_pending();
   	}

   
   With this change, the send line 196 of the test runs, and the receive happens in the outer task, but it still not terminate sometimes, which confused me.

It could be that somehow,

1. while worker 1 is searching for work in the global queue, worker 2 does not transition to searching because of this condition, then it fails to find work.
2. then, worker 1 runs the task and enqueues a task to its local queue, and gets stuck at rx.recv()
3. then worker 2 parks, the transition_to_parked runs, and there are zero workers in a searching state, and so it doesn't try to notify, and it gets parked
4. the final result is that one worker is executing the task, waiting on the channel, and the other worker is parked.

Instead of always trying to notify, changing the condition below with (is_searching as usize) == prev.num_searching() has the same effect the change in point 3 above.

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

Lines 188 to 198 in 6fcd9c0

	/// Returns `true` if this is the final searching worker.
	fn dec_num_unparked(cell: &AtomicUsize, is_searching: bool) -> bool {
	let mut dec = 1 << UNPARK_SHIFT;
	if is_searching {
	dec += 1;
	}
	let prev = State(cell.fetch_sub(dec, SeqCst));
	is_searching && prev.num_searching() == 1
	}