Then you could use https://docs.prefect.io/concepts/results/#caching-of-results-in-memory

I’m not sure what the implications are though. It’s definitely risky to clear it out.

Here's what our flow looks like:

@flow(
    name="Multi Greeter",
    task_runner=SequentialTaskRunner(),
)
async def multi_greeter():
    names = [
        "Kelvin",
        "Zanie",
        "Marvin",
    ] * 10 # Batch size of 30

    for batch in range(1000): # Approx number of batches
        # First set of mapped tasks
        greet_states = await greet.map(name=names, return_state=True)  # type: ignore
        greetings = [state.result() for state in greet_states if state.is_completed()]

        # Second set of mapped tasks that take previous results as input
        somethings_states = await get_something_from_greetings.map(message=greetings, return_state=True)  # type: ignore
        somethings = [state.result() for state in somethings_states if state.is_completed()]

        task_that_saves_to_database(somethings, return_state=True)

greet

,

get_seomthing_from_greeting~s~

, and

task_that_saves_to_database

, are all configured to use

cache_result_in_memory=False

,

I don’t think

cache_result_in_memory

accounts for the task runner result passing mechanism though if you’re using the sequential one

There’s an example in the doc I linked for that case.

One option would be to manage your results manually, i.e. put them into a global dictionary and clear them as needed — do not return any data from the task itself.

That’s kind of meh but if you’re both disk and memory constrained it’s a little tricky for us to know when it’s okay to release the result

We can’t know if you’ll use the task downstream since the DAG is dynamic

The global dictionary makes a lot of sense

I think I'll refactor one flow and see what the performance is like

Hey @Zanie I took your advice and refactored some of the flows to use

global

variables for responses. Essentially, the global variables are reset to empty lists / dicts whenever a new batch of mapped tasks begin. I"m also invoking a method that resets the

_results

dict on the task runner. The memory usage growth is much less than before, but there is still a steady increase. Flows started at an avg mem consumption of 10% and after 14 hours they're at 25%. I feel like I'm missing something. The barebones python code shouldn't see such linear increases in memory usage over time. I'd really appreciate any help with this matter! Thank you again for your guidance

Is there a way to disable in-memory tracking of task run states or maybe have it only persist to a database / secondary memory?

Can you put groups of tasks in subflows? If you return a value from the subflow (i.e.

True

) then we should release all those task states from memory.

Will keep you posted on the update

I'm wondering if specifying

task_runner=SequentialTaskRunner()

for the

batch_subflow

flow would change anything. It is still creating a new task runner instance for each call to

batch_subflow

right?