hash_composed_bins_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_STORAGE_HASH_HASH_COMPOSED_BINS_IMPL_HPP_
#define FOEDUS_STORAGE_HASH_HASH_COMPOSED_BINS_IMPL_HPP_

#include <stdint.h>

#include <memory>

#include "foedus/compiler.hpp"
#include "foedus/error_stack.hpp"
#include "foedus/cache/fwd.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/snapshot/snapshot_id.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/storage/hash/fwd.hpp"
#include "foedus/storage/hash/hash_id.hpp"
#include "foedus/storage/hash/hash_page_impl.hpp"

namespace foedus {
namespace storage {
namespace hash {

typedef HashIntermediatePage HashRootInfoPage;

struct ComposedBin final {
  HashBin             bin_;
  SnapshotPagePointer page_id_;
};

const uint16_t kHashComposedBinsPageMaxBins = (kPageSize - 80) / sizeof(ComposedBin);

struct HashComposedBinsPage final {
  PageHeader          header_;      // +40 -> 40
  HashBinRange        bin_range_;   // +16 -> 56
  SnapshotPagePointer next_page_;   // +8  -> 64
  uint32_t            bin_count_;   // +4  -> 68
  uint32_t            dummy_;       // +4  -> 72
  uint64_t            padding_;     // +8  -> 80
  ComposedBin         bins_[kHashComposedBinsPageMaxBins];    // -> 4096
};

struct ComposedBinsBuffer final {
  enum Constants {
    kMinBufferSize = 1 << 6,
  };
  cache::SnapshotFileSet* fileset_;
  SnapshotPagePointer   head_page_id_;
  uint32_t              total_pages_;
  uint32_t              buffer_size_;
  uint32_t              buffer_pos_;
  uint32_t              buffer_count_;

  uint32_t              cursor_buffer_;

  uint16_t              cursor_bin_;
  uint16_t              cursor_bin_count_;

  HashComposedBinsPage* buffer_;

  void init(
    cache::SnapshotFileSet* fileset,
    SnapshotPagePointer head_page_id,
    uint32_t total_pages,
    uint32_t buffer_size,
    HashComposedBinsPage* buffer);

  static void assure_read_buffer_size(memory::AlignedMemory* read_buffer, uint32_t inputs);

  ErrorCode next_pages();
  ErrorCode next_bin() ALWAYS_INLINE;
  const HashComposedBinsPage& get_cur_page() const ALWAYS_INLINE {
    return buffer_[cursor_buffer_];
  }
  const ComposedBin& get_cur_bin() const ALWAYS_INLINE {
    ASSERT_ND(has_more());
    ASSERT_ND(cursor_bin_count_ == get_cur_page().bin_count_);
    ASSERT_ND(cursor_bin_ < cursor_bin_count_);
    return get_cur_page().bins_[cursor_bin_];
  }
  bool has_more() const { return buffer_pos_ < total_pages_; }
};

struct ComposedBinsMergedStream final {
  typedef HashIntermediatePage* PagePtr;

  std::unique_ptr< ComposedBinsBuffer[] > inputs_memory_;

  ComposedBinsBuffer*   inputs_;
  uint32_t              input_count_;
  snapshot::SnapshotId  snapshot_id_;
  uint8_t               levels_;
  uint8_t               padding_;

  PagePtr               cur_path_[kHashMaxLevels];

  ErrorStack init(
    const HashRootInfoPage* const*  inputs,
    uint32_t                  input_count,
    PagePtr                   root_page,
    uint16_t                  root_child_index,
    memory::AlignedMemory*    read_buffer,
    cache::SnapshotFileSet*   fileset,
    snapshot::SnapshotWriter* writer,
    uint32_t*                 writer_buffer_pos,
    uint32_t*                 writer_higher_buffer_pos,
    bool*                     had_any_change);

  ErrorCode process_a_bin(uint32_t* installed_count, HashBin* next_lowest_bin);

  ErrorCode switch_path(
    HashBin                   lowest_bin,
    cache::SnapshotFileSet*   fileset,
    snapshot::SnapshotWriter* writer,
    uint32_t*                 writer_buffer_pos,
    uint32_t*                 writer_higher_buffer_pos);

  ErrorCode open_path(
    HashBin bin,
    uint8_t fixed_upto_level,
    cache::SnapshotFileSet* fileset,
    snapshot::SnapshotWriter* writer,
    uint32_t* writer_buffer_pos,
    uint32_t* writer_higher_buffer_pos);

  ErrorCode assure_writer_buffer(
    snapshot::SnapshotWriter* writer,
    uint32_t*                 writer_buffer_pos,
    uint32_t                  writer_higher_buffer_pos);
};

inline ErrorCode ComposedBinsBuffer::next_bin() {
  if (LIKELY(cursor_bin_ + 1U < cursor_bin_count_)) {
    ++cursor_bin_;
    return kErrorCodeOk;
  } else if (LIKELY(cursor_buffer_ + 1U < buffer_count_)) {
    ++cursor_buffer_;
    cursor_bin_ = 0;
    cursor_bin_count_ = buffer_[cursor_buffer_].bin_count_;
    return kErrorCodeOk;
  } else {
    return next_pages();
  }
}

static_assert(sizeof(HashRootInfoPage) == kPageSize, "incorrect sizeof(RootInfoPage)");
static_assert(sizeof(HashComposedBinsPage) == kPageSize, "incorrect sizeof(HashComposedBinsPage)");

}  // namespace hash
}  // namespace storage
}  // namespace foedus
#endif  // FOEDUS_STORAGE_HASH_HASH_COMPOSED_BINS_IMPL_HPP_