New ConcurrentForEach library

Author: ChrisLambrou

Source/Redgate.MicroLibraries.sln

@@ -15,6 +15,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "ULibs.FullExceptionString",
 EndProject
 Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ULibs.TinyJsonDeser", "ULibs.TinyJsonDeser\ULibs.TinyJsonDeser.csproj", "{F46791E7-5434-400B-B5B4-6659909C262E}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "ULibs.ConcurrentForEach", "ULibs.ConcurrentForEach\ULibs.ConcurrentForEach.csproj", "{B9950135-5A39-4B4C-8A61-D6105CA03396}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -45,6 +47,10 @@ Global
 		{F46791E7-5434-400B-B5B4-6659909C262E}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{F46791E7-5434-400B-B5B4-6659909C262E}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{F46791E7-5434-400B-B5B4-6659909C262E}.Release|Any CPU.Build.0 = Release|Any CPU
+		{B9950135-5A39-4B4C-8A61-D6105CA03396}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{B9950135-5A39-4B4C-8A61-D6105CA03396}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{B9950135-5A39-4B4C-8A61-D6105CA03396}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{B9950135-5A39-4B4C-8A61-D6105CA03396}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE

Source/ULibs.ConcurrentForEach/AssemblyInfo.cs

@@ -0,0 +1,3 @@
+using System.Runtime.CompilerServices;
+
+[assembly: InternalsVisibleTo("Ulibs.Tests")]

Source/ULibs.ConcurrentForEach/Concurrent.cs

@@ -0,0 +1,137 @@
+using System;
+using System.Collections.Generic;
+/***using System.Diagnostics.CodeAnalysis;***/
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace /***$rootnamespace$.***/ULibs.ConcurrentForEach
+{
+    /***[ExcludeFromCodeCoverage]***/
+    internal static class Concurrent
+    {
+        /// <summary>
+        /// Applies an asynchronous operation to each element in a sequence.
+        /// </summary>
+        /// <typeparam name="T">The type of elements in the sequence.</typeparam>
+        /// <param name="source">The input sequence of elements.</param>
+        /// <param name="func">The asynchronous operation applied to each element in the sequence. It's strongly
+        /// recommended that the operation should take care to handle its own expected exceptions.</param>
+        /// <param name="maxConcurrentTasks">
+        /// <para>
+        /// The maximum number of concurrent operations. Must be greater than or equal to 1.
+        /// </para>
+        /// <para>
+        /// The number of simultaneous operations can vary depending on the use case. For cpu intensive
+        /// operations, consider using <see cref="Environment.ProcessorCount">Environment.ProcessorCount</see>.
+        /// For operations that invoke the same web service for each item, RFC 7230 suggests that the number
+        /// of simultaneous requests/connections should be limited (https://tools.ietf.org/html/rfc7230#section-6.4).
+        /// A search for the connection limits used by common web-browsers suggests that a value in the range 6-8 is
+        /// appropriate (any more, and you risk triggering abuse detection mechanisms). For operations that invoke a
+        /// different web service for each item, a search for the connection limits used by common web-browsers
+        /// suggests that a value in the range 10-20 is appropriate.
+        /// </para>
+        /// </param>
+        /// <param name="cancellationToken">Used to cancel the operations.</param>
+        /// <returns>A task that can be awaited upon for all operations to complete. Awaiting on the task will
+        /// raise an <see cref="AggregateException"/> if any operation fails, or work is cancelled via the
+        /// <paramref name="cancellationToken"/>.</returns>
+        public static Task ForEach<T>(
+            this IEnumerable<T> source,
+            Func<T, Task> func,
+            int maxConcurrentTasks,
+            CancellationToken cancellationToken)
+        {
+            if (func == null) throw new ArgumentNullException(nameof(func));
+            return source.ForEach((item, _) => func(item), maxConcurrentTasks, cancellationToken);
+        }
+
+        /// <summary>
+        /// Applies an asynchronous operation to each element in a sequence.
+        /// </summary>
+        /// <typeparam name="T">The type of elements in the sequence.</typeparam>
+        /// <param name="source">The input sequence of elements.</param>
+        /// <param name="func">The asynchronous operation applied to each element in the sequence. It's strongly
+        /// recommended that the operation should take care to handle its own expected exceptions.</param>
+        /// <param name="maxConcurrentTasks">
+        /// <para>
+        /// The maximum number of concurrent operations. Must be greater than or equal to 1.
+        /// </para>
+        /// <para>
+        /// The number of simultaneous operations can vary depending on the use case. For cpu intensive
+        /// operations, consider using <see cref="Environment.ProcessorCount">Environment.ProcessorCount</see>.
+        /// For operations that invoke the same web service for each item, RFC 7230 suggests that the number
+        /// of simultaneous requests/connections should be limited (https://tools.ietf.org/html/rfc7230#section-6.4).
+        /// A search for the connection limits used by common web-browsers suggests that a value in the range 6-8 is
+        /// appropriate (any more, and you risk triggering abuse detection mechanisms). For operations that invoke a
+        /// different web service for each item, a search for the connection limits used by common web-browsers
+        /// suggests that a value in the range 10-20 is appropriate.
+        /// </para>
+        /// </param>
+        /// <param name="cancellationToken">Used to cancel the operations.</param>
+        /// <returns>A task that can be awaited upon for all operations to complete. Awaiting on the task will
+        /// raise an <see cref="AggregateException"/> if any operation fails, or work is cancelled via the
+        /// <paramref name="cancellationToken"/>.</returns>
+        public static async Task ForEach<T>(
+            this IEnumerable<T> source,
+            Func<T, CancellationToken, Task> func,
+            int maxConcurrentTasks,
+            CancellationToken cancellationToken)
+        {
+            if (maxConcurrentTasks < 1)
+                throw new ArgumentException("Value cannot be less than 1", nameof(maxConcurrentTasks));
+            if (source == null) throw new ArgumentNullException(nameof(source));
+            if (func == null) throw new ArgumentNullException(nameof(func));
+
+            using (var semaphore = new SemaphoreSlim(maxConcurrentTasks, maxConcurrentTasks))
+            {
+                var tasks = new List<Task>();
+                foreach (var item in source)
+                {
+                    // Wait for the next available slot.
+                    try
+                    {
+                        await semaphore.WaitAsync(cancellationToken);
+                    }
+                    catch (OperationCanceledException exception)
+                    {
+                        tasks.Add(Task.FromException(exception));
+                        break;
+                    }
+
+                    // Discard completed tasks. Not strictly necessary, but keeps the list size down.
+                    tasks.RemoveAll(task => task.IsCompleted);
+
+                    // Kick-off the next task.
+                    tasks.Add(CreateTask(func, item, cancellationToken).ReleaseSemaphoreOnCompletion(semaphore));
+                }
+
+                await Task.WhenAll(tasks);
+            }
+        }
+
+        private static Task CreateTask<T>(
+            Func<T, CancellationToken, Task> func, T item, CancellationToken cancellationToken)
+        {
+            try
+            {
+                return func(item, cancellationToken);
+            }
+            catch (Exception exception)
+            {
+                return Task.FromException(exception);
+            }
+        }
+
+        private static async Task ReleaseSemaphoreOnCompletion(this Task task, SemaphoreSlim semaphore)
+        {
+            try
+            {
+                await task;
+            }
+            finally
+            {
+                semaphore.Release();
+            }
+        }
+    }
+}

Source/ULibs.ConcurrentForEach/README.md

@@ -0,0 +1 @@
+Provides a way to asynchronously apply an operation to each element in a sequence, whilst limiting the maximum number of concurrent operations.

Source/ULibs.ConcurrentForEach/RELEASENOTES.md

@@ -0,0 +1,7 @@
+# ULibs.ConcurrentForEach release notes
+
+## 1.0.0
+
+### Features
+
+- New `ForEach` extension method for `IEnumerable<T>` that applies an asynchronous operation to each element, whilst limiting the maximum number of concurrent operations.

Source/ULibs.ConcurrentForEach/ULibs.ConcurrentForEach.csproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+  <PropertyGroup>
+    <TargetFramework>netstandard2.0</TargetFramework>
+    <Copyright>2018</Copyright>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Microsoft.DotNet.Analyzers.Compatibility" Version="0.2.12-alpha" />
+  </ItemGroup>
+
+</Project>

Source/ULibs.ConcurrentForEach/ULibs.ConcurrentForEach.nuspec

@@ -0,0 +1,19 @@
+<?xml version="1.0"?>
+<!-- http://docs.nuget.org/docs/reference/nuspec-reference -->
+<package >
+  <metadata>
+    <id>RedGate.ULibs.ConcurrentForEach.Sources</id>
+    <version>$version$</version>
+    <authors>red-gate</authors>
+    <owners>red-gate</owners>
+    <requireLicenseAcceptance>false</requireLicenseAcceptance>
+    <summary><![CDATA[$summary$]]></summary>
+    <description><![CDATA[$description$]]></description>
+    <releaseNotes><![CDATA[$releaseNotes$]]></releaseNotes>
+    <copyright>Copyright $copyrightYear$ Red Gate Software Ltd</copyright>
+    <projectUrl>https://github.com/red-gate/MicroLibraries</projectUrl>
+  </metadata>
+  <files>
+    <file src="*.pp" target="content\App_Packages\RedGate.ULibs.ConcurrentForEach.$version$"/>
+  </files>
+</package>

Source/Ulibs.Tests/ConcurrentForEach/ConcurrentTests.cs

@@ -0,0 +1,168 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using NUnit.Framework;
+using ULibs.ConcurrentForEach;
+
+namespace Ulibs.Tests.ConcurrentForEach
+{
+    [TestFixture]
+    public class ConcurrentTests
+    {
+        private readonly Random _random = new Random();
+
+        private int GetRandomDelay()
+        {
+            lock (_random)
+            {
+                return 50 + _random.Next(50);
+            }
+        }
+
+        [Test]
+        public void ForEach_ArgChecks()
+        {
+            IEnumerable<int> sequence = Enumerable.Range(0, 10);
+            IEnumerable<int> nullSequence = null;
+
+            Func<int, Task> operation = i => Task.CompletedTask;
+            Func<int, Task> nullOperation = null;
+
+            Func<int, CancellationToken, Task> cancellableOperation = (i, c) => Task.CompletedTask;
+            Func<int, CancellationToken, Task> nullCancellableOperation = null;
+
+            Assert.That(() => sequence.ForEach(nullOperation, 1, CancellationToken.None), Throws.ArgumentNullException);
+            Assert.That(() => sequence.ForEach(nullCancellableOperation, 1, CancellationToken.None), Throws.ArgumentNullException);
+            Assert.That(() => nullSequence.ForEach(operation, 1, CancellationToken.None), Throws.ArgumentNullException);
+            Assert.That(() => nullSequence.ForEach(cancellableOperation, 1, CancellationToken.None), Throws.ArgumentNullException);
+            Assert.That(() => sequence.ForEach(operation, 0, CancellationToken.None), Throws.ArgumentException);
+            Assert.That(() => sequence.ForEach(cancellableOperation, 0, CancellationToken.None), Throws.ArgumentException);
+        }
+
+        [Test]
+        public void ForEach_ExecutesTheOperationAgainstEachElement()
+        {
+            // ARRANGE
+            var count = 0;
+            Task Operation(int value) => Task.Run(() => Interlocked.Increment(ref count));
+            var sequence = Enumerable.Range(0, 10);
+
+            // ACT
+            var task = sequence.ForEach(Operation, 2, CancellationToken.None);
+
+            // ASSERT
+            Assert.That(task.Wait(2000), Is.True);
+            Assert.That(count, Is.EqualTo(10));
+        }
+
+        [Test]
+        public void ForEach_LimitsTheNumberOfSimultaneousOperations()
+        {
+            // ARRANGE
+            var currentOperationCount = 0;
+            var maxOperationCount = 0;
+            var monitor = new object();
+            async Task Operation(int value)
+            {
+                lock (monitor)
+                {
+                    currentOperationCount++;
+                    maxOperationCount = Math.Max(maxOperationCount, currentOperationCount);
+                }
+                await Task.Delay(GetRandomDelay());
+                lock (monitor)
+                {
+                    currentOperationCount--;
+                }
+            }
+            var sequence = Enumerable.Range(0, 10);
+
+            // ACT
+            var task = sequence.ForEach(Operation, 2, CancellationToken.None);
+
+            // ASSERT
+            Assert.That(task.Wait(2000), Is.True);
+            Assert.That(maxOperationCount, Is.EqualTo(2));
+        }
+
+        [Test]
+        public void ForEach_ExecutesTheOperationAgainstAllElements_InSpiteOfExceptions()
+        {
+            // ARRANGE
+            var count = 0;
+
+            async Task Operation(int value)
+            {
+                Interlocked.Increment(ref count);
+                if (value % 5 == 0)
+                {
+                    // Sometimes the func should fail immediately.
+                    throw new ApplicationException("Error");
+                }
+                await Task.Delay(GetRandomDelay());
+                if (value % 2 == 0)
+                {
+                    // Sometimes the func should fail after the first await.
+                    throw new ApplicationException("Error");
+                }
+            }
+            var sequence = Enumerable.Range(0, 10);
+
+            // ACT
+            var task = sequence.ForEach(Operation, 2, CancellationToken.None);
+
+            // ASSERT
+            Assert.That(() => Assert.That(task.Wait(2000), Is.True),
+                        Throws.InstanceOf<AggregateException>());
+            Assert.That(count, Is.EqualTo(10));
+        }
+
+        [Test]
+        public void ForEach_AbortsWhenTheCancellationTokenIsCancelled_WhereEachTaskReceivesTheCancellationSignal()
+        {
+            // ARRANGE
+            var count = 0;
+            Task Operation(int value, CancellationToken token) => Task.Run(async () =>
+            {
+                Interlocked.Increment(ref count);
+                await Task.Delay(GetRandomDelay(), token);
+            });
+            var sequence = Enumerable.Range(0, 10);
+            var cancellationTokenSource = new CancellationTokenSource(200);
+            var cancellationToken = cancellationTokenSource.Token;
+
+            // ACT
+            var task = sequence.ForEach(Operation, 2, cancellationToken);
+
+            // ASSERT
+            Assert.That(() => Assert.That(task.Wait(2000), Is.True),
+                        Throws.InstanceOf<AggregateException>());
+            Assert.That(count, Is.GreaterThan(0).And.LessThan(10));
+        }
+
+        [Test]
+        public void ForEach_AbortsWhenTheCancellationTokenIsCancelled_WhereNoTaskReceivesTheCancellationSignal()
+        {
+            // ARRANGE
+            var count = 0;
+            Task Operation(int value) => Task.Run(async () =>
+            {
+                Interlocked.Increment(ref count);
+                await Task.Delay(GetRandomDelay());
+            });
+            var sequence = Enumerable.Range(0, 10);
+            var cancellationTokenSource = new CancellationTokenSource(200);
+            var cancellationToken = cancellationTokenSource.Token;
+
+            // ACT
+            var task = sequence.ForEach(Operation, 2, cancellationToken);
+
+            // ASSERT
+            Assert.That(() => Assert.That(task.Wait(2000), Is.True),
+                        Throws.InstanceOf<AggregateException>());
+            Assert.That(count, Is.GreaterThan(0).And.LessThan(10));
+        }
+    }
+}