前言
续上文
一、突发 IO
对于那些需要小型且可能是非连续的数据块（或突发）以及频繁变化的元数据的应用程序，突发输入/输出（BurstIO）提供了满足这些要求的数据传输容器和接口。这个接口仅支持数据向量的传输：float, double, octet (int8/uint8), short (int16), ushort (uint16), long (int32), ulong (uint32), longlong (int64), 和 ulonglong(uint64)。与批量输入/输出（BulkIO）类似，BurstIO 提供了突发信号相关信息（SRI）和精确时间戳，但是它通过每个数据突发中的带内信息提供这些信息。由于元数据的增加的开销要求，通过将多个突发分组为单次传输，无论是通过编程还是通过可配置的策略设置，BurstIO 可以实现其最高吞吐量，以尝试最大化效率并限制延迟。
1、数据传输
突发输入/输出（BurstIO）数据传输通过 REDHAWK 资源（组件或设备）的 BurstIO 端口对象的 pushBurst() 和 pushBursts() 方法调用来实现。资源可以使用这些 push 方法将突发及其关联的元数据从一个资源传输到另一个资源，这一过程在资源的服务功能内进行。与批量输入/输出（BulkIO）相似，BurstIO 接口为突发的每个数据向量提供相同的 BULKIO::PrecisionUTCTime 时间戳。BurstIO 定义了一个新的 BurstSRI 信号相关信息（SRI）对象，使开发人员能够进一步描述信号环境和数据转换。这些字段在突发信号相关信息（SRI）中有进一步的描述。
①、输入
拥有提供端口（输入端口）的资源，使用 getBurst() 方法从端口抓取数据。此方法从输入端口的数据队列返回一个 PacketType 对象（在突发包访问器中描述），如果队列为空，则返回 null/None 值。
以下代码片段是 getBurst() 方法的一个示例。
/** Grab data from the port's getBurst method */ burstio::BurstShortIn::PacketType *pkt; pkt = inShortPort->getBurst( bulkio::Const::NON_BLOCKING ); // check if a valid packet was returned if ( pkt == NULL ) { return NOOP; } // check for EOS if ( pkt->getEOS() ) { outShortPort->pushBurst(pkt->getSequence(), pkt->getSRI(), pkt->getEOS()); } ... perform algorithm on the data: pkt->getData() ... or pkt->getSequence()
②、输出
由于 BurstIO 数据的异步性，该接口使开发者能够控制从资源中输出（输出）突发。将突发数据从一个资源推送到另一个资源的两个主要方法调用是：pushBursts() 和 pushBurst()。pushBursts() 允许直接将多个突发作为 BurstType 对象的序列直接下游发送，而 pushBurst() 提供了一个接口来排队推送单个突发，但根据突发的数量、总队列大小和发送间隔遵循策略指令。这两种方法都使用指定的路由约束和连接过滤器路由突发数据，这些可以通过以下接口控制：
// this route streams with Stream ID == "data-stream-one" to a connection // identified as "connection-one" shortBurstPort->addConnectionFilter("data-stream-one", "connection-one");
或者：
// update connection filter using the Component's connection property // "myConnectionTable" shortBurstPort->updateConnectionFilter(myConnectionTable); // this sets the stream filter to only route streams to specific connections shortBurstPort->setRoutingMode(burstio::ROUTE_CONNECTION_STREAMS);
pushBurst() 和 pushBursts() 方法之间的主要区别在于管理数据传输的方式和时间的能力。只有使用 pushBurst() 排队的突发流量受到策略约束的控制，而对 pushBursts() 的调用则直接发送到下游连接的资源。
// this method will limit the maximum number of bursts that // can be queued before they are sent shortBurstPort->setMaxBursts(size_t count); // this method will enable threshold monitoring for the amount of sample // data that exceeds this limit before sending data downstream shortBurstPort->setByteThreshold(size_t bytes); // this method will enable the latency time between the sending of // available data downstream shortBurstPort->setLatencyThreshold( long usec );
以下代码片段是一个使用 pushBurst() 方法调用将向量数据样本排队到端口的示例。
std::vector< BurstShortOut::NativeType > data; my_transform(data); BURSTIO::BurstSRI sri; burstio::BurstShortOut::BurstType burst; burst.SRI = sri; burst.EOS = false; burst.T = burstio::utils::now(); burst.data.length(data.size()); for(int i=0; i< data.size(); i++ ) burst.data[i] = data[i]; // this queues a single burst shortBurstPort->pushBurst( burst ); // or std::vector< BurstShortOut::NativeType > data; my_transform(data); // this queues a single burst shortBurstPort->pushBurst( data, sri, burstio::utils::now() );
以下代码片段是一个使用 pushBursts() 方法调用的向量数据样本的示例。此调用中的突发直接传递到下游的连接资源。
std::vector< BurstShortOut::NativeType > data; my_transform(data); BurstShortOut::BurstSequenceType bursts; bursts.length(1); burstio::BurstShortOut::BurstType burst; burst.SRI = sri; burst.EOS = false; burst.T = burstio::utils::now(); burst.data.length(data.size()); for(int i=0; i< data.size(); i++ ) burst.data[i] = data[i]; bursts[0] = burst; // this pushes the burst directly downstream because // it is a sequence of bursts shortBurstPort->pushBursts(bursts);
2、突发信号相关信息 (SRI)
BurstSRI 对象随每个数据突发一起传送，并描述数据生产者的数据负载和处理状态。下表仅描述了在资源之间传递突发数据时数据结构的必需字段。
3、多输出端口
每个输出突发输入/输出（BurstIO）端口类型都提供了基于流 ID 和连接 ID 过滤来自资源的突发数据的能力。要使用端口的多出能力，资源必须包含类似于以下的代码：
<structsequence id="connectionTable"> <struct id="connectionTable::connection_descriptor" name="connection_descriptor"> <simple id="connectionTable::connection_id" name="connection_id" type="string"> <kind kindtype="configure"/> </simple> <simple id="connectionTable::stream_id" name="stream_id" type="string"> <kind kindtype="configure"/> </simple> <simple id="connectionTable::port_name" name="port_name" type="string"> <kind kindtype="configure"/> </simple> </struct> <configurationkind kindtype="configure"/> </structsequence>
为了将特定的数据流引导到特定的连接，需要将 connectionTable 对象传递给端口的 updateConnectionFilter 方法。当路由模式设置为 ROUTE_CONNECTION_STREAMS 时，端口将会将过滤状态应用于通过资源的 BurstIO 端口传出的任何突发流量。要将突发传递给现有连接，端口的过滤器表中必须存在与下游资源的突发的流 ID 和连接 ID 匹配的项。
4、使用复数数据
每个传入数据的 BurstPacket 提供了 getComplex()` 方法，以表示向量是否包含复数样本（它由实部和虚部组成）。复数数据以交替的实部和虚部值发送。开发者可以以任何方式处理这些数据；然而，本节描述了将数据转换为更易于处理形式的常用方法。
①、在 C++ 中转换复数数据
在 C++ 中，传入的突发输入/输出（BurstIO）数据向量可以被强制转换为复数值的 std::vector。例如：
BurstShortIn::BurstPacket *pkt = myShortPort->getPacket(bulkio::Const::BLOCKING); if ( pkt->isComplex() ) { BurstShortIn::ComplexType cplx_data = pkt->getComplexData(); // ... do some processing with cplx_data }
5、时间戳
以下代码段提供了一个示例，展示如何构造一个要在突发信号相关信息（SRI）中发送的 BULKIO::PrecisionUTCTime 时间戳。
/** * To create a time stamp from the current time of day */ BULKIO::PrecisionUTCTime tstamp = burstio::utils::now();
6、端口统计
所有突发输入/输出（BurstIO）端口支持批量输入/输出（BulkIO）统计接口，并添加了跟踪特定于突发的指标的额外关键词。统计数据是在 10 次 pushBurst 调用的窗口中跟踪的。输入端口包含单个 PortStatistics 结构，而输出端口包含一系列 PortStatistics 结构；每个连接一个结构。有关 BULKIO::PortStatistics 的更多信息，请参见端口统计。输入和输出端口的额外 BurstIO 指标在以下表格中描述：
①、C++
以下示例说明了一个组件，该组件对传入的突发数据进行转换，并将结果向下游推送。
burstio::BurstShortIn::PacketType *pkt; pkt = inShortPort->getPacket(bulkio::Const::NON_BLOCKING); // check if a valid packet was returned if ( pkt == NULL ) { return NOOP; } // check for EOS if ( pkt->getEOS() ) { outShortPort->pushBurst(pkt->getSequence(), pkt->getSRI(), pkt->getEOS()); } // do some processing.....to the burst contents BurstShortOut::SequenceType data = do_some_magic(pkt->getSequence()); // we changed the data so calc new time stamp.... BULKIO::PrecisionUTCTime newTS = calc_timestamp(pkt->getTime()); outShortPort->pushBurst(data, pkt->getSRI(), newTS, pkt->getEOS());
二、消息传递
消息传递依赖于 CORBA 的事件结构作为传输结构。在 CORBA 的事件 API 中，消息通过使用函数 push() 以 Any 类型传递。
虽然 CORBA 管理数据的编组和传递，但它并未提供任何固有于事件的机制来描述 Any 类型的内容。REDHAWK 决定利用现有的负载结构描述符来描述消息的负载，即属性接口描述语言（IDL）。选择此接口消除了创建描述消息的新 IDL 的需要。此外，已有一种 XML 结构映射到高效的二进制数据结构，允许使用 XML 来描述消息内容，同时消除了在消息传递机制中引入 XML 解析器的需要。
为了支持这项额外功能，REDHAWK 扩展了属性描述符，允许属性具有消息类型。唯一可以具有有效消息类型的属性是结构。
1、消息生产者
在创建新组件或编辑现有组件时，可以创建一个消息生产者。创建消息生产者后，您必须注册您的代码，以便从端口发送消息。以下程序解释了如何创建消息生产者并发送消息。
①、创建一个消息生产者
使用 REDHAWK IDE 向组件或设备添加消息生产者端口，请遵循以下步骤：
- 1）从项目资源管理器视图中，双击组件的软件包描述符（SPD）文件。此时将显示组件编辑器。
- 2）在组件编辑器中，选择“属性”标签。组件编辑器的属性标签页将被显示。
- 3）要添加结构属性，请点击“添加结构”。此时将显示属性标签的结构属性部分。
- 4）在结构属性部分，输入产生的消息的名称。ID 默认为您输入的名称。从“类型（Kind）”下拉菜单中，选择“消息（message）。
REDHAWK——连接（续）（二）https://developer.aliyun.com/article/1474068
