Use IO task marker in scheduling

[jrbourbeau]

Companion PR to dask/dask#11558 where we incorporate the new io_task attribute into our scheduling logic

[github-actions]

Unit Test Results

See test report for an extended history of previous test failures. This is useful for diagnosing flaky tests.

    21 files   -      6      21 suites   - 6   9h 24m 43s ⏱️ - 2h 1m 38s
 4 128 tests +     1   4 011 ✅ +     2    116 💤 +  4  1 ❌  - 5 
39 621 runs   - 12 093  37 715 ✅  - 11 692  1 903 💤  - 396  3 ❌  - 5 

For more details on these failures, see this check.

Results for commit 2514021. ± Comparison against base commit 8f1b241.

This pull request skips 4 tests.

distributed.cli.tests.test_dask_scheduler ‑ test_signal_handling[Signals.SIGINT]
distributed.cli.tests.test_dask_scheduler ‑ test_signal_handling[Signals.SIGTERM]
distributed.tests.test_client ‑ test_computation_object_code_not_available
distributed.tests.test_scheduler ‑ test_decide_worker_rootish_while_last_worker_is_retiring

♻️ This comment has been updated with latest results.

[graingert]

Suggested change

	while not s.tasks:
	await asyncio.sleep(0.01)
	await async_poll_for(lambda: s.tasks, 5)