apollo::cyber::transport::ShmDispatcher类 参考

#include <shm_dispatcher.h>

类 apollo::cyber::transport::ShmDispatcher 继承关系图:

apollo::cyber::transport::ShmDispatcher 的协作图:

Public 类型
using	SegmentContainer = std::unordered_map< uint64_t, SegmentPtr >

Public 成员函数
virtual	~ShmDispatcher ()
void	Shutdown () override
template<typename MessageT >
void	AddListener (const RoleAttributes &self_attr, const MessageListener< MessageT > &listener)
template<typename MessageT >
void	AddListener (const RoleAttributes &self_attr, const RoleAttributes &opposite_attr, const MessageListener< MessageT > &listener)
template<typename MessageT >
void	AddArenaListener (const RoleAttributes &self_attr, const MessageListener< MessageT > &listener)
template<typename MessageT >
void	AddArenaListener (const RoleAttributes &self_attr, const RoleAttributes &opposite_attr, const MessageListener< MessageT > &listener)
Public 成员函数 继承自 apollo::cyber::transport::Dispatcher
	Dispatcher ()
virtual	~Dispatcher ()
template<typename MessageT >
void	AddListener (const RoleAttributes &self_attr, const MessageListener< MessageT > &listener)
template<typename MessageT >
void	AddListener (const RoleAttributes &self_attr, const RoleAttributes &opposite_attr, const MessageListener< MessageT > &listener)
template<typename MessageT >
void	RemoveListener (const RoleAttributes &self_attr)
template<typename MessageT >
void	RemoveListener (const RoleAttributes &self_attr, const RoleAttributes &opposite_attr)
bool	HasChannel (uint64_t channel_id)

额外继承的成员函数
Protected 属性 继承自 apollo::cyber::transport::Dispatcher
std::atomic< bool >	is_shutdown_
AtomicHashMap< uint64_t, ListenerHandlerBasePtr >	msg_listeners_
base::AtomicRWLock	rw_lock_

详细描述

在文件 shm_dispatcher.h 第 50 行定义.

成员类型定义说明

SegmentContainer

using apollo::cyber::transport::ShmDispatcher::SegmentContainer = std::unordered_map<uint64_t, SegmentPtr>

在文件 shm_dispatcher.h 第 53 行定义.

构造及析构函数说明

~ShmDispatcher()

apollo::cyber::transport::ShmDispatcher::~ShmDispatcher ( )

virtual

在文件 shm_dispatcher.cc 第 32 行定义.

{ Shutdown(); }

成员函数说明

AddArenaListener() [1/2]

template<typename MessageT >

void apollo::cyber::transport::ShmDispatcher::AddArenaListener	(	const RoleAttributes &	self_attr,
		const MessageListener< MessageT > &	listener
	)

在文件 shm_dispatcher.h 第 102 行定义.

                                               {
  if (is_shutdown_.load()) {
    return;
  }
  uint64_t channel_id = self_attr.channel_id();
 
  std::shared_ptr<ListenerHandler<MessageT>> handler;
  ListenerHandlerBasePtr* handler_base = nullptr;
  if (arena_msg_listeners_.Get(channel_id, &handler_base)) {
    handler =
        std::dynamic_pointer_cast<ListenerHandler<MessageT>>(*handler_base);
    if (handler == nullptr) {
      AERROR << "please ensure that readers with the same channel["
             << self_attr.channel_name()
             << "] in the same process have the same message type";
      return;
    }
  } else {
    ADEBUG << "new reader for channel:"
           << GlobalData::GetChannelById(channel_id);
    handler.reset(new ListenerHandler<MessageT>());
    arena_msg_listeners_.Set(channel_id, handler);
  }
  handler->Connect(self_attr.id(), listener);
}

AddArenaListener() [2/2]

template<typename MessageT >

void apollo::cyber::transport::ShmDispatcher::AddArenaListener	(	const RoleAttributes &	self_attr,
		const RoleAttributes &	opposite_attr,
		const MessageListener< MessageT > &	listener
	)

在文件 shm_dispatcher.h 第 131 行定义.

                                               {
  if (is_shutdown_.load()) {
    return;
  }
  uint64_t channel_id = self_attr.channel_id();
  std::shared_ptr<ListenerHandler<MessageT>> handler;
  ListenerHandlerBasePtr* handler_base = nullptr;
  if (arena_msg_listeners_.Get(channel_id, &handler_base)) {
    handler =
        std::dynamic_pointer_cast<ListenerHandler<MessageT>>(*handler_base);
    if (handler == nullptr) {
      AERROR << "please ensuore that readers with the same channel["
             << self_attr.channel_name()
             << "] in the same process have the same message type";
      return;
    }
  } else {
    ADEBUG << "new reader for channel:"
           << GlobalData::GetChannelById(channel_id);
    handler.reset(new ListenerHandler<MessageT>());
    arena_msg_listeners_.Set(channel_id, handler);
  }
  handler->Connect(self_attr.id(), listener);
}

AddListener() [1/2]

template<typename MessageT >

void apollo::cyber::transport::ShmDispatcher::AddListener	(	const RoleAttributes &	self_attr,
		const MessageListener< MessageT > &	listener
	)

在文件 shm_dispatcher.h 第 159 行定义.

                                                                           {
  // FIXME: make it more clean
  if (cyber::common::GlobalData::Instance()->IsChannelEnableArenaShm(
          self_attr.channel_id()) &&
      self_attr.message_type() != message::MessageType<message::RawMessage>() &&
      self_attr.message_type() !=
          message::MessageType<message::PyMessageWrap>()) {
    auto listener_adapter = [listener, self_attr](
                                const std::shared_ptr<ReadableBlock>& rb,
                                const MessageInfo& msg_info) {
      auto msg = std::make_shared<MessageT>();
      // TODO(ALL): read config from msg_info
      auto arena_manager = ProtobufArenaManager::Instance();
      auto msg_wrapper = arena_manager->CreateMessageWrapper();
      memcpy(msg_wrapper->GetData(), rb->buf, 1024);
      MessageT* msg_p;
      if (!message::ParseFromArenaMessageWrapper(msg_wrapper.get(), msg.get(),
                                                 &msg_p)) {
        AERROR << "ParseFromArenaMessageWrapper failed";
      }
      // msg->CopyFrom(*msg_p);
      // msg = arena_manager->LoadMessage<MessageT>(msg_wrapper.get())
      auto segment = arena_manager->GetSegment(self_attr.channel_id());
      auto msg_addr = reinterpret_cast<uint64_t>(msg_p);
      msg.reset(reinterpret_cast<MessageT*>(msg_addr),
                [arena_manager, segment, msg_wrapper](MessageT* p) {
                  // fprintf(stderr, "msg deleter invoked\n");
                  // auto related_blocks =
                  //     arena_manager->GetMessageRelatedBlocks(msg_wrapper.get());
                  // for (auto block_index : related_blocks) {
                  //   // segment->ReleaseBlockForReadByIndex(block_index);
                  //   segment->RemoveBlockReadLock(block_index);
                  // }
                });
      auto related_blocks_for_lock =
          arena_manager->GetMessageRelatedBlocks(msg_wrapper.get());
      for (int i = 0; i < related_blocks_for_lock.size(); ++i) {
        auto block_index = related_blocks_for_lock[i];
        if (!segment->AddBlockReadLock(block_index)) {
          AWARN << "failed to acquire block for read, channel: "
                << self_attr.channel_id() << " index: " << block_index;
          for (int j = 0; j < i; ++j) {
            // restore the lock
            segment->RemoveBlockReadLock(related_blocks_for_lock[j]);
          }
          return;
        }
      }
 
      auto send_time = msg_info.send_time();
 
      statistics::Statistics::Instance()->AddRecvCount(self_attr,
                                                       msg_info.seq_num());
      statistics::Statistics::Instance()->SetTotalMsgsStatus(
          self_attr, msg_info.seq_num());
 
      auto recv_time = Time::Now().ToNanosecond();
 
      // sampling in microsecond
      auto tran_diff = (recv_time - send_time) / 1000;
      if (tran_diff > 0) {
        // sample transport latency in microsecond
        statistics::Statistics::Instance()->SamplingTranLatency<uint64_t>(
            self_attr, tran_diff);
      }
      statistics::Statistics::Instance()->SetProcStatus(self_attr,
                                                        recv_time / 1000);
      listener(msg, msg_info);
      auto related_blocks =
          arena_manager->GetMessageRelatedBlocks(msg_wrapper.get());
      for (auto block_index : related_blocks) {
        // segment->ReleaseBlockForReadByIndex(block_index);
        segment->RemoveBlockReadLock(block_index);
      }
    };
 
    AddArenaListener<ReadableBlock>(self_attr, listener_adapter);
  } else {
    auto listener_adapter = [listener, self_attr](
                                const std::shared_ptr<ReadableBlock>& rb,
                                const MessageInfo& msg_info) {
      auto msg = std::make_shared<MessageT>();
      // TODO(ALL): read config from msg_info
      RETURN_IF(!message::ParseFromArray(
          rb->buf, static_cast<int>(rb->block->msg_size()), msg.get()));
 
      auto send_time = msg_info.send_time();
 
      statistics::Statistics::Instance()->AddRecvCount(self_attr,
                                                       msg_info.seq_num());
      statistics::Statistics::Instance()->SetTotalMsgsStatus(
          self_attr, msg_info.seq_num());
 
      auto recv_time = Time::Now().ToNanosecond();
 
      // sampling in microsecond
      auto tran_diff = (recv_time - send_time) / 1000;
      if (tran_diff > 0) {
        // sample transport latency in microsecond
        statistics::Statistics::Instance()->SamplingTranLatency<uint64_t>(
            self_attr, tran_diff);
      }
      statistics::Statistics::Instance()->SetProcStatus(self_attr,
                                                        recv_time / 1000);
      listener(msg, msg_info);
    };
 
    Dispatcher::AddListener<ReadableBlock>(self_attr, listener_adapter);
  }
  AddSegment(self_attr);
}

AddListener() [2/2]

template<typename MessageT >

void apollo::cyber::transport::ShmDispatcher::AddListener	(	const RoleAttributes &	self_attr,
		const RoleAttributes &	opposite_attr,
		const MessageListener< MessageT > &	listener
	)

在文件 shm_dispatcher.h 第 273 行定义.

                                                                           {
  // FIXME: make it more clean
  if (cyber::common::GlobalData::Instance()->IsChannelEnableArenaShm(
          self_attr.channel_id()) &&
      self_attr.message_type() != message::MessageType<message::RawMessage>() &&
      self_attr.message_type() !=
          message::MessageType<message::PyMessageWrap>()) {
    auto listener_adapter = [listener, self_attr](
                                const std::shared_ptr<ReadableBlock>& rb,
                                const MessageInfo& msg_info) {
      auto msg = std::make_shared<MessageT>();
      auto arena_manager = ProtobufArenaManager::Instance();
      auto msg_wrapper = arena_manager->CreateMessageWrapper();
      memcpy(msg_wrapper->GetData(), rb->buf, 1024);
      MessageT* msg_p;
      if (!message::ParseFromArenaMessageWrapper(msg_wrapper.get(), msg.get(),
                                                 &msg_p)) {
        AERROR << "ParseFromArenaMessageWrapper failed";
      }
      // msg->CopyFrom(*msg_p);
      // msg = arena_manager->LoadMessage<MessageT>(msg_wrapper.get())
      auto segment = arena_manager->GetSegment(self_attr.channel_id());
      auto msg_addr = reinterpret_cast<uint64_t>(msg_p);
      msg.reset(reinterpret_cast<MessageT*>(msg_addr),
                [arena_manager, segment, msg_wrapper](MessageT* p) {
                  // fprintf(stderr, "msg deleter invoked\n");
                  // auto related_blocks =
                  //     arena_manager->GetMessageRelatedBlocks(msg_wrapper.get());
                  // for (auto block_index : related_blocks) {
                  //   // segment->ReleaseBlockForReadByIndex(block_index);
                  //   segment->RemoveBlockReadLock(block_index);
                  // }
                });
      auto related_blocks_for_lock =
          arena_manager->GetMessageRelatedBlocks(msg_wrapper.get());
      for (int i = 0; i < related_blocks_for_lock.size(); ++i) {
        auto block_index = related_blocks_for_lock[i];
        if (!segment->AddBlockReadLock(block_index)) {
          AWARN << "failed to acquire block for read, channel: "
                << self_attr.channel_id() << " index: " << block_index;
          for (int j = 0; j < i; ++j) {
            // restore the lock
            segment->RemoveBlockReadLock(related_blocks_for_lock[j]);
          }
          return;
        }
      }
 
      auto send_time = msg_info.send_time();
 
      statistics::Statistics::Instance()->AddRecvCount(self_attr,
                                                       msg_info.seq_num());
      statistics::Statistics::Instance()->SetTotalMsgsStatus(
          self_attr, msg_info.seq_num());
 
      auto recv_time = Time::Now().ToNanosecond();
 
      // sampling in microsecond
      auto tran_diff = (recv_time - send_time) / 1000;
      if (tran_diff > 0) {
        statistics::Statistics::Instance()->SamplingTranLatency<uint64_t>(
            self_attr, tran_diff);
      }
      statistics::Statistics::Instance()->SetProcStatus(self_attr,
                                                        recv_time / 1000);
 
      listener(msg, msg_info);
      auto related_blocks =
          arena_manager->GetMessageRelatedBlocks(msg_wrapper.get());
      for (auto block_index : related_blocks) {
        // segment->ReleaseBlockForReadByIndex(block_index);
        segment->RemoveBlockReadLock(block_index);
      }
    };
 
    AddArenaListener<ReadableBlock>(self_attr, opposite_attr, listener_adapter);
  } else {
    auto listener_adapter = [listener, self_attr](
                                const std::shared_ptr<ReadableBlock>& rb,
                                const MessageInfo& msg_info) {
      auto msg = std::make_shared<MessageT>();
      RETURN_IF(!message::ParseFromArray(
          rb->buf, static_cast<int>(rb->block->msg_size()), msg.get()));
 
      auto send_time = msg_info.send_time();
      auto msg_seq_num = msg_info.seq_num();
 
      statistics::Statistics::Instance()->AddRecvCount(self_attr, msg_seq_num);
      statistics::Statistics::Instance()->SetTotalMsgsStatus(self_attr,
                                                             msg_seq_num);
 
      auto recv_time = Time::Now().ToNanosecond();
 
      // sampling in microsecond
      auto tran_diff = (recv_time - send_time) / 1000;
      if (tran_diff > 0) {
        statistics::Statistics::Instance()->SamplingTranLatency<uint64_t>(
            self_attr, tran_diff);
      }
      statistics::Statistics::Instance()->SetProcStatus(self_attr,
                                                        recv_time / 1000);
 
      listener(msg, msg_info);
    };
 
    Dispatcher::AddListener<ReadableBlock>(self_attr, opposite_attr,
                                           listener_adapter);
  }
  AddSegment(self_attr);
}

Shutdown()

void apollo::cyber::transport::ShmDispatcher::Shutdown ( )

overridevirtual

重载 apollo::cyber::transport::Dispatcher .

在文件 shm_dispatcher.cc 第 34 行定义.

                             {
  if (is_shutdown_.exchange(true)) {
    return;
  }
 
  if (thread_.joinable()) {
    thread_.join();
  }
 
  {
    ReadLockGuard<AtomicRWLock> lock(segments_lock_);
    segments_.clear();
  }
}

---

该类的文档由以下文件生成:

• cyber/transport/dispatcher/shm_dispatcher.h
• cyber/transport/dispatcher/shm_dispatcher.cc