code: def local_ray_callbacks(callbacks=None):
      """
      Allows Dask-Ray callbacks to work with nested schedulers.
  
      Callbacks will only be used by the first started scheduler they encounter.
      This means that only the outermost scheduler will use global callbacks.
      """
      global_callbacks = callbacks is None
      if global_callbacks:
          callbacks, RayDaskCallback.ray_active = (
              RayDaskCallback.ray_active,
              set(),
          )
      try:
          yield callbacks or ()
      finally:
          if global_callbacks:
              RayDaskCallback.ray_active = callbacks
  
imports: import contextlib
  from collections import namedtuple, defaultdict
  from datetime import datetime
  from dask.callbacks import Callback
label_desc: Use image node_id 3 for calling a global function with example usage: local_ray_callbacks(callbacks) without return types