Repository: dask/distributed

Issue:
LocalCluster: discrepancy default option code & documentation
<!-- Please include a self-contained copy-pastable example that generates the issue if possible.

Please be concise with code posted. See guidelines below on how to provide a good bug report:

- Craft Minimal Bug Reports http://matthewrocklin.com/blog/work/2018/02/28/minimal-bug-reports
- Minimal Complete Verifiable Examples https://stackoverflow.com/help/mcve

Bug reports that follow these guidelines are easier to diagnose, and so are often handled much more quickly.
-->

**Describe the issue**:
The default scheduler port of a LocalCluster is 0 (random port), not 8786 as mentioned in [the documentation](https://distributed.dask.org/en/stable/api.html#distributed.LocalCluster).



**Minimal Complete Verifiable Example**:

```python
# Put your MCVE code here
```

**Anything else we need to know?**:

**Environment**:

- Dask version: dask-2023.7.1 distributed-2023.7.1
- Python version: 3.11.0
- Operating System:
- Install method (conda, pip, source):


Discussion:
GPUtester: Can one of the admins verify this patch?

Admins can comment `ok to test` to allow this one PR to run or `add to allowlist` to allow all future PRs from the same author to run.
github-actions[bot]: ## Unit Test Results
_See [test report](https://dask.github.io/distributed/test_report.html) for an extended history of previous test failures. This is useful for diagnosing flaky tests._

       20 files  ±0         20 suites  ±0   12h 7m 3s [:stopwatch:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.9.0/README.md#the-symbols "duration of all tests") + 40m 11s
  3 752 tests ±0    3 642 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.9.0/README.md#the-symbols "passed tests") +1     104 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.9.0/README.md#the-symbols "skipped / disabled tests") ±0  6 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.9.0/README.md#the-symbols "failed tests")  - 1 
36 477 runs  ±0  34 807 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.9.0/README.md#the-symbols "passed tests") +3  1 663 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.9.0/README.md#the-symbols "skipped / disabled tests") ±0  7 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v2.9.0/README.md#the-symbols "failed tests")  - 3 

For more details on these failures, see [this check](https://github.com/dask/distributed/runs/15624066550).

Results for commit c1470f74. ± Comparison against base commit eb297b3e.

[test-results]:data:application/gzip;base64,H4sIACMEzWQC/02NSQ6DMAwAv4Jy7iGL60A/U4WApagsVZZT1b/XqBA4WTO2xh9BYRqTeDRa3hqRSsgnDSW6HNaFGQxqw4q3edsbe9cHPlPxfnMIF/cKb3ZKQlXkwsQKqxhjXCOb7VcsS/o3rN2xdqGVp9u7iOZQe7eeXLN+neeQGYRXYCVZUDR0hK7t+06SQQKlwbY8rZfQORTfH/8EyuoUAQAA


Patch:
diff --git a/distributed/deploy/local.py b/distributed/deploy/local.py
index 9a9fde7c..69f5d8af 100644
--- a/distributed/deploy/local.py
+++ b/distributed/deploy/local.py
@@ -49,7 +49,7 @@ class LocalCluster(SpecCluster):
     threads_per_worker: int
         Number of threads per each worker
     scheduler_port: int
-        Port of the scheduler.  8786 by default, use 0 to choose a random port
+        Port of the scheduler. Use 0 to choose a random port (default). 8786 is a common choice.
     silence_logs: logging level
         Level of logs to print out to stdout.  ``logging.WARN`` by default.
         Use a falsey value like False or None for no change.
diff --git a/distributed/scheduler.py b/distributed/scheduler.py
index 9e7fc368..9e1b16fd 100644
--- a/distributed/scheduler.py
+++ b/distributed/scheduler.py
@@ -92,6 +92,7 @@ from distributed.metrics import monotonic, time
 from distributed.multi_lock import MultiLockExtension
 from distributed.node import ServerNode
 from distributed.proctitle import setproctitle
+from distributed.protocol import deserialize
 from distributed.protocol.pickle import dumps, loads
 from distributed.protocol.serialize import Serialized, ToPickle, serialize
 from distributed.publish import PublishExtension
@@ -112,6 +113,7 @@ from distributed.utils import (
     key_split_group,
     log_errors,
     no_default,
+    offload,
     recursive_to_dict,
     validate_key,
     wait_for,
@@ -124,6 +126,7 @@ from distributed.utils_comm import (
 )
 from distributed.utils_perf import disable_gc_diagnosis, enable_gc_diagnosis
 from distributed.variable import VariableExtension
+from distributed.worker import dumps_task
 
 if TYPE_CHECKING:
     # TODO import from typing (requires Python >=3.10)
@@ -4318,88 +4321,94 @@ class Scheduler(SchedulerState, ServerNode):
         }
         return msg
 
-    def update_graph(
+    def _match_graph_with_tasks(
+        self, dsk: dict[str, Any], dependencies: dict[str, set[str]], keys: set[str]
+    ) -> set[str]:
+        n = 0
+        lost_keys = set()
+        while len(dsk) != n:  # walk through new tasks, cancel any bad deps
+            n = len(dsk)
+            for k, deps in list(dependencies.items()):
+                if any(
+                    dep not in self.tasks and dep not in dsk for dep in deps
+                ):  # bad key
+                    lost_keys.add(k)
+                    logger.info("User asked for computation on lost data, %s", k)
+                    del dsk[k]
+                    del dependencies[k]
+                    if k in keys:
+                        keys.remove(k)
+                del deps
+        # Avoid computation that is already finished
+        done = set()  # tasks that are already done
+        for k, v in dependencies.items():
+            if v and k in self.tasks:
+                ts = self.tasks[k]
+                if ts.state in ("memory", "erred"):
+                    done.add(k)
+
+        if done:
+            dependents = dask.core.reverse_dict(dependencies)
+            stack = list(done)
+            while stack:  # remove unnecessary dependencies
+                key = stack.pop()
+                try:
+                    deps = dependencies[key]
+                except KeyError:
+                    deps = {ts.key for ts in self.tasks[key].dependencies}
+                for dep in deps:
+                    if dep in dependents:
+                        child_deps = dependents[dep]
+                    elif dep in self.tasks:
+                        child_deps = {ts.key for ts in self.tasks[key].dependencies}
+                    else:
+                        child_deps = set()
+                    if all(d in done for d in child_deps):
+                        if dep in self.tasks and dep not in done:
+                            done.add(dep)
+                            stack.append(dep)
+        for anc in done:
+            dsk.pop(anc, None)
+            dependencies.pop(anc, None)
+        return lost_keys
+
+    def _create_taskstate_from_graph(
         self,
+        *,
+        start: float,
+        dsk: dict,
+        dependencies: dict,
+        keys: set[str],
+        ordered: dict[str, int],
         client: str,
-        graph_header: dict,
-        graph_frames: list[bytes],
-        keys: list[str],
-        internal_priority: dict[str, int] | None,
+        annotations_by_type: dict,
+        global_annotations: dict | None,
+        stimulus_id: str,
         submitting_task: str | None,
         user_priority: int | dict[str, int] = 0,
         actors: bool | list[str] | None = None,
         fifo_timeout: float = 0.0,
         code: tuple[SourceCode, ...] = (),
-        annotations: dict | None = None,
-        stimulus_id: str | None = None,
     ) -> None:
-        start = time()
-        try:
-            # TODO: deserialization + materialization should be offloaded to a
-            # thread since this is non-trivial compute time that blocks the
-            # event loop. This likely requires us to use a lock since we need to
-            # guarantee ordering of update_graph calls (as long as there is just
-            # a single offload thread, this is not a problem)
-            from distributed.protocol import deserialize
-
-            graph = deserialize(graph_header, graph_frames).data
-        except Exception as e:
-            msg = """\
-                Error during deserialization of the task graph. This frequently occurs if the Scheduler and Client have different environments. For more information, see https://docs.dask.org/en/stable/deployment-considerations.html#consistent-software-environments
-            """
-            try:
-                raise RuntimeError(textwrap.dedent(msg)) from e
-            except RuntimeError as e:
-                err = error_message(e)
-                for key in keys:
-                    self.report(
-                        {
-                            "op": "task-erred",
-                            "key": key,
-                            "exception": err["exception"],
-                            "traceback": err["traceback"],
-                        }
-                    )
+        """
+        Take a low level graph and create the necessary scheduler state to
+        compute it.
 
-            return
-        annotations = annotations or {}
-        if isinstance(annotations, ToPickle):  # type: ignore
-            # FIXME: what the heck?
-            annotations = annotations.data  # type: ignore
-
-        stimulus_id = stimulus_id or f"update-graph-{time()}"
-        (
-            dsk,
-            dependencies,
-            annotations_by_type,
-        ) = self.materialize_graph(graph, annotations)
-
-        if internal_priority is None:
-            # Removing all non-local keys before calling order()
-            dsk_keys = set(dsk)  # intersection() of sets is much faster than dict_keys
-            stripped_deps = {
-                k: v.intersection(dsk_keys)
-                for k, v in dependencies.items()
-                if k in dsk_keys
-            }
-            internal_priority = dask.order.order(dsk, dependencies=stripped_deps)
+        WARNING
+        -------
+        This method must not be made async since nothing here is concurrency
+        safe. All interactions with TaskState objects here should be happening
+        in the same event loop tick.
+        """
+
+        lost_keys = self._match_graph_with_tasks(dsk, dependencies, keys)
 
-        requested_keys = set(keys)
-        del keys
         if len(dsk) > 1:
             self.log_event(
                 ["all", client], {"action": "update_graph", "count": len(dsk)}
             )
-        self._pop_known_tasks(
-            known_tasks=self.tasks, dsk=dsk, dependencies=dependencies
-        )
 
-        if lost_keys := self._pop_lost_tasks(
-            dsk=dsk,
-            known_tasks=self.tasks,
-            dependencies=dependencies,
-            keys=requested_keys,
-        ):
+        if lost_keys:
             self.report({"op": "cancelled-keys", "keys": lost_keys}, client=client)
             self.client_releases_keys(
                 keys=lost_keys, client=client, stimulus_id=stimulus_id
@@ -4414,13 +4423,14 @@ class Scheduler(SchedulerState, ServerNode):
 
         if code:  # add new code blocks
             computation.code.add(code)
-        if annotations:
+        if global_annotations:
             # FIXME: This is kind of inconsistent since it only includes global
             # annotations.
-            computation.annotations.update(annotations)
+            computation.annotations.update(global_annotations)
+        del global_annotations
 
         runnable, touched_tasks, new_tasks = self._generate_taskstates(
-            keys=requested_keys,
+            keys=keys,
             dsk=dsk,
             dependencies=dependencies,
             computation=computation,
@@ -4432,7 +4442,7 @@ class Scheduler(SchedulerState, ServerNode):
         )
 
         self._set_priorities(
-            internal_priority=internal_priority,
+            internal_priority=ordered,
             submitting_task=submitting_task,
             user_priority=user_priority,
             fifo_timeout=fifo_timeout,
@@ -4440,11 +4450,11 @@ class Scheduler(SchedulerState, ServerNode):
             tasks=runnable,
         )
 
-        self.client_desires_keys(keys=requested_keys, client=client)
+        self.client_desires_keys(keys=keys, client=client)
 
         # Add actors
         if actors is True:
-            actors = list(requested_keys)
+            actors = list(keys)
         for actor in actors or []:
             ts = self.tasks[actor]
             ts.actor = True
@@ -4496,7 +4506,7 @@ class Scheduler(SchedulerState, ServerNode):
                     self,
                     client=client,
                     tasks=[ts.key for ts in touched_tasks],
-                    keys=requested_keys,
+                    keys=keys,
                     dependencies=dependencies,
                     annotations=dict(annotations_for_plugin),
                     priority=priority,
@@ -4510,6 +4520,88 @@ class Scheduler(SchedulerState, ServerNode):
             if ts.state in ("memory", "erred"):
                 self.report_on_key(ts=ts, client=client)
 
+    @log_errors
+    async def update_graph(
+        self,
+        client: str,
+        graph_header: dict,
+        graph_frames: list[bytes],
+        keys: set[str],
+        internal_priority: dict[str, int] | None,
+        submitting_task: str | None,
+        user_priority: int | dict[str, int] = 0,
+        actors: bool | list[str] | None = None,
+        fifo_timeout: float = 0.0,
+        code: tuple[SourceCode, ...] = (),
+        annotations: dict | None = None,
+        stimulus_id: str | None = None,
+    ) -> None:
+        # FIXME: Apparently empty dicts arrive as a ToPickle object
+        if isinstance(annotations, ToPickle):
+            annotations = annotations.data  # type: ignore[unreachable]
+        start = time()
+        try:
+            try:
+                graph = deserialize(graph_header, graph_frames).data
+                del graph_header, graph_frames
+            except Exception as e:
+                msg = """\
+                    Error during deserialization of the task graph. This frequently occurs if the Scheduler and Client have different environments. For more information, see https://docs.dask.org/en/stable/deployment-considerations.html#consistent-software-environments
+                """
+                raise RuntimeError(textwrap.dedent(msg)) from e
+            (
+                dsk,
+                dependencies,
+                annotations_by_type,
+            ) = await offload(
+                _materialize_graph,
+                graph=graph,
+                global_annotations=annotations or {},
+            )
+            del graph
+            if not internal_priority:
+                # Removing all non-local keys before calling order()
+                dsk_keys = set(
+                    dsk
+                )  # intersection() of sets is much faster than dict_keys
+                stripped_deps = {
+                    k: v.intersection(dsk_keys)
+                    for k, v in dependencies.items()
+                    if k in dsk_keys
+                }
+                internal_priority = await offload(
+                    dask.order.order, dsk=dsk, dependencies=stripped_deps
+                )
+
+            self._create_taskstate_from_graph(
+                dsk=dsk,
+                client=client,
+                dependencies=dependencies,
+                keys=set(keys),
+                ordered=internal_priority or {},
+                submitting_task=submitting_task,
+                user_priority=user_priority,
+                actors=actors,
+                fifo_timeout=fifo_timeout,
+                code=code,
+                annotations_by_type=annotations_by_type,
+                # FIXME: This is just used to attach to Computation
+                # objects. This should be removed
+                global_annotations=annotations,
+                start=start,
+                stimulus_id=stimulus_id or f"update-graph-{start}",
+            )
+        except RuntimeError as e:
+            err = error_message(e)
+            for key in keys:
+                self.report(
+                    {
+                        "op": "task-erred",
+                        "key": key,
+                        "exception": err["exception"],
+                        "traceback": err["traceback"],
+                    }
+                )
         end = time()
         self.digest_metric("update-graph-duration", end - start)
 
@@ -4670,127 +4762,6 @@ class Scheduler(SchedulerState, ServerNode):
                     isinstance(el, (int, float)) for el in ts.priority
                 )
 
-    @staticmethod
-    def _pop_lost_tasks(
-        dsk: dict, keys: set[str], known_tasks: dict[str, TaskState], dependencies: dict
-    ) -> set[str]:
-        n = 0
-        out = set()
-        while len(dsk) != n:  # walk through new tasks, cancel any bad deps
-            n = len(dsk)
-            for k, deps in list(dependencies.items()):
-                if any(
-                    dep not in known_tasks and dep not in dsk for dep in deps
-                ):  # bad key
-                    out.add(k)
-                    logger.info("User asked for computation on lost data, %s", k)
-                    del dsk[k]
-                    del dependencies[k]
-                    if k in keys:
-                        keys.remove(k)
-        return out
-
-    @staticmethod
-    def _pop_known_tasks(
-        known_tasks: dict[str, TaskState], dsk: dict, dependencies: dict
-    ) -> set[str]:
-        # Avoid computation that is already finished
-        already_in_memory = set()  # tasks that are already done
-        for k, v in dependencies.items():
-            if v and k in known_tasks:
-                ts = known_tasks[k]
-                if ts.state in ("memory", "erred"):
-                    already_in_memory.add(k)
-
-        done = set(already_in_memory)
-        if already_in_memory:
-            dependents = dask.core.reverse_dict(dependencies)
-            stack = list(already_in_memory)
-            while stack:  # remove unnecessary dependencies
-                key = stack.pop()
-                try:
-                    deps = dependencies[key]
-                except KeyError:
-                    deps = known_tasks[key].dependencies
-                for dep in deps:
-                    if dep in dependents:
-                        child_deps = dependents[dep]
-                    elif dep in known_tasks:
-                        child_deps = known_tasks[dep].dependencies
-                    else:
-                        child_deps = set()
-                    if all(d in done for d in child_deps):
-                        if dep in known_tasks and dep not in done:
-                            done.add(dep)
-                            stack.append(dep)
-        for anc in done:
-            dsk.pop(anc, None)
-            dependencies.pop(anc, None)
-        return done
-
-    @staticmethod
-    def materialize_graph(
-        hlg: HighLevelGraph, global_annotations: dict
-    ) -> tuple[dict, dict, dict]:
-        from distributed.worker import dumps_task
-
-        dsk = dask.utils.ensure_dict(hlg)
-
-        annotations_by_type: defaultdict[str, dict[str, Any]] = defaultdict(dict)
-        for type_, value in global_annotations.items():
-            annotations_by_type[type_].update(
-                {stringify(k): (value(k) if callable(value) else value) for k in dsk}
-            )
-        for layer in hlg.layers.values():
-            if layer.annotations:
-                annot = layer.annotations
-                for annot_type, value in annot.items():
-                    annotations_by_type[annot_type].update(
-                        {
-                            stringify(k): (value(k) if callable(value) else value)
-                            for k in layer
-                        }
-                    )
-
-        dependencies, _ = get_deps(dsk)
-
-        # Remove `Future` objects from graph and note any future dependencies
-        dsk2 = {}
-        fut_deps = {}
-        for k, v in dsk.items():
-            dsk2[k], futs = unpack_remotedata(v, byte_keys=True)
-            if futs:
-                fut_deps[k] = futs
-        dsk = dsk2
-
-        # - Add in deps for any tasks that depend on futures
-        for k, futures in fut_deps.items():
-            dependencies[k].update(f.key for f in futures)
-        new_dsk = {}
-        exclusive = set(hlg)
-        for k, v in dsk.items():
-            new_k = stringify(k)
-            new_dsk[new_k] = stringify(v, exclusive=exclusive)
-        dsk = new_dsk
-        dependencies = {
-            stringify(k): {stringify(dep) for dep in deps}
-            for k, deps in dependencies.items()
-        }
-
-        # Remove any self-dependencies (happens on test_publish_bag() and others)
-        for k, v in dependencies.items():
-            deps = set(v)
-            if k in deps:
-                deps.remove(k)
-            dependencies[k] = deps
-
-        # Remove aliases
-        for k in list(dsk):
-            if dsk[k] is k:
-                del dsk[k]
-        dsk = valmap(dumps_task, dsk)
-        return dsk, dependencies, dict(annotations_by_type)
-
     def stimulus_queue_slots_maybe_opened(self, *, stimulus_id: str) -> None:
         """Respond to an event which may have opened spots on worker threadpools
 
@@ -8505,3 +8476,66 @@ class CollectTaskMetaDataPlugin(SchedulerPlugin):
                 self.metadata[key] = ts.metadata
                 self.state[key] = finish
                 self.keys.discard(key)
+
+
+def _materialize_graph(
+    graph: HighLevelGraph, global_annotations: dict
+) -> tuple[dict, dict, dict]:
+    dsk = dask.utils.ensure_dict(graph)
+    annotations_by_type: defaultdict[str, dict[str, Any]] = defaultdict(dict)
+    for annotations_type, value in global_annotations.items():
+        annotations_by_type[annotations_type].update(
+            {stringify(k): (value(k) if callable(value) else value) for k in dsk}
+        )
+
+    for layer in graph.layers.values():
+        if layer.annotations:
+            annot = layer.annotations
+            for annot_type, value in annot.items():
+                annotations_by_type[annot_type].update(
+                    {
+                        stringify(k): (value(k) if callable(value) else value)
+                        for k in layer
+                    }
+                )
+    dependencies, _ = get_deps(dsk)
+
+    # Remove `Future` objects from graph and note any future dependencies
+    dsk2 = {}
+    fut_deps = {}
+    for k, v in dsk.items():
+        dsk2[k], futs = unpack_remotedata(v, byte_keys=True)
+        if futs:
+            fut_deps[k] = futs
+    dsk = dsk2
+
+    # - Add in deps for any tasks that depend on futures
+    for k, futures in fut_deps.items():
+        dependencies[k].update(f.key for f in futures)
+    new_dsk = {}
+    # Annotation callables are evaluated on the non-stringified version of
+    # the keys
+    exclusive = set(graph)
+    for k, v in dsk.items():
+        new_k = stringify(k)
+        new_dsk[new_k] = stringify(v, exclusive=exclusive)
+    dsk = new_dsk
+    dependencies = {
+        stringify(k): {stringify(dep) for dep in deps}
+        for k, deps in dependencies.items()
+    }
+
+    # Remove any self-dependencies (happens on test_publish_bag() and others)
+    for k, v in dependencies.items():
+        deps = set(v)
+        if k in deps:
+            deps.remove(k)
+        dependencies[k] = deps
+
+    # Remove aliases
+    for k in list(dsk):
+        if dsk[k] is k:
+            del dsk[k]
+    dsk = valmap(dumps_task, dsk)
+
+    return dsk, dependencies, annotations_by_type
diff --git a/distributed/worker.py b/distributed/worker.py
index 9e6cb988..737938f5 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -1535,23 +1535,24 @@ class Worker(BaseWorker, ServerNode):
             logger.info("Closed worker has not yet started: %s", self.status)
         if not executor_wait:
             logger.info("Not waiting on executor to close")
-        self.status = Status.closing
 
-        # Stop callbacks before giving up control in any `await`.
-        # We don't want to heartbeat while closing.
-        for pc in self.periodic_callbacks.values():
-            pc.stop()
+        # This also informs the scheduler about the status update
+        self.status = Status.closing
+        setproctitle("dask worker [closing]")
 
-        self.stop()
+        if nanny and self.nanny:
+            with self.rpc(self.nanny) as r:
+                await r.close_gracefully(reason=reason)
 
         # Cancel async instructions
         await BaseWorker.close(self, timeout=timeout)
 
-        for preload in self.preloads:
-            try:
-                await preload.teardown()
-            except Exception:
-                logger.exception("Failed to tear down preload")
+        teardowns = [
+            plugin.teardown(self)
+            for plugin in self.plugins.values()
+            if hasattr(plugin, "teardown")
+        ]
+        await asyncio.gather(*(td for td in teardowns if isawaitable(td)))
 
         for extension in self.extensions.values():
             if hasattr(extension, "close"):
@@ -1559,19 +1560,18 @@ class Worker(BaseWorker, ServerNode):
                 if isawaitable(result):
                     await result
 
-        if nanny and self.nanny:
-            with self.rpc(self.nanny) as r:
-                await r.close_gracefully(reason=reason)
+        self.stop_services()
 
-        setproctitle("dask worker [closing]")
+        for preload in self.preloads:
+            try:
+                await preload.teardown()
+            except Exception:
+                logger.exception("Failed to tear down preload")
 
-        teardowns = [
-            plugin.teardown(self)
-            for plugin in self.plugins.values()
-            if hasattr(plugin, "teardown")
-        ]
+        for pc in self.periodic_callbacks.values():
+            pc.stop()
 
-        await asyncio.gather(*(td for td in teardowns if isawaitable(td)))
+        self.stop()
 
         if self._client:
             # If this worker is the last one alive, clean up the worker
@@ -1597,8 +1597,6 @@ class Worker(BaseWorker, ServerNode):
 
         await self.scheduler.close_rpc()
 
-        self.stop_services()
-
         # Give some time for a UCX scheduler to complete closing endpoints
         # before closing self.batched_stream, otherwise the local endpoint
         # may be closed too early and errors be raised on the scheduler when
@@ -1649,10 +1647,11 @@ class Worker(BaseWorker, ServerNode):
         await self.rpc.close()
 
         self.status = Status.closed
+        setproctitle("dask worker [closed]")
+
         await ServerNode.close(self)
 
         self.__exit_stack.__exit__(None, None, None)
-        setproctitle("dask worker [closed]")
         return "OK"
 
     async def close_gracefully(
diff --git a/docs/source/changelog.rst b/docs/source/changelog.rst
index 41d6ef03..6c7e033c 100644
--- a/docs/source/changelog.rst
+++ b/docs/source/changelog.rst
@@ -1,6 +1,53 @@
 Changelog
 =========
 
+.. _v2023.8.0:
+
+2023.8.0
+--------
+
+Released on August 4, 2023
+
+Enhancements
+^^^^^^^^^^^^
+- Offload CPU intensive sections of update graph to unblock event loop (:pr:`8049`) `Florian Jetter`_
+- Log worker close reason in events (:pr:`8042`) `Florian Jetter`_
+- Exclude comm handshake from connect timeout (:pr:`7698`) `Florian Jetter`_
+- Automatically restart P2P shuffles when output worker leaves (:pr:`7970`) `Hendrik Makait`_
+- Add ``Client.unregister_scheduler_plugin`` method (:pr:`7968`) `Brian Phillips`_
+- Fix log message (:pr:`8029`) `Hendrik Makait`_
+- Send shards grouped by input chunk in P2P rechunking (:pr:`8010`) `Hendrik Makait`_
+
+Bug Fixes
+^^^^^^^^^
+- Close state machine and add-ins first in ``Worker.close`` (:pr:`8066`) `Hendrik Makait`_
+- Fix ``decide_worker`` picking a closing worker (:pr:`8032`) `crusaderky`_
+- Raise ``CommClosedError`` in ``get_stream_address`` (:pr:`8020`) `jochenott`_
+- Respect average ``nthreads`` in adaptive (:pr:`8041`) `Matthew Rocklin`_
+- Use queued tasks in adaptive target (:pr:`8037`) `Matthew Rocklin`_
+- Restore support for yield unsafe ``Client`` context managers and deprecate that support (:pr:`7987`) `Thomas Grainger`_
+
+Documentation
+^^^^^^^^^^^^^
+- Change ``worker_saturation`` default value to 1.1 in the documention (:pr:`8040`) `minhnguyenxuan60`_
+- Clarified ``concurrent.futures`` section in ``client.rst`` (:pr:`8048`) `mercyo12`_
+
+Maintenance
+^^^^^^^^^^^
+- Fix flaky ``test_worker_metrics`` (:pr:`8069`) `crusaderky`_
+- Use SPDX in ``license`` metadata (:pr:`8065`) `jakirkham`_
+- Rebalance ``ci1`` markers (:pr:`8061`) `Florian Jetter`_
+- Ensure stream messages are always ordered (:pr:`8059`) `Florian Jetter`_
+- Simplify update graph (:pr:`8047`) `Florian Jetter`_
+- Provide close reason when signal is caught (:pr:`8045`) `Florian Jetter`_
+- Allow unclosed comms in tests (:pr:`8057`) `Florian Jetter`_
+- Cosmetic tweak to ``adaptive_target`` (:pr:`8052`) `crusaderky`_
+- Fix linting (:pr:`8046`) `Florian Jetter`_
+- Update gpuCI ``RAPIDS_VER`` to ``23.10`` (:pr:`8033`)
+- Test against more recent ``pyarrow`` versions (:pr:`8021`) `James Bourbeau`_
+- Add a test for ``GraphLayout`` with ``scatter`` (:pr:`8025`) `Irina Truong`_
+- Fix compatibility variable naming  (:pr:`8030`) `Hendrik Makait`_
+
 .. _v2023.7.1:
 
 2023.7.1
@@ -5144,3 +5191,7 @@ significantly without many new features.
 .. _`Patrick Hoefler`: https://github.com/phofl
 .. _`Irina Truong`: https://github.com/j-bennet
 .. _`Joris Van den Bossche`: https://github.com/jorisvandenbossche
+.. _`Brian Phillips`: https://github.com/bphillips-exos
+.. _`jochenott`: https://github.com/jochenott
+.. _`minhnguyenxuan60`: https://github.com/minhnguyenxuan60
+.. _`mercyo12`: https://github.com/mercyo12
diff --git a/pyproject.toml b/pyproject.toml
index 8c6c28da..0d42107e 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -27,7 +27,7 @@ requires-python = ">=3.9"
 dependencies = [
     "click >= 8.0",
     "cloudpickle >= 1.5.0",
-    "dask == 2023.7.1",
+    "dask == 2023.8.0",
     "jinja2 >= 2.10.3",
     "locket >= 1.0.0",
     "msgpack >= 1.0.0",


Test patch:
diff --git a/distributed/tests/test_scheduler.py b/distributed/tests/test_scheduler.py
index 275148f6..af476c09 100644
--- a/distributed/tests/test_scheduler.py
+++ b/distributed/tests/test_scheduler.py
@@ -303,6 +303,15 @@ async def test_decide_worker_rootish_while_last_worker_is_retiring(c, s, a):
         while a.state.executing_count != 1 or b.state.executing_count != 1:
             await asyncio.sleep(0.01)
 
+        # Rootish is a dynamic property as it is defined right now. Since the
+        # above submit calls are individual update_graph calls, waiting for
+        # tasks to be in executing state on the worker is not sufficient to
+        # guarantee that all the y tasks are already on the scheduler. Only
+        # after at least 5 have been registered, will the task be flagged as
+        # rootish
+        while "y-2" not in s.tasks or not s.is_rootish(s.tasks["y-2"]):
+            await asyncio.sleep(0.01)
+
         # - y-2 has no restrictions
         # - TaskGroup(y) has more than 4 tasks (total_nthreads * 2)
         # - TaskGroup(y) has less than 5 dependency groups
@@ -1358,7 +1367,7 @@ async def test_update_graph_culls(s, a, b):
     )
 
     header, frames = serialize(ToPickle(dsk), on_error="raise")
-    s.update_graph(
+    await s.update_graph(
         graph_header=header,
         graph_frames=frames,
         keys=["y"],
@@ -3678,7 +3687,7 @@ async def test_gather_on_worker_bad_recipient(c, s, a, b):
     """The recipient is missing"""
     x = await c.scatter("x")
     await b.close()
-    assert s.workers.keys() == {a.address}
+    await async_poll_for(lambda: s.workers.keys() == {a.address}, timeout=5)
     out = await s.gather_on_worker(b.address, {x.key: [a.address]})
     assert out == {x.key}
 
diff --git a/distributed/tests/test_steal.py b/distributed/tests/test_steal.py
index a0dadfcc..10569265 100644
--- a/distributed/tests/test_steal.py
+++ b/distributed/tests/test_steal.py
@@ -641,10 +641,11 @@ async def test_steal_when_more_tasks(c, s, a, *rest):
             "slowidentity": 0.2,
             "slow2": 1,
         },
-        "distributed.scheduler.work-stealing-interval": "20ms",
     },
 )
 async def test_steal_more_attractive_tasks(c, s, a, *rest):
+    ext = s.extensions["stealing"]
+
     def slow2(x):
         sleep(1)
         return x
@@ -652,9 +653,27 @@ async def test_steal_more_attractive_tasks(c, s, a, *rest):
     x = c.submit(mul, b"0", 100000000, workers=a.address)  # 100 MB
     await wait(x)
 
+    # The submits below are all individual update_graph calls which are very
+    # likely submitted in the same batch.
+    # Prior to https://github.com/dask/distributed/pull/8049, the entire batch
+    # would be processed by the scheduler in the same event loop tick.
+    # Therefore, the first PC `stealing.balance` call would be guaranteed to see
+    # all the tasks and make the correct decision.
+    # After the PR, the batch is processed in multiple event loop ticks, so the
+    # first PC `stealing.balance` call would potentially only see the first
+    # tasks and would try to rebalance them instead of the slow and heavy one.
+    # To guarantee that the stealing extension sees all tasks, we're stopping
+    # the callback and are calling balance ourselves once we are certain the
+    # tasks are all on the scheduler.
+    # Related https://github.com/dask/distributed/pull/5443
+    await ext.stop()
     futures = [c.submit(slowidentity, x, pure=False, delay=0.2) for i in range(10)]
     future = c.submit(slow2, x, priority=-1)
 
+    while future.key not in s.tasks:
+        await asyncio.sleep(0.01)
+    # Now call it once explicitly to move the heavy task
+    ext.balance()
     while not any(w.state.tasks for w in rest):
         await asyncio.sleep(0.01)