Skip to content

Instantly share code, notes, and snippets.

@atifaziz
Last active March 1, 2023 18:39
Show Gist options
  • Save atifaziz/b0b6db4ffda6423362f482318fccddd5 to your computer and use it in GitHub Desktop.
Save atifaziz/b0b6db4ffda6423362f482318fccddd5 to your computer and use it in GitHub Desktop.
Extension to run tasks concurrently and return an array when all have completed
#region Copyright (c) Microsoft. All rights reserved.
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
// SOFTWARE
//
#endregion
static partial class Extensions
{
static readonly Task IncompleteTask = new TaskCompletionSource().Task;
public static async Task<(T Item, Task<TResult> Task)[]>
ToCompletedTaskArrayAsync<T, TResult>(
this IEnumerable<T> source,
Func<T, CancellationToken, Task<TResult>> function,
int concurrency, // 0 = unbounded
CancellationToken cancellationToken)
{
ArgumentNullException.ThrowIfNull(source, nameof(source));
ArgumentNullException.ThrowIfNull(function, nameof(function));
if (concurrency < 0) throw new ArgumentOutOfRangeException(nameof(concurrency), concurrency, null);
using var concurrencySemaphore = concurrency > 0 ? new SemaphoreSlim(concurrency) : null;
using var completionSemaphore = new SemaphoreSlim(0, int.MaxValue);
var pendingTaskList = new List<Task<(int, (T, Task<TResult>))>?>(source is ICollection { Count: var count } ? count : 4);
var i = 0;
foreach (var item in source)
pendingTaskList.Add(RunAsync(i++, item));
var results = new (T, Task<TResult>)[pendingTaskList.Count];
var si = 0;
for (var pendingCount = pendingTaskList.Count; pendingCount > 0; pendingCount--)
{
await completionSemaphore.WaitAsync().ConfigureAwait(false);
while (true)
{
i = pendingTaskList.FindIndex(si, t => t is { IsCompleted: true });
if (i >= 0)
break;
si = 0;
}
var completedTask = pendingTaskList[i]!;
pendingTaskList[i] = null;
if (i + 1 == pendingTaskList.Count)
{
var rsi = pendingTaskList.Count;
while (rsi > 0 && pendingTaskList[rsi - 1] is null)
rsi--;
pendingTaskList.RemoveRange(rsi, pendingTaskList.Count - rsi);
si = 0;
}
else
{
si = i + 1;
}
(i, var result) = await completedTask.ConfigureAwait(false);
results[i] = result;
}
return results;
async Task<(int, (T, Task<TResult>))> RunAsync(int i, T item)
{
try
{
if (concurrencySemaphore is not null)
await concurrencySemaphore.WaitAsync().ConfigureAwait(false);
var task = function(item, cancellationToken);
_ = await Task.WhenAny(task, IncompleteTask).ConfigureAwait(false);
return (i, (item, task));
}
finally
{
if (concurrencySemaphore is not null)
concurrencySemaphore.Release();
completionSemaphore.Release();
}
}
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment