The TDLR; on easily managing .NET concurrent asynchronous threads

Ever had the requirement for your code to perform some series of background actions, say to query an API for weather data for various locations, but your provider caps you at having only 2 max concurrent connections at any given time? If so, then you’ve already gone through the wealth of information found online for a solution. I’ve seen quite a few myself and yes, they all work just fine!

For me, I like to keep things as simple as possible… so with that in mind and knowing that you’re most likely here looking for quick code samples, here is my TDLR; for handling this type of scenario:

//Currently only allowing 2 concurrent threads to run at any given time var threadBroker = new SemaphoreSlim(2);

Task.WhenAll(mockActions.Select(iAction => Task.Run(() => { try { threadBroker.WaitAsync().Wait(); iAction(); } finally { threadBroker.Release(); } }))).Wait();

NOTE: You'll want to implement your own exception handling and logging inside your try/finally block ,which I have left out for brevity.

Now before we actually get into what's going on, here's the above wrapped into a nice little extension method...

public static class AsyncExtensions { public static void ExecuteAsync(this IEnumerable actions, int maxCurrentThreads = 1) { var threadBroker = new SemaphoreSlim(maxCurrentThreads);
Task.WhenAll(actions.Select(iAction => Task.Run(() => { try { threadBroker.WaitAsync().Wait(); iAction(); } finally { threadBroker.Release(); } }))).Wait(); } }

Then, using the same mockActions object above, here's is how we consume the extension method:

mockActions.ExecuteAsync(2);

This really simplifies most of the more common concurrent asynchronous threading tasks that I run into on a day to day basis...

So what's going on here?

The key component of what we're utilizing here is SemaphoreSlim, which allows for managing the total number of concurrent threads that we want to have executing at any given time.

This is ideal for situations where the thread wait times are "expected to be very short" (that's Microsoft's official documentation). The reason for this is because SemaphoreSlim implements something called SpinWait, which is fast however keeps the CPU active while waiting for a thread resource to free up.

In situations where the wait times might be on the longer side, you can use the SemaphoreSlim's big brother, Semaphore. Semaphore isn't quite as fast as it's little brother however is a bit more robust.

Now let's look at a more realistic example utilizing our extension method while sticking with our weather API scenario:

var locker = new object(); var results = new List<string>();

var zipCodes = new[] { "1111111", "2222222", "3333333", "4444444", "5555555" };

zipCodes.Aggregate(new List(), (list, zip) => { list.Add(() => { var jsonResult = string.Format("Returned weather Json for zip {0}", zip); //Mock API call for the zip code's weather

    //Must ensure only one thread at a time can access "results" list
    lock (locker)
    {
        results.Add(jsonResult);
    }
});

return list;

}).ExecuteAsync(2);

//Now simply do something with your "results" weather data

posted on Monday, June 12, 2017 4:26 PM Print

New on Geeks with Blogs