System.Threading.Tasks - Limitare il numero di task concomitanti

Question

ho appena iniziato a guardare la nuova bontà "System.Threading.Tasks" in .Net 4.0 e vorrebbero sapere se c'è qualche costruire in supporto per limitando il numero di attività simultanee che vengono eseguite contemporaneamente o se questo deve essere gestito manualmente.

per esempio: se ho bisogno di chiamare un metodo di calcolo 100 volte, c'è un modo per impostare 100 attività, ma hanno solo 5 eseguire contemporaneamente? La risposta potrebbe essere solo quella di creare 5 attività, chiamare Task.WaitAny e creare una nuova attività al termine di ciascuna precedente. Voglio solo assicurarmi che non manchi un trucco se esiste un modo migliore per farlo.

In sostanza, c'è un costruito in modo per farlo:

Dim taskArray() = {New Task(Function() DoComputation1()), 
        New Task(Function() DoComputation2()), 
        ... 
        New Task(Function() DoComputation100())} 

Dim maxConcurrentThreads As Integer = 5 
RunAllTasks(taskArray, maxConcurrentThreads) 

Grazie per qualsiasi aiuto.

Answer 1

So che questo è quasi un anno, ma ho trovato un modo molto più semplice per raggiungere questo obiettivo, così ho pensato che avrei condiviso:

Dim actionsArray() As Action = 
    new Action(){ 
     New Action(Sub() DoComputation1()), 
     New Action(Sub() DoComputation2()), 
     ... 
     New Action(Sub() DoComputation100()) 
     } 

System.Threading.Tasks.Parallel.Invoke(New Tasks.ParallelOptions() With {.MaxDegreeOfParallelism = 5}, actionsArray) 

Voila!

Answer 2

E non sembra che, anche se si potrebbe creare una sottoclasse di TaskScheduler che implementa tale comportamento.

Answer 3

Risposta breve: Se ciò che si desidera è limitare il numero di attività di lavoro in modo che non saturino il servizio Web, quindi penso che il tuo approccio sia corretto.

Risposta lunga: Il nuovo motore System.Threading.Tasks in .NET 4.0 viene eseguito su .NET ThreadPool. Poiché esiste sempre un solo ThreadPool per processo e il valore predefinito è un massimo di 250 thread di lavoro. Pertanto, se si dovesse impostare il numero massimo di ThreadPool di thread da un numero più modesto, si può essere in grado di ridurre il numero di thread contemporaneamente esecuzione, e quindi attività utilizzando l'API ThreadPool.SetMaxThreads (...).

TUTTAVIA, si noti che si potrebbe non essere soli nell'utilizzo del ThreadPool poiché molte altre classi che si utilizzano possono anche mettere in coda elementi al ThreadPool. Pertanto, ci sono buone probabilità che tu possa finire paralizzando il resto della tua app facendo questo. Si noti inoltre che, poiché il ThreadPool impiega un algoritmo per ottimizzare l'utilizzo di core alla base di una determinata macchina, limitando il numero di thread ThreadPool può mettere in coda ad un numero arbitrariamente basso può causare alcuni problemi di prestazioni piuttosto catastrofici.

Anche in questo caso, se si desidera eseguire un numero limitato di attività/thread di lavoro per esercitare alcune attività, solo la creazione di un numero limitato di attività (rispetto a 100) rappresenta l'approccio migliore.

Answer 4

So che questo è un vecchio thread, ma ho solo voluto condividere la mia soluzione a questo problema: uso i semafori.

(Questo è in C#)

private void RunAllActions(IEnumerable<Action> actions, int maxConcurrency) 
{ 
    using(SemaphoreSlim concurrencySemaphore = new SemaphoreSlim(maxConcurrency)) 
    { 
     foreach(Action action in actions) 
     { 
      Task.Factory.StartNew(() => 
      { 
       concurrencySemaphore.Wait(); 
       try 
       { 
        action(); 
       } 
       finally 
       { 
        concurrencySemaphore.Release(); 
       } 
      }); 
     } 
    } 
} 

Answer 5

Una soluzione potrebbe essere quella di dare un'occhiata al codice pre-fatte da Microsoft here.

La descrizione è simile a questa: "Fornisce un task scheduler che garantisce un livello di concorrenza massimo durante l'esecuzione in cima al ThreadPool.", E per quanto ho potuto testarlo sembra fare il trucco, in allo stesso modo della proprietà MaxDegreeOfParallelism in ParallelOptions.

Answer 6

Se il programma utilizza i servizi Web, il numero di connessioni simultanee sarà limitato alla proprietà ServicePointManager.DefaultConnectionLimit. Se vuoi 5 connessioni simultanee non è sufficiente usare la soluzione di Arrow_Raider. Dovresti anche aumentare ServicePointManager.DefaultConnectionLimit perché è solo 2 per impostazione predefinita.

Answer 7

C# equivalente di campione fornito dal James

Action[] actionsArray = new Action[] { 
new Action(() => DoComputation1()), 
new Action(() => DoComputation2()), 
    //... 
new Action(() => DoComputation100()) 
    }; 

    System.Threading.Tasks.Parallel.Invoke(new Tasks.ParallelOptions {MaxDegreeOfParallelism = 5 }, actionsArray) 

Answer 8

My blog post mostra come fare questo sia con le attività e con azioni, e fornisce un progetto di esempio è possibile scaricare ed eseguire di vedere sia in azione.

Con azioni

Se utilizzando azioni, è possibile utilizzare la funzione built-in .Net Parallel.Invoke. Qui lo limitiamo a eseguire al massimo 5 thread in parallelo.

var listOfActions = new List<Action>(); 
for (int i = 0; i < 100; i++) 
{ 
    // Note that we create the Action here, but do not start it. 
    listOfActions.Add(() => DoSomething()); 
} 

var options = new ParallelOptions {MaxDegreeOfParallelism = 5}; 
Parallel.Invoke(options, listOfActions.ToArray()); 

con compiti

Dal momento che si sta utilizzando Task qui però, non v'è alcuna funzione built-in. Tuttavia, puoi utilizzare quello che fornisco sul mio blog.

/// <summary> 
    /// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel. 
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para> 
    /// </summary> 
    /// <param name="tasksToRun">The tasks to run.</param> 
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param> 
    /// <param name="cancellationToken">The cancellation token.</param> 
    public static void StartAndWaitAllThrottled(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, CancellationToken cancellationToken = new CancellationToken()) 
    { 
     StartAndWaitAllThrottled(tasksToRun, maxTasksToRunInParallel, -1, cancellationToken); 
    } 

    /// <summary> 
    /// Starts the given tasks and waits for them to complete. This will run, at most, the specified number of tasks in parallel. 
    /// <para>NOTE: If one of the given tasks has already been started, an exception will be thrown.</para> 
    /// </summary> 
    /// <param name="tasksToRun">The tasks to run.</param> 
    /// <param name="maxTasksToRunInParallel">The maximum number of tasks to run in parallel.</param> 
    /// <param name="timeoutInMilliseconds">The maximum milliseconds we should allow the max tasks to run in parallel before allowing another task to start. Specify -1 to wait indefinitely.</param> 
    /// <param name="cancellationToken">The cancellation token.</param> 
    public static void StartAndWaitAllThrottled(IEnumerable<Task> tasksToRun, int maxTasksToRunInParallel, int timeoutInMilliseconds, CancellationToken cancellationToken = new CancellationToken()) 
    { 
     // Convert to a list of tasks so that we don&#39;t enumerate over it multiple times needlessly. 
     var tasks = tasksToRun.ToList(); 

     using (var throttler = new SemaphoreSlim(maxTasksToRunInParallel)) 
     { 
      var postTaskTasks = new List<Task>(); 

      // Have each task notify the throttler when it completes so that it decrements the number of tasks currently running. 
      tasks.ForEach(t => postTaskTasks.Add(t.ContinueWith(tsk => throttler.Release()))); 

      // Start running each task. 
      foreach (var task in tasks) 
      { 
       // Increment the number of tasks currently running and wait if too many are running. 
       throttler.Wait(timeoutInMilliseconds, cancellationToken); 

       cancellationToken.ThrowIfCancellationRequested(); 
       task.Start(); 
      } 

      // Wait for all of the provided tasks to complete. 
      // We wait on the list of "post" tasks instead of the original tasks, otherwise there is a potential race condition where the throttler&#39;s using block is exited before some Tasks have had their "post" action completed, which references the throttler, resulting in an exception due to accessing a disposed object. 
      Task.WaitAll(postTaskTasks.ToArray(), cancellationToken); 
     } 
    } 

E poi creare l'elenco delle attività e chiamare la funzione di farli correre, con dire un massimo di 5 contemporaneamente alla volta, si potrebbe fare questo:

var listOfTasks = new List<Task>(); 
for (int i = 0; i < 100; i++) 
{ 
    var count = i; 
    // Note that we create the Task here, but do not start it. 
    listOfTasks.Add(new Task(() => Something())); 
} 
Tasks.StartAndWaitAllThrottled(listOfTasks, 5);