https://prefect.io logo
Join Slack
Powered by
<@ULVA73B9P> how do I prevent memory leaks on long...
# ask-marvin
j
@Marvin how do I prevent memory leaks on long-running loops with task decorator? I have these flags already set:
Copy code
cache_policy=NO_CACHE,
persist_result=False,
cache_result_in_memory=False,
m
✅ thought for 146.0 seconds
Those flags are the right first step, but they only disable result caching/persistence. They do not guarantee constant memory usage in a long-lived flow process. What to do next 1. Prefer direct task calls in the loop If you do not need parallelism, use:
Copy code
python
   @flow
   def run():
       for item in items:
           value = my_task(item)
           # use value and let it go out of scope
instead of accumulating 
.submit()
futures:
Copy code
python
   # avoid for very long loops
   futures = [my_task.submit(item) for item in items]
2. Do not retain futures or results yourself Common source of “Prefect memory leak” reports is really: - appending futures to a list - appending results to a list - keeping large objects in closures / globals If you use 
.submit()
, process in small batches:
Copy code
python
   @flow
   def run(items, batch_size=100):
       for i in range(0, len(items), batch_size):
           batch = items[i:i+batch_size]
           futures = [my_task.submit(item) for item in batch]
           results = [f.result() for f in futures]
           # handle results here
           del futures, results
3. Avoid generator-style tasks for huge streams Prefect currently tracks yielded generator results in a way that can grow over time in long-running executions. If a task yields many items, prefer chunking:
Copy code
python
   @task(cache_policy=NO_CACHE, persist_result=False, cache_result_in_memory=False)
   def process_chunk(chunk):
       for item in chunk:
           handle(item)

   @flow
   def run(all_items, chunk_size=1000):
       for i in range(0, len(all_items), chunk_size):
           process_chunk(all_items[i:i+chunk_size])
4. Keep the loop body inside a task/flow chunk if possible For truly long-running work, a common pattern is: - top-level flow orchestrates - each iteration batch runs in a shorter-lived task or subflow - process exits between scheduled runs / deployments if possible Long-lived Python processes are harder to keep flat in memory regardless of Prefect. 5. If you need isolation, use process-based execution Prefect has 
ProcessPoolTaskRunner
, which can help if the leak is in task code or libraries:
Copy code
python
   from prefect import flow, task
   from prefect.task_runners import ProcessPoolTaskRunner

   @task(cache_policy=NO_CACHE, persist_result=False, cache_result_in_memory=False)
   def my_task(x):
       return do_work(x)

   @flow(task_runner=ProcessPoolTaskRunner())
   def run(items):
       for item in items:
           my_task.submit(item)
This adds overhead, but worker processes release memory more predictably. Important nuance Even with:
Copy code
python
cache_policy=NO_CACHE,
persist_result=False,
cache_result_in_memory=False,
Prefect may still hold some memory for: - task states - futures - dependency tracking - logging/context objects - retries - generator yields So if memory still climbs, the biggest questions are: - Are you using direct calls or 
.submit()
? - Are you storing futures/results anywhere? - Are task outputs large? - Is the task a generator / yielding many items? - Is the real leak in user code / pandas / requests / DB client / model objects? My practical recommendation For the flattest memory profile, use this shape:
Copy code
python
from prefect import flow, task
from prefect.cache_policies import NO_CACHE

@task(
    cache_policy=NO_CACHE,
    persist_result=False,
    cache_result_in_memory=False,
)
def process_one(item):
    return do_work(item)

@flow
def run(items):
    for item in items:
        result = process_one(item)
        consume(result)
        del result
If you need concurrency, batch 
.submit()
calls and clear references after each batch. If you want, I can help you refactor your exact loop into a lower-memory Prefect pattern.
j
@Marvin, it seems to be resulting from a for loop (inside a task), that calls a sub-task. It is an async method and I'm doing
Copy code
for i in range(start, end):
    await func(i)
m
✅ thought for 17.5 seconds
Yep — that pattern is a likely culprit. When you 
await
a Prefect task inside another task in a long-running loop, you're still creating a task run and state for every iteration. Your caching flags stop result caching, but they do not stop Prefect from tracking task run metadata/state during that parent run. Best fix: if 
func(i)
is tiny and you're calling it thousands/millions of times, do not make 
func
a Prefect task. Make it a plain async function instead, and only put Prefect boundaries around the larger unit of work.
Copy code
python
from prefect import task

async def func(i):
    return await do_work(i)

@task
async def parent(start, end):
    for i in range(start, end):
        await func(i)
Instead of:
Copy code
python
from prefect import task

@task
async def func(i):
    return await do_work(i)

@task
async def parent(start, end):
    for i in range(start, end):
        await func(i)
Why this helps - plain async function calls do not create Prefect task runs - no per-iteration state objects/futures - much less orchestration overhead - better for hot inner loops Rule of thumb Use a Prefect task only when you want orchestration features for that call: - retries - caching - observability in the UI - concurrency control - separate execution boundary If it's just a small helper in a tight loop, keep it as a normal Python function/coroutine. Better structure Make the outer chunk the task, not each item:
Copy code
python
from prefect import flow, task

async def process_one(i):
    await do_work(i)

@task
async def process_range(start, end):
    for i in range(start, end):
        await process_one(i)

@flow
async def my_flow():
    await process_range(0, 100000)
If you need retries per item, batch them instead of one Prefect task per element:
Copy code
python
from prefect import flow, task

async def process_one(i):
    await do_work(i)

@task(retries=2)
async def process_batch(start, end):
    for i in range(start, end):
        await process_one(i)

@flow
async def my_flow():
    batch_size = 1000
    for start in range(0, 100000, batch_size):
        end = min(start + batch_size, 100000)
        await process_batch(start, end)
If you truly need each 
i
to be a Prefect task run then don't keep it in one huge parent task forever. Break it up: - submit/map in bounded batches - await the batch - drop references - move to next batch But if you're doing:
Copy code
python
for i in range(start, end):
    await func(i)
inside a task, and 
func
is also a 
@task
, then yes — I'd strongly suspect orchestration/state accumulation rather than result caching. If you want, paste a small version of your task definitions and I can show the safest rewrite.
j
@Marvin are you suggesting that running the loop from the 
@flow
method would be better?
m
✅ thought for 11.8 seconds
Yes — often, yes. But the more precise recommendation is: Put the loop at the highest level where you do not need per-iteration Prefect orchestration. So in your case, these are the usual options: Best for memory: loop in the 
@flow
, and call either: - a plain async function, or - a coarser-grained task Example:
Copy code
python
from prefect import flow, task

async def process_one(i):
    await do_work(i)

@task(
    persist_result=False,
    cache_result_in_memory=False,
)
async def process_batch(start, end):
    for i in range(start, end):
        await process_one(i)

@flow
async def my_flow():
    for start in range(0, 100000, 1000):
        end = min(start + 1000, 100000)
        await process_batch(start, end)
That is generally better than:
Copy code
python
@task
async def process_one(i):
    await do_work(i)

@task
async def parent(start, end):
    for i in range(start, end):
        await process_one(i)
because the second version creates a Prefect task run for every 
i
. Important distinction Looping in a 
@flow
is better if the thing inside the loop is not another super-fine-grained Prefect task. These patterns are usually good: 1. Loop in flow, call plain function
Copy code
python
@flow
async def my_flow():
    for i in range(start, end):
        await process_one_plain(i)
2. Loop in flow, call batch task
Copy code
python
@flow
async def my_flow():
    for batch in batches:
        await process_batch(batch)
3. Loop in task, call plain function
Copy code
python
@task
async def process_batch(start, end):
    for i in range(start, end):
        await process_one_plain(i)
This is also fine. What is usually bad for memory / overhead
Copy code
python
@task
async def process_one(i): ...

@task
async def parent():
    for i in range(...):
        await process_one(i)
because that creates lots of nested task runs and state tracking. Rule of thumb - Inner hot loop -> plain Python / plain async - Chunk or batch boundary -> Prefect task - Whole orchestration -> Prefect flow So yes, moving the loop to the 
@flow
can be better, but the real win is avoiding a Prefect task call on every iteration. If you'd like, I can show you which of these 3 rewrites fits your case best: 1. loop in flow + plain async helper 2. loop in task + plain async helper 3. flow submits batch tasks
j
@Marvin is using Prefect sub-tasks a recommended paradigm then?
m
✅ thought for 13.9 seconds
Yes — sub-tasks are absolutely a recommended Prefect pattern. But the key is: use them at the right granularity. Recommended use of sub-tasks Sub-tasks are great when each call is a meaningful orchestration unit, for example when you want: - retries - caching - separate visibility in the UI - concurrency limits - failure isolation - parallel execution with 
.submit()
/ 
.map()
Examples: - 
extract()
, 
transform()
, 
load()
- process one file - process one partition - process one API page or one customer batch Those are good task boundaries. Not recommended Using a sub-task for a very hot inner loop like:
Copy code
python
for i in range(10_000_000):
    await tiny_task(i)
That is usually too fine-grained. Even if each task does little work, Prefect still has orchestration overhead for each one: - creating task runs - tracking state - dependency bookkeeping - logging/context handling - retries metadata So sub-tasks are recommended, but not for millions of tiny calls inside a single long-lived task. Good mental model Use Prefect tasks for business/work units, not for every Python function. Good:
Copy code
python
@task
async def process_batch(batch):
    for item in batch:
        await process_one_plain(item)
Less good:
Copy code
python
@task
async def process_one(item):
    ...

@task
async def process_all(items):
    for item in items:
        await process_one(item)
unless 
process_one
is substantial enough to justify its own task run. How to choose task granularity Ask: 1. Do I need to see each iteration in the UI? 2. Do I need retries for each iteration? 3. Do I need parallelism for each iteration? 4. Is each iteration expensive enough that orchestration overhead is small relative to work? 5. Would I be comfortable with one task run per item at this scale? If the answers are mostly “no”, make it a plain function. Practical recommendation for your case If you're worried about memory in a long-running loop, I'd recommend: - keep sub-tasks, but move them up to batch-level - keep the inner per-item function plain async/python - avoid one Prefect task run per item Example:
Copy code
python
from prefect import flow, task

async def process_one(i):
    await do_work(i)

@task(retries=2, persist_result=False, cache_result_in_memory=False)
async def process_batch(start, end):
    for i in range(start, end):
        await process_one(i)

@flow
async def run():
    batch_size = 1000
    for start in range(0, 100000, batch_size):
        end = min(start + batch_size, 100000)
        await process_batch(start, end)
That still uses sub-tasks — just at a healthier level. So the short answer is: Yes, sub-tasks are recommended. No, ultra-fine-grained sub-tasks inside massive long-running loops are usually not. If you want, I can help you define a good task granularity for your specific workload.
j
@Marvin, what if I want retries w/ Prefect for each of those small sub-tasks though separately, vs. at the large level?
m
✅ thought for 15.7 seconds
Yes — that’s the tradeoff. If you want separate Prefect retries per small unit, then each small unit really does need to be its own Prefect task run. That’s valid, but you generally should not do it as one huge nested loop inside a long-lived parent task. Recommended pattern: move those small retriable units to the flow level and run them in bounded batches. Instead of:
Copy code
python
@task(retries=3)
async def process_one(i):
    await do_work(i)

@task
async def process_all(start, end):
    for i in range(start, end):
        await process_one(i)
prefer something like:
Copy code
python
from prefect import flow, task

@task(
    retries=3,
    persist_result=False,
    cache_result_in_memory=False,
)
async def process_one(i):
    await do_work(i)

@flow
async def run(start, end, batch_size=100):
    for batch_start in range(start, end, batch_size):
        batch_end = min(batch_start + batch_size, end)

        futures = []
        for i in range(batch_start, batch_end):
            futures.append(process_one.submit(i))

        for fut in futures:
            await fut.result_async()

        del futures
Why this is better - each 
i
still gets its own Prefect task run - each 
i
still gets its own retries - you avoid one giant parent task holding onto an enormous nested execution history - batching limits how many futures/states are live at once Important caveat If you do want individual retries, there is no free lunch: Prefect must track each task run somehow. So memory and orchestration overhead will always be higher than using plain functions. The goal is not to eliminate overhead — it’s to bound it. Best practices for this pattern 1. Batch submissions Don’t submit 1M tiny tasks at once. 2. Keep batch size moderate Try 
50
, 
100
, 
500
, maybe 
1000
, depending on payload size and concurrency. 3. Do not keep all futures forever Await a batch, then drop references. 4. Make each small task return little or nothing If possible:
Copy code
python
   @task(..., persist_result=False, cache_result_in_memory=False)
   async def process_one(i) -> None:
       await do_work(i)
5. Consider a concurrency limit If 
do_work
hits an external service/db, combine retries with a work pool or tag-based concurrency limit if needed. Docs: https://docs.prefect.io/ Even better than nested sub-task calls If you need per-item retries, flow -> many small tasks is generally healthier than task -> loop -> sub-task. So: Better
Copy code
python
@flow
async def run():
    for ...
        fut = process_one.submit(...)
Worse for long-running memory
Copy code
python
@task
async def parent():
    for ...
        await process_one(...)
One more design option If these units are extremely small, Prefect retries may be too expensive relative to the work. In that case, consider: - keep them as plain async calls - implement per-item retry in your own code using 
tenacity
or similar - use Prefect task retries at the batch/task boundary only That often gives better performance and lower memory while still being robust. So the answer is: If you need Prefect-managed retries per small unit, use separate task runs — but submit them from the flow in bounded batches, not from one enormous looping task. If you want, I can give you: - an async batch pattern - a sync batch pattern - or a pattern using 
.map()
and waiting per batch
j
@Marvin what if I remove the task decorator from the "main task" and keep it in the sub-tasks?
m
✅ thought for 14.2 seconds
Yes — that can be a good compromise. If your current shape is:
Copy code
python
@task
async def process_one(i):
    await do_work(i)

@task
async def main_task(start, end):
    for i in range(start, end):
        await process_one(i)
then removing 
@task
from 
main_task
and making it a plain async helper, called from a 
@flow
, is usually better than nested task-inside-task for this case. Example:
Copy code
python
from prefect import flow, task

@task(
    retries=3,
    persist_result=False,
    cache_result_in_memory=False,
)
async def process_one(i):
    await do_work(i)

async def main_loop(start, end):
    for i in range(start, end):
        await process_one(i)

@flow
async def run(start, end):
    await main_loop(start, end)
Why this is better - you remove one orchestration layer - no parent task run is tracking a huge long-lived loop - only the sub-tasks are Prefect-managed - each 
process_one(i)
can still have its own retries That said, it may still not be ideal if 
end - start
is very large. Why? Because you still create one Prefect task run per 
i
, so you still have: - per-task state tracking - futures/results/state objects - orchestration overhead So this is better than task -> sub-task loop, but bounded batching from the flow is usually better still. Better version of that pattern
Copy code
python
from prefect import flow, task

@task(
    retries=3,
    persist_result=False,
    cache_result_in_memory=False,
)
async def process_one(i):
    await do_work(i)

async def main_loop(start, end, batch_size=100):
    for batch_start in range(start, end, batch_size):
        batch_end = min(batch_start + batch_size, end)

        futures = []
        for i in range(batch_start, batch_end):
            futures.append(process_one.submit(i))

        for fut in futures:
            await fut.result_async()

        del futures

@flow
async def run(start, end):
    await main_loop(start, end)
Comparison Current pattern
Copy code
python
@task
async def main_task():
    for i in ...:
        await process_one(i)
- worst of the three for your scenario Remove decorator from main task
Copy code
python
async def main_loop():
    for i in ...:
        await process_one(i)
- better Flow + bounded submission/waiting
Copy code
python
@flow
async def run():
    for batch in ...:
        futures = [process_one.submit(i) ...]
        ...
- usually best when you want per-item Prefect retries One subtle point If you write:
Copy code
python
for i in range(start, end):
    await process_one(i)
and 
process_one
is a task, that is still sequential orchestration. It works, but it may still accumulate state over a very long run. So yes, removing the decorator from the “main task” helps, but if memory is still the concern, I’d strongly consider: - batching - awaiting each batch - deleting references before continuing So the short answer is: Yes — removing 
@task
from the main looping function is likely an improvement. But if you keep individual Prefect sub-tasks per item, batching from the flow is still the safer long-running pattern. If you want, I can rewrite your exact structure into the least-invasive version.
7 Views
Open in Slack
PreviousNext
Powered by