Allow worker to prioritize tasks based on memory production/consumption

[mrocklin]

This adds a worker.py::TaskPrefix class that tracks consumption and
production of all computed tasks, grouped by task prefix.

Then we use these values when determining priorities,
(de)prioritizing tasks that (produce)/consume five times more data than
they (consume)/produce.

See #5250 for background

[mrocklin]

OK, I've added pausing, although it's ugly and not very future-reader friendly. This will have to be cleaned up, but it is probably simple enough for people to take a look at if they're interested.

[jrbourbeau]

Thanks for pushing this up @mrocklin. I'll plan to read through #5250 and review this tomorrow

[TomNicholas]

I'm coming up against workers over-eagerly consuming memory, and wondering what the status of this effort is?

Really I'm just looking for a way to get workers to deprioritise certain memory-consuming root tasks (xr.open_dataset tasks). I can open another issue if that's worthwhile.

[gjoseph92]

@TomNicholas see #5223 and #5555 for discussion of the underlying problem. We're focused on core stability issues (deadlocks) right now, so changes to the scheduling algorithm to address this are not getting attention at the moment.

Opening a separate issue to discuss workarounds for root task overproduction might be useful. There may be tricks you can play using worker resources, though I haven't had much success with them personally.