TPL数据流，对核心设计感到困惑

Question

我使用了相当多的TPL数据流，但在一个我无法解决的问题上却步履蹒跚：

我有以下架构：

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链接到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();
    }
}

Answer 1

好的，最后一次;-)

简介：

场景1中观察到的时间增量可以由垃圾收集器的不同行为充分解释。。

当运行场景1链接transformManyBlocks时，运行时行为会在主线程上创建新项(列表)期间触发垃圾回收，而在运行带有transformManyBlockEmptys链接的场景1时则不是这样。

注意，创建一个新的引用类型实例(Object1)将导致调用在GC堆中分配内存，这反过来可能触发GC集合运行。由于创建了相当多的Object1实例(和列表)，垃圾收集器有相当多的工作要做--扫描堆以查找(可能)不可访问的对象。

因此，所观察到的差异可以通过以下任何一种方法尽量减少：

• 将Object1从类转换为结构(从而确保没有在堆上分配实例的内存)。
• 保持对生成列表的引用(从而减少垃圾收集器需要识别不可访问对象的时间)。
• 在将所有项目投递到网络之前生成所有项目。

(注意:我无法解释为什么垃圾收集器在场景1 "transformManyBlock“与场景1 "transformManyBlockEmpty”中的行为不同，但是通过ConcurrencyVisualizer收集的数据清楚地显示了两者之间的区别。)

结果：

(在核心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. 场景1(如发布的，即链接到transformManyBlock)： TransformBlock : ms: 6826所用时间 TransformBlock : ms: 6826所用时间
2. 场景1(链接到transformManyBlockEmpty)： TransformBlock : ms: 3140所用时间 TransformBlock : ms: 3140所用时间
3. 场景1(循环体中的transformManyBlock，Thread.Sleep(200) )： TransformBlock : ms: 4949所用时间 TransformBlock : ms: 4950所用的时间
4. 场景2(发布但修改为报告时间)： TransformBlocks 21和22完成。SW21: 619 ms，SW22: 669 ms TransformBlocks 11和12完成。SW11: 669 ms，SW12: 667 ms

接下来，我修改了场景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. 场景1 (transformManyBlock)： TransformBlock : ms: 1029所用时间 TransformBlock : ms: 1029所用时间
2. 场景1 (transformManyBlockEmpty)： TransformBlock : ms: 975所用的时间 TransformBlock : ms: 975所用的时间
3. 场景1(循环体中的transformManyBlock，Thread.Sleep(200) )： TransformBlock : ms: 972所用时间 TransformBlock : ms: 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. 场景1 (transformManyBlock)： TransformBlock : ms: 6052所用时间 TransformBlock : ms: 6052所用时间
2. 场景1 (transformManyBlockEmpty)： TransformBlock : ms: 5524所用时间 TransformBlock : ms: 5524所用时间
3. 场景1(循环体中的transformManyBlock，Thread.Sleep(200) )： TransformBlock : ms: 5098所用时间 TransformBlock : ms: 5098所用时间

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

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

在场景1中，在TransformMany lambda中执行一个紧循环，这将占用CPU，并将饥饿其他线程以获得处理器资源。这就是为什么可以观察到延迟执行完成继续任务的原因。在场景二中，在Thread.Sleep lambda中执行一个TransformMany，给其他线程执行完成延续任务的机会。观察到的运行时行为差异与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