log_reducer_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_SNAPSHOT_LOG_REDUCER_IMPL_HPP_
#define FOEDUS_SNAPSHOT_LOG_REDUCER_IMPL_HPP_
#include <stdint.h>

#include <atomic>
#include <iosfwd>
#include <map>
#include <memory>
#include <string>
#include <vector>

#include "foedus/attachable.hpp"
#include "foedus/fwd.hpp"
#include "foedus/initializable.hpp"
#include "foedus/assorted/cacheline.hpp"
#include "foedus/assorted/raw_atomics.hpp"
#include "foedus/cache/snapshot_file_set.hpp"
#include "foedus/fs/fwd.hpp"
#include "foedus/fs/path.hpp"
#include "foedus/log/fwd.hpp"
#include "foedus/log/log_id.hpp"
#include "foedus/memory/aligned_memory.hpp"
#include "foedus/snapshot/fwd.hpp"
#include "foedus/snapshot/log_reducer_ref.hpp"
#include "foedus/snapshot/mapreduce_base_impl.hpp"
#include "foedus/snapshot/snapshot_id.hpp"
#include "foedus/snapshot/snapshot_writer_impl.hpp"
#include "foedus/soc/shared_cond.hpp"
#include "foedus/soc/shared_memory_repo.hpp"
#include "foedus/storage/fwd.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/thread/condition_variable_impl.hpp"
#include "foedus/thread/fwd.hpp"

namespace foedus {
namespace snapshot {

enum ReducerConstants {
  kFlagNoMoreWriters = 0x0001,
};

union ReducerBufferStatus {
  uint64_t word;
  struct Components {
    uint16_t        active_writers_;
    uint16_t        flags_;
    BufferPosition  tail_position_;
  } components;

  bool is_no_more_writers() const {
    return (components.flags_ & kFlagNoMoreWriters) != 0;
  }
  bool is_clear() const { return word == 0; }
  uint16_t get_active_writers() const { return components.active_writers_; }
  BufferPosition get_tail_position() const { return components.tail_position_; }
  uint64_t get_tail_bytes() const { return from_buffer_position(components.tail_position_); }
};

struct BlockHeaderBase {
  enum Constants {
    kFullBlockHeaderMagicWord = 0xDEADBEEF,
    kFillerBlockHeaderMagicWord = 0x8BADF00D,
  };
  bool is_full_block() const {
    ASSERT_ND(magic_word_ == kFullBlockHeaderMagicWord
      || magic_word_ == kFillerBlockHeaderMagicWord);
    return magic_word_ == kFullBlockHeaderMagicWord;
  }
  bool is_filler() const {
    ASSERT_ND(magic_word_ == kFullBlockHeaderMagicWord
      || magic_word_ == kFillerBlockHeaderMagicWord);
    return magic_word_ == kFillerBlockHeaderMagicWord;
  }

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

  uint32_t            magic_word_;
  BufferPosition      block_length_;
};

struct FullBlockHeader : BlockHeaderBase {
  storage::StorageId  storage_id_;
  uint32_t            log_count_;
  uint32_t            shortest_key_length_;
  uint32_t            longest_key_length_;
  inline void assert_key_length() const {
    ASSERT_ND(shortest_key_length_ > 0);
    ASSERT_ND(shortest_key_length_ <= longest_key_length_);
  }
};


struct FillerBlockHeader : public BlockHeaderBase {};

struct LogReducerControlBlock {
  // this is backed by shared memory. not instantiation. just reinterpret_cast.
  LogReducerControlBlock() = delete;
  ~LogReducerControlBlock() = delete;

  void initialize() {
    clear();
  }
  void clear() {
    current_buffer_ = 0;
    buffer_status_[0].store(0U);
    buffer_status_[1].store(0U);
    total_storage_count_ = 0;
  }
  void uninitialize() {
  }

  ReducerBufferStatus get_buffer_status_atomic(uint32_t index) const {
    ReducerBufferStatus ret;
    ret.word = buffer_status_[index % 2].load();
    return ret;
  }
  std::atomic<uint64_t>* get_buffer_status_address(uint32_t index) {
    return &buffer_status_[index % 2];
  }

  ReducerBufferStatus get_current_buffer_status() const {
    return get_buffer_status_atomic(current_buffer_);
  }
  ReducerBufferStatus get_non_current_buffer_status() const {
    return get_buffer_status_atomic(current_buffer_ + 1U);
  }

  std::atomic<uint64_t> buffer_status_[2];

  std::atomic<uint32_t> current_buffer_;

  std::atomic<uint32_t> total_storage_count_;

  uint16_t              id_;
};


class LogReducer final : public MapReduceBase {
 public:
  explicit LogReducer(Engine* engine);

  ErrorStack  initialize_once() override;
  ErrorStack  uninitialize_once() override;

  std::string to_string() const override { return std::string("Reducer-") + std::to_string(id_); }
  friend std::ostream&    operator<<(std::ostream& o, const LogReducer& v);

 protected:
  ErrorStack  handle_process() override;

 private:
  struct MergeContext {
    explicit MergeContext(uint32_t dumped_files_count_);
    ~MergeContext();

    const uint32_t                            dumped_files_count_;
    memory::AlignedMemory                     io_memory_;
    std::vector< memory::AlignedMemorySlice > io_buffers_;

    std::vector< std::unique_ptr<SortedBuffer> >  sorted_buffers_;

    std::vector< std::unique_ptr<fs::DirectIoFile> > sorted_files_auto_ptrs_;

    SortedBuffer**                          tmp_sorted_buffer_array_;
    uint32_t                                tmp_sorted_buffer_count_;

    storage::StorageId  get_min_storage_id() const;
    void                set_tmp_sorted_buffer_array(storage::StorageId storage_id);
  };

  LogReducerControlBlock* control_block_;

  void*                   buffers_[2];

  uint64_t                buffer_half_size_bytes_;

  storage::Page*          root_info_pages_;

  cache::SnapshotFileSet  previous_snapshot_files_;

  memory::AlignedMemory   dump_io_buffer_;

  memory::AlignedMemory   sort_buffer_;

  memory::AlignedMemory   positions_buffers_;

  memory::AlignedMemorySlice input_positions_slice_;
  memory::AlignedMemorySlice output_positions_slice_;

  memory::AlignedMemory   writer_pool_memory_;
  memory::AlignedMemory   writer_intermediate_memory_;

  uint32_t      sorted_runs_;

  void expand_if_needed(
    uint64_t required_size,
    memory::AlignedMemory *memory,
    const std::string& name);
  void expand_positions_buffers_if_needed(uint64_t required_size_per_buffer);

  fs::Path get_sorted_run_file_path(uint32_t sorted_run) const;

  ErrorStack dump_buffer();
  ErrorStack dump_buffer_wait_for_writers(uint32_t buffer_index) const;

  void dump_buffer_scan_block_headers(
    char* buffer_base,
    BufferPosition tail_position,
    std::map<storage::StorageId, std::vector<BufferPosition> > *blocks) const;

  ErrorStack dump_buffer_sort_storage(
    const LogBuffer &buffer,
    storage::StorageId storage_id,
    const std::vector<BufferPosition>& log_positions,
    uint32_t* out_shortest_key_length,
    uint32_t* out_longest_key_length,
    uint32_t* written_count);

  ErrorStack dump_buffer_sort_storage_write(
    const LogBuffer &buffer,
    storage::StorageId storage_id,
    const BufferPosition* sorted_logs,
    uint32_t shortest_key_length,
    uint32_t longest_key_length,
    uint32_t log_count,
    fs::DirectIoFile *dump_file);
  uint64_t dump_block_header(
    const LogBuffer &buffer,
    storage::StorageId storage_id,
    const BufferPosition* sorted_logs,
    uint32_t shortest_key_length,
    uint32_t longest_key_length,
    uint32_t log_count,
    void* destination) const;

  ErrorStack  merge_sort();

  void        merge_sort_check_buffer_status() const;
  ErrorStack  merge_sort_dump_last_buffer();

  void        merge_sort_allocate_io_buffers(MergeContext* context) const;
  ErrorStack  merge_sort_open_sorted_runs(MergeContext* context) const;
  ErrorStack  merge_sort_initialize_sort_buffers(MergeContext* context) const;
  ErrorCode   merge_sort_advance_sort_buffers(
    SortedBuffer* buffer,
    storage::StorageId processed_storage_id) const;

  uint32_t    get_max_storage_count() const;
};


static_assert(
  sizeof(LogReducerControlBlock) <= soc::NodeMemoryAnchors::kLogReducerMemorySize,
  "LogReducerControlBlock is too large.");

}  // namespace snapshot
}  // namespace foedus
#endif  // FOEDUS_SNAPSHOT_LOG_REDUCER_IMPL_HPP_