実行中のタスクの最大数を保持するコレクション

Question

Taskオブジェクトを実行している最大数を保持するコレクション/バッグ（何か）が必要です。新しい実行を追加すると、実行中のタスクの最大数が増えた場合に、追加する新しいタスクの空きスロットがあるまで、呼び出しにスレッドをブロックする必要があります（スレッドを安全にするためにスレッドを追加しようとするスレッドが多数あります）が達成された。これは私がこれまでに持っているもので、うまくいきます。

public class ConcurrentTaskLimiter 
{ 
    public int MaxWorkingTasks { get; } 
    private readonly Task[] _tasks; 
    private readonly bool[] _finished; 

    public ConcurrentTaskLimiter(int maxWorkingTasks) 
    { 
     MaxWorkingTasks = maxWorkingTasks; 
     if ((1 <= maxWorkingTasks) == false) 
      throw new ArgumentOutOfRangeException(nameof(maxWorkingTasks), maxWorkingTasks, "Must be >= 1"); 
     _tasks = new Task[maxWorkingTasks]; 
     _finished = new bool[maxWorkingTasks]; 

     for (int i = 0; i < MaxWorkingTasks; i++) 
     { 
      _tasks[i] = Task.FromResult(0); // use this as finished tasks 
      _finished[i] = true; 
     } 
    } 

    public void BlockAdd(Task t) 
    { 
     if (t == null) 
      throw new ArgumentNullException(nameof(t)); 

     if (t.Status == TaskStatus.Canceled 
      || t.Status == TaskStatus.Faulted 
      || t.Status == TaskStatus.RanToCompletion) 
      return; 

     lock (this) 
     { 
      int i; 
      while (true) 
      { 
       for (i = 0; i < MaxWorkingTasks; i++) 
       { 
        if (_finished[i]) 
        { 
         _tasks[i] = t; 
         _finished[i] = false; 
         return; 
        } 
       } 
       i = Task.WaitAny(_tasks); 
       _finished[i] = true; 
      } 
     } 
    } 
} 

このコードに問題はありませんか？または、この種のタスクを処理できるクラスがいくつか組み込まれていますか？:)

Answer 1

SynchronizedCollection<T>を使用すると、すぐにスレッドの安全が確保されます。またはSystem.Collections.Concurrentのコレクションの1つを使用してください。私はSystem.Collections.Concurrentのものを使っています。なぜなら、それらははるかに新しい、より効率的ですからです。

Answer 2

.NETには、BlockingCollection<T>というクラスがあり、タスクを大幅に簡素化できます。

ブロッキングコレクションのインスタンスは、保持できるタスクの数の上限で初期化できます。これを行うと、タスクを追加するとコレクションが容量を超えてしまうため、Addの呼び出しがブロックされます。

Answer 3

私はより洗練されたソリューションを提案します。

public class TasksPool 
    { 
     public int MaxSize { get; private set; } 

     public TasksPool(int maxSize) 
     { 
      if (maxSize < 1) throw new IndexOutOfRangeException("Should be 1 or more"); 
      MaxSize = maxSize; 
      _semaphore = new SemaphoreSlim(maxSize-1); 
     } 

     private readonly SemaphoreSlim _semaphore; 

     public void Add(Task t, CancellationToken token) 
     { 
      _semaphore.Wait(token); 

      if (token.IsCancellationRequested) return; 
      t.ContinueWith(q => _semaphore.Release(), token); 
     } 
    }