AMM: Don't schedule tasks to paused workers

[crusaderky]

In scope

• Prevent Scheduler from assigning new tasks to workers that are paused or closing_gracefully
• Fix a bunch of bugs in the Worker.status -> WorkerState.status sync
• Treat closing_gracefully the same as running in most cases

Follow-ups

• AMM: AMM to avoid paused workers #5440
• AMM: scatter and replicate to avoid paused workers #5441
• AMM: Graceful Worker Retirement #5381

Out of scope

• Steal away tasks from a worker in paused or closing_gracefully status (Tasks are not stolen if worker runs out of memory #3761)
• Prevent workers that are closing_gracefully from running queued tasks

[crusaderky]

I further broke down this PR. It's now ready for review.

[fjetter]

Suggested change

	client_msgs: dict = {}
	worker_msgs: dict = {}
	parent._transitions(recs, client_msgs, worker_msgs)
	self.send_all(client_msgs, worker_msgs)
	self.transitions(recs)

[crusaderky]

Aren't client_msgs and worker_msgs filled in place by _transitions()? I fell for this already before.

[fjetter]

Yes, if you're using the underscored method, you'll need to call the send yourself. If you use the non-underscored one this will be done. You basically just copied what the non-underscored method does,

distributed/distributed/scheduler.py

Lines 7145 to 7155 in f3aa9d1

	def transitions(self, recommendations: dict):
	"""Process transitions until none are left
	This includes feedback from previous transitions and continues until we
	reach a steady state
	"""
	parent: SchedulerState = cast(SchedulerState, self)
	client_msgs: dict = {}
	worker_msgs: dict = {}
	parent._transitions(recommendations, client_msgs, worker_msgs)
	self.send_all(client_msgs, worker_msgs)