Come eseguire correttamente più attività asincrone in parallelo?

Question

Cosa fare se è necessario eseguire più attività di I/O asincrone in parallelo, ma è necessario assicurarsi che non siano in esecuzione più di X processi I/O contemporaneamente; e le attività di elaborazione degli I/O pre e post non dovrebbero avere tale limitazione.

Ecco uno scenario: diciamo che ci sono 1000 attività; ognuno di essi accetta una stringa di testo come parametro di input; trasforma quel testo (elaborazione pre I/O) quindi scrive il testo trasformato in un file. L'obiettivo è di fare in modo che la logica di pre-elaborazione utilizzi il 100% di CPU/core e parte I/O delle attività eseguite con un massimo di 10 gradi di parallelismo (massimo 10 simultaneamente aperto per scrivere file alla volta).

È possibile fornire un codice di esempio su come farlo con C#/.NET 4.5?

http://blogs.msdn.com/b/csharpfaq/archive/2012/01/23/using-async-for-file-access-alan-berman.aspx

Answer 1

Penso che usare TPL Dataflow per questo sarebbe una buona idea: si creano blocchi pre e post processo con parallelismo illimitato, un blocco di scrittura di file con parallelismo limitato e si collegano tra loro. Qualcosa di simile:

var unboundedParallelismOptions = 
    new ExecutionDataflowBlockOptions 
    { 
     MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded 
    }; 

var preProcessBlock = new TransformBlock<string, string>(
    s => PreProcess(s), unboundedParallelismOptions); 

var writeToFileBlock = new TransformBlock<string, string>(
    async s => 
      { 
       await WriteToFile(s); 
       return s; 
      }, 
    new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = 10 }); 

var postProcessBlock = new ActionBlock<string>(
    s => PostProcess(s), unboundedParallelismOptions); 

var propagateCompletionOptions = 
    new DataflowLinkOptions { PropagateCompletion = true }; 

preProcessBlock.LinkTo(writeToFileBlock, propagateCompletionOptions); 
writeToFileBlock.LinkTo(postProcessBlock, propagateCompletionOptions); 

// use something like await preProcessBlock.SendAsync("text") here 

preProcessBlock.Complete(); 
await postProcessBlock.Completion; 

Dove WriteToFile() potrebbe assomigliare a questo:

private static async Task WriteToFile(string s) 
{ 
    using (var writer = new StreamWriter(GetFileName())) 
     await writer.WriteAsync(s); 
} 

Answer 2

Sembra che ci si vuole prendere in considerazione un Djikstra i semafori per controllare l'accesso alla partenza di compiti.

Tuttavia, questo suona come un tipico problema di coda/numero fisso di consumatori, che potrebbe essere un modo più appropriato per strutturarlo.