async_logger.h

Go to the documentation of this file.

/******************************************************************************
 * Copyright 2018 The Apollo Authors. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *****************************************************************************/

#ifndef CYBER_LOGGER_ASYNC_LOGGER_H_
#define CYBER_LOGGER_ASYNC_LOGGER_H_

#include <atomic>
#include <condition_variable>
#include <cstdint>
#include <ctime>
#include <deque>
#include <iostream>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <unordered_map>
#include <utility>
#include <vector>

#include "glog/logging.h"

#include "cyber/common/macros.h"
#include "cyber/logger/log_file_object.h"

namespace apollo {
namespace cyber {
namespace logger {

class AsyncLogger : public google::base::Logger {
 public:
  explicit AsyncLogger(google::base::Logger* wrapped);

  ~AsyncLogger();

  void Start();

  void Stop();

  void Write(bool force_flush, time_t timestamp, const char* message,
             int message_len) override;

  void Flush() override;

  uint32_t LogSize() override;

  std::thread* LogThread() { return &log_thread_; }

 private:
  // A buffered message.
  //
  // TODO(todd): using std::string for buffered messages is convenient but not
  // as efficient as it could be. It's better to make the buffers just be
  // Arenas and allocate both the message data and Msg struct from them, forming
  // a linked list.
  struct Msg {
    time_t ts;
    std::string message;
    int32_t level;
    Msg() : ts(0), message(), level(google::INFO) {}
    Msg(time_t ts, std::string&& message, int32_t level)
        : ts(ts), message(std::move(message)), level(level) {}
    Msg(const Msg& rsh) {
      ts = rsh.ts;
      message = rsh.message;
      level = rsh.level;
    }
    Msg(Msg&& rsh) {
      ts = rsh.ts;
      message = rsh.message;
      level = rsh.level;
    }
    Msg& operator=(Msg&& rsh) {
      ts = rsh.ts;
      message = std::move(rsh.message);
      level = rsh.level;
      return *this;
    }
    Msg& operator=(const Msg& rsh) {
      ts = rsh.ts;
      message = rsh.message;
      level = rsh.level;
      return *this;
    }
  };

  void RunThread();
  void FlushBuffer(const std::unique_ptr<std::deque<Msg>>& msg);

  google::base::Logger* const wrapped_;
  std::thread log_thread_;

  // Count of how many times the writer thread has flushed the buffers.
  // 64 bits should be enough to never worry about overflow.
  std::atomic<uint64_t> flush_count_ = {0};

  // Count of how many times the writer thread has dropped the log messages.
  // 64 bits should be enough to never worry about overflow.
  uint64_t drop_count_ = 0;

  // The buffer to which application threads append new log messages.
  std::unique_ptr<std::deque<Msg>> active_buf_;

  // The buffer currently being flushed by the logger thread, cleared
  // after a successful flush.
  std::unique_ptr<std::deque<Msg>> flushing_buf_;

  // Trigger for the logger thread to stop.
  enum State { INITTED, RUNNING, STOPPED };
  std::atomic<State> state_ = {INITTED};
  std::atomic_flag flag_ = ATOMIC_FLAG_INIT;
  std::unordered_map<std::string, std::unique_ptr<LogFileObject>>
      module_logger_map_;

  DISALLOW_COPY_AND_ASSIGN(AsyncLogger);
};

}  // namespace logger
}  // namespace cyber
}  // namespace apollo

#endif  // CYBER_LOGGER_ASYNC_LOGGER_H_