DurableFunction generating n tasks inside function

[eddynaka]

Hello,

I have a question related to how the best way is to do the following:
imagine a durable function that receives a zip and in the zip you have multiple files.
From that, the durable function calls a new function to unzip and process each file.

Today, this function that is processing those files are opening one thread for each file and, once all finishes, I'm merging their results.

With that, if I receive a zip with 1k files, I will start 1k tasks.

Below, one example:

var tasks = new List<Task<ObjectX>>();
foreach (string path in paths)
{
   tasks.Add(Task.Factory.StartNew(() => someProcessing));
}

await Task.WhenAll(tasks);

Thank you!

[ConnorMcMahon]

You would essentially just put your someProcessing as an Activity Function. These activity functions could then be executed in parallel across the various VMs your function application is scaled out to. In addition, the need to repeat this processing work if your function crashes becomes more minimal because the results of each step will be stored persistently.

The code would look almost exactly the same, but you would replace Task.Factory.StartNew(() => someProcessing) with context.CallActivityAsync(nameof(SomeProcessingActivity), path).

This is a very common pattern that we call fan-out/fan-in.

[eddynaka]

I see.

The issue in this approach is that after processing the file I have to generate a "report".
With that, I have to wait for all threads.

If I do fan-out/fan-in, I wouldn't have control of that.

How does AzFunctions interprets ProcessorCount / Threads / etc?
Is there a limitation? Can I use ProcessorCount to restrict it?

[ConnorMcMahon]

You absolutely can do that with the fan-out/fan-in pattern. Running Task.WhenAll() on all of the activity tasks will wait until every activity is finished. You can then have a follow up activity that generates the report now that all of the files are completed processing. You can even have each activity return the metadata necessary to generate the report.

In general, you are just replacing Threads with activities, and we will handle the scaling and everything behind the scenes. You should not be thinking about threading in your orchestration logic in Durable Functions.

[eddynaka]

just understanding the design...

1. Orchestrator receives the call and invokes await starterClient.StartNewAsync("funcName1", obj1)
2. funcName1 dispatches tasks calling context.CallActivityAsync("funcName2", obj2) for each file (imagine 100)
3. When all get back, I can merge the results.

when I call (2), does that mean that it can use a different VM? or is it going to be always in the same VM/function?

The reason why I'm asking: my current function do the following:

1. unzips the file
2. creates a context (this is used for caching + properties) and start dispatching threads
3. wait for all and merge results

If (2) generates a request outside the context, the file won't be there to be analyzed, for example.

[ConnorMcMahon]

So in general, if you wanted the benefits of Durable Functions, you would need to refactor your file processing code to not have shared context between the different activities, as activities are not guaranteed to execute on the same process. This is because they can be scaled to other VMs, or can start executing after a process crash kills the process that scheduled it. Because of that requirement, all of the context you want to persist needs to be serializable.

The idea in Durable Functions is that you spread your work across separate stateless activity functions and use the orchestration function to coordinate those stateless activity functions. Therefore, all of the information for the activities need to be passed around as serializable inputs/outputs. In your case, this would mean passing around references to the files and reloading the files in each activity. This may result in additional work beyond having it stored in a single function, but that is in exchange for removing the need for duplicate work in the case of a process crash and for the ability to scale your work items across many VMs.