Parallel.For vs ThreadPool und async/erwarten

Question

Ich versuche, 5000 Dateien asynchron zu verarbeiten, ohne den Threadpool uneingeschränkt wachsen. Die Parallel.For Schleife gibt mir jedoch keine konsistent korrekte Antwort (Anzahl kommt zu kurz), während die Task.Run ist.

Was mache ich falsch in der Parallel.For Schleife, die diese falschen Antworten verursacht?

using System; 
using System.Collections.Generic; 
using System.IO; 
using System.Linq; 
using System.Threading; 
using System.Threading.Tasks; 

class Program 
{ 
    static volatile int count = 0; 
    static volatile int total = 0; 
    static void Main(string[] args) 
    { 
     Parallel.For(0, 5000, new ParallelOptions { MaxDegreeOfParallelism = 10 }, 
      async (index) => 
      { 
       string filePath = $"C:\\temp\\files\\out{index}.txt"; 
       var bytes = await ReadFileAsync(filePath); 
       Interlocked.Add(ref total, bytes.Length); 
       Interlocked.Increment(ref count); 
      }); 
     Console.WriteLine(count); 
     Console.WriteLine(total); 

     count = 0; 
     total = 0; 
     List<Task> tasks = new List<Task>(); 
     foreach (int index in Enumerable.Range(0, 5000)) 
     { 
      tasks.Add(Task.Run(async() => 
      { 
       string filePath = $"C:\\temp\\files\\out{index}.txt"; 
       var bytes = await ReadFileAsync(filePath); 
       Interlocked.Add(ref total, bytes.Length); 
       Interlocked.Increment(ref count); 
      })); 
     } 
     Task.WhenAll(tasks).Wait(); 
     Console.WriteLine(count); 
     Console.WriteLine(total); 
    } 
    public static async Task<byte[]> ReadFileAsync(string filePath) 
    { 
     byte[] bytes = new byte[4096]; 
     using (var sourceStream = new FileStream(filePath, 
       FileMode.Open, FileAccess.Read, FileShare.Read, 
       bufferSize: 4096, useAsync: true)) 
     { 
      await sourceStream.ReadAsync(bytes, 0, 4096); 
     }; 
     return bytes; 
    } 
} 

Answer 1

Parallel.For ist nicht async bewusst.

Als solche funktioniert die Parallel.For nicht wie erwartet. Da auf die vom asynchronen Lambda erzeugte Task nicht gewartet wird, werden alle Iterationen in der Zeit abgeschlossen, die erforderlich ist, um die Tasks zu erstellen und nicht zu vervollständigen.

nach dem Parallel.For, eine Anzahl von Iterationen wird immer noch eine anhängige Aufgabe hat, die noch nicht abgeschlossen ist, und daher Ihre Ergänzungen count und total noch nicht abgeschlossen.

Stephen Toub hat eine asynchrone Version von Parallel.ForEach implementiert. (ForEachAsync) Die Umsetzung wird wie folgt:

public static Task ForEachAsync<T>(this IEnumerable<T> source, int dop, Func<T, Task> body) 
{ 
    return Task.WhenAll(
     from partition in Partitioner.Create(source).GetPartitions(dop) 
     select Task.Run(async delegate { 
      using (partition) 
       while (partition.MoveNext()) 
        await body(partition.Current); 
     })); 
} 

So könnten Sie Ihre Schleife umschreiben:

Enumerable.Range(0, 5000).ForEachAsync(10, async (index)=>{ 
    //$$$ 
});