Message Passing

> 原文：http://gnuradio.org/doc/doxygen/page_msg_passing.html

# **消息处理机制**

## 介绍

GNU Radio原本只是一个在block与block之间传递的数据流系统。这样的系统对于采样序列、比特流来说还行，但是如果有控制数据（control data）、元数据（metedata）、分组结构（packet structures），那么这样的结构就显得并不是很友好。

我们通过引入tag stream机制（参见[Stream Tags](http://gnuradio.org/doc/doxygen/page_stream_tags.html)）部分解决了这个问题。tag stream是相对正常的data stream并行的一个流。不同的是，tag stream用来存储元数据（metadata）和控制信息（control infomation）。数据流中的特定样本会被打上tags，然后流向下游。这种机制允许其他block辨别具有某些特定行为的粒度（item）。

tag stream的优点：tag stream相对于data来说是并行的。
tag stream的缺点：tag stream仅仅只能在work函数中获取，并且只能流向一个方向。

我们有许多理由去设计一个更通用的、更广泛的消息传递体系。首先，需要允许下游的blocks和上游的blocks通信。其次，需要允许GNU Radio和应用之间更简单的通信。新的消息传递接口（message passing interface）解决了这些问题，其是基于异步的实现方式。

## 消息传递API（Message Passing API）

[gr::basic_block](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html)实现了消息传递接口，gr::basic_block也是GNU Radio中所有blocks的父类。每个block有消息队列（messages queues）来存储输入的messages，也可以将messages发送到其他blocks的消息队列中。blocks也会区分输入和输出端口。

block会在构造函数（constructor）中声明输入和输出消息端口（message port）。消息端口以PMT对象表示，注册新的端口的API如下所示：
```
void message_port_register_in(pmt::pmt_t port_id)
void message_port_register_out(pmt::pmt_t port_id)
```
其他模块如果想要在某个端口接受messages，那么可以订阅（subscribe）这个port_id。当block有messages去发送时，可以指定port_id进行发送。订阅（subscribe）和发布（publish）messages的API如下：
```
void message_port_pub(pmt::pmt_t port_id,
 pmt::pmt_t msg);
void message_port_sub(pmt::pmt_t port_id,
 pmt::pmt_t target);
void message_port_unsub(pmt::pmt_t port_id,
 pmt::pmt_t target);
```
订阅某个block输出消息端口的任何blocks，都会在该端口发布消息时接受到该消息。具体地说，当block发布message时，该block会迭代所有订阅的block，然后用[gr::basic_block::_post](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html#ab82376fe13f515cb904c0d2e9fec87b6)方法发送message到订阅的blocks的消息队列中去。

## 消息处理函数（Message Handler Functions）

订阅的block必须声明消息处理函数（message handler function）去处理接受到的messages。在用[gr::basic_block::message_port_register_in](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html#a362b6de38292cca9c1d56439a6efad04)声明输入端口之后，我们需要将该端口绑定到消息处理函数，这里用的是Boost的bind方法：
```
set_msg_handler(pmt::pmt_t port_id,
 boost::bind(&block_class::message_handler_function, this, _1));
```

port_id是注册端口时的PMT对象。block_class::message_handler_function是用来处理消息的函数。this和_1是钦定的，不用管太多。所有消息处理函数的原型是：
```
void block_class::message_handler_function(pmt::pmt_t msg)
```

下面，我们介绍一个例子。

## 流图中连接消息（Connecting Messages through the Flowgraph）

[gr::hier_block2::msg_connect](http://gnuradio.org/doc/doxygen/classgr_1_1hier__block2.html#a8abff96c559930893d3ffdb22a3c43fd)方法让我们更容易订阅其他blocks的消息。这个消息连接函数如下使用，其中，src是一个拥有输出消息端口（命名为pdus）的block，dbg是一个拥有输入消息端口（命名为print）的block。
```
self.tb.msg_connect(src, "pdus", dbg, "print")
```

src在端口pdus发布的所有messages都会被dbg的端口print所接收。注意我们仅仅用了字符串来定义端口，而非PMT对象。这让开发者更容易使用（python中可以用pmt::intern函数来创建PMT对象，c++中为pmt::intern函数）。

查询block输入和输出端口的API：
```
pmt::pmt_t message_ports_in();
pmt::pmt_t message_prots_out();
```
返回值为元素为PMT对象的PMT向量，需要PMT的操作符才能解析。

block有一些内置的方法来处理messages的post和receive(这里的post指接受消息，放入队列，唤醒线程)。blocks的[gr::basic_block::_post](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html#ab82376fe13f515cb904c0d2e9fec87b6)方法 接受message并将其放入队列。在发射模型中，我们可以用gr::basic_block::_post方法访问消息队列。如果block处于闲置状态，那么一旦message被post到队列中，然后线程就会被唤醒，调用message handler。

我们用Boost bind方法将消息端口和消息处理函数绑定，当新消息被压入端口的消息队列中去，就会调用消息处理函数去处理消息。

## Python blocks中的消息处理（Message Passing in Python Blocks）

文档这部分没有东西。。囧。。

## 代码示例（Code Examples）

接下来是GNU Radio中利用消息处理的一些代码片段。我们会利用[gr::blocks::message_debug](http://gnuradio.org/doc/doxygen/classgr_1_1blocks_1_1message__debug.html)和[gr::blocks::tagged_stream_to_pdu](http://gnuradio.org/doc/doxygen/classgr_1_1blocks_1_1tagged__stream__to__pdu.html)来展示输入和输出消息传递。

[gr::blocks::message_debug](http://gnuradio.org/doc/doxygen/classgr_1_1blocks_1_1message__debug.html)用于调试消息传递系统。其有三个消息端口：print, store和pdu_print。print端口将所有消息标准输出。store端口会存储所有post到它的消息。pdu_print将PDU消息格式化打印。store端口可以和[gr::block::message_debug::get_message(int i)](http://gnuradio.org/doc/doxygen/classgr_1_1blocks_1_1message__debug.html#ad3cc1e2fa2412307b3cb5b33c14f0c8b)一起工作，这个函数是用来向后获取第i个message的。(`这个函数的解释是照原文翻译的，但是感觉怪怪的`)

该block的构造函数如下：
```
{
 message_port_register_in(pmt::mp("print"));
 set_msg_handler(pmt::mp("print"),
 boost::bind(&message_debug_impl::print, this, _1));
 message_port_register_in(pmt::mp("store"));
 set_msg_handler(pmt::mp("store"),
 boost::bind(&message_debug_impl::store, this, _1));
 message_port_register_in(pmt::mp("print_pdu"));
 set_msg_handler(pmt::mp("print_pdu"),
 boost::bind(&message_debug_impl::print_pdu, this, _1));
}
```

三个端口分别按照对应的名称注册。我们可以用[gr::basic_block::set_msg_handler](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html#a7fed11ec01538bfea999fb37687b43b4)函数为端口绑定回调函数。gr::blocks::message_debug_impl::print, gr::blocks::message_debug_impl::store, and gr::blocks::message_debug_impl::print_pdu都是回调函数，，用于处理对应的messages的。下面是print的函数:
```
void
message_debug_impl::print(pmt::pmt_t msg)
{
 std::cout << "***** MESSAGE DEBUG PRINT ********\n";
 pmt::print(msg);
 std::cout << "**********************************\n";
}
```
这个函数接受PMT类型的message作为参数，并打印出。该方法[pmt::print](http://gnuradio.org/doc/doxygen/namespacepmt.html#acf4beefce8c6c5c70bdce4e246a54886)是PMT库中的函数，用于友好的打印PMT。

[gr::blocks::tagged_stream_to_pdu](http://gnuradio.org/doc/doxygen/classgr_1_1blocks_1_1tagged__stream__to__pdu.html)block仅定义了一个输出消息端口。此时，构造函数如下：
```
{
 message_port_register_out(pdu_port_id);
}
```
pdu_port_id被定义在pdu.h中，其值为pdus，这样就仅仅创建了一个输出消息端口，而没有创建输入消息端口。

此block是用于接收带有stream tags的样本流，然后从中构建PDU消息。在GNU Radio中，我们把一个组成为(metadata, data)的PMT pair定义为一个PDU。其中，metadata描述pair中data部分，比如，data段的长度和其他信息（样本速率等等）。PMT vectors知道自己的长度，所以实际上长度值其实也不那么必要（废话好多，神烦）。。。metadata是一个PMT字典，而data是PMT类型的向量，向量元素的类型均是一致的，比如bytes、floats或者complex等。

最后，当PDU Message准备好之后，该block调用gr::block::tagged_stream_to_pdu_impl::send_message方法：
```
void
tagged_stream_to_pdu_impl::send_message()
{
 if(pmt::length(d_pdu_vector) != d_pdu_length) {
 throw std::runtime_error("msg length not correct");
 }
 pmt::pmt_t msg = pmt::cons(d_pdu_meta,
 d_pdu_vector);
 message_port_pub(pdu_port_id, msg);
 d_pdu_meta = pmt::PMT_NIL;
 d_pdu_vector = pmt::PMT_NIL;
 d_pdu_length = 0;
 d_pdu_remain = 0;
 d_inpdu = false;
}
```
该方法对PDU做了一些错误检查，并且在最后清除变量的值。注意其中发布message的哪一行，该block通过gr::basic_block::message_port_pub将message传播到所有的订阅block。

这里有一个类似的block[gr::blocks::pdu_to_tagged_stream](http://gnuradio.org/doc/doxygen/classgr_1_1blocks_1_1pdu__to__tagged__stream.html)做的是相反的工作。它等待PDU messages，然后从中解析出metadata和data，并处理之。metadata字典被分离成键值对，然后从中创建stream tags，data被转化成输出items。接下来描述如何用gr::blocks::pdu_to_tagged_stream将PDUs传入流图中。

# Posting from External Sources

这里要讨论的最后一部分是，从external source（注意哦，和前面的msg->msg的区别是，这里的pdu不是来自一个输出消息端口，所以无法进行订阅和绑定）中获取message。我们可以用[gr::basic_block::_post](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html#ab82376fe13f515cb904c0d2e9fec87b6)方法直接将其传递到指定端口。任何具有输入消息端口的block都可以用这个方法来获取外界的message。

接下来用[gr::blocks::pdu_to_tagged_stream](http://gnuradio.org/doc/doxygen/classgr_1_1blocks_1_1pdu__to__tagged__stream.html)block作为某个流图的信源。它等待PDUs消息，然后将其转化为正常的数据流。payload被发送作为正常的数据流，而metadata会被解析成tags，最后发送带有标签的数据流。

假设我们创建了一个src的block，来讲PDU转化为stream，其有一个pdus输入消息端口。PDUs来自另外一个block或者流图，但是这里我们以手动创建的方式来模拟代替。
```
port = pmt.intern("pdus")
msg = pmt.cons(pmt.PMT_NIL,
 pmt.make_u8vector(16, 0xFF))
src.to_basic_block()._post(port, msg)
```
这里，用pmt.PMT_NIT或者pmt::PMT_NIL对象来表示空的metadata，payload现在是16个元素、每个元素全1的字节。我们调用[gr::basic_block::to_basic_block](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html#a84bf6924dfa0bb724caca46575669557)方法，然后调用了[gr::basic_block::_post](http://gnuradio.org/doc/doxygen/classgr_1_1basic__block.html#ab82376fe13f515cb904c0d2e9fec87b6)方法来讲PDU数据传到src的正确端口里。注意，这部分代码是在src的上游block里进行的。

以上，所有的机制已经被讨论，并且已经用qa_pdu.py里的QA代码验证过。

# 将messages当做命令使用

Messages可以用来作为对blocks的命令。比如：

* [gr::qtgui::freq_sink_c](http://gnuradio.org/doc/doxygen/classgr_1_1qtgui_1_1freq__sink__c.html):频率轴的比例可以被messages调制
* [gr::uhd::usrp_source](http://gnuradio.org/doc/doxygen/classgr_1_1uhd_1_1usrp__source.html) and [gr::uhd::usrp_sink](http://gnuradio.org/doc/doxygen/classgr_1_1uhd_1_1usrp__sink.html):许多发射接收机设置可以用命令形式的messages来操作，例如频率、增益、LO偏置
* [gr::digital::header_payload_demux](http://gnuradio.org/doc/doxygen/classgr_1_1digital_1_1header__payload__demux.html),这接受header parser block的ack来知晓有多少items需要处理

这种情况下，虽然没有什么特殊要求，但是一般被建议用以下的形式：

* pmt::cons(KEY, VALUE):这种形式用于单独的值。KEY可以对应参数名，VALUE可以对应值。在QT GUI Frequency Sink中，KEY对"freq"，VALUE即单位HZ的中心频率。
* pmt::dict((KEY1: VALUE1), (KEY2: VALUE2), ...):这其实本质上和上一条一样，但是可以用于多个键值对。当一个单独的命令有多个参数，这样就不用拆成多个命令。（例如， USRP block的命令里也许有一个timestamp和一个center frequency的参数）

两种情况下，所有的KEYs都是PMT对象。值则视情况而定。

也许不用这种格式，也是可以的，，比如QT Frequency sink简单地将float值作为一个command message，但是仍然可以工作很好。可以，我们有些理由需要遵守这种规范：

* 互操作性：越多的人用规范的格式， 那么来自不同开发者的模块可以更好的一起工作
* 可检测性：如果message包含键值对，那么message debug block将会显示更多有用的信息。
* 直观：这种形式非常直观，很容易阅读以及理解。

逆风起笔