使Parallel.ForEach等待工作，直到一個插槽打開

Question

我使用Parallel.ForEach工作一堆項目。問題是，我想根據打開的工作人員（插槽）的數量來確定哪些項目能夠工作的優先順序。例如。如果我在8個並行的東西上工作，並在任務1-4之間打開一個插槽，我希望將簡單的工作分配給這些插槽。插槽的下半部分將得到努力工作。通過這種方式，我不會把所有8個插槽都捆綁在一起進行艱苦/長時間運行的工作，易/快速的項目將首先運行。我實現了這個如下：

守則

const int workers = 8; 
List<Thing> thingsToDo = ...; //Get the things that need to be done. 
Thing[] currentlyWorkingThings = new Thing[workers]; //One slot for each worker. 

void Run() { 
    Parallel.ForEach(PrioritizeThings(thingsToDo), o => { 
     int index = 0; 

     //"PrioritizeTasks" added this thing to the list of currentlyWorkingThings. 
     //Find my position in this list. 
     lock (currentlyWorkingThings) 
      index = currentlyWorkingThings.IndexOf(o); 

     //Do work on this thing... 

     //Then remove it from the list of currently working things, thereby 
     // opening a new slot when this worker returns/finishes. 
     lock (currentlyWorkingThings) 
      currentlyWorkingThings[index] = null; 
    }); 
} 

IEnumerable<Thing> PrioritizeThings(List<Thing> thingsToDo) { 
    int slots = workers; 
    int halfSlots = (int)Math.Ceiling(slots/2f); 

    //Sort thingsToDo by their difficulty, easiest first. 

    //Loop until we've worked every Thing. 
    while (thingsToDo.Count > 0) { 
     int slotToFill = ...; //Find the first open slot. 
     Thing nextThing = null; 

     lock (currentlyWorkingThings) { 
      //If the slot is in the "top half", get the next easy thing - otherwise 
      // get the next hard thing. 
      if (slotToFill < halfSlots) 
       nextThing = thingsToDo.First(); 
      else 
       nextThing = thingsToDo.Last(); 

      //Add the nextThing to the list of currentlyWorkingThings and remove it from 
      // the list of thingsToDo. 
      currentlyWorkingThings[slotToFill] = nextThing; 
      thingsToDo.Remove(nextThing); 
     } 

     //Return the nextThing to work. 
     yield return nextThing; 
    } 
} 

的問題

所以我看到這裏的問題是，Parallel請求，接下來的事情去努力從PrioritizeThings開始插槽（在現有物品完成之前）。我認爲Parallel正在展望未來，並提前準備好工作。我希望它不要這樣做，並且只在完成完成後才填寫工人/插槽。我想要解決這個問題的唯一方法是在PrioritizeThings中添加一個睡眠/等待循環，在它看到一個合法的開放時隙之前，它不會返回一個工作。但我不喜歡這樣，我希望有一些方法可以使Parallel等待更長時間才能開始工作。有什麼建議麼？

Answer 1

有一種方法可以支持您正在描述的情況。

當您創建ForEach時，您需要傳遞ParallelOptions以及非標準TaskScheduler。難的是創造一個TaskSchedueler做優先系統對你來說，幸運的是微軟發佈的包含一個這樣的調度例子一包名爲「ParallelExtensionsExtras」以其調度QueuedTaskScheduler

private static void Main(string[] args) 
{ 
    int totalMaxConcurrancy = Environment.ProcessorCount; 
    int highPriorityMaxConcurrancy = totalMaxConcurrancy/2; 

    if (highPriorityMaxConcurrancy == 0) 
     highPriorityMaxConcurrancy = 1; 

    QueuedTaskScheduler qts = new QueuedTaskScheduler(TaskScheduler.Default, totalMaxConcurrancy); 
    var highPriortiyScheduler = qts.ActivateNewQueue(0); 
    var lowPriorityScheduler = qts.ActivateNewQueue(1); 

    BlockingCollection<Foo> highPriorityWork = new BlockingCollection<Foo>(); 
    BlockingCollection<Foo> lowPriorityWork = new BlockingCollection<Foo>(); 

    List<Task> processors = new List<Task>(2); 

    processors.Add(Task.Factory.StartNew(() => 
    { 
     Parallel.ForEach(highPriorityWork.GetConsumingPartitioner(), //.GetConsumingPartitioner() is also from ParallelExtensionExtras, it gives better performance than .GetConsumingEnumerable() with Parallel.ForEeach(
         new ParallelOptions() { TaskScheduler = highPriortiyScheduler, MaxDegreeOfParallelism = highPriorityMaxConcurrancy }, 
         ProcessWork); 
    }, TaskCreationOptions.LongRunning)); 

    processors.Add(Task.Factory.StartNew(() => 
    { 
     Parallel.ForEach(lowPriorityWork.GetConsumingPartitioner(), 
         new ParallelOptions() { TaskScheduler = lowPriorityScheduler}, 
         ProcessWork); 
    }, TaskCreationOptions.LongRunning)); 


    //Add some work to do here to the highPriorityWork or lowPriorityWork collections 


    //Lets the blocking collections know we are no-longer going to be adding new items so it will break out of the `ForEach` once it has finished the pending work. 
    highPriorityWork.CompleteAdding(); 
    lowPriorityWork.CompleteAdding(); 

    //Waits for the two collections to compleatly empty before continueing 
    Task.WaitAll(processors.ToArray()); 
} 

private static void ProcessWork(Foo work) 
{ 
    //... 
} 

即使你有Parallel.ForEach兩個實例運行它們兩者的組合總數不會超過您通過MaxConcurrency傳入QueuedTaskScheduler構造函數中的值，並且如果兩項工作都有工作要做，則優先清空highPriorityWork集合（最多可達到所有可用插槽的1/2，以便您不會窒息低優先級隊列，您可以輕鬆地將其調整爲更高或更低的比例取決於你的性能需求）。

如果你不希望高優先級總是贏，而你更希望有一個在兩個列表之間交替的「循環」風格的計劃程序（所以你不希望快速項目總是贏，但只是讓他們隨着緩慢的項目洗牌），你可以設置相同的優先級到兩個或更多的隊列（或只是使用RoundRobinTaskSchedulerQueue這是做同樣的事情）