apollo事件通信机制

        我目前用的是基于ros版本的apollo,在apollo内部有两种消息传输机制,一种是基于ros的回调函数的订阅<详细介绍>来实现各个模块之间的数据通信,还有一种就是事件的通信机制,主要用于感知模块内部的通信,这种通信是基于进程间的内存共享来实现消息的传递.下面我主要介绍第二种消息传递的方式:

• 先看一下事件的实际应用

发布事件:

void AsyncFusionSubnode::PublishDataAndEvent(
    const double timestamp, const std::string &device_id,
    const SharedDataPtr<FusionItem> &data) {
  CommonSharedDataKey key(timestamp, device_id);
  bool fusion_succ = fusion_data_->Add(key, data);
  if (!fusion_succ) {
    AERROR << "fusion shared data addkey failure";
  }

  ADEBUG << "adding key in fusion shared data " << key.ToString();

  for (size_t idx = 0; idx < pub_meta_events_.size(); ++idx) {
    const EventMeta &event_meta = pub_meta_events_[idx];
    Event event;
    event.event_id = event_meta.event_id;
    event.timestamp = timestamp;
    event.reserve = device_id;
    event_manager_->Publish(event);
  }
}

        事件发布的函数中要注意两点:一是fusion_data_->Add(key,data)就是把要传递的信息data与key绑定在一起,然后用fusion_data_添加,第二个是event_manager_->Publish(event)把事件添加进队列.先看一下Event的结构:

struct Event {
  EventID event_id = 0;
  double timestamp = 0.0;
  std::string reserve;
  // TODO(Yangguang Li):
  double local_timestamp = 0.0;  // local timestamp to compute process delay.

  Event() { local_timestamp = TimeUtil::GetCurrentTime(); }

  std::string to_string() const {
    std::ostringstream oss;
    oss << "event_id: " << event_id
        << " timestamp: " << GLOG_TIMESTAMP(timestamp)
        << " reserve: " << reserve;
    return oss.str();
  }
};

        仔细观察会发现发布的event并没有包含任何data的信息,只是实例化了一个event,包含id timespace device_id,那这个data信息是怎么传递出去的呢?再回过头来看这个与data绑定在一起的key是如何生成的,CommonSharedDataKey key(timestamp, device_id);

struct CommonSharedDataKey {
  CommonSharedDataKey() = default;
  CommonSharedDataKey(const double ts, const std::string &id)
      : timestamp(ts), device_id(id) {}
  virtual std::string ToString() const {
    return device_id +
           (boost::format("%ld") %
            static_cast<int64_t>(timestamp * FLAGS_stamp_enlarge_factor))
               .str();
  }
  double timestamp = 0.0;
  std::string device_id = "";
}

       看到这里是不是有点明白了,data本身没有传递,只是这个data根据device_id和时间戳生成了一个只属于这个data的钥匙key,然后把这个能生成key的device_id和时间戳发布到事件队列中,接收的时候先按id接受event,然后获取事件中的device_id和时间戳来生成key,然后通过key拿到data.(device_id:std::string类型的变量,某种传感器的name,timespace:时间戳,因为每一刻的时间戳都不同,所以不同时刻发出去的事件key都是独一无二的).

• 我们再继续深入,考虑一个问题

       一个节点会一直不停的发布事件,这么多事件是如何有序的被另一个节点接受然后解析,并且被解析过后的事件是如何处理的?

上文提到过一个东西"事件队列",

using EventQueue = FixedSizeConQueue<Event>;

       感知内部会每一个event安排一个queue,这个队列的最大size是5(可以考虑一下为什么这个size要设置成5),当然了这个队列是自己重新改造的符合线程安全的队列,那么我们总共有多少个事件的?接触过apollo的应该更容易理解,apollo有一个config文件,这个文件中每一个edges就是一个事件,感知这个模块其实就是一个有向无环图,串联这些节点的就是这些edges,我们通过读取config文件来初始化这个各个节点,各个边,各个sharedData,在初始化edges时,会初始化一个map,如下:

EventQueueMap event_queue_map_;

这个map的key值是时间ID,value值是一个事件队列,先根据每一个事件ID为每个事件new一个专属队列,队列最大size是5

      event_queue_map_[event_pb.id()].reset(
          new EventQueue(5));

       事件队列准备好之后,接下来就是要使用它了.你们也想到了,发布事件就是按照事件ID往这个队列push元素,订阅事件就是从这个队列pop元素,这样一来,事件通信机制的基本框架就搭好了

订阅事件:

bool AsyncFusionSubnode::SubscribeEvents(const EventMeta &event_meta,
                                         std::vector<Event> *events) const {
  Event event;
  // blocking call for each of these events
  while (event_manager_->Subscribe(event_meta.event_id, &event, true)) {
    ADEBUG << "starting subscribing event " << event_meta.event_id;
    // events->push_back(event);
  }

  // only obtain latest event from a sensor queue
  if (event.event_id != 0 && event.timestamp != 0.0) {
    events->push_back(event);
  }
  return true;
}

搞懂了事件之后,要明白就两个节点之间通过事件通信还要明白另一个东西,sharedData

       看sharedData之前,先理一下总体思路:按照我们之前的分析,A节点需要发布信息,它不会把这个数据量巨大结构体发给另一个节点,而是通过底层的事件队列来完成的.明确我们要发出的信息,一个事件ID,一个时间戳,一个Name,发出的同时用时间戳和Name生成一个唯一Key来指向这个数据量巨大的结构体.那么节点B在接受信息的时候只要能复现这个key,是不是就能拿到这个数据量巨大的结构体了.

• 明白了总体思路,就来看一下具体的实现:

这个是我们在发布信息的时候把data跟生成的key绑定在一起,

bool fusion_succ = fusion_data_->Add(key, data);

template <class M>
bool CommonSharedData<M>::Add(const std::string &key,
                              const SharedDataPtr<M> &data) {
  MutexLock lock(&mutex_);
  auto ret = data_map_.emplace(SharedDataPair(key, data));
  if (!ret.second) {
    AWARN << "Duplicate key: " << key;
    return false;
  }

  const uint64_t timestamp = ::time(NULL);
  data_added_time_map_.emplace(DataKeyTimestampPair(key, timestamp));
  ++stat_.add_cnt;
  return true;
}

template <class M>
bool CommonSharedData<M>::Add(const CommonSharedDataKey &key,
                              const SharedDataPtr<M> &data) {
  // update latest_timestamp for SharedData
  latest_timestamp_ = key.timestamp;
  return Add(key.ToString(), data);
}

      其实绑定也很简单,就是在common_shared_data.cc中维护了一个map(typedef std::unordered_map<std::string, SharedDataPtr<M>> SharedDataMap;),key值是string,value值是一个智能指针,这个指针指向的就是我们数据量巨大的结构体.我们把这个key值,跟指向data结构体的指针,尾插到这个map中.所以我们用一个sharedData类型对象,只要能得到key,是不是就能拿到这个数据量巨大的结构体了.当然这个结构体可以是任何东西,在感知内部,这个结构体往往是某种传感器一帧的障碍物信息.