We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
psutil causes Nanny to crash
<!-- 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.
-->

**What happened**:

We are getting the following exceptions occasionally from our workers resulting the whole process to stall eventually:

```
FileNotFoundError: [Errno 2] No such file or directory: '/proc/12/statm'
Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/tornado/ioloop.py", line 905, in _run
    return self.callback()
  File "/usr/local/lib/python3.8/site-packages/distributed/system_monitor.py", line 121, in update
    read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (
AttributeError: 'NoneType' object has no attribute 'read_bytes'

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/psutil/_common.py", line 441, in wrapper
    ret = self._cache[fun]
AttributeError: 'Process' object has no attribute '_cache'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "/usr/local/lib/python3.8/site-packages/tornado/ioloop.py", line 905, in _run
    return self.callback()
  File "/usr/local/lib/python3.8/site-packages/distributed/worker_memory.py", line 322, in memory_monitor
    memory = proc.memory_info().rss
  File "/usr/local/lib/python3.8/site-packages/psutil/_common.py", line 444, in wrapper
    return fun(self)
  File "/usr/local/lib/python3.8/site-packages/psutil/__init__.py", line 1061, in memory_info
    return self._proc.memory_info()
  File "/usr/local/lib/python3.8/site-packages/psutil/_pslinux.py", line 1661, in wrapper
    return fun(self, *args, **kwargs)
  File "/usr/local/lib/python3.8/site-packages/psutil/_pslinux.py", line 1895, in memory_info
    with open_binary("%s/%s/statm" % (self._procfs_path, self.pid)) as f:
  File "/usr/local/lib/python3.8/site-packages/psutil/_common.py", line 711, in open_binary
    return open(fname, "rb", **kwargs)
```

If I understand the code correctly the following is happening:

`psutil.disk_io_counters()` returns a named tuple:

https://github.com/dask/distributed/blob/2022.04.0/distributed/system_monitor.py#L39

so that the internal `self._collect_disk_io_counters` is _not_ set to `False` but `psutil.disk_io_counters()` returns `None` instead of an expected named tuple when called within the update method:

https://github.com/dask/distributed/blob/2022.04.0/distributed/system_monitor.py#L115

later thus causing the Nanny to crash.

It seems to be an issue of `psutil` in the first place but I think the `SystemMonitor` could be more resilient if that happens.

**What you expected to happen**:

`SystemMonitor` should not raise an exception if `psutil.disk_io_counters()` returns `None`: https://github.com/dask/distributed/blob/034b4d4c5cfad2fe3b672425eb96b826d79d9415/distributed/system_monitor.py#L115

**Minimal Complete Verifiable Example**:

```python
# Put your MCVE code here
```

**Anything else we need to know?**:

Should I prepare a PR with the suggested changes?

**Environment**:

- Dask version: 2022.4.0
- Python version: 3.8
- Operating System: Debian GNU/Linux 10 (buster)
- Install method (conda, pip, source): pip

<!-- If you are reporting an issue such as scale stability, cluster deadlock.
Please provide a cluster dump state with this issue, by running client.dump_cluster_state()

https://distributed.dask.org/en/stable/api.html?highlight=dump_cluster_state#distributed.Client.dump_cluster_state

-->

<details>
<summary>Cluster Dump State:</summary>

</details>

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `distributed/system_monitor.py`
Content:
```
1 from collections import deque
2 
3 import psutil
4 
5 from distributed.compatibility import WINDOWS
6 from distributed.metrics import time
7 
8 try:
9     from distributed.diagnostics import nvml
10 except Exception:
11     nvml = None  # type: ignore
12 
13 
14 class SystemMonitor:
15     def __init__(self, n=10000):
16         self.proc = psutil.Process()
17 
18         self.time = deque(maxlen=n)
19         self.cpu = deque(maxlen=n)
20         self.memory = deque(maxlen=n)
21         self.count = 0
22 
23         self.quantities = {"cpu": self.cpu, "memory": self.memory, "time": self.time}
24 
25         try:
26             ioc = psutil.net_io_counters()
27         except Exception:
28             self._collect_net_io_counters = False
29         else:
30             self.last_time = time()
31             self.read_bytes = deque(maxlen=n)
32             self.write_bytes = deque(maxlen=n)
33             self.quantities["read_bytes"] = self.read_bytes
34             self.quantities["write_bytes"] = self.write_bytes
35             self._last_io_counters = ioc
36             self._collect_net_io_counters = True
37 
38         try:
39             disk_ioc = psutil.disk_io_counters()
40         except Exception:
41             self._collect_disk_io_counters = False
42         else:
43             if disk_ioc is None:  # diskless machine
44                 self._collect_disk_io_counters = False
45             else:
46                 self.last_time_disk = time()
47                 self.read_bytes_disk = deque(maxlen=n)
48                 self.write_bytes_disk = deque(maxlen=n)
49                 self.quantities["read_bytes_disk"] = self.read_bytes_disk
50                 self.quantities["write_bytes_disk"] = self.write_bytes_disk
51                 self._last_disk_io_counters = disk_ioc
52                 self._collect_disk_io_counters = True
53 
54         if not WINDOWS:
55             self.num_fds = deque(maxlen=n)
56             self.quantities["num_fds"] = self.num_fds
57 
58         if nvml.device_get_count() > 0:
59             gpu_extra = nvml.one_time()
60             self.gpu_name = gpu_extra["name"]
61             self.gpu_memory_total = gpu_extra["memory-total"]
62             self.gpu_utilization = deque(maxlen=n)
63             self.gpu_memory_used = deque(maxlen=n)
64             self.quantities["gpu_utilization"] = self.gpu_utilization
65             self.quantities["gpu_memory_used"] = self.gpu_memory_used
66 
67         self.update()
68 
69     def recent(self):
70         try:
71             return {k: v[-1] for k, v in self.quantities.items()}
72         except IndexError:
73             return {k: None for k, v in self.quantities.items()}
74 
75     def get_process_memory(self) -> int:
76         """Sample process memory, as reported by the OS.
77         This one-liner function exists so that it can be easily mocked in unit tests,
78         as the OS allocating and releasing memory is highly volatile and a constant
79         source of flakiness.
80         """
81         return self.proc.memory_info().rss
82 
83     def update(self):
84         with self.proc.oneshot():
85             cpu = self.proc.cpu_percent()
86             memory = self.get_process_memory()
87         now = time()
88 
89         self.cpu.append(cpu)
90         self.memory.append(memory)
91         self.time.append(now)
92         self.count += 1
93 
94         result = {"cpu": cpu, "memory": memory, "time": now, "count": self.count}
95 
96         if self._collect_net_io_counters:
97             try:
98                 ioc = psutil.net_io_counters()
99             except Exception:
100                 pass
101             else:
102                 last = self._last_io_counters
103                 duration = now - self.last_time
104                 read_bytes = (ioc.bytes_recv - last.bytes_recv) / (duration or 0.5)
105                 write_bytes = (ioc.bytes_sent - last.bytes_sent) / (duration or 0.5)
106                 self.last_time = now
107                 self._last_io_counters = ioc
108                 self.read_bytes.append(read_bytes)
109                 self.write_bytes.append(write_bytes)
110                 result["read_bytes"] = read_bytes
111                 result["write_bytes"] = write_bytes
112 
113         if self._collect_disk_io_counters:
114             try:
115                 disk_ioc = psutil.disk_io_counters()
116             except Exception:
117                 pass
118             else:
119                 last_disk = self._last_disk_io_counters
120                 duration_disk = now - self.last_time_disk
121                 read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (
122                     duration_disk or 0.5
123                 )
124                 write_bytes_disk = (disk_ioc.write_bytes - last_disk.write_bytes) / (
125                     duration_disk or 0.5
126                 )
127                 self.last_time_disk = now
128                 self._last_disk_io_counters = disk_ioc
129                 self.read_bytes_disk.append(read_bytes_disk)
130                 self.write_bytes_disk.append(write_bytes_disk)
131                 result["read_bytes_disk"] = read_bytes_disk
132                 result["write_bytes_disk"] = write_bytes_disk
133 
134         if not WINDOWS:
135             num_fds = self.proc.num_fds()
136             self.num_fds.append(num_fds)
137             result["num_fds"] = num_fds
138 
139         if nvml.device_get_count() > 0:
140             gpu_metrics = nvml.real_time()
141             self.gpu_utilization.append(gpu_metrics["utilization"])
142             self.gpu_memory_used.append(gpu_metrics["memory-used"])
143             result["gpu_utilization"] = gpu_metrics["utilization"]
144             result["gpu_memory_used"] = gpu_metrics["memory-used"]
145 
146         return result
147 
148     def __repr__(self):
149         return "<SystemMonitor: cpu: %d memory: %d MB fds: %d>" % (
150             self.cpu[-1],
151             self.memory[-1] / 1e6,
152             -1 if WINDOWS else self.num_fds[-1],
153         )
154 
155     def range_query(self, start):
156         if start == self.count:
157             return {k: [] for k in self.quantities}
158 
159         istart = start - (self.count - len(self.cpu))
160         istart = max(0, istart)
161 
162         seq = [i for i in range(istart, len(self.cpu))]
163 
164         d = {k: [v[i] for i in seq] for k, v in self.quantities.items()}
165         return d
166
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/distributed/system_monitor.py b/distributed/system_monitor.py
--- a/distributed/system_monitor.py
+++ b/distributed/system_monitor.py
@@ -116,20 +116,23 @@
             except Exception:
                 pass
             else:
-                last_disk = self._last_disk_io_counters
-                duration_disk = now - self.last_time_disk
-                read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (
-                    duration_disk or 0.5
-                )
-                write_bytes_disk = (disk_ioc.write_bytes - last_disk.write_bytes) / (
-                    duration_disk or 0.5
-                )
-                self.last_time_disk = now
-                self._last_disk_io_counters = disk_ioc
-                self.read_bytes_disk.append(read_bytes_disk)
-                self.write_bytes_disk.append(write_bytes_disk)
-                result["read_bytes_disk"] = read_bytes_disk
-                result["write_bytes_disk"] = write_bytes_disk
+                if disk_ioc is None:  # diskless machine
+                    self._collect_disk_io_counters = False
+                else:
+                    last_disk = self._last_disk_io_counters
+                    duration_disk = now - self.last_time_disk
+                    read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (
+                        duration_disk or 0.5
+                    )
+                    write_bytes_disk = (
+                        disk_ioc.write_bytes - last_disk.write_bytes
+                    ) / (duration_disk or 0.5)
+                    self.last_time_disk = now
+                    self._last_disk_io_counters = disk_ioc
+                    self.read_bytes_disk.append(read_bytes_disk)
+                    self.write_bytes_disk.append(write_bytes_disk)
+                    result["read_bytes_disk"] = read_bytes_disk
+                    result["write_bytes_disk"] = write_bytes_disk
 
         if not WINDOWS:
             num_fds = self.proc.num_fds()

{"golden_diff": "diff --git a/distributed/system_monitor.py b/distributed/system_monitor.py\n--- a/distributed/system_monitor.py\n+++ b/distributed/system_monitor.py\n@@ -116,20 +116,23 @@\n             except Exception:\n                 pass\n             else:\n-                last_disk = self._last_disk_io_counters\n-                duration_disk = now - self.last_time_disk\n-                read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (\n-                    duration_disk or 0.5\n-                )\n-                write_bytes_disk = (disk_ioc.write_bytes - last_disk.write_bytes) / (\n-                    duration_disk or 0.5\n-                )\n-                self.last_time_disk = now\n-                self._last_disk_io_counters = disk_ioc\n-                self.read_bytes_disk.append(read_bytes_disk)\n-                self.write_bytes_disk.append(write_bytes_disk)\n-                result[\"read_bytes_disk\"] = read_bytes_disk\n-                result[\"write_bytes_disk\"] = write_bytes_disk\n+                if disk_ioc is None:  # diskless machine\n+                    self._collect_disk_io_counters = False\n+                else:\n+                    last_disk = self._last_disk_io_counters\n+                    duration_disk = now - self.last_time_disk\n+                    read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (\n+                        duration_disk or 0.5\n+                    )\n+                    write_bytes_disk = (\n+                        disk_ioc.write_bytes - last_disk.write_bytes\n+                    ) / (duration_disk or 0.5)\n+                    self.last_time_disk = now\n+                    self._last_disk_io_counters = disk_ioc\n+                    self.read_bytes_disk.append(read_bytes_disk)\n+                    self.write_bytes_disk.append(write_bytes_disk)\n+                    result[\"read_bytes_disk\"] = read_bytes_disk\n+                    result[\"write_bytes_disk\"] = write_bytes_disk\n \n         if not WINDOWS:\n             num_fds = self.proc.num_fds()\n", "issue": "psutil causes Nanny to crash\n<!-- Please include a self-contained copy-pastable example that generates the issue if possible.\r\n\r\nPlease be concise with code posted. See guidelines below on how to provide a good bug report:\r\n\r\n- Craft Minimal Bug Reports http://matthewrocklin.com/blog/work/2018/02/28/minimal-bug-reports\r\n- Minimal Complete Verifiable Examples https://stackoverflow.com/help/mcve\r\n\r\nBug reports that follow these guidelines are easier to diagnose, and so are often handled much more quickly.\r\n-->\r\n\r\n**What happened**:\r\n\r\nWe are getting the following exceptions occasionally from our workers resulting the whole process to stall eventually:\r\n\r\n```\r\nFileNotFoundError: [Errno 2] No such file or directory: '/proc/12/statm'\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.8/site-packages/tornado/ioloop.py\", line 905, in _run\r\n    return self.callback()\r\n  File \"/usr/local/lib/python3.8/site-packages/distributed/system_monitor.py\", line 121, in update\r\n    read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (\r\nAttributeError: 'NoneType' object has no attribute 'read_bytes'\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.8/site-packages/psutil/_common.py\", line 441, in wrapper\r\n    ret = self._cache[fun]\r\nAttributeError: 'Process' object has no attribute '_cache'\r\n\r\nDuring handling of the above exception, another exception occurred:\r\n\r\nTraceback (most recent call last):\r\n  File \"/usr/local/lib/python3.8/site-packages/tornado/ioloop.py\", line 905, in _run\r\n    return self.callback()\r\n  File \"/usr/local/lib/python3.8/site-packages/distributed/worker_memory.py\", line 322, in memory_monitor\r\n    memory = proc.memory_info().rss\r\n  File \"/usr/local/lib/python3.8/site-packages/psutil/_common.py\", line 444, in wrapper\r\n    return fun(self)\r\n  File \"/usr/local/lib/python3.8/site-packages/psutil/__init__.py\", line 1061, in memory_info\r\n    return self._proc.memory_info()\r\n  File \"/usr/local/lib/python3.8/site-packages/psutil/_pslinux.py\", line 1661, in wrapper\r\n    return fun(self, *args, **kwargs)\r\n  File \"/usr/local/lib/python3.8/site-packages/psutil/_pslinux.py\", line 1895, in memory_info\r\n    with open_binary(\"%s/%s/statm\" % (self._procfs_path, self.pid)) as f:\r\n  File \"/usr/local/lib/python3.8/site-packages/psutil/_common.py\", line 711, in open_binary\r\n    return open(fname, \"rb\", **kwargs)\r\n```\r\n\r\nIf I understand the code correctly the following is happening:\r\n\r\n`psutil.disk_io_counters()` returns a named tuple:\r\n\r\nhttps://github.com/dask/distributed/blob/2022.04.0/distributed/system_monitor.py#L39\r\n\r\nso that the internal `self._collect_disk_io_counters` is _not_ set to `False` but `psutil.disk_io_counters()` returns `None` instead of an expected named tuple when called within the update method:\r\n\r\nhttps://github.com/dask/distributed/blob/2022.04.0/distributed/system_monitor.py#L115\r\n\r\nlater thus causing the Nanny to crash.\r\n\r\nIt seems to be an issue of `psutil` in the first place but I think the `SystemMonitor` could be more resilient if that happens.\r\n\r\n**What you expected to happen**:\r\n\r\n`SystemMonitor` should not raise an exception if `psutil.disk_io_counters()` returns `None`: https://github.com/dask/distributed/blob/034b4d4c5cfad2fe3b672425eb96b826d79d9415/distributed/system_monitor.py#L115\r\n\r\n**Minimal Complete Verifiable Example**:\r\n\r\n```python\r\n# Put your MCVE code here\r\n```\r\n\r\n**Anything else we need to know?**:\r\n\r\nShould I prepare a PR with the suggested changes?\r\n\r\n**Environment**:\r\n\r\n- Dask version: 2022.4.0\r\n- Python version: 3.8\r\n- Operating System: Debian GNU/Linux 10 (buster)\r\n- Install method (conda, pip, source): pip\r\n\r\n<!-- If you are reporting an issue such as scale stability, cluster deadlock.\r\nPlease provide a cluster dump state with this issue, by running client.dump_cluster_state()\r\n\r\nhttps://distributed.dask.org/en/stable/api.html?highlight=dump_cluster_state#distributed.Client.dump_cluster_state\r\n\r\n-->\r\n\r\n<details>\r\n<summary>Cluster Dump State:</summary>\r\n\r\n</details>\n", "before_files": [{"content": "from collections import deque\n\nimport psutil\n\nfrom distributed.compatibility import WINDOWS\nfrom distributed.metrics import time\n\ntry:\n    from distributed.diagnostics import nvml\nexcept Exception:\n    nvml = None  # type: ignore\n\n\nclass SystemMonitor:\n    def __init__(self, n=10000):\n        self.proc = psutil.Process()\n\n        self.time = deque(maxlen=n)\n        self.cpu = deque(maxlen=n)\n        self.memory = deque(maxlen=n)\n        self.count = 0\n\n        self.quantities = {\"cpu\": self.cpu, \"memory\": self.memory, \"time\": self.time}\n\n        try:\n            ioc = psutil.net_io_counters()\n        except Exception:\n            self._collect_net_io_counters = False\n        else:\n            self.last_time = time()\n            self.read_bytes = deque(maxlen=n)\n            self.write_bytes = deque(maxlen=n)\n            self.quantities[\"read_bytes\"] = self.read_bytes\n            self.quantities[\"write_bytes\"] = self.write_bytes\n            self._last_io_counters = ioc\n            self._collect_net_io_counters = True\n\n        try:\n            disk_ioc = psutil.disk_io_counters()\n        except Exception:\n            self._collect_disk_io_counters = False\n        else:\n            if disk_ioc is None:  # diskless machine\n                self._collect_disk_io_counters = False\n            else:\n                self.last_time_disk = time()\n                self.read_bytes_disk = deque(maxlen=n)\n                self.write_bytes_disk = deque(maxlen=n)\n                self.quantities[\"read_bytes_disk\"] = self.read_bytes_disk\n                self.quantities[\"write_bytes_disk\"] = self.write_bytes_disk\n                self._last_disk_io_counters = disk_ioc\n                self._collect_disk_io_counters = True\n\n        if not WINDOWS:\n            self.num_fds = deque(maxlen=n)\n            self.quantities[\"num_fds\"] = self.num_fds\n\n        if nvml.device_get_count() > 0:\n            gpu_extra = nvml.one_time()\n            self.gpu_name = gpu_extra[\"name\"]\n            self.gpu_memory_total = gpu_extra[\"memory-total\"]\n            self.gpu_utilization = deque(maxlen=n)\n            self.gpu_memory_used = deque(maxlen=n)\n            self.quantities[\"gpu_utilization\"] = self.gpu_utilization\n            self.quantities[\"gpu_memory_used\"] = self.gpu_memory_used\n\n        self.update()\n\n    def recent(self):\n        try:\n            return {k: v[-1] for k, v in self.quantities.items()}\n        except IndexError:\n            return {k: None for k, v in self.quantities.items()}\n\n    def get_process_memory(self) -> int:\n        \"\"\"Sample process memory, as reported by the OS.\n        This one-liner function exists so that it can be easily mocked in unit tests,\n        as the OS allocating and releasing memory is highly volatile and a constant\n        source of flakiness.\n        \"\"\"\n        return self.proc.memory_info().rss\n\n    def update(self):\n        with self.proc.oneshot():\n            cpu = self.proc.cpu_percent()\n            memory = self.get_process_memory()\n        now = time()\n\n        self.cpu.append(cpu)\n        self.memory.append(memory)\n        self.time.append(now)\n        self.count += 1\n\n        result = {\"cpu\": cpu, \"memory\": memory, \"time\": now, \"count\": self.count}\n\n        if self._collect_net_io_counters:\n            try:\n                ioc = psutil.net_io_counters()\n            except Exception:\n                pass\n            else:\n                last = self._last_io_counters\n                duration = now - self.last_time\n                read_bytes = (ioc.bytes_recv - last.bytes_recv) / (duration or 0.5)\n                write_bytes = (ioc.bytes_sent - last.bytes_sent) / (duration or 0.5)\n                self.last_time = now\n                self._last_io_counters = ioc\n                self.read_bytes.append(read_bytes)\n                self.write_bytes.append(write_bytes)\n                result[\"read_bytes\"] = read_bytes\n                result[\"write_bytes\"] = write_bytes\n\n        if self._collect_disk_io_counters:\n            try:\n                disk_ioc = psutil.disk_io_counters()\n            except Exception:\n                pass\n            else:\n                last_disk = self._last_disk_io_counters\n                duration_disk = now - self.last_time_disk\n                read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (\n                    duration_disk or 0.5\n                )\n                write_bytes_disk = (disk_ioc.write_bytes - last_disk.write_bytes) / (\n                    duration_disk or 0.5\n                )\n                self.last_time_disk = now\n                self._last_disk_io_counters = disk_ioc\n                self.read_bytes_disk.append(read_bytes_disk)\n                self.write_bytes_disk.append(write_bytes_disk)\n                result[\"read_bytes_disk\"] = read_bytes_disk\n                result[\"write_bytes_disk\"] = write_bytes_disk\n\n        if not WINDOWS:\n            num_fds = self.proc.num_fds()\n            self.num_fds.append(num_fds)\n            result[\"num_fds\"] = num_fds\n\n        if nvml.device_get_count() > 0:\n            gpu_metrics = nvml.real_time()\n            self.gpu_utilization.append(gpu_metrics[\"utilization\"])\n            self.gpu_memory_used.append(gpu_metrics[\"memory-used\"])\n            result[\"gpu_utilization\"] = gpu_metrics[\"utilization\"]\n            result[\"gpu_memory_used\"] = gpu_metrics[\"memory-used\"]\n\n        return result\n\n    def __repr__(self):\n        return \"<SystemMonitor: cpu: %d memory: %d MB fds: %d>\" % (\n            self.cpu[-1],\n            self.memory[-1] / 1e6,\n            -1 if WINDOWS else self.num_fds[-1],\n        )\n\n    def range_query(self, start):\n        if start == self.count:\n            return {k: [] for k in self.quantities}\n\n        istart = start - (self.count - len(self.cpu))\n        istart = max(0, istart)\n\n        seq = [i for i in range(istart, len(self.cpu))]\n\n        d = {k: [v[i] for i in seq] for k, v in self.quantities.items()}\n        return d\n", "path": "distributed/system_monitor.py"}], "after_files": [{"content": "from collections import deque\n\nimport psutil\n\nfrom distributed.compatibility import WINDOWS\nfrom distributed.metrics import time\n\ntry:\n    from distributed.diagnostics import nvml\nexcept Exception:\n    nvml = None  # type: ignore\n\n\nclass SystemMonitor:\n    def __init__(self, n=10000):\n        self.proc = psutil.Process()\n\n        self.time = deque(maxlen=n)\n        self.cpu = deque(maxlen=n)\n        self.memory = deque(maxlen=n)\n        self.count = 0\n\n        self.quantities = {\"cpu\": self.cpu, \"memory\": self.memory, \"time\": self.time}\n\n        try:\n            ioc = psutil.net_io_counters()\n        except Exception:\n            self._collect_net_io_counters = False\n        else:\n            self.last_time = time()\n            self.read_bytes = deque(maxlen=n)\n            self.write_bytes = deque(maxlen=n)\n            self.quantities[\"read_bytes\"] = self.read_bytes\n            self.quantities[\"write_bytes\"] = self.write_bytes\n            self._last_io_counters = ioc\n            self._collect_net_io_counters = True\n\n        try:\n            disk_ioc = psutil.disk_io_counters()\n        except Exception:\n            self._collect_disk_io_counters = False\n        else:\n            if disk_ioc is None:  # diskless machine\n                self._collect_disk_io_counters = False\n            else:\n                self.last_time_disk = time()\n                self.read_bytes_disk = deque(maxlen=n)\n                self.write_bytes_disk = deque(maxlen=n)\n                self.quantities[\"read_bytes_disk\"] = self.read_bytes_disk\n                self.quantities[\"write_bytes_disk\"] = self.write_bytes_disk\n                self._last_disk_io_counters = disk_ioc\n                self._collect_disk_io_counters = True\n\n        if not WINDOWS:\n            self.num_fds = deque(maxlen=n)\n            self.quantities[\"num_fds\"] = self.num_fds\n\n        if nvml.device_get_count() > 0:\n            gpu_extra = nvml.one_time()\n            self.gpu_name = gpu_extra[\"name\"]\n            self.gpu_memory_total = gpu_extra[\"memory-total\"]\n            self.gpu_utilization = deque(maxlen=n)\n            self.gpu_memory_used = deque(maxlen=n)\n            self.quantities[\"gpu_utilization\"] = self.gpu_utilization\n            self.quantities[\"gpu_memory_used\"] = self.gpu_memory_used\n\n        self.update()\n\n    def recent(self):\n        try:\n            return {k: v[-1] for k, v in self.quantities.items()}\n        except IndexError:\n            return {k: None for k, v in self.quantities.items()}\n\n    def get_process_memory(self) -> int:\n        \"\"\"Sample process memory, as reported by the OS.\n        This one-liner function exists so that it can be easily mocked in unit tests,\n        as the OS allocating and releasing memory is highly volatile and a constant\n        source of flakiness.\n        \"\"\"\n        return self.proc.memory_info().rss\n\n    def update(self):\n        with self.proc.oneshot():\n            cpu = self.proc.cpu_percent()\n            memory = self.get_process_memory()\n        now = time()\n\n        self.cpu.append(cpu)\n        self.memory.append(memory)\n        self.time.append(now)\n        self.count += 1\n\n        result = {\"cpu\": cpu, \"memory\": memory, \"time\": now, \"count\": self.count}\n\n        if self._collect_net_io_counters:\n            try:\n                ioc = psutil.net_io_counters()\n            except Exception:\n                pass\n            else:\n                last = self._last_io_counters\n                duration = now - self.last_time\n                read_bytes = (ioc.bytes_recv - last.bytes_recv) / (duration or 0.5)\n                write_bytes = (ioc.bytes_sent - last.bytes_sent) / (duration or 0.5)\n                self.last_time = now\n                self._last_io_counters = ioc\n                self.read_bytes.append(read_bytes)\n                self.write_bytes.append(write_bytes)\n                result[\"read_bytes\"] = read_bytes\n                result[\"write_bytes\"] = write_bytes\n\n        if self._collect_disk_io_counters:\n            try:\n                disk_ioc = psutil.disk_io_counters()\n            except Exception:\n                pass\n            else:\n                if disk_ioc is None:  # diskless machine\n                    self._collect_disk_io_counters = False\n                else:\n                    last_disk = self._last_disk_io_counters\n                    duration_disk = now - self.last_time_disk\n                    read_bytes_disk = (disk_ioc.read_bytes - last_disk.read_bytes) / (\n                        duration_disk or 0.5\n                    )\n                    write_bytes_disk = (\n                        disk_ioc.write_bytes - last_disk.write_bytes\n                    ) / (duration_disk or 0.5)\n                    self.last_time_disk = now\n                    self._last_disk_io_counters = disk_ioc\n                    self.read_bytes_disk.append(read_bytes_disk)\n                    self.write_bytes_disk.append(write_bytes_disk)\n                    result[\"read_bytes_disk\"] = read_bytes_disk\n                    result[\"write_bytes_disk\"] = write_bytes_disk\n\n        if not WINDOWS:\n            num_fds = self.proc.num_fds()\n            self.num_fds.append(num_fds)\n            result[\"num_fds\"] = num_fds\n\n        if nvml.device_get_count() > 0:\n            gpu_metrics = nvml.real_time()\n            self.gpu_utilization.append(gpu_metrics[\"utilization\"])\n            self.gpu_memory_used.append(gpu_metrics[\"memory-used\"])\n            result[\"gpu_utilization\"] = gpu_metrics[\"utilization\"]\n            result[\"gpu_memory_used\"] = gpu_metrics[\"memory-used\"]\n\n        return result\n\n    def __repr__(self):\n        return \"<SystemMonitor: cpu: %d memory: %d MB fds: %d>\" % (\n            self.cpu[-1],\n            self.memory[-1] / 1e6,\n            -1 if WINDOWS else self.num_fds[-1],\n        )\n\n    def range_query(self, start):\n        if start == self.count:\n            return {k: [] for k in self.quantities}\n\n        istart = start - (self.count - len(self.cpu))\n        istart = max(0, istart)\n\n        seq = [i for i in range(istart, len(self.cpu))]\n\n        d = {k: [v[i] for i in seq] for k, v in self.quantities.items()}\n        return d\n", "path": "distributed/system_monitor.py"}]}