code: def memray_scheduler(
      directory: str | pathlib.Path = "memray-profiles",
      report_args: Sequence[str]
      | Literal[False] = ("flamegraph", "--temporal", "--leaks"),
      **memray_kwargs: Any,
  ) -> Iterator[None]:
      """Generate a Memray profile on the Scheduler and download the generated report.
  
      Example::
  
          with memray_scheduler():
              client.submit(my_function).result()
  
          # Or even while the computation is already running
  
          fut = client.submit(my_function)
  
          with memray_scheduler():
              time.sleep(10)
  
          fut.result()
  
      Parameters
      ----------
      directory : str
          The directory to save the reports to.
      report_args : tuple[str]
          Particularly for native_traces=True, the reports have to be
          generated on the same host using the same Python interpreter as the
          profile was generated. Otherwise, native traces will yield unusable
          results. Therefore, we're generating the reports on the Scheduler and
          download them afterwards. You can modify the report generation by
          providing additional arguments and we will generate the reports as::
  
              memray *report_args -f <filename> -o <filename>.html
  
          If the raw data should be fetched instead of the report, set this to
          False.
      **memray_kwargs
          Keyword arguments to be passed to memray.Tracker, e.g.
          {"native_traces": True}
      """
      directory = pathlib.Path(directory)
      client = get_client()
      filename = uuid.uuid4().hex
      assert client.run_on_scheduler(
          _start_memray, filename=filename, **memray_kwargs
      )
      # Sleep for a brief moment such that we get
      # a clear profiling signal when everything starts
      time.sleep(0.1)
      yield
      directory.mkdir(exist_ok=True)
  
      client = get_client()
  
      profile = client.run_on_scheduler(
          _fetch_memray_profile,
          filename=filename,
          report_args=report_args,
      )
      path = directory / "scheduler"
      if report_args:
          suffix = ".html"
      else:
          suffix = ".memray"
      with open(str(path) + suffix, "wb") as fd:
          fd.write(profile)
  
file_path: distributed/distributed/diagnostics/memray.py
imports: from __future__ import annotations
  import contextlib
  import logging
  import pathlib
  import subprocess
  import time
  import uuid
  from collections.abc import Iterator, Sequence
  from typing import Any, Literal
  from urllib.parse import quote
  from toolz.itertoolz import partition
  from distributed import get_client
  from distributed.worker import Worker
label_desc: Use image node_id 4 for calling a global function with example usage: memray_scheduler(directory, report_args) without return types