DumpFileSortedBuffer::DumpFileSortedBuffer(
  fs::DirectIoFile* file, memory::AlignedMemorySlice io_buffer)
  : SortedBuffer(
    reinterpret_cast<char*>(io_buffer.get_block()),
    io_buffer.get_size(),
    fs::file_size(file->get_path())),
    file_(file),
    io_buffer_(io_buffer) {
  ASSERT_ND(buffer_size_ % kAlignment == 0);
  ASSERT_ND(total_size_ % kAlignment == 0);
}

inline uint32_t ConstDiv::rem64(uint64_t n, uint32_t d, uint64_t q) const {
#ifndef NDEBUG
  ASSERT_ND(d == d_);
#endif  // NDEBUG
  ASSERT_ND(n / d == q);
  if (flags_ & kFlagPowerOfTwo) {
    return n & ((1ULL << d_highest_bits_) - 1ULL);
  } else {
    return n - d * q;
  }
}

xct::McsBlockIndex ThreadPimpl::mcs_initial_lock(xct::McsLock* mcs_lock) {
  assert_mcs_aligned(mcs_lock);
  ASSERT_ND(!mcs_lock->is_locked());
  // so far we allow only 2^16 MCS blocks per transaction. we might increase later.
  ASSERT_ND(current_xct_.get_mcs_block_current() < 0xFFFFU);
  xct::McsBlockIndex block_index = current_xct_.increment_mcs_block_current();
  ASSERT_ND(block_index > 0);
  mcs_init_block(mcs_lock, block_index, false);
  mcs_lock->reset(id_, block_index);
  assorted::memory_fence_acq_rel();
  return block_index;
}

ErrorCode ThreadPimpl::follow_page_pointers_for_write_batch(
  uint16_t batch_size,
  storage::VolatilePageInit page_initializer,
  storage::DualPagePointer** pointers,
  storage::Page** parents,
  const uint16_t* index_in_parents,
  storage::Page** out) {
  // REMINDER: Remember that it might be parents == out. It's not an issue in this function, tho.
  // this method is not quite batched as it doesn't need to be.
  // still, less branches because we can assume all of them need a writable volatile page.
  for (uint16_t b = 0; b < batch_size; ++b) {
    storage::DualPagePointer* pointer = pointers[b];
    if (pointer == nullptr) {
      out[b] = nullptr;
      continue;
    }
    ASSERT_ND(!parents[b]->get_header().snapshot_);
    storage::Page** page = out + b;
    storage::VolatilePagePointer volatile_pointer = pointer->volatile_pointer_;
    if (!volatile_pointer.is_null()) {
      *page = global_volatile_page_resolver_.resolve_offset(volatile_pointer);
    } else if (pointer->snapshot_pointer_ == 0) {
      // we need a volatile page. so construct it from snapshot
      CHECK_ERROR_CODE(install_a_volatile_page(pointer, page));
    } else {
      ASSERT_ND(page_initializer);
      // we must not install a new volatile page in snapshot page. We must not hit this case.
      memory::PagePoolOffset offset = core_memory_->grab_free_volatile_page();
      if (UNLIKELY(offset == 0)) {
        return kErrorCodeMemoryNoFreePages;
      }
      storage::Page* new_page = local_volatile_page_resolver_.resolve_offset_newpage(offset);
      storage::VolatilePagePointer new_page_id;
      new_page_id.components.numa_node = numa_node_;
      new_page_id.components.offset = offset;
      storage::VolatilePageInitArguments args = {
        holder_,
        new_page_id,
        new_page,
        parents[b],
        index_in_parents[b]
      };
      page_initializer(args);
      storage::assert_valid_volatile_page(new_page, offset);
      ASSERT_ND(new_page->get_header().snapshot_ == false);

      *page = place_a_new_volatile_page(offset, pointer);
    }
    ASSERT_ND(out[b] != nullptr);
  }
  return kErrorCodeOk;
}

uint32_t PagePoolOffsetAndEpochChunk::get_safe_offset_count(const Epoch& threshold) const {
  ASSERT_ND(is_sorted());
  OffsetAndEpoch dummy;
  dummy.safe_epoch_ = threshold.value();
  struct CompareEpoch {
    bool operator() (const OffsetAndEpoch& left, const OffsetAndEpoch& right) {
      return Epoch(left.safe_epoch_) < Epoch(right.safe_epoch_);
    }
  };
  const OffsetAndEpoch* result = std::lower_bound(chunk_, chunk_ + size_, dummy, CompareEpoch());
  ASSERT_ND(result);
  ASSERT_ND(result - chunk_ <= size_);
  return result - chunk_;
}

bool LogMapper::add_new_bucket(storage::StorageId storage_id) {
  if (buckets_allocated_count_ >= buckets_memory_.get_size() / sizeof(Bucket)) {
    // we allocated all buckets_memory_! we have to flush the buckets now.
    // this shouldn't happen often.
    LOG(WARNING) << to_string() << " ran out of buckets_memory_, so it has to flush buckets before"
      " processing one IO buffer. This shouldn't happen often. check your log_mapper_bucket_kb_"
      " setting. this=" << *this;
    return false;
  }

  Bucket* base_address = reinterpret_cast<Bucket*>(buckets_memory_.get_block());
  Bucket* new_bucket = base_address + buckets_allocated_count_;
  ++buckets_allocated_count_;
  new_bucket->storage_id_ = storage_id;
  new_bucket->counts_ = 0;
  new_bucket->next_bucket_ = nullptr;

  BucketHashList* hashlist = find_storage_hashlist(storage_id);
  if (hashlist) {
    // just add this as a new tail
    ASSERT_ND(hashlist->storage_id_ == storage_id);
    ASSERT_ND(hashlist->tail_->is_full());
    hashlist->tail_->next_bucket_ = new_bucket;
    hashlist->tail_ = new_bucket;
    ++hashlist->bucket_counts_;
  } else {
    // we don't even have a linked list for this.
    // If this happens often, maybe we should have 65536 hash buckets...
    if (hashlist_allocated_count_ >= kBucketHashListMaxCount) {
      LOG(WARNING) << to_string() << " ran out of hashlist memory, so it has to flush buckets now"
        " This shouldn't happen often. We must consider increasing kBucketHashListMaxCount."
        " this=" << *this;
      return false;
    }

    // allocate from the pool
    BucketHashList* new_hashlist =
      reinterpret_cast<BucketHashList*>(tmp_hashlist_buffer_slice_.get_block())
      + hashlist_allocated_count_;
    ++hashlist_allocated_count_;

    new_hashlist->storage_id_ = storage_id;
    new_hashlist->head_ = new_bucket;
    new_hashlist->tail_ = new_bucket;
    new_hashlist->bucket_counts_ = 1;

    add_storage_hashlist(new_hashlist);
  }
  return true;
}

void LogMapper::send_bucket_partition_general(
  const Bucket* bucket,
  storage::StorageType storage_type,
  storage::PartitionId partition,
  const BufferPosition* positions) {
  uint64_t written = 0;
  uint32_t log_count = 0;
  uint32_t shortest_key_length = 0xFFFF;
  uint32_t longest_key_length = 0;
  // stitch the log entries in send buffer
  char* send_buffer = reinterpret_cast<char*>(tmp_send_buffer_slice_.get_block());
  const char* io_base = reinterpret_cast<const char*>(io_buffer_.get_block());
  ASSERT_ND(tmp_send_buffer_slice_.get_size() == kSendBufferSize);

  for (uint32_t i = 0; i < bucket->counts_; ++i) {
    uint64_t pos = from_buffer_position(positions[i]);
    const log::LogHeader* header = reinterpret_cast<const log::LogHeader*>(io_base + pos);
    ASSERT_ND(header->storage_id_ == bucket->storage_id_);
    uint16_t log_length = header->log_length_;
    ASSERT_ND(log_length > 0);
    ASSERT_ND(log_length % 8 == 0);
    if (written + log_length > kSendBufferSize) {
      // buffer full. send out.
      send_bucket_partition_buffer(
        bucket,
        partition,
        send_buffer,
        log_count,
        written,
        shortest_key_length,
        longest_key_length);
      log_count = 0;
      written = 0;
      shortest_key_length = 0xFFFF;
      longest_key_length = 0;
    }
    std::memcpy(send_buffer + written, header, header->log_length_);
    written += header->log_length_;
    ++log_count;
    update_key_lengthes(header, storage_type, &shortest_key_length, &longest_key_length);
  }
  send_bucket_partition_buffer(
    bucket,
    partition,
    send_buffer,
    log_count,
    written,
    shortest_key_length,
    longest_key_length);
}

ErrorStack inserts_varlen_task(const proc::ProcArguments& args) {
  EXPECT_EQ(sizeof(uint32_t), args.input_len_);
  uint32_t id = *reinterpret_cast<const uint32_t*>(args.input_buffer_);
  EXPECT_NE(id, 2U);

  thread::Thread* context = args.context_;
  storage::hash::HashStorage hash(args.engine_, kName);
  ASSERT_ND(hash.exists());
  xct::XctManager* xct_manager = args.engine_->get_xct_manager();
  Epoch commit_epoch;

  WRAP_ERROR_CODE(xct_manager->begin_xct(context, xct::kSerializable));

  char buffer[16];
  std::memset(buffer, 0, sizeof(buffer));
  for (uint32_t i = 0; i < kRecords / 2U; ++i) {
    uint64_t rec = id * kRecords / 2U + i;
    // first 8 bytes, mod 17 to have next layers.
    assorted::write_bigendian<uint64_t>(static_cast<uint64_t>(rec % 17U), buffer);
    // and 1-4 bytes of decimal representation in text
    std::string str = std::to_string(rec);
    std::memcpy(buffer + sizeof(uint64_t), str.data(), str.size());
    uint16_t len = sizeof(uint64_t) + str.size();
    uint64_t data = rec + kDataAddendum;
    ErrorCode ret = hash.insert_record(context, buffer, len, &data, sizeof(data));
    EXPECT_EQ(kErrorCodeOk, ret) << i;
  }

  // CHECK_ERROR(hash.debugout_single_thread(args.engine_));
  WRAP_ERROR_CODE(xct_manager->precommit_xct(context, &commit_epoch));
  // CHECK_ERROR(hash.debugout_single_thread(args.engine_));
  WRAP_ERROR_CODE(xct_manager->wait_for_commit(commit_epoch));
  return kRetOk;
}

ErrorStack verify_varlen_task(const proc::ProcArguments& args) {
  thread::Thread* context = args.context_;
  storage::hash::HashStorage hash(args.engine_, kName);
  ASSERT_ND(hash.exists());
  CHECK_ERROR(hash.verify_single_thread(context));
  // CHECK_ERROR(hash.debugout_single_thread(args.engine_));
  xct::XctManager* xct_manager = args.engine_->get_xct_manager();
  WRAP_ERROR_CODE(xct_manager->begin_xct(context, xct::kSerializable));

  char buffer[16];
  std::memset(buffer, 0, sizeof(buffer));
  for (uint32_t i = 0; i < kRecords; ++i) {
    uint64_t rec = i;
    assorted::write_bigendian<uint64_t>(static_cast<uint64_t>(rec % 17U), buffer);
    std::string str = std::to_string(rec);
    std::memcpy(buffer + sizeof(uint64_t), str.data(), str.size());
    uint16_t len = sizeof(uint64_t) + str.size();

    uint64_t data;
    uint16_t capacity = sizeof(data);
    ErrorCode ret = hash.get_record(context, buffer, len, &data, &capacity);
    EXPECT_EQ(kErrorCodeOk, ret) << i;
    EXPECT_EQ(i + kDataAddendum, data) << i;
    EXPECT_EQ(sizeof(data), capacity) << i;
  }

  Epoch commit_epoch;
  ErrorCode committed = xct_manager->precommit_xct(context, &commit_epoch);
  EXPECT_EQ(kErrorCodeOk, committed);
  return kRetOk;
}

void HashDataPage::release_pages_recursive(
  const memory::GlobalVolatilePageResolver& page_resolver,
  memory::PageReleaseBatch* batch) {
  if (next_page_.volatile_pointer_.components.offset != 0) {
    HashDataPage* next = reinterpret_cast<HashDataPage*>(
      page_resolver.resolve_offset(next_page_.volatile_pointer_));
    ASSERT_ND(next->header().get_in_layer_level() == 0);
    ASSERT_ND(next->get_bin() == get_bin());
    next->release_pages_recursive(page_resolver, batch);
    next_page_.volatile_pointer_.components.offset = 0;
  }

  VolatilePagePointer volatile_id;
  volatile_id.word = header().page_id_;
  batch->release(volatile_id);
}

ErrorStack verify_task(const proc::ProcArguments& args) {
  thread::Thread* context = args.context_;
  storage::masstree::MasstreeStorage masstree(args.engine_, kName);
  ASSERT_ND(masstree.exists());
  CHECK_ERROR(masstree.verify_single_thread(context));
  xct::XctManager* xct_manager = args.engine_->get_xct_manager();
  WRAP_ERROR_CODE(xct_manager->begin_xct(context, xct::kSerializable));

  for (uint32_t i = 0; i < kRecords; ++i) {
    uint64_t rec = i;
    storage::masstree::KeySlice slice = storage::masstree::normalize_primitive<uint64_t>(rec);
    uint64_t data;
    uint16_t capacity = sizeof(data);
    ErrorCode ret = masstree.get_record_normalized(context, slice, &data, &capacity, true);
/*
    if (ret != kErrorCodeOk || rec != data) {
      CHECK_ERROR(masstree.verify_single_thread(context));
      CHECK_ERROR(masstree.debugout_single_thread(args.engine_));
      std::cout << "asdasd" << std::endl;
    }
*/
    EXPECT_EQ(kErrorCodeOk, ret) << i;
    EXPECT_EQ(rec, data) << i;
    EXPECT_EQ(sizeof(data), capacity) << i;
  }

  Epoch commit_epoch;
  ErrorCode committed = xct_manager->precommit_xct(context, &commit_epoch);
  EXPECT_EQ(kErrorCodeOk, committed);
  return kRetOk;
}

ErrorStack TpccLoadTask::load_customers_in_district(Wid wid, Did did) {
  LOG(INFO) << "Loading Customer for DID=" << static_cast<int>(did) << ", WID=" << wid
    << ": " << engine_->get_memory_manager()->dump_free_memory_stat();

  // insert to customers_secondary at the end after sorting
  memory::AlignedMemory secondary_keys_buffer;
  struct Secondary {
    char  last_[17];      // +17 -> 17
    char  first_[17];     // +17 -> 34
    char  padding_[2];    // +2 -> 36
    Cid   cid_;           // +4 -> 40
    static bool compare(const Secondary &left, const Secondary& right) ALWAYS_INLINE {
      int cmp = std::memcmp(left.last_, right.last_, sizeof(left.last_));
      if (cmp < 0) {
        return true;
      } else if (cmp > 0) {
        return false;
      }
      cmp = std::memcmp(left.first_, right.first_, sizeof(left.first_));
      if (cmp < 0) {
        return true;
      } else if (cmp > 0) {
        return false;
      }
        ASSERT_ND(left.cid_ != right.cid_);
      if (left.cid_ < right.cid_) {
        return true;
      } else {
        return false;
      }
    }
  };

TpccLoadTask::TpccLoadTask(char* ctime_buffer) {
  // Initialize timestamp (for date columns)
  time_t t_clock;
  ::time(&t_clock);
  timestamp_ = ::ctime_r(&t_clock, ctime_buffer);
  ASSERT_ND(timestamp_);
}

void ThreadPimpl::flush_retired_volatile_page(
  uint16_t node,
  Epoch current_epoch,
  memory::PagePoolOffsetAndEpochChunk* chunk) {
  if (chunk->size() == 0) {
    return;
  }
  uint32_t safe_count = chunk->get_safe_offset_count(current_epoch);
  while (safe_count < chunk->size() / 10U) {
    LOG(WARNING) << "Thread-" << id_ << " can return only "
      << safe_count << " out of " << chunk->size()
      << " retired pages to node-" << node  << " in epoch=" << current_epoch
      << ". This means the thread received so many retired pages in a short time period."
      << " Will adavance an epoch to safely return the retired pages."
      << " This should be a rare event.";
    engine_->get_xct_manager()->advance_current_global_epoch();
    current_epoch = engine_->get_xct_manager()->get_current_global_epoch();
    LOG(INFO) << "okay, advanced epoch. now we should be able to return more pages";
    safe_count = chunk->get_safe_offset_count(current_epoch);
  }

  VLOG(0) << "Thread-" << id_ << " batch-returning retired volatile pages to node-" << node
    << " safe_count/count=" << safe_count << "/" << chunk->size() << ". epoch=" << current_epoch;
  memory::PagePool* volatile_pool
    = engine_->get_memory_manager()->get_node_memory(node)->get_volatile_pool();
  volatile_pool->release(safe_count, chunk);
  ASSERT_ND(!chunk->full());
}

DivvyupPageGrabBatch::DivvyupPageGrabBatch(Engine* engine)
: engine_(engine), node_count_(engine->get_options().thread_.group_count_) {
  chunks_ = new PagePoolOffsetChunk[node_count_];
  for (uint16_t node = 0; node < node_count_; ++node) {
    ASSERT_ND(chunks_[node].empty());
  }
}

 void get_assignments(uint32_t count, uint32_t* result_from, uint32_t* result_to) const {
   if (load_threads_ == 1U) {
     ASSERT_ND(thread_ordinal_ == 0);
     *result_from = 0;
     *result_to = count;
   } else {
     ASSERT_ND(thread_ordinal_ < load_threads_);
     uint32_t div = count / load_threads_;
     ASSERT_ND(div > 0);
     EXPECT_GT(div, 0);
     *result_from = div * thread_ordinal_;
     *result_to = div * (thread_ordinal_ + 1U);
     if (thread_ordinal_ + 1U == load_threads_) {
       *result_to = count;
     }
   }
 }

bool ImpersonateSession::is_running() const {
  ASSERT_ND(thread_->get_control_block()->current_ticket_ >= ticket_);
  if (thread_->get_control_block()->current_ticket_ != ticket_) {
    return false;
  }
  return (thread_->get_control_block()->status_ == kWaitingForExecution ||
    thread_->get_control_block()->status_ == kRunningTask);
}

void hash_data_volatile_page_init(const VolatilePageInitArguments& args) {
  ASSERT_ND(args.parent_);
  ASSERT_ND(args.page_);
  StorageId storage_id = args.parent_->get_header().storage_id_;
  HashDataPage* page = reinterpret_cast<HashDataPage*>(args.page_);
  PageType parent_type = args.parent_->get_header().get_page_type();
  uint64_t bin;
  if (parent_type == kHashIntermediatePageType) {
    const HashIntermediatePage* parent
      = reinterpret_cast<const HashIntermediatePage*>(args.parent_);

    ASSERT_ND(args.index_in_parent_ < kHashIntermediatePageFanout);
    ASSERT_ND(parent->get_level() == 0);
    ASSERT_ND(parent->get_bin_range().length() == kHashIntermediatePageFanout);
    bin = parent->get_bin_range().begin_ + args.index_in_parent_;
  } else {
    ASSERT_ND(parent_type == kHashDataPageType);
    ASSERT_ND(args.index_in_parent_ == 0);
    const HashDataPage* parent = reinterpret_cast<const HashDataPage*>(args.parent_);
    bin = parent->get_bin();
  }
  HashStorage storage(args.context_->get_engine(), storage_id);
  uint8_t bin_bits = storage.get_bin_bits();
  uint8_t bin_shifts = storage.get_bin_shifts();
  page->initialize_volatile_page(storage_id, args.page_id, args.parent_, bin, bin_bits, bin_shifts);
}

void MetaLogger::meta_logger_main() {
  LOG(INFO) << "Meta-logger started";
  while (!stop_requested_) {
    {
      uint64_t demand = control_block_->logger_wakeup_.acquire_ticket();
      if (!stop_requested_ && !control_block_->has_waiting_log()) {
        VLOG(0) << "Meta-logger going to sleep";
        control_block_->logger_wakeup_.timedwait(demand, 100000ULL);
      }
    }
    VLOG(0) << "Meta-logger woke up";
    if (stop_requested_) {
      break;
    } else if (!control_block_->has_waiting_log()) {
      continue;
    }

    // we observed buffer_used_ > 0 BEFORE the fence. Now we can read buffer_ safely.
    assorted::memory_fence_acq_rel();
    ASSERT_ND(control_block_->buffer_used_ > 0);
    uint32_t write_size = sizeof(control_block_->buffer_);
    ASSERT_ND(control_block_->buffer_used_ <= write_size);
    LOG(INFO) << "Meta-logger got a log (" << control_block_->buffer_used_ << " b) to write out";
    FillerLogType* filler = reinterpret_cast<FillerLogType*>(
      control_block_->buffer_ + control_block_->buffer_used_);
    filler->populate(write_size - control_block_->buffer_used_);
    ErrorCode er = current_file_->write_raw(write_size, control_block_->buffer_);
    if (er != kErrorCodeOk) {
      LOG(FATAL) << "Meta-logger couldn't write a log. error=" << get_error_message(er)
        << ", os_error=" << assorted::os_error();
    }

    // also fsync on the file and parent. every, single, time.
    // we don't care performance on metadata logger. just make it simple and robust.
    bool synced = fs::fsync(current_file_->get_path(), true);
    if (!synced) {
      LOG(FATAL) << "Meta-logger couldn't fsync. os_error=" << assorted::os_error();
    }
    assorted::memory_fence_release();
    control_block_->buffer_used_ = 0;
    assorted::memory_fence_release();
    control_block_->durable_offset_ += write_size;
  }
  LOG(INFO) << "Meta-logger terminated";
}

void hash_intermediate_volatile_page_init(const VolatilePageInitArguments& args) {
  ASSERT_ND(args.parent_);  // because this is always called for non-root pages.
  ASSERT_ND(args.page_);
  ASSERT_ND(args.index_in_parent_ < kHashIntermediatePageFanout);
  StorageId storage_id = args.parent_->get_header().storage_id_;
  HashIntermediatePage* page = reinterpret_cast<HashIntermediatePage*>(args.page_);

  ASSERT_ND(args.parent_->get_header().get_page_type() == kHashIntermediatePageType);
  const HashIntermediatePage* parent = reinterpret_cast<const HashIntermediatePage*>(args.parent_);

  ASSERT_ND(parent->get_level() > 0);
  parent->assert_range();
  HashBin interval = kHashMaxBins[parent->get_level()];
  HashBin parent_begin = parent->get_bin_range().begin_;
  HashBin begin = parent_begin + args.index_in_parent_ * interval;
  uint8_t level = parent->get_level() - 1U;
  page->initialize_volatile_page(storage_id, args.page_id, parent, level, begin);
}