Offload CPU intensive sections of update_graph #7980

Since the introduction of HLGs and particularly since moving to pickle to serialize graphs, Scheduler.update_graph is a very CPU intensive operation.

Most notably, these three steps are eating up a significant amount of time, depending on the size of the graph and the size of the submitted payload.

• Graph deserialization (unpickle)
• Graph materialization
• dask.order

While all of this is running, the scheduler event loop is blocked such that the scheduler and the dashboard appear to be frozen. We're currently not even issuing any log messages telling the user that this is happening.

These CPU intensive tasks should be offloaded to a dedicated thread to allow the remaining scheduler to operate as usual. However, we cannot simply offload the entire Scheduler.update_graph method since the mutation to the scheduler state is not thread safe. For a scheduler that is not at rest, this could cause state corruption since the update_graph modifications and any live updates would interfere.
We also have to ensure that it is not possible to run update_graph concurrently since update_graph is depending on ordering.

I suspect that the easiest way to achieve this without introducing additional race conditions requires a gentle refactoring that looks roughly like this

async def update_graph(self, ...):
    async with self._update_graph_lock:
        ... = await offload(materialize_and_prep_graph, ...)
        # Everything below this may mutate state but everything has to run in the same tick!
        self._generate_taskstates(...)
        ....

def materialize_and_prep_graph(
    self,
    graph_header,
    graph_frames,
    keys,
    internal_priority,
    ...
):
    # Everything we offload should be pure. Might be easiest to ensure if this is a static method/function
    graph = deserialize_graph(graph_header, graph_frames)
    ... = self.materialize_graph(graph)
    if not internal_priority:
        internal_priority = dask.order.order(graph)
    ...