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):
    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 KeyboardInterrupt:
            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'])
Asynchronous get function
dataset/ETHPy150Open dask/dask/dask/async.py/get_async