Limitation de la simultanéité des processus de génération et d'attente des processus de travail

Question

J'ai un programme C# qui met en file d'attente un ensemble d'éléments de travail. Pour chaque élément de travail, je génère une autre application (que je ne peux pas modifier) ​​pour traiter cet élément. Ce processus est relativement gourmand en ressources processeur. Je voudrais limiter le nombre d'instances de cette application.

J'ai pensé à utiliser PLINQ:

Parallel.ForEach(
    workItems, 
    new ParallelOptions { MaxDegreeOfParallelism = 4 }, 
    x => Process.Start("worker", x).WaitForExit()); 

Mais ma préoccupation est que chaque travailleur parallèle utilisera simplement un thread pour attendre que le processus correspondant.

J'ai aussi vu que PLINQ essayait de grouper les articles retournés, ce qui signifie qu'il pourrait bloquer en attendant un lot de taille appropriée. Donc, peut-être un modèle mono-producteur/consommateur multiple utilisant BlockingCollection fonctionnerait? Le problème avec cela est que j'aurais un thread pour chaque travailleur parallèle, ce qui serait (?) Pire que la solution PLINQ. Etant donné que ce qui précède est simplifié et que j'ai effectivement un TaskCompletionSource connecté à chaque processus de travail (via l'événement Exited), est-ce que je pourrais utiliser quelque chose de la TPL pour faire cela sans bloquer les threads d'arrière-plan?

Answer 1

Si vous pouvez envelopper le début Process et terminer en utilisant l'événement Process.Exited dans un Task:

Task WrapExternalProcess(WorkItem workItem) { ... } 

Vous pouvez utiliser continuations pour éliminer le blocage total. Quelque chose comme cela ferait:

Task DoAllWork(IEnumerable<WorkItem> workItems) 
{ 
    int THREAD_COUNT = 4; 

    var bag = new ConcurrentBag<WorkItem>(workItems); 
    var ce = new CountdownEvent(THREAD_COUNT); 
    var tcs = new TaskCompletionSource<bool>(); 

    for (int i = 0 ; i < THREAD_COUNT ; i++) Work(bag, ce, tcs); 

    return tcs.Task; 
} 

void Work(
    ConcurrentBag<WorkItem> bag, 
    CountdownEvent ce, 
    TaskCompletionSource<bool> tcs 
) 
{ 
    WorkItem workItem; 
    if (bag.TryTake(out workItem)) 
    { 
     WrapExternalProcess(workItem) 
      .ContinueWith(t => Work(bag, ce, tcs)); 
    } 
    else // no more work 
    { 
     // If I'm the last thread to finish 
     if (ce.Signal()) tcs.SetResult(true); 
    } 
} 

Answer 2

J'ai mis quelque chose comme ce qui suit:

var sem = new SemaphoreSlim(4); 
foreach (var item in workItems) 
{ 
    sem.Wait(); 

    ProcessAsync(item).ContinueWith(_ => sem.Release()); 
} 

Cela me donne un seul fil qui maintient un nombre limité de processus de fond en vol à la fois.