mmap_reader.cpp

#include "mmap_reader.h"

#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>

#include <cstring>
#include <stdexcept>
#include <system_error>
#include <utility>

namespace {
  void throw_errno(const std::string& what) {
    throw std::system_error(errno, std::generic_category(), what);
}
}  // namespace

MmapReader::MmapReader(const std::string& path) {
  fd_ = ::open(path.c_str(), O_RDONLY);
  if (fd_ < 0) throw_errno("open(" + path + ")");

  struct stat st;
  if (::fstat(fd_, &st) != 0) {
    int e = errno;
    ::close(fd_);
    fd_ = -1;
    throw std::system_error(e, std::generic_category(), "fstat(" + path + ")");
  }
  size_ = static_cast<std::size_t>(st.st_size);

  if (size_ == 0) {
    // mmap() rejects a zero length; an empty file simply yields no chunks.
    data_ = nullptr;
    return;
  }

  int flags = MAP_PRIVATE;
#ifdef MAP_POPULATE
  // Linux: prefault all pages now so processing doesn't stall on minor faults.
  flags |= MAP_POPULATE;
#endif
  void* p = ::mmap(nullptr, size_, PROT_READ, flags, fd_, 0);
  if (p == MAP_FAILED) {
    int e = errno;
    ::close(fd_);
    fd_ = -1;
    throw std::system_error(e, std::generic_category(), "mmap(" + path + ")");
  }
  data_ = static_cast<const char*>(p);

  // We scan front-to-back exactly once: ask the kernel for aggressive readahead
  // and to drop pages behind us, which keeps a >RAM file from thrashing.
  ::madvise(const_cast<char*>(data_), size_, MADV_SEQUENTIAL | MADV_WILLNEED);
}

void MmapReader::reset() noexcept {
  if (data_ != nullptr && size_ > 0) {
    ::munmap(const_cast<char*>(data_), size_);
  }
  if (fd_ >= 0) {
    ::close(fd_);
  }
  fd_ = -1;
  data_ = nullptr;
  size_ = 0;
}

MmapReader::~MmapReader() { reset(); }

MmapReader::MmapReader(MmapReader&& other) noexcept
    : fd_(other.fd_), data_(other.data_), size_(other.size_) {
  other.fd_ = -1;
  other.data_ = nullptr;
  other.size_ = 0;
}

MmapReader& MmapReader::operator=(MmapReader&& other) noexcept {
  if (this != &other) {
    reset();
    fd_ = other.fd_;
    data_ = other.data_;
    size_ = other.size_;
    other.fd_ = -1;
    other.data_ = nullptr;
    other.size_ = 0;
  }
  return *this;
}

std::vector<Chunk> MmapReader::chunks(std::size_t target_size) const {
  std::vector<Chunk> result;
  if (size_ == 0 || target_size == 0) return result;

  const char* const base = data_;
  const char* const end = data_ + size_;
  result.reserve(size_ / target_size + 1);

  const char* cur = base;
  while (cur < end) {
    const char* chunk_end = cur + target_size;
    if (chunk_end >= end) {
      chunk_end = end;  // last chunk runs to EOF
    } else {
      // Extend to just past the next newline so we never split a record.
      const char* nl = static_cast<const char*>(
          ::memchr(chunk_end, '\n', static_cast<std::size_t>(end - chunk_end)));
      chunk_end = (nl == nullptr) ? end : nl + 1;
    }
    result.push_back(Chunk{cur, chunk_end});
    cur = chunk_end;
  }
  return result;
}