Repository: dask/distributed

Issue:
Fine performance metrics: measure time spent entering/exiting thread pools
- Part of #7665 
- Related to #7655 
- Related to #5882

When investigating #7655, I got the suspicion that we may have a bottleneck caused by the fact that there's only one worker in the offload executor.
Additionally, there's a known issue (#5882) where, whenever a task calls `rejoin()`, you will have one less thread available then what the WorkerStateMachine believes is available, which will cause `Worker.execute` to get stuck for a long time on the call to `run_in_executor`.

In both cases, as of #7586 this time is displayed as `("execute", <prefix>, "other", "seconds")`.
Wrap these two calls to `run_in_executor` in  `with context_meter.meter("offload")` and `with context_meter.meter("executor")` respectively.

Discussion:
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._

       26 files  +       1         26 suites  +1   14h 0m 18s [:stopwatch:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "duration of all tests") + 55m 46s
  3 548 tests ±       0    3 433 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests")  -        4     105 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests")  -   1  9 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") +4  1 [:fire:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "test errors") +1 
44 885 runs  +2 308  42 636 [:heavy_check_mark:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "passed tests") +2 217  2 239 [:zzz:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "skipped / disabled tests") +86  9 [:x:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "failed tests") +4  1 [:fire:](https://github.com/EnricoMi/publish-unit-test-result-action/blob/v1.20/README.md#the-symbols "test errors") +1 

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

Results for commit a639fc40. ± Comparison against base commit 1ccc312a.

[test-results]:data:application/gzip;base64,H4sIAOn9LmQC/03Myw7CIBCF4VdpWLvgMkwHX8Yg0oTYi6Gwanx3QS3t8v9mcjY2hNGv7NpJvHRszSEd9cjRprDMpTUHQYXKNdW70tDytmbnqoFShz3Dq5jgutFgw1jINPAxLrE+FYl5rrsARPqf+y5IVNjstyulMjuddr99nnXLNIVUgllUZnDApQHnSVirEbxW/I69QSHAaUdkemLvD0zr/JwUAQAA

crusaderky: Believe it or not, all 11 CI failures are unrelated to this change. Ready for review and merge.

Patch:
diff --git a/.github/workflows/update-gpuci.yaml b/.github/workflows/update-gpuci.yaml
index 08f07301..ec172425 100644
--- a/.github/workflows/update-gpuci.yaml
+++ b/.github/workflows/update-gpuci.yaml
@@ -54,7 +54,7 @@ jobs:
           regex: false
 
       - name: Create Pull Request
-        uses: peter-evans/create-pull-request@v4
+        uses: peter-evans/create-pull-request@v5
         if: ${{ env.UCX_PY_VER != env.NEW_UCX_PY_VER }}  # make sure new ucx-py nightlies are available
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
diff --git a/continuous_integration/environment-3.10.yaml b/continuous_integration/environment-3.10.yaml
index aa5b9dc6..48172dd6 100644
--- a/continuous_integration/environment-3.10.yaml
+++ b/continuous_integration/environment-3.10.yaml
@@ -47,4 +47,4 @@ dependencies:
       - git+https://github.com/dask/zict
       - git+https://github.com/fsspec/filesystem_spec
       - keras
-      - gilknocker>=0.3.0
+      - gilknocker>=0.4.0
diff --git a/continuous_integration/environment-3.11.yaml b/continuous_integration/environment-3.11.yaml
index 5ec7e6b9..379f4bba 100644
--- a/continuous_integration/environment-3.11.yaml
+++ b/continuous_integration/environment-3.11.yaml
@@ -47,4 +47,4 @@ dependencies:
       - git+https://github.com/dask/zict
       - git+https://github.com/fsspec/filesystem_spec
       - keras
-      - gilknocker>=0.3.0
+      - gilknocker>=0.4.0
diff --git a/continuous_integration/environment-3.8.yaml b/continuous_integration/environment-3.8.yaml
index 62b303a5..5a41cef5 100644
--- a/continuous_integration/environment-3.8.yaml
+++ b/continuous_integration/environment-3.8.yaml
@@ -48,4 +48,4 @@ dependencies:
       - git+https://github.com/dask/dask
       - git+https://github.com/jcrist/crick  # Only tested here
       - keras
-      - gilknocker>=0.3.0
+      - gilknocker>=0.4.0
diff --git a/continuous_integration/environment-3.9.yaml b/continuous_integration/environment-3.9.yaml
index f3040ba7..a43f24cd 100644
--- a/continuous_integration/environment-3.9.yaml
+++ b/continuous_integration/environment-3.9.yaml
@@ -50,4 +50,4 @@ dependencies:
   - pip:
       - git+https://github.com/dask/dask
       - keras
-      - gilknocker>=0.3.0
+      - gilknocker>=0.4.0
diff --git a/distributed/http/scheduler/prometheus/core.py b/distributed/http/scheduler/prometheus/core.py
index 5f2e78fa..66067563 100644
--- a/distributed/http/scheduler/prometheus/core.py
+++ b/distributed/http/scheduler/prometheus/core.py
@@ -52,6 +52,13 @@ class SchedulerMetricCollector(PrometheusCollector):
         )
         yield worker_states
 
+        if self.server.monitor.monitor_gil_contention:
+            yield CounterMetricFamily(
+                self.build_name("gil_contention"),
+                "GIL contention metric",
+                value=self.server.monitor._cumulative_gil_contention,
+            )
+
         tasks = GaugeMetricFamily(
             self.build_name("tasks"),
             "Number of tasks known by scheduler",
diff --git a/distributed/http/worker/prometheus/core.py b/distributed/http/worker/prometheus/core.py
index b8a60775..23aef2d9 100644
--- a/distributed/http/worker/prometheus/core.py
+++ b/distributed/http/worker/prometheus/core.py
@@ -59,6 +59,13 @@ class WorkerMetricCollector(PrometheusCollector):
             value=ws.transfer_incoming_count,
         )
 
+        if self.server.monitor.monitor_gil_contention:
+            yield CounterMetricFamily(
+                self.build_name("gil_contention"),
+                "GIL contention metric",
+                value=self.server.monitor._cumulative_gil_contention,
+            )
+
         yield GaugeMetricFamily(
             self.build_name("threads"),
             "Number of worker threads",
diff --git a/distributed/system_monitor.py b/distributed/system_monitor.py
index 94c9a012..85bdae0d 100644
--- a/distributed/system_monitor.py
+++ b/distributed/system_monitor.py
@@ -30,6 +30,8 @@ class SystemMonitor:
     _last_host_cpu_counters: Any  # dynamically-defined psutil namedtuple
     _last_gil_contention: float  # 0-1 value
 
+    _cumulative_gil_contention: float
+
     gpu_name: str | None
     gpu_memory_total: int
 
@@ -108,6 +110,7 @@ class SystemMonitor:
                 self.monitor_gil_contention = False
             else:
                 self.quantities["gil_contention"] = deque(maxlen=maxlen)
+                self._cumulative_gil_contention = 0.0
                 raw_interval = dask.config.get(
                     "distributed.admin.system-monitor.gil.interval",
                 )
@@ -191,6 +194,7 @@ class SystemMonitor:
 
         if self.monitor_gil_contention:
             self._last_gil_contention = self._gilknocker.contention_metric
+            self._cumulative_gil_contention += self._last_gil_contention
             result["gil_contention"] = self._last_gil_contention
             self._gilknocker.reset_contention_metric()
 
diff --git a/distributed/utils.py b/distributed/utils.py
index 4c1a6642..52a2cc11 100644
--- a/distributed/utils.py
+++ b/distributed/utils.py
@@ -67,7 +67,7 @@ from dask.utils import parse_timedelta as _parse_timedelta
 from dask.widgets import get_template
 
 from distributed.compatibility import WINDOWS
-from distributed.metrics import monotonic, time
+from distributed.metrics import context_meter, monotonic, time
 
 try:
     from dask.context import thread_state
@@ -1460,7 +1460,8 @@ def offload(
     run_in_executor_with_context
     https://bugs.python.org/issue34014
     """
-    return run_in_executor_with_context(_offload_executor, func, *args, **kwargs)
+    with context_meter.meter("offload"):
+        return run_in_executor_with_context(_offload_executor, func, *args, **kwargs)
 
 
 class EmptyContext:
diff --git a/distributed/worker.py b/distributed/worker.py
index 1513a045..f443237c 100644
--- a/distributed/worker.py
+++ b/distributed/worker.py
@@ -2269,20 +2269,29 @@ class Worker(BaseWorker, ServerNode):
                         self.scheduler_delay,
                     )
                 elif "ThreadPoolExecutor" in str(type(e)):
-                    result = await run_in_executor_with_context(
-                        e,
-                        apply_function,
-                        function,
-                        args2,
-                        kwargs2,
-                        self.execution_state,
-                        key,
-                        self.active_threads,
-                        self.active_threads_lock,
-                        self.scheduler_delay,
-                    )
+                    # The 'executor' time metric should be almost zero most of the time,
+                    # e.g. thread synchronization overhead only, since thread-noncpu and
+                    # thread-cpu inside the thread detract from it. However, it may
+                    # become substantial in case of misalignment between the size of the
+                    # thread pool and the number of running tasks in the worker state
+                    # machine (e.g. https://github.com/dask/distributed/issues/5882)
+                    with context_meter.meter("executor"):
+                        result = await run_in_executor_with_context(
+                            e,
+                            apply_function,
+                            function,
+                            args2,
+                            kwargs2,
+                            self.execution_state,
+                            key,
+                            self.active_threads,
+                            self.active_threads_lock,
+                            self.scheduler_delay,
+                        )
                 else:
-                    # Can't capture contextvars across processes
+                    # Can't capture contextvars across processes. If this is a
+                    # ProcessPoolExecutor, the 'executor' time metric will show the
+                    # whole runtime inside the executor.
                     with context_meter.meter("executor"):
                         result = await self.loop.run_in_executor(
                             e,
diff --git a/docs/source/prometheus.rst b/docs/source/prometheus.rst
index d2707c25..3c9225e7 100644
--- a/docs/source/prometheus.rst
+++ b/docs/source/prometheus.rst
@@ -26,6 +26,15 @@ dask_scheduler_clients
     Number of clients connected
 dask_scheduler_desired_workers
     Number of workers scheduler needs for task graph
+dask_scheduler_gil_contention_total
+    Value representing cumulative total of GIL contention,
+    in the form of summed percentages.
+
+    .. note::
+       Requires ``gilknocker`` to be installed, and 
+       ``distributed.admin.system-monitor.gil.enabled``
+       configuration to be set.
+
 dask_scheduler_workers
     Number of workers known by scheduler
 dask_scheduler_tasks
@@ -117,6 +126,15 @@ dask_worker_tasks
     Number of tasks at worker
 dask_worker_threads
     Number of worker threads
+dask_worker_gil_contention_total
+    Value representing cumulative total GIL contention on worker,
+    in the form of summed percentages.
+
+    .. note::
+       Requires ``gilknocker`` to be installed, and
+       ``distributed.admin.system-monitor.gil.enabled``
+       configuration to be set.
+
 dask_worker_latency_seconds
     Latency of worker connection
 dask_worker_memory_bytes


Test patch:
diff --git a/distributed/http/tests/test_core.py b/distributed/http/tests/test_core.py
index 4cf85bba..7f43d581 100644
--- a/distributed/http/tests/test_core.py
+++ b/distributed/http/tests/test_core.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import pathlib
+from unittest import mock
 
 import pytest
 from tornado.httpclient import AsyncHTTPClient
@@ -16,13 +17,26 @@ async def test_scheduler(c, s, a, b):
     assert response.code == 200
 
 
-@gen_cluster(client=True, nthreads=[("", 1)])
-async def test_prometheus_api_doc(c, s, a):
+@mock.patch("warnings.warn", return_value=None)
+@gen_cluster(
+    client=True,
+    nthreads=[("", 1)],
+    config={"distributed.admin.system-monitor.gil.enabled": True},
+)
+async def test_prometheus_api_doc(c, s, a, _):
     """Test that the Sphinx documentation of Prometheus endpoints matches the
     implementation.
     """
     pytest.importorskip("prometheus_client")
 
+    documented = set()
+    root_dir = pathlib.Path(__file__).parent.parent.parent.parent
+    with open(root_dir / "docs" / "source" / "prometheus.rst") as fh:
+        for row in fh:
+            row = row.strip()
+            if row.startswith("dask_"):
+                documented.add(row)
+
     # Some metrics only appear if there are tasks on the cluster
     fut = c.submit(inc, 1)
     await fut
@@ -53,14 +67,15 @@ async def test_prometheus_api_doc(c, s, a):
             "dask_worker_transfer_bandwidth_median_bytes",
         }
 
-    implemented = scheduler_metrics | worker_metrics | crick_metrics
+    try:
+        import gilknocker  # noqa: F401
 
-    documented = set()
-    root_dir = pathlib.Path(__file__).parent.parent.parent.parent
-    with open(root_dir / "docs" / "source" / "prometheus.rst") as fh:
-        for row in fh:
-            row = row.strip()
-            if row.startswith("dask_"):
-                documented.add(row)
+        gil_metrics = set()  # Already in worker_metrics
+    except ImportError:
+        gil_metrics = {
+            "dask_scheduler_gil_contention_total",
+            "dask_worker_gil_contention_total",
+        }
 
+    implemented = scheduler_metrics | worker_metrics | crick_metrics | gil_metrics
     assert documented == implemented
diff --git a/distributed/tests/test_nanny.py b/distributed/tests/test_nanny.py
index 03be3172..4b3ff3ee 100644
--- a/distributed/tests/test_nanny.py
+++ b/distributed/tests/test_nanny.py
@@ -706,28 +706,24 @@ async def test_malloc_trim_threshold(c, s, a):
     using mimalloc by default. If it does, a thorough benchmarking exercise is needed.
     """
     da = pytest.importorskip("dask.array")
-
-    a = da.random.random(
-        2**29 // 8,  # 0.5 GiB,
-        chunks=160 * 2**10 // 8,  # 160 kiB
-    ).persist()
-    await wait(a)
+    arr = da.random.random(2**29 // 8, chunks="512 kiB")  # 0.5 GiB
+    arr = arr.persist()
+    await wait(arr)
     # Wait for heartbeat
-    while s.memory.process < 2**29:
-        await asyncio.sleep(0.01)
-    del a
+    await async_poll_for(lambda: s.memory.process > 2**29, timeout=5)
+    del arr
 
     # This is the delicate bit, as it relies on
     # 1. PyMem_Free() to be quick to invoke glibc free() when memory becomes available
     # 2. glibc free() to be quick to invoke the kernel's sbrk() when the same happens
     #
     # At the moment of writing, the readings are:
-    # - 122 MiB after starting a new worker
-    # - 139 MiB after computing a trivial dask.array collection
-    # - 185 MiB at the end of this test, with MALLOC_TRIM_THRESHOLD_=65536
-    # - 698 MiB at the end of this test, without MALLOC_TRIM_THRESHOLD_
-    while s.memory.process > 250 * 2**20:
-        await asyncio.sleep(0.01)
+    # - 132        MiB after starting a new worker
+    # - 670~725[1] MiB after arr has been fully computed
+    # - 156~210[1] MiB at the end of this test, with MALLOC_TRIM_THRESHOLD_=65536
+    # - 620~670[1] MiB at the end of this test, without MALLOC_TRIM_THRESHOLD_
+    # [1] depends on distributed.scheduler.worker-saturation
+    await async_poll_for(lambda: s.memory.process < 300 * 2**20, timeout=5)
 
 
 @gen_cluster(client=True, nthreads=[])
diff --git a/distributed/tests/test_worker_metrics.py b/distributed/tests/test_worker_metrics.py
index 881ee21f..2eb7c44f 100644
--- a/distributed/tests/test_worker_metrics.py
+++ b/distributed/tests/test_worker_metrics.py
@@ -1,9 +1,8 @@
 from __future__ import annotations
 
 import asyncio
-import secrets
 import time
-from collections.abc import Hashable
+from collections.abc import Collection, Hashable
 from concurrent.futures import ProcessPoolExecutor
 from time import sleep
 
@@ -11,8 +10,8 @@ import pytest
 
 import dask
 
+import distributed
 from distributed import Event, Worker, wait
-from distributed.comm.utils import OFFLOAD_THRESHOLD
 from distributed.compatibility import WINDOWS
 from distributed.metrics import context_meter, meter
 from distributed.utils_test import (
@@ -25,14 +24,16 @@ from distributed.utils_test import (
 )
 
 
-def get_digests(w: Worker, allow: str | None = None) -> dict[Hashable, float]:
+def get_digests(w: Worker, allow: str | Collection[str] = ()) -> dict[Hashable, float]:
     # import pprint; pprint.pprint(dict(w.digests_total))
+    if isinstance(allow, str):
+        allow = (allow,)
     digests = {
         k: v
         for k, v in w.digests_total.items()
         if k
         not in {"latency", "tick-duration", "transfer-bandwidth", "transfer-duration"}
-        and (allow is None or allow in k)
+        and (any(a in k for a in allow) or not allow)
     }
     assert all(v >= 0 for v in digests.values()), digests
     return digests
@@ -80,6 +81,7 @@ async def test_task_lifecycle(c, s, a, b):
         # -> Run in thread
         ("execute", "z", "thread-cpu", "seconds"),
         ("execute", "z", "thread-noncpu", "seconds"),
+        ("execute", "z", "executor", "seconds"),
         # Spill output; added by _transition_to_memory
         ("execute", "z", "serialize", "seconds"),
         ("execute", "z", "compress", "seconds"),
@@ -150,15 +152,23 @@ async def test_run_spec_deserialization(c, s, a):
 
 
 @gen_cluster(client=True)
-async def test_offload(c, s, a, b):
+async def test_offload(c, s, a, b, monkeypatch):
     """Test that functions wrapped by offload() are metered"""
-    nbytes = int(OFFLOAD_THRESHOLD * 1.1)
-    x = c.submit(secrets.token_bytes, nbytes, key="x", workers=[a.address])
+    monkeypatch.setattr(distributed.comm.utils, "OFFLOAD_THRESHOLD", 1)
+    monkeypatch.setattr(distributed.worker, "OFFLOAD_THRESHOLD", 1)
+
+    x = c.submit(inc, 1, key="x", workers=[a.address])
     y = c.submit(lambda x: None, x, key="y", workers=[b.address])
     await y
 
-    assert 0 < a.digests_total["get-data", "serialize", "seconds"] < 1
-    assert 0 < b.digests_total["gather-dep", "deserialize", "seconds"] < 1
+    assert list(get_digests(b, {"offload", "serialize", "deserialize"})) == [
+        ("gather-dep", "offload", "seconds"),
+        ("gather-dep", "deserialize", "seconds"),
+        ("execute", "y", "offload", "seconds"),
+        ("execute", "y", "deserialize", "seconds"),
+        ("get-data", "offload", "seconds"),
+        ("get-data", "serialize", "seconds"),
+    ]
 
 
 @gen_cluster(client=True, nthreads=[("", 1)])
@@ -315,11 +325,7 @@ async def test_memory_monitor(c, s, a):
     x = c.submit(inc, 1, key="x")
     await async_poll_for(lambda: a.data.disk, timeout=5)
 
-    assert list(get_digests(a)) == [
-        ("execute", "x", "deserialize", "seconds"),
-        ("execute", "x", "thread-cpu", "seconds"),
-        ("execute", "x", "thread-noncpu", "seconds"),
-        ("execute", "x", "other", "seconds"),
+    assert list(get_digests(a, "memory-monitor")) == [
         ("memory-monitor", "serialize", "seconds"),
         ("memory-monitor", "compress", "seconds"),
         ("memory-monitor", "disk-write", "seconds"),
@@ -344,11 +350,13 @@ async def test_user_metrics_sync(c, s, a):
         ("execute", "x", "I/O", "seconds"),
         ("execute", "x", "thread-cpu", "seconds"),
         ("execute", "x", "thread-noncpu", "seconds"),
+        ("execute", "x", "executor", "seconds"),
         ("execute", "x", "other", "seconds"),
     ]
     assert get_digests(a)["execute", "x", "I/O", "seconds"] == 5
     assert get_digests(a)["execute", "x", "thread-cpu", "seconds"] == 0
     assert get_digests(a)["execute", "x", "thread-noncpu", "seconds"] == 0
+    assert get_digests(a)["execute", "x", "executor", "seconds"] == 0
     assert get_digests(a)["execute", "x", "other", "seconds"] == 0