c# - 如何以同步方式合并两个 TPL DataFlow 管道？
问题描述
我想编写一个应用程序来评估来自两个传感器的传感器数据。两个传感器都以对象的形式发送数据，Package这些对象被拆分为Frame对象。APackage本质上是a Tuple<Timestamp, Data[]>，aFrame是a Tuple<Timestamp, Data>。然后我需要始终Frame使用来自两个来源的最早时间戳。
所以基本上我的对象流是
Package -(1:n)-> Frame \
}-pair synchronized-> Tuple<Frame, Frame>
Package -(1:n)-> Frame /
例子
假设每个Package包含 2 或 3 个值（现实：5-7）和以 1 递增的整数时间戳（现实：~200Hz => ~5ms 增量）。“数据”只是timestamp * 100为了简单起见。
Packages (timestamp, values[])
Source 1:
{(19, [1700, 1800, 1900]), (22, [2000, 2100, 2200]), (26, [2500, 2600]),
(29, [2700, 2800, 2900]), ...}
Source 2:
{(17, [1500, 1600, 1700]), (19, [1800, 1900]), (21, [2000, 2100]),
(26, [2400, 2500, 2600]), ...}
之后(1:n)的步骤：
Frames (timestamp, value)
Source 1:
{(17, 1700), (18, 1800), (19, 1900), (20, 2000), (21, 2100),
(22, 2200), (25, 2500), (26, 2600), (27, 2700), (28, 2800),
(29, 2900), ...}
Source 2:
{(15, 1500), (16, 1600), (17, 1700), (18, 1800), (19, 1900),
(20, 2000), (21, 2100), (24, 2400), (25, 2500), (26, 2600), ...}
步骤后pair synchronized：
Merged tuples (timestamp, source1, source2)
{(15, null, 1500), (16, null, 1600), (17, 1700, 1700), (18, 1800, 1800),
(19, 1900, 1900), (20, 2000, 2000), (21, 2100, 2100), (22, 2200, null),
(24, null, 2400), (25, 2500, 2500), (26, 2600, 2600), ...}
请注意，由于两个源都没有发送值，23因此缺少时间戳。这只是一个副作用。我可以放一个空元组，也可以不放，没关系。元组是否是or也无关紧要，即or 。(27, 2700, 2700)((27, 2700), (27, 2700))Tuple<Timestamp, Data, Data>Tuple<Frame, Frame>
如果我得到正确的文档，我很确定这(1:n)部分应该是。TransformManyBlock<Package, Frame>
但是我该使用哪个块 pair synchronized ？起初，我认为这JoinBlock<Frame, Frame>将是我正在寻找的东西，但它似乎只是将两个元素按索引配对。但是，由于既不能确保两个管道都以相同的时间戳开始，也不能确保两个管道总是产生稳定的连续时间戳流（因为有时具有几帧的包可能会在传输中丢失），这不是一个选择。所以我需要更多的是一个“MergeBlock”，它可以决定两个输入流的哪个元素接下来传播到输出（如果有的话）。
我想我必须自己写这样的东西。但是我无法编写正确处理两个 ISourceBlock 变量和一个 ITargetBlock 变量的代码。我基本上被卡住了：
private void MergeSynchronized(
ISourceBlock<Frame> source1,
ISourceBlock<Frame> source2,
ITargetBlock<Tuple<Frame, Frame>> target)
{
var frame1 = source1.Receive();
var frame2 = source2.Receive();
//Loop {
// Depending on the timestamp [mis]match,
// either pair frame1+frame2 or frame1+null or null+frame2, and
// replace whichever frame(s) was/were propagated already
// with the next frame from the respective pipeline
//}
}
我什至不确定这个草稿：该方法应该是async我可以使用的var frame1 = await source1.ReceiveAsnyc();吗？循环的条件是什么？在哪里以及如何检查是否完成？如何解决我的代码意味着我必须等到流中的间隙结束才能意识到存在间隙的明显问题？
我考虑的替代方法是在管道中添加一个额外的块，确保每个传感器有足够的“哨兵帧”放入管道中，以便始终对齐每个管道中的第一个将对齐正确的两个。我猜这将是一种TransformManyBlock读取帧，将“预期”时间戳与实际时间戳进行比较，然后为丢失的时间戳插入哨兵帧，直到帧的时间戳再次正确为止。
或者该pair synchronized部分是停止使用 TPL 数据流对象并启动已经与该Data部分一起使用的实际代码的地方吗？
解决方案
TPL DataFlow API 的问题在于，一切都是内部/私有和/或密封的。这使您没有太多扩展 API 的可能性。
无论如何，对于您的问题，实现一个新的 SynchronizedJoinBlock 类可能是个好主意。实际的业务逻辑位于 GetMessagesRecursive 方法中：
public sealed class SynchronizedJoinBlock<T1, T2>
: IReceivableSourceBlock<Tuple<T1, T2>>
{
private readonly object _syncObject = new object();
private readonly Func<T1, T2, int> _compareFunction;
private readonly Queue<T1> _target1Messages;
private readonly Queue<T2> _target2Messages;
private readonly TransformManyBlock<T1, Tuple<T1, T2>> _target1;
private readonly TransformManyBlock<T2, Tuple<T1, T2>> _target2;
private readonly BatchedJoinBlock<Tuple<T1, T2>, Tuple<T1, T2>> _batchedJoinBlock;
private readonly TransformManyBlock<Tuple<IList<Tuple<T1, T2>>, IList<Tuple<T1, T2>>>, Tuple<T1, T2>> _transformManyBlock;
public ITargetBlock<T1> Target1 => _target1;
public ITargetBlock<T2> Target2 => _target2;
public Task Completion => _transformManyBlock.Completion;
public SynchronizedJoinBlock(Func<T1, T2, int> compareFunction)
{
_compareFunction = compareFunction
?? throw new ArgumentNullException(nameof(compareFunction));
_batchedJoinBlock = new BatchedJoinBlock<Tuple<T1, T2>, Tuple<T1, T2>>(1);
_target1Messages = new Queue<T1>();
_target2Messages = new Queue<T2>();
Func<ICollection<Tuple<T1, T2>>> getMessagesFunction = () =>
{
lock (_syncObject)
{
if (_target1Messages.Count > 0 && _target2Messages.Count > 0)
{
return GetMessagesRecursive(_target1Messages.Peek(), _target2Messages.Peek()).ToArray();
}
else
{
return new Tuple<T1, T2>[0];
}
}
};
_target1 = new TransformManyBlock<T1, Tuple<T1, T2>>((element) =>
{
_target1Messages.Enqueue(element);
return getMessagesFunction();
});
_target1.LinkTo(_batchedJoinBlock.Target1, new DataflowLinkOptions() { PropagateCompletion = true });
_target2 = new TransformManyBlock<T2, Tuple<T1, T2>>((element) =>
{
_target2Messages.Enqueue(element);
return getMessagesFunction();
});
_target2.LinkTo(_batchedJoinBlock.Target2, new DataflowLinkOptions() { PropagateCompletion = true });
_transformManyBlock = new TransformManyBlock<Tuple<IList<Tuple<T1, T2>>, IList<Tuple<T1, T2>>>, Tuple<T1, T2>>(
element => element.Item1.Concat(element.Item2)
);
_batchedJoinBlock.LinkTo(_transformManyBlock, new DataflowLinkOptions() { PropagateCompletion = true });
}
private IEnumerable<Tuple<T1, T2>> GetMessagesRecursive(T1 value1, T2 value2)
{
int result = _compareFunction(value1, value2);
if (result == 0)
{
yield return Tuple.Create(_target1Messages.Dequeue(), _target2Messages.Dequeue());
}
else if (result < 0)
{
yield return Tuple.Create(_target1Messages.Dequeue(), default(T2));
if (_target1Messages.Count > 0)
{
foreach (var item in GetMessagesRecursive(_target1Messages.Peek(), value2))
{
yield return item;
}
}
}
else
{
yield return Tuple.Create(default(T1), _target2Messages.Dequeue());
if (_target2Messages.Count > 0)
{
foreach (var item in GetMessagesRecursive(value1, _target2Messages.Peek()))
{
yield return item;
}
}
}
}
public void Complete()
{
_target1.Complete();
_target2.Complete();
}
Tuple<T1, T2> ISourceBlock<Tuple<T1, T2>>.ConsumeMessage(
DataflowMessageHeader messageHeader,
ITargetBlock<Tuple<T1, T2>> target, out bool messageConsumed)
{
return ((ISourceBlock<Tuple<T1, T2>>)_transformManyBlock)
.ConsumeMessage(messageHeader, target, out messageConsumed);
}
void IDataflowBlock.Fault(Exception exception)
{
((IDataflowBlock)_transformManyBlock).Fault(exception);
}
public IDisposable LinkTo(ITargetBlock<Tuple<T1, T2>> target,
DataflowLinkOptions linkOptions)
{
return _transformManyBlock.LinkTo(target, linkOptions);
}
void ISourceBlock<Tuple<T1, T2>>.ReleaseReservation(
DataflowMessageHeader messageHeader, ITargetBlock<Tuple<T1, T2>> target)
{
((ISourceBlock<Tuple<T1, T2>>)_transformManyBlock)
.ReleaseReservation(messageHeader, target);
}
bool ISourceBlock<Tuple<T1, T2>>.ReserveMessage(
DataflowMessageHeader messageHeader, ITargetBlock<Tuple<T1, T2>> target)
{
return ((ISourceBlock<Tuple<T1, T2>>)_transformManyBlock)
.ReserveMessage(messageHeader, target);
}
public bool TryReceive(Predicate<Tuple<T1, T2>> filter, out Tuple<T1, T2> item)
{
return _transformManyBlock.TryReceive(filter, out item);
}
public bool TryReceiveAll(out IList<Tuple<T1, T2>> items)
{
return _transformManyBlock.TryReceiveAll(out items);
}
}
推荐阅读
- botframework - 从 Simple Azure bot 调用 Graph API 的步骤
- python - 为什么这个幂函数在负指数情况下返回零？
- regex - 如何将一个字符串与另一个字符串进行比较，其中一个字符串之间有空格
- javascript - 使用 datazoom 进行缩放时出现 eCharts xAxis 错误
- javascript - 将 WebRTC 中的 MediaStream 服务器端（带有视频和音频）录制到文件中
- ssl - 如何绕过 Angular2-Nativescript 上的 SSL 错误？
- javascript - 尝试将图像放入蛇游戏时出现“NS_ERROR_NOT_AVAILABLE”
- azure - PowerShell 失败时将 Azure AD 帐户添加为 SQL 登录
- javascript - 使用 onclick 从另一个函数中调用函数
- bash - If-Statement 疯了（还是我？）