log_mapper_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_MAPPER_IMPL_HPP_
#define FOEDUS_SNAPSHOT_LOG_MAPPER_IMPL_HPP_
#include <stdint.h>

#include <iosfwd>
#include <string>

#include "foedus/compiler.hpp"
#include "foedus/fwd.hpp"
#include "foedus/initializable.hpp"
#include "foedus/fs/fwd.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/mapreduce_base_impl.hpp"
#include "foedus/storage/storage_id.hpp"
#include "foedus/thread/fwd.hpp"
#include "foedus/thread/stoppable_thread_impl.hpp"
#include "foedus/thread/thread_id.hpp"

namespace foedus {
namespace snapshot {
class LogMapper final : public MapReduceBase {
 public:
  LogMapper(Engine* engine, uint16_t local_ordinal);

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

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

 protected:
  ErrorStack  handle_process() override;

 private:
  enum Constsants {
    kBucketSize = 1 << 16,
    kBucketMaxCount = (kBucketSize - 16) / 4,
    kBucketHashListMaxCount = 1 << 12,
    kSendBufferSize = 1 << 20,
  };

  struct Bucket final {
    inline bool is_full() const ALWAYS_INLINE { return counts_ >= kBucketMaxCount; }

    storage::StorageId  storage_id_;   // +4 => 4
    uint32_t            counts_;       // +4 => 8
    Bucket*             next_bucket_;  // +8 => 16
    BufferPosition      log_positions_[kBucketMaxCount];  // + 4 * kBucketMaxCount => kBucketSize
  };
  STATIC_SIZE_CHECK(sizeof(Bucket), kBucketSize)

  
  struct BucketHashList {
    storage::StorageId  storage_id_;  // +4 => 4
    uint16_t            bucket_counts_;  // +2 => 6
    uint16_t            dummy_;       // +2 => 8
    Bucket*             head_;        // +8 => 16
    Bucket*             tail_;        // +8 => 24
    BucketHashList*     hashlist_next_;  // +8 => 32
  };

  struct PartitionSortEntry {
    inline void set(storage::PartitionId partition, BufferPosition position) ALWAYS_INLINE {
      partition_ = partition;
      position_ = position;
    }

    inline bool operator<(const PartitionSortEntry& other) const ALWAYS_INLINE {
      return partition_ < other.partition_;
    }
    inline bool operator<=(const PartitionSortEntry& other) const ALWAYS_INLINE {
      return partition_ <= other.partition_;
    }
    inline bool operator==(const PartitionSortEntry& other) const ALWAYS_INLINE {
      return partition_ == other.partition_;
    }
    inline bool operator!=(const PartitionSortEntry& other) const ALWAYS_INLINE {
      return partition_ != other.partition_;
    }
    inline bool operator>(const PartitionSortEntry& other) const ALWAYS_INLINE {
      return partition_ > other.partition_;
    }
    inline bool operator>=(const PartitionSortEntry& other) const ALWAYS_INLINE {
      return partition_ >= other.partition_;
    }

    uint16_t              filler1_;  // +2 => 2
    uint8_t               filler2_;  // +1 => 3
    storage::PartitionId  partition_;  // +1 => 4
    BufferPosition        position_;  // +4 => 8
  };

  struct IoBufStatus {
    uint64_t size_inbuf_aligned_;
    uint64_t size_infile_aligned_;

    uint64_t next_infile_;
    uint64_t buf_infile_aligned_;
    uint64_t cur_inbuf_;
    uint64_t end_inbuf_aligned_;
    uint64_t end_infile_;
    bool more_in_the_file_;
    bool first_read_;
    bool ended_;
    log::LogFileOrdinal cur_file_ordinal_;

    uint64_t to_infile(uint64_t inbuf) const { return inbuf + buf_infile_aligned_; }
  };

  memory::AlignedMemory   io_buffer_;

  memory::AlignedMemory   buckets_memory_;

  memory::AlignedMemory   tmp_memory_;

  memory::AlignedMemory   presort_buffer_;
  memory::AlignedMemory   presort_ouputs_;
  memory::AlignedMemory   presort_reordered_;

  memory::AlignedMemorySlice  tmp_send_buffer_slice_;

  memory::AlignedMemorySlice  tmp_position_array_slice_;

  memory::AlignedMemorySlice  tmp_sort_array_slice_;

  memory::AlignedMemorySlice  tmp_hashlist_buffer_slice_;

  memory::AlignedMemorySlice  tmp_partition_array_slice_;

  uint32_t                buckets_allocated_count_;

  uint32_t                hashlist_allocated_count_;

  uint64_t                processed_log_count_;

  BucketHashList*       storage_hashlists_[256];

  ErrorStack  handle_process_buffer(const fs::DirectIoFile &file, IoBufStatus* status);

  bool        bucket_log(storage::StorageId storage_id, uint64_t pos) ALWAYS_INLINE;

  bool        add_new_bucket(storage::StorageId storage_id);

  void        flush_all_buckets();

  void        flush_bucket(const BucketHashList& hashlist);

  void        send_bucket_partition(Bucket* bucket, storage::PartitionId partition);
  void send_bucket_partition_general(
    const Bucket* bucket,
    storage::StorageType storage_type,
    storage::PartitionId partition,
    const BufferPosition* positions);
  void        send_bucket_partition_presort(
    Bucket* bucket,
    storage::StorageType storage_type,
    storage::PartitionId partition);
  void        send_bucket_partition_buffer(
    const Bucket* bucket,
    storage::PartitionId partition,
    const char* send_buffer,
    uint32_t log_count,
    uint64_t written,
    uint32_t shortest_key_length,
    uint32_t longest_key_length);

  void        clear_storage_buckets();

  inline BucketHashList* find_storage_hashlist(storage::StorageId storage_id) ALWAYS_INLINE {
    uint8_t index = static_cast<uint8_t>(storage_id);
    BucketHashList* hashlist = storage_hashlists_[index];
    if (UNLIKELY(hashlist == nullptr)) {
      return nullptr;
    } else {
      while (UNLIKELY(hashlist->storage_id_ != storage_id)) {
        hashlist = hashlist->hashlist_next_;
        if (hashlist == nullptr) {
          return nullptr;
        }
      }
    }

    return hashlist;
  }
  void add_storage_hashlist(BucketHashList* new_hashlist);

  void report_completion(double elapsed_sec);
};


}  // namespace snapshot
}  // namespace foedus
#endif  // FOEDUS_SNAPSHOT_LOG_MAPPER_IMPL_HPP_