cuckoo: use datapool for cuckoo data storage

This patch uses the datapool interface to allow for durable
operations of the cuckoo storage engine.

To enable durable storage, provide a path to a file in
"cuckoo_datapool" config option. If no file is provided, the
datapool falls back to shared memory.

src/storage/cuckoo/CMakeLists.txt

@@ -1 +1,2 @@
 add_library(cuckoo cuckoo.c)
+target_link_libraries(cuckoo datapool)

src/storage/cuckoo/cuckoo.c

@@ -5,6 +5,8 @@
 #include <hash/cc_murmur3.h>
 #include <cc_mm.h>
 
+#include <datapool/datapool.h>
+
 /* TODO(yao): make D and iv[] configurable */
 #include <stdlib.h>
 #include <sysexits.h>
@@ -37,6 +39,7 @@ static uint32_t iv[D] = {
     0x4dd2be0a
 };
 
+static struct datapool *pool; /* data pool mapping for the hash table */
 static void* ds; /* data store is also the hash table */
 static size_t item_size = CUCKOO_ITEM_SIZE;
 static uint32_t max_nitem = CUCKOO_NITEM;
@@ -279,11 +282,13 @@ cuckoo_setup(cuckoo_options_st *options, cuckoo_metrics_st *metrics)
     }
 
     hash_size = item_size * max_nitem;
-    ds = cc_zalloc(hash_size);
-    if (ds == NULL) {
+    pool = datapool_open(option_str(&options->cuckoo_datapool),
+        hash_size, NULL);
+    if (pool == NULL) {
         log_crit("cuckoo data store allocation failed");
         exit(EX_CONFIG);
     }
+    ds = datapool_addr(pool);
 
     cuckoo_init = true;
 }
@@ -296,7 +301,7 @@ cuckoo_teardown(void)
     if (!cuckoo_init) {
         log_warn("%s has never been setup", CUCKOO_MODULE_NAME);
     } else {
-        cc_free(ds);
+        datapool_close(pool);
     }
 
     cuckoo_metrics = NULL;

src/storage/cuckoo/cuckoo.h

@@ -17,6 +17,7 @@
 #define CUCKOO_NITEM 1024
 #define CUCKOO_POLICY CUCKOO_POLICY_RANDOM
 #define CUCKOO_MAX_TTL (30 * 24 * 60 * 60) /* 30 days */
+#define CUCKOO_DATAPOOL NULL
 
 /*          name                type                default             description */
 #define CUCKOO_OPTION(ACTION)                                                                    \
@@ -25,7 +26,8 @@
     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_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)

test/storage/CMakeLists.txt

@@ -1,2 +1,5 @@
 add_subdirectory(cuckoo)
 add_subdirectory(slab)
+if(USE_PMEM)
+    add_subdirectory(cuckoo_pmem)
+endif(USE_PMEM)

test/storage/cuckoo_pmem/CMakeLists.txt

@@ -0,0 +1,13 @@
+set(suite cuckoo_pmem)
+set(test_name check_${suite})
+
+set(source check_${suite}.c)
+
+add_executable(${test_name} ${source})
+target_link_libraries(${test_name} cuckoo)
+target_link_libraries(${test_name} time)
+target_link_libraries(${test_name} ccommon-static ${CHECK_LIBRARIES})
+target_link_libraries(${test_name} pthread m)
+
+add_dependencies(check ${test_name})
+add_test(${test_name} ${test_name})

test/storage/cuckoo_pmem/check_cuckoo_pmem.c

@@ -0,0 +1,216 @@
+#include <storage/cuckoo/item.h>
+#include <storage/cuckoo/cuckoo.h>
+
+#include <time/time.h>
+
+#include <cc_bstring.h>
+#include <cc_mm.h>
+
+#include <check.h>
+#include <string.h>
+#include <stdio.h>
+
+/* define for each suite, local scope due to macro visibility rule */
+#define SUITE_NAME "cuckoo_pmem"
+#define DEBUG_LOG  SUITE_NAME ".log"
+#define DATAPOOL_PATH "./datapool.pelikan"
+
+void test_insert_basic(uint32_t policy, bool cas);
+void test_insert_collision(uint32_t policy, bool cas);
+void test_cas(uint32_t policy);
+void test_delete_basic(uint32_t policy, bool cas);
+void test_expire_basic(uint32_t policy, bool cas);
+void test_expire_truncated(uint32_t policy, bool cas);
+
+cuckoo_options_st options = { CUCKOO_OPTION(OPTION_INIT) };
+cuckoo_metrics_st metrics = { CUCKOO_METRIC(METRIC_INIT) };
+
+/*
+ * utilities
+ */
+static void
+test_setup(uint32_t policy, bool cas, delta_time_i ttl)
+{
+    option_load_default((struct option *)&options, OPTION_CARDINALITY(options));
+    options.cuckoo_policy.val.vuint = policy;
+    options.cuckoo_item_cas.val.vbool = cas;
+    options.cuckoo_max_ttl.val.vuint = ttl;
+    options.cuckoo_datapool.val.vstr = DATAPOOL_PATH;
+
+    cuckoo_setup(&options, &metrics);
+}
+
+static void
+test_teardown(int un)
+{
+    cuckoo_teardown();
+    if (un)
+        unlink(DATAPOOL_PATH);
+}
+
+static void
+test_reset(uint32_t policy, bool cas, delta_time_i ttl, int un)
+{
+    test_teardown(un);
+    test_setup(policy, cas, ttl);
+}
+
+static void
+test_assert_entry_exists(struct bstring *key, struct val *val)
+{
+    struct item *it = cuckoo_get(key);
+    ck_assert_msg(it != NULL, "cuckoo_get returned NULL");
+    ck_assert_int_eq(it->vlen, val->vstr.len);
+    ck_assert_int_eq(it->klen, key->len);
+    ck_assert_int_eq(it->vlen, val->vstr.len);
+    struct bstring testval;
+    item_value_str(&testval, it);
+    ck_assert_int_eq(it->vlen, testval.len);
+    ck_assert_int_eq(cc_memcmp(testval.data, val->vstr.data, testval.len), 0);
+}
+
+/**
+ * Tests basic functionality for cuckoo_insert and cuckoo_get with small key/val.
+ * Checks that the commands succeed and that the item returned is well-formed.
+ */
+void
+test_insert_basic(uint32_t policy, bool cas)
+{
+#define KEY "key"
+#define VAL "value"
+
+    struct bstring key;
+    struct val val;
+    struct item *it;
+
+    test_reset(policy, cas, CUCKOO_MAX_TTL, 0);
+
+    bstring_set_literal(&key, KEY);
+
+    val.type = VAL_TYPE_STR;
+    bstring_set_literal(&val.vstr, VAL);
+
+    time_update();
+    it = cuckoo_insert(&key, &val, INT32_MAX);
+    ck_assert_msg(it != NULL, "cuckoo_insert not OK");
+
+    test_assert_entry_exists(&key, &val);
+
+    test_reset(policy, cas, CUCKOO_MAX_TTL, 0);
+
+    test_assert_entry_exists(&key, &val);
+
+#undef KEY
+#undef VAL
+}
+
+void
+test_insert_collision(uint32_t policy, bool cas)
+{
+    struct bstring key;
+    struct val val;
+    struct item *it;
+    int hits = 0;
+    char keystring[CC_UINTMAX_MAXLEN];
+    uint64_t i, testval;
+
+    test_reset(policy, cas, CUCKOO_MAX_TTL, 1);
+
+    time_update();
+    for (i = 0; i < CUCKOO_NITEM + 1; i++) {
+        key.len = sprintf(keystring, "%"PRIu64, i);
+        key.data = keystring;
+
+        val.type = VAL_TYPE_INT;
+        val.vint = i;
+
+        it = cuckoo_insert(&key, &val, INT32_MAX);
+        ck_assert_msg(it != NULL, "cuckoo_insert not OK");
+    }
+
+    test_reset(policy, cas, CUCKOO_MAX_TTL, 0);
+
+    for (i = 0; i < CUCKOO_NITEM + 1; i++) {
+        key.len = sprintf(keystring, "%"PRIu64, i);
+        key.data = keystring;
+
+        it = cuckoo_get(&key);
+        if (it == NULL) {
+            continue;
+        }
+        hits++;
+        ck_assert_int_eq(it->klen, key.len);
+        testval = item_value_int(it);
+        ck_assert_int_eq(testval, i);
+    }
+
+    ck_assert_msg(hits > (double)CUCKOO_NITEM * 9 / 10, "hit rate is lower than expected when hash collision occurs");
+    ck_assert_msg(hits <= CUCKOO_NITEM, "hit rate is too high, expected more evicted values");
+}
+
+
+START_TEST(test_insert_basic_random_true)
+{
+    test_insert_basic(CUCKOO_POLICY_RANDOM, true);
+}
+END_TEST
+
+START_TEST(test_insert_basic_expire_true)
+{
+    test_insert_basic(CUCKOO_POLICY_EXPIRE, true);
+}
+END_TEST
+
+START_TEST(test_insert_collision_random_false)
+{
+    test_insert_collision(CUCKOO_POLICY_RANDOM, false);
+}
+END_TEST
+
+START_TEST(test_insert_collision_expire_true)
+{
+    test_insert_collision(CUCKOO_POLICY_EXPIRE, true);
+}
+END_TEST
+
+/*
+ * test suite
+ */
+static Suite *
+cuckoo_suite(void)
+{
+    Suite *s = suite_create(SUITE_NAME);
+
+    /* basic requests */
+    TCase *tc_basic = tcase_create("basic");
+    suite_add_tcase(s, tc_basic);
+
+    tcase_add_test(tc_basic, test_insert_basic_random_true);
+    tcase_add_test(tc_basic, test_insert_basic_expire_true);
+
+    TCase *tc_collision = tcase_create("collision");
+    suite_add_tcase(s, tc_collision);
+
+    tcase_add_test(tc_collision, test_insert_collision_random_false);
+    tcase_add_test(tc_collision, test_insert_collision_expire_true);
+
+    return s;
+}
+
+int
+main(void)
+{
+    int nfail;
+
+    Suite *suite = cuckoo_suite();
+    SRunner *srunner = srunner_create(suite);
+    srunner_set_log(srunner, DEBUG_LOG);
+    srunner_run_all(srunner, CK_ENV); /* set CK_VEBOSITY in ENV to customize */
+    nfail = srunner_ntests_failed(srunner);
+    srunner_free(srunner);
+
+    /* teardown */
+    test_teardown(1);
+
+    return (nfail == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
+}