code: def __init__(
      self,
      maxlen: int | None | NoDefault = no_default,
      monitor_disk_io: bool | None = None,
      monitor_host_cpu: bool | None = None,
      monitor_gil_contention: bool | None = None,
  ):
      self.proc = psutil.Process()
      self.count = 0
  
      if maxlen is no_default:
          maxlen = dask.config.get(
              "distributed.admin.system-monitor.log-length"
          )
      if isinstance(maxlen, int):
          maxlen = max(1, maxlen)
      elif maxlen is not None:  # pragma: nocover
          raise TypeError(f"maxlen must be int or None; got {maxlen!r}")
      self.maxlen = maxlen
  
      self.last_time = monotonic()
  
      self.quantities = {
          "cpu": deque(maxlen=maxlen),
          "memory": deque(maxlen=maxlen),
          "time": deque(maxlen=maxlen),
      }
  
      try:
          self._last_net_io_counters = psutil.net_io_counters()
      except Exception:
          # FIXME is this possible?
          self.monitor_net_io = False  # pragma: nocover
      else:
          self.monitor_net_io = True
          self.quantities["host_net_io.read_bps"] = deque(maxlen=maxlen)
          self.quantities["host_net_io.write_bps"] = deque(
              maxlen=maxlen
          )
  
      if monitor_disk_io is None:
          monitor_disk_io = dask.config.get(
              "distributed.admin.system-monitor.disk"
          )
      if monitor_disk_io:
          try:
              disk_ioc = psutil.disk_io_counters()
          except Exception:
              # FIXME occurs when psutil version doesn't have handling for given platform / kernel;
              # should we explicitly error in this case?
              monitor_disk_io = False  # pragma: nocover
          else:
              if disk_ioc is None:  # pragma: nocover
                  # diskless machine
                  monitor_disk_io = False
              else:
                  self._last_disk_io_counters = disk_ioc
                  self.quantities["host_disk_io.read_bps"] = deque(
                      maxlen=maxlen
                  )
                  self.quantities["host_disk_io.write_bps"] = deque(
                      maxlen=maxlen
                  )
      self.monitor_disk_io = monitor_disk_io
  
      if monitor_host_cpu is None:
          monitor_host_cpu = dask.config.get(
              "distributed.admin.system-monitor.host-cpu"
          )
      self.monitor_host_cpu = monitor_host_cpu
      if monitor_host_cpu:
          self._last_host_cpu_counters = hostcpu_c = psutil.cpu_times()
          # This is a namedtuple whose fields change based on OS and kernel version
          for k in hostcpu_c._fields:
              self.quantities["host_cpu." + k] = deque(maxlen=maxlen)
  
      if monitor_gil_contention is None:
          monitor_gil_contention = dask.config.get(
              "distributed.admin.system-monitor.gil.enabled"
          )
      self.monitor_gil_contention = monitor_gil_contention
      if self.monitor_gil_contention:
          try:
              from gilknocker import KnockKnock
          except ImportError:
              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",
              )
              interval = (
                  parse_timedelta(raw_interval, default="us") * 1e6
              )
  
              self._gilknocker = KnockKnock(
                  polling_interval_micros=int(interval)
              )
              self._gilknocker.start()
  
      if not WINDOWS:
          self.quantities["num_fds"] = deque(maxlen=maxlen)
  
      if nvml.device_get_count() > 0:
          gpu_extra = nvml.one_time()
          self.gpu_name = gpu_extra["name"]
          self.gpu_memory_total = gpu_extra["memory-total"]
          self.quantities["gpu_utilization"] = deque(maxlen=maxlen)
          self.quantities["gpu_memory_used"] = deque(maxlen=maxlen)
      else:
          self.gpu_name = None
          self.gpu_memory_total = -1
  
      self.update()
  
file_path: distributed/distributed/system_monitor.py
imports: from __future__ import annotations
  import sys
  from collections import deque
  from typing import Any
  import psutil
  import dask.config
  from dask.typing import NoDefault, no_default
  from dask.utils import parse_timedelta
  from distributed.compatibility import WINDOWS
  from distributed.diagnostics import nvml
  from distributed.metrics import monotonic, time
label_desc: Use image node_id 1 to create a new SystemMonitor object  with example: obj = SystemMonitor(maxlen, monitor_disk_io, monitor_host_cpu, monitor_gil_contention)