working commit (temporary progress, doesn't build)

1 files changed, 60 insertions, 65 deletions

diff --git a/benchmarks/tail-latency/insert_query_threads.cpp b/benchmarks/tail-latency/insert_query_threads.cpp
index 759de5d..1188ce0 100644
--- a/benchmarks/tail-latency/insert_query_threads.cpp
+++ b/benchmarks/tail-latency/insert_query_threads.cpp
@@ -39,7 +39,7 @@ typedef de::DEConfiguration<Shard, Q, de::DeletePolicy::TOMBSTONE,
 std::atomic<size_t> idx;
 std::atomic<bool> inserts_done = false;
 
-ssize_t query_ratio = 8;
+ssize_t query_ratio = 0;
 
 std::atomic<size_t> total_res = 0;
 size_t reccnt = 0;
@@ -72,22 +72,10 @@ void insert_thread(Ext *extension, std::vector<Rec> *records, size_t start_idx,
   TIMER_INIT();
 
   TIMER_START();
-
   for (size_t i=start_idx; i<stop_idx; i++) {
-    //extension->print_structure();
-
     while (!extension->insert((*records)[i])) {
       usleep(1);
-      //fprintf(stderr, "[D] Failed to insert\n");
     }
-
-    //fprintf(stderr, "[D] Record inserted\n");
-
-    if (extension->get_record_count() != i + 1) {
-      fprintf(stderr, "[E]: invalid record count %ld %ld\n", extension->get_record_count(), i+1);
-      extension->print_structure();
-      exit(EXIT_FAILURE);
-    } 
   }
 
   TIMER_STOP();
@@ -115,81 +103,88 @@ int main(int argc, char **argv) {
   std::vector<size_t> sfs = {8}; //, 4, 8, 16, 32, 64, 128, 256, 512, 1024};
   size_t buffer_size = 8000;
   std::vector<size_t> policies = {
-      5
+      0
   };
 
-  std::vector<size_t> thread_counts = {8};
+  std::vector<size_t> thread_counts = {1};
+  std::vector<size_t> modifiers = {0, 1, 2, 3};
 
   size_t insert_threads = 1;
-  size_t query_threads = 16;
+  size_t query_threads = 0;
 
   reccnt = n;
 
   for (auto pol : policies) {
     for (auto internal_thread_cnt : thread_counts) {
-      auto policy = get_policy<Shard, Q>(sfs[0], buffer_size, pol, n);
-      auto config = Conf(std::move(policy));
-      config.recon_enable_maint_on_flush = true;
-      config.recon_maint_disabled = false;
-      //config.buffer_flush_trigger = 4000;
-      config.maximum_threads = internal_thread_cnt;
-
-      g_thrd_cnt = internal_thread_cnt;
-
-      total_insert_time.store(0);
-      total_query_time.store(0);
-      total_query_count.store(0);
-      
-      auto extension = new Ext(std::move(config));
-
-      /* warmup structure w/ 10% of records */
-      size_t warmup = 0 * n;
-      for (size_t k = 0; k < warmup; k++) {
-        while (!extension->insert(data[k])) {
-          usleep(1);
+      for (auto mod : modifiers) {
+        auto policy = get_policy<Shard, Q>(sfs[0], buffer_size, pol, n, mod);
+        auto config = Conf(std::move(policy));
+        config.recon_enable_maint_on_flush = true;
+        config.recon_maint_disabled = false;
+        // config.buffer_flush_trigger = 4000;
+        config.maximum_threads = internal_thread_cnt;
+
+        g_thrd_cnt = internal_thread_cnt;
+
+        total_insert_time.store(0);
+        total_query_time.store(0);
+        total_query_count.store(0);
+
+        auto extension = new Ext(std::move(config));
+
+        /* warmup structure w/ 10% of records */
+        size_t warmup = 0 * n;
+        for (size_t k = 0; k < warmup; k++) {
+          while (!extension->insert(data[k])) {
+            usleep(1);
+          }
         }
-      }
 
-      extension->await_version();
+        extension->await_version();
 
-      idx.store(warmup);
+        idx.store(warmup);
 
-      std::thread i_thrds[insert_threads];
-      std::thread q_thrds[query_threads];
+        std::thread i_thrds[insert_threads];
+        std::thread q_thrds[query_threads];
 
-      size_t per_insert_thrd = (n - warmup) / insert_threads;
-      size_t start = warmup;
+        size_t per_insert_thrd = (n - warmup) / insert_threads;
+        size_t start = warmup;
 
-      for (size_t i=0; i<insert_threads; i++) {
-        i_thrds[i] = std::thread(insert_thread, extension, &data, start, start + per_insert_thrd);
-        start += per_insert_thrd;
-      }
+        for (size_t i = 0; i < insert_threads; i++) {
+          i_thrds[i] = std::thread(insert_thread, extension, &data, start,
+                                   start + per_insert_thrd);
+          start += per_insert_thrd;
+        }
 
-      for (size_t i=0; i<query_threads; i++) {
-        q_thrds[i] = std::thread(query_thread, extension, &queries);
-      }
+        for (size_t i = 0; i < query_threads; i++) {
+          q_thrds[i] = std::thread(query_thread, extension, &queries);
+        }
 
-      for (size_t i=0; i<insert_threads; i++) {
-        i_thrds[i].join();
-      }
+        for (size_t i = 0; i < insert_threads; i++) {
+          i_thrds[i].join();
+        }
 
-      inserts_done.store(true);
+        inserts_done.store(true);
 
-      for (size_t i=0; i<query_threads; i++) {
-        q_thrds[i].join();
-      }
+        for (size_t i = 0; i < query_threads; i++) {
+          q_thrds[i].join();
+        }
 
-      fprintf(stderr, "%ld\n", total_res.load());
+        fprintf(stderr, "%ld\n", total_res.load());
 
-      size_t insert_tput = ((double)(n - warmup) / (double) total_insert_time) *1e9;
-      size_t query_lat = (double) total_query_time.load() / (double) total_query_count.load();
+        size_t insert_tput =
+            ((double)(n - warmup) / (double)total_insert_time) * 1e9;
+        size_t query_lat =
+            (double)total_query_time.load() / (double)total_query_count.load();
 
-      fprintf(stdout, "%ld\t%ld\t%ld\n", internal_thread_cnt, insert_tput, query_lat);
-      fflush(stdout);
+        fprintf(stdout, "%ld\t%ld\t%ld\t%ld\n", internal_thread_cnt, mod,
+                insert_tput, query_lat);
+        fflush(stdout);
 
-      total_res.store(0);
-      inserts_done.store(false);
-      delete extension;
+        total_res.store(0);
+        inserts_done.store(false);
+        delete extension;
+      }
     }
   }