c# - TPL Dataflow，对核心设计感到困惑

Question

我一直在使用 TPL Dataflow，但遇到了一个我无法解决的问题：

我有以下架构：

BroadCastBlock<List<object1>>-> 2 个不同的TransformBlock<List<Object1>, Tuple<int, List<Object1>>>-> 都链接到TransformManyBlock<Tuple<int, List<Object1>>, Object2>

我在链末端的 TransformManyBlock 中改变 lambda 表达式：(a) 对流式元组执行操作的代码，(b) 根本没有代码。

在 TransformBlocks 中，我测量从第一项到达开始到 TransformBlock.Completion 指示块完成时停止的时间（broadCastBlock 链接到 transfrom 块，propagateCompletion 设置为 true）。

我无法调和的是，为什么 (b) 情况下的 transformBlocks 完成速度比 (a) 快 5-6 倍。这完全违背了整个 TDF 设计意图的意图。转换块中的项目被传递到 transfromManyBlock，因此 transformManyBlock 对影响转换块何时完成的项目所做的一切都无关紧要。我看不出transfromManyBlock中发生的任何事情可能与前面的TransformBlocks有关的单一原因。

谁能调和这个奇怪的观察？

这是一些显示差异的代码。运行代码时，请确保更改以下两行：

        tfb1.transformBlock.LinkTo(transformManyBlock);
        tfb2.transformBlock.LinkTo(transformManyBlock);

至：

        tfb1.transformBlock.LinkTo(transformManyBlockEmpty);
        tfb2.transformBlock.LinkTo(transformManyBlockEmpty);

为了观察前面的 transformBlocks 在运行时的差异。

class Program
{
    static void Main(string[] args)
    {
        Test test = new Test();
        test.Start();
    }
}

class Test
{
    private const int numberTransformBlocks = 2;
    private int currentGridPointer;
    private Dictionary<int, List<Tuple<int, List<Object1>>>> grid;

    private BroadcastBlock<List<Object1>> broadCastBlock;
    private TransformBlockClass tfb1;
    private TransformBlockClass tfb2;

    private TransformManyBlock<Tuple<int, List<Object1>>, Object2> 
               transformManyBlock;
    private TransformManyBlock<Tuple<int, List<Object1>>, Object2> 
               transformManyBlockEmpty;
    private ActionBlock<Object2> actionBlock;

    public Test()
    {
        grid = new Dictionary<int, List<Tuple<int, List<Object1>>>>();

        broadCastBlock = new BroadcastBlock<List<Object1>>(list => list);

        tfb1 = new TransformBlockClass();
        tfb2 = new TransformBlockClass();

        transformManyBlock = new TransformManyBlock<Tuple<int, List<Object1>>, Object2>
                (newTuple =>
            {
                for (int counter = 1; counter <= 10000000;  counter++)
                {
                    double result = Math.Sqrt(counter + 1.0);
                }

                return new Object2[0];

            });

        transformManyBlockEmpty 
            = new TransformManyBlock<Tuple<int, List<Object1>>, Object2>(
                  tuple =>
            {
                return new Object2[0];
            });

        actionBlock = new ActionBlock<Object2>(list =>
            {
                int tester = 1;
                //flush transformManyBlock
            });

        //linking
        broadCastBlock.LinkTo(tfb1.transformBlock
                              , new DataflowLinkOptions 
                                  { PropagateCompletion = true }
                              );
        broadCastBlock.LinkTo(tfb2.transformBlock
                              , new DataflowLinkOptions 
                                  { PropagateCompletion = true }
                              );

        //link either to ->transformManyBlock or -> transformManyBlockEmpty
        tfb1.transformBlock.LinkTo(transformManyBlock);
        tfb2.transformBlock.LinkTo(transformManyBlock);

        transformManyBlock.LinkTo(actionBlock
                                  , new DataflowLinkOptions 
                                       { PropagateCompletion = true }
                                  );
        transformManyBlockEmpty.LinkTo(actionBlock
                                       , new DataflowLinkOptions 
                                            { PropagateCompletion = true }
                                       );

        //completion
        Task.WhenAll(tfb1.transformBlock.Completion
                     , tfb2.transformBlock.Completion)
                       .ContinueWith(_ =>
            {
                transformManyBlockEmpty.Complete();
                transformManyBlock.Complete();
            });

        transformManyBlock.Completion.ContinueWith(_ =>
            {
                Console.WriteLine("TransformManyBlock (with code) completed");
            });

        transformManyBlockEmpty.Completion.ContinueWith(_ =>
        {
            Console.WriteLine("TransformManyBlock (empty) completed");
        });

    }

    public void Start()
    {
        const int numberBlocks = 100;
        const int collectionSize = 300000;


        //send collection numberBlock-times
        for (int i = 0; i < numberBlocks; i++)
        {
            List<Object1> list = new List<Object1>();
            for (int j = 0; j < collectionSize; j++)
            {
                list.Add(new Object1(j));
            }

            broadCastBlock.Post(list);
        }

        //mark broadCastBlock complete
        broadCastBlock.Complete();

        Console.WriteLine("Core routine finished");
        Console.ReadLine();
    }
}

class TransformBlockClass
{
    private Stopwatch watch;
    private bool isStarted;
    private int currentIndex;

    public TransformBlock<List<Object1>, Tuple<int, List<Object1>>> transformBlock;

    public TransformBlockClass()
    {
        isStarted = false;
        watch = new Stopwatch();

        transformBlock = new TransformBlock<List<Object1>, Tuple<int, List<Object1>>>
           (list =>
        {
            if (!isStarted)
            {
                StartUp();
                isStarted = true;
            }

            return new Tuple<int, List<Object1>>(currentIndex++, list);
        });

        transformBlock.Completion.ContinueWith(_ =>
            {
                ShutDown();
            });
    }

    private void StartUp()
    {
        watch.Start();
    }

    private void ShutDown()
    {
        watch.Stop();

        Console.WriteLine("TransformBlock : Time elapsed in ms: " 
                              + watch.ElapsedMilliseconds);
    }
}

class Object1
{
    public int val { get; private set; }

    public Object1(int val)
    {
        this.val = val;
    }
}

class Object2
{
    public int value { get; private set; }
    public List<Object1> collection { get; private set; }

    public Object2(int value, List<Object1> collection)
    {
        this.value = value;
        this.collection = collection;
    }    
}

*编辑：我发布了另一个代码片段，这次使用值类型的集合，我无法重现我在上面的代码中观察到的问题。是否是传递引用类型并同时对它们进行操作（即使在不同的数据流块中）可能会阻塞并导致争用？*

class Program
{
    static void Main(string[] args)
    {
        Test test = new Test();
        test.Start();
    }
}

class Test
{
    private BroadcastBlock<List<int>> broadCastBlock;
    private TransformBlock<List<int>, List<int>> tfb11;
    private TransformBlock<List<int>, List<int>> tfb12;
    private TransformBlock<List<int>, List<int>> tfb21;
    private TransformBlock<List<int>, List<int>> tfb22;
    private TransformManyBlock<List<int>, List<int>> transformManyBlock1;
    private TransformManyBlock<List<int>, List<int>> transformManyBlock2;
    private ActionBlock<List<int>> actionBlock1;
    private ActionBlock<List<int>> actionBlock2;

    public Test()
    {
        broadCastBlock = new BroadcastBlock<List<int>>(item => item);

        tfb11 = new TransformBlock<List<int>, List<int>>(item =>
            {
                return item;
            });

        tfb12 = new TransformBlock<List<int>, List<int>>(item =>
            {
                return item;
            });

        tfb21 = new TransformBlock<List<int>, List<int>>(item =>
            {
                return item;
            });

        tfb22 = new TransformBlock<List<int>, List<int>>(item =>
            {
                return item;
            });

        transformManyBlock1 = new TransformManyBlock<List<int>, List<int>>(item =>
            {
                Thread.Sleep(100);
                //or you can replace the Thread.Sleep(100) with actual work, 
                //no difference in results. This shows that the issue at hand is 
                //unrelated to starvation of threads.

                return new List<int>[1] { item };
            });

        transformManyBlock2 = new TransformManyBlock<List<int>, List<int>>(item =>
            {
                return new List<int>[1] { item };
            });

        actionBlock1 = new ActionBlock<List<int>>(item =>
            {
                //flush transformManyBlock
            });

        actionBlock2 = new ActionBlock<List<int>>(item =>
        {
            //flush transformManyBlock
        });

        //linking
        broadCastBlock.LinkTo(tfb11, new DataflowLinkOptions 
                                      { PropagateCompletion = true });
        broadCastBlock.LinkTo(tfb12, new DataflowLinkOptions 
                                      { PropagateCompletion = true });
        broadCastBlock.LinkTo(tfb21, new DataflowLinkOptions 
                                      { PropagateCompletion = true });
        broadCastBlock.LinkTo(tfb22, new DataflowLinkOptions 
                                      { PropagateCompletion = true });

        tfb11.LinkTo(transformManyBlock1);
        tfb12.LinkTo(transformManyBlock1);
        tfb21.LinkTo(transformManyBlock2);
        tfb22.LinkTo(transformManyBlock2);

        transformManyBlock1.LinkTo(actionBlock1
                                   , new DataflowLinkOptions 
                                     { PropagateCompletion = true }
                                   );
        transformManyBlock2.LinkTo(actionBlock2
                                   , new DataflowLinkOptions 
                                     { PropagateCompletion = true }
                                   );

        //completion
        Task.WhenAll(tfb11.Completion, tfb12.Completion).ContinueWith(_ =>
            {
                Console.WriteLine("TransformBlocks 11 and 12 completed");
                transformManyBlock1.Complete();
            });

        Task.WhenAll(tfb21.Completion, tfb22.Completion).ContinueWith(_ =>
            {
                Console.WriteLine("TransformBlocks 21 and 22 completed");
                transformManyBlock2.Complete();
            });

        transformManyBlock1.Completion.ContinueWith(_ =>
            {
                Console.WriteLine
                    ("TransformManyBlock (from tfb11 and tfb12) finished");
            });

        transformManyBlock2.Completion.ContinueWith(_ =>
            {
                Console.WriteLine
                    ("TransformManyBlock (from tfb21 and tfb22) finished");
            });
    }

    public void Start()
    {
        const int numberBlocks = 100;
        const int collectionSize = 300000;

        //send collection numberBlock-times
        for (int i = 0; i < numberBlocks; i++)
        {
            List<int> list = new List<int>();
            for (int j = 0; j < collectionSize; j++)
            {
                list.Add(j);
            }

            broadCastBlock.Post(list);
        }

        //mark broadCastBlock complete
        broadCastBlock.Complete();

        Console.WriteLine("Core routine finished");
        Console.ReadLine();
    }
}

score 3 · Accepted Answer

好的，最后的尝试;-)

概要：

场景 1 中观察到的时间增量可以通过垃圾收集器的不同行为来完全解释。

运行场景 1 链接 transformManyBlocks 时，运行时行为是在主线程上创建新项目（列表）期间触发垃圾收集，而运行场景 1 链接 transformManyBlockEmptys 时则不是这种情况。

请注意，创建新的引用类型实例 (Object1) 会导致调用在 GC 堆中分配内存，这反过来可能会触发 GC 收集运行。由于创建了相当多的 Object1 实例（和列表），垃圾收集器需要做更多的工作来扫描堆中（可能）无法访问的对象。

因此，观察到的差异可以通过以下任何方式最小化：

将 Object1 从类转换为结构（从而确保实例的内存不在堆上分配）。
保持对生成列表的引用（从而减少垃圾收集器识别不可达对象所需的时间）。
在将它们发布到网络之前生成所有项目。

（注意：我无法解释为什么垃圾收集器在场景 1“transformManyBlock”与场景 1“transformManyBlockEmpty”中的行为不同，但通过 ConcurrencyVisualizer 收集的数据清楚地显示了差异。）

结果：

（测试在 Core i7 980X、6 核、启用 HT 的情况下运行）：

我将场景 2 修改如下：

// Start a stopwatch per tfb
int tfb11Cnt = 0;
Stopwatch sw11 = new Stopwatch();
tfb11 = new TransformBlock<List<int>, List<int>>(item =>
{
    if (Interlocked.CompareExchange(ref tfb11Cnt, 1, 0) == 0)
        sw11.Start();

    return item;
});

// [...]

// completion
Task.WhenAll(tfb11.Completion, tfb12.Completion).ContinueWith(_ =>
{

     Console.WriteLine("TransformBlocks 11 and 12 completed. SW11: {0}, SW12: {1}",
     sw11.ElapsedMilliseconds, sw12.ElapsedMilliseconds);
     transformManyBlock1.Complete();
});

结果：

方案 1（如发布的，即链接到 transformManyBlock）：
TransformBlock ：经过的时间（以毫秒为单位）：6826
TransformBlock ：经过的时间（以毫秒为单位）：6826
方案 1（链接到 transformManyBlockEmpty）：
TransformBlock：经过的时间（以毫秒为单位）：3140
TransformBlock：经过的时间（以毫秒为单位）：3140
场景 1（transformManyBlock，循环体中的 Thread.Sleep(200)）：
TransformBlock：经过的时间（以毫秒为单位）：4949
TransformBlock：经过的时间（以毫秒为单位）：4950
场景 2（已发布但已修改为报告时间）：
TransformBlocks 21 和 22 已完成。SW21：619 毫秒，SW22：669 毫秒
TransformBlocks 11 和 12 完成。SW11：669 毫秒，SW12：667 毫秒

接下来，我更改了场景 1 和 2，以便在将输入数据发布到网络之前准备好输入数据：

// Scenario 1
//send collection numberBlock-times
var input = new List<List<Object1>>(numberBlocks);
for (int i = 0; i < numberBlocks; i++)
{
    var list = new List<Object1>(collectionSize);
    for (int j = 0; j < collectionSize; j++)
    {
        list.Add(new Object1(j));
    }
    input.Add(list);
}

foreach (var inp in input)
{
    broadCastBlock.Post(inp);
    Thread.Sleep(10);
}

// Scenario 2
//send collection numberBlock-times
var input = new List<List<int>>(numberBlocks);
for (int i = 0; i < numberBlocks; i++)
{
    List<int> list = new List<int>(collectionSize);
    for (int j = 0; j < collectionSize; j++)
    {
        list.Add(j);
    }

    //broadCastBlock.Post(list);
    input.Add(list);
 }

 foreach (var inp in input)
 {
     broadCastBlock.Post(inp);
     Thread.Sleep(10);
 }

结果：

方案 1（transformManyBlock）：
TransformBlock：经过的时间（以毫秒为单位）：1029
TransformBlock：经过的时间（以毫秒为单位）：1029
方案 1（transformManyBlockEmpty）：
TransformBlock：经过的时间（以毫秒为单位）：975
TransformBlock：经过的时间（以毫秒为单位）：975
场景 1（transformManyBlock，循环体中的 Thread.Sleep(200)）：
TransformBlock：经过的时间（以毫秒为单位）：972
TransformBlock：经过的时间（以毫秒为单位）：972

最后，我将代码改回了原始版本，但保留了对创建列表的引用：

var lists = new List<List<Object1>>();
for (int i = 0; i < numberBlocks; i++)
{
    List<Object1> list = new List<Object1>();
    for (int j = 0; j < collectionSize; j++)
    {
        list.Add(new Object1(j));
    }
    lists.Add(list);                
    broadCastBlock.Post(list);
}

结果：

方案 1（transformManyBlock）：
TransformBlock：经过的时间（以毫秒为单位）：6052
TransformBlock：经过的时间（以毫秒为单位）：6052
方案 1（transformManyBlockEmpty）：
TransformBlock：经过的时间（以毫秒为单位）：5524
TransformBlock：经过的时间（以毫秒为单位）：5524
方案 1（transformManyBlock，循环体中的 Thread.Sleep(200)）：
TransformBlock：经过的时间（以毫秒为单位）：5098
TransformBlock：经过的时间（以毫秒为单位）：5098

同样，将 Object1 从一个类更改为一个结构会导致两个块大约同时完成（大约快 10 倍）。

更新：以下答案不足以解释观察到的行为。

在场景一中，TransformMany lambda 内部执行了一个紧密循环，这将占用 CPU 并使其他线程无法获得处理器资源。这就是为什么可以观察到 Completion 延续任务执行延迟的原因。在场景二中，一个 Thread.Sleep 在 TransformMany lambda 中执行，让其他线程有机会执行 Completion 继续任务。观察到的运行时行为差异与 TPL 数据流无关。为了改善观察到的增量，在场景 1 中在循环体内引入一个 Thread.Sleep 就足够了：

for (int counter = 1; counter <= 10000000;  counter++)
{
   double result = Math.Sqrt(counter + 1.0);
   // Back off for a little while
   Thread.Sleep(200);
}

（以下是我的原始答案。我没有仔细阅读OP的问题，只是在阅读了他的评论后才明白他在问什么。我仍然把它留在这里作为参考。）

你确定你测量的是正确的东西吗？请注意，当您执行以下操作时：transformBlock.Completion.ContinueWith(_ => ShutDown());那么您的时间测量将受到 TaskScheduler 行为的影响（例如，在继续任务开始执行之前需要多长时间）。虽然我无法观察到您在我的机器上看到的差异，但在使用专用线程测量时间时，我得到了更精确的结果（就 tfb1 和 tfb2 完成时间之间的增量而言）：

       // Within your Test.Start() method...
       Thread timewatch = new Thread(() =>
       {
           var sw = Stopwatch.StartNew();
           tfb1.transformBlock.Completion.Wait();
           Console.WriteLine("tfb1.transformBlock completed within {0} ms",
                              sw.ElapsedMilliseconds);
        });

        Thread timewatchempty = new Thread(() =>
        {
            var sw = Stopwatch.StartNew();
            tfb2.transformBlock.Completion.Wait();
            Console.WriteLine("tfb2.transformBlock completed within {0} ms", 
                               sw.ElapsedMilliseconds);
        });

        timewatch.Start();
        timewatchempty.Start();

        //send collection numberBlock-times
        for (int i = 0; i < numberBlocks; i++)
        {
          // ... rest of the code

c# - TPL Dataflow，对核心设计感到困惑

1 回答 1

概要：

结果：

Related

Reference