【Apollo自动驾驶-从理论到代码】cyber/node模块

/* 作者水平有限，欢迎批评指正，内容持续完善中！！*/

主要文件

类图

处理流程

Node特点及须知

1. Node中可以创建Reader、Writer、Service、Client对象。
2. 其中上面提到的Reader、Writer、Service、Client对象由对应的类创建。
3. 根据要创建的类型，选择NodeChannelImpl或者NodeServiceImpl类进行Reader、Writer、Service、Client对象的创建。

代码详解

由于node目录和service目录的关联性较大，下面将会一起介绍两处的代码。

CreateNode函数的调用点

在Node类中，使用了友元函数CreateNode()。该函数位于命名空间apollo::cyber中，在模块代码中，会通过调用该函数创建节点Node，如：

./modules/canbus/tools/teleop.cc:97:    node_ = CreateNode("teleop");
./modules/data/tools/smart_recorder/realtime_record_processor.cc:120:  smart_recorder_node_ = CreateNode(absl::StrCat("smart_recorder_", getpid()));
./modules/routing/tools/routing_cast.cc:33:      apollo::cyber::CreateNode("routing_cast"));

那现在深入到CreateNode函数中，该函数位于cyber.cc中，cyber.cc文件中只定义了一个CreateNode函数，由此可见，Node和Cyber的密切关系：

namespace apollo {
namespace cyber {using apollo::cyber::common::GlobalData;
using apollo::cyber::proto::RunMode;std::unique_ptr<Node> CreateNode(const std::string& node_name,const std::string& name_space) {bool is_reality_mode = GlobalData::Instance()->IsRealityMode();if (is_reality_mode && !OK()) {// add some hint logAERROR << "please initialize cyber firstly.";return nullptr;}return std::unique_ptr<Node>(new Node(node_name, name_space));
}

如上，在CreateNode函数中，直接返回了Node类型的对象，这里真正创建了Node对象。并且传入了node_name、name_space作为参数。下面进入到node.h文件中，关注节点的实际创建过程。

Node的构造函数

首先在Node的构造函数中，使用node_name创建了NodeChannelImpl和NodeServiceImpl对象。后面会分析二者的创建与执行过程。

Node::Node(const std::string& node_name, const std::string& name_space): node_name_(node_name), name_space_(name_space) {node_channel_impl_.reset(new NodeChannelImpl(node_name));node_service_impl_.reset(new NodeServiceImpl(node_name));
}

NodeChannelImpl的构造函数

此处只是做了简单的初始化工作。如果是真实模式的话，会启动node_manager。

  /*** @brief Construct a new Node Channel Impl object** @param node_name node name*/explicit NodeChannelImpl(const std::string& node_name): is_reality_mode_(true), node_name_(node_name) {node_attr_.set_host_name(common::GlobalData::Instance()->HostName());node_attr_.set_host_ip(common::GlobalData::Instance()->HostIp());node_attr_.set_process_id(common::GlobalData::Instance()->ProcessId());node_attr_.set_node_name(node_name);uint64_t node_id = common::GlobalData::RegisterNode(node_name);node_attr_.set_node_id(node_id);is_reality_mode_ = common::GlobalData::Instance()->IsRealityMode();if (is_reality_mode_) {node_manager_ =service_discovery::TopologyManager::Instance()->node_manager();node_manager_->Join(node_attr_, RoleType::ROLE_NODE);}}

NodeServiceImpl的构造函数

创建Reader

下面以teleop.cc文件为例，介绍如何创建reader及writer。同时带大家进入相关底层代码一探究竟。如下代码，分别调用了node_对象的两个接口CreateReader和CreateWriter创建reader和writer。

./modules/canbus/tools/teleop.cc:97:    node_ = CreateNode("teleop");chassis_reader_ = node_->CreateReader<Chassis>(FLAGS_chassis_topic, [this](const std::shared_ptr<Chassis> &chassis) {OnChassis(*chassis);});control_command_writer_ =node_->CreateWriter<ControlCommand>(FLAGS_control_command_topic);

在node.h文件中，CreateReader函数共有三个重载函数，意味着会根据创建时刻的传递类型，进行重载。如下为三个函数的实现，三者的区别为第一个参数分别为RoleAttributes、ReaderConfig、string。具体差别见代码中的注释。由此可知，这里主要做了配置参数的初始化。最终使用模板函数node_channel_impl_->template CreateReader(）创建reader。

//使用RoleAttributes创建Reader。
//config: 包括channel name, qos
//reader_func: 当消息到达时的回调函数
template <typename MessageT>
auto Node::CreateReader(const proto::RoleAttributes& role_attr,const CallbackFunc<MessageT>& reader_func)-> std::shared_ptr<Reader<MessageT>> {std::lock_guard<std::mutex> lg(readers_mutex_);if (readers_.find(role_attr.channel_name()) != readers_.end()) {AWARN << "Failed to create reader: reader with the same channel already ""exists.";return nullptr;}auto reader = node_channel_impl_->template CreateReader<MessageT>(role_attr, reader_func);if (reader != nullptr) {readers_.emplace(std::make_pair(role_attr.channel_name(), reader));}return reader;
}//使用reader config创建Reader。
//config: 包括channel name, qos ,pending queue size
//reader_func: 当消息到达时的回调函数
template <typename MessageT>
auto Node::CreateReader(const ReaderConfig& config,const CallbackFunc<MessageT>& reader_func)-> std::shared_ptr<cyber::Reader<MessageT>> {std::lock_guard<std::mutex> lg(readers_mutex_);if (readers_.find(config.channel_name) != readers_.end()) {AWARN << "Failed to create reader: reader with the same channel already ""exists.";return nullptr;}auto reader =node_channel_impl_->template CreateReader<MessageT>(config, reader_func);if (reader != nullptr) {readers_.emplace(std::make_pair(config.channel_name, reader));}return reader;
}//使用channel name创建Reader。qos和其他配置使用默认。
//channel_name: reader订阅的channel
//reader_func: 当消息到达时的回调函数
template <typename MessageT>
auto Node::CreateReader(const std::string& channel_name,const CallbackFunc<MessageT>& reader_func)-> std::shared_ptr<Reader<MessageT>> {std::lock_guard<std::mutex> lg(readers_mutex_);if (readers_.find(channel_name) != readers_.end()) {AWARN << "Failed to create reader: reader with the same channel already ""exists.";return nullptr;}auto reader = node_channel_impl_->template CreateReader<MessageT>(channel_name, reader_func);if (reader != nullptr) {readers_.emplace(std::make_pair(channel_name, reader));}return reader;
}

对于上面三个接口，可用表格总结如下：

创建Writer、Service、Client

结合上面Reader的创建过程，Writer、Service、Client的创建类似。总的来说，会根据不同的参数类型，选择不同的重载函数，从而创建不同的对象。下面用表格方式进行全局展示，前文中各概念的关系。

Reader的实质创建

Reader类型的选择

在node_channel_impl.h中经过函数重载，最终的Reader的创建使用下面函数，参数有三个：

根据是否为实时模式，使用不同的Reader类创建对象，然后执行reader的Init函数。

template <typename MessageT>
auto NodeChannelImpl::CreateReader(const proto::RoleAttributes& role_attr,const CallbackFunc<MessageT>& reader_func,uint32_t pending_queue_size)-> std::shared_ptr<Reader<MessageT>> {if (!role_attr.has_channel_name() || role_attr.channel_name().empty()) {AERROR << "Can't create a reader with empty channel name!";return nullptr;}proto::RoleAttributes new_attr(role_attr);FillInAttr<MessageT>(&new_attr);std::shared_ptr<Reader<MessageT>> reader_ptr = nullptr;if (!is_reality_mode_) {reader_ptr =std::make_shared<blocker::IntraReader<MessageT>>(new_attr, reader_func);} else {reader_ptr = std::make_shared<Reader<MessageT>>(new_attr, reader_func,pending_queue_size);}RETURN_VAL_IF_NULL(reader_ptr, nullptr);RETURN_VAL_IF(!reader_ptr->Init(), nullptr);return reader_ptr;
}

Reader的Init()过程

在Reader的构造函数中，除了处理入参外，还对blocker_进行了初始化。对象构造完成后，会调用Init()函数进行初始化。下面在代码中进行注释说明初始化的过程。

template <typename MessageT>
bool Reader<MessageT>::Init() {if (init_.exchange(true)) {return true;}//此处根据reader_func创建lamba表达式（未命名的内联函数），其实就是消息真正执行者。std::function<void(const std::shared_ptr<MessageT>&)> func;if (reader_func_ != nullptr) {func = [this](const std::shared_ptr<MessageT>& msg) {this->Enqueue(msg);this->reader_func_(msg);};} else {func = [this](const std::shared_ptr<MessageT>& msg) { this->Enqueue(msg); };}//重点介绍：
//这里调用了单实例的sched，全局调度模块。使用节点名称和通道名称作为协程的ID。
//通过协程工厂，将前面的func函数和dv进行关联，并放入sched创建Task。auto sched = scheduler::Instance();croutine_name_ = role_attr_.node_name() + "_" + role_attr_.channel_name();auto dv = std::make_shared<data::DataVisitor<MessageT>>(role_attr_.channel_id(), pending_queue_size_);// Using factory to wrap templates.croutine::RoutineFactory factory =croutine::CreateRoutineFactory<MessageT>(std::move(func), dv);if (!sched->CreateTask(factory, )) {AERROR << "Create Task Failed!";init_.store(false);return false;}//创建当前Node的receiver,并且加入拓扑网络中，后面会介绍TopologyManager的具体作用，在Topo中如何进行发现与通信。receiver_ = ReceiverManager<MessageT>::Instance()->GetReceiver(role_attr_);this->role_attr_.set_id(receiver_->id().HashValue());channel_manager_ =service_discovery::TopologyManager::Instance()->channel_manager();JoinTheTopology();return true;
}

/未完待续/