Updates for dynamic rate limitingnew-concurrency

6 files changed, 201 insertions, 24 deletions

diff --git a/CMakeLists.txt b/CMakeLists.txt
index 6ad9040..81e03f1 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -363,6 +363,11 @@ if (tail_bench)
     target_include_directories(stall_benchmark_vptree PRIVATE include external external/m-tree/cpp external/PGM-index/include external/PLEX/include benchmarks/include external/psudb-common/cpp/include)
     target_link_options(stall_benchmark_vptree PUBLIC -mcx16)
 
+    add_executable(stall_benchmark_tput_vptree ${CMAKE_CURRENT_SOURCE_DIR}/benchmarks/tail-latency/stall_benchmark_tput_vptree.cpp)
+    target_link_libraries(stall_benchmark_tput_vptree PUBLIC gsl pthread atomic)
+    target_include_directories(stall_benchmark_tput_vptree PRIVATE include external external/m-tree/cpp external/PGM-index/include external/PLEX/include benchmarks/include external/psudb-common/cpp/include)
+    target_link_options(stall_benchmark_tput_vptree PUBLIC -mcx16)
+
     add_executable(isam_construction_times ${CMAKE_CURRENT_SOURCE_DIR}/benchmarks/tail-latency/isam_construction_times.cpp)
     target_link_libraries(isam_construction_times PUBLIC gsl pthread atomic)
     target_include_directories(isam_construction_times PRIVATE include external external/m-tree/cpp external/PGM-index/include external/PLEX/include benchmarks/include external/psudb-common/cpp/include)
diff --git a/benchmarks/tail-latency/isam_latency_dist.cpp b/benchmarks/tail-latency/isam_latency_dist.cpp
index c116413..e9faa47 100644
--- a/benchmarks/tail-latency/isam_latency_dist.cpp
+++ b/benchmarks/tail-latency/isam_latency_dist.cpp
@@ -25,10 +25,10 @@
 typedef de::Record<uint64_t, uint64_t> Rec;
 typedef de::ISAMTree<Rec>Shard;
 typedef de::rc::Query<Shard> Q;
-typedef de::DynamicExtension<Shard, Q, de::DeletePolicy::TOMBSTONE, de::SerialScheduler> Ext;
+typedef de::DynamicExtension<Shard, Q, de::DeletePolicy::TOMBSTONE, de::FIFOScheduler> Ext;
 typedef Q::Parameters QP;
 typedef de::DEConfiguration<Shard, Q, de::DeletePolicy::TOMBSTONE,
-                            de::SerialScheduler>
+                            de::FIFOScheduler>
     Conf;
 
 void usage(char *progname) {
@@ -50,9 +50,9 @@ int main(int argc, char **argv) {
     auto data = read_sosd_file<Rec>(d_fname, n);
     auto queries = read_range_queries<QP>(q_fname, .0001);
 
-    std::vector<int> policies = {0, 1};
-    std::vector<size_t> buffers = {1000, 8000, 16000};
-    std::vector<size_t> sfs = {4};
+    std::vector<int> policies = {6};
+    std::vector<size_t> buffers = {12000};
+    std::vector<size_t> sfs = {3};
     
     for (size_t l=0; l<policies.size(); l++) {
     for (size_t j=0; j<buffers.size(); j++) {
@@ -63,7 +63,7 @@ int main(int argc, char **argv) {
         config.recon_enable_maint_on_flush = true;
         config.recon_maint_disabled = false;
         
-        auto extension = new Ext(std::move(config));
+        auto extension = new Ext(std::move(config), .999);
 
         /* warmup structure w/ 10% of records */
         size_t warmup = .3 * n;
@@ -89,24 +89,24 @@ int main(int argc, char **argv) {
 
         extension->await_version();
         
-        /* repeat the queries a bunch of times */
-        for (size_t l=0; l<10; l++) {
-        for (size_t i=0; i<queries.size(); i++) {
-            TIMER_START();
-            auto q = queries[i];
-            auto res = extension->query(std::move(q));
-            res.get();
-            TIMER_STOP();
+        // /* repeat the queries a bunch of times */
+        // for (size_t l=0; l<10; l++) {
+        // for (size_t i=0; i<queries.size(); i++) {
+        //     TIMER_START();
+        //     auto q = queries[i];
+        //     auto res = extension->query(std::move(q));
+        //     res.get();
+        //     TIMER_STOP();
 
-            fprintf(stdout, "Q\t%ld\t%ld\t%d\t%ld\n", sfs[k], buffers[j], policies[l], TIMER_RESULT());
-        }
-        }
+        //     fprintf(stdout, "Q\t%ld\t%ld\t%d\t%ld\n", sfs[k], buffers[j], policies[l], TIMER_RESULT());
+        // }
+        // }
 
 
-        QP p = {0, 10000};
-        auto res =extension->query(std::move(p));
+        // QP p = {0, 10000};
+        // auto res =extension->query(std::move(p));
 
-        fprintf(stderr, "%ld\n", res.get());
+        // fprintf(stderr, "%ld\n", res.get());
         delete extension;
     }}}
 
diff --git a/benchmarks/tail-latency/stall_benchmark_tput_vptree.cpp b/benchmarks/tail-latency/stall_benchmark_tput_vptree.cpp
new file mode 100644
index 0000000..1ae5996
--- /dev/null
+++ b/benchmarks/tail-latency/stall_benchmark_tput_vptree.cpp
@@ -0,0 +1,154 @@
+/*
+ *
+ */
+
+#include <cstdlib>
+#define ENABLE_TIMER
+// #define DE_PRINT_SHARD_COUNT
+// #define TS_TEST
+
+#include <thread>
+
+#include "file_util.h"
+#include "framework/DynamicExtension.h"
+#include "framework/interface/Record.h"
+#include "framework/scheduling/FIFOScheduler.h"
+#include "framework/scheduling/SerialScheduler.h"
+#include "framework/util/Configuration.h"
+#include "query/knn.h"
+#include "shard/VPTree.h"
+#include "standard_benchmarks.h"
+#include "util/types.h"
+
+#include "framework/reconstruction/FixedShardCountPolicy.h"
+
+#include <gsl/gsl_rng.h>
+
+#include "psu-util/timer.h"
+
+typedef Word2VecRec Rec;
+typedef de::VPTree<Rec> Shard;
+typedef de::knn::Query<Shard> Q;
+typedef de::DynamicExtension<Shard, Q, de::DeletePolicy::TOMBSTONE,
+                             de::FIFOScheduler>
+    Ext;
+typedef Q::Parameters QP;
+typedef de::DEConfiguration<Shard, Q, de::DeletePolicy::TOMBSTONE,
+                            de::FIFOScheduler>
+    Conf;
+
+std::atomic<size_t> idx;
+std::atomic<bool> inserts_done = false;
+
+ssize_t query_ratio = 0;
+
+std::atomic<size_t> total_res = 0;
+size_t reccnt = 0;
+
+size_t g_thrd_cnt = 0;
+
+std::atomic<size_t> total_insert_time = 0;
+std::atomic<size_t> total_insert_count = 0;
+std::atomic<size_t> total_query_time = 0;
+std::atomic<size_t> total_query_count = 0;
+
+void insert_thread(Ext *extension, std::vector<Rec> *records, size_t start_idx,
+                   size_t stop_idx, gsl_rng *rng) {
+
+  for (size_t i = start_idx; i < stop_idx; i++) {
+    while (!extension->insert((*records)[i], rng)) {
+      usleep(1);
+    }
+  }
+}
+
+void usage(char *progname) {
+  fprintf(stderr, "%s reccnt datafile rate_limit policy\n", progname);
+}
+
+int main(int argc, char **argv) {
+
+  if (argc < 5) {
+    usage(argv[0]);
+    exit(EXIT_FAILURE);
+  }
+
+  size_t n = atol(argv[1]);
+  std::string d_fname = std::string(argv[2]);
+  double rate_limit = std::atof(argv[3]);
+  size_t pol = std::atol(argv[4]);
+  assert(pol >= 0 && pol <= 6);
+
+  auto data = read_vector_file<Rec, 300>(d_fname, n);
+
+  size_t buffer_size = 1000;
+  size_t scale_factor = 8;
+  double modifier = 0;
+  size_t insert_threads = 1;
+  size_t internal_thread_cnt = 32;
+  reccnt = n;
+
+  gsl_rng *rng = gsl_rng_alloc(gsl_rng_mt19937);
+
+  auto policy =
+      get_policy<Shard, Q>(scale_factor, buffer_size, pol, n, modifier);
+  auto config = Conf(std::move(policy));
+  config.recon_enable_maint_on_flush = true;
+  config.recon_maint_disabled = false;
+  config.buffer_size = buffer_size;
+  config.buffer_flush_trigger = buffer_size;
+  // config.buffer_flush_trigger = 4000;
+  config.maximum_threads = internal_thread_cnt;
+  config.rt_level_scale = scale_factor;
+
+  if (rate_limit == 0) {
+    config.dynamic_ratelimiting = true;
+  }
+
+  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), rate_limit);
+
+  /* warmup structure w/ 30% of records */
+  size_t warmup = .3 * n;
+  for (size_t k = 0; k < warmup; k++) {
+    while (!extension->insert(data[k])) {
+      usleep(1);
+    }
+  }
+
+  extension->await_version();
+
+  idx.store(warmup);
+
+  std::thread i_thrds[insert_threads];
+
+  size_t per_insert_thrd = (n - warmup) / insert_threads;
+  size_t start = warmup;
+
+  TIMER_INIT();
+  TIMER_START();
+  for (size_t i = 0; i < insert_threads; i++) {
+    i_thrds[i] = std::thread(insert_thread, extension, &data, start,
+                             start + per_insert_thrd, rng);
+    start += per_insert_thrd;
+  }
+
+  for (size_t i = 0; i < insert_threads; i++) {
+    i_thrds[i].join();
+  }
+  TIMER_STOP();
+
+  size_t insert_tput = ((double)(n - warmup) / (double)TIMER_RESULT()) * 1e9;
+  fprintf(stdout, "Insertion Tput: %ld\n", insert_tput);
+
+  inserts_done.store(true);
+  inserts_done.store(false);
+  delete extension;
+
+  fflush(stderr);
+}
diff --git a/benchmarks/tail-latency/stall_benchmark_vptree.cpp b/benchmarks/tail-latency/stall_benchmark_vptree.cpp
index 26df573..f89617b 100644
--- a/benchmarks/tail-latency/stall_benchmark_vptree.cpp
+++ b/benchmarks/tail-latency/stall_benchmark_vptree.cpp
@@ -57,13 +57,12 @@ void insert_thread(Ext *extension, std::vector<Rec> *records, size_t start_idx,
   TIMER_INIT();
 
   for (size_t i = start_idx; i < stop_idx; i++) {
-    TIMER_START();
+  TIMER_START();
     while (!extension->insert((*records)[i], rng)) {
       usleep(1);
     }
     TIMER_STOP();
-
-    fprintf(stdout, "I\t%ld\n", TIMER_RESULT());
+      fprintf(stdout, "I\t%ld\n", TIMER_RESULT());
   }
 }
 
@@ -103,6 +102,11 @@ int main(int argc, char **argv) {
   config.buffer_flush_trigger = buffer_size;
   // config.buffer_flush_trigger = 4000;
   config.maximum_threads = internal_thread_cnt;
+  config.rt_level_scale = scale_factor;
+
+  if (rate_limit == 0) {
+    config.dynamic_ratelimiting = true;
+  }
 
   g_thrd_cnt = internal_thread_cnt;
 
diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h
index 63264a0..c062d4a 100644
--- a/include/framework/DynamicExtension.h
+++ b/include/framework/DynamicExtension.h
@@ -10,6 +10,7 @@
 #pragma once
 
 #include <atomic>
+#include <chrono>
 #include <cstdio>
 #include <gsl/gsl_rng.h>
 #include <mutex>
@@ -544,7 +545,13 @@ private:
 
       reconstruction_results<ShardType> flush_recon;
       flush_recon.target_level = 0;
+      auto start = std::chrono::high_resolution_clock::now();
       flush_recon.new_shard = std::make_shared<ShardType>(std::move(buffview));
+      auto stop = std::chrono::high_resolution_clock::now();
+      flush_recon.runtime =
+          std::chrono::duration_cast<std::chrono::nanoseconds>(stop - start)
+              .count();
+      flush_recon.reccnt = flush_recon.new_shard->get_record_count();
 
       reconstructions.push_back(flush_recon);
 
@@ -637,6 +644,13 @@ private:
 
     /* apply our updates to the copied structure (adding/removing shards) */
     for (auto recon : reconstructions) {
+      if (recon.target_level != 0) {
+        #ifdef DE_PRINT_MODEL_OUTPUT
+        fprintf(stdout, "M\t%ld\t%ld\t%ld\t%ld\n", recon.target_level,
+                args->predicted_runtime, recon.runtime, recon.reccnt);
+        #endif
+      }
+
       auto grow = args->version->get_mutable_structure()->apply_reconstruction(
           recon, args->version->get_id());
       if (grow) {
diff --git a/include/framework/structure/InternalLevel.h b/include/framework/structure/InternalLevel.h
index cf2b16a..88b3fef 100644
--- a/include/framework/structure/InternalLevel.h
+++ b/include/framework/structure/InternalLevel.h
@@ -262,7 +262,7 @@ private:
   ssize_t m_level_no;
   std::vector<shard_ptr> m_shards;
 
-  const size_t m_window_size = 15;
+  const size_t m_window_size = 1;
 
   std::deque<size_t> m_rt_window;
 };