问题标签 [bufferblock]

假设我有BufferBlock一个有限容量为 2 的 a，我将它链接（使用LinkTo()）一个具有 MaxDegreeOfParallelism = 2 的 ActionBlock。现在，我知道我将能够立即将 2 个项目发送到缓冲区，并且 ActionBlock 将立即开始处理它们. 但是，假设这些操作需要几秒钟才能完成。当第一个操作正在运行时，我是否能够再向缓冲区发送 2 个项目，或者仅在消耗它的操作完成后才从缓冲区中删除一个项目？

我正在尝试使用着色器存储缓冲区对象（又名缓冲区块），但有几件事我没有完全掌握。我要做的是在其中存储不确定数量的灯光的（简化）数据n，以便我的着色器可以遍历它们并执行计算。

首先让我说我得到了正确的结果，并且没有来自 OpenGL 的错误。然而，它困扰我不知道它为什么工作。

所以，在我的着色器中，我得到了以下信息：

在我的应用程序中：

在最后一个循环中，我存储了一个PointLightData结构，其偏移量等于它的大小乘以我已经存储的它们的数量（所以0第一个是偏移量）。

所以，正如我所说，一切似乎都是正确的。绑定点正确设置为零，我为我的对象分配了足够的内存等。图形结果还可以。

现在回答我的问题。我正在std430用作布局 - 事实上，如果我将其更改std140为我最初所做的那样，它就会中断。这是为什么？我的假设是std430着色器PointLights缓冲区块生成的布局与编译器为我的应用程序PointLightData结构生成的布局很匹配（正如您在该循环中看到的那样，我一个接一个地存储）。你认为是这样吗？

现在，假设我的假设是正确的，显而易见的解决方案是自己进行大小和偏移量的映射，用glGetUniformIndicesand查询opengl glGetActiveUniformsiv（后者用GL_UNIFORM_SIZEand调用GL_UNIFORM_OFFSET），但我偷偷怀疑这两个家伙只有使用统一块而不是像我正在尝试做的缓冲块。至少，当我执行以下操作时，OpenGL 会发脾气，给我一个1281错误并返回一个非常奇怪的数字作为索引（类似3432898282或其他）：

我这样说是否正确glGetUniformIndices并且glGetActiveUniformsiv不适用于缓冲块？

如果他们不这样做，或者它工作的事实就像我想象的只是巧合，我怎么能手动进行映射？我检查了编程指南的附录 H，结构数组的措辞有些混乱。如果我无法查询 OpenGL 的大小/偏移量，我想我可以手动计算它们（虽然很麻烦），但我也很感激那里的一些帮助。

我正在尝试为 TPL BufferBlock 创建单元测试，并希望测试是否引发了异常。但是，测试在抛出异常之前通过。

编辑

此外，这是一个长期运行的过程，所以我不称之为完成。此过程一直运行到应用程序关闭

这是代码：

这是测试代码

ArgumentException 异常在 SetMessagePriority 中被正确抛出，因为“MessageType”等于 MessageType.Unknown。但是，在引发 ArgumentException 时，单元测试 EnqueueTest_UnknownMessageType 已成功通过，因为

在抛出异常之前返回“true”。如何编写测试 EnqueueTest_UnknownMessageType 以使其因抛出异常而失败？

我试过添加

测试，但它仍然通过

我想使用连续运行的 BufferBlock 来实现消费者/生产者模式，类似于这里的问题和这里的代码。

我尝试使用像 OP 这样的 ActionBlock，但如果缓冲块已满且新消息在其队列中，则新消息永远不会添加到 ConcurrentDictionary _queue。

在下面的代码中，当使用此调用将新消息添加到缓冲区块时，永远不会调用 ConsumeAsync 方法：_messageBufferBlock.SendAsync(message)

如何更正下面的代码，以便每次使用添加新消息时调用 ConsumeAsync 方法_messageBufferBlock.SendAsync(message)？

编辑 我已经删除了所有额外的代码并添加了注释。

我编写了一个小程序来测试使用 BufferBlock (System.Threading.Tasks.Dataflow) 来实现双优先级消费者-生产者队列。

消费者应始终首先使用高优先级队列中的任何项目。

在这个初始测试中，我让生产者运行的速度比消费者慢得多，所以数据应该按照它进入的顺序出来，不管优先级如何。

但是，我发现Task.WhenAny()直到两个队列中都有东西（或有完成），结果才完成，因此表现得像Task.WhenAll().

我以为我理解async/ await，并且我仔细阅读了 Cleary 的“C# Cookbook 中的并发性”。但是，发生了一些我不明白的事情。

有任何想法吗？

代码：

输出：

我认为这是非常基本的方法，但我还没有找到任何例子。我有一个生产者和一个消费者，我想在至少处理 x 个对象时完成管道。此外，我需要知道收到了哪些对象。

我就是这样做的：

我对 buffer.TryReceiveAll 有点困惑。等待消费任务和 TryReceiveAll 有什么区别？为什么 TryReceiveAll 在我的场景中为假？我想我实现目标的方法仍然有问题。

我有一个像这样的 BufferBlock 设置。

让多个消费者从不同的线程调用“FetchWork”函数

有时，同一个工作项最终会出现在多个消费者中！InputQueue 中的工作项数量越多，在 GetWork 中收到重复项的机会就越大。我的理解是通过 ReceiveAsync 获取的项目是原子的，一旦读取一个项目，就不会再次读取。这不会发生在这里。我有40 个并行消费者调用 GetWork。

我想实现类似两阶段提交协议的东西来消费消息。

为了做到这一点，我实现了ITargetBlock自己：

在文档中它说ConsumeMessage可以返回与以前提供的不同的实例。

我想知道它发生在哪些情况和方式？

我有一个问题，我有分配数据进入系统，因为它被存储在磁盘上，然后通过管道进行转换。

管道中的每个步骤都可能受 CPU 限制，但我的整体问题是内存，这意味着我必须对我的处理量设置一个限制。

在尝试使用 BoundedChannels 进行初始实现后，很容易看到内存消耗迅速增长，超出了可用的范围（我的开发机器分配了更多内存，所以这里没问题，但它会崩溃）......

总体概念是：

问题是，如果通道例如允许 100.000 个元素，当第一个被填充时，一切正常，但显然整个管道的限制随着我们执行步骤而增加，3 步骤限制突然变为 100.000 * 3...

办公室。我们可以说限制是 33.000，这在整个管道中总共约为 100.000，但是我希望允许平衡所有通道的限制，这样如果我们添加更多步骤或调整，我们就不必调整它这取决于它最密集的地方......

我不太清楚是否允许这种管道的数据流框架？

（对不起，这很抽象，另一种可能是我必须从下往上设计）