异步任务，视频缓冲

Posted 2023-03-31

【中文标题】异步任务，视频缓冲

【英文标题】：Asynchronous Task, video buffering

【发布时间】：2021-09-19 17:12:10

【问题描述】：

我正在尝试理解 C# 中的任务，但仍有一些问题。我正在尝试创建一个包含视频的应用程序。主要目的是从文件中读取视频（我正在使用 Emgu.CV）并通过 TCP/IP 将其发送到板上进行处理，然后以流（实时）方式返回。首先，我是按顺序完成的。因此，阅读Bitmap，从板上发送-接收，并进行绘图。但是读取位图并绘制它们需要太多时间。我想要一个传输、接收 FIFO 缓冲区来保存视频帧，以及一个不同的任务来完成发送接收每个帧的工作。所以我想并行进行。我想我应该创建 3 个任务：

        tasks.Add(Task.Run(() => Video_load(video_path)));
        tasks.Add(Task.Run(() => Video_Send_Recv(video_path)));
        tasks.Add(Task.Run(() => VideoDisp_hw(32)));

我想“并行”运行。我应该使用什么类型的对象？并发队列？缓冲块？还是只是一个列表？

感谢您的建议！我想问一些事情。我正在尝试使用 2 个 TPL 块创建一个简单的控制台程序。 1 块将是转换块（接收消息，即“开始”）并将数据加载到列表，另一个块将是 ActionBlock（仅从列表中读取数据并打印它们）。下面是代码：

namespace TPL_Dataflow

    class Program
    
        static void Main(string[] args)
        
            Console.WriteLine("Hello World!");
            Random randn = new Random();

            var loadData = new TransformBlock<string, List<int>>(async sample_string =>
           
               List<int> input_data = new List<int>();
               int cnt = 0;

                if (sample_string == "start")
                
                   Console.WriteLine("Inside loadData");
                   while (cnt < 16)
                   
                       input_data.Add(randn.Next(1, 255));
                       await Task.Delay(1500);
                       Console.WriteLine("Cnt");
                       cnt++;
                   
                                    
                else
                
                    Console.WriteLine("Not started yet");

                
            return input_data;
           );


            var PrintData = new ActionBlock<List<int>>(async input_data =>
            
                while(input_data.Count > 0)
                


                    Console.WriteLine("output Data = " + input_data.First());
                    await Task.Delay(1000);
                    input_data.RemoveAt(0);
                    
                
 

              );

            var linkOptions = new DataflowLinkOptions  PropagateCompletion = true ;
            loadData.LinkTo(PrintData, input_data => input_data.Count() >0  );
            //loadData.LinkTo(PrintData, linkOptions);
            
            loadData.SendAsync("start");
            loadData.Complete();
            PrintData.Completion.Wait();

        
    

但它似乎以串行方式工作..我做错了什么？我尝试异步执行while循环。我想同时做两件事。当列表中的数据可用时，然后绘制。

【问题讨论】：

overview 中的 TPL 数据流库指的是一个处理来自网络摄像头的图像数据的应用程序，作为其原型用例。所以你可能想试一试。

您好@TheodorZoulias，感谢您的帮助！那么，据我了解，它可能是一个源块（加载视频）、一个传播块（接收和传输数据）和一个目标块（将数据写入图像输出）？我正在尝试理解 TPL 数据流，但仍然遇到一些问题......有没有最简单的方法通过简单的编程来完成并行性？任务没有帮助，对吧？

TPL Dataflow 的思想是创建一个由块组成的管道（通常是TransformBlocks），然后将原始数据馈送到第一个块，让数据流过管道，被处理一路上。你可以看到一个例子here。使用此库创建多步处理器比手动管理任务要容易得多。它有一个学习曲线，但不是太陡峭。一两天的学习，你就可以成功使用了。

我在第一篇文章中添加了更新

【参考方案1】：

您可以使用TransformManyBlock<string, int> 作为生产者块，并使用ActionBlock<int> 作为消费者块。 TransformManyBlock 将使用接受 Func<string, IEnumerable<int>> 委托的构造函数进行实例化，并传递一个 iterator method（下例中的 Produce 方法），该方法会一一产生值：

Random random = new Random();

var producer = new TransformManyBlock<string, int>(Produce);

IEnumerable<int> Produce(string message)

    if (message == "start")
    
        int cnt = 0;
        while (cnt < 16)
        
            int value;
            lock (random) value = random.Next(1, 255);
            Console.WriteLine($"Producing #value");
            yield return value;
            Thread.Sleep(1500);
            cnt++;
        
    
    else
    
        yield break;
    


var consumer = new ActionBlock<int>(async value =>

    Console.WriteLine($"Received: value");
    await Task.Delay(1000);
);

producer.LinkTo(consumer, new()  PropagateCompletion = true );

producer.Post("start");
producer.Complete();
consumer.Completion.Wait();

不幸的是，生产者必须在产生每个值（Thread.Sleep(1500);）之间的空闲期间阻塞工作线程，因为TransformManyBlock 目前没有接受Func<string, IAsyncEnumerable<int>> 的构造函数。这可能会在TPL Dataflow 库的下一个版本中得到修复。您可以跟踪this GitHub issue，以了解此功能何时发布。

---

替代解决方案：您可以不将生产者和消费者显式链接，而是将它们取消链接，并手动将生产者生成的值发送给消费者。在这种情况下，两个块都是ActionBlocks：

Random random = new Random();

var consumer = new ActionBlock<int>(async value =>

    Console.WriteLine($"Received: value");
    await Task.Delay(1000);
);

var producer = new ActionBlock<string>(async message =>

    if (message == "start")
    
        int cnt = 0;
        while (cnt < 16)
        
            int value;
            lock (random) value = random.Next(1, 255);
            Console.WriteLine($"Producing #value");
            var accepted = await consumer.SendAsync(value);
            if (!accepted) break; // The consumer has failed
            await Task.Delay(1500);
            cnt++;
        
    
);

PropagateCompletion(producer, consumer);

producer.Post("start");
producer.Complete();
consumer.Completion.Wait();

async void PropagateCompletion(IDataflowBlock source, IDataflowBlock target)

    try  await source.Completion.ConfigureAwait(false);  catch  
    var ex = source.Completion.IsFaulted ? source.Completion.Exception : null;
    if (ex != null) target.Fault(ex); else target.Complete();

这种方法的主要困难是如何将生产者的完成传播给消费者，以便最终完成两个块。显然你不能使用new DataflowLinkOptions PropagateCompletion = true 配置，因为这些块没有显式链接。你也不能Complete手动消费者，因为在这种情况下它会过早地停止接受来自生产者的值。这个问题的解决方法就是上面例子中的PropagateCompletion方法。

【讨论】：

非常感谢您的帮助！！我试图加载位图而不是代码，但图像没有更新。我应该使用锁还是什么？此外，一段时间后，应用程序由于内存分配过多而崩溃。有没有办法在 TransformManyBLock 中进行“流量控制”？ （抱歉一直问，我对 C# 和面向对象编程很陌生...）

嗨@ThKont，我很高兴这很有帮助。 TPL 数据流有一个很好的方法来控制内存：BoundedCapacity 选项。通常管道中的所有块都应该配置有这个选项，除了第一个。关于非更新图像，如果您没有找到解决方案，您可以考虑发布一个新问题，以及一个最小且可重现的示例，以便我们查看。