lander/lsm/src/store/lsm_store_entry.c

#include <errno.h>
#include <fcntl.h>
#include <pthread.h>
#include <stdlib.h>
#include <string.h>
#include <sys/stat.h>

#include "lsm.h"
#include "lsm/store.h"
#include "lsm/store_internal.h"
#include "lsm/str.h"

lsm_error lsm_entry_init(lsm_entry **ptr) {
  lsm_entry *entry = calloc(1, sizeof(lsm_entry));

  if (entry == NULL) {
    return lsm_error_failed_alloc;
  }

  *ptr = entry;

  return lsm_error_ok;
}

void lsm_entry_free(lsm_entry *entry) {
  if (entry->attrs.count > 0) {
    for (size_t i = 0; i < entry->attrs.count; i++) {
      lsm_str_free(entry->attrs.items[i].str);
    }

    free(entry->attrs.items);
  }

  free(entry);
}

lsm_error lsm_entry_clone(lsm_entry **out, const lsm_entry *entry) {
  lsm_entry *new;
  LSM_RES(lsm_entry_init(&new));

  lsm_str_init_copy_n(&new->key, lsm_str_ptr(entry->key),
                      lsm_str_len(entry->key));

  for (int i = 0; i < 4; i++) {
    new->attrs.bitmap[i] = entry->attrs.bitmap[i];
  }

  new->attrs.count = entry->attrs.count;
  new->attrs.items = malloc(sizeof(lsm_attr) * entry->attrs.count);

  for (size_t i = 0; i < entry->attrs.count; i++) {
    new->attrs.items[i].type = entry->attrs.items[i].type;
    lsm_str_init_copy_n(&new->attrs.items[i].str,
                        lsm_str_ptr(entry->attrs.items[i].str),
                        lsm_str_len(entry->attrs.items[i].str));
  }

  new->data_len = entry->data_len;
  new->idx_file_offset = entry->idx_file_offset;

  *out = new;

  return lsm_error_ok;
}

lsm_error lsm_entry_wrapper_init(lsm_entry_wrapper **ptr) {
  lsm_entry_wrapper *wrap = calloc(1, sizeof(lsm_entry_wrapper));

  if (wrap == NULL) {
    return lsm_error_failed_alloc;
  }

  pthread_rwlock_init(&wrap->lock, NULL);

  *ptr = wrap;

  return lsm_error_ok;
}

void lsm_entry_wrapper_free(lsm_entry_wrapper *wrapper) { free(wrapper); }

bool lsm_entry_attr_present(const lsm_entry *entry, uint8_t type) {
  return (entry->attrs.bitmap[type / 64] & (((uint64_t)1) << (type % 64))) != 0;
}

lsm_error lsm_entry_attr_get(const lsm_str **out, const lsm_entry *entry,
                             uint8_t type) {
  if (!lsm_entry_attr_present(entry, type)) {
    return lsm_error_not_found;
  }

  uint64_t i = 0;

  while (entry->attrs.items[i].type != type) {
    i++;
  }

  *out = entry->attrs.items[i].str;

  return lsm_error_ok;
}

lsm_error lsm_entry_attr_get_uint64_t(uint64_t *out, const lsm_entry *entry,
                                      uint8_t type) {
  const lsm_str *s;

  LSM_RES(lsm_entry_attr_get(&s, entry, type));

  uint64_t num = 0;

  for (uint8_t i = 0; i < sizeof(uint64_t) / sizeof(char); i++) {
    ((char *)&num)[i] = lsm_str_char(s, i);
  }

  *out = num;

  return lsm_error_ok;
}

lsm_error lsm_entry_attr_get_uint8_t(uint8_t *out, const lsm_entry *entry,
                                     uint8_t type) {
  const lsm_str *s;

  LSM_RES(lsm_entry_attr_get(&s, entry, type));

  *out = lsm_str_char(s, 0);

  return lsm_error_ok;
}

lsm_error lsm_entry_attr_remove(lsm_str **out, lsm_entry *entry, uint8_t type) {
  if (!lsm_entry_attr_present(entry, type)) {
    return lsm_error_not_found;
  }

  if (entry->attrs.count == 1) {
    *out = entry->attrs.items[0].str;

    free(entry->attrs.items);
    entry->attrs.items = NULL;
    entry->attrs.count = 0;
    entry->attrs.bitmap[type / 64] &= ~(((uint64_t)1) << (type % 64));

    return lsm_error_ok;
  }

  uint64_t i = 0;

  while (entry->attrs.items[i].type != type) {
    i++;
  }

  lsm_attr *new_attrs = malloc((entry->attrs.count - 1) * sizeof(lsm_attr));

  if (new_attrs == NULL) {
    return lsm_error_failed_alloc;
  }

  if (out != NULL) {
    *out = entry->attrs.items[i].str;
  }

  memcpy(new_attrs, entry->attrs.items, i * sizeof(lsm_attr));
  memcpy(&new_attrs[i], &entry->attrs.items[i + 1],
         (entry->attrs.count - i - 1) * sizeof(lsm_attr));

  free(entry->attrs.items);

  entry->attrs.items = new_attrs;
  entry->attrs.count--;
  entry->attrs.bitmap[type / 64] &= ~(((uint64_t)1) << (type % 64));

  return lsm_error_ok;
}

lsm_error lsm_entry_attr_insert(lsm_entry *entry, uint8_t type, lsm_str *data) {
  // Remove a previous version of the attribute
  lsm_str *out;
  if (lsm_entry_attr_remove(&out, entry, type) == lsm_error_ok) {
    lsm_str_free(out);
  }

  lsm_attr *new_attrs =
      realloc(entry->attrs.items, (entry->attrs.count + 1) * sizeof(lsm_attr));

  if (new_attrs == NULL) {
    return lsm_error_failed_alloc;
  }

  new_attrs[entry->attrs.count].type = type;
  new_attrs[entry->attrs.count].str = data;

  entry->attrs.items = new_attrs;
  entry->attrs.count++;
  entry->attrs.bitmap[type / 64] |= ((uint64_t)1) << (type % 64);

  return lsm_error_ok;
}

lsm_error lsm_entry_attr_insert_uint64_t(lsm_entry *entry, uint8_t type,
                                         uint64_t data) {
  lsm_str *s;
  LSM_RES(
      lsm_str_init_copy_n(&s, (char *)&data, sizeof(uint64_t) / sizeof(char)));

  return lsm_entry_attr_insert(entry, type, s);
}

lsm_error lsm_entry_attr_insert_uint8_t(lsm_entry *entry, uint8_t type,
                                        uint8_t data) {
  lsm_str *s;
  LSM_RES(
      lsm_str_init_copy_n(&s, (char *)&data, sizeof(uint8_t) / sizeof(char)));

  return lsm_entry_attr_insert(entry, type, s);
}

uint64_t lsm_entry_data_path_len(const lsm_store *store,
                                 const lsm_entry *entry) {
  const lsm_str *data_path = store->data_path;
  const lsm_str *key = entry->key;

  uint8_t levels =
      key->len <= LSM_STORE_DATA_LEVELS ? key->len : LSM_STORE_DATA_LEVELS;

  return data_path->len + 1 + 2 * levels + key->len +
         strlen(LSM_DATA_FILE_SUFFIX);
}

void lsm_entry_data_path(char *path, const lsm_store *store,
                         const lsm_entry *entry) {
  const lsm_str *data_path = store->data_path;
  const lsm_str *key = entry->key;

  uint8_t levels =
      key->len > LSM_STORE_DATA_LEVELS ? LSM_STORE_DATA_LEVELS : key->len;

  memcpy(path, lsm_str_ptr(data_path), data_path->len);
  path[data_path->len] = '/';

  uint64_t index = data_path->len + 1;

  // Create each directory in the file hierarchy
  // cppcheck-suppress knownConditionTrueFalse
  for (uint8_t i = 0; i < levels; i++) {
    path[index] = lsm_str_char(key, i);
    path[index + 1] = '/';

    index += 2;
  }

  memcpy(&path[index], lsm_str_ptr(key), lsm_str_len(key));

  index += lsm_str_len(key);
  strcpy(&path[index], LSM_DATA_FILE_SUFFIX);
}

lsm_error lsm_entry_data_mkdirs(const lsm_store *store,
                                const lsm_entry *entry) {
  char path[lsm_entry_data_path_len(store, entry) + 1];
  lsm_entry_data_path(path, store, entry);

  const lsm_str *data_path = store->data_path;
  const lsm_str *key = entry->key;

  uint8_t levels =
      key->len <= LSM_STORE_DATA_LEVELS ? key->len : LSM_STORE_DATA_LEVELS;

  // Create all required directories in the path
  // cppcheck-suppress knownConditionTrueFalse
  for (uint8_t i = 0; i < levels; i++) {
    path[data_path->len + 2 * (i + 1)] = '\0';

    if ((mkdir(path, 0755) != 0) && (errno != EEXIST)) {
      return lsm_error_failed_io;
    }

    path[data_path->len + 2 * (i + 1)] = '/';
  }

  return lsm_error_ok;
}

lsm_error lsm_entry_data_open(FILE **out, const lsm_store *store,
                              const lsm_entry *entry, const char *mode) {
  char path[lsm_entry_data_path_len(store, entry) + 1];
  lsm_entry_data_path(path, store, entry);

  FILE *f = fopen(path, mode);

  if (f == NULL) {
    return lsm_error_failed_io;
  }

  if (out != NULL) {
    *out = f;
  }

  return lsm_error_ok;
}

lsm_error lsm_entry_data_remove(const lsm_store *store,
                                const lsm_entry *entry) {
  if (entry->data_len > 0) {
    char data_path[lsm_entry_data_path_len(store, entry) + 1];
    lsm_entry_data_path(data_path, store, entry);

    if (remove(data_path) != 0) {
      return lsm_error_failed_io;
    }
  }

  return lsm_error_ok;
}