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_SNAPSHOT_LOG_BUFFER_HPP_
#define FOEDUS_SNAPSHOT_LOG_BUFFER_HPP_

#include <iosfwd>
#include <string>

#include "foedus/assert_nd.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/error_code.hpp"
#include "foedus/fs/fwd.hpp"
#include "foedus/log/common_log_types.hpp"
#include "foedus/memory/aligned_memory.hpp"
#include "foedus/snapshot/snapshot_id.hpp"

namespace foedus {
namespace snapshot {
struct LogBuffer {
  explicit LogBuffer(char* base_address) : base_address_(base_address) {}
  char* const base_address_;

  inline log::RecordLogType* resolve(BufferPosition position) const {
    return reinterpret_cast<log::RecordLogType*>(
      base_address_ + from_buffer_position(position));
  }
  inline BufferPosition compact(const log::RecordLogType* address) const {
    return to_buffer_position(reinterpret_cast<const char*>(address) - base_address_);
  }
};

class SortedBuffer {
 public:
  SortedBuffer(char* buffer, uint64_t buffer_size, uint64_t total_size)
    : buffer_(buffer),
      buffer_size_(buffer_size),
      offset_(0),
      total_size_(total_size),
      cur_block_storage_id_(0),
      cur_block_log_count_(0),
      cur_block_abosulte_begin_(0),
      cur_block_abosulte_end_(0) {
    assert_checks();
  }
  virtual ~SortedBuffer() {}

  uint64_t    to_relative_pos(uint64_t absolute_pos) const {
    ASSERT_ND(absolute_pos >= offset_);
    ASSERT_ND(absolute_pos <= offset_ + buffer_size_);
    return absolute_pos - offset_;
  }

  uint64_t    to_absolute_pos(uint64_t relative_pos) const {
    ASSERT_ND(relative_pos < buffer_size_);
    return relative_pos + offset_;
  }

  const char* get_buffer() const { return buffer_; }

  uint64_t    get_buffer_size() const { return buffer_size_; }

  uint64_t    get_offset() const { return offset_; }

  uint64_t    get_total_size() const { return total_size_; }

  storage::StorageId  get_cur_block_storage_id() const { return cur_block_storage_id_; }
  uint32_t            get_cur_block_log_count() const { return cur_block_log_count_; }

  uint64_t            get_cur_block_abosulte_begin() const { return cur_block_abosulte_begin_; }
  uint64_t            get_cur_block_abosulte_end() const { return cur_block_abosulte_end_; }
  uint32_t  get_cur_block_shortest_key_length() const { return cur_block_shortest_key_length_; }
  uint32_t  get_cur_block_longest_key_length() const { return cur_block_longest_key_length_; }

  void        set_current_block(
    storage::StorageId storage_id,
    uint32_t           log_count,
    uint64_t           abosulte_begin,
    uint64_t           abosulte_end,
    uint32_t           shortest_key_length,
    uint32_t           longest_key_length) {
    cur_block_storage_id_ = storage_id;
    cur_block_log_count_ = log_count;
    cur_block_abosulte_begin_ = abosulte_begin;
    cur_block_abosulte_end_ = abosulte_end;
    cur_block_shortest_key_length_ = shortest_key_length;
    cur_block_longest_key_length_ = longest_key_length;
    assert_checks();
  }
  bool        is_valid_current_block() const { return cur_block_storage_id_ != 0; }
  void        invalidate_current_block() { cur_block_storage_id_ = 0; }

  virtual ErrorCode     wind(uint64_t next_absolute_pos) = 0;

  virtual std::string   to_string() const = 0;

  virtual void          describe(std::ostream* o) const = 0;

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

  void assert_checks() {
    ASSERT_ND(offset_ <= total_size_);
    if (cur_block_storage_id_ != 0) {
      ASSERT_ND(cur_block_log_count_ > 0);
      ASSERT_ND(cur_block_abosulte_begin_ <= cur_block_abosulte_end_);
      ASSERT_ND(cur_block_abosulte_end_ <= total_size_);

      // did we overlook the begin/end of the block?
      ASSERT_ND(cur_block_abosulte_begin_ <= offset_ + buffer_size_);
      ASSERT_ND(cur_block_abosulte_end_ >= offset_);
      ASSERT_ND(cur_block_shortest_key_length_ <= cur_block_longest_key_length_);
      ASSERT_ND(cur_block_shortest_key_length_ > 0);
      ASSERT_ND(cur_block_longest_key_length_ <= 0xFFFFU);
    }
  }

 protected:
  char* const         buffer_;
  const uint64_t      buffer_size_;
  uint64_t            offset_;
  const uint64_t      total_size_;

  storage::StorageId  cur_block_storage_id_;
  uint32_t            cur_block_log_count_;
  uint64_t            cur_block_abosulte_begin_;
  uint64_t            cur_block_abosulte_end_;
  uint32_t            cur_block_shortest_key_length_;
  uint32_t            cur_block_longest_key_length_;

  void describe_base_elements(std::ostream* o) const;
};

class InMemorySortedBuffer CXX11_FINAL : public SortedBuffer {
 public:
  InMemorySortedBuffer(char* buffer, uint64_t buffer_size)
    : SortedBuffer(buffer, buffer_size, buffer_size) {
  }
  ~InMemorySortedBuffer() CXX11_OVERRIDE {}

  std::string to_string() const CXX11_OVERRIDE { return "<in-memory buffer>"; }
  ErrorCode   wind(uint64_t next_absolute_pos) CXX11_OVERRIDE {
    ASSERT_ND(next_absolute_pos <= buffer_size_);
    return kErrorCodeOk;
  }
  void        describe(std::ostream* o) const CXX11_OVERRIDE;
};

class DumpFileSortedBuffer CXX11_FINAL : public SortedBuffer {
 public:
  DumpFileSortedBuffer(fs::DirectIoFile *file, memory::AlignedMemorySlice io_buffer);
  ~DumpFileSortedBuffer() CXX11_OVERRIDE {}

  std::string to_string() const CXX11_OVERRIDE;
  ErrorCode   wind(uint64_t next_absolute_pos) CXX11_OVERRIDE;
  void        describe(std::ostream* o) const CXX11_OVERRIDE;

  fs::DirectIoFile*                 get_file()      const { return file_; }
  const memory::AlignedMemorySlice& get_io_buffer() const { return io_buffer_; }

 private:
  enum Constants {
    kAlignment = 1 << 12,
  };
  fs::DirectIoFile* const           file_;
  memory::AlignedMemorySlice const  io_buffer_;
};

}  // namespace snapshot
}  // namespace foedus
#endif  // FOEDUS_SNAPSHOT_LOG_BUFFER_HPP_