https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> i have a map of tasks, but all of the...
# ask-marvin
o
@Marvin i have a map of tasks, but all of them need one task in serialexection, is it possible to call a task from a task that has task concurrency?
m
✅ thought for 160.8 seconds
Yes — a task can call another task in Prefect 3.x. But for your case, the better question is: how do I let many tasks run concurrently, while forcing one specific step to run one-at-a-time? The usual answer is: Don’t rely on nesting alone for serialization. Use a concurrency limit on that one task. ### Recommended pattern If you have: - many items processed in parallel - one bottleneck task that must run serially do this:
Copy code
python
from prefect import flow, task

@task
def parallel_step(x):
    return x * 2

@task(tags=["serial-step"])
def serial_step(x):
    print(f"serial processing {x}")
    return x + 1

@flow
def my_flow(items):
    a = parallel_step.map(items)
    b = serial_step.map(a.result())
    return b.result()
Then create a concurrency limit of 1 for that tag. I’d verify the CLI first, and in Prefect 3 the command is:
Copy code
bash
prefect concurrency-limit --help
If you want, I can give you the exact command for creating that limit in your environment. ### Important behavior If you do this inside a task:
Copy code
python
@task
def outer(x):
    return inner(x)
that works, but it’s effectively synchronous. If you do:
Copy code
python
@task
def outer(x):
    fut = inner.submit(x)
    return fut.result()
that also works, but can be awkward if you’re using a bounded task runner, because nested submissions can create blocking behavior. ### Best practice For your use case, I’d recommend: - keep orchestration at the flow level when possible - use 
.map()
or 
.submit()
for parallel work - use a concurrency limit = 1 on the serial task That gives you: - parallel upstream tasks - only one execution of the bottleneck task at a time - cleaner dependency management ### Alternative: explicit chaining If you truly need strict serial order across all mapped items, you can chain with `wait_for`:
Copy code
python
from prefect import flow, task

@task
def serial_step(x):
    return x + 1

@flow
def my_flow(items):
    futures = []
    previous = None

    for item in items:
        if previous is None:
            f = serial_step.submit(item)
        else:
            f = serial_step.submit(item, wait_for=[previous])
        futures.append(f)
        previous = f

    return [f.result() for f in futures]
That enforces one-after-another ordering, not just “max concurrency 1”. ### Short answer - Yes, tasks can call tasks. - Yes, you can mix mapped concurrency with one serial step. - Best solution: put a concurrency limit of 
1
on the serial task instead of trying to manage it purely through nested task calls. Helpful docs: - Tasks - Run work concurrently If you want, I can show the exact Prefect 3 pattern for: 1. serializing one mapped task with a tag limit, or 2. preserving the original item order strictly.
o
@Marvin i hear you but the parralel task results can be two old, imagine 100 parrallilsed tasks are done, but the seriealized arent ready yet, and results of the parrallel vanish due timeout. so i cant do all parrallel tasks at once. each of them must to the serial write and in due time, so starting the next parrallel task before write of the old one ready is not an option
m
✅ thought for 24.1 seconds
Got it — you don’t want: 1. all parallel work to finish first 2. then queue up the serialized writes afterward Because the intermediate results may expire / become invalid before the serial write happens. So your real requirement is more like: - each item does 
parallel_compute -> serial_write
- many computes can be in flight - but writes must happen one at a time - and compute for item 
N+1
should not get too far ahead of the serialized write queue That means you need backpressure, not just a serial task. ### Important point If you allow 100 compute tasks to run at once, and the write step is single-file, then you can still build up a queue of 100 results waiting to be written. A concurrency limit on the write task alone does not solve that. ### What to do instead You want to limit the whole pipeline width so only a small number of items are in progress at once. A good pattern is: - one task does the “parallel” compute for a single item - one task does the serialized write - the flow only keeps a small number of item pipelines active at a time For example, process items in small batches:
Copy code
python
from prefect import flow, task

@task
def compute(item):
    # expensive / parallelizable step
    return f"result-{item}"

@task(tags=["serial-write"])
def write_result(result):
    # must run one at a time
    print(f"writing {result}")

@flow
def my_flow(items, window_size=5):
    for i in range(0, len(items), window_size):
        batch = items[i:i + window_size]

        computed = compute.map(batch)
        write_result.map(computed.result())
This does not launch all 100 at once. It launches only 
window_size
computes, then writes those, then moves on. ### If results must be written immediately Then the safest design is actually not to map all compute tasks at once. Instead, submit only a few item pipelines at a time:
Copy code
python
from prefect import flow, task

@task
def compute(item):
    return f"result-{item}"

@task(tags=["serial-write"])
def write_result(result):
    print(f"writing {result}")

@flow
def my_flow(items, in_flight=3):
    for i in range(0, len(items), in_flight):
        chunk = items[i:i + in_flight]

        futures = [compute.submit(item) for item in chunk]
        results = [f.result() for f in futures]

        write_futures = [write_result.submit(r) for r in results]
        [f.result() for f in write_futures]
This gives you bounded parallelism: - only 
in_flight
computes happen at once - writes remain serialized via tag concurrency limit - you never accumulate a huge stale backlog ### If each item must write before more work starts If you mean literally: - item A compute - item A write - only then item B compute then that is no longer parallel across items. That’s just serial processing:
Copy code
python
from prefect import flow, task

@task
def compute(item):
    return f"result-{item}"

@task
def write_result(result):
    print(f"writing {result}")

@flow
def my_flow(items):
    for item in items:
        result = compute.submit(item).result()
        write_result.submit(result).result()
### Best mental model You likely want one of these: Option A: small bounded pipeline window - a few computes in flight - one write at a time - prevents stale results Option B: fully serial per item - safest - least throughput ### What not to do Avoid this shape if results expire quickly:
Copy code
python
computed = compute.map(all_items)
write_result.map(computed.result())
because it creates a barrier: - all compute first - then all writes That’s exactly the stale-result problem you described. ### My recommendation Use bounded chunking / windowing plus a serial write task. If you want, I can give you a Prefect 3 pattern that does a rolling window of, say, 5 items at a time, so compute and write overlap without letting the write queue grow unbounded.
o
@Marvin i really do not see why you are trying to avoid a nested task?
m
✅ thought for 13.6 seconds
I’m not avoiding it because it’s impossible — nested tasks are allowed. I’m avoiding it because for your requirement nested tasks alone usually don’t control backlog the way people expect. If you want, you can do this:
Copy code
python
from prefect import flow, task

@task
def compute(item):
    return f"result-{item}"

@task(tags=["serial-write"])
def write_result(result):
    print(f"writing {result}")

@task
def process_one(item):
    result = compute(item)          # nested task call
    write_result(result)            # nested task call
    return result

@flow
def my_flow(items):
    futures = [process_one.submit(item) for item in items]
    return [f.result() for f in futures]
### Why that may or may not help Here 
process_one
runs concurrently for many items. But inside each `process_one`: - 
compute(item)
runs - then 
write_result(result)
runs If 
write_result
is constrained to one-at-a-time, then many 
process_one
task runs can still pile up waiting for that serialized step. So nested tasks do express the dependency: 
compute -> write
What they do not automatically guarantee is: - bounded queue size - freshness protection - backpressure strong enough for your timeout constraints ### The real distinction Nested tasks solve: - “can item B’s write depend on item B’s compute?” → yes Nested tasks do not by themselves solve: - “how do I stop too many computed results from sitting around waiting for the single writer?” That’s why I was steering toward bounded concurrency/windowing. ### If you want nested tasks anyway Then the pattern is:
Copy code
python
from prefect import flow, task

@task
def compute(item):
    return f"result-{item}"

@task(tags=["serial-write"])
def write_result(result):
    print(f"writing {result}")
    return True

@task
def process_one(item):
    result = compute(item)
    write_result(result)
    return result

@flow
def my_flow(items):
    futures = process_one.map(items)
    return futures.result()
### But one warning If you map 
process_one
over 100 items, you still have 100 item-pipelines active. If all computes finish quickly and writes are slow, those 100 mapped parent tasks can still sit around waiting on the serialized write step. So yes: - nested task calls are valid - they may even be a clean way to model your pipeline - but they are not enough if your real issue is “results expire while waiting for the serial bottleneck” ### Short answer You absolutely can call a task from a task. I wasn’t discouraging nested tasks because they’re unsupported — I was warning that they don’t fully solve the timeout/backpressure problem unless you also limit how many 
process_one
pipelines are active. If you want, I can show you the exact nested-task version with bounded concurrency, which sounds like the thing you actually want: - nested 
compute -> write
- only 
N
item pipelines active at once - serialized writer.
o
@Marvin what happens if a task is orderend to spawen a nested task (but its labeled concurrency is occupied) ?
m
✅ thought for 112.3 seconds
Short answer: it waits. If a nested task is ready to run, but its concurrency-limited tag is fully occupied, Prefect does not immediately fail it. The child task stays blocked until a slot becomes available. ### What that means in practice If you have something like:
Copy code
python
@task(tags=["serial-write"])
def write_result(x):
    ...

@task
def process_one(x):
    y = compute(x)
    write_result(y)
and 
"serial-write"
is already occupied, then 
write_result(y)
will wait for the tag slot. ### What happens to the parent task? That depends on how the nested task is called: #### 1. Direct nested call
Copy code
python
write_result(y)
Then the parent task waits too. 
process_one
is blocked until 
write_result
can acquire the concurrency slot and finish. #### 2. Submitted nested task
Copy code
python
f = write_result.submit(y)
f.result()
This is similar from your code’s perspective: the parent still waits at 
f.result()
. If you do:
Copy code
python
f = write_result.submit(y)
return f
then the parent can finish without waiting, and the child waits on the concurrency slot independently. ### Important subtlety If the parent task itself also holds a concurrency-limited slot while waiting for the child, you can create a deadlock. For example:
Copy code
python
@task(tags=["serial-write"])
def parent(x):
    child(x)   # child also needs "serial-write"
If the limit for 
"serial-write"
is 
1
, then: - parent acquires the only slot - parent calls child - child tries to acquire the same slot - child waits forever, because parent is holding it So: If parent and child share the same occupied concurrency limit, and parent waits for child, you can deadlock. ### For your case If your goal is: - 
compute(item)
can run concurrently - 
write(item)
must be serial then this is okay:
Copy code
python
@task
def compute(item):
    return item

@task(tags=["serial-write"])
def write(item):
    ...

@task
def process_one(item):
    result = compute(item)
    write(result)
Here: - 
process_one
tasks may run in parallel - each one blocks when it reaches 
write
- only one 
write
runs at a time - others wait That part works. ### But the practical consequence Even though the nested child waits correctly, you can still end up with: - many parent tasks sitting around blocked - many computed results waiting for the serial writer So the behavior is: - not fail - not skip - not auto-retry in the user sense - it waits until a slot opens ### CLI for tag concurrency limits Verified CLI:
Copy code
bash
prefect concurrency-limit --help
Useful docs: - Tasks - Run work concurrently If you want, I can sketch the exact safe nested-task pattern for your pipeline, including how to avoid deadlocking the parent while still forcing 
write
to serialize.
o
@Marvin how to create a task tag that is equal to the flow run name?
m
✅ thought for 108.1 seconds
Yes — but not in the 
@task
decorator directly. Task decorator tags are static:
Copy code
python
@task(tags=["fixed-tag"])
def my_task():
    ...
If you want a tag equal to the current flow run name, use the 
tags(...)
context manager at runtime. ### Example
Copy code
python
from prefect import flow, task, tags
from prefect.runtime import flow_run

@task
def my_task():
    pass

@flow
def my_flow():
    run_name = flow_run.name

    with tags(run_name):
        my_task()
That applies the current flow run name as a dynamic tag to 
my_task
. ### If you want both static and dynamic tags They combine:
Copy code
python
from prefect import flow, task, tags
from prefect.runtime import flow_run

@task(tags=["serial-write"])
def my_task():
    pass

@flow
def my_flow():
    with tags(flow_run.name):
        my_task()
So the task run gets both: - 
serial-write
- 
<current flow run name>
### Caveat If the flow run name contains characters you don’t want to use as a tag, sanitize it first:
Copy code
python
from prefect import flow, task, tags
from prefect.runtime import flow_run

@task
def my_task():
    pass

@flow
def my_flow():
    run_tag = flow_run.name.replace(" ", "-").lower()
    with tags(run_tag):
        my_task()
### If you need this around 
.submit()
or 
.map()
That works too:
Copy code
python
@flow
def my_flow(items):
    with tags(flow_run.name):
        futures = my_task.map(items)
### Key point Use: - 
prefect.runtime.flow_run.name
to get the run name - 
with tags(...)
to apply it dynamically Helpful references: - Tasks - `tags` context manager If you want, I can show how to use the flow run ID or name as a per-flow concurrency tag for your nested serial-write pattern.
o
@Marvin suppse there are two flow runs, can they both exute the task labled with write if its slot is 1? so is the task label concurrency limit scoped to flow level?
m
✅ thought for 10.5 seconds
No — if both task runs have the same tag and that tag’s concurrency limit is 
1
, then across both flow runs only one of those task runs can run at a time. So this is not scoped to a single flow run. ### Meaning If you do:
Copy code
python
@task(tags=["write"])
def write_result(x):
    ...
and create:
Copy code
bash
prefect concurrency-limit create write 1
then all task runs tagged 
write
compete for that same single slot: - flow run A: 
write_result(...)
- flow run B: 
write_result(...)
Only one can execute at once. The other waits. ### So the scope is effectively - per tag - shared across the Prefect API / workspace - not per flow run ### If you want per-flow-run isolation Then you need a different tag per flow run, which is exactly why you were asking about tagging with the flow run name. For example:
Copy code
python
from prefect import flow, task, tags
from prefect.runtime import flow_run

@task
def write_result(x):
    ...

@flow
def my_flow(items):
    with tags(flow_run.name):
        write_result.map(items)
But note the implication: - if every flow run gets a unique tag - and you set concurrency limit 
1
on that unique tag then serialization happens within that flow run only, not across all flow runs. ### So: - tag = 
"write"
with limit 
1
-> global serialization across all flow runs using that tag - tag = 
flow_run.name
with limit 
1
-> serialization only within that specific flow run, assuming the limit exists for that tag ### Important practical caveat Prefect tag concurrency limits are for known tags. If you generate a brand-new tag per run, you’d need a corresponding concurrency limit for that tag, which is usually not practical unless you manage them programmatically. That’s why most people use: - one shared tag for global serialization, or - another concurrency mechanism if they need per-run dynamic isolation ### Direct answer No, task tag concurrency limits are not flow-scoped. If two different flow runs use the same tag and its limit is 
1
, they will block each other. Docs: - Tasks - Run work concurrently If you want, I can help you choose between: 1. global single writer across all flow runs, or 2. one writer per flow run.
o
@Marvin i think you forgot that only the write operation must be limited per flow run, not the fetch opersion
m
✅ thought for 15.9 seconds
Yes — you’re right. I was over-applying the tag. If only the write must be limited within each flow run, and fetch should stay unconstrained, then: - 
fetch
should not get the per-flow tag/concurrency control - only 
write
should get it ### But there’s an important limitation Prefect’s built-in tag concurrency limits are global per tag, not automatically “per flow run”. So if you want: - flow run A: one write at a time - flow run B: one write at a time - A and B should not block each other then a shared tag like 
"write"
with limit 
1
is wrong, because that serializes writes across all flow runs. ### What you actually need You need a dynamic tag on the write task only, such as: - 
write-<flow_run_id>
or - 
write-<flow_run_name>
Then all writes within one flow run share the same tag, but writes from different flow runs have different tags. Conceptually:
Copy code
python
fetch(item)          # no special tag
write(result)        # tag = f"write-{current_flow_run_id}"
### However: another practical limitation For Prefect tag-based concurrency limits to work, the limit has to exist for that tag. So if you generate a unique write tag per flow run, you’d also need a concurrency limit of 
1
created for each of those dynamic tags. That makes dynamic per-flow tag limits awkward unless you create/manage them yourself. ### Therefore, for per-flow-run-only write serialization, the most reliable option is often not tag concurrency Instead, do the serialization in the flow logic itself, and only for writes. For example:
Copy code
python
from prefect import flow, task

@task
def fetch(item):
    return f"data-{item}"

@task
def write(data):
    print(f"writing {data}")

@flow
def my_flow(items):
    fetch_futures = [fetch.submit(item) for item in items]

    for future in fetch_futures:
        data = future.result()
        write.submit(data).result()
That gives you: - fetches can all run concurrently - writes happen one-at-a-time within this flow run - another flow run can do the same independently ### But note This still allows all fetches to finish early and wait in memory/results, which earlier you said may be a problem. So if you need both: - fetch parallelism - write serial per flow run - avoid large stale backlog then you want a bounded number of item pipelines active per flow run Example:
Copy code
python
from prefect import flow, task
from prefect.futures import wait

@task
def fetch(item):
    return f"data-{item}"

@task
def write(data):
    print(f"writing {data}")

@flow
def my_flow(items, window=5):
    for i in range(0, len(items), window):
        chunk = items[i:i+window]
        futures = [fetch.submit(item) for item in chunk]

        for f in futures:
            write.submit(f.result()).result()
### Key correction You are correct that only write should be constrained. The fetch task should remain untagged / unconstrained unless you separately want a fetch limit. ### Bottom line - No, you should not tag both fetch and write if only write needs serialization. - Yes, the write-only limit should apply only to 
write
. - But Prefect tag concurrency is global per tag, not automatically per flow run. - So per-flow-run serialization of only write is best implemented either: - by your flow structure, or - with dynamically managed per-run tags/limits on 
write
only. If you want, I can show the cleanest nested-task pattern for exactly this: - 
fetch
concurrent - 
write
serial - serialization only within each flow run - no cross-flow blocking.
o
@Marvin Concurrency slot is empty but still the subtask does not start Submitting task process_interval to thread pool executor...
m
✅ thought for 225.1 seconds
That symptom strongly suggests thread-pool deadlock, not a free concurrency slot problem. If you see:
Copy code
text
Submitting task process_interval to thread pool executor...
and then it never actually starts, the usual cause is: - a parent task is already running in the thread pool - it submits a nested child task - then waits on that child - but all worker threads are already occupied by similar parent tasks - so the child is queued and can never get a worker So even if the Prefect concurrency slot is free, the subtask may still not run because the task runner thread pool has no free worker. ### Typical deadlock shape
Copy code
python
@task
def child(x):
    return x * 2

@task
def parent(x):
    fut = child.submit(x)
    return fut.result()   # blocks parent thread

@flow(task_runner=ThreadPoolTaskRunner(max_workers=2))
def my_flow():
    futures = [parent.submit(i) for i in range(2)]
    return [f.result() for f in futures]
What happens: - both 
parent
tasks occupy the two worker threads - each parent submits a 
child
- each parent waits on 
child.result()
- but no worker thread is left to run either child - everything stalls ### Why this is different from tag concurrency If it were a tag concurrency limit issue, you’d expect the task to be waiting on a lease/slot. But your log:
Copy code
text
Submitting task ... to thread pool executor
means the task was already handed to the executor layer. That points more to executor starvation/deadlock than Prefect tag limits. ### Most likely causes 1. Bounded 
ThreadPoolTaskRunner(max_workers=...)
 2. Nested 
.submit()
inside tasks 3. Calling 
.result()
inside the parent task 4. Multiple parents doing this at once ### How to fix it #### Option 1: increase 
max_workers
Quickest test:
Copy code
python
from prefect import flow
from prefect.task_runners import ThreadPoolTaskRunner

@flow(task_runner=ThreadPoolTaskRunner(max_workers=20))
def my_flow():
    ...
If the problem disappears, it was almost certainly pool starvation/deadlock. #### Option 2: avoid waiting inside the parent task Instead of:
Copy code
python
@task
def parent(x):
    fut = child.submit(x)
    return fut.result()
restructure so the flow waits, not the task. #### Option 3: use direct nested call if you actually want synchronous execution If you really mean “run child inline as part of parent”, then:
Copy code
python
@task
def parent(x):
    return child(x)
That avoids executor submission for the child, though it changes behavior. #### Option 4: move orchestration to the flow level This is generally the safest pattern in Prefect. ### For your case Since you do want nested behavior, the key question is: Are you doing something like this inside the task?
Copy code
python
subtask.submit(...).result()
If yes, that is the prime suspect. ### Very short diagnosis - empty concurrency slot does not mean the task can start - the task may still be stuck waiting for a thread pool worker - nested submit + 
.result()
inside tasks is the classic cause If you paste your 
@flow
, 
@task
, and task runner config, I can tell you exactly whether it’s deadlocking and show the minimal change to fix it.
o
@Marvin can i set concurrency limit per task?
m
✅ thought for 8.1 seconds
Yes, effectively, but not as a dedicated 
@task(concurrency_limit=...)
parameter. In Prefect 3.x, per-task concurrency is usually done by giving that task a unique tag and applying a concurrency limit to that tag. ### Example
Copy code
python
from prefect import task

@task(tags=["my-write-task"])
def write_result(x):
    ...
Then create a limit for that tag:
Copy code
bash
prefect concurrency-limit create my-write-task 1
That makes 
write_result
run with concurrency 
1
as long as no other tasks use that same tag. ### Important nuance This is really: - concurrency limit per tag - not a first-class “per task object” setting So if two different tasks share the same tag, they share the same limit. If you want it to behave like per task, use a tag unique to that task. ### If you need different limits for different tasks
Copy code
python
@task(tags=["fetch-task"])
def fetch(...):
    ...

@task(tags=["write-task"])
def write(...):
    ...
Then:
Copy code
bash
prefect concurrency-limit create fetch-task 10
prefect concurrency-limit create write-task 1
### Verified CLI
Copy code
bash
prefect concurrency-limit --help
Docs: - Tasks - Run work concurrently If you want, I can also show how to set a limit that applies to only one specific task but not its subtasks or sibling tasks.
Open in Slack
PreviousNext
Powered by