code: def get_async(
      submit,
      num_workers,
      dsk,
      result,
      cache=None,
      get_id=default_get_id,
      rerun_exceptions_locally=None,
      pack_exception=default_pack_exception,
      raise_exception=reraise,
      callbacks=None,
      dumps=identity,
      loads=identity,
      chunksize=None,
      **kwargs,
  ):
      """Asynchronous get function
  
      This is a general version of various asynchronous schedulers for dask.  It
      takes a ``concurrent.futures.Executor.submit`` function to form a more
      specific ``get`` method that walks through the dask array with parallel
      workers, avoiding repeat computation and minimizing memory use.
  
      Parameters
      ----------
      submit : function
          A ``concurrent.futures.Executor.submit`` function
      num_workers : int
          The number of workers that task submissions can be spread over
      dsk : dict
          A dask dictionary specifying a workflow
      result : key or list of keys
          Keys corresponding to desired data
      cache : dict-like, optional
          Temporary storage of results
      get_id : callable, optional
          Function to return the worker id, takes no arguments. Examples are
          `threading.current_thread` and `multiprocessing.current_process`.
      rerun_exceptions_locally : bool, optional
          Whether to rerun failing tasks in local process to enable debugging
          (False by default)
      pack_exception : callable, optional
          Function to take an exception and ``dumps`` method, and return a
          serialized tuple of ``(exception, traceback)`` to send back to the
          scheduler. Default is to just raise the exception.
      raise_exception : callable, optional
          Function that takes an exception and a traceback, and raises an error.
      callbacks : tuple or list of tuples, optional
          Callbacks are passed in as tuples of length 5. Multiple sets of
          callbacks may be passed in as a list of tuples. For more information,
          see the dask.diagnostics documentation.
      dumps: callable, optional
          Function to serialize task data and results to communicate between
          worker and parent.  Defaults to identity.
      loads: callable, optional
          Inverse function of `dumps`.  Defaults to identity.
      chunksize: int, optional
          Size of chunks to use when dispatching work. Defaults to 1.
          If -1, will be computed to evenly divide ready work across workers.
  
      See Also
      --------
      threaded.get
      """
      chunksize = chunksize or config.get("chunksize", 1)
  
      queue = Queue()
  
      if isinstance(result, list):
          result_flat = set(flatten(result))
      else:
          result_flat = {result}
      results = set(result_flat)
  
      dsk = dict(dsk)
      with local_callbacks(callbacks) as callbacks:
          _, _, pretask_cbs, posttask_cbs, _ = unpack_callbacks(
              callbacks
          )
          started_cbs = []
          succeeded = False
          # if start_state_from_dask fails, we will have something
          # to pass to the final block.
          state = {}
          try:
              for cb in callbacks:
                  if cb[0]:
                      cb[0](dsk)
                  started_cbs.append(cb)
  
              keyorder = order(dsk)
  
              state = start_state_from_dask(
                  dsk, cache=cache, sortkey=keyorder.get
              )
  
              for _, start_state, _, _, _ in callbacks:
                  if start_state:
                      start_state(dsk, state)
  
              if rerun_exceptions_locally is None:
                  rerun_exceptions_locally = config.get(
                      "rerun_exceptions_locally", False
                  )
  
              if state["waiting"] and not state["ready"]:
                  raise ValueError("Found no accessible jobs in dask")
  
              def fire_tasks(chunksize):
                  """Fire off a task to the thread pool"""
                  # Determine chunksize and/or number of tasks to submit
                  nready = len(state["ready"])
                  if chunksize == -1:
                      ntasks = nready
                      chunksize = -(ntasks // -num_workers)
                  else:
                      used_workers = -(
                          len(state["running"]) // -chunksize
                      )
                      avail_workers = max(num_workers - used_workers, 0)
                      ntasks = min(nready, chunksize * avail_workers)
  
                  # Prep all ready tasks for submission
                  args = []
                  for _ in range(ntasks):
                      # Get the next task to compute (most recently added)
                      key = state["ready"].pop()
                      # Notify task is running
                      state["running"].add(key)
                      for f in pretask_cbs:
                          f(key, dsk, state)
  
                      # Prep args to send
                      data = {
                          dep: state["cache"][dep]
                          for dep in get_dependencies(dsk, key)
                      }
                      args.append(
                          (
                              key,
                              dumps((dsk[key], data)),
                              dumps,
                              loads,
                              get_id,
                              pack_exception,
                          )
                      )
  
                  # Batch submit
                  for i in range(-(len(args) // -chunksize)):
                      each_args = args[
                          i * chunksize : (i + 1) * chunksize
                      ]
                      if not each_args:
                          break
                      fut = submit(batch_execute_tasks, each_args)
                      fut.add_done_callback(queue.put)
  
              # Main loop, wait on tasks to finish, insert new ones
              while (
                  state["waiting"] or state["ready"] or state["running"]
              ):
                  fire_tasks(chunksize)
                  for key, res_info, failed in queue_get(
                      queue
                  ).result():
                      if failed:
                          exc, tb = loads(res_info)
                          if rerun_exceptions_locally:
                              data = {
                                  dep: state["cache"][dep]
                                  for dep in get_dependencies(dsk, key)
                              }
                              task = dsk[key]
                              _execute_task(
                                  task, data
                              )  # Re-execute locally
                          else:
                              raise_exception(exc, tb)
                      res, worker_id = loads(res_info)
                      state["cache"][key] = res
                      finish_task(
                          dsk, key, state, results, keyorder.get
                      )
                      for f in posttask_cbs:
                          f(key, res, dsk, state, worker_id)
  
              succeeded = True
  
          finally:
              for _, _, _, _, finish in started_cbs:
                  if finish:
                      finish(dsk, state, not succeeded)
  
      return nested_get(result, state["cache"])
  
imports: from __future__ import annotations
  import os
  from collections.abc import Mapping, Sequence
  from concurrent.futures import Executor, Future
  from functools import partial
  from queue import Empty, Queue
  from dask import config
  from dask.callbacks import local_callbacks, unpack_callbacks
  from dask.core import _execute_task, flatten, get_dependencies, has_tasks, reverse_dict
  from dask.order import order
  from dask.typing import Key
label_desc: Use image node_id 11 for calling a global function with example usage: get_async(submit, num_workers, dsk, result, cache, get_id, rerun_exceptions_locally, pack_exception, raise_exception, callbacks, dumps, loads, chunksize) and returns: nested_get