Short answer: yes — that assumption is off.

max_workers

limits how many tasks run concurrently, not how many are created/scheduled. In your loop, you’re scheduling a new batch before the previous batch has finished because you never block on those futures. That’s why “Pending” explodes. What to change You need to add backpressure between batches by waiting on the current batch before scheduling the next. Two easy ways: - Gate on the mapped tasks - Or gate on the downstream ingestion task (which already depends on the mapped tasks via

wait_for

) Example with explicit gating per batch

from prefect import flow
from prefect.futures import wait
from prefect.task_runners import ProcessPoolTaskRunner

@flow(task_runner=ProcessPoolTaskRunner(max_workers=os.cpu_count() - 8))
def pocket(uid: str, pid: str, pocket: str):
    batch_size = 256
    protein_pdbqt = f"{PROTEIN_DIR}/{pid}/{pid}.pdbqt"

    for i in range(0, len(sdf_files), batch_size):
        batch = sdf_files[i:i + batch_size]

        futures = run_binana_on_sdf.map(
            sdf_filepath=[str(f) for f in batch],
            protein_pdbqt=protein_pdbqt,
            outputdir=BINANA_OUTPUT_DIR,
        )

        # Option A: wait for the mapped tasks to finish before moving on
        wait(futures)

        # Then run ingestion and block before scheduling the next batch
        ingest_future = ingest_binana_output(
            [str(path) for path in batch],
            unidock_results_map,
            wait_for=futures,
        )
        ingest_future.result()  # important: prevents the next batch from being scheduled early

Why this works -

run_binana_on_sdf.map(...)

creates a task per file immediately. Without a blocking call, your for-loop keeps going and creates the next batch right away. -

wait(futures)

or

ingest_future.result()

forces the flow to pause before scheduling the next batch, keeping “Pending” roughly bounded to the current batch (+ its ingestion task). -

max_workers

still controls how many of those are actually running at once. If you want a global/cluster-wide cap on concurrent BINANA tasks across all flows, add a concurrency slot around the work inside the task:

from prefect.concurrency.sync import concurrency

@task
def run_binana_on_sdf(...):
    with concurrency("binana", capacity=56):  # pick your capacity
        ...  # do work

Note: this controls concurrent “Running” tasks, not how many are created. It complements, but does not replace, the batch gating shown above. Quick check questions - Are both

run_binana_on_sdf

and

ingest_binana_output

decorated with

@task

? If not, the

wait_for

won’t have any effect. - Do you want to ingest per-batch (as shown) or stream ingestion as items complete (we can switch to

as_completed

if you prefer streaming)?

Thanks for the details — a few important clarifications that explain what you’re seeing: - Mapping creates task runs immediately. Even if you “wait”, the current batch’s mapped task runs are created in the API right away and will show as Pending until they start Running. That’s expected for the batch you just submitted. - If you see thousands of Pending, it usually means the next batches are being submitted before the current batch finishes (i.e., the loop isn’t actually being gated), or there’s additional mapping happening inside your tasks. Given you tried

wait(futures)

and

.result()

with no change, I suspect either: - The wait is not placed where it actually blocks the loop, or - There’s additional mapping/submission inside

ingest_binana_output

(or inside

run_binana_on_sdf

) that’s creating more tasks, or - An import mixup (e.g., you’re calling

asyncio.wait

instead of

prefect.futures.wait

). Can you share exactly where you put the wait lines? A “known good” per-batch pattern Try this exact structure to force a hard gate between batches. Note the explicit import and the fact we block before the loop continues.

from prefect import flow, task
from prefect.futures import wait as prefect_wait
from prefect.task_runners import ProcessPoolTaskRunner
import os

@task
def run_binana_on_sdf(sdf_filepath: str, protein_pdbqt: str, outputdir: str):
    # do work
    ...

@task
def ingest_binana_output(paths: list[str], unidock_results_map: dict):
    # ingest per-batch
    ...

@flow(task_runner=ProcessPoolTaskRunner(max_workers=max(1, os.cpu_count() - 8)))
def pocket(uid: str, pid: str, pocket: str):
    batch_size = 256
    protein_pdbqt = f"{PROTEIN_DIR}/{pid}/{pid}.pdbqt"

    for i in range(0, len(sdf_files), batch_size):
        batch = sdf_files[i:i + batch_size]

        futures = run_binana_on_sdf.map(
            sdf_filepath=[str(f) for f in batch],
            protein_pdbqt=protein_pdbqt,
            outputdir=BINANA_OUTPUT_DIR,
        )

        # HARD GATE 1: block on the batch’s mapped tasks
        prefect_wait(list(futures))  # ensure you’re using prefect.futures.wait

        # Ingest, then HARD GATE 2: block before scheduling the next batch
        ingest_future = ingest_binana_output(
            [str(p) for p in batch],
            unidock_results_map,
        )
        ingest_future.result()

This keeps Pending bounded to approximately the current batch (+ the single ingestion task). If this still shows thousands of Pending, it means something else is submitting additional work (often nested mapping). If

ingest_binana_output

maps or submits tasks internally, that will multiply counts; same for

run_binana_on_sdf

. If you prefer streaming but still a hard cap, use a rolling window (at most N in-flight): - Keep a list of at most 256

.submit(...)

futures (not

map

), - Use

prefect.futures.as_completed

to drain them, - Then call

ingest_binana_output(...).result()

per window. Global caps (optional) These control how many tasks can run, not how many are created, but they’re useful when multiple flows share resources: - Tag-based concurrency for tasks: decorate

run_binana_on_sdf

with a tag and set a global limit: - Create the limit:

prefect concurrency-limit create binana 56

- Tag the task:

@task(tags=["binana"])

Docs: Tag-based concurrency limits - Work pool concurrency for flow runs: limit how many flow runs execute at once on a pool: -

prefect work-pool set-concurrency-limit "my-pool" 10

If the real goal is to lower orchestration overhead and keep Pending extremely small, consider batching inside a single task (no mapping at all): one

process_batch

task that loops over the 256 items internally. That creates one TaskRun per batch, keeps the UI quiet, and is often faster for very short per-item work. Next debugging steps - Where exactly did you put the wait? Can you paste that snippet?

- Does

ingest_binana_output

or

run_binana_on_sdf

submit or map other tasks internally? - Roughly how many files are in

sdf_files

? - Which Prefect version are you on? References - Futures helpers (wait/as_completed): Futures API - Concurrency helpers (context manager): Concurrency API - Mapping overview: Blog: Task mapping at scale

Good news: this isn’t about ProcessPoolTaskRunner. The core issue is your

.map(...)

call is unintentionally mapping over your strings. In Prefect, mapping expands any argument that is iterable. Strings are iterables, so this line will try to map across each character of

protein_pdbqt

and

BINANA_OUTPUT_DIR

too:

futures = run_binana_on_sdf.map(
    sdf_filepath=[str(f) for f in batch],
    protein_pdbqt=protein_pdbqt,          # <-- string, iterable
    outputdir=BINANA_OUTPUT_DIR,          # <-- string, iterable
)

That can create way more task runs than you expect. Fix it by marking those as unmapped constants:

from prefect import unmapped

futures = run_binana_on_sdf.map(
    sdf_filepath=[str(f) for f in batch],
    protein_pdbqt=unmapped(protein_pdbqt),
    outputdir=unmapped(BINANA_OUTPUT_DIR),
)

Then your per-batch gating with

prefect.futures.wait

or

future.result()

will actually bound Pending to the current batch. Putting it together with hard gating

from prefect import flow, task, unmapped
from prefect.futures import wait as prefect_wait
from prefect.task_runners import ProcessPoolTaskRunner
import os

@task
def run_binana_on_sdf(sdf_filepath: str, protein_pdbqt: str, outputdir: str):
    ...

@task
def ingest_binana_output(paths: list[str], unidock_results_map: dict):
    ...

@flow(task_runner=ProcessPoolTaskRunner(max_workers=max(1, os.cpu_count() - 8)))
def pocket(uid: str, pid: str, pocket: str):
    batch_size = 256
    protein_pdbqt = f"{PROTEIN_DIR}/{pid}/{pid}.pdbqt"

    for i in range(0, len(sdf_files), batch_size):
        batch = sdf_files[i:i + batch_size]

        futures = run_binana_on_sdf.map(
            sdf_filepath=[str(f) for f in batch],
            protein_pdbqt=unmapped(protein_pdbqt),
            outputdir=unmapped(BINANA_OUTPUT_DIR),
        )

        prefect_wait(list(futures))  # wait for current batch to finish

        ingest_binana_output([str(p) for p in batch], unidock_results_map).result()

If you still want a strict cap on the number of created task runs at any time, use a rolling window with

.submit

instead of `.map`:

from prefect.futures import as_completed

@flow(task_runner=ProcessPoolTaskRunner(max_workers=max(1, os.cpu_count() - 8)))
def pocket(uid: str, pid: str, pocket: str):
    window = 256
    batch_paths, inflight = [], []

    protein_pdbqt = f"{PROTEIN_DIR}/{pid}/{pid}.pdbqt"

    for path in sdf_files:
        fut = run_binana_on_sdf.submit(
            sdf_filepath=str(path),
            protein_pdbqt=protein_pdbqt,
            outputdir=BINANA_OUTPUT_DIR,
        )
        inflight.append(fut)
        batch_paths.append(str(path))

        if len(inflight) >= window:
            # drain the window
            for _ in as_completed(inflight):
                pass
            ingest_binana_output(batch_paths, unidock_results_map).result()
            inflight.clear()
            batch_paths.clear()

    if inflight:
        for _ in as_completed(inflight):
            pass
        ingest_binana_output(batch_paths, unidock_results_map).result()

Optional global limits - Task-level concurrency (across flows): tag

run_binana_on_sdf

and create a limit -

@task(tags=["binana"])

-

prefect concurrency-limit create binana 56

Docs: Tag-based concurrency limits - Work pool concurrency (flow runs):

prefect work-pool set-concurrency-limit "my-pool" 10

(limits concurrent flow runs on the pool) If switching to

unmapped(...)

doesn’t bring Pending down to “one batch at a time,” share the exact

.map(...)

call and any mapping that might occur inside your tasks — but in my experience, the string-iterable mapping is the culprit here.

I hear you — this feels frustrating. A few key points that explain what you’re seeing and how to make it predictable: Why you see “hundreds Running” even with max_workers=96 - In Prefect 3, “Running” means “the engine is executing the task or it’s queued in the task runner’s executor”. With ProcessPoolTaskRunner, we submit all mapped tasks to the process pool; only up to

max_workers

actually run at once, but Prefect marks them Running before they actually acquire a process slot. So the UI can show many more “Running” than CPU slots. It’s unintuitive, but expected. Why “Pending” grows - Pending will grow if the flow keeps submitting new tasks. That happens if: - The loop isn’t truly gated (e.g., waiting on the wrong futures or waiting in a place that doesn’t block the loop). - There’s additional mapping inside your tasks (nested task submissions). - Or strings weren’t marked

unmapped(...)

and you accidentally mapped over them (you said you tried this, but I’m keeping it on the list since it’s the most common multiplier). Concrete ways to hard-cap what gets created and what shows as Running 1) Enforce a rolling window with submit + as_completed (no map) This guarantees at most N tasks are ever in-flight/created at once and avoids surprises with mapping. It also makes your batching/ingestion per window straightforward.

from prefect import flow, task
from prefect.futures import as_completed
from prefect.task_runners import ProcessPoolTaskRunner
import os

@task
def run_binana_on_sdf(sdf_filepath: str, protein_pdbqt: str, outputdir: str):
    ...

@task
def ingest_binana_output(paths: list[str], unidock_results_map: dict):
    ...

@flow(task_runner=ProcessPoolTaskRunner(max_workers=max(1, os.cpu_count() - 8)))
def pocket(uid: str, pid: str, pocket: str):
    window = 256
    protein_pdbqt = f"{PROTEIN_DIR}/{pid}/{pid}.pdbqt"
    inflight, batch_paths = [], []

    for path in sdf_files:
        fut = run_binana_on_sdf.submit(
            sdf_filepath=str(path),
            protein_pdbqt=protein_pdbqt,
            outputdir=BINANA_OUTPUT_DIR,
        )
        inflight.append(fut)
        batch_paths.append(str(path))

        if len(inflight) >= window:
            # drain the window
            for _ in as_completed(inflight):
                pass
            ingest_binana_output(batch_paths, unidock_results_map).result()
            inflight.clear()
            batch_paths.clear()

    # drain remainder
    if inflight:
        for _ in as_completed(inflight):
            pass
        ingest_binana_output(batch_paths, unidock_results_map).result()

- This pattern prevents “pending explosion” because you never create more than

window

task runs before draining. 2) If you want the UI to reflect your CPU cap, add a task-level concurrency limit via a tag - This stops tasks from entering Running beyond your limit (they’ll sit in AwaitingConcurrencySlot instead of Running). - Decorate your BINANA task:

@task(tags=["binana"])

- Create a limit (e.g., 96): -

prefect concurrency-limit create binana 96

- Inspect slots: -

prefect concurrency-limit inspect binana

Docs: Tag-based concurrency limits 3) If BINANA spawns threads itself, prevent OS-level oversubscription - External tools often use OpenMP/BLAS threads per process. If each Prefect task runs BINANA which uses many threads, you’ll exceed CPU count regardless of process pool size. Set environment variables (e.g.

OMP_NUM_THREADS=1

,

MKL_NUM_THREADS=1

,

OPENBLAS_NUM_THREADS=1

) for the task environment. 4) If your per-item work is fast, batch inside a single task - One task per 256 items avoids thousands of API state transitions and keeps both “Pending” and “Running” small. It’s often the fastest approach for short operations. Why the earlier gating may not have changed the UI

- Even when you block correctly, the current batch’s tasks will flip to Running in the UI immediately (engine-side), then queue inside the process pool. You’ll still see >96 Running in the UI, but only ~96 actually execute concurrently. The rolling-window approach above is the most reliable way to bound both the number created and the number marked Running at any one time. If this still doesn’t behave as expected, can you share: - Prefect version - Whether this is running as a deployment (and on what worker type) - A minimal snippet of the two task definitions (to see if there’s nested mapping or additional submissions) - Rough count of

sdf_files

Happy to iterate on a minimal repro with you. Links for reference: - Futures helpers (as_completed/wait): Futures API - Concurrency limits: Tag-based concurrency limits - Work pool concurrency (limits concurrent flow runs, not tasks):

prefect work-pool set-concurrency-limit "my-pool" 10

if i allow all the tasks to be created @ the same time, then Prefect (eventually) locks up

this postpones said lockup by about 24 hours

and let Ray handle the "many small tasks" work

like we want to enable arbitrary scale for OSS in terms of run objects in the API but there's a non-trivial amount of config you'd have to do and honestly prob some perf bottlenecks we still need to sniff out if you feel like you have any color to add here about what you ran into that'd be super appreciated/valuable https://github.com/PrefectHQ/prefect/discussions/18150

or if you prefer, a new discussion