hash_page_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_PAGE_HPP_
#define FOEDUS_STORAGE_HASH_HASH_PAGE_HPP_

#include <stdint.h>

#include <cstring>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/fwd.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/storage/page.hpp"
#include "foedus/storage/record.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/storage/hash/hash_combo.hpp"
#include "foedus/storage/hash/hash_hashinate.hpp"
#include "foedus/storage/hash/hash_id.hpp"
#include "foedus/xct/xct_id.hpp"

namespace foedus {
namespace storage {
namespace hash {

class HashIntermediatePage final {
 public:
  HashIntermediatePage() = delete;
  HashIntermediatePage(const HashIntermediatePage& other) = delete;
  HashIntermediatePage& operator=(const HashIntermediatePage& other) = delete;

  PageHeader&             header() { return header_; }
  const PageHeader&       header() const { return header_; }
  VolatilePagePointer     get_volatile_page_id() const {
    ASSERT_ND(!header_.snapshot_);
    return VolatilePagePointer(header_.page_id_);
  }
  SnapshotPagePointer     get_snapshot_page_id() const {
    ASSERT_ND(header_.snapshot_);
    return static_cast<SnapshotPagePointer>(header_.page_id_);
  }
  DualPagePointer&        get_pointer(uint16_t index) { return pointers_[index]; }
  const DualPagePointer&  get_pointer(uint16_t index) const { return pointers_[index]; }
  DualPagePointer*        get_pointer_address(uint16_t index) { return pointers_ + index; }
  const DualPagePointer*  get_pointer_address(uint16_t index) const { return pointers_ + index; }

  void                    initialize_volatile_page(
    StorageId storage_id,
    VolatilePagePointer page_id,
    const HashIntermediatePage* parent,
    uint8_t level,
    HashBin start_bin);

  void                    initialize_snapshot_page(
    StorageId storage_id,
    SnapshotPagePointer page_id,
    uint8_t level,
    HashBin start_bin);

  void release_pages_recursive_parallel(Engine* engine);
  void release_pages_recursive(
    const memory::GlobalVolatilePageResolver& page_resolver,
    memory::PageReleaseBatch* batch);

  const HashBinRange&   get_bin_range() const { return bin_range_; }
  inline uint8_t   get_level() const { return header_.get_in_layer_level(); }

  inline void      assert_bin(HashBin bin) const ALWAYS_INLINE {
    ASSERT_ND(bin_range_.contains(bin));
  }
  inline void      assert_range() const ALWAYS_INLINE {
    ASSERT_ND(bin_range_.length() == kHashMaxBins[get_level() + 1U]);
  }

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

  uint64_t get_dummy_padding_unused() const { return padding_; }

 private:
  PageHeader          header_;        // +40 -> 40

  uint64_t            padding_;       // +8 -> 48

  HashBinRange        bin_range_;     // +16 -> 64

  DualPagePointer     pointers_[kHashIntermediatePageFanout];
};

class HashDataPage final {
 public:
  friend struct ReserveRecords;
  struct Slot {
    xct::RwLockableXctId tid_;            // +16 -> 16

    uint16_t offset_;                   // +2  -> 18
    uint16_t physical_record_length_;   // +2  -> 20
    uint16_t key_length_;               // +2  -> 22
    uint16_t payload_length_;           // +2  -> 24

    HashValue hash_;                    // +8  -> 32

    inline uint16_t get_aligned_key_length() const { return assorted::align8(key_length_); }
    inline uint16_t get_max_payload() const {
      return physical_record_length_ - get_aligned_key_length();
    }
  };

  // A page object is never explicitly instantiated. You must reinterpret_cast.
  HashDataPage() = delete;
  HashDataPage(const HashDataPage& other) = delete;
  HashDataPage& operator=(const HashDataPage& other) = delete;

  void initialize_volatile_page(
    StorageId storage_id,
    VolatilePagePointer page_id,
    const Page* parent,
    HashBin bin,
    uint8_t bin_shifts);
  void initialize_snapshot_page(
    StorageId storage_id,
    SnapshotPagePointer page_id,
    HashBin bin,
    uint8_t bin_bits,
    uint8_t bin_shifts);
  void release_pages_recursive(
    const memory::GlobalVolatilePageResolver& page_resolver,
    memory::PageReleaseBatch* batch);


  // simple accessors
  PageHeader&               header() { return header_; }
  const PageHeader&         header() const { return header_; }
  bool                      is_locked() const { return header_.page_version_.is_locked(); }

  VolatilePagePointer       get_volatile_page_id() const {
    ASSERT_ND(!header_.snapshot_);
    return VolatilePagePointer(header_.page_id_);
  }
  SnapshotPagePointer       get_snapshot_page_id() const {
    ASSERT_ND(header_.snapshot_);
    return static_cast<SnapshotPagePointer>(header_.page_id_);
  }

  inline const DualPagePointer&  next_page() const ALWAYS_INLINE { return next_page_; }
  inline DualPagePointer&   next_page() ALWAYS_INLINE { return next_page_; }
  inline const DualPagePointer*  next_page_address() const ALWAYS_INLINE { return &next_page_; }
  inline DualPagePointer*   next_page_address() ALWAYS_INLINE { return &next_page_; }

  inline const DataPageBloomFilter& bloom_filter() const ALWAYS_INLINE { return bloom_filter_; }
  inline DataPageBloomFilter& bloom_filter() ALWAYS_INLINE { return bloom_filter_; }

  inline uint16_t           get_record_count() const ALWAYS_INLINE { return header_.key_count_; }

  inline const Slot&        get_slot(DataPageSlotIndex record) const ALWAYS_INLINE {
    return reinterpret_cast<const Slot*>(this + 1)[-record - 1];
  }
  inline Slot&              get_slot(DataPageSlotIndex record) ALWAYS_INLINE {
    return reinterpret_cast<Slot*>(this + 1)[-record - 1];
  }
  inline Slot&              get_new_slot(DataPageSlotIndex record) ALWAYS_INLINE {
    // Only in this method, record is okay to be == key_count
    return reinterpret_cast<Slot*>(this + 1)[-record - 1];
  }
  inline const Slot*        get_slot_address(DataPageSlotIndex record) const ALWAYS_INLINE {
    return reinterpret_cast<const Slot*>(this + 1) - record - 1;
  }
  inline Slot*              get_slot_address(DataPageSlotIndex record) ALWAYS_INLINE {
    return reinterpret_cast<Slot*>(this + 1) - record - 1;
  }

  inline DataPageSlotIndex  to_slot_index(const Slot* slot) const ALWAYS_INLINE {
    ASSERT_ND(this);
    ASSERT_ND(slot);
    int64_t index = reinterpret_cast<const Slot*>(this + 1) - slot - 1;
    ASSERT_ND(index >= 0);
    ASSERT_ND(index < static_cast<int64_t>(sizeof(data_) / sizeof(Slot)));
    return static_cast<DataPageSlotIndex>(index);
  }

  inline char*        record_from_offset(uint16_t offset) { return data_ + offset; }
  inline const char*  record_from_offset(uint16_t offset) const { return data_ + offset; }

  // methods about available/required spaces

  inline uint16_t     available_space() const {
    uint16_t consumed = next_offset() + get_record_count() * sizeof(Slot);
    ASSERT_ND(consumed <= sizeof(data_));
    if (consumed > sizeof(data_)) {  // just to be conservative on release build
      return 0;
    }
    return sizeof(data_) - consumed;
  }

  inline uint16_t     next_offset() const {
    DataPageSlotIndex count = get_record_count();
    if (count == 0) {
      return 0;
    }
    const Slot& last_slot = get_slot(count - 1);
    ASSERT_ND((last_slot.offset_ + last_slot.physical_record_length_) % 8 == 0);
    return last_slot.offset_ + last_slot.physical_record_length_;
  }

  inline static uint16_t required_space(uint16_t key_length, uint16_t payload_length) {
    return assorted::align8(key_length) + assorted::align8(payload_length) + sizeof(Slot);
  }


  // search, insert, migrate, etc to manipulate records in this page

  DataPageSlotIndex reserve_record(
    HashValue hash,
    const BloomFilterFingerprint& fingerprint,
    const void* key,
    uint16_t key_length,
    uint16_t payload_length);

  void create_record_in_snapshot(
    xct::XctId xct_id,
    HashValue hash,
    const BloomFilterFingerprint& fingerprint,
    const void* key,
    uint16_t key_length,
    const void* payload,
    uint16_t payload_length) ALWAYS_INLINE;

  DataPageSlotIndex search_key_physical(
    HashValue hash,
    const BloomFilterFingerprint& fingerprint,
    const void* key,
    KeyLength key_length,
    DataPageSlotIndex record_count,
    DataPageSlotIndex check_from = 0) const;

  inline bool compare_slot_key(
    DataPageSlotIndex index,
    HashValue hash,
    const void* key,
    uint16_t key_length) const {
    ASSERT_ND(index < get_record_count());  // record count purely increasing
    const Slot& slot = get_slot(index);
    ASSERT_ND(hashinate(record_from_offset(slot.offset_), slot.key_length_) == slot.hash_);
    // quick check first
    if (slot.hash_ != hash || slot.key_length_ != key_length) {
      return false;
    }
    const void* data = record_from_offset(slot.offset_);
    return std::memcmp(data, key, key_length) == 0;
  }

  HashBin     get_bin() const { return bin_; }
  inline void assert_bin(HashBin bin) const ALWAYS_INLINE { ASSERT_ND(bin_ == bin); }

  void protected_set_bin_shifts(uint8_t bin_shifts) { header_.masstree_layer_ = bin_shifts; }
  inline uint8_t get_bin_shifts() const { return header_.masstree_layer_; }

  void        assert_entries() const ALWAYS_INLINE {
#ifndef NDEBUG
    assert_entries_impl();
#endif  // NDEBUG
  }
  void        assert_entries_impl() const;
  friend std::ostream& operator<<(std::ostream& o, const HashDataPage& v);

 private:
  PageHeader      header_;        // +40 -> 40

  HashBin         bin_;           // +8 -> 48

  DualPagePointer next_page_;     // +16 -> 64

  DataPageBloomFilter bloom_filter_;  // +64 -> 128

  char            data_[kHashDataPageDataSize];
};


void hash_intermediate_volatile_page_init(const VolatilePageInitArguments& args);

void hash_data_volatile_page_init(const VolatilePageInitArguments& args);

inline void HashDataPage::create_record_in_snapshot(
  xct::XctId xct_id,
  HashValue hash,
  const BloomFilterFingerprint& fingerprint,
  const void* key_arg,
  uint16_t key_length,
  const void* payload_arg,
  uint16_t payload_length) {
  ASSERT_ND(header_.snapshot_);
  ASSERT_ND(available_space() >= required_space(key_length, payload_length));
  ASSERT_ND(reinterpret_cast<uintptr_t>(this) % kPageSize == 0);
  ASSERT_ND(reinterpret_cast<uintptr_t>(key_arg) % 8 == 0);
  ASSERT_ND(reinterpret_cast<uintptr_t>(payload_arg) % 8 == 0);

  const void* key = ASSUME_ALIGNED(key_arg, 8U);
  const void* payload = ASSUME_ALIGNED(payload_arg, 8U);
  DataPageSlotIndex index = get_record_count();
  Slot& slot = get_slot(index);
  slot.tid_.xct_id_ = xct_id;
  slot.offset_ = next_offset();
  slot.hash_ = hash;
  slot.key_length_ = key_length;
  uint16_t aligned_key_length = assorted::align8(key_length);
  slot.physical_record_length_ = aligned_key_length + assorted::align8(payload_length);
  slot.payload_length_ = payload_length;

  ASSERT_ND(reinterpret_cast<uintptr_t>(record_from_offset(slot.offset_)) % 8 == 0);
  char* record = reinterpret_cast<char*>(ASSUME_ALIGNED(record_from_offset(slot.offset_), 8U));

  ASSERT_ND(key_length > 0);
  std::memcpy(record, key, key_length);
  if (key_length != aligned_key_length) {
    std::memset(record + key_length, 0, aligned_key_length - key_length % 8);
  }

  if (payload_length > 0) {
    char* dest = reinterpret_cast<char*>(ASSUME_ALIGNED(record + aligned_key_length, 8U));
    uint16_t aligned_payload_length = assorted::align8(payload_length);
    std::memcpy(dest, payload, payload_length);
    if (key_length != aligned_payload_length) {
      std::memset(dest + payload_length, 0, aligned_payload_length - payload_length);
    }
  }

  bloom_filter_.add(fingerprint);

  header_.increment_key_count();
}


static_assert(
  sizeof(HashIntermediatePage) == kPageSize,
  "sizeof(HashIntermediatePage) is not kPageSize");

static_assert(
  sizeof(HashDataPage) == kPageSize,
  "sizeof(HashDataPage) is not kPageSize");

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