jcrist: Ping @pitrou for review.
diff --git a/dask/diagnostics/profile.py b/dask/diagnostics/profile.py
index 48e68a565..ec855173e 100644
--- a/dask/diagnostics/profile.py
+++ b/dask/diagnostics/profile.py
@@ -140,29 +140,35 @@ class ResourceProfiler(Callback):
     data will only be collected while a dask scheduler is active.
     """
     def __init__(self, dt=1):
-        self._tracker = _Tracker(dt)
-        self._tracker.start()
-        self.results = []
+        self._dt = dt
         self._entered = False
+        self._tracker = None
+        self.results = []
+
+    def _is_running(self):
+        return self._tracker is not None and self._tracker.is_alive()
 
     def _start_collect(self):
-        assert self._tracker.is_alive(), "Resource tracker is shutdown"
+        if not self._is_running():
+            self._tracker = _Tracker(self._dt)
+            self._tracker.start()
         self._tracker.parent_conn.send('collect')
 
     def _stop_collect(self):
-        if self._tracker.is_alive():
+        if self._is_running():
             self._tracker.parent_conn.send('send_data')
             self.results.extend(starmap(ResourceData, self._tracker.parent_conn.recv()))
 
     def __enter__(self):
-        self.clear()
         self._entered = True
+        self.clear()
         self._start_collect()
         return super(ResourceProfiler, self).__enter__()
 
     def __exit__(self, *args):
         self._entered = False
         self._stop_collect()
+        self.close()
         super(ResourceProfiler, self).__exit__(*args)
 
     def _start(self, dsk):
@@ -174,7 +180,9 @@ class ResourceProfiler(Callback):
 
     def close(self):
         """Shutdown the resource tracker process"""
-        self._tracker.shutdown()
+        if self._is_running():
+            self._tracker.shutdown()
+            self._tracker = None
 
     __del__ = close
 

ResourceProfiler keeps running after `__exit__`
If you write something like:
```
    with ResourceProfiler(dt=0.01) as rprof:
        # ...
```

then the profiler process keeps running at the end of the `with` block. You have to call `rprof.close()` as well, which is unexpected.
(noticed in the distributed test suite)

diff --git a/dask/diagnostics/tests/test_profiler.py b/dask/diagnostics/tests/test_profiler.py
index 5c062640b..cf374d7a1 100644
--- a/dask/diagnostics/tests/test_profiler.py
+++ b/dask/diagnostics/tests/test_profiler.py
@@ -83,15 +83,20 @@ def test_resource_profiler():
     assert len(results) > 0
     assert all(isinstance(i, tuple) and len(i) == 3 for i in results)
 
+    # Tracker stopped on exit
+    assert not rprof._is_running()
+
     rprof.clear()
     assert rprof.results == []
 
+    # Close is idempotent
     rprof.close()
-    assert not rprof._tracker.is_alive()
+    assert not rprof._is_running()
 
-    with pytest.raises(AssertionError):
-        with rprof:
-            get(dsk, 'e')
+    # Restarts tracker if already closed
+    with rprof:
+        get(dsk2, 'c')
+    assert len(rprof.results) > 0
 
 
 @pytest.mark.skipif("not psutil")
@@ -114,7 +119,7 @@ def test_resource_profiler_multiple_gets():
     assert all(isinstance(i, tuple) and len(i) == 3 for i in results)
 
     rprof.close()
-    assert not rprof._tracker.is_alive()
+    assert not rprof._is_running()
 
 
 def test_cache_profiler():

pip install -e .[complete]
parse_log_pytest
requirements.txt
pytest moto blosc cachey graphviz pyarrow pandas_datareader cityhash flake8 mmh3 pytest-xdist xxhash backports.lzma mock
apt-get update apt-get install -y gcc
3.9
docs/requirements-docs.txt
pytest --no-header -rA --tb=line --color=no -p no:cacheprovider -W ignore::DeprecationWarning
alabaster==0.7.16
babel==2.17.0
backports.lzma==0.0.14
blosc==1.11.2
boto3==1.37.23
botocore==1.37.23
cachey==0.2.1
certifi==2025.1.31
cffi==1.17.1
charset-normalizer==3.4.1
cityhash==0.4.8
click==8.1.8
cloudpickle==3.1.1
cryptography==44.0.2
-e git+https://github.com/dask/dask.git@d0fac49699b46078b070d318d4c9a02612d0bd97#egg=dask
distributed==1.19.3
docutils==0.21.2
exceptiongroup==1.2.2
execnet==2.1.1
flake8==7.2.0
fsspec==2025.3.1
graphviz==0.20.3
HeapDict==1.0.1
idna==3.10
imagesize==1.4.1
importlib_metadata==8.6.1
iniconfig==2.1.0
Jinja2==3.1.6
jmespath==1.0.1
locket==1.0.0
lxml==5.3.1
MarkupSafe==3.0.2
mccabe==0.7.0
mmh3==5.1.0
mock==5.2.0
moto==5.1.2
msgpack==1.1.0
msgpack-python==0.5.6
numpy==2.0.2
numpydoc==0.6.0
packaging==24.2
pandas==2.2.3
pandas-datareader==0.10.0
partd==1.4.2
pluggy==1.5.0
psutil==7.0.0
pyarrow==19.0.1
pycodestyle==2.13.0
pycparser==2.22
pyflakes==3.3.1
Pygments==2.19.1
pytest==8.3.5
pytest-xdist==3.6.1
python-dateutil==2.9.0.post0
pytz==2025.2
PyYAML==6.0.2
requests==2.32.3
responses==0.25.7
s3transfer==0.11.4
six==1.17.0
snowballstemmer==2.2.0
sortedcontainers==2.4.0
Sphinx==7.4.7
sphinx-rtd-theme==3.0.2
sphinxcontrib-applehelp==2.0.0
sphinxcontrib-devhelp==2.0.0
sphinxcontrib-htmlhelp==2.1.0
sphinxcontrib-jquery==4.1
sphinxcontrib-jsmath==1.0.1
sphinxcontrib-qthelp==2.0.0
sphinxcontrib-serializinghtml==2.0.0
tblib==3.1.0
tomli==2.2.1
toolz==1.0.0
tornado==6.4.2
tzdata==2025.2
urllib3==1.26.20
Werkzeug==3.1.3
xmltodict==0.14.2
xxhash==3.5.0
zict==3.0.0
zipp==3.21.0

name: dask
channels:
  - defaults
  - https://repo.anaconda.com/pkgs/main
  - https://repo.anaconda.com/pkgs/r
  - conda-forge
dependencies:
  - _libgcc_mutex=0.1=main
  - _openmp_mutex=5.1=1_gnu
  - ca-certificates=2025.2.25=h06a4308_0
  - ld_impl_linux-64=2.40=h12ee557_0
  - libffi=3.4.4=h6a678d5_1
  - libgcc-ng=11.2.0=h1234567_1
  - libgomp=11.2.0=h1234567_1
  - libstdcxx-ng=11.2.0=h1234567_1
  - ncurses=6.4=h6a678d5_0
  - openssl=3.0.16=h5eee18b_0
  - pip=25.0=py39h06a4308_0
  - python=3.9.21=he870216_1
  - readline=8.2=h5eee18b_0
  - setuptools=75.8.0=py39h06a4308_0
  - sqlite=3.45.3=h5eee18b_0
  - tk=8.6.14=h39e8969_0
  - wheel=0.45.1=py39h06a4308_0
  - xz=5.6.4=h5eee18b_1
  - zlib=1.2.13=h5eee18b_1
  - pip:
      - alabaster==0.7.16
      - babel==2.17.0
      - backports-lzma==0.0.14
      - blosc==1.11.2
      - boto3==1.37.23
      - botocore==1.37.23
      - cachey==0.2.1
      - certifi==2025.1.31
      - cffi==1.17.1
      - charset-normalizer==3.4.1
      - cityhash==0.4.8
      - click==8.1.8
      - cloudpickle==3.1.1
      - cryptography==44.0.2
      - distributed==1.19.3
      - docutils==0.21.2
      - exceptiongroup==1.2.2
      - execnet==2.1.1
      - flake8==7.2.0
      - fsspec==2025.3.1
      - graphviz==0.20.3
      - heapdict==1.0.1
      - idna==3.10
      - imagesize==1.4.1
      - importlib-metadata==8.6.1
      - iniconfig==2.1.0
      - jinja2==3.1.6
      - jmespath==1.0.1
      - locket==1.0.0
      - lxml==5.3.1
      - markupsafe==3.0.2
      - mccabe==0.7.0
      - mmh3==5.1.0
      - mock==5.2.0
      - moto==5.1.2
      - msgpack==1.1.0
      - msgpack-python==0.5.6
      - numpy==2.0.2
      - numpydoc==0.6.0
      - packaging==24.2
      - pandas==2.2.3
      - pandas-datareader==0.10.0
      - partd==1.4.2
      - pluggy==1.5.0
      - psutil==7.0.0
      - pyarrow==19.0.1
      - pycodestyle==2.13.0
      - pycparser==2.22
      - pyflakes==3.3.1
      - pygments==2.19.1
      - pytest==8.3.5
      - pytest-xdist==3.6.1
      - python-dateutil==2.9.0.post0
      - pytz==2025.2
      - pyyaml==6.0.2
      - requests==2.32.3
      - responses==0.25.7
      - s3transfer==0.11.4
      - six==1.17.0
      - snowballstemmer==2.2.0
      - sortedcontainers==2.4.0
      - sphinx==7.4.7
      - sphinx-rtd-theme==3.0.2
      - sphinxcontrib-applehelp==2.0.0
      - sphinxcontrib-devhelp==2.0.0
      - sphinxcontrib-htmlhelp==2.1.0
      - sphinxcontrib-jquery==4.1
      - sphinxcontrib-jsmath==1.0.1
      - sphinxcontrib-qthelp==2.0.0
      - sphinxcontrib-serializinghtml==2.0.0
      - tblib==3.1.0
      - tomli==2.2.1
      - toolz==1.0.0
      - tornado==6.4.2
      - tzdata==2025.2
      - urllib3==1.26.20
      - werkzeug==3.1.3
      - xmltodict==0.14.2
      - xxhash==3.5.0
      - zict==3.0.0
      - zipp==3.21.0
prefix: /opt/conda/envs/dask

dask/diagnostics/tests/test_profiler.py::test_resource_profiler dask/diagnostics/tests/test_profiler.py::test_resource_profiler_multiple_gets
dask/diagnostics/tests/test_profiler.py::test_profiler dask/diagnostics/tests/test_profiler.py::test_profiler_works_under_error dask/diagnostics/tests/test_profiler.py::test_two_gets dask/diagnostics/tests/test_profiler.py::test_cache_profiler dask/diagnostics/tests/test_profiler.py::test_register[Profiler] dask/diagnostics/tests/test_profiler.py::test_register[<lambda>] dask/diagnostics/tests/test_profiler.py::test_register[CacheProfiler]
dask/diagnostics/profile.py
dask/diagnostics/profile.py