hash_composer_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_COMPOSER_IMPL_HPP_
#define FOEDUS_STORAGE_HASH_HASH_COMPOSER_IMPL_HPP_

#include <stdint.h>

#include <iosfwd>
#include <string>

#include "foedus/compiler.hpp"
#include "foedus/fwd.hpp"
#include "foedus/cache/fwd.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/storage/composer.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/hash/fwd.hpp"
#include "foedus/storage/hash/hash_composed_bins_impl.hpp"
#include "foedus/storage/hash/hash_id.hpp"
#include "foedus/storage/hash/hash_page_impl.hpp"
#include "foedus/storage/hash/hash_storage.hpp"
#include "foedus/storage/hash/hash_tmpbin.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace storage {
namespace hash {
class HashComposer final {
 public:
  explicit HashComposer(Composer *parent);

  std::string to_string() const;

  ErrorStack compose(const Composer::ComposeArguments& args);
  ErrorStack construct_root(const Composer::ConstructRootArguments& args);


  Composer::DropResult  drop_volatiles(const Composer::DropVolatilesArguments& args);
  void                  drop_root_volatile(const Composer::DropVolatilesArguments& args);

  static void           launch_construct_root_multi_level(
    HashComposer* pointer,
    const Composer::ConstructRootArguments* args,
    uint16_t numa_node,
    HashIntermediatePage* root_page,
    ErrorStack* out_error) {
    *out_error = pointer->construct_root_multi_level(*args, numa_node, root_page);
  }

 private:
  Engine* const             engine_;
  const StorageId           storage_id_;
  const HashStorage         storage_;

  const memory::GlobalVolatilePageResolver& volatile_resolver_;

  HashDataPage* resolve_data(VolatilePagePointer pointer) const ALWAYS_INLINE;
  HashIntermediatePage* resolve_intermediate(VolatilePagePointer pointer) const ALWAYS_INLINE;

  ErrorStack construct_root_single_level(
    const Composer::ConstructRootArguments& args,
    HashIntermediatePage* root_page);
  ErrorStack construct_root_multi_level(
    const Composer::ConstructRootArguments& args,
    uint16_t numa_node,
    HashIntermediatePage* root_page);

  void drop_volatiles_child(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* child_pointer,
    uint8_t parent_level,
    Composer::DropResult *result);
  Composer::DropResult drop_volatiles_recurse(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer);

  bool can_drop_volatile_bin(VolatilePagePointer head, Epoch valid_until) const;

  void drop_volatile_entire_bin(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer_to_head) const;

  bool is_to_keep_volatile(uint16_t level);
  void drop_all_recurse(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer);
};

class HashComposeContext {
 public:
  HashComposeContext(
    Engine*                           engine,
    snapshot::MergeSort*              merge_sort,
    snapshot::SnapshotWriter*         snapshot_writer,
    cache::SnapshotFileSet*           previous_snapshot_files,
    Page*                             root_info_page);

  ErrorStack execute();

 private:
  ErrorCode apply_batch(uint64_t cur, uint64_t next);

  ErrorStack finalize();

  ErrorCode dump_data_pages();

  bool verify_new_pointer(storage::SnapshotPagePointer pointer) const;
  bool verify_old_pointer(storage::SnapshotPagePointer pointer) const;

  HashDataPage* resolve_data(VolatilePagePointer pointer) const ALWAYS_INLINE;
  HashIntermediatePage* resolve_intermediate(VolatilePagePointer pointer) const ALWAYS_INLINE;

  bool is_initial_snapshot() const { return previous_root_page_pointer_ == 0; }


  HashIntermediatePage*   get_cur_path_page(uint8_t level) const { return cur_path_ + level; }
  HashIntermediatePage*   get_cur_path_lowest() const {
    return get_cur_path_page(cur_path_lowest_level_);
  }
  SnapshotPagePointer     get_cur_path_bin_head(HashBin bin) const;

  ErrorStack              init_cur_path();
  ErrorCode               update_cur_path(HashBin bin);
  ErrorCode               update_cur_path_if_needed(HashBin bin) ALWAYS_INLINE;
  bool                    verify_cur_path() const;


  ErrorStack              close_cur_bin();
  ErrorStack              open_cur_bin(HashBin bin);


  ErrorStack              init_intermediates();
  HashComposedBinsPage*   get_intermediate_head(uint8_t root_index) const {
    ASSERT_ND(root_index < storage_.get_root_children());
    return intermediate_base_ + root_index;
  }
  HashComposedBinsPage*   get_intermediate_tail(uint8_t root_index) const;

  void                    update_cur_intermediate_tail(HashBin bin);
  void                    update_cur_intermediate_tail_if_needed(HashBin bin) ALWAYS_INLINE;

  ErrorCode               append_to_intermediate(SnapshotPagePointer page_id, HashBin bin);

  ErrorCode               expand_intermediate_pool_if_needed();

  ErrorStack install_snapshot_data_pages(uint64_t* installed_count) const;
  ErrorStack install_snapshot_data_pages_root_child(
    const HashComposedBinsPage* composed,
    HashIntermediatePage* volatile_root_child,
    uint64_t* installed_count) const;

  Engine* const                   engine_;
  snapshot::MergeSort* const      merge_sort_;
  const Epoch                     system_initial_epoch_;
  const StorageId                 storage_id_;
  const snapshot::SnapshotId      snapshot_id_;
  const HashStorage               storage_;
  snapshot::SnapshotWriter* const snapshot_writer_;
  cache::SnapshotFileSet*  const  previous_snapshot_files_;
  HashRootInfoPage* const         root_info_page_;

  const bool                      partitionable_;
  const uint8_t                   levels_;
  const uint8_t                   bin_bits_;
  const uint8_t                   bin_shifts_;
  const uint16_t                  root_children_;
  const uint16_t                  numa_node_;
  const HashBin                   total_bin_count_;
  const SnapshotPagePointer       previous_root_page_pointer_;

  const memory::GlobalVolatilePageResolver& volatile_resolver_;


  HashIntermediatePage*           cur_path_;
  memory::AlignedMemory           cur_path_memory_;

  uint8_t                         cur_path_lowest_level_;

  HashBinRange                    cur_path_valid_range_;

  HashBin                         cur_bin_;
  const HashBin                   kCurBinNotOpened = (1ULL << kHashMaxBinBits);

  HashTmpBin                      cur_bin_table_;

  memory::AlignedMemory           data_page_io_memory_;

  HashComposedBinsPage*           cur_intermediate_tail_;

  uint32_t                  allocated_pages_;
  uint32_t                  allocated_intermediates_;
  uint32_t                  max_pages_;
  uint32_t                  max_intermediates_;
  HashDataPage*             page_base_;
  HashComposedBinsPage*     intermediate_base_;
};


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