cache_bench_host.c

Go to the documentation of this file.

#include "cache_bench_host.h"
 
#include <errno.h>
#include <fcntl.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <unistd.h>
 
#include "ra8_attributes.h"
 
typedef enum : uint8_t {
  k_cb_host_retry_limit = 32U, 
} cb_host_retry_t;
 
typedef enum : uint16_t {
  k_cb_host_create_mode = 0600U, 
} cb_host_mode_t;

RA8_INTERNAL
static cb_io_status_t
internal_sink_write(void* ctx, const uint8_t* data, size_t length, size_t* out_written)
{
  if ((ctx == nullptr) || (out_written == nullptr)) {
    return k_cb_io_fault;
  }
  const int fd = *(const int*)ctx;
  for (uint8_t attempt = 0U; attempt < (uint8_t)k_cb_host_retry_limit; ++attempt) {
    const ssize_t result = write(fd, data, length);
    if (result >= 0) {
      *out_written = (size_t)result;
      return k_cb_io_ok;
    }
    if (errno != EINTR) {
      break;
    }
  }
  *out_written = 0U;
  return k_cb_io_fault;
}
 
void cb_host_standard_sinks(cb_sink_t* output, cb_sink_t* error)
{
  static int s_output_fd = STDOUT_FILENO;
  static int s_error_fd  = STDERR_FILENO;
  if (output != nullptr) {
    *output = (cb_sink_t){.write = internal_sink_write, .ctx = &s_output_fd};
  }
  if (error != nullptr) {
    *error = (cb_sink_t){.write = internal_sink_write, .ctx = &s_error_fd};
  }
}

RA8_INTERNAL
static cb_io_status_t
internal_source_read(void* ctx, uint64_t offset, uint8_t* dst, size_t capacity, size_t* out_read)
{
  cb_host_source_t* source = (cb_host_source_t*)ctx;
  if ((source == nullptr) || (dst == nullptr) || (out_read == nullptr) || (offset > source->size)) {
    return k_cb_io_fault;
  }
  struct stat info = {};
  if ((fstat(source->fd, &info) != 0) || (info.st_size < 0) ||
      ((uint64_t)info.st_size != source->size)) {
    return k_cb_io_mutated;
  }
  for (uint8_t attempt = 0U; attempt < (uint8_t)k_cb_host_retry_limit; ++attempt) {
    const ssize_t result = pread(source->fd, dst, capacity, (off_t)offset);
    if (result >= 0) {
      *out_read = (size_t)result;
      return k_cb_io_ok;
    }
    if (errno != EINTR) {
      break;
    }
  }
  *out_read = 0U;
  return k_cb_io_fault;
}
 
cb_io_status_t cb_host_source_open(const char* path, cb_host_source_t* binding, cb_source_t* source)
{
  if ((path == nullptr) || (binding == nullptr) || (source == nullptr)) {
    return k_cb_io_fault;
  }
  *binding     = (cb_host_source_t){.fd = -1};
  const int fd = open(path, O_RDONLY | O_CLOEXEC | O_NOFOLLOW);
  if (fd < 0) {
    return k_cb_io_fault;
  }
  struct stat info = {};
  if ((fstat(fd, &info) != 0) || !S_ISREG(info.st_mode) || (info.st_size < 0)) {
    (void)close(fd);
    return k_cb_io_fault;
  }
  *binding = (cb_host_source_t){.fd = fd, .size = (uint64_t)info.st_size};
  *source  = (cb_source_t){.read = internal_source_read, .ctx = binding, .size = binding->size};
  return k_cb_io_ok;
}
 
cb_io_status_t cb_host_source_close(cb_host_source_t* binding)
{
  if ((binding == nullptr) || (binding->fd < 0)) {
    return k_cb_io_fault;
  }
  const int result = close(binding->fd);
  binding->fd      = -1;
  return (result == 0) ? k_cb_io_ok : k_cb_io_fault;
}

RA8_INTERNAL
static cb_io_status_t
internal_split_path(const char* path, char* parent, size_t parent_capacity, const char** base)
{
  const char* slash          = strrchr(path, '/');
  *base                      = (slash == nullptr) ? path : slash + 1;
  const size_t parent_length = (slash == nullptr) ? 1U : (size_t)(slash - path);
  if ((**base == '\0') || (parent_length >= parent_capacity)) {
    return k_cb_io_capacity;
  }
  size_t output_length = parent_length;
  if (slash == nullptr) {
    parent[0] = '.';
  } else if (parent_length == 0U) {
    parent[0]     = '/';
    output_length = 1U;
  } else {
    memcpy(parent, path, parent_length);
  }
  parent[output_length] = '\0';
  return k_cb_io_ok;
}

RA8_INTERNAL
static cb_io_status_t
internal_create_temp_exclusive(cb_host_output_t* binding, const char* parent, const char* base)
{
  for (uint8_t attempt = 0U; attempt < (uint8_t)k_cb_host_retry_limit; ++attempt) {
    const int length = snprintf(binding->temporary,
                                sizeof(binding->temporary),
                                "%s/.%s.cache-bench.%ld.%u",
                                parent,
                                base,
                                (long)getpid(),
                                (unsigned int)attempt);
    if ((length < 0) || ((size_t)length >= sizeof(binding->temporary))) {
      return k_cb_io_capacity;
    }
    binding->fd = open(binding->temporary,
                       O_WRONLY | O_CREAT | O_EXCL | O_CLOEXEC,
                       (mode_t)k_cb_host_create_mode);
    if (binding->fd >= 0) {
      binding->transactional = true;
      return k_cb_io_ok;
    }
    if (errno != EEXIST) {
      break;
    }
  }
  return k_cb_io_fault;
}
 
cb_io_status_t cb_host_output_open(const char* path, cb_host_output_t* binding, cb_sink_t* sink)
{
  if ((binding == nullptr) || (sink == nullptr)) {
    return k_cb_io_fault;
  }
  *binding = (cb_host_output_t){.fd = -1, .dir_fd = -1};
  if (path == nullptr) {
    binding->fd = STDOUT_FILENO;
    *sink       = (cb_sink_t){.write = internal_sink_write, .ctx = &binding->fd};
    return k_cb_io_ok;
  }
  const size_t path_length = strlen(path);
  if (path_length >= sizeof(binding->destination)) {
    return k_cb_io_capacity;
  }
  memcpy(binding->destination, path, path_length + 1U);
  char           parent[k_cb_host_path_capacity];
  const char*    base   = nullptr;
  cb_io_status_t status = internal_split_path(path, parent, sizeof(parent), &base);
  if (status != k_cb_io_ok) {
    return status;
  }
  binding->dir_fd = open(parent, O_RDONLY | O_DIRECTORY | O_CLOEXEC);
  if (binding->dir_fd < 0) {
    return k_cb_io_fault;
  }
  status = internal_create_temp_exclusive(binding, parent, base);
  if (status == k_cb_io_ok) {
    *sink = (cb_sink_t){.write = internal_sink_write, .ctx = &binding->fd};
    return k_cb_io_ok;
  }
  cb_host_output_abort(binding);
  return status;
}
 
cb_io_status_t cb_host_output_commit(cb_host_output_t* binding)
{
  if ((binding == nullptr) || (binding->fd < 0)) {
    return k_cb_io_fault;
  }
  if (!binding->transactional) {
    return k_cb_io_ok;
  }
  bool ok     = fsync(binding->fd) == 0;
  ok          = (close(binding->fd) == 0) && ok;
  binding->fd = -1;
  if (ok) {
    ok = rename(binding->temporary, binding->destination) == 0;
  }
  if (ok) {
    ok = fsync(binding->dir_fd) == 0;
  }
  const bool dir_ok = close(binding->dir_fd) == 0;
  binding->dir_fd   = -1;
  if (!ok) {
    (void)unlink(binding->temporary);
  }
  binding->transactional = false;
  return (ok && dir_ok) ? k_cb_io_ok : k_cb_io_fault;
}
 
void cb_host_output_abort(cb_host_output_t* binding)
{
  if (binding == nullptr) {
    return;
  }
  if (binding->fd >= 0) {
    (void)close(binding->fd);
    binding->fd = -1;
  }
  if (binding->transactional) {
    (void)unlink(binding->temporary);
  }
  if (binding->dir_fd >= 0) {
    (void)close(binding->dir_fd);
    binding->dir_fd = -1;
  }
  binding->transactional = false;
}

RA8_INTERNAL
static cb_io_status_t
internal_scratch_transfer(int fd, bool write_mode, uint64_t offset, void* data, size_t length)
{
  size_t   complete = 0U;
  uint32_t attempts = 0U;
  while (complete < length) {
    const ssize_t result =
      write_mode
        ? pwrite(fd,
                 &((const uint8_t*)data)[complete],
                 length - complete,
                 (off_t)(offset + complete))
        : pread(fd, &((uint8_t*)data)[complete], length - complete, (off_t)(offset + complete));
    if (result > 0) {
      complete += (size_t)result;
      attempts = 0U;
      continue;
    }
    if ((result < 0) && (errno == EINTR) && (attempts < (uint32_t)k_cb_host_retry_limit)) {
      attempts++;
      continue;
    }
    return k_cb_io_fault;
  }
  return k_cb_io_ok;
}

RA8_INTERNAL
static cb_io_status_t internal_scratch_read(void* ctx, uint64_t offset, void* data, size_t length)
{
  return internal_scratch_transfer(((cb_host_scratch_t*)ctx)->fd, false, offset, data, length);
}

RA8_INTERNAL
static cb_io_status_t internal_scratch_write(void* ctx, uint64_t offset, void* data, size_t length)
{
  return internal_scratch_transfer(((cb_host_scratch_t*)ctx)->fd, true, offset, data, length);
}
 
cb_io_status_t cb_host_scratch_open(cb_host_scratch_t* binding, cb_scratch_t* scratch)
{
  if ((binding == nullptr) || (scratch == nullptr)) {
    return k_cb_io_fault;
  }
  char path[] = "/tmp/ra8-cache-bench.XXXXXX";
  binding->fd = mkstemp(path);
  if (binding->fd < 0) {
    return k_cb_io_fault;
  }
  if (unlink(path) != 0) {
    (void)close(binding->fd);
    binding->fd = -1;
    return k_cb_io_fault;
  }
  *scratch =
    (cb_scratch_t){.read = internal_scratch_read, .write = internal_scratch_write, .ctx = binding};
  return k_cb_io_ok;
}
 
cb_io_status_t cb_host_scratch_close(cb_host_scratch_t* binding)
{
  if ((binding == nullptr) || (binding->fd < 0)) {
    return k_cb_io_fault;
  }
  const int result = close(binding->fd);
  binding->fd      = -1;
  return (result == 0) ? k_cb_io_ok : k_cb_io_fault;
}