masstree_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_MASSTREE_MASSTREE_COMPOSER_IMPL_HPP_
#define FOEDUS_STORAGE_MASSTREE_MASSTREE_COMPOSER_IMPL_HPP_

#include <stdint.h>

#include <iosfwd>
#include <string>

#include "foedus/fwd.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/snapshot/snapshot_id.hpp"
#include "foedus/storage/composer.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/storage/masstree/fwd.hpp"
#include "foedus/storage/masstree/masstree_page_impl.hpp"
#include "foedus/storage/masstree/masstree_storage.hpp"

namespace foedus {
namespace storage {
namespace masstree {
class MasstreeComposer final {
 public:
  explicit MasstreeComposer(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);

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

  ErrorStack check_buddies(const Composer::ConstructRootArguments& args) const;

  MasstreePage* resolve_volatile(VolatilePagePointer pointer);

  Composer::DropResult drop_volatiles_recurse(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer);
  Composer::DropResult drop_volatiles_intermediate(
    const Composer::DropVolatilesArguments& args,
    MasstreeIntermediatePage* page);
  Composer::DropResult drop_volatiles_border(
    const Composer::DropVolatilesArguments& args,
    MasstreeBorderPage* page);
  bool is_updated_pointer(
    const Composer::DropVolatilesArguments& args,
    SnapshotPagePointer pointer) const;
  bool is_to_keep_volatile(uint8_t layer, uint16_t btree_level) const;

  void drop_foster_twins(const Composer::DropVolatilesArguments& args, MasstreePage* page);
  void drop_all_recurse(
    const Composer::DropVolatilesArguments& args,
    DualPagePointer* pointer);
  void drop_all_recurse_page_only(
    const Composer::DropVolatilesArguments& args,
    MasstreePage* page);
};


class MasstreeComposeContext {
 public:
  enum Constants {
    kMaxLevels = 32,
    kMaxLayers = 16,
    kMaxLogGroupSize = 1 << 14,
    kTmpBoundaryArraySize = kMaxLogGroupSize,
  };

  struct PathLevel {
    memory::PagePoolOffset head_;
    memory::PagePoolOffset tail_;
    uint16_t  layer_;
    SlotIndex next_original_;
    SlotIndex next_original_mini_;
    KeyLength next_original_remainder_;
    uint32_t  page_count_;
    uint32_t  filler_;
    KeySlice  low_fence_;
    KeySlice  high_fence_;
    KeySlice  next_original_slice_;

    bool has_next_original() const { return next_original_ != 0xFFFFU; }
    void set_no_more_next_original() {
      next_original_ = 0xFFFFU;
      next_original_mini_ = 0xFFFFU;
      next_original_remainder_ = kInitiallyNextLayer;
      next_original_slice_ = kSupremumSlice;
    }
    bool contains_slice(KeySlice slice) const {
      ASSERT_ND(low_fence_ <= slice);  // as logs are sorted, this must not happen
      return high_fence_ == kSupremumSlice || slice < high_fence_;  // careful on supremum case
    }
    bool contains_key(const char* key, KeyLength key_length) const {
      ASSERT_ND(is_key_aligned_and_zero_padded(key, key_length));
      KeySlice slice = normalize_be_bytes_full_aligned(key + layer_ * kSliceLen);
      return contains_slice(slice);
    }
    bool needs_to_consume_original(KeySlice slice, KeyLength key_length) const {
      return has_next_original()
        && (
          next_original_slice_ < slice
          || (next_original_slice_ == slice
              && next_original_remainder_ > kSliceLen
              && key_length > (layer_ + 1U) * kSliceLen));
    }

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

  struct FenceAndPointer {
    KeySlice low_fence_;
    SnapshotPagePointer pointer_;
  };

  struct PageBoundaryInfo {
    uint8_t btree_level_;
    uint8_t removed_;
    uint8_t layer_;
    uint8_t local_snapshot_pointer_high_;
    uint32_t local_snapshot_pointer_low_;

    KeySlice slices_[2];

    PageBoundaryInfo() = delete;
    ~PageBoundaryInfo() = delete;

    uint32_t dynamic_sizeof() const ALWAYS_INLINE { return (layer_ + 2U) * sizeof(KeySlice) + 8U; }
    static uint32_t calculate_hash(
      uint8_t btree_level,
      uint8_t layer,
      const KeySlice* prefixes,
      KeySlice low_fence,
      KeySlice high_fence) ALWAYS_INLINE {
      // good old multiply-hash
      uint64_t hash = btree_level + (layer * 256U);
      const uint64_t kMult = 0x7ada20dc734afb6fULL;
      for (uint8_t i = 0; i < layer; ++i) {
        hash = hash * kMult + prefixes[i];
      }
      hash = hash * kMult + low_fence;
      hash = hash * kMult + high_fence;
      uint32_t compressed = static_cast<uint32_t>(hash >> 32) ^ static_cast<uint32_t>(hash);
      return compressed;
    }
    SnapshotLocalPageId get_local_page_id() const ALWAYS_INLINE {
      return static_cast<SnapshotLocalPageId>(local_snapshot_pointer_high_) << 32
        | static_cast<SnapshotLocalPageId>(local_snapshot_pointer_low_);
    }

    bool exact_match(
      uint8_t btree_level,
      uint8_t layer,
      const KeySlice* prefixes,
      KeySlice low,
      KeySlice high) const ALWAYS_INLINE {
      if (layer != layer_ || btree_level != btree_level_ || removed_) {
        return false;
      }
      for (uint8_t i = 0; i < layer_; ++i) {
        if (prefixes[i] != slices_[i]) {
          return false;
        }
      }
      return low == slices_[layer_] && high == slices_[layer_ + 1];
    }
  };

  struct PageBoundarySort {
    uint32_t                  hash_;
    snapshot::BufferPosition  info_pos_;
    bool operator<(const PageBoundarySort& rhs) const ALWAYS_INLINE {
      return hash_ < rhs.hash_;
    }
    static bool static_less_than(
      const PageBoundarySort& lhs,
      const PageBoundarySort& rhs) ALWAYS_INLINE {
      return lhs.hash_ < rhs.hash_;
    }
  };

  MasstreeComposeContext(
    Engine* engine,
    snapshot::MergeSort* merge_sort,
    const Composer::ComposeArguments& args);
  ErrorStack execute();

 private:
  snapshot::SnapshotWriter* get_writer()  const { return args_.snapshot_writer_; }
  cache::SnapshotFileSet*   get_files()   const { return args_.previous_snapshot_files_; }
  memory::PagePoolOffset    allocate_page() {
    ASSERT_ND(allocated_pages_ < max_pages_);
    memory::PagePoolOffset new_offset = allocated_pages_;
    ++allocated_pages_;
    return new_offset;
  }

  MasstreePage*             get_page(memory::PagePoolOffset offset) const ALWAYS_INLINE;
  MasstreePage*             get_original(memory::PagePoolOffset offset) const ALWAYS_INLINE;
  MasstreeIntermediatePage* as_intermdiate(MasstreePage* page) const ALWAYS_INLINE;
  MasstreeBorderPage*       as_border(MasstreePage* page) const ALWAYS_INLINE;
  uint16_t get_cur_prefix_length() const { return get_last_layer() * sizeof(KeySlice); }
  uint8_t                   get_last_level_index() const {
    ASSERT_ND(cur_path_levels_ > 0);
    return cur_path_levels_ - 1U;
  }
  PathLevel*                get_last_level() { return cur_path_ + get_last_level_index(); }
  const PathLevel*          get_last_level() const  { return cur_path_ + get_last_level_index(); }
  uint8_t                   get_last_layer() const  {
    if (cur_path_levels_ == 0) {
      return 0;
    } else {
      return get_last_level()->layer_;
    }
  }
  void                      store_cur_prefix(uint8_t layer, KeySlice prefix_slice);

  PathLevel*                get_second_last_level() {
    ASSERT_ND(cur_path_levels_ > 1U);
    return cur_path_ + (cur_path_levels_ - 2U);
  }

  ErrorStack  execute_a_log(uint32_t cur);
  ErrorStack  execute_same_key_group(uint32_t from, uint32_t to);
  ErrorStack  execute_insert_group(uint32_t from, uint32_t to);
  uint32_t    execute_insert_group_get_cur_run(
    uint32_t cur,
    uint32_t to,
    KeySlice* min_slice,
    KeySlice* max_slice);
  ErrorCode   execute_insert_group_append_loop(uint32_t from, uint32_t to, uint32_t hint_count);
  ErrorStack  execute_delete_group(uint32_t from, uint32_t to);
  ErrorStack  execute_update_group(uint32_t from, uint32_t to);
  ErrorStack  execute_overwrite_group(uint32_t from, uint32_t to);

  void        write_dummy_page_zero();

  // init/uninit called only once or at most a few times
  ErrorStack  init_root();
  ErrorStack  open_first_level(const char* key, KeyLength key_length);
  ErrorStack  open_next_level(
    const char* key,
    KeyLength key_length,
    MasstreePage* parent,
    KeySlice prefix_slice,
    SnapshotPagePointer* next_page_id);
  void        open_next_level_create_layer(
    const char* key,
    KeyLength key_length,
    MasstreeBorderPage* parent,
    KeySlice prefix_slice,
    SlotIndex parent_index);

  ErrorStack  flush_buffer();
  inline ErrorStack flush_if_nearly_full() {
    uint64_t threshold = max_pages_ * 8ULL / 10ULL;
    if (UNLIKELY(allocated_pages_ >= threshold || allocated_pages_ + 256U >= max_pages_)) {
      CHECK_ERROR(flush_buffer());
    }
    return kRetOk;
  }

  ErrorStack  adjust_path(const char* key, KeyLength key_length);

  ErrorStack  consume_original_upto_border(KeySlice slice, KeyLength key_length, PathLevel* level);
  ErrorStack  consume_original_upto_intermediate(KeySlice slice, PathLevel* level);
  ErrorStack  consume_original_all();
  ErrorStack  close_level_grow_subtree(
    SnapshotPagePointer* root_pointer,
    KeySlice subtree_low,
    KeySlice subtree_high);
  ErrorStack  pushup_non_root();

  // next methods called for each log entry. must be efficient! So these methods return ErrorCode.

  bool        does_need_to_adjust_path(const char* key, KeyLength key_length) const ALWAYS_INLINE;

  ErrorStack  close_path_layer(uint16_t max_layer);
  ErrorStack  close_last_level();
  ErrorStack  close_first_level();
  ErrorStack  close_all_levels();

  void        append_border(
    KeySlice slice,
    xct::XctId xct_id,
    KeyLength remainder_length,
    const char* suffix,
    PayloadLength payload_count,
    const char* payload,
    PathLevel* level);
  void        append_border_next_layer(
    KeySlice slice,
    xct::XctId xct_id,
    SnapshotPagePointer pointer,
    PathLevel* level);
  void        append_border_newpage(KeySlice slice, PathLevel* level);
  void        append_intermediate(
    KeySlice low_fence,
    SnapshotPagePointer pointer,
    PathLevel* level);
  void        append_intermediate_newpage_and_pointer(
    KeySlice low_fence,
    SnapshotPagePointer pointer,
    PathLevel* level);


  void refresh_page_boundary_info_variables();
  ErrorCode close_level_register_page_boundaries();
  void remove_old_page_boundary_info(SnapshotPagePointer page_id, MasstreePage* page);
  PageBoundaryInfo* get_page_boundary_info(snapshot::BufferPosition pos) ALWAYS_INLINE {
    ASSERT_ND(pos <= page_boundary_info_cur_pos_);
    return reinterpret_cast<PageBoundaryInfo*>(page_boundary_info_ + pos * 8ULL);
  }
  const PageBoundaryInfo* get_page_boundary_info(snapshot::BufferPosition pos) const ALWAYS_INLINE {
    ASSERT_ND(pos <= page_boundary_info_cur_pos_);
    return reinterpret_cast<PageBoundaryInfo*>(page_boundary_info_ + pos * 8ULL);
  }
  void assert_page_boundary_not_exists(
    uint8_t btree_level,
    uint8_t layer,
    const KeySlice* prefixes,
    KeySlice low,
    KeySlice high) const ALWAYS_INLINE;
  void sort_page_boundary_info();
  SnapshotPagePointer lookup_page_boundary_info(
    uint8_t btree_level,
    uint8_t layer,
    const KeySlice* prefixes,
    KeySlice low,
    KeySlice high) const ALWAYS_INLINE;
  ErrorStack install_snapshot_pointers(uint64_t* installed_count) const;
  ErrorCode install_snapshot_pointers_recurse(
    const memory::GlobalVolatilePageResolver& resolver,
    uint8_t layer,
    KeySlice* prefixes,
    MasstreePage* volatile_page,
    uint64_t* installed_count) const ALWAYS_INLINE;
  ErrorCode install_snapshot_pointers_recurse_intermediate(
    const memory::GlobalVolatilePageResolver& resolver,
    uint8_t layer,
    KeySlice* prefixes,
    MasstreeIntermediatePage* volatile_page,
    uint64_t* installed_count) const;
  ErrorCode install_snapshot_pointers_recurse_border(
    const memory::GlobalVolatilePageResolver& resolver,
    uint8_t layer,
    KeySlice* prefixes,
    MasstreeBorderPage* volatile_page,
    uint64_t* installed_count) const;

  Engine* const             engine_;
  snapshot::MergeSort* const  merge_sort_;
  const StorageId           id_;
  MasstreeStorage           storage_;
  const Composer::ComposeArguments& args_;

  const snapshot::SnapshotId snapshot_id_;
  const uint16_t            numa_node_;
  const uint32_t            max_pages_;

  MasstreeIntermediatePage* const root_;

  Page* const       page_base_;
  Page* const       original_base_;

  // const members up to here.

  SnapshotPagePointer       page_id_base_;

  char*                     page_boundary_info_;
  PageBoundarySort*         page_boundary_sort_;

  snapshot::BufferPosition  page_boundary_info_cur_pos_;
  uint32_t                  page_boundary_elements_;

  snapshot::BufferPosition  page_boundary_info_capacity_;
  uint32_t                  max_page_boundary_elements_;

  uint32_t                  allocated_pages_;

  uint8_t                   root_index_;
  uint8_t                   root_index_mini_;

  uint8_t                   cur_path_levels_;
  PathLevel                 cur_path_[kMaxLevels];
  char                      cur_prefix_be_[kMaxLayers * kSliceLen];
  KeySlice                  cur_prefix_slices_[kMaxLayers];
  KeySlice                  tmp_boundary_array_[kTmpBoundaryArraySize];
};

}  // namespace masstree
}  // namespace storage
}  // namespace foedus
#endif  // FOEDUS_STORAGE_MASSTREE_MASSTREE_COMPOSER_IMPL_HPP_