| 106 | done_counter = 0 |
| 107 | |
| 108 | async def run_one(i: int, inp: InputT) -> None: |
| 109 | nonlocal done_counter |
| 110 | async with sem: |
| 111 | try: |
| 112 | results[i] = await flow_fn(inp, cfg=cfg, **flow_kwargs) |
| 113 | except Exception as exc: |
| 114 | errors.append((i, exc)) |
| 115 | if on_error == "raise": |
| 116 | raise |
| 117 | finally: |
| 118 | # asyncio is single-threaded; this increment + read + |
| 119 | # callback all happen synchronously between await points. |
| 120 | done_counter += 1 |
| 121 | if on_progress is not None: |
| 122 | on_progress(done_counter, len(inputs)) |
| 123 | |
| 124 | # Same call shape for both modes — `run_one` swallows its own |
| 125 | # exception when on_error="skip", and re-raises (cancelling siblings |