storage-cfs.c

Go to the documentation of this file.

/*
 * Copyright (c) 2010, Swedish Institute of Computer Science
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. Neither the name of the Institute nor the names of its contributors
 *    may be used to endorse or promote products derived from this software
 *    without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE INSTITUTE AND CONTRIBUTORS ``AS IS'' AND
 * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
 * ARE DISCLAIMED.  IN NO EVENT SHALL THE INSTITUTE OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
 * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
 * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
 * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
 * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
 * SUCH DAMAGE.
 */
 
/**
 * \file
 *      Contiki File System (CFS) backend for the storage abstraction
 *      used by the database.
 * \author
 *      Nicolas Tsiftes <nvt@sics.se>
 */
 
#include <stdio.h>
#include <string.h>
 
#include "cfs/cfs.h"
#include "cfs/cfs-coffee.h"
#include "lib/random.h"
 
#define DEBUG DEBUG_NONE
#include "net/ipv6/uip-debug.h"
 
#include "db-options.h"
#include "storage.h"
 
struct attribute_record {
  char name[ATTRIBUTE_NAME_LENGTH];
  uint8_t domain;
  uint8_t element_size;
};
 
struct index_record {
  char attribute_name[ATTRIBUTE_NAME_LENGTH];
  char file_name[DB_MAX_FILENAME_LENGTH];
  uint8_t type;
};
 
#if DB_FEATURE_COFFEE
#define DB_COFFEE_CATALOG_SIZE RELATION_NAME_LENGTH +                     \
                               (DB_MAX_ATTRIBUTES_PER_RELATION *          \
                               sizeof(struct attribute_record))
#endif
 
#define ROW_XOR 0xf6U
 
static bool
merge_strings(char *dest, size_t dest_size, char *prefix, char *suffix)
{
  size_t prefix_len = strlen(prefix);
  size_t suffix_len = strlen(suffix);
 
  if(dest_size < prefix_len + suffix_len + 1) {
    return false;
  }
 
  memcpy(dest, prefix, prefix_len);
  memcpy(dest + prefix_len, suffix, suffix_len);
  dest[prefix_len + suffix_len] = '\0';
 
  return true;
}
 
char *
storage_generate_file(char *prefix, unsigned long size)
{
  static char filename[ATTRIBUTE_NAME_LENGTH + sizeof(".ffff")];
#if !DB_FEATURE_COFFEE
  int fd;
#endif
 
  snprintf(filename, sizeof(filename), "%s.%x", prefix,
           (unsigned)(random_rand() & 0xffff));
 
#if DB_FEATURE_COFFEE
  PRINTF("DB: Reserving %lu bytes in %s\n", size, filename);
  if(cfs_coffee_reserve(filename, size) < 0) {
    PRINTF("DB: Failed to reserve\n");
    return NULL;
  }
  return filename;
#else
  fd = cfs_open(filename, CFS_WRITE);
  cfs_close(fd);
  return fd < 0 ? NULL : filename;
#endif /* DB_FEATURE_COFFEE */
}
 
db_result_t
storage_load(relation_t *rel)
{
  PRINTF("DB: Opening the tuple file %s\n", rel->tuple_filename);
  rel->tuple_storage = cfs_open(rel->tuple_filename,
                                CFS_READ | CFS_WRITE | CFS_APPEND);
  if(rel->tuple_storage < 0) {
    PRINTF("DB: Failed to open the tuple file\n");
    return DB_STORAGE_ERROR;
  }
 
  return DB_OK;
}
 
void
storage_unload(relation_t *rel)
{
  if(RELATION_HAS_TUPLES(rel)) {
    PRINTF("DB: Unload tuple file %s\n", rel->tuple_filename);
 
    cfs_close(rel->tuple_storage);
    rel->tuple_storage = -1;
  }
}
 
db_result_t
storage_get_relation(relation_t *rel, char *name)
{
  int fd;
  int r;
  int i;
  struct attribute_record record;
  db_result_t result;
 
  fd = cfs_open(name, CFS_READ);
  if(fd < 0) {
    return DB_STORAGE_ERROR;
  }
 
  r = cfs_read(fd, rel->name, sizeof(rel->name));
  if(r != sizeof(rel->name)) {
    cfs_close(fd);
    PRINTF("DB: Failed to read name, got %d of %d bytes\n",
           r, sizeof(rel->name));
    return DB_STORAGE_ERROR;
  }
 
  r = cfs_read(fd, rel->tuple_filename, sizeof(rel->tuple_filename));
  if(r != sizeof(rel->name)) {
    cfs_close(fd);
    PRINTF("DB: Failed to read tuple filename\n");
    return DB_STORAGE_ERROR;
  }
 
  rel->tuple_filename[sizeof(rel->tuple_filename) - 1] ^= ROW_XOR;
 
  /* Read attribute records. */
  result = DB_OK;
  for(i = 0;; i++) {
    r = cfs_read(fd, &record, sizeof(record));
    if(r == 0) {
      break;
    }
    if(r != sizeof(record)) {
      PRINTF("DB: Failed to read attribute record %d (r = %d)\n", i, r);
      result = DB_STORAGE_ERROR;
      break;
    }
 
    if(relation_attribute_add(rel, DB_MEMORY, record.name,
                              record.domain, record.element_size) == NULL) {
      PRINTF("DB: Failed to add the attribute %s\n", record.name);
      result = DB_STORAGE_ERROR;
      break;
    }
  }
 
  (void)i;
  PRINTF("DB: Read %d attributes\n", i);
 
  cfs_close(fd);
  return result;
}
 
db_result_t
storage_put_relation(relation_t *rel)
{
  int fd;
  int r;
  char *str;
  unsigned char *last_byte;
 
  PRINTF("DB: put_relation(%s)\n", rel->name);
 
  cfs_remove(rel->name);
 
#if DB_FEATURE_COFFEE
  cfs_coffee_reserve(rel->name, DB_COFFEE_CATALOG_SIZE);
#endif
 
  fd = cfs_open(rel->name, CFS_WRITE | CFS_READ);
  if(fd < 0) {
    return DB_STORAGE_ERROR;
  }
 
  r = cfs_write(fd, rel->name, sizeof(rel->name));
  if(r != sizeof(rel->name)) {
    cfs_close(fd);
    cfs_remove(rel->name);
    return DB_STORAGE_ERROR;
  }
 
  if(rel->tuple_filename[0] == '\0') {
    str = storage_generate_file("tuple", DB_COFFEE_RESERVE_SIZE);
    if(str == NULL) {
      cfs_close(fd);
      cfs_remove(rel->name);
      return DB_STORAGE_ERROR;
    }
 
    strncpy(rel->tuple_filename, str, sizeof(rel->tuple_filename) - 1);
    rel->tuple_filename[sizeof(rel->tuple_filename) - 1] = '\0';
  }
 
  /*
   * Encode the last byte to ensure that the filename is not
   * null-terminated. This will make the Coffee FS determine
   * the correct length when re-opening the file.
   */
  last_byte = (unsigned char *)&rel->tuple_filename[sizeof(rel->tuple_filename) - 1];
  *last_byte ^= ROW_XOR;
 
  r = cfs_write(fd, rel->tuple_filename, sizeof(rel->tuple_filename));
 
  *last_byte ^= ROW_XOR;
 
  if(r != sizeof(rel->tuple_filename)) {
    cfs_close(fd);
    cfs_remove(rel->tuple_filename);
    return DB_STORAGE_ERROR;
  }
 
  PRINTF("DB: Saved relation %s\n", rel->name);
 
  cfs_close(fd);
  return DB_OK;
}
 
db_result_t
storage_put_attribute(relation_t *rel, attribute_t *attr)
{
  int fd;
  struct attribute_record record;
  int r;
 
  PRINTF("DB: put_attribute(%s, %s)\n", rel->name, attr->name);
 
  fd = cfs_open(rel->name, CFS_WRITE | CFS_APPEND);
  if(fd < 0) {
    return DB_STORAGE_ERROR;
  }
 
  memset(&record.name, 0, sizeof(record.name));
  memcpy(record.name, attr->name, sizeof(record.name));
  record.domain = attr->domain;
  record.element_size = attr->element_size;
  r = cfs_write(fd, &record, sizeof(record));
  if(r != sizeof(record)) {
    cfs_close(fd);
    cfs_remove(rel->name);
    return DB_STORAGE_ERROR;
  }
 
  cfs_close(fd);
  return DB_OK;
}
 
db_result_t
storage_drop_relation(relation_t *rel, int remove_tuples)
{
  if(remove_tuples && RELATION_HAS_TUPLES(rel)) {
    cfs_remove(rel->tuple_filename);
  }
  return cfs_remove(rel->name) < 0 ? DB_STORAGE_ERROR : DB_OK;
}
 
#if DB_FEATURE_REMOVE
db_result_t
storage_rename_relation(char *old_name, char *new_name)
{
  db_result_t result;
  int old_fd;
  int new_fd;
  int r;
  char buf[64];
 
  result = DB_STORAGE_ERROR;
  old_fd = new_fd = -1;
 
  old_fd = cfs_open(old_name, CFS_READ);
  new_fd = cfs_open(new_name, CFS_WRITE);
  if(old_fd < 0 || new_fd < 0) {
    goto error;
  }
 
  for(;;) {
    r = cfs_read(old_fd, buf, sizeof(buf));
    if(r < 0) {
      goto error;
    } else if(r == 0) {
      break;
    }
    if(cfs_write(new_fd, buf, r) != r) {
      goto error;
    }
  };
 
  cfs_remove(old_name);
  result = DB_OK;
 
error:
  cfs_close(old_fd);
  cfs_close(new_fd);
 
  if(result != DB_OK) {
    cfs_remove(new_name);
  }
  return result;
}
#endif /* DB_FEATURE_REMOVE */
 
db_result_t
storage_get_index(index_t *index, relation_t *rel, attribute_t *attr)
{
  char filename[INDEX_NAME_LENGTH + 1];
  int fd;
  int r;
  struct index_record record;
  db_result_t result;
 
  if(!merge_strings(filename, sizeof(filename), rel->name, INDEX_NAME_SUFFIX)) {
    return DB_NAME_ERROR;
  }
 
  fd = cfs_open(filename, CFS_READ);
  if(fd < 0) {
    return DB_STORAGE_ERROR;
  }
 
  for(result = DB_STORAGE_ERROR;;) {
    r = cfs_read(fd, &record, sizeof(record));
    if(r < sizeof(record)) {
      break;
    }
    if(strcmp(attr->name, record.attribute_name) == 0) {
      PRINTF("DB: Found the index record for %s.%s: type %d, filename %s\n",
        rel->name, attr->name, record.type, record.file_name);
      index->type = record.type;
      memcpy(index->descriptor_file, record.file_name,
             sizeof(index->descriptor_file));
      result = DB_OK;
    }
  }
 
  cfs_close(fd);
 
  return result;
}
 
db_result_t
storage_put_index(index_t *index)
{
  char filename[INDEX_NAME_LENGTH + 1];
  int fd;
  int r;
  struct index_record record;
  db_result_t result;
 
  if(!merge_strings(filename, sizeof(filename), index->rel->name,
                    INDEX_NAME_SUFFIX)) {
    return DB_NAME_ERROR;
  }
 
  fd = cfs_open(filename, CFS_WRITE | CFS_APPEND);
  if(fd < 0) {
    return DB_STORAGE_ERROR;
  }
 
  strcpy(record.attribute_name, index->attr->name);
  memcpy(record.file_name, index->descriptor_file, sizeof(record.file_name));
  record.type = index->type;
 
  result = DB_OK;
  r = cfs_write(fd, &record, sizeof(record));
  if(r < sizeof(record)) {
    result = DB_STORAGE_ERROR;
  } else {
    PRINTF("DB: Wrote an index record for %s.%s, type %d\n",
      index->rel->name, index->attr->name, record.type);
  }
 
  cfs_close(fd);
 
  return result;
}
 
db_result_t
storage_get_row(relation_t *rel, tuple_id_t *tuple_id, storage_row_t row)
{
  int r;
  tuple_id_t nrows;
 
  if(DB_ERROR(storage_get_row_amount(rel, &nrows))) {
    return DB_STORAGE_ERROR;
  }
 
  if(*tuple_id >= nrows) {
    return DB_FINISHED;
  }
 
  if(cfs_seek(rel->tuple_storage, *tuple_id * rel->row_length, CFS_SEEK_SET) ==
              (cfs_offset_t)-1) {
    return DB_STORAGE_ERROR;
  }
 
  r = cfs_read(rel->tuple_storage, row, rel->row_length);
  if(r < 0) {
    PRINTF("DB: Reading failed on fd %d\n", rel->tuple_storage);
    return DB_STORAGE_ERROR;
  } else if(r == 0) {
    return DB_FINISHED;
  } else if(r < rel->row_length) {
    PRINTF("DB: Incomplete record: %d < %d\n", r, rel->row_length);
    return DB_STORAGE_ERROR;
  }
 
  row[rel->row_length - 1] ^= ROW_XOR;
 
  PRINTF("DB: Read %d bytes from relation %s\n", rel->row_length, rel->name);
 
  return DB_OK;
}
 
db_result_t
storage_put_row(relation_t *rel, storage_row_t row)
{
  cfs_offset_t end;
  unsigned remaining;
  int r;
  unsigned char *last_byte;
#if DB_FEATURE_INTEGRITY
  int missing_bytes;
  char buf[rel->row_length];
#endif
 
  end = cfs_seek(rel->tuple_storage, 0, CFS_SEEK_END);
  if(end == (cfs_offset_t)-1) {
    return DB_STORAGE_ERROR;
  }
 
#if DB_FEATURE_INTEGRITY
  missing_bytes = end % rel->row_length;
  if(missing_bytes > 0) {
    memset(buf, 0xff, sizeof(buf));
    r = cfs_write(rel->tuple_storage, buf, sizeof(buf));
    if(r != missing_bytes) {
      return DB_STORAGE_ERROR;
    }
  }
#endif
 
  /* Ensure that last written byte is separated from 0, to make file
     lengths correct in Coffee. */
  last_byte = row + rel->row_length - 1;
  *last_byte ^= ROW_XOR;
 
  remaining = rel->row_length;
  do {
    r = cfs_write(rel->tuple_storage, row, remaining);
    if(r < 0) {
      PRINTF("DB: Failed to store %u bytes\n", remaining);
      *last_byte ^= ROW_XOR;
      return DB_STORAGE_ERROR;
    }
    row += r;
    remaining -= r;
  } while(remaining > 0);
 
  PRINTF("DB: Stored a of %d bytes\n", rel->row_length);
 
  *last_byte ^= ROW_XOR;
 
  return DB_OK;
}
 
db_result_t
storage_get_row_amount(relation_t *rel, tuple_id_t *amount)
{
  cfs_offset_t offset;
 
  if(rel->row_length == 0) {
    *amount = 0;
  } else {
    offset = cfs_seek(rel->tuple_storage, 0, CFS_SEEK_END);
    if(offset == (cfs_offset_t)-1) {
      return DB_STORAGE_ERROR;
    }
 
    *amount = (tuple_id_t)(offset / rel->row_length);
  }
 
  return DB_OK;
}
 
db_storage_id_t
storage_open(const char *filename)
{
  int fd;
 
  fd = cfs_open(filename, CFS_WRITE | CFS_READ);
#if DB_FEATURE_COFFEE
  if(fd >= 0) {
    cfs_coffee_set_io_semantics(fd, CFS_COFFEE_IO_FLASH_AWARE);
  }
#endif
  return fd;
}
 
void
storage_close(db_storage_id_t fd)
{
  cfs_close(fd);
}
 
db_result_t
storage_read(db_storage_id_t fd,
             void *buffer, unsigned long offset, unsigned length)
{
  char *ptr;
  int r;
 
  /* Extend the file if necessary, so that previously unwritten bytes
     will be read in as zeroes. */
  if(cfs_seek(fd, offset + length, CFS_SEEK_SET) == (cfs_offset_t)-1) {
    return DB_STORAGE_ERROR;
  }
 
  if(cfs_seek(fd, offset, CFS_SEEK_SET) == (cfs_offset_t)-1) {
    return DB_STORAGE_ERROR;
  }
 
  ptr = buffer;
  while(length > 0) {
    r = cfs_read(fd, ptr, length);
    if(r <= 0) {
      return DB_STORAGE_ERROR;
    }
    ptr += r;
    length -= r;
  }
 
  return DB_OK;
}
 
db_result_t
storage_write(db_storage_id_t fd,
              void *buffer, unsigned long offset, unsigned length)
{
  char *ptr;
  int r;
 
  if(cfs_seek(fd, offset, CFS_SEEK_SET) == (cfs_offset_t)-1) {
    return DB_STORAGE_ERROR;
  }
 
  ptr = buffer;
  while(length > 0) {
    r = cfs_write(fd, ptr, length);
    if(r <= 0) {
      return DB_STORAGE_ERROR;
    }
    ptr += r;
    length -= r;
  }
 
  return DB_OK;
}