使用相同的模型并行反序列化(高效)

Question

我正在尝试读取使用ProtoBuf.NET使用.NET任务序列化的多个文件，如：

public static ResultsDump Amalgamate(RuntimeTypeModel model, IEnumerable<string> files)
{      
  var readDumpTasks = 
    files.Select(fn =>
      Task<ResultsDump>.Factory.StartNew(() => {
        try {
          using (var dumpFile = new FileStream(fn, FileMode.Open))
          {
            var miniDump = (ResultsDump)model.Deserialize(dumpFile, null, typeof(ResultsDump));
            if (miniDump == null) {
              throw new Exception(string.Format("Failed to deserialize dump file {0}", fn));
            }
            //readDumps.Add(miniDump);
            return miniDump;
          }
        }
        catch (Exception e) {
          throw new Exception(string.Format("cannot read dump file {0}: {1}", fn, e.Message), e);
        }
      })).ToArray();

  Task.WaitAll(readDumpTasks);

  var allDumps = readDumpTasks.Select(t => t.Result).ToList();

  // Goes on.. irrelevant to the question
}

由于某些原因，的CPU使用率并不真正超过单个核心。Protobuf.NET中有什么固有的锁不喜欢并发地反序列化多个文件吗？

我尝试过多个RuntimeTypeModel实例以及一个实例，它似乎总是在非常“低”的CPU使用率水平上达到峰值。

难道我指责ProtoBuf.NET是错的吗？这是.NET内存分配器/ TPL吗？

Answer 1

protobuf-net中有意进行非常有限的锁定；它只在检查类型(第一次运行)以确定需要什么时才真正锁定。一旦该模型被理解，它是无锁的，它被设计成并行的。

如前所述(注释)，IO极有可能是您的瓶颈。实际上，并行访问同一个物理磁盘/纺锤体通常会大大降低吞吐量，因为缓冲区的竞争更加激烈，而且它必须进行更多的查找而不是连续的读取。

这应该很容易测试/验证:对于测试运行，而不是从磁盘读取，首先加载到内存；

var ms = new MemoryStream(
    File.ReadAllBytes(path));

在加载所有文件后，现在执行相同的代码，但将MemoryStream作为输入传递。如果它仍然不缩放，它可能是一个bug。然而，我强烈地怀疑，在这一点上，您会发现它很好地并行。

下面是一个有用的示例，对我来说，它用并发反序列化填充我的所有内核：

using System.Collections.Generic;
using ProtoBuf;
using System;
using System.IO;
using System.Threading.Tasks;


internal class Program
{
    private static void Main()
    {
        var foo = new Foo { Bars = new List<Bar>() };
        var rand = new Random(1234);
        for (int i = 0; i < 1000; i++)
        {
            var bar = new Bar
            {
                A = rand.Next(),
                B = rand.Next(),
                C = rand.Next(),
                D = rand.Next(),
                E = rand.Next(),
                F = rand.Next(),
                G = rand.Next(),
                H = rand.Next()
            };
            foo.Bars.Add(bar);
        }
        var ms = new MemoryStream();
        Serializer.Serialize(ms, foo);
        var bytes = ms.ToArray();
        const int count = 100000;
        Parallel.For(0, count, x =>
        {
            Serializer.Deserialize<Foo>(new MemoryStream(bytes));
        });
    }
}
[ProtoContract]
internal class Foo
{
    [ProtoMember(1)]
    public List<Bar> Bars { get; set; }
}
[ProtoContract]
internal class Bar
{
    [ProtoMember(1)]
    public int A { get; set; }
    [ProtoMember(2)]
    public int B { get; set; }
    [ProtoMember(3)]
    public int C { get; set; }
    [ProtoMember(4)]
    public int D { get; set; }
    [ProtoMember(5)]
    public int E { get; set; }
    [ProtoMember(6)]
    public int F { get; set; }
    [ProtoMember(7)]
    public int G { get; set; }
    [ProtoMember(8)]
    public int H { get; set; }
}