Initial implementation of file-backed durable slab storage

src/datapool/datapool.h

@@ -1,12 +1,14 @@
 #pragma once
 
 #include <stddef.h>
-
+#include <stdbool.h>
 struct datapool;
 
-struct datapool *datapool_open(const char *path, size_t size,
-	int *fresh);
+struct datapool *datapool_open(const char *path, const char *user_signature,
+    size_t size, int *fresh, bool prefault);
 void datapool_close(struct datapool *pool);
 
 void *datapool_addr(struct datapool *pool);
 size_t datapool_size(struct datapool *pool);
+void datapool_set_user_data(const struct datapool *pool, const void *user_data, size_t user_size);
+void datapool_get_user_data(const struct datapool *pool, void *user_data, size_t user_size);

src/datapool/datapool_pmem.c

@@ -20,12 +20,18 @@
  * Big enough to fit all necessary metadata, but most of this size is left
  * unused for future expansion.
  */
-#define DATAPOOL_HEADER_LEN 4096
+
+#define DATAPOOL_INTERNAL_HEADER_LEN 2048
+#define DATAPOOL_USER_LAYOUT_LEN       48
+#define DATAPOOL_USER_HEADER_LEN     2048
+#define DATAPOOL_HEADER_LEN (DATAPOOL_INTERNAL_HEADER_LEN + DATAPOOL_USER_HEADER_LEN)
 #define DATAPOOL_VERSION 1
 
 #define DATAPOOL_FLAG_DIRTY (1 << 0)
 #define DATAPOOL_VALID_FLAGS (DATAPOOL_FLAG_DIRTY)
 
+#define PAGE_SIZE 4096
+
 /*
  * Header at the beginning of the file, it's verified every time the pool is
  * opened.
@@ -35,7 +41,10 @@ struct datapool_header {
     uint64_t version;
     uint64_t size;
     uint64_t flags;
-    uint8_t unused[DATAPOOL_HEADER_LEN - 32];
+    uint8_t unused[DATAPOOL_INTERNAL_HEADER_LEN - 32];
+
+    uint8_t user_signature[DATAPOOL_USER_LAYOUT_LEN];
+    uint8_t user_data[DATAPOOL_USER_HEADER_LEN - DATAPOOL_USER_LAYOUT_LEN];
 };
 
 struct datapool {
@@ -62,10 +71,19 @@ datapool_sync(struct datapool *pool)
     ASSERT(ret == 0);
 }
 
+static bool
+datapool_valid_user_signature(struct datapool *pool, const char *user_name)
+{
+    if (cc_strcmp(pool->hdr->user_signature, user_name)) {
+        return false;
+    }
+    return true;
+}
+
 static bool
 datapool_valid(struct datapool *pool)
 {
-    if (memcmp(pool->hdr->signature,
+    if (cc_memcmp(pool->hdr->signature,
           DATAPOOL_SIGNATURE, DATAPOOL_SIGNATURE_LEN) != 0) {
         log_info("no signature found in datapool");
         return false;
@@ -102,22 +120,23 @@ datapool_valid(struct datapool *pool)
 }
 
 static void
-datapool_initialize(struct datapool *pool)
+datapool_initialize(struct datapool *pool, const char *user_name)
 {
     log_info("initializing fresh datapool");
 
     /* 1. clear the header from any leftovers */
-    memset(pool->hdr, 0, DATAPOOL_HEADER_LEN);
+    cc_memset(pool->hdr, 0, DATAPOOL_HEADER_LEN);
     datapool_sync_hdr(pool);
 
     /* 2. fill in the data */
     pool->hdr->version = DATAPOOL_VERSION;
     pool->hdr->size = pool->mapped_len;
     pool->hdr->flags = 0;
+    cc_memcpy(pool->hdr->user_signature, user_name, cc_strlen(user_name));
     datapool_sync_hdr(pool);
 
     /* 3. set the signature */
-    memcpy(pool->hdr->signature, DATAPOOL_SIGNATURE, DATAPOOL_SIGNATURE_LEN);
+    cc_memcpy(pool->hdr->signature, DATAPOOL_SIGNATURE, DATAPOOL_SIGNATURE_LEN);
     datapool_sync_hdr(pool);
 }
 
@@ -143,14 +162,24 @@ datapool_flag_clear(struct datapool *pool, int flag)
  * finish successfully.
  */
 struct datapool *
-datapool_open(const char *path, size_t size, int *fresh)
+datapool_open(const char *path, const char *user_signature, size_t size, int *fresh, bool prefault)
 {
     struct datapool *pool = cc_alloc(sizeof(*pool));
     if (pool == NULL) {
         log_error("unable to create allocate memory for pmem mapping");
         goto err_alloc;
     }
 
+    if (user_signature == NULL) {
+        log_error("empty user signature");
+        goto err_map;
+    }
+
+    if (cc_strnlen(user_signature, DATAPOOL_USER_LAYOUT_LEN) == DATAPOOL_USER_LAYOUT_LEN ) {
+        log_error("user signature is too long %zu", cc_strlen(user_signature));
+        goto err_map;
+    }
+
     size_t map_size = size + sizeof(struct datapool_header);
 
     if (path == NULL) { /* fallback to DRAM if pmem is not configured */
@@ -169,6 +198,15 @@ datapool_open(const char *path, size_t size, int *fresh)
         goto err_map;
     }
 
+    if (prefault) {
+        log_info("prefault datapool");
+        volatile char *cur_addr = pool->addr;
+        char *addr_end = (char *)cur_addr + map_size;
+        for (; cur_addr < addr_end; cur_addr += PAGE_SIZE) {
+            *cur_addr = *cur_addr;
+        }
+    }
+
     log_info("mapped datapool %s with size %llu, is_pmem: %d",
         path, pool->mapped_len, pool->is_pmem);
 
@@ -184,13 +222,23 @@ datapool_open(const char *path, size_t size, int *fresh)
             *fresh = 1;
         }
 
-        datapool_initialize(pool);
+        datapool_initialize(pool, user_signature);
+    } else if (!datapool_valid_user_signature(pool, user_signature)) {
+        log_error("wrong user signature (%s) used for pool", user_signature);
+        goto err_map_adr;
     }
 
     datapool_flag_set(pool, DATAPOOL_FLAG_DIRTY);
 
     return pool;
 
+err_map_adr:
+    if (pool->file_backed) {
+        int ret = pmem_unmap(pool->addr, pool->mapped_len);
+        ASSERT(ret == 0);
+    } else {
+        cc_free(pool->addr);
+    }
 err_map:
     cc_free(pool);
 err_alloc:
@@ -225,3 +273,16 @@ datapool_size(struct datapool *pool)
     return pool->mapped_len - sizeof(struct datapool_header);
 }
 
+void
+datapool_set_user_data(const struct datapool *pool, const void *user_data, size_t user_size)
+{
+    ASSERT(user_size < DATAPOOL_USER_HEADER_LEN - DATAPOOL_USER_LAYOUT_LEN);
+    cc_memcpy(pool->hdr->user_data, user_data, user_size);
+}
+
+void
+datapool_get_user_data(const struct datapool *pool, void *user_data, size_t user_size)
+{
+    ASSERT(user_size < DATAPOOL_USER_HEADER_LEN - DATAPOOL_USER_LAYOUT_LEN);
+    cc_memcpy(user_data, pool->hdr->user_data, user_size);
+}

src/datapool/datapool_shm.c

@@ -8,7 +8,7 @@
 #include <cc_mm.h>
 
 struct datapool *
-datapool_open(const char *path, size_t size, int *fresh)
+datapool_open(const char *path, const char *user_signature, size_t size, int *fresh, bool prefault)
 {
     if (path != NULL) {
         log_warn("attempted to open a file-based data pool without"
@@ -41,3 +41,19 @@ datapool_size(struct datapool *pool)
     return cc_alloc_usable_size(pool);
 }
 
+/*
+ * NOTE: Abstraction in datapool required defining functions below
+ *       datapool_get_user_data is currently used only in in pmem implementation
+ *       datapool_set_user_data is called during teardown e.g. slab
+ */
+void
+datapool_set_user_data(const struct datapool *pool, const void *user_data, size_t user_size)
+{
+
+}
+
+void
+datapool_get_user_data(const struct datapool *pool, void *user_data, size_t user_size)
+{
+    NOT_REACHED();
+}

src/storage/cuckoo/cuckoo.c

@@ -283,7 +283,8 @@ cuckoo_setup(cuckoo_options_st *options, cuckoo_metrics_st *metrics)
 
     hash_size = item_size * max_nitem;
     pool = datapool_open(option_str(&options->cuckoo_datapool),
-        hash_size, NULL);
+        option_str(&options->cuckoo_datapool_name), hash_size,
+        NULL, option_bool(&options->cuckoo_datapool_prefault));
     if (pool == NULL) {
         log_crit("cuckoo data store allocation failed");
         exit(EX_CONFIG);

src/storage/cuckoo/cuckoo.h

@@ -18,16 +18,21 @@
 #define CUCKOO_POLICY CUCKOO_POLICY_RANDOM
 #define CUCKOO_MAX_TTL (30 * 24 * 60 * 60) /* 30 days */
 #define CUCKOO_DATAPOOL NULL
+#define CUCKOO_DATAPOOL_NAME "cuckoo_datapool"
+#define CUCKOO_PREFAULT false
+
+/*          name                      type                default                  description */
+#define CUCKOO_OPTION(ACTION)                                                                          \
+    ACTION( cuckoo_displace,          OPTION_TYPE_UINT,   CUCKOO_DISPLACE,         "# displaces allowed"   )\
+    ACTION( cuckoo_item_cas,          OPTION_TYPE_BOOL,   CUCKOO_ITEM_CAS,         "support cas in items"  )\
+    ACTION( cuckoo_item_size,         OPTION_TYPE_UINT,   CUCKOO_ITEM_SIZE,        "item size (inclusive)" )\
+    ACTION( cuckoo_nitem,             OPTION_TYPE_UINT,   CUCKOO_NITEM,            "# items allocated"     )\
+    ACTION( cuckoo_policy,            OPTION_TYPE_UINT,   CUCKOO_POLICY,           "evict policy"          )\
+    ACTION( cuckoo_max_ttl,           OPTION_TYPE_UINT,   CUCKOO_MAX_TTL,          "max ttl in seconds"    )\
+    ACTION( cuckoo_datapool,          OPTION_TYPE_STR,    CUCKOO_DATAPOOL,         "path to data pool"     )\
+    ACTION( cuckoo_datapool_name,     OPTION_TYPE_STR,    CUCKOO_DATAPOOL_NAME,    "cuckoo datapool name"  )\
+    ACTION( cuckoo_datapool_prefault, OPTION_TYPE_BOOL,   CUCKOO_PREFAULT,         "prefault data pool"    )
 
-/*          name                type                default             description */
-#define CUCKOO_OPTION(ACTION)                                                                    \
-    ACTION( cuckoo_displace,    OPTION_TYPE_UINT,   CUCKOO_DISPLACE,    "# displaces allowed"   )\
-    ACTION( cuckoo_item_cas,    OPTION_TYPE_BOOL,   CUCKOO_ITEM_CAS,    "support cas in items"  )\
-    ACTION( cuckoo_item_size,   OPTION_TYPE_UINT,   CUCKOO_ITEM_SIZE,   "item size (inclusive)" )\
-    ACTION( cuckoo_nitem,       OPTION_TYPE_UINT,   CUCKOO_NITEM,       "# items allocated"     )\
-    ACTION( cuckoo_policy,      OPTION_TYPE_UINT,   CUCKOO_POLICY,      "evict policy"          )\
-    ACTION( cuckoo_max_ttl,     OPTION_TYPE_UINT,   CUCKOO_MAX_TTL,     "max ttl in seconds"    )\
-    ACTION( cuckoo_datapool,    OPTION_TYPE_STR,    CUCKOO_DATAPOOL,    "path to data pool"     )
 
 typedef struct {
     CUCKOO_OPTION(OPTION_DECLARE)

src/storage/slab/CMakeLists.txt

@@ -4,3 +4,4 @@ set(SOURCE
     slab.c)
 
 add_library(slab ${SOURCE})
+target_link_libraries(slab datapool)

src/storage/slab/item.c

@@ -102,21 +102,24 @@ _item_dealloc(struct item **it_p)
 }
 
 /*
- * Link an item into the hash table
+ * (Re)Link an item into the hash table
  */
 static void
-_item_link(struct item *it)
+_item_link(struct item *it, bool relink)
 {
     ASSERT(it->magic == ITEM_MAGIC);
-    ASSERT(!(it->is_linked));
     ASSERT(!(it->in_freeq));
 
+    if (!relink) {
+        ASSERT(!(it->is_linked));
+
+        it->is_linked = 1;
+        slab_deref(item_to_slab(it)); /* slab ref'ed in _item_alloc */
+    }
+
     log_verb("link it %p of id %"PRIu8" at offset %"PRIu32, it, it->id,
             it->offset);
 
-    it->is_linked = 1;
-    slab_deref(item_to_slab(it)); /* slab ref'ed in _item_alloc */
-
     hashtable_put(it, hash_table);
 
     INCR(slab_metrics, item_linked_curr);
@@ -128,14 +131,20 @@ _item_link(struct item *it)
     PERSLAB_INCR_N(it->id, item_val_byte, it->vlen);
 }
 
+void
+item_relink(struct item *it)
+{
+    _item_link(it, true);
+}
+
 void
 item_insert(struct item *it, const struct bstring *key)
 {
     ASSERT(it != NULL && key != NULL);
 
     item_delete(key);
 
-    _item_link(it);
+    _item_link(it, false);
     log_verb("insert it %p of id %"PRIu8" for key %.*s", it, it->id, key->len,
         key->data);
 }

src/storage/slab/item.h

@@ -210,5 +210,8 @@ void item_update(struct item *it, const struct bstring *val);
 /* Remove item from cache */
 bool item_delete(const struct bstring *key);
 
+/* Relink item */
+void item_relink(struct item *it);
+
 /* flush the cache */
 void item_flush(void);