merge_sort.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_SNAPSHOT_MERGE_SORT_HPP_
#define FOEDUS_SNAPSHOT_MERGE_SORT_HPP_

#include <algorithm>

#include "foedus/assert_nd.hpp"
#include "foedus/compiler.hpp"
#include "foedus/epoch.hpp"
#include "foedus/error_stack.hpp"
#include "foedus/initializable.hpp"
#include "foedus/assorted/cacheline.hpp"
#include "foedus/log/common_log_types.hpp"
#include "foedus/memory/fwd.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/snapshot/snapshot_id.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/storage/array/array_log_types.hpp"
#include "foedus/storage/masstree/masstree_log_types.hpp"

namespace foedus {
namespace snapshot {
class MergeSort CXX11_FINAL : public DefaultInitializable {
 public:
  typedef uint32_t MergedPosition;
  typedef uint16_t InputIndex;

  enum Constants {
    kMaxMergedPosition = 1 << 23,
    kLogChunk = 1 << 10,
    kDefaultChunkBatch = 1 << 8,
    kMaxLevels = 32,
    kWindowChunkReserveBytes = 1 << 16,
  };

  struct SortEntry {
    inline void set(
      uint64_t  key,
      uint16_t  compressed_epoch,
      uint32_t  in_epoch_ordinal,
      bool      needs_additional_check,
      MergedPosition position) ALWAYS_INLINE {
      ASSERT_ND(in_epoch_ordinal < (1U << 24));
      ASSERT_ND(position < kMaxMergedPosition);
      data_
        = static_cast<__uint128_t>(key) << 64
          | static_cast<__uint128_t>(compressed_epoch) << 48
          | static_cast<__uint128_t>(in_epoch_ordinal) << 24
          | static_cast<__uint128_t>(needs_additional_check ? 1U << 23 : 0)
          | static_cast<__uint128_t>(position);
    }
    inline uint64_t get_key() const ALWAYS_INLINE {
      return static_cast<uint64_t>(data_ >> 64);
    }
    inline bool needs_additional_check() const ALWAYS_INLINE {
      return data_ & (1U << 23);
    }
    inline MergedPosition get_position() const ALWAYS_INLINE {
      return static_cast<MergedPosition>(data_) & 0x7FFFFFU;
    }

    __uint128_t data_;
  };

  struct PositionEntry {
    uint16_t        input_index_;     // +2 -> 2
    uint16_t        log_type_;        // +2 -> 4
    BufferPosition  input_position_;  // +4 -> 8

    inline log::LogCode get_log_type() const ALWAYS_INLINE {
      return static_cast<log::LogCode>(log_type_);
    }
  };

  struct InputStatus CXX11_FINAL {
    const char*     window_;
    uint64_t        cur_relative_pos_;
    uint64_t        chunk_relative_pos_;
    char            padding_[6];
    uint64_t        previous_chunk_relative_pos_;
    uint64_t        window_offset_;
    uint64_t        window_size_;
    uint64_t        end_absolute_pos_;

    inline void assert_consistent() const ALWAYS_INLINE {
#ifndef NDEBUG
      // be careful on infinite loop! don't call any function here
      ASSERT_ND(window_);
      ASSERT_ND(cur_relative_pos_ <= window_size_);
      ASSERT_ND(chunk_relative_pos_ <= window_size_);
      ASSERT_ND(previous_chunk_relative_pos_ <= window_size_);
      ASSERT_ND(window_offset_ <= end_absolute_pos_);
      ASSERT_ND(cur_relative_pos_ + window_offset_ <= end_absolute_pos_);
      ASSERT_ND(chunk_relative_pos_ + window_offset_ <= end_absolute_pos_);
      ASSERT_ND(previous_chunk_relative_pos_ + window_offset_ <= end_absolute_pos_);
      ASSERT_ND(cur_relative_pos_ <= chunk_relative_pos_);
      ASSERT_ND(previous_chunk_relative_pos_ <= chunk_relative_pos_);
      if (window_offset_ + cur_relative_pos_ != end_absolute_pos_) {
        const log::RecordLogType* l
          = reinterpret_cast<const log::RecordLogType*>(window_ + chunk_relative_pos_);
        ASSERT_ND(chunk_relative_pos_ + l->header_.log_length_ <= window_size_);
      }
#endif  // NDEBUG
    }
    inline const log::RecordLogType* get_cur_log() const ALWAYS_INLINE {
      assert_consistent();
      return reinterpret_cast<const log::RecordLogType*>(window_ + cur_relative_pos_);
    }
    inline const log::RecordLogType* get_chunk_log() const ALWAYS_INLINE {
      assert_consistent();
      return reinterpret_cast<const log::RecordLogType*>(window_ + chunk_relative_pos_);
    }
    inline const log::RecordLogType* from_byte_pos(uint64_t pos) const ALWAYS_INLINE {
      assert_consistent();
      ASSERT_ND(pos < window_size_);
      return reinterpret_cast<const log::RecordLogType*>(window_ + pos);
    }
    inline const log::RecordLogType* from_compact_pos(BufferPosition pos) const ALWAYS_INLINE {
      assert_consistent();
      ASSERT_ND(pos * 8ULL < window_size_);
      return reinterpret_cast<const log::RecordLogType*>(window_ + pos * 8ULL);
    }
    inline uint64_t to_absolute_pos(uint64_t relative_pos) const ALWAYS_INLINE {
      assert_consistent();
      return window_offset_ + relative_pos;
    }
    inline bool is_last_window() const ALWAYS_INLINE {
      assert_consistent();
      return to_absolute_pos(window_size_) >= end_absolute_pos_;
    }
    inline bool is_ended() const ALWAYS_INLINE {
      assert_consistent();
      return to_absolute_pos(cur_relative_pos_) == end_absolute_pos_;
    }
    inline bool is_last_chunk_in_window() const ALWAYS_INLINE {
      if (is_last_window()) {
        // in last window, we accurately determines the last chunk
        uint64_t chunk_abs_pos = to_absolute_pos(chunk_relative_pos_);
        ASSERT_ND(chunk_abs_pos <= end_absolute_pos_);
        if (chunk_abs_pos >= end_absolute_pos_) {
          return true;
        }
        uint16_t length = get_chunk_log()->header_.log_length_;
        ASSERT_ND(length > 0);
        ASSERT_ND(chunk_abs_pos + length <= end_absolute_pos_);
        return chunk_abs_pos + length >= end_absolute_pos_;
      } else {
        // in this case, we conservatively determines the last chunk
        return chunk_relative_pos_ + kWindowChunkReserveBytes >= window_size_;
      }
    }
    inline bool is_last_chunk_overall() const ALWAYS_INLINE {
      return is_last_window() && is_last_chunk_in_window();
    }
  };

  struct AdjustComparatorMasstree CXX11_FINAL {
    AdjustComparatorMasstree(PositionEntry* position_entries, InputStatus* inputs_status)
      : position_entries_(position_entries), inputs_status_(inputs_status) {}
    inline bool operator() (const SortEntry& left, const SortEntry& right) const ALWAYS_INLINE {
      ASSERT_ND(left.get_key() <= right.get_key());
      if (left.get_key() < right.get_key()) {
        return true;
      }
      MergedPosition left_index = left.get_position();
      MergedPosition right_index = right.get_position();
      const PositionEntry& left_pos = position_entries_[left_index];
      const PositionEntry& right_pos = position_entries_[right_index];
      const storage::masstree::MasstreeCommonLogType* left_casted
        = reinterpret_cast<const storage::masstree::MasstreeCommonLogType*>(
          inputs_status_[left_pos.input_index_].from_compact_pos(left_pos.input_position_));
      const storage::masstree::MasstreeCommonLogType* right_casted
        = reinterpret_cast<const storage::masstree::MasstreeCommonLogType*>(
          inputs_status_[right_pos.input_index_].from_compact_pos(right_pos.input_position_));
      int cmp = storage::masstree::MasstreeCommonLogType::compare_logs(left_casted, right_casted);
      return cmp < 0 || (cmp == 0 && left_index < right_index);
    }
    PositionEntry* position_entries_;
    InputStatus* inputs_status_;
  };

  struct GroupifyResult {
    uint32_t count_;            // +4 -> 4
    bool has_common_key_;       // +1 -> 5
    bool has_common_log_code_;  // +1 -> 6
    uint16_t log_code_;         // +2 -> 8
    inline log::LogCode get_log_type() const ALWAYS_INLINE {
      return static_cast<log::LogCode>(log_code_);
    }
  };

  MergeSort(
    storage::StorageId id,
    storage::StorageType type,
    Epoch base_epoch,
    SortedBuffer* const* inputs,
    uint16_t inputs_count,
    uint16_t max_original_pages,
    memory::AlignedMemory* const work_memory,
    uint16_t chunk_batch_size = kDefaultChunkBatch);

  ErrorStack next_batch();

  uint32_t fetch_logs(uint32_t sort_pos, uint32_t count, log::RecordLogType const** out) const;

  GroupifyResult groupify(uint32_t begin, uint32_t limit = 0) const ALWAYS_INLINE;

  void assert_sorted();

  ErrorStack initialize_once() CXX11_OVERRIDE;
  ErrorStack uninitialize_once() CXX11_OVERRIDE { return kRetOk; }

  inline bool is_no_merging() const ALWAYS_INLINE { return inputs_count_ == 1U; }

  inline bool is_ended_all() const {
    for (InputIndex i = 0; i < inputs_count_; ++i) {
      if (!inputs_status_[i].is_ended()) {
        return false;
      }
    }
    return true;
  }

  inline storage::StorageId get_storage_id() const ALWAYS_INLINE { return id_; }
  inline Epoch          get_base_epoch() const ALWAYS_INLINE { return base_epoch_; }
  inline MergedPosition get_current_count() const ALWAYS_INLINE { return current_count_; }
  inline InputIndex     get_inputs_count() const ALWAYS_INLINE { return inputs_count_; }
  inline const PositionEntry* get_position_entries() const ALWAYS_INLINE {
    return position_entries_;
  }
  inline log::LogCode get_log_type_from_sort_position(uint32_t sort_pos) const ALWAYS_INLINE {
    MergedPosition pos = sort_entries_[sort_pos].get_position();
    return position_entries_[pos].get_log_type();
  }
  inline const SortEntry* get_sort_entries() const ALWAYS_INLINE { return sort_entries_; }
  inline const log::RecordLogType* resolve_merged_position(MergedPosition pos) const ALWAYS_INLINE {
    ASSERT_ND(pos < current_count_);
    const PositionEntry& entry = position_entries_[pos];
    return inputs_status_[entry.input_index_].from_compact_pos(entry.input_position_);
  }
  inline const log::RecordLogType* resolve_sort_position(uint32_t sort_pos) const ALWAYS_INLINE {
    ASSERT_ND(sort_pos < current_count_);
    MergedPosition pos = sort_entries_[sort_pos].get_position();
    return resolve_merged_position(pos);
  }
  void change_log_type_at(uint32_t sort_pos, log::LogCode before, log::LogCode after) {
    // we have to change two places; PositionEntry and header of the log itself.
    ASSERT_ND(sort_pos < current_count_);
    ASSERT_ND(get_log_type_from_sort_position(sort_pos) == before);
    MergedPosition pos = sort_entries_[sort_pos].get_position();
    PositionEntry& entry = position_entries_[pos];
    ASSERT_ND(entry.get_log_type() == before);
    entry.log_type_ = after;
    ASSERT_ND(entry.get_log_type() == after);
    ASSERT_ND(get_log_type_from_sort_position(sort_pos) == after);

    log::RecordLogType* record = const_cast<log::RecordLogType*>(resolve_merged_position(pos));
    ASSERT_ND(record->header_.get_type() == before);
    record->header_.log_type_code_ = after;
    ASSERT_ND(record->header_.get_type() == after);
  }

  inline storage::Page* get_original_pages() const ALWAYS_INLINE { return original_pages_; }
  inline bool are_all_single_layer_logs() const ALWAYS_INLINE { return longest_key_length_ <= 8U; }
  inline uint16_t get_longest_key_length() const ALWAYS_INLINE { return longest_key_length_; }

 private:
  const storage::StorageId      id_;
  const storage::StorageType    type_;
  const Epoch                   base_epoch_;
  const uint16_t                shortest_key_length_;
  const uint16_t                longest_key_length_;
  SortedBuffer* const*          inputs_;
  const InputIndex              inputs_count_;
  const uint16_t                max_original_pages_;
  const uint16_t                chunk_batch_size_;
  memory::AlignedMemory* const  work_memory_;

  MergedPosition                current_count_;
  uint32_t                      buffer_capacity_;

  // followings are backed by work_memory_, and allocatd at initialize_once.
  SortEntry*                    sort_entries_;
  PositionEntry*                position_entries_;
  storage::Page*                original_pages_;
  InputStatus*                  inputs_status_;

  void next_batch_one_input();

  ErrorStack advance_window();

  void next_chunk(InputIndex input_index);

  InputIndex determine_min_input() const;

  InputIndex pick_chunks();

  void batch_sort(InputIndex min_input);
  void batch_sort_prepare(InputIndex min_input);
  void batch_sort_adjust_sort();

  int compare_logs(const log::RecordLogType* lhs, const log::RecordLogType* rhs) const;

  void append_logs(InputIndex input_index, uint64_t upto_relative_pos);

  uint16_t populate_entry_array(InputIndex input_index, uint64_t relative_pos) ALWAYS_INLINE;
  uint16_t populate_entry_hash(InputIndex input_index, uint64_t relative_pos) ALWAYS_INLINE;
  uint16_t populate_entry_masstree(InputIndex input_index, uint64_t relative_pos) ALWAYS_INLINE;

  // these methods are called very frequently (potentially for each log). worth explicit inlining.
  // all of them return the position upto (inclusive) which the logs have the common feature.
  uint32_t groupify_find_common_keys_8b(uint32_t begin, uint32_t limit) const ALWAYS_INLINE;
  uint32_t groupify_find_common_keys_general(uint32_t begin, uint32_t limit) const ALWAYS_INLINE;
  uint32_t groupify_find_common_log_type(uint32_t begin, uint32_t limit) const ALWAYS_INLINE;
};


inline bool is_array_log_type(uint16_t log_type) {
  return log_type == log::kLogCodeArrayOverwrite || log_type == log::kLogCodeArrayIncrement;
}
inline bool is_hash_log_type(uint16_t log_type) {
  return
    log_type == log::kLogCodeHashOverwrite
    || log_type == log::kLogCodeHashInsert
    || log_type == log::kLogCodeHashDelete
    || log_type == log::kLogCodeHashUpdate;
}
inline bool is_masstree_log_type(uint16_t log_type) {
  return
    log_type == log::kLogCodeMasstreeInsert
    || log_type == log::kLogCodeMasstreeDelete
    || log_type == log::kLogCodeMasstreeUpdate
    || log_type == log::kLogCodeMasstreeOverwrite;
}

inline MergeSort::GroupifyResult MergeSort::groupify(uint32_t begin, uint32_t limit) const {
  ASSERT_ND(begin < current_count_);
  GroupifyResult result;
  result.count_ = 1;
  result.has_common_key_ = false;
  result.has_common_log_code_ = false;
  result.log_code_ = log::kLogCodeInvalid;
  if (UNLIKELY(begin + 1U == current_count_)) {
    return result;
  }

  // first, check common keys
  uint32_t cur;
  if (shortest_key_length_ == 8U && longest_key_length_ == 8U) {
    cur = groupify_find_common_keys_8b(begin, limit);
  } else {
    cur = groupify_find_common_keys_general(begin, limit);
  }

  ASSERT_ND(cur >= begin);
  ASSERT_ND(cur < current_count_);
  if (UNLIKELY(cur != begin)) {
    result.has_common_key_ = true;
    result.count_ = cur - begin + 1U;
    return result;
  }

  // if keys are different, check common log types.
  cur = groupify_find_common_log_type(begin, limit);
  ASSERT_ND(cur >= begin);
  ASSERT_ND(cur < current_count_);
  if (UNLIKELY(cur != begin)) {
    result.has_common_log_code_ = true;
    result.log_code_ = position_entries_[sort_entries_[cur].get_position()].log_type_;
    result.count_ = cur - begin + 1U;
    return result;
  } else {
    return result;
  }
}

inline uint32_t MergeSort::groupify_find_common_keys_8b(uint32_t begin, uint32_t limit) const {
  ASSERT_ND(shortest_key_length_ == 8U && longest_key_length_ == 8U);
  ASSERT_ND(begin + 1U < current_count_);
  const uint32_t end = limit ? std::min<uint32_t>(current_count_, begin + limit) : current_count_;
  uint32_t cur = begin;
  // 8 byte key is enough to determine equality.
  while (sort_entries_[cur].get_key() == sort_entries_[cur + 1U].get_key()
      && LIKELY(cur + 1U < end)) {
    ++cur;
  }
  return cur;
}

inline uint32_t MergeSort::groupify_find_common_keys_general(uint32_t begin, uint32_t limit) const {
  ASSERT_ND(type_ != storage::kArrayStorage);
  ASSERT_ND(begin + 1U < current_count_);
  const uint32_t end = limit ? std::min<uint32_t>(current_count_, begin + limit) : current_count_;
  uint32_t cur = begin;
  // we might need more. a bit slower.
  while (sort_entries_[cur].get_key() == sort_entries_[cur + 1U].get_key()
      && LIKELY(cur + 1U < end)) {
    if (sort_entries_[cur].needs_additional_check()
      || sort_entries_[cur + 1U].needs_additional_check()) {
      MergedPosition cur_pos = sort_entries_[cur].get_position();
      MergedPosition next_pos = sort_entries_[cur + 1U].get_position();
      ASSERT_ND(is_masstree_log_type(position_entries_[cur_pos].get_log_type()));
      ASSERT_ND(is_masstree_log_type(position_entries_[next_pos].get_log_type()));
      const storage::masstree::MasstreeCommonLogType* cur_log
        = reinterpret_cast<const storage::masstree::MasstreeCommonLogType*>(
            resolve_merged_position(cur_pos));
      const storage::masstree::MasstreeCommonLogType* next_log
        = reinterpret_cast<const storage::masstree::MasstreeCommonLogType*>(
            resolve_merged_position(next_pos));
      uint16_t key_len = cur_log->key_length_;
      ASSERT_ND(key_len != 8U || next_log->key_length_ != 8U);
      if (key_len != next_log->key_length_) {
        break;
      } else if (key_len > 8U) {
        if (std::memcmp(cur_log->get_key() + 8, next_log->get_key() + 8, key_len - 8U) != 0) {
          break;
        }
      }
    }
    ++cur;
  }
  return cur;
}

inline uint32_t MergeSort::groupify_find_common_log_type(uint32_t begin, uint32_t limit) const {
  ASSERT_ND(begin + 1U < current_count_);

  // First, let's avoid (relatively) expensive checks if there is no common log type,
  // and assume the worst case; we are calling this method for each log.
  uint32_t cur = begin;
  const uint32_t end = limit ? std::min<uint32_t>(current_count_, begin + limit) : current_count_;
  PositionEntry cur_pos = position_entries_[sort_entries_[cur].get_position()];
  PositionEntry next_pos = position_entries_[sort_entries_[cur + 1U].get_position()];
  if (LIKELY(cur_pos.log_type_ != next_pos.log_type_)) {
    return cur;
  }

  // the LIKELY above will be false if there is a group, but then the cost of branch misdetection
  // is amortized by the (hopefully) large number of logs processed together below.

  // okay, from now on, there likely is a number of logs with same log type.
  // this method has to read potentially a large number of position entries, which might be
  // not contiguous. thus, let's do parallel prefetching.
  const uint16_t kFetchSize = 8;
  cur = begin + 1U;
  while (LIKELY(cur + kFetchSize < end)) {
    for (uint16_t i = 0; i < kFetchSize; ++i) {
      assorted::prefetch_cacheline(position_entries_ + sort_entries_[cur + i].get_position());
    }
    for (uint16_t i = 0; i < kFetchSize; ++i) {
      PositionEntry cur_pos = position_entries_[sort_entries_[cur + i].get_position()];
      PositionEntry next_pos = position_entries_[sort_entries_[cur + i + 1U].get_position()];
      // now that we assume there is a large group, this is UNLIKELY.
      if (UNLIKELY(cur_pos.log_type_ != next_pos.log_type_)) {
        return cur + i;
      }
    }
    cur += kFetchSize;
  }

  // last bits. no optimization needed.
  while (cur + 1U < end) {
    PositionEntry cur_pos = position_entries_[sort_entries_[cur].get_position()];
    PositionEntry next_pos = position_entries_[sort_entries_[cur + 1U].get_position()];
    if (cur_pos.log_type_ != next_pos.log_type_) {
      return cur;
    }
    ++cur;
  }
  return cur;
}

}  // namespace snapshot
}  // namespace foedus
#endif  // FOEDUS_SNAPSHOT_MERGE_SORT_HPP_