Fan-out/Fan-in in Durable Functions

1. Understanding the Fan-Out/Fan-In Pattern

Imagine you need to process a hundred images to generate a thumbnail for each. With function chaining, you would process them one by one, which would be slow. A much more efficient approach is to start all 100 thumbnail generation tasks at the same time and then, once they are all complete, gather the results. This is the essence of the fan-out/fan-in pattern.

• Fan-Out: The orchestrator function starts multiple activity functions in parallel.
• Fan-In: The orchestrator waits for all the parallel activities to finish.
• Aggregate: After the "fan-in," the orchestrator can aggregate the results from all the completed activities.

This pattern should feel conceptually familiar given your experience with concurrent programming in .NET. The key difference here is that the Durable Functions framework manages the distribution and reliability of these parallel tasks, which could be running across many different virtual machines.

To solidify your understanding, please read the following introduction to the pattern.

2. Implementing the Fan-Out/Fan-In Orchestrator

Now, let's implement this pattern. We will modify the project you created in the last lesson. Our goal is to create an orchestration that:

1. Gets a list of work items (e.g., a list of document IDs to process).
2. "Fans out" by starting a parallel activity function for each document.
3. "Fans in" by waiting for all processing activities to complete.
4. Aggregates the results (e.g., calculates the total number of words processed).

The core of this logic lies in how we call the activities from the orchestrator. Instead of await-ing each call, we launch them all and store their Task objects in a list. Then, we use Task.WhenAll to wait for all of them to complete.

Let's look at the C# code that enables this.

3. Hands-On: Building the Orchestration

Now, let's apply this in your project. You can add a new file or modify the one from the previous lesson. We'll create three new functions: an orchestrator, and two activities.

Step 1: Create the Activity Functions

First, let's create the activity functions that will do the work. The first gets our list of items, and the second processes a single item.

// In a new C# file (e.g., FanOutFanIn.cs)

[Function(nameof(GetWorkItems))]
public static List<string> GetWorkItems([ActivityTrigger] object input, FunctionContext context)
{
    var logger = context.GetLogger(nameof(GetWorkItems));
    logger.LogInformation("Getting work items...");
    // In a real app, this would come from a database or blob storage.
    return new List<string>
    {
        "Report-Alpha.docx",
        "Budget-2024-Final.xlsx",
        "Project-Phoenix-Spec.pdf",
        "Meeting-Notes-Q3.txt"
    };
}

[Function(nameof(ProcessWorkItem))]
public static async Task<int> ProcessWorkItem([ActivityTrigger] string documentName, FunctionContext context)
{
    var logger = context.GetLogger(nameof(ProcessWorkItem));
    logger.LogInformation("Processing document: {documentName}", documentName);

    // Simulate work, like counting words or processing data.
    await Task.Delay(TimeSpan.FromSeconds(5)); 
    
    // Return a result, e.g., the number of characters in the name.
    return documentName.Length;
}

Step 2: Create the Orchestrator Function

This is the most important part. The orchestrator will call GetWorkItems once, then loop through the results to "fan out" calls to ProcessWorkItem.

// In the same C# file

[Function(nameof(FanOutFanInOrchestrator))]
public static async Task<long> FanOutFanInOrchestrator(
    [OrchestrationTrigger] TaskOrchestrationContext context)
{
    var logger = context.CreateReplaySafeLogger(nameof(FanOutFanInOrchestrator));
    logger.LogInformation("Orchestration started.");

    // 1. Get the list of work to be done.
    var workItems = await context.CallActivityAsync<List<string>>(nameof(GetWorkItems), null);

    // 2. Fan-Out: Start all processing tasks in parallel.
    logger.LogInformation("Fanning out to {count} activities.", workItems.Count);
    var processingTasks = new List<Task<int>>();
    foreach (var workItem in workItems)
    {
        // We DON'T await here. We add the task to a list.
        Task<int> task = context.CallActivityAsync<int>(nameof(ProcessWorkItem), workItem);
        processingTasks.Add(task);
    }

    // 3. Fan-In: Wait for all the parallel tasks to complete.
    await Task.WhenAll(processingTasks);
    logger.LogInformation("All activities completed (fanned in).");

    // 4. Aggregate the results.
    int totalChars = processingTasks.Sum(task => task.Result);
    logger.LogInformation("Aggregation complete. Total characters: {total}", totalChars);
    
    return totalChars;
}

Step 3: Create an HTTP Starter

Finally, create a new HTTP-triggered client function to start this specific orchestration.

// In the same C# file

[Function("FanOutFanIn_HttpStart")]
public static async Task<HttpResponseData> FanOutFanIn_HttpStart(
    [HttpTrigger(AuthorizationLevel.Anonymous, "get", "post")] HttpRequestData req,
    [DurableClient] DurableTaskClient client,
    FunctionContext executionContext)
{
    var logger = executionContext.GetLogger("FanOutFanIn_HttpStart");

    // Function input comes from the request content.
    string instanceId = await client.ScheduleNewOrchestrationInstanceAsync(
        nameof(FanOutFanInOrchestrator));

    logger.LogInformation("Started orchestration with ID = '{instanceId}'.", instanceId);

    // Returns an HTTP 202 response with location header that contains
    // the URI of the management API to query the status of the orchestration.
    return client.CreateCheckStatusResponse(req, instanceId);
}

Step 4: Run and Test
Run your project (F5 in Visual Studio). Use Postman or a browser to send a POST request to the FanOutFanIn_HttpStart endpoint.

1. You will get back a response with a statusQueryGetUri.
2. If you query this URI immediately, you'll see the status is "Running".
3. Wait a few seconds. Notice in the function console that all "Processing document..." logs appear almost simultaneously. This is the fan-out in action!
4. After about 5 seconds (the delay in our activity), query the status URI again. The status should now be "Completed", and the output field should show the sum of the lengths of all the document names.

4. Real-World Considerations

The basic pattern is powerful, but in production environments, you need to consider a few more things, like controlling the level of parallelism and handling errors.

Let's watch a video that uses a different real-world scenario to demonstrate the fan-out/fan-in pattern and discusses some of these advanced concepts.

As you build more complex systems, two topics become critical:

• Controlling Parallelism: What if GetWorkItems returned 10,000 items? Fanning out to 10,000 parallel activities could overwhelm your function app or downstream services. A common solution is to process items in batches.
• Error Handling: What if one of the 10,000 activities fails? Task.WhenAll will throw an exception as soon as one task fails. You need a strategy for handling partial failures.

The following blog post provides excellent code examples for these advanced scenarios.

Conclusion

Excellent work! You've now implemented one of the most common and powerful patterns in distributed computing. By dispatching work in parallel, you can build highly scalable and performant serverless applications.

Key Takeaways:

• The fan-out/fan-in pattern is used to execute multiple independent tasks concurrently.
• Fan-out is achieved by calling activity functions in a loop without await and collecting the Task objects in a list.
• Fan-in is achieved by using await Task.WhenAll() to wait for every task in the list to complete.
• After the fan-in, you can aggregate the results from the completed tasks.
• For production systems, consider controlling the degree of parallelism with batching and implementing robust error handling and retry policies.

Next Steps

So far, we've seen orchestrations that run to completion relatively quickly. But what about very long-running processes that might take hours or even days? How does an external client start such a process and get the result later? In our next lesson, we will explore the async HTTP API pattern, which is designed specifically for this scenario.