MutableBuffer tests and bugfixes

3 files changed, 454 insertions, 1 deletions

diff --git a/include/framework/MutableBuffer.h b/include/framework/MutableBuffer.h
index 567b1d7..b3acfee 100644
--- a/include/framework/MutableBuffer.h
+++ b/include/framework/MutableBuffer.h
@@ -95,7 +95,7 @@ public:
     Record<K, V, W>* sorted_output() {
         TIMER_INIT();
         TIMER_START();
-        std::sort(m_data, m_data + m_reccnt.load(), memtable_record_cmp);
+        std::sort(m_data, m_data + m_reccnt.load(), memtable_record_cmp<K,V,W>);
         TIMER_STOP();
 
         #ifdef INSTRUMENT_MERGING
diff --git a/tests/mutable_buffer_tests.cpp b/tests/mutable_buffer_tests.cpp
new file mode 100644
index 0000000..8768551
--- /dev/null
+++ b/tests/mutable_buffer_tests.cpp
@@ -0,0 +1,300 @@
+#include <string>
+#include <thread>
+#include <gsl/gsl_rng.h>
+#include <vector>
+#include <algorithm>
+
+#include "testing.h"
+#include "framework/MutableBuffer.h"
+
+#include <check.h>
+
+using namespace de;
+
+START_TEST(t_create)
+{
+    auto rng = gsl_rng_alloc(gsl_rng_mt19937);
+    auto buffer = new WeightedMBuffer(100, true, 50, rng);
+
+    ck_assert_ptr_nonnull(buffer);
+    ck_assert_int_eq(buffer->get_capacity(), 100);
+    ck_assert_int_eq(buffer->get_record_count(), 0);
+    ck_assert_int_eq(buffer->is_full(), false);
+    ck_assert_ptr_nonnull(buffer->sorted_output());
+    ck_assert_int_eq(buffer->get_tombstone_count(), 0);
+    ck_assert_int_eq(buffer->get_tombstone_capacity(), 50);
+
+    delete buffer;
+    gsl_rng_free(rng);
+}
+END_TEST
+
+
+START_TEST(t_insert)
+{
+    auto rng = gsl_rng_alloc(gsl_rng_mt19937);
+    auto buffer = new WeightedMBuffer(100, true, 50, rng);
+
+    uint64_t key = 0;
+    uint32_t val = 5;
+
+    for (size_t i=0; i<99; i++) {
+        ck_assert_int_eq(buffer->append(key, val, 1, false), 1);
+        ck_assert_int_eq(buffer->check_tombstone(key, val), 0);
+
+        key++;
+        val++;
+
+        ck_assert_int_eq(buffer->get_record_count(), i+1);
+        ck_assert_int_eq(buffer->get_tombstone_count(), 0);
+        ck_assert_int_eq(buffer->is_full(), 0);
+    }
+
+    ck_assert_int_eq(buffer->append(key, val, 1.0, false), 1);
+
+    key++;
+    val++;
+
+    ck_assert_int_eq(buffer->is_full(), 1);
+    ck_assert_int_eq(buffer->append(key, val, 1.0, false), 0);
+
+    delete buffer;
+    gsl_rng_free(rng);
+
+}
+END_TEST
+
+
+START_TEST(t_insert_tombstones)
+{
+    auto rng = gsl_rng_alloc(gsl_rng_mt19937);
+    auto buffer = new WeightedMBuffer(100, true, 50, rng);
+
+    uint64_t key = 0;
+    uint32_t val = 5;
+    size_t ts_cnt = 0;
+
+    for (size_t i=0; i<99; i++) {
+        bool ts = false;
+        if (i % 2 == 0) {
+            ts_cnt++;
+            ts=true;
+        }
+
+        ck_assert_int_eq(buffer->append(key, val, 1.0, ts), 1);
+        ck_assert_int_eq(buffer->check_tombstone(key, val), ts);
+
+        key++;
+        val++;
+
+        ck_assert_int_eq(buffer->get_record_count(), i+1);
+        ck_assert_int_eq(buffer->get_tombstone_count(), ts_cnt);
+        ck_assert_int_eq(buffer->is_full(), 0);
+    }
+
+    // inserting one more tombstone should not be possible
+    ck_assert_int_eq(buffer->append(key, val, 1.0, true), 0);
+
+
+    ck_assert_int_eq(buffer->append(key, val, 1.0, false), 1);
+
+    key++;
+    val++;
+
+    ck_assert_int_eq(buffer->is_full(), 1);
+    ck_assert_int_eq(buffer->append(key, val, 1.0, false), 0);
+
+    delete buffer;
+    gsl_rng_free(rng);
+}
+END_TEST
+
+
+START_TEST(t_truncate)
+{
+    auto rng = gsl_rng_alloc(gsl_rng_mt19937);
+    auto buffer = new WeightedMBuffer(100, true, 100, rng);
+
+    uint64_t key = 0;
+    uint32_t val = 5;
+    size_t ts_cnt = 0;
+
+    for (size_t i=0; i<100; i++) {
+        bool ts = false;
+        if (i % 2 == 0) {
+            ts_cnt++;
+            ts=true;
+        }
+
+        ck_assert_int_eq(buffer->append(key, val, 1.0, ts), 1);
+        ck_assert_int_eq(buffer->check_tombstone(key, val), ts);
+
+        key++;
+        val++;
+
+        ck_assert_int_eq(buffer->get_record_count(), i+1);
+        ck_assert_int_eq(buffer->get_tombstone_count(), ts_cnt);
+    }
+
+    ck_assert_int_eq(buffer->is_full(), 1);
+    ck_assert_int_eq(buffer->append(key, val, 1.0, false), 0);
+
+    ck_assert_int_eq(buffer->truncate(), 1);
+
+    ck_assert_int_eq(buffer->is_full(), 0);
+    ck_assert_int_eq(buffer->get_record_count(), 0);
+    ck_assert_int_eq(buffer->get_tombstone_count(), 0);
+    ck_assert_int_eq(buffer->append(key, val, 1.0, false), 1);
+
+    delete buffer;
+    gsl_rng_free(rng);
+
+}
+END_TEST
+
+
+START_TEST(t_sorted_output)
+{
+    size_t cnt = 100;
+
+    auto rng = gsl_rng_alloc(gsl_rng_mt19937);
+    auto buffer = new WeightedMBuffer(cnt, true, cnt/2, rng);
+
+
+    std::vector<uint64_t> keys(cnt);
+    for (size_t i=0; i<cnt-2; i++) {
+        keys[i] = rand();
+    }
+
+    // duplicate final two records for tombstone testing
+    // purposes
+    keys[cnt-2] =  keys[cnt-3];
+    keys[cnt-1] =  keys[cnt-2];
+
+    uint32_t val = 12345;
+    for (size_t i=0; i<cnt-2; i++) {
+        buffer->append(keys[i], val, 1.0, false);
+    }
+
+    buffer->append(keys[cnt-2], val, 1.0, true);
+    buffer->append(keys[cnt-1], val, 1.0, true);
+
+
+    WeightedRec *sorted_records = buffer->sorted_output();
+    std::sort(keys.begin(), keys.end());
+
+    for (size_t i=0; i<cnt; i++) {
+        ck_assert_int_eq(sorted_records[i].key, keys[i]);
+    }
+
+    delete buffer;
+    gsl_rng_free(rng);
+}
+END_TEST
+
+
+void insert_records(std::vector<std::pair<uint64_t, uint32_t>> *values, size_t start, size_t stop, WeightedMBuffer *buffer)
+{
+    for (size_t i=start; i<stop; i++) {
+        buffer->append((*values)[i].first, (*values)[i].second, 1.0);
+    }
+
+}
+
+START_TEST(t_multithreaded_insert)
+{
+    size_t cnt = 10000;
+    auto rng = gsl_rng_alloc(gsl_rng_mt19937);
+    auto buffer = new WeightedMBuffer(cnt, true, cnt/2, rng);
+
+    std::vector<std::pair<uint64_t, uint32_t>> records(cnt);
+    for (size_t i=0; i<cnt; i++) {
+        records[i] = {rand(), rand()};
+    }
+
+    // perform a t_multithreaded insertion
+    size_t thread_cnt = 8;
+    size_t per_thread = cnt / thread_cnt;
+    std::vector<std::thread> workers(thread_cnt);
+    size_t start = 0;
+    size_t stop = start + per_thread;
+    for (size_t i=0; i<thread_cnt; i++) {
+        workers[i] = std::thread(insert_records, &records, start, stop, buffer);
+        start = stop;
+        stop = std::min(start + per_thread, cnt);
+    }
+
+    for (size_t i=0; i<thread_cnt; i++) {
+        if (workers[i].joinable()) {
+            workers[i].join();
+        }
+    }
+
+    ck_assert_int_eq(buffer->is_full(), 1);
+    ck_assert_int_eq(buffer->get_record_count(), cnt);
+
+    std::sort(records.begin(), records.end());
+    WeightedRec *sorted_records = buffer->sorted_output();
+    for (size_t i=0; i<cnt; i++) {
+        ck_assert_int_eq(sorted_records[i].key, records[i].first);
+    }
+
+    delete buffer;
+    gsl_rng_free(rng);
+}
+END_TEST
+
+
+Suite *unit_testing()
+{
+    Suite *unit = suite_create("Mutable Buffer Unit Testing");
+    TCase *initialize = tcase_create("de::MutableBuffer Constructor Testing");
+    tcase_add_test(initialize, t_create);
+
+    suite_add_tcase(unit, initialize);
+
+
+    TCase *append = tcase_create("de::MutableBuffer::append Testing");
+    tcase_add_test(append, t_insert);
+    tcase_add_test(append, t_insert_tombstones);
+    tcase_add_test(append, t_multithreaded_insert);
+
+    suite_add_tcase(unit, append);
+
+
+    TCase *truncate = tcase_create("de::MutableBuffer::truncate Testing");
+    tcase_add_test(truncate, t_truncate);
+
+    suite_add_tcase(unit, truncate);
+
+
+    TCase *sorted_out = tcase_create("de::MutableBuffer::sorted_output");
+    tcase_add_test(sorted_out, t_sorted_output);
+
+    suite_add_tcase(unit, sorted_out);
+
+    return unit;
+}
+
+
+int run_unit_tests()
+{
+    int failed = 0;
+    Suite *unit = unit_testing();
+    SRunner *unit_runner = srunner_create(unit);
+
+    srunner_run_all(unit_runner, CK_NORMAL);
+    failed = srunner_ntests_failed(unit_runner);
+    srunner_free(unit_runner);
+
+    return failed;
+}
+
+
+int main() 
+{
+    int unit_failed = run_unit_tests();
+
+    return (unit_failed == 0) ? EXIT_SUCCESS : EXIT_FAILURE;
+}
+
diff --git a/tests/testing.h b/tests/testing.h
new file mode 100644
index 0000000..234deef
--- /dev/null
+++ b/tests/testing.h
@@ -0,0 +1,153 @@
+/*
+ *
+ *
+ */
+
+#pragma once
+
+#include <string>
+
+#include <unistd.h>
+#include <fcntl.h>
+
+#include "util/types.h"
+#include "util/base.h"
+#include "framework/MutableBuffer.h"
+#include "framework/InternalLevel.h"
+
+typedef de::Record<uint64_t, uint32_t, uint64_t> WeightedRec;
+typedef de::MutableBuffer<uint64_t, uint32_t, uint64_t> WeightedMBuffer;
+typedef de::InternalLevel<uint64_t, uint32_t, uint64_t> WeightedLevel;
+
+typedef de::Record<uint64_t, uint32_t> UnweightedRec;
+typedef de::MutableBuffer<uint64_t, uint32_t> UnweightedMBuffer;
+typedef de::InternalLevel<uint64_t, uint32_t> UnweightedLevel;
+
+static gsl_rng *g_rng = gsl_rng_alloc(gsl_rng_mt19937);
+
+static bool initialize_test_file(std::string fname, size_t page_cnt)
+{
+    auto flags = O_RDWR | O_CREAT | O_TRUNC;
+    mode_t mode = 0640;
+    char *page = nullptr;
+
+    int fd = open(fname.c_str(), flags, mode);
+    if (fd == -1) {
+        goto error;
+    }
+
+    page = (char *) aligned_alloc(de::SECTOR_SIZE, de::PAGE_SIZE);
+    if (!page) {
+        goto error_opened;
+    }
+
+    for (size_t i=0; i<=page_cnt; i++) {
+        *((int *) page) = i;
+        if (write(fd, page, de::PAGE_SIZE) == -1) {
+            goto error_alloced;
+        }
+    }
+
+    free(page);
+
+    return 1;
+
+error_alloced:
+    free(page);
+
+error_opened:
+    close(fd);
+
+error:
+    return 0;
+}
+
+static bool roughly_equal(int n1, int n2, size_t mag, double epsilon) {
+    return ((double) std::abs(n1 - n2) / (double) mag) < epsilon;
+}
+
+static WeightedMBuffer *create_test_mbuffer(size_t cnt)
+{
+    auto buffer = new WeightedMBuffer(cnt, true, cnt, g_rng);
+
+    for (size_t i = 0; i < cnt; i++) {
+        uint64_t key = rand();
+        uint32_t val = rand();
+
+        buffer->append(key, val);
+    }
+
+    return buffer;
+}
+
+static WeightedMBuffer *create_test_mbuffer_tombstones(size_t cnt, size_t ts_cnt) 
+{
+    auto buffer = new WeightedMBuffer(cnt, true, ts_cnt, g_rng);
+
+    std::vector<std::pair<uint64_t, uint32_t>> tombstones;
+
+    for (size_t i = 0; i < cnt; i++) {
+        uint64_t key = rand();
+        uint32_t val = rand();
+
+        if (i < ts_cnt) {
+            tombstones.push_back({key, val});
+        }
+
+        buffer->append(key, val);
+    }
+
+    for (size_t i=0; i<ts_cnt; i++) {
+        buffer->append(tombstones[i].first, tombstones[i].second, 1.0, true);
+    }
+
+    return buffer;
+}
+
+static WeightedMBuffer *create_weighted_mbuffer(size_t cnt)
+{
+    auto buffer = new WeightedMBuffer(cnt, true, cnt, g_rng);
+    
+    // Put in half of the count with weight one.
+    uint64_t key = 1;
+    for (size_t i=0; i< cnt / 2; i++) {
+        buffer->append(key, i, 2);
+    }
+
+    // put in a quarter of the count with weight two.
+    key = 2;
+    for (size_t i=0; i< cnt / 4; i++) {
+        buffer->append(key, i, 4);
+    }
+
+    // the remaining quarter with weight four.
+    key = 3;
+    for (size_t i=0; i< cnt / 4; i++) {
+        buffer->append(key, i, 8);
+    }
+
+    return buffer;
+}
+
+static WeightedMBuffer *create_double_seq_mbuffer(size_t cnt, bool ts=false) 
+{
+    auto buffer = new WeightedMBuffer(cnt, true, cnt, g_rng);
+
+    for (size_t i = 0; i < cnt / 2; i++) {
+        uint64_t key = i;
+        uint32_t val = i;
+
+        buffer->append(key, val, 1.0, ts);
+    }
+
+    for (size_t i = 0; i < cnt / 2; i++) {
+        uint64_t key = i;
+        uint32_t val = i + 1;
+
+        buffer->append(key, val, 1.0, ts);
+    }
+
+    return buffer;
+}
+
+