Throttling Multiple Tasks to Process Requests in C#

• Download source - 20.3 KB

Introduction

Task Parallel Library is a powerful feature introduced by Microsoft as part of the .NET Framework. This acts as a wrapper to existing ThreadPool which is now managed by CLR optimally. This saves us from the lengthy boilerplate code of maintaining and disposing the threads. There are many articles that cover the advantages of TPL and its advantages over traditional thread pool.

Here is a very good article that gives a detailed overview of TPL.

This is a short tip that will cover a common use case while using TPL: Limiting the number of tasks that execute in parallel.

Background

We use Task Parallel Library when processing multiple requests. A service or a worker process can load a set of requests and process all of them parallelly using TPL. This is a very common application of TPL, where a windows service application polls for any new requests and processes them in parallel. In some cases, we might want to control the number of concurrent processes.

About the Code

Attached is a .NET 4.5 Console application with structural methods that make use of the extension method.

He have a method fun() that takes the processId as parameter and sleeps for some random time. This will simulate over async task.

public async Task fun(int processId)
       {
           await Task.Run( () =>{
               Random rand = new Random();
               Console.WriteLine("Processing " + processId);
               Thread.Sleep(rand.Next(1500));
               Console.WriteLine("Done processing - " + processId);
           });
       }

Extension Method

public static async Task executeParallel<T>
      (this IEnumerable<T> items, int limit, Func<T, Task> action)
     {
         var allTasks = new List<Task>(); //Store all Tasks
         var activeTasks = new List<Task>();
         foreach (var item in items)
         {
             if (activeTasks.Count >= limit)
             {
                 var completedTask = await Task.WhenAny(activeTasks);
                 activeTasks.Remove(completedTask);
             }
             var task = action(item);
             allTasks.Add(task);
             activeTasks.Add(task);
         }
         await Task.WhenAll(allTasks); // Wait for all task to complete
     }

items denote the list of input arguments for our parallel process. In other words, we are iterating over the pending requests in the queue and throttling will be handled by the logic inside the executeParallel method.

The extension method will be called as follows:

await queue.executeParallel(limit, fun); 

Main Method

The main method will take an input of number of processes. We can see the throttle in work.

static void Main(string[] args)
    {
      List<int> queue = new List<int>();
      ProcessingService service = new ProcessingService();
      int n, limit = 3 ; // any arbitrary limit     
    
        Console.WriteLine("Enter number of process to Add: ");
        n = Convert.ToInt32(Console.ReadLine());
        for(int processid = 0; processid <= n; processid++)
        {         
          queue.Add(processid);
        }
      service.process(queue,limit);

      Console.Read();
    }

Throttling in Action

We can see the next process being picked up automatically and at any time, there will be at most 3 processes running.

History

• 31^st October, 2018: Initial version