mmap_batch.cc

#include <iostream>
#include <unistd.h>
#include <cstring>
#include <sys/mman.h>
#include <vector>
#include <thread>
#include <sys/stat.h> 
#include <fcntl.h>

#define likely(x)      __builtin_expect(!!(x), 1)
#define unlikely(x)    __builtin_expect(!!(x), 0)

static char data[4096] __attribute__((aligned(4096))) = {'a'};
static int32_t map_size = 4096 * 4;

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);
}

int main() {
  std::vector<std::thread> threads;
  for(int i = 0; i < 64; i++) {
    std::thread worker(writer, i);
    threads.push_back(std::move(worker));
  }
  for (int i = 0; i < 64; i++) {
    threads[i].join();
  }

  return 0;
}

// g++ -std=c++11 -O2 -pthread mmap_batch.cc -o benchmark