Fixed a few concurrency bugs

11 files changed, 51 insertions, 53 deletions

diff --git a/tests/include/concurrent_extension.h b/tests/include/concurrent_extension.h
index 63ccf87..e8f4df0 100644
--- a/tests/include/concurrent_extension.h
+++ b/tests/include/concurrent_extension.h
@@ -327,7 +327,6 @@ START_TEST(t_static_structure)
     }
 
     gsl_rng_free(rng);
-    delete flat;
     delete test_de;
 }
 END_TEST
diff --git a/tests/include/dynamic_extension.h b/tests/include/dynamic_extension.h
index c378ea9..8021e9a 100644
--- a/tests/include/dynamic_extension.h
+++ b/tests/include/dynamic_extension.h
@@ -293,7 +293,6 @@ START_TEST(t_static_structure)
     }
 
     gsl_rng_free(rng);
-    delete flat;
     delete test_de;
 }
 END_TEST
diff --git a/tests/include/irs.h b/tests/include/irs.h
index 35a4bbb..8a05761 100644
--- a/tests/include/irs.h
+++ b/tests/include/irs.h
@@ -42,7 +42,7 @@ static gsl_rng *g_rng;
 START_TEST(t_irs)
 {
     auto buffer = create_sequential_mbuffer<R>(100, 1000);
-    auto shard = Shard(buffer->get_buffer_view());
+    auto shard = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     size_t k = 5;
     irs::Query<Shard>::Parameters parms;
@@ -80,7 +80,7 @@ START_TEST(t_buffer_irs)
     parms.rng = g_rng;
 
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         auto query = irs::Query<Shard>::local_preproc_buffer(&view, &parms);
         irs::Query<Shard>::distribute_query(&parms, {}, query);
         auto result = irs::Query<Shard>::local_query_buffer(query); 
@@ -103,8 +103,8 @@ START_TEST(t_irs_merge)
     auto buffer1 = create_sequential_mbuffer<R>(100, 200);
     auto buffer2 = create_sequential_mbuffer<R>(400, 1000);
 
-    auto shard1 = Shard(buffer1->get_buffer_view());
-    auto shard2 = Shard(buffer2->get_buffer_view());
+    auto shard1 = Shard(buffer1->get_buffer_view(buffer1->debug_get_head()));
+    auto shard2 = Shard(buffer2->get_buffer_view(buffer2->debug_get_head()));
 
     size_t k = 10;
     irs::Query<Shard>::Parameters parms;
diff --git a/tests/include/pointlookup.h b/tests/include/pointlookup.h
index f3a03dd..a365144 100644
--- a/tests/include/pointlookup.h
+++ b/tests/include/pointlookup.h
@@ -37,10 +37,10 @@
 START_TEST(t_point_lookup_query)
 {
     auto buffer = create_test_mbuffer<R>(1000);
-    auto shard = Shard(buffer->get_buffer_view());
+    auto shard = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     {
-        auto bv = buffer->get_buffer_view();
+        auto bv = buffer->get_buffer_view(buffer->debug_get_head());
         for (size_t i=0; i<bv.get_record_count(); i++) {
             auto key = bv.get(i)->rec.key;
 
@@ -73,7 +73,7 @@ START_TEST(t_buffer_point_lookup)
 
     auto buffer = create_test_mbuffer<R>(1000);
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         for (int i=view.get_record_count()-1; i>=0; i--) {
             pl::Query<Shard>::Parameters parms = {view.get(i)->rec.key};
             auto local_query = pl::Query<Shard>::local_preproc_buffer(&view, &parms);
diff --git a/tests/include/rangecount.h b/tests/include/rangecount.h
index b77c77d..2640e84 100644
--- a/tests/include/rangecount.h
+++ b/tests/include/rangecount.h
@@ -38,7 +38,7 @@
 START_TEST(t_range_count)
 {
     auto buffer = create_sequential_mbuffer<R>(100, 1000);
-    auto shard = Shard(buffer->get_buffer_view());
+    auto shard = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     rc::Query<Shard>::Parameters parms = {300, 500};
 
@@ -61,7 +61,7 @@ START_TEST(t_buffer_range_count)
     rc::Query<Shard>::Parameters parms = {300, 500};
 
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         auto query = rc::Query<Shard>::local_preproc_buffer(&view, &parms);
         auto result = rc::Query<Shard>::local_query_buffer(query); 
         delete query;
@@ -79,8 +79,8 @@ START_TEST(t_range_count_merge)
     auto buffer1 = create_sequential_mbuffer<R>(100, 200);
     auto buffer2 = create_sequential_mbuffer<R>(400, 1000);
 
-    auto shard1 = Shard(buffer1->get_buffer_view());
-    auto shard2 = Shard(buffer2->get_buffer_view());
+    auto shard1 = Shard(buffer1->get_buffer_view(buffer1->debug_get_head()));
+    auto shard2 = Shard(buffer2->get_buffer_view(buffer2->debug_get_head()));
 
     rc::Query<Shard>::Parameters parms = {150, 500};
 
diff --git a/tests/include/rangequery.h b/tests/include/rangequery.h
index f7bb7c1..076a37f 100644
--- a/tests/include/rangequery.h
+++ b/tests/include/rangequery.h
@@ -40,7 +40,7 @@
 START_TEST(t_range_query)
 {
     auto buffer = create_sequential_mbuffer<R>(100, 1000);
-    auto shard = Shard(buffer->get_buffer_view());
+    auto shard = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     rq::Query<Shard>::Parameters parms = {300, 500};
 
@@ -67,7 +67,7 @@ START_TEST(t_buffer_range_query)
     rq::Query<Shard>::Parameters parms = {300, 500};
 
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         auto query = rq::Query<Shard>::local_preproc_buffer(&view, &parms);
         auto result = rq::Query<Shard>::local_query_buffer(query); 
         delete query;
@@ -89,8 +89,8 @@ START_TEST(t_range_query_merge)
     auto buffer1 = create_sequential_mbuffer<R>(100, 200);
     auto buffer2 = create_sequential_mbuffer<R>(400, 1000);
 
-    auto shard1 = Shard(buffer1->get_buffer_view());
-    auto shard2 = Shard(buffer2->get_buffer_view());
+    auto shard1 = Shard(buffer1->get_buffer_view(buffer1->debug_get_head()));
+    auto shard2 = Shard(buffer2->get_buffer_view(buffer2->debug_get_head()));
 
     rq::Query<Shard>::Parameters parms = {150, 500};
 
diff --git a/tests/include/shard_standard.h b/tests/include/shard_standard.h
index 0b5ab00..996d947 100644
--- a/tests/include/shard_standard.h
+++ b/tests/include/shard_standard.h
@@ -58,7 +58,7 @@ START_TEST(t_mbuffer_init)
         buffer->append(r);
     }
 
-    Shard* shard = new Shard(buffer->get_buffer_view());
+    Shard* shard = new Shard(buffer->get_buffer_view(buffer->debug_get_head()));
     ck_assert_uint_eq(shard->get_record_count(), 512);
 
     delete buffer;
@@ -73,9 +73,9 @@ START_TEST(t_shard_init)
     auto mbuffer2 = create_test_mbuffer<R>(n);
     auto mbuffer3 = create_test_mbuffer<R>(n);
 
-    auto shard1 = new Shard(mbuffer1->get_buffer_view());
-    auto shard2 = new Shard(mbuffer2->get_buffer_view());
-    auto shard3 = new Shard(mbuffer3->get_buffer_view());
+    auto shard1 = new Shard(mbuffer1->get_buffer_view(mbuffer1->debug_get_head()));
+    auto shard2 = new Shard(mbuffer2->get_buffer_view(mbuffer2->debug_get_head()));
+    auto shard3 = new Shard(mbuffer3->get_buffer_view(mbuffer3->debug_get_head()));
 
     std::vector<const Shard*> shards = {shard1, shard2, shard3};
     auto shard4 = new Shard(shards);
@@ -122,8 +122,8 @@ START_TEST(t_full_cancelation)
     auto buffer = create_double_seq_mbuffer<R>(n, false);
     auto buffer_ts = create_double_seq_mbuffer<R>(n, true);
 
-    Shard* shard = new Shard(buffer->get_buffer_view());
-    Shard* shard_ts = new Shard(buffer_ts->get_buffer_view());
+    Shard* shard = new Shard(buffer->get_buffer_view(buffer->debug_get_head()));
+    Shard* shard_ts = new Shard(buffer_ts->get_buffer_view(buffer_ts->debug_get_head()));
 
     ck_assert_int_eq(shard->get_record_count(), n);
     ck_assert_int_eq(shard->get_tombstone_count(), 0);
@@ -151,10 +151,10 @@ START_TEST(t_point_lookup)
     size_t n = 10000;
 
     auto buffer = create_double_seq_mbuffer<R>(n, false);
-    auto isam = Shard(buffer->get_buffer_view());
+    auto isam = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
 
         for (size_t i=0; i<n; i++) {
             auto rec = view.get(i);
@@ -177,7 +177,7 @@ START_TEST(t_point_lookup_miss)
     size_t n = 10000;
 
     auto buffer = create_double_seq_mbuffer<R>(n, false);
-    auto isam = Shard(buffer->get_buffer_view());
+    auto isam = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     for (uint32_t i=n + 100; i<2*n; i++) {
         R r = R{i, i};
diff --git a/tests/include/shard_string.h b/tests/include/shard_string.h
index 7a3d761..bd428ee 100644
--- a/tests/include/shard_string.h
+++ b/tests/include/shard_string.h
@@ -48,7 +48,7 @@ START_TEST(t_mbuffer_init)
         buffer->append(recs[i]);
     }
 
-    Shard* shard = new Shard(buffer->get_buffer_view());
+    Shard* shard = new Shard(buffer->get_buffer_view(buffer->debug_get_head()));
     ck_assert_uint_eq(shard->get_record_count(), 512);
 
     delete buffer;
@@ -63,9 +63,9 @@ START_TEST(t_shard_init)
     auto mbuffer2 = create_test_mbuffer<R>(n);
     auto mbuffer3 = create_test_mbuffer<R>(n);
 
-    auto shard1 = new Shard(mbuffer1->get_buffer_view());
-    auto shard2 = new Shard(mbuffer2->get_buffer_view());
-    auto shard3 = new Shard(mbuffer3->get_buffer_view());
+    auto shard1 = new Shard(mbuffer1->get_buffer_view(mbuffer1->debug_get_head()));
+    auto shard2 = new Shard(mbuffer2->get_buffer_view(mbuffer2->debug_get_head()));
+    auto shard3 = new Shard(mbuffer3->get_buffer_view(mbuffer3->debug_get_head()));
 
     std::vector<const Shard*> shards = {shard1, shard2, shard3};
     auto shard4 = new Shard(shards);
@@ -110,10 +110,10 @@ START_TEST(t_point_lookup)
     size_t n = 10000;
 
     auto buffer = create_test_mbuffer<R>(n);
-    auto shard = Shard(buffer->get_buffer_view());
+    auto shard = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
 
         for (size_t i=0; i<n; i++) {
             auto rec = view.get(i);
@@ -137,7 +137,7 @@ START_TEST(t_point_lookup_miss)
     size_t n = 10000;
 
     auto buffer = create_test_mbuffer<R>(n);
-    auto shard = Shard(buffer->get_buffer_view());
+    auto shard = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     for (size_t i=n + 100; i<2*n; i++) {
         const char *c = "computer";
diff --git a/tests/include/wss.h b/tests/include/wss.h
index 01327d2..5185567 100644
--- a/tests/include/wss.h
+++ b/tests/include/wss.h
@@ -41,7 +41,7 @@ typedef wss::Query<Shard> Q;
 START_TEST(t_wss_query)
 {
     auto buffer = create_weighted_mbuffer<R>(1000);
-    auto shard = Shard(buffer->get_buffer_view());
+    auto shard = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
     auto rng = gsl_rng_alloc(gsl_rng_mt19937);
 
     size_t k = 20;
@@ -77,7 +77,7 @@ START_TEST(t_buffer_wss_query)
     parms.sample_size = k;
 
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         auto query = Q::local_preproc_buffer(&view, &parms);
         Q::distribute_query(&parms, {}, query);
         auto result = Q::local_query_buffer(query);
diff --git a/tests/mutable_buffer_tests.cpp b/tests/mutable_buffer_tests.cpp
index 16f9269..816c8b0 100644
--- a/tests/mutable_buffer_tests.cpp
+++ b/tests/mutable_buffer_tests.cpp
@@ -38,7 +38,7 @@ START_TEST(t_create)
     ck_assert_int_eq(buffer->get_tombstone_count(), 0);
 
     {
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         ck_assert_int_eq(view.get_tombstone_count(), 0);
         ck_assert_int_eq(view.get_record_count(), 0);
     }
@@ -68,7 +68,7 @@ START_TEST(t_insert)
         cnt++;
 
         ck_assert_int_eq(buffer->get_record_count(), cnt);
-        ck_assert_int_eq(buffer->get_buffer_view().get_record_count(), cnt);
+        ck_assert_int_eq(buffer->get_buffer_view(buffer->debug_get_head()).get_record_count(), cnt);
         ck_assert_int_eq(buffer->get_tail(), cnt);
     }
 
@@ -85,7 +85,7 @@ START_TEST(t_insert)
         cnt++;
 
         ck_assert_int_eq(buffer->get_record_count(), cnt);
-        ck_assert_int_eq(buffer->get_buffer_view().get_record_count(), cnt);
+        ck_assert_int_eq(buffer->get_buffer_view(buffer->debug_get_head()).get_record_count(), cnt);
 
         ck_assert_int_eq(buffer->get_tombstone_count(), 0);
         ck_assert_int_eq(buffer->is_at_low_watermark(), true);
@@ -131,14 +131,14 @@ START_TEST(t_advance_head)
     Wrapped<Rec> *view_records = new Wrapped<Rec>[buffer->get_record_count()];
     {
         /* get a view of the pre-advanced state */
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         ck_assert_int_eq(view.get_record_count(), cnt);
         view.copy_to_buffer((psudb::byte *) view_records);
 
         /* advance the head */
         ck_assert_int_eq(buffer->advance_head(new_head), 1);
         ck_assert_int_eq(buffer->get_record_count(), 25);
-        ck_assert_int_eq(buffer->get_buffer_view().get_record_count(), 25);
+        ck_assert_int_eq(buffer->get_buffer_view(buffer->debug_get_head()).get_record_count(), 25);
         ck_assert_int_eq(view.get_record_count(), cnt);
         ck_assert_int_eq(buffer->get_available_capacity(), 200 - cnt);
 
@@ -272,7 +272,7 @@ START_TEST(t_bview_get)
 
     {
         /* get a view of the pre-advanced state */
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         auto reccnt = view.get_record_count();
 
         /* scan the records in the view */
@@ -291,7 +291,7 @@ START_TEST(t_bview_get)
 
     {
         /* get a new view (should have fewer records) */
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         auto reccnt = view.get_record_count();
 
         /* verify the scan again */
@@ -311,7 +311,7 @@ START_TEST(t_bview_get)
 
     {
         /* get a new view (should have fewer records) */
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         auto reccnt = view.get_record_count();
 
         /* verify the scan again */
@@ -364,7 +364,7 @@ START_TEST(t_bview_delete)
     Rec fdr2 = {300, 300};
     {
         /* get a new view (should have fewer records) */
-        auto view = buffer->get_buffer_view();
+        auto view = buffer->get_buffer_view(buffer->debug_get_head());
         ck_assert_int_eq(view.delete_record(dr1), 1);
         ck_assert_int_eq(view.delete_record(dr2), 1);
         ck_assert_int_eq(view.delete_record(dr3), 1);
diff --git a/tests/vptree_tests.cpp b/tests/vptree_tests.cpp
index 49964e5..3bbf7c1 100644
--- a/tests/vptree_tests.cpp
+++ b/tests/vptree_tests.cpp
@@ -33,7 +33,7 @@ START_TEST(t_mbuffer_init)
         buffer->append({i, i});
     }
 
-    Shard* shard = new Shard(buffer->get_buffer_view());
+    Shard* shard = new Shard(buffer->get_buffer_view(buffer->debug_get_head()));
     ck_assert_uint_eq(shard->get_record_count(), n);
 
     delete buffer;
@@ -48,9 +48,9 @@ START_TEST(t_wss_init)
     auto mbuffer2 = create_test_mbuffer<R>(n);
     auto mbuffer3 = create_test_mbuffer<R>(n);
 
-    auto shard1 = new Shard(mbuffer1->get_buffer_view());
-    auto shard2 = new Shard(mbuffer2->get_buffer_view());
-    auto shard3 = new Shard(mbuffer3->get_buffer_view());
+    auto shard1 = new Shard(mbuffer1->get_buffer_view(mbuffer1->debug_get_head()));
+    auto shard2 = new Shard(mbuffer2->get_buffer_view(mbuffer2->debug_get_head()));
+    auto shard3 = new Shard(mbuffer3->get_buffer_view(mbuffer3->debug_get_head()));
 
     std::vector<const Shard *> shards = {shard1, shard2, shard3};
     auto shard4 = new Shard(shards);
@@ -74,10 +74,10 @@ START_TEST(t_point_lookup)
     size_t n = 16;
 
     auto buffer = create_sequential_mbuffer<R>(0, n);
-    auto wss = Shard(buffer->get_buffer_view());
+    auto wss = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     {
-        auto bv = buffer->get_buffer_view();
+        auto bv = buffer->get_buffer_view(buffer->debug_get_head());
 
         for (size_t i=0; i<n; i++) {
             PRec r;
@@ -102,7 +102,7 @@ START_TEST(t_point_lookup_miss)
     size_t n = 10000;
 
     auto buffer = create_sequential_mbuffer<R>(0, n);
-    auto wss = Shard(buffer->get_buffer_view());
+    auto wss = Shard(buffer->get_buffer_view(buffer->debug_get_head()));
 
     for (size_t i=n + 100; i<2*n; i++) {
         PRec r;
@@ -131,7 +131,7 @@ START_TEST(t_buffer_query)
     p.point = target;
 
     {
-        auto bv = buffer->get_buffer_view();
+        auto bv = buffer->get_buffer_view(buffer->debug_get_head());
         auto query = Q::local_preproc_buffer(&bv, &p);
         auto result = Q::local_query_buffer(query);
         delete query;
@@ -151,7 +151,7 @@ START_TEST(t_knn_query)
     size_t n = 1000;
     auto buffer = create_sequential_mbuffer<R>(0, n);
 
-    auto vptree = VPTree<PRec>(buffer->get_buffer_view());
+    auto vptree = VPTree<PRec>(buffer->get_buffer_view(buffer->debug_get_head()));
 
     Q::Parameters p;