INNER CODE UNIT · Python

execute_input

Hebbian-Robotics/hflow · src/hflow/app.py:512

    async def execute_input(value: _ConcurrentInputT) -> _ConcurrentResultT:
        return await operation(value)

    def submit_next_input() -> None:
        nonlocal next_input_index
        if next_input_index < len(inputs):
            input_index_by_task[asyncio.create_task(execute_input(inputs[next_input_index]))] = (
                next_input_index
            )
            next_input_index += 1

    for _ in range(min(int(concurrency_limit), len(inputs))):
        submit_next_input()
    try:
        while input_index_by_task:
            completed_tasks, _ = await asyncio.wait(
                input_index_by_task, return_when=asyncio.FIRST_COMPLETED
            )

View source record →