void MapRegion(int fd, uint64_t file_offset, char** base) {
  void* ptr = mmap(nullptr, map_size, PROT_READ | PROT_WRITE,
           MAP_SHARED,
           fd,
           file_offset);
  if (unlikely(ptr == MAP_FAILED)) {
    *base = nullptr;
    return;
  }
  *base = reinterpret_cast<char*>(ptr);
}

void UnMapRegion(char* base) {
  munmap(base, map_size);
}

void writer(int index) {
  std::string fname = "data" + std::to_string(index);
  std::string batch = "batch" + std::to_string(index);

  char* base = nullptr;
  char* cursor = nullptr;
  uint64_t mmap_offset = 0, file_offset = 0;

  int data_fd = ::open(fname.c_str(), O_RDWR | O_CREAT | O_DIRECT, 0645);
  int batch_fd = ::open(batch.c_str(), O_RDWR | O_CREAT | O_DIRECT, 0645);
  posix_fallocate(data_fd, 0, (4096UL * 1000000));
  posix_fallocate(batch_fd, 0, map_size);

  MapRegion(batch_fd, 0, &base);
  if (unlikely(base == nullptr)) {
    return;
  }
  cursor = base;
  file_offset += map_size;

  for (int32_t i = 0; i < 1000000; i++) {
    if (unlikely(mmap_offset >= map_size)) {
      pwrite64(data_fd, base, map_size, file_offset);
      cursor = base;

      file_offset += map_size;
      mmap_offset = 0;
    }
    memcpy(cursor, data, 4096);
    cursor += 4096;
    mmap_offset += 4096;
  }
  UnMapRegion(base);

  close(data_fd);
  close(batch_fd);
}