def get_async(apply_async, num_workers, dsk, result, cache=None,
              queue=None, get_id=default_get_id, raise_on_exception=False,
              rerun_exceptions_locally=None, callbacks=None, **kwargs):
    """ Asynchronous get function

    This is a general version of various asynchronous schedulers for dask.  It
    takes a an apply_async function as found on Pool objects to form a more
    specific ``get`` method that walks through the dask array with parallel
    workers, avoiding repeat computation and minimizing memory use.

    This function evaluates the entire graph, regardless of the given output
    keys.  You may want to cull your graph ahead of time with
    ``dask.optimize.cull``.

    Parameters
    ----------

    apply_async : function
        Asynchronous apply function as found on Pool or ThreadPool
    num_workers : int
        The number of active tasks we should have at any one time
    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)
    callbacks : tuple or list of tuples, optional
        Callbacks are passed in as tuples of length 4. Multiple sets of
        callbacks may be passed in as a list of tuples. For more information,
        see the dask.diagnostics documentation.

    See Also
    --------

    threaded.get
    """
    assert queue

    if callbacks is None:
        callbacks = _globals['callbacks']
    start_cbs, start_state_cbs, pretask_cbs, posttask_cbs, finish_cbs = unpack_callbacks(callbacks)

    if isinstance(result, list):
        result_flat = set(flatten(result))
    else:
        result_flat = set([result])
    results = set(result_flat)

    dsk = dsk.copy()
    for f in start_cbs:
        f(dsk)

    dsk = cull(dsk, list(results))

    keyorder = order(dsk)

    state = start_state_from_dask(dsk, cache=cache, sortkey=keyorder.get)

    for f in start_state_cbs:
        f(dsk, state)

    if rerun_exceptions_locally is None:
        rerun_exceptions_locally = _globals.get('rerun_exceptions_locally', False)

    if state['waiting'] and not state['ready']:
        raise ValueError("Found no accessible jobs in dask")

    def fire_task():
        """ Fire off a task to the thread pool """
        # Choose a good task to compute
        key = state['ready'].pop()
        state['running'].add(key)
        for f in pretask_cbs:
            f(key, dsk, state)

        # Prep data to send
        data = dict((dep, state['cache'][dep])
                    for dep in get_dependencies(dsk, key))
        # Submit
        apply_async(execute_task, args=[key, dsk[key], data, queue,
                                        get_id, raise_on_exception])

    # Seed initial tasks into the thread pool
    while state['ready'] and len(state['running']) < num_workers:
        fire_task()

    # Main loop, wait on tasks to finish, insert new ones
    while state['waiting'] or state['ready'] or state['running']:
        try:
            key, res, tb, worker_id = queue.get()
        except __HOLE__:
            for f in finish_cbs:
                f(dsk, state, True)
            raise
        if isinstance(res, Exception):
            for f in finish_cbs:
                f(dsk, state, True)
            if rerun_exceptions_locally:
                data = dict((dep, state['cache'][dep])
                            for dep in get_dependencies(dsk, key))
                task = dsk[key]
                _execute_task(task, data)  # Re-execute locally
            else:
                raise(remote_exception(res, tb))
        state['cache'][key] = res
        finish_task(dsk, key, state, results, keyorder.get)
        for f in posttask_cbs:
            f(key, res, dsk, state, worker_id)
        while state['ready'] and len(state['running']) < num_workers:
            fire_task()

    # Final reporting
    while state['running'] or not queue.empty():
        key, res, tb, worker_id = queue.get()

    for f in finish_cbs:
        f(dsk, state, False)

    return nested_get(result, state['cache'])
dataset/ETHPy150Open dask/dask/dask/async.py/get_async