Recreate slab lruq

- divide datapool metadata header to internal and user space
- update next free item in slab after recreate
- extend cc_queue operations to support pointer arithmetics

deps/ccommon/include/cc_queue.h

@@ -127,6 +127,7 @@ extern "C" {
  * _INSERT_TAIL         -       -       +       +       +
  * _REMOVE_HEAD         +       -       +       -       -
  * _REMOVE              +       +       +       +       +
+ * _REINIT              -       -       -       +       -
  *
  */
 
@@ -691,6 +692,23 @@ struct {                                                                \
         (head2)->tqh_last = &(head2)->tqh_first;                        \
 } while (0)
 
+#define TAILQ_REINIT(head, var, field, offset) do {                     \
+    TAILQ_FIRST((head)) = var;                                          \
+    (head)->tqh_last = NULL;                                            \
+    TAILQ_FOREACH(var, head, s_tqe) {                                   \
+        if ((TAILQ_NEXT((var), field)) != NULL) {                       \
+            TAILQ_NEXT((var), field) =                                  \
+                (void *)((char *)(TAILQ_NEXT((var), field)) + (offset));\
+        }                                                               \
+        if ((var) == TAILQ_FIRST(head)) {                               \
+            (var)->field.tqe_prev = &TAILQ_FIRST(head);                 \
+        } else {                                                        \
+            (var)->field.tqe_prev =                                     \
+                (void *)((char *)((var)->field.tqe_prev) + (offset));   \
+        }                                                               \
+    }                                                                   \
+} while (0)
+
 /*
  * Circular queue declarations.
  */

src/datapool/datapool.h

@@ -10,3 +10,5 @@ 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,7 +20,10 @@
  * 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_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)
@@ -31,11 +34,16 @@
  * opened.
  */
 struct datapool_header {
-    uint8_t signature[DATAPOOL_SIGNATURE_LEN];
-    uint64_t version;
-    uint64_t size;
-    uint64_t flags;
-    uint8_t unused[DATAPOOL_HEADER_LEN - 32];
+    struct {
+        uint8_t data[DATAPOOL_USER_HEADER_LEN];
+    } user;
+    struct {
+        uint8_t signature[DATAPOOL_SIGNATURE_LEN];
+        uint64_t version;
+        uint64_t size;
+        uint64_t flags;
+        uint8_t unused[DATAPOOL_INTERNAL_HEADER_LEN - 32];
+    } internal;
 };
 
 struct datapool {
@@ -65,35 +73,35 @@ datapool_sync(struct datapool *pool)
 static bool
 datapool_valid(struct datapool *pool)
 {
-    if (memcmp(pool->hdr->signature,
+    if (cc_memcmp(pool->hdr->internal.signature,
           DATAPOOL_SIGNATURE, DATAPOOL_SIGNATURE_LEN) != 0) {
         log_info("no signature found in datapool");
         return false;
     }
 
-    if (pool->hdr->version != DATAPOOL_VERSION) {
+    if (pool->hdr->internal.version != DATAPOOL_VERSION) {
         log_info("incompatible datapool version (is: %d, expecting: %d)",
-            pool->hdr->version, DATAPOOL_SIGNATURE);
+            pool->hdr->internal.version, DATAPOOL_SIGNATURE);
         return false;
     }
 
-    if (pool->hdr->size == 0) {
+    if (pool->hdr->internal.size == 0) {
         log_error("datapool has 0 size");
         return false;
     }
 
-    if (pool->hdr->size > pool->mapped_len) {
+    if (pool->hdr->internal.size > pool->mapped_len) {
         log_error("datapool has invalid size (is: %d, expecting: %d)",
-            pool->mapped_len, pool->hdr->size);
+            pool->mapped_len, pool->hdr->internal.size);
         return false;
     }
 
-    if (pool->hdr->flags & ~DATAPOOL_VALID_FLAGS) {
+    if (pool->hdr->internal.flags & ~DATAPOOL_VALID_FLAGS) {
         log_error("datapool has invalid flags set");
         return false;
     }
 
-    if (pool->hdr->flags & DATAPOOL_FLAG_DIRTY) {
+    if (pool->hdr->internal.flags & DATAPOOL_FLAG_DIRTY) {
         log_info("datapool has a valid header but is dirty");
         return false;
     }
@@ -107,31 +115,31 @@ datapool_initialize(struct datapool *pool)
     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;
+    pool->hdr->internal.version = DATAPOOL_VERSION;
+    pool->hdr->internal.size = pool->mapped_len;
+    pool->hdr->internal.flags = 0;
     datapool_sync_hdr(pool);
 
     /* 3. set the signature */
-    memcpy(pool->hdr->signature, DATAPOOL_SIGNATURE, DATAPOOL_SIGNATURE_LEN);
+    cc_memcpy(pool->hdr->internal.signature, DATAPOOL_SIGNATURE, DATAPOOL_SIGNATURE_LEN);
     datapool_sync_hdr(pool);
 }
 
 static void
 datapool_flag_set(struct datapool *pool, int flag)
 {
-    pool->hdr->flags |= flag;
+    pool->hdr->internal.flags |= flag;
     datapool_sync_hdr(pool);
 }
 
 static void
 datapool_flag_clear(struct datapool *pool, int flag)
 {
-    pool->hdr->flags &= ~flag;
+    pool->hdr->internal.flags &= ~flag;
     datapool_sync_hdr(pool);
 }
 
@@ -225,3 +233,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);
+    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);
+    cc_memcpy(user_data, pool->hdr->user.data, user_size);
+}

src/datapool/datapool_shm.c

@@ -41,3 +41,14 @@ datapool_size(struct datapool *pool)
     return cc_alloc_usable_size(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)
+{
+    NOT_REACHED();
+}

src/storage/slab/slab.c

@@ -25,6 +25,11 @@ struct slab_heapinfo {
     struct slab_tqh slab_lruq;   /* lru slab q */
 };
 
+struct slab_pool_metadata {
+    void        *usr_pool_addr; /* previous pool address */
+    struct slab *slab_lruq_head;/* lru slab q head */
+};
+
 static struct datapool *pool_slab;              /* data pool mapping for the slabs */
 static int pool_slab_state;                     /* data pool state */
 perslab_metrics_st perslab[SLABCLASS_MAX_ID];
@@ -163,16 +168,19 @@ _slab_recreate_items(struct slab *slab)
     uint32_t i;
 
     p = &slabclass[slab->id];
-
+    p->nfree_item = p->nitem;
+    p->next_item_in_slab = (struct item *)&slab->data[0];
     for (i = 0; i < p->nitem; i++) {
         it = _slab_to_item(slab, i, p->size);
         if (it->is_linked) {
             INCR(slab_metrics, item_curr);
             INCR(slab_metrics, item_alloc);
             PERSLAB_INCR(slab->id, item_curr);
             item_relink(it);
+            p->nfree_item--;
+            p->next_item_in_slab = (struct item *)((char *)p->next_item_in_slab + p->size);
         } else if (it->in_freeq) {
-            _slab_put_item_into_freeq(it,slab->id);
+            _slab_put_item_into_freeq(it, slab->id);
         }
     }
 }
@@ -189,16 +197,37 @@ _slab_table_update(struct slab *slab)
               heapinfo.nslab - 1);
 }
 
+/*
+ * Recreate lru slab q
+ * After restart mapping from datapool could be different
+ * and all addresses must be updated
+ */
+static void
+_slab_lruq_rebuild(char *heap_start)
+{
+    struct slab_pool_metadata heap_metadata;
+
+    datapool_get_user_data(pool_slab, &heap_metadata, sizeof(struct slab_pool_metadata));
+
+    ptrdiff_t offset = heap_start - (char *)heap_metadata.usr_pool_addr;
+
+    struct slab *slab = (void *)((char *)heap_metadata.slab_lruq_head + offset);
+
+    TAILQ_REINIT(&heapinfo.slab_lruq, slab, s_tqe, offset);
+}
+
 /*
  * Recreate slabs structure when persistent memory features are enabled (USE_PMEM)
  */
 static void
 _slab_recovery(void)
 {
     uint32_t i;
-    uint8_t * heap_start = datapool_addr(pool_slab);
-    /* TODO: recreate heapinfo.slab_lruq */
-    for(i = 0; i < heapinfo.max_nslab; i++) {
+
+    char *heap_start = datapool_addr(pool_slab);
+
+    _slab_lruq_rebuild(heap_start);
+    for (i = 0; i < heapinfo.max_nslab; i++) {
         struct slab *slab = (struct slab *) heap_start;
         if (slab->initialized) {
             INCR(slab_metrics, slab_req);
@@ -322,6 +351,13 @@ _slab_heapinfo_setup(void)
 static void
 _slab_heapinfo_teardown(void)
 {
+    struct slab_pool_metadata pool_metadata =
+    {
+        datapool_addr(pool_slab),
+        TAILQ_FIRST(&heapinfo.slab_lruq)
+    };
+
+    datapool_set_user_data(pool_slab, &pool_metadata, sizeof(struct slab_pool_metadata));
     datapool_close(pool_slab);
     pool_slab = NULL;
 }