code: def apply_async(self, func, callback=None):
      cf_future = concurrent.futures.Future()
      cf_future.get = cf_future.result  # achieve AsyncResult API
  
      async def f(func, callback):
          batch, tasks = await self._to_func_args(func)
          key = f"{repr(batch)}-{uuid4().hex}"
  
          dask_future = self.client.submit(
              batch, tasks=tasks, key=key, **self.submit_kwargs
          )
          self.waiting_futures.add(dask_future)
          self._callbacks[dask_future] = callback
          self._results[dask_future] = cf_future
  
      self.client.loop.add_callback(f, func, callback)
  
      return cf_future
  
imports: from __future__ import print_function, division, absolute_import
  import asyncio
  import concurrent.futures
  import contextlib
  import time
  from uuid import uuid4
  import weakref
  from .parallel import parallel_config
  from .parallel import AutoBatchingMixin, ParallelBackendBase
label_desc: Use image node_id 8 for calling the DaskDistributedBackend obj's underlying member method code with example usage: obj.apply_async(func, callback) and returns: cf_future