crash fixes

src/rdb.c

@@ -1291,12 +1291,10 @@ int rdbSaveBackground(char *filename, rdbSaveInfo *rsi) {
     openChildInfoPipe();
 
     start = ustime();
-    handle_prefork();
     if ((childpid = fork()) == 0) {
         int retval;
 
         /* Child */
-        handle_postfork(0);
         closeListeningSockets(0);
         redisSetProcTitle("redis-rdb-bgsave");
         retval = rdbSave(filename,rsi);
@@ -1315,7 +1313,6 @@ int rdbSaveBackground(char *filename, rdbSaveInfo *rsi) {
         exitFromChild((retval == C_OK) ? 0 : 1);
     } else {
         /* Parent */
-        handle_postfork(childpid);
         server.stat_fork_time = ustime()-start;
         server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
         latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);

src/sdsalloc.h

@@ -38,9 +38,6 @@
 
 #include "zmalloc.h"
 #include "storage.h"
-//#define s_malloc zmalloc
-#define s_malloc salloc
-//#define s_realloc zrealloc
-#define s_realloc srealloc
-//#define s_free zfree
-#define s_free sfree
+#define s_malloc zmalloc
+#define s_realloc zrealloc
+#define s_free zfree

src/storage.c

@@ -12,30 +12,41 @@
 struct memkind *mkdisk = NULL;
 static char *PMEM_DIR = "/mnt/btrfs_scratch/";
 
-
+void handle_prefork();
+void handle_postfork_parent();
+void handle_postfork_child();
 
 #define OBJECT_PAGE_BUFFER_SIZE 8192 //(size in objs)
+#define OBJ_PAGE_BITS_PER_WORD 64
 struct object_page
 {
     uint64_t allocmap[OBJECT_PAGE_BUFFER_SIZE/(8*sizeof(uint64_t))];
-    robj rgobj[OBJECT_PAGE_BUFFER_SIZE];
     struct object_page *pnext;
+    char rgb[];
 };
-#define OBJ_PAGE_BITS_PER_WORD 64
-struct object_page *headObjpage = NULL;
 
-void storage_init()
+struct alloc_pool
 {
-    int errv = memkind_create_pmem(PMEM_DIR, 0, &mkdisk);
-    if (errv)
-    {
-        fprintf(stderr, "Memory pool creation failed: %d\n", errv);
-        exit(EXIT_FAILURE);
-    }
-    headObjpage = memkind_calloc(MEMKIND_HUGETLB, 1, sizeof(struct object_page));
-}
+    unsigned cbObject;
+    struct object_page *pobjpageHead;
+};
 
-int IdxAllocObject(struct object_page *page)
+
+struct object_page *pool_allocate_page(int cbObject)
+{
+    size_t cb = (((size_t)cbObject) * OBJECT_PAGE_BUFFER_SIZE) + sizeof(struct object_page);
+    return scalloc(cb, 1, MALLOC_SHARED);
+}
+void pool_initialize(struct alloc_pool *ppool, int cbObject)
+{
+    if ((cbObject % 8) != 0)
+    {
+        cbObject += 8 - (cbObject % 8);
+    }
+    ppool->cbObject = cbObject;
+    ppool->pobjpageHead = pool_allocate_page(cbObject);
+}
+static int IdxAllocObject(struct object_page *page)
 {
     for (size_t iword = 0; iword < OBJ_PAGE_BITS_PER_WORD; ++iword)
     {
@@ -48,49 +59,93 @@ int IdxAllocObject(struct object_page *page)
                 bitword >>= 1;
                 ++ibit;
             }
-            page->allocmap[iword] |= 1 << ibit;
+            page->allocmap[iword] |= 1ULL << ibit;
             return (iword * OBJ_PAGE_BITS_PER_WORD) + ibit;
         }
     }
     return -1;
 }
-
-struct redisObject *salloc_obj()
+void *pool_alloc(struct alloc_pool *ppool)
 {
-    struct object_page *cur = headObjpage;
+    struct object_page *cur = ppool->pobjpageHead;
     for (;;)
     {
         int idx = IdxAllocObject(cur);
         if (idx >= 0)
         {
-            return &cur->rgobj[idx];
+            return cur->rgb + (((size_t)ppool->cbObject) * idx);
         }
 
         if (cur->pnext == NULL)
         {
-            cur->pnext = memkind_calloc(MEMKIND_HUGETLB, 1, sizeof(struct object_page));
+            cur->pnext = pool_allocate_page(ppool->cbObject);
         }
 
         cur = cur->pnext;
     }
 }
-void sfree_obj(struct redisObject *obj)
+
+#pragma weak serverLog
+void serverLog(int level, const char*fmt, ...){}
+
+void pool_free(struct alloc_pool *ppool, void *pv)
 {
-    struct object_page *cur = headObjpage;
-    for (;;)
+    struct object_page *cur = ppool->pobjpageHead;
+    char *obj = pv;
+
+    for (;cur != NULL;)
     {
-        if (obj >= cur->rgobj && (obj < (cur->rgobj + OBJECT_PAGE_BUFFER_SIZE)))
+        if (obj >= cur->rgb && (obj < (cur->rgb + (OBJECT_PAGE_BUFFER_SIZE * ppool->cbObject))))
         {
             // Its on this page
-            int idx = obj - cur->rgobj;
-            cur->allocmap[idx / OBJ_PAGE_BITS_PER_WORD] &= ~(1 << (idx % OBJ_PAGE_BITS_PER_WORD));
-            break;
+            int idx = (obj - cur->rgb) / ppool->cbObject;
+            cur->allocmap[idx / OBJ_PAGE_BITS_PER_WORD] &= ~(1ULL << (idx % OBJ_PAGE_BITS_PER_WORD));
+            return;
         }
         cur = cur->pnext;
     }
+    serverLog(LOG_CRIT, "obj not from pool");
+    sfree(obj); // we don't know where it came from
     return;
 }
 
+#define EMBSTR_ROBJ_SIZE (sizeof(robj)+sizeof(struct sdshdr8)+OBJ_ENCODING_EMBSTR_SIZE_LIMIT+1)
+struct alloc_pool poolobj;
+struct alloc_pool poolembstrobj;
+
+void storage_init()
+{
+    int errv = memkind_create_pmem(PMEM_DIR, 0, &mkdisk);
+    if (errv)
+    {
+        fprintf(stderr, "Memory pool creation failed: %d\n", errv);
+        exit(EXIT_FAILURE);
+    }
+    pool_initialize(&poolobj, sizeof(robj));
+    pool_initialize(&poolembstrobj, EMBSTR_ROBJ_SIZE);
+
+    pthread_atfork(handle_prefork, handle_postfork_parent, handle_postfork_child);
+}
+
+
+
+struct redisObject *salloc_obj()
+{
+    return pool_alloc(&poolobj);
+}
+void sfree_obj(struct redisObject *obj)
+{
+    pool_free(&poolobj, obj);
+}
+struct redisObject *salloc_objembstr()
+{
+    return pool_alloc(&poolembstrobj);
+}
+void sfree_objembstr(robj *obj)
+{
+    pool_free(&poolembstrobj, obj);
+}
+
 void *salloc(size_t cb, enum MALLOC_CLASS class)
 {
     switch (class)
@@ -137,19 +192,16 @@ void handle_prefork()
     }
 }
 
-void handle_postfork(int pid)
+void handle_postfork_parent()
 {
-    if (pid != 0)
-    {
-        // Parent, close fdNew
-        close(fdNew);
-        fdNew = -1;
-    }
-    else
-    {
-        int fdOriginal = memkind_fd(mkdisk);
-        memkind_pmem_remapfd(mkdisk, fdNew);
-        close(fdOriginal);
-    }
-    
+    // Parent, close fdNew
+    close(fdNew);
+    fdNew = -1;
+}
+
+void handle_postfork_child()
+{
+    int fdOriginal = memkind_fd(mkdisk);
+    memkind_pmem_remapfd(mkdisk, fdNew);
+    close(fdOriginal);
 }

src/storage.h

@@ -1,6 +1,8 @@
 #ifndef __STORAGE_H__
 #define __STORAGE_H__
 
+#define OBJ_ENCODING_EMBSTR_SIZE_LIMIT 44   // Note: also defined in object.c - should always match
+
 enum MALLOC_CLASS
 {
     MALLOC_LOCAL,
@@ -11,13 +13,12 @@ void storage_init(void);
 
 struct redisObject *salloc_obj();
 void sfree_obj(struct redisObject *obj);
+struct redisObject *salloc_objembstr();
+void sfree_objembstr(struct redisObject *obj);
 
 void *salloc(size_t cb, enum MALLOC_CLASS class);
 void *scalloc(size_t cb, size_t c, enum MALLOC_CLASS class);
 void sfree(void*);
 void *srealloc(void *pv, size_t cb);
 
-void handle_prefork();
-void handle_postfork(int pid);
-
 #endif