code: def span(*tags: str) -> Iterator[str]:
      """Tag group of tasks to be part of a certain group, called a span.
  
      This context manager can be nested, thus creating sub-spans. If you close and
      re-open a span context manager with the same tag, you'll end up with two separate
      spans.
  
      Every cluster defines a global "default" span when no span has been defined by the
      client; the default span is automatically closed and reopened when all tasks
      associated to it have been completed; in other words the cluster is idle save for
      tasks that are explicitly annotated by a span. Note that, in some edge cases, you
      may end up with overlapping default spans, e.g. if a worker crashes and all unique
      tasks that were in memory on it need to be recomputed.
  
      You may capture the ID of a span on the client to match it with the
      :class:`~distributed.spans.Span` objects the scheduler:
  
      >>> client = Client()
      >>> with span("my workflow") as span_id:
      ...     client.submit(lambda: "Hello world!").result()
      >>> client.cluster.scheduler.extensions["spans"].spans[span_id]
      Span<name=('my workflow',), id=5dc9b908-116b-49a5-b0d7-5a681f49a111>
  
      Notes
      -----
      You may retrieve the current span with ``dask.get_annotations().get("span")``.
      You can do so in the client code as well as from inside a task.
      """
      if not tags:
          raise ValueError("Must specify at least one span tag")
  
      annotation = dask.get_annotations().get("span")
      prev_tags = annotation["name"] if annotation else ()
      # You must specify the full history of IDs, not just the parent, because
      # otherwise you would not be able to uniquely identify grandparents when
      # they have no tasks of their own.
      prev_ids = annotation["ids"] if annotation else ()
      ids = tuple(str(uuid.uuid4()) for _ in tags)
      with dask.annotate(
          span={"name": prev_tags + tags, "ids": prev_ids + ids}
      ):
          yield ids[-1]
  
imports: from __future__ import annotations
  import uuid
  import weakref
  from collections import defaultdict
  from collections.abc import Hashable, Iterable, Iterator
  from contextlib import contextmanager
  from itertools import islice
  from typing import TYPE_CHECKING, Any
  import dask.config
  from dask.typing import Key
  from distributed.collections import sum_mappings
  from distributed.itertools import ffill
  from distributed.metrics import time
label_desc: Use image node_id 1 for calling a global function with example usage: span() without return types