thread_log_buffer.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_THREAD_LOG_BUFFER_HPP_
#define FOEDUS_LOG_THREAD_LOG_BUFFER_HPP_
#include <stdint.h>

#include <iosfwd>

#include "foedus/cxx11.hpp"
#include "foedus/epoch.hpp"
#include "foedus/fwd.hpp"
#include "foedus/initializable.hpp"
#include "foedus/log/fwd.hpp"
#include "foedus/memory/aligned_memory.hpp"
#include "foedus/thread/thread_id.hpp"

namespace foedus {
namespace log {
struct ThreadEpockMark {
  Epoch       old_epoch_;
  Epoch       new_epoch_;
  uint64_t    offset_begin_;
  uint64_t    offset_end_;

  ThreadEpockMark() {
    old_epoch_.reset();
    new_epoch_.reset();
    offset_begin_ = 0;
    offset_end_ = 0;
  }
  ThreadEpockMark(Epoch old_epoch, Epoch new_epoch, uint64_t offset_begin) {
    old_epoch_ = old_epoch;
    new_epoch_ = new_epoch;
    offset_begin_ = offset_begin;
    offset_end_ = 0;
  }
  friend std::ostream& operator<<(std::ostream& o, const ThreadEpockMark& v);
};

struct ThreadLogBufferMeta CXX11_FINAL {
  enum Constants {
    kMaxNonDurableEpochs = 64,
  };

  ThreadLogBufferMeta();

  void  assert_consistent() const;
  friend std::ostream& operator<<(std::ostream& o, const ThreadLogBufferMeta& v);

  static uint32_t increment_mark_index(uint32_t index) {
    return (index + 1) % kMaxNonDurableEpochs;
  }

  thread::ThreadId                thread_id_;
  uint16_t                        padding1_;
  uint32_t                        padding2_;

  uint64_t                        buffer_size_;
  uint64_t                        buffer_size_safe_;

  uint64_t                        offset_head_;

  uint64_t                        offset_durable_;

  uint64_t                        offset_committed_;

  uint64_t                        offset_tail_;

  ThreadEpockMark                 thread_epoch_marks_[kMaxNonDurableEpochs];
  uint32_t                        oldest_mark_index_;
  uint32_t                        current_mark_index_;
};

class ThreadLogBuffer CXX11_FINAL : public DefaultInitializable {
 public:
  friend class Logger;
  static uint64_t distance(uint64_t buffer_size, uint64_t from, uint64_t to) ALWAYS_INLINE {
    ASSERT_ND(from < buffer_size);
    ASSERT_ND(to < buffer_size);
    if (to >= from) {
      return to - from;
    } else {
      return to + buffer_size - from;  // wrap around
    }
  }
  static void advance(uint64_t buffer_size, uint64_t *target, uint64_t amount) ALWAYS_INLINE {
    ASSERT_ND(*target < buffer_size);
    ASSERT_ND(amount < buffer_size);
    *target += amount;
    if (*target >= buffer_size) {
      *target -= buffer_size;
    }
  }

  ThreadLogBuffer(Engine* engine, thread::ThreadId thread_id);
  ErrorStack  initialize_once() CXX11_OVERRIDE;
  ErrorStack  uninitialize_once() CXX11_OVERRIDE;

  ThreadLogBuffer() CXX11_FUNC_DELETE;
  ThreadLogBuffer(const ThreadLogBuffer &other) CXX11_FUNC_DELETE;
  ThreadLogBuffer& operator=(const ThreadLogBuffer &other) CXX11_FUNC_DELETE;

  void        assert_consistent() const {
#ifndef NDEBUG
    meta_.assert_consistent();
#endif  // NDEBUG
  }
  thread::ThreadId get_thread_id() const { return meta_.thread_id_; }

  char*       reserve_new_log(uint16_t log_length) ALWAYS_INLINE {
    if (UNLIKELY(log_length + meta_.offset_tail_ >= meta_.buffer_size_)) {
      // now we need wrap around. to simplify, let's avoid having a log entry spanning the
      // end of the buffer. put a filler log to fill the rest.
      fillup_tail();
      ASSERT_ND(meta_.offset_tail_ == 0);
    }
    // also make sure tail isn't too close to head (full). in that case, we wait for loggers
    if (UNLIKELY(
      head_to_tail_distance() + log_length >= meta_.buffer_size_safe_)) {
      wait_for_space(log_length);
      ASSERT_ND(head_to_tail_distance() + log_length < meta_.buffer_size_safe_);
    }
    ASSERT_ND(head_to_tail_distance() + log_length < meta_.buffer_size_safe_);
    char *buffer = buffer_ + meta_.offset_tail_;
    advance(meta_.buffer_size_, &meta_.offset_tail_, log_length);
    return buffer;
  }
  uint64_t    head_to_tail_distance() const ALWAYS_INLINE {
    return distance(meta_.buffer_size_, meta_.offset_head_, meta_.offset_tail_);
  }

  void        publish_committed_log(Epoch commit_epoch) ALWAYS_INLINE {
    Epoch last_epoch = get_last_epoch();
    ASSERT_ND(commit_epoch >= last_epoch);
    if (UNLIKELY(commit_epoch > last_epoch)) {
      on_new_epoch_observed(commit_epoch);  // epoch switches!
    } else if (UNLIKELY(commit_epoch < last_epoch)) {
      // This MUST not happen because it means an already durable epoch received a new log!
      crash_stale_commit_epoch(commit_epoch);
    }
    meta_.offset_committed_ = meta_.offset_tail_;
  }

  void        discard_current_xct_log() {
    meta_.offset_tail_ = meta_.offset_committed_;
  }
  Epoch       get_last_epoch() const {
    return meta_.thread_epoch_marks_[meta_.current_mark_index_].new_epoch_;
  }

  void        wait_for_space(uint16_t required_space);

  uint64_t    get_offset_head() const { return meta_.offset_head_; }

  uint64_t    get_offset_durable() const { return meta_.offset_durable_; }

  uint64_t    get_offset_committed() const { return meta_.offset_committed_; }

  uint64_t    get_offset_tail() const { return meta_.offset_tail_; }


  const ThreadLogBufferMeta& get_meta() const { return meta_; }
  const char* get_buffer() const { return buffer_; }

  struct OffsetRange {
    uint64_t begin_;
    uint64_t end_;
    bool is_empty() const { return begin_ == end_; }
  };
  OffsetRange get_logs_to_write(Epoch written_epoch);
  void        on_log_written(Epoch written_epoch);


  friend std::ostream& operator<<(std::ostream& o, const ThreadLogBuffer& v);

 private:
  void        on_new_epoch_observed(Epoch commit_epoch);

  void        crash_stale_commit_epoch(Epoch commit_epoch);

  void        fillup_tail();

  Engine* const             engine_;
  ThreadLogBufferMeta       meta_;


  char*                     buffer_;
};
}  // namespace log
}  // namespace foedus
#endif  // FOEDUS_LOG_THREAD_LOG_BUFFER_HPP_