c# - 长时间运行的进程的并行化和性能优化

Question

我想并行处理逐帧处理多个视频剪辑的应用程序。每个剪辑的每个帧的顺序很重要(显然)。我决定使用 TPL 数据流，因为我相信这是数据流的一个很好的例子(电影帧就是数据)。

所以我有一个从数据库加载帧的进程(比如说一批 500 个，全部聚在一起)

Example sequence:    
|mid:1 fr:1|mid:1 fr:2|mid:2 fr:1|mid:3 fr:1|mid:1 fr:3|mid:2 fr:2|mid:2 fr:3|mid:1 fr:4|

并将它们发布到 BufferBlock。对于这个 BufferBlock，我将 ActionBlocks 与过滤器链接起来，使每个 MovieID 有一个 ActionBlock，这样我就可以进行某种数据分区。每个 ActionBlock 都是顺序的，但理想情况下，多部电影的多个 ActionBlock 可以并行运行。

我确实让上述网络工作并且它确实并行运行，但根据我的计算，只有八到十个 ActionBlocks 在同时执行。我对每个 ActionBlock 的运行时间进行了计时，大约为 100-200 毫秒。我可以采取哪些步骤至少使并发性翻倍？

我曾尝试将 Action 委托(delegate)转换为异步方法，并在 ActionBlock Action 委托(delegate)中使数据库访问异步，但没有帮助。

编辑:我实现了额外级别的数据分区:奇数 ID 电影的帧在 ServerA 上处理，偶数电影的帧在 ServerB 上处理。应用程序的两个实例都访问同一个数据库。如果我的问题是 DB IO，那么我不会看到处理的总帧数有任何改善(或者非常小，低于 20%)。但我确实看到它翻了一番。因此，这让我得出结论，Threadpool 并没有产生更多的线程来并行处理更多的帧(两台服务器都是四核的，分析器显示每个应用程序大约有 25-30 个线程)。

Answer 1

一些假设:

• 根据您的示例数据，您收到的电影帧(可能还有电影中的帧)乱序

• 您的 ActionBlock<T> 实例是通用的；它们都调用相同的方法进行处理，您只需根据每个电影 ID 创建它们的列表(您事先有一个电影 ID 列表)，如下所示:

// The movie IDs
IEnumerable<int> movieIds = ...;

// The actions.
var actions = movieIds.Select(
    i => new { Id = i, Action = new ActionBlock<Frame>(MethodToProcessFrame) });

// The buffer block.
BufferBlock<Frame> buffer = ...;

// Link everything up.
foreach (var action in actions) 
{
    // Not necessary in C# 5.0, but still, good practice.
    // The copy of the action.
    var actionCopy = action;

    // Link.
    bufferBlock.LinkTo(actionCopy.Action, f => f.MovieId == actionCopy.Id);
}

如果是这种情况，则说明您创建了太多 ActionBlock<T>没有被赋予工作的实例；因为你的帧(和可能的电影)是乱序的，你不能保证所有的 ActionBlock<T>实例将有工作要做。

此外，当您创建 ActionBlock<T>实例它将使用 MaxDegreeOfParallelism 创建为 1，这意味着它是线程安全的，因为只有一个线程可以同时访问该 block 。

此外，TPL DataFlow 库最终依赖于 Task<TResult> class ，它默认在线程池上进行调度。线程池将在这里做一些事情:

• 确保所有处理器核心都饱和。这与确保您的 ActionBlock<T> 非常不同实例已经饱和，这个是您应该关注的指标

• 确保在处理器内核饱和时，确保工作均匀分布，并确保没有太多正在执行的并发任务(上下文切换很昂贵).

看起来你处理电影的方法是通用的，传入什么电影的帧并不重要(如果它确实重要，那么你需要更新你的问题有了它，因为它改变了很多东西)。这也意味着它是线程安全的。

此外，如果可以假设一帧的处理不依赖于任何先前帧的处理(或者，看起来电影的帧是按顺序排列的)，您可以使用单个 ActionBlock<T>但调整 MaxDegreeOfParallelism值，像这样:

// The buffer block.
BufferBlock<Frame> buffer = ...;

// Have *one* ActionBlock<T>
var action = new ActionBlock<Frame>(MethodToProcessFrame,
    // This is where you tweak the concurrency:
    new ExecutionDataflowBlockOptions {
        MaxDegreeOfParallelism = 4,
    }
);

// Link.  No filter needed.
bufferBlock.LinkTo(action);

现在，您的 ActionBlock<T>将总是饱和。诚然，任何负责任的任务调度程序(默认情况下是线程池)仍然会限制最大并发量，但它会在同一时间合理地做尽可能多的事情。

为此，如果您的操作是真正线程安全的，您可以设置 MaxDegreeOfParallelism至 DataflowBlockOptions.Unbounded ，像这样:

// Have *one* ActionBlock<T>
var action = new ActionBlock<Frame>(MethodToProcessFrame,
    // This is where you tweak the concurrency:
    new ExecutionDataflowBlockOptions {
        // We're thread-safe, let the scheduler determine
        // how nuts we can go.
        MaxDegreeOfParallelism = DataflowBlockOptions.Unbounded,
    }
);

当然，所有这些都假设其他一切都是最优的(I/O 读/写等)