code: def dependent_keys(tasks, complete=False):
      """
      All keys that need to compute for these keys to finish.
  
      If *complete* is false, omit tasks that are busy processing or
      have finished executing.
      """
      out = set()
      errors = set()
      stack = list(tasks)
      while stack:
          ts = stack.pop()
          key = ts.key
          if key in out:
              continue
          if not complete and ts.who_has:
              continue
          if ts.exception is not None:
              errors.add(key)
              if not complete:
                  continue
  
          out.add(key)
          stack.extend(ts.dependencies)
      return out, errors
  
imports: from __future__ import annotations
  import asyncio
  import logging
  import warnings
  from collections import defaultdict
  from timeit import default_timer
  from typing import ClassVar
  from tlz import groupby, valmap
  from dask.base import tokenize
  from dask.utils import key_split
  from distributed.diagnostics.plugin import SchedulerPlugin
  from distributed.metrics import time
label_desc: Use image node_id 1 for calling a global function with example usage: dependent_keys(tasks, complete) and returns: out,  errors