C# 可以使用 Barrier 或类似的东西来多次同步任务吗？

Question

我了解障碍可用于在第二阶段运行之前让多个任务同步完成。

我想让几个任务同步多个步骤，如下所示：

state is 1;
Task1 runs and pauses waiting for state to become 2;
Task2 runs and pauses waiting for state to become 2;
Task2 is final Task and causes the state to progress to state 2;
Task1 runs and pauses waiting for state to become 3;
Task2 runs and pauses waiting for state to become 3;
Task2 is final Task and causes the state to progress to state 3;
state 3 is final state and so all tasks exit.

我知道我可以在每个状态结束时启动新任务，但由于每个任务不会花费太长时间，我想避免为每个步骤创建新任务。

我可以使用 for 循环同步运行上述内容，但最终状态可能是 100,000，因此我希望使用多个线程来更快地运行该进程，因为该进程受 CPU 限制。

我尝试使用计数器来跟踪每个任务在完成时增加的已完成任务的数量。如果任务是要完成的最终任务，那么它将状态更改为下一个状态。所有已完成的任务然后使用while (iterationState == state) await Task.Yield 等待，但性能很糟糕，在我看来这是一种非常粗暴的做法。

完成上述工作的最有效方法是什么？必须有一个优化的工具来完成这项工作？

我正在使用Parallel.For，创建 300 个任务，每个任务需要运行多达 100,000 个状态。通过一个状态运行的每个任务在不到一秒的时间内完成，创建 300 * 100,000 个任务是一个巨大的开销，即使使用单个线程，同步运行整个任务也会更快。

所以我想创建 300 个任务并让这些任务在 100,000 个状态中同步移动。希望仅创建 300 个任务而不是 300 * 100,000 个任务的开销，以及任务之间优化同步的开销，将比在单个线程上同步执行时运行得更快。

每个状态必须完全完成才能运行下一个状态。

那么 - 这种情况下的最佳同步技术是什么？谢谢！

Answer 1

while (iterationState == state) await Task.Yield 确实是在 300 个任务之间同步的糟糕解决方案（不，300 个不一定非常昂贵：您只会分配合理数量的线程）。

这里的关键问题不是Parallel.For，它是在 300 个任务之间同步以高效地等待，直到每个任务都完成给定阶段。

这里最简单和最干净的解决方案可能是在阶段上有一个 for 循环，在你想要并行化的位上有一个 parallel.for：

for (int stage = 0; stage < 10000; stage++)
{
   // the next line blocks until all 300 have completed
   // will use thread pool threads as necessary
   Parallel.For( ... 300 items to process this stage ... );
}

无需额外的同步原语，无需消耗 CPU 的自旋等待，无需在线程之间进行不必要的颠簸以查看它们是否已准备好继续进行。

Answer 2

我想我理解你想要做什么，所以这里有一个建议的方法来处理它。注意 - 我使用 Action 作为阻塞集合的类型，但您可以将其更改为最适合您的场景的类型。

// Shared variables
CountdownEvent workItemsCompleted = new CountdownEvent(300);
BlockingCollection<Action> workItems = new BlockingCollection<Action>();
CancellationTokenSource cancelSource = new CancellationTokenSource();

// Work Item Queue Thread
for(int i=1; i < stages; ++i)
{
    workItemsCompleted.Reset(300);
    for(int j=0; j < workItemsForStage[i].Count; ++j)
    {
         workItems.Add(() => {}) // Add your work item here
    }
    workItemsCompleted.Wait(token) // token should be passed in from cancelSource.Token
}

// Worker threads that are making use of the queue
// token should be passed to the threads from cancelSource.Token
while(!token.IsCancelled)
{
    var item = workItems.Take(token); // Blocks until available item or token is cancelled
    item();
    workItemsCompleted.Signal();
}

如果需要，您可以在主线程中使用cancelSource 来取消正在运行的操作。在您的工作线程中，您需要处理OperationCancelledException。使用此设置，您可以根据需要启动尽可能多的工作线程，并轻松对获得最佳性能的地方进行基准测试（可能只使用 10 个工作线程等）。只需启动尽可能多的工作人员，然后在工作项队列线程中排队工作项。它基本上是一种生产者-消费者类型的模型，只是生产者将一个阶段的工作排队，然后阻塞，直到该阶段完成，然后再排队下一轮的工作。