def feed_queue(objects, func, get_deps, results, state):
    """
    Starts producer threads for any objects which are ready to be processed
    (i.e. they have no dependencies which haven't been successfully processed).

    Shortcuts any objects whose dependencies have failed and places an
    (object, None, UpstreamError()) tuple on the results queue.
    """
    pending = state.pending()
    log.debug('Pending: {}'.format(pending))

    for obj in pending:
        deps = get_deps(obj)

        if any(state.failed in dep for dep in deps):
            log.debug('{} has upstream errors - not processing'.format(obj))
            results.put((obj, None, UpstreamError()))
            state.failed.add(obj)
        elif all(
            dep not in objects or dep in state.finished
            for dep in deps
        ):
            log.debug('Starting producer thread for {}'.format(obj))
            t = Thread(target=producer, args=(obj, func, results))
            t.daemon = True
            t.start()
            state.started.add(obj)

    if state.is_done():
        results.put(STOP)
dataset/ETHPy150Open docker/compose/compose/parallel.py/feed_queue/swapped operands of `in`