masstree_cursor.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_CURSOR_HPP_
#define FOEDUS_STORAGE_MASSTREE_MASSTREE_CURSOR_HPP_

#include <stdint.h>

#include <cstring>
#include <string>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/cxx11.hpp"
#include "foedus/engine.hpp"
#include "foedus/assorted/endianness.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/storage/masstree/fwd.hpp"
#include "foedus/storage/masstree/masstree_id.hpp"
#include "foedus/storage/masstree/masstree_record_location.hpp"
#include "foedus/storage/masstree/masstree_storage.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace storage {
namespace masstree {
class MasstreeCursor CXX11_FINAL {
 public:
  struct Route {
    enum MovedPageSearchStatus {
      kNone = 0,
      kMovedPageSearchedNeither = 1,
      kMovedPageSearchedOne = 2,
      kMovedPageSearchedBoth = 3,
    };
    MasstreePage* page_;
    PageVersionStatus stable_;
    SlotIndex index_;
    SlotIndex index_mini_;
    SlotIndex key_count_;
    SlotIndex key_count_mini_;

    bool      snapshot_;
    Layer     layer_;

    MovedPageSearchStatus moved_page_search_status_;

    KeySlice latest_separator_;

    SlotIndex order_[kBorderPageMaxSlots];

    SlotIndex get_cur_original_index() const ALWAYS_INLINE {
      return order_[index_];
    }
    SlotIndex get_original_index(SlotIndex index) const ALWAYS_INLINE {
      return order_[index];
    }
    bool    is_valid_record() const ALWAYS_INLINE {
      return index_ < key_count_;
    }
    bool    was_stably_moved() const { return stable_.is_moved(); }
    void setup_order();
  };
  enum SearchType {
    kForwardInclusive = 0,
    kForwardExclusive,
    kBackwardInclusive,
    kBackwardExclusive,
  };
  enum KeyCompareResult {
    kCurKeySmaller,
    kCurKeyEquals,
    kCurKeyLarger,
    // the following two are only when cur key points to next layer. instead, no equals for layer
    kCurKeyBeingsWith,
    kCurKeyContains,
  };
  enum Constants {
    kMaxRecords = kBorderPageMaxSlots,
    kMaxRoutes = kPageSize / sizeof(Route),
    kKeyLengthExtremum = 0,
  };

  MasstreeCursor(MasstreeStorage storage, thread::Thread* context);

  thread::Thread*   get_context() { return context_; }
  MasstreeStorage&  get_storage() { return storage_; }
  bool              is_for_writes() const { return for_writes_; }
  bool              is_forward_cursor() const { return forward_cursor_; }

  ErrorCode   open(
    const char* begin_key = CXX11_NULLPTR,
    KeyLength begin_key_length = kKeyLengthExtremum,
    const char* end_key = CXX11_NULLPTR,
    KeyLength end_key_length = kKeyLengthExtremum,
    bool forward_cursor = true,
    bool for_writes = false,
    bool begin_inclusive = true,
    bool end_inclusive = false);

  ErrorCode   open_normalized(
    KeySlice begin_key,
    KeySlice end_key,
    bool forward_cursor = true,
    bool for_writes = false,
    bool begin_inclusive = true,
    bool end_inclusive = false) {
    ASSERT_ND((forward_cursor && end_key >= begin_key) ||
      (!forward_cursor && end_key <= begin_key));
    KeySlice begin_key_be = assorted::htobe<KeySlice>(begin_key);
    KeySlice end_key_be = assorted::htobe<KeySlice>(end_key);
    return open(
      reinterpret_cast<const char*>(&begin_key_be),
      sizeof(KeySlice),
      reinterpret_cast<const char*>(&end_key_be),
      sizeof(KeySlice),
      forward_cursor,
      for_writes,
      begin_inclusive,
      end_inclusive);
  }

  bool      is_valid_record() const ALWAYS_INLINE {
    return route_count_ > 0 && !reached_end_ && cur_route()->is_valid_record();
  }
  KeyLength   get_key_length() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    return cur_key_length_;
  }
  const KeySlice* get_cur_route_prefix_slices() const ALWAYS_INLINE {
    return cur_route_prefix_slices_;
  }
  const char* get_cur_route_prefix_be() const ALWAYS_INLINE {
    return cur_route_prefix_be_;
  }
  KeySlice    get_key_in_layer_slice() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    return cur_key_in_layer_slice_;
  }
  const char* get_key_suffix() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    return cur_key_suffix_;
  }
  KeySlice    get_normalized_key() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    ASSERT_ND(cur_key_length_ <= sizeof(KeySlice));
    return cur_key_in_layer_slice_;
  };
  void        copy_combined_key(char* buffer) const;
  void        copy_combined_key_part(KeyLength offset, KeyLength len, char* buffer) const;
  std::string get_combined_key() const ALWAYS_INLINE {
    char buf[kMaxKeyLength];
    copy_combined_key(buf);
    return std::string(buf, cur_key_length_);
  }

  const char* get_payload() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    return cur_payload_;
  }
  PayloadLength  get_payload_length() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    return cur_payload_length_;
  }

  ErrorCode next();


  ErrorCode delete_record();

  ErrorCode overwrite_record(
    const void* payload,
    PayloadLength payload_offset,
    PayloadLength payload_count);
  template <typename PAYLOAD>
  ErrorCode overwrite_record_primitive(
    PAYLOAD payload,
    PayloadLength payload_offset);

  template <typename PAYLOAD>
  ErrorCode increment_record(PAYLOAD* value, PayloadLength payload_offset);

 private:
  Engine* const         engine_;
  MasstreeStorage       storage_;
  thread::Thread* const context_;
  xct::Xct* const current_xct_;

  bool        for_writes_;
  bool        forward_cursor_;
  bool        end_inclusive_;
  bool        reached_end_;

  KeyLength   end_key_length_;
  KeyLength   cur_key_length_;
  PayloadLength cur_payload_length_;
  KeyLength   search_key_length_;
  SearchType  search_type_;
  bool        search_key_in_layer_extremum_;

  uint16_t    route_count_;

  bool        should_skip_cur_route_;

  bool        cur_key_next_layer_;
  KeyLength   cur_key_in_layer_remainder_;
  KeySlice    cur_key_in_layer_slice_;
  RecordLocation cur_key_location_;
  // xct::XctId  cur_key_observed_owner_id_;
  xct::RwLockableXctId* cur_key_owner_id_address;

  char*       end_key_;
  KeySlice*   end_key_slices_;

  KeySlice*   cur_route_prefix_slices_;
  char*       cur_route_prefix_be_;
  const char* cur_key_suffix_;

  const char* cur_payload_;

  char*       search_key_;
  KeySlice*   search_key_slices_;

  PageVersionStatus cur_page_stable_;

  Route*      routes_;

  ErrorCode push_route(MasstreePage* page);
  ErrorCode fetch_cur_record_logical(MasstreeBorderPage* page, SlotIndex record);
  void      check_end_key();
  bool      is_cur_key_next_layer() const { return cur_key_location_.observed_.is_next_layer(); }
  KeyCompareResult compare_cur_key_aginst_search_key(KeySlice slice, uint8_t layer) const;
  KeyCompareResult compare_cur_key_aginst_end_key() const;
  KeyCompareResult compare_cur_key(
    KeySlice slice,
    uint8_t layer,
    const char* full_key,
    KeyLength full_length) const;

  SlotIndex     get_cur_index() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    return cur_route()->get_cur_original_index();
  }
  MasstreePage* get_cur_page() const ALWAYS_INLINE {
    ASSERT_ND(is_valid_record());
    return cur_route()->page_;
  }

  Route*        cur_route() ALWAYS_INLINE {
    ASSERT_ND(route_count_ > 0);
    return routes_ + route_count_ - 1;
  }
  const Route*  cur_route() const ALWAYS_INLINE {
    ASSERT_ND(route_count_ > 0);
    return routes_ + route_count_ - 1;
  }

  template <typename T>
  ErrorCode allocate_if_not_exist(T** pointer);

  bool is_search_key_extremum() const ALWAYS_INLINE {
    return search_key_length_ == kKeyLengthExtremum;
  }
  bool is_end_key_supremum() const ALWAYS_INLINE {
    return end_key_length_ == kKeyLengthExtremum;
  }

  ErrorCode follow_foster_border(KeySlice slice);
  void extract_separators(KeySlice* separator_low, KeySlice* separator_high) const ALWAYS_INLINE;


  ErrorCode locate_layer(uint8_t layer);
  ErrorCode locate_border(KeySlice slice);
  ErrorCode locate_next_layer();
  ErrorCode locate_descend(KeySlice slice);

  // proceed_xxx is for next
  ErrorCode proceed_route();
  ErrorCode proceed_route_border();
  ErrorCode proceed_route_intermediate();
  ErrorCode proceed_pop();
  ErrorCode proceed_next_layer();
  ErrorCode proceed_deeper();
  ErrorCode proceed_deeper_border();
  ErrorCode proceed_deeper_intermediate();
  void      proceed_route_intermediate_rebase_separator();

  void set_should_skip_cur_route();

  MasstreePage* resolve_volatile(VolatilePagePointer ptr) const;

  void assert_modify() const ALWAYS_INLINE {
#ifndef NDEBUG
    ASSERT_ND(!should_skip_cur_route_);
    ASSERT_ND(for_writes_);
    ASSERT_ND(is_valid_record());
    ASSERT_ND(!cur_route()->snapshot_);
    ASSERT_ND(reinterpret_cast<Page*>(get_cur_page())->get_header().get_page_type()
      == kMasstreeBorderPageType);
#endif  // NDEBUG
  }
  void assert_route() const ALWAYS_INLINE {
#ifndef NDEBUG
    assert_route_impl();
#endif  // NDEBUG
  }
  void assert_route_impl() const;
};

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