Realtime processor against recorded tasks that are being recorded 更多...

#include <realtime_record_processor.h>

类 apollo::data::RealtimeRecordProcessor 继承关系图:

apollo::data::RealtimeRecordProcessor 的协作图:

Public 成员函数
	RealtimeRecordProcessor (const std::string &source_record_dir, const std::string &restored_output_dir)

bool	Init (const SmartRecordTrigger &trigger_conf) override

bool	Process () override

std::string	GetDefaultOutputFile () const override

void	MonitorStatus ()

virtual	~RealtimeRecordProcessor ()=default

Public 成员函数继承自 apollo::data::RecordProcessor
	RecordProcessor (const std::string &source_record_dir, const std::string &restored_output_dir)

virtual	~RecordProcessor ()

额外继承的成员函数
Protected 成员函数继承自 apollo::data::RecordProcessor
bool	InitTriggers (const SmartRecordTrigger &trigger_conf)

bool	ShouldRestore (const cyber::record::RecordMessage &msg) const

Protected 属性继承自 apollo::data::RecordProcessor
const std::string	source_record_dir_

const std::string	restored_output_dir_

std::vector< std::unique_ptr< TriggerBase > >	triggers_

std::unique_ptr< cyber::record::RecordWriter >	writer_ = nullptr

详细描述

Realtime processor against recorded tasks that are being recorded

在文件 realtime_record_processor.h 第 38 行定义.

构造及析构函数说明

◆ RealtimeRecordProcessor()

apollo::data::RealtimeRecordProcessor::RealtimeRecordProcessor	(	const std::string &	source_record_dir,
		const std::string &	restored_output_dir
	)

在文件 realtime_record_processor.cc 第 92 行定义.

    : RecordProcessor(source_record_dir, restored_output_dir) {
  default_output_filename_ = restored_output_dir_;
  default_output_filename_.erase(
      std::remove(default_output_filename_.begin(),
                  default_output_filename_.end(), '-'),
      default_output_filename_.end());
  default_output_filename_ =
      GetFileName(absl::StrCat(default_output_filename_, ".record"), false);
}

◆ ~RealtimeRecordProcessor()

virtual apollo::data::RealtimeRecordProcessor::~RealtimeRecordProcessor ( )

virtualdefault

成员函数说明

◆ GetDefaultOutputFile()

std::string apollo::data::RealtimeRecordProcessor::GetDefaultOutputFile ( ) const

inlineoverridevirtual

实现了 apollo::data::RecordProcessor.

在文件 realtime_record_processor.h 第 44 行定义.

                                                  {
    return absl::StrCat(restored_output_dir_, "/", default_output_filename_);
  };

◆ Init()

bool apollo::data::RealtimeRecordProcessor::Init ( const SmartRecordTrigger & trigger_conf )

overridevirtual

重载 apollo::data::RecordProcessor .

在文件 realtime_record_processor.cc 第 105 行定义.

                                                                         {
  // Init input/output, for realtime processor create both
  // input and output dir if they do not exist
  if (!EnsureDirectory(source_record_dir_) ||
      !EnsureDirectory(restored_output_dir_)) {
    AERROR << "unable to init input/output dir: " << source_record_dir_ << "/"
           << restored_output_dir_;
    return false;
  }
  if (!RemoveAllFiles(source_record_dir_)) {
    AERROR << "unable to clear input dir: " << source_record_dir_;
    return false;
  }
  // Init recorder
  cyber::Init("smart_recorder");
  smart_recorder_node_ = CreateNode(absl::StrCat("smart_recorder_", getpid()));
  if (smart_recorder_node_ == nullptr) {
    AERROR << "create smart recorder node failed: " << getpid();
    return false;
  }
  recorder_status_writer_ =
      smart_recorder_node_->CreateWriter<SmartRecorderStatus>(
          FLAGS_recorder_status_topic);
  max_backward_time_ = trigger_conf.max_backward_time();
  min_restore_chunk_ = trigger_conf.min_restore_chunk();
  std::vector<std::string> all_channels;
  std::vector<std::string> black_channels;
  const std::set<std::string>& all_channels_set =
      ChannelPool::Instance()->GetAllChannels();
  std::copy(all_channels_set.begin(), all_channels_set.end(),
            std::back_inserter(all_channels));
  recorder_ = std::make_shared<Recorder>(
      absl::StrCat(source_record_dir_, "/", default_output_filename_), false,
      all_channels, black_channels, HeaderBuilder::GetHeader());
  // Init base
  if (!RecordProcessor::Init(trigger_conf)) {
    AERROR << "base init failed";
    return false;
  }
  reused_record_num_ = trigger_conf.reused_record_num();
  record_files_.clear();
  return true;
}

◆ MonitorStatus()

void apollo::data::RealtimeRecordProcessor::MonitorStatus ( )

在文件 realtime_record_processor.cc 第 221 行定义.

                                            {
  int status_counter = 0;
  while (!cyber::IsShutdown()) {
    static constexpr int kCheckingFrequency = 100;
    static constexpr int kPublishStatusFrequency = 30;
    std::this_thread::sleep_for(std::chrono::milliseconds(kCheckingFrequency));
    if (++status_counter % kPublishStatusFrequency == 0) {
      status_counter = 0;
      PublishStatus(RecordingState::RECORDING, "smart recorder recording");
      // AINFO << "smart recorder recording status check every 3000ms a time.";
      ProcessRestoreRecord(source_record_dir_);
    }
  }
  recorder_->Stop();
  is_terminating_ = true;
  AINFO << "wait for a while trying to complete the restore work";
  static constexpr int kMessageInterval = 1000;
  int interval_counter = 0;
  while (++interval_counter * kMessageInterval < recorder_wait_time_) {
    MonitorManager::Instance()->LogBuffer().WARN(
        "SmartRecorder is terminating...");
    std::this_thread::sleep_for(std::chrono::milliseconds(kMessageInterval));
  }
}

◆ Process()

bool apollo::data::RealtimeRecordProcessor::Process ( )