

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\n\n    This is a general version of various asynchronous schedulers for dask.  It\n    takes a an apply_async function as found on Pool objects to form a more\n    specific ``get`` method that walks through the dask array with parallel\n    workers, avoiding repeat computation and minimizing memory use.\n\n    This function evaluates the entire graph, regardless of the given output\n    keys.  You may want to cull your graph ahead of time with\n    ``dask.optimize.cull``.\n\n    Parameters\n    ----------\n\n    apply_async : function\n        Asynchronous apply function as found on Pool or ThreadPool\n    num_workers : int\n        The number of active tasks we should have at any one time\n    dsk: dict\n        A dask dictionary specifying a workflow\n    result : key or list of keys\n        Keys corresponding to desired data\n    cache : dict-like, optional\n        Temporary storage of results\n    get_id : callable, optional\n        Function to return the worker id, takes no arguments. Examples are\n        `threading.current_thread` and `multiprocessing.current_process`.\n    rerun_exceptions_locally : bool, optional\n        Whether to rerun failing tasks in local process to enable debugging\n        (False by default)\n    callbacks : tuple or list of tuples, optional\n        Callbacks are passed in as tuples of length 4. Multiple sets of\n        callbacks may be passed in as a list of tuples. For more information,\n        see the dask.diagnostics documentation.\n\n    See Also\n    --------\n\n    threaded.get\n    '
    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 '
        key = state['ready'].pop()
        state['running'].add(key)
        for f in pretask_cbs:
            f(key, dsk, state)
        data = dict(((dep, state['cache'][dep]) for dep in get_dependencies(dsk, key)))
        apply_async(execute_task, args=[key, dsk[key], data, queue, get_id, raise_on_exception])
    while (state['ready'] and (len(state['running']) < num_workers)):
        fire_task()
    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)
            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()
    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/original