"Infinite" asynchronous parallel foreach loop

user_12753809 0
Viewed 482

I have a List<string> containing 50K to 100K words

I would like to iterate through it in a parallel and asynchronous fashion

As an example, I could use

while (true)
{
   Parallel.ForEach(words, new ParallelOptions { MaxDegreeOfParallelism = 100 }, ...)
}

But problems are:

  1. Parallel.ForEach isn't asynchronous
  2. When we arrive at the end of the list, we have to wait for each thread to end before the while (true) statement continues
  3. Which means that there aren't always 100 threads running, which is what I want

How would I be able to achieve this?

Please let me know if this is confusing, or if I'm bad at explaining things.

async-await parallel-processing asynchronous c#
1 Answers

Here is a totally contrived async friendly TPL DataFlow example of how you could achieve what you ask.

  1. It's suitable for an async IO workload
  2. It's cancellable
  3. It limits max parallelism
  4. It has bounded capacity so there is always 100 jobs available
  5. It's infinite

Given

private static CancellationTokenSource _cs;
private static CancellationToken _token;
private static ActionBlock<string> _block;

private static async Task MethodAsync(string something)
{
   // Your async workload
}

public static async Task EndlessRunner(string[] someArray)
{
   try
   {
      var index = 0;
      while (!_token.IsCancellationRequested)
      {
         await _block.SendAsync(someArray[index],_token);
         if (++index >= someArray.Length) index = 0;
      }
   }
   catch (OperationCanceledException)
   {
      Console.WriteLine("Cancelled");
   }
}

Example

private static async Task Main()
{
   _cs = new CancellationTokenSource();
   _token = _cs.Token;

   _block = new ActionBlock<string>(
      MethodAsync, 
      new ExecutionDataflowBlockOptions()
      {
         EnsureOrdered = false,
         MaxDegreeOfParallelism = 100,
         BoundedCapacity = 100,
         CancellationToken = _cs.Token,
         SingleProducerConstrained = true
      });

   var someList = Enumerable
      .Range(0,5000)
      .Select(I => $"something {I}")
      .ToArray();

   Task.Run(() => EndlessRunner(someList));

   Console.ReadKey();

   _cs.Cancel();

   _block.Complete();
   await _block.Completion;

}
user_16129750
Related