human
I wrote the below code

    queue = Queue()
    if isinstance(result, list):
        result_flat = set(flatten(result))
    else:
        result_flat = set([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
        try:
            for cb in callbacks:
                if cb[0]:
                    cb[0](dsk)
                started_cbs.append(cb)
            (dsk, dependencies) = cull(dsk, list(results))
            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 = _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, dumps((dsk[key], data)), dumps, loads, get_id, pack_exception), callback=queue.put)
            while (state['ready'] and (len(state['running']) < num_workers)):
                fire_task()
            while (state['waiting'] or state['ready'] or state['running']):
                (key, res_info, failed) = queue_get(queue)
                if failed:
                    (exc, tb) = loads(res_info)
                    if rerun_exceptions_locally:
                        data = dict(((dep, state['cache'][dep]) for...

What's a good function header?

Here's a good function header?
assistant
#Here's a good function header.

def get_async(apply_async, 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, **kwargs):