logger_impl.hpp

Go to the documentation of this file.

/*
 * Copyright (c) 2014-2015, Hewlett-Packard Development Company, LP.
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License as published by the Free
 * Software Foundation; either version 2 of the License, or (at your option)
 * any later version.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details. You should have received a copy of the GNU General Public
 * License along with this program; if not, write to the Free Software
 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
 *
 * HP designates this particular file as subject to the "Classpath" exception
 * as provided by HP in the LICENSE.txt file that accompanied this code.
 */
#ifndef FOEDUS_LOG_LOGGER_IMPL_HPP_
#define FOEDUS_LOG_LOGGER_IMPL_HPP_
#include <stdint.h>

#include <atomic>
#include <iosfwd>
#include <mutex>
#include <string>
#include <thread>
#include <vector>

#include "foedus/attachable.hpp"
#include "foedus/epoch.hpp"
#include "foedus/fwd.hpp"
#include "foedus/initializable.hpp"
#include "foedus/fs/fwd.hpp"
#include "foedus/fs/path.hpp"
#include "foedus/log/epoch_history.hpp"
#include "foedus/log/fwd.hpp"
#include "foedus/log/log_id.hpp"
#include "foedus/log/logger_ref.hpp"
#include "foedus/memory/aligned_memory.hpp"
#include "foedus/savepoint/fwd.hpp"
#include "foedus/soc/shared_memory_repo.hpp"
#include "foedus/soc/shared_mutex.hpp"
#include "foedus/soc/shared_polling.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/thread/thread_id.hpp"

namespace foedus {
namespace log {

struct LoggerControlBlock {
  enum Constants {
    kMaxEpochHistory = 1 << 16,
  };

  // this is backed by shared memory. not instantiation. just reinterpret_cast.
  LoggerControlBlock() = delete;
  ~LoggerControlBlock() = delete;

  void initialize() {
    wakeup_cond_.initialize();
    epoch_history_mutex_.initialize();
    stop_requested_ = false;
    epoch_history_head_ = 0;
    epoch_history_count_ = 0;
  }
  void uninitialize() {
    epoch_history_mutex_.uninitialize();
  }


  bool is_epoch_history_empty() const { return epoch_history_count_ == 0; }
  uint32_t get_tail_epoch_history() const {
    return wrap_epoch_history_index(epoch_history_head_ + epoch_history_count_ - 1U);
  }

  static uint32_t wrap_epoch_history_index(uint32_t index) { return index % kMaxEpochHistory; }

  std::atomic< Epoch::EpochInteger >  durable_epoch_;
  soc::SharedPolling              wakeup_cond_;

  Epoch                           marked_epoch_;

  // the followings are also shared just for savepoint.

  std::atomic< LogFileOrdinal >   oldest_ordinal_;
  std::atomic< uint64_t >         oldest_file_offset_begin_;
  std::atomic< LogFileOrdinal >   current_ordinal_;

  std::atomic< uint64_t >         current_file_durable_offset_;

  std::atomic<bool>               stop_requested_;

  soc::SharedMutex  epoch_history_mutex_;

  uint32_t          epoch_history_head_;
  uint32_t          epoch_history_count_;

  EpochHistory      epoch_histories_[kMaxEpochHistory];
};

class Logger final : public DefaultInitializable, public LoggerRef {
 public:
  Logger(
    Engine* engine,
    LoggerControlBlock* control_block,
    LoggerId id,
    thread::ThreadGroupId numa_node,
    uint8_t in_node_ordinal,
    const fs::Path &log_folder,
    const std::vector< thread::ThreadId > &assigned_thread_ids)
  : LoggerRef(engine, control_block, id, numa_node, in_node_ordinal),
    log_folder_(log_folder),
    assigned_thread_ids_(assigned_thread_ids) {}
  ErrorStack  initialize_once() override;
  ErrorStack  uninitialize_once() override;

  Logger() = delete;
  Logger(const Logger &other) = delete;
  Logger& operator=(const Logger &other) = delete;

  LogFileOrdinal get_current_ordinal() const { return control_block_->current_ordinal_; }
  bool is_stop_requested() const { return control_block_->stop_requested_; }

  std::string             to_string() const;
  friend std::ostream&    operator<<(std::ostream& o, const Logger& v);

 private:
  void        handle_logger();

  ErrorStack  update_durable_epoch(Epoch new_durable_epoch, bool had_any_log);
  Epoch calculate_min_durable_epoch();

  ErrorStack  switch_file_if_required();

  ErrorStack  check_lagging_threads(Epoch new_epoch);

  ErrorStack  log_epoch_switch(Epoch new_epoch);

  ErrorStack  write_dummy_epoch_mark();

  ErrorStack  write_one_epoch(Epoch write_epoch);
  ErrorStack  write_one_epoch_piece(
    const ThreadLogBuffer& buffer,
    Epoch write_epoch,
    uint64_t from_offset,
    uint64_t upto_offset);

  void        assert_consistent();
  void        assert_written_logs(Epoch write_epoch, const char* logs, uint64_t bytes) const;

  const fs::Path                  log_folder_;
  const std::vector< thread::ThreadId > assigned_thread_ids_;

  std::thread                     logger_thread_;

  memory::AlignedMemory           fill_buffer_;

  fs::DirectIoFile*               current_file_;
  fs::Path                        current_file_path_;

  std::vector< thread::Thread* >  assigned_threads_;

  std::mutex                      epoch_switch_mutex_;
};
static_assert(
  sizeof(LoggerControlBlock) <= soc::NodeMemoryAnchors::kLoggerMemorySize,
  "LoggerControlBlock is too large.");
}  // namespace log
}  // namespace foedus
#endif  // FOEDUS_LOG_LOGGER_IMPL_HPP_