Updated throughput bench to use SOSD

2 files changed, 39 insertions, 21 deletions

diff --git a/benchmarks/include/data-proc.h b/benchmarks/include/data-proc.h
index f758ed4..dbac671 100644
--- a/benchmarks/include/data-proc.h
+++ b/benchmarks/include/data-proc.h
@@ -10,6 +10,8 @@
 
 #include "psu-ds/BTree.h"
 
+#pragma once
+
 typedef uint64_t key_type;
 typedef uint64_t value_type;
 typedef uint64_t weight_type;
@@ -242,3 +244,15 @@ static bool build_delete_vec(std::vector<R> &to_delete, std::vector<R> &vec, siz
 td:
     return true;
 }
+
+static std::vector<int64_t> read_sosd_file(std::string &fname, size_t n) {
+    std::fstream file;
+    file.open(fname, std::ios::in | std::ios::binary);
+
+    std::vector<int64_t> records(n);
+    for (size_t i=0; i<n; i++) {
+        file.read((char*) &(records[i]), sizeof(int64_t));
+    }
+
+    return records;
+}
diff --git a/benchmarks/insert_query_tput.cpp b/benchmarks/insert_query_tput.cpp
index 05715b1..40a5f8d 100644
--- a/benchmarks/insert_query_tput.cpp
+++ b/benchmarks/insert_query_tput.cpp
@@ -10,6 +10,7 @@
 #include "shard/ISAMTree.h"
 #include "query/irs.h"
 #include "framework/interface/Record.h"
+#include "include/data-proc.h"
 
 #include <gsl/gsl_rng.h>
 
@@ -20,23 +21,22 @@ typedef de::Record<int64_t, int64_t> Rec;
 typedef de::ISAMTree<Rec> ISAM;
 typedef de::irs::Query<ISAM, Rec> Q;
 typedef de::DynamicExtension<Rec, ISAM, Q> Ext;
+typedef de::irs::Parms<Rec> QP;
 
 std::atomic<bool> inserts_done = false;
 
-void query_thread(Ext *extension, size_t n) {
+void query_thread(Ext *extension, std::vector<QP> *queries) {
     gsl_rng *rng = gsl_rng_alloc(gsl_rng_mt19937);
-    size_t range = n*.0001;
+    size_t total = 0;
 
-    int64_t total = 0;
-
-    de::irs::Parms<Rec> *q = new de::irs::Parms<Rec>();
     while (!inserts_done.load()) {
-        size_t start = gsl_rng_uniform_int(rng, n - range);
-        q->lower_bound = start;
-        q->upper_bound = start + range;
-        q->sample_size = 100;
-        q->rng = rng;
-        auto res = extension->query(q);
+        auto q_idx = gsl_rng_uniform_int(rng, queries->size());
+
+        auto q = (*queries)[q_idx];
+        q.rng = rng;
+        q.sample_size = 1000;
+
+        auto res = extension->query(&q);
         auto r = res.get();
         total += r.size();
         usleep(1);
@@ -45,15 +45,14 @@ void query_thread(Ext *extension, size_t n) {
     fprintf(stderr, "%ld\n", total);
 
     gsl_rng_free(rng);
-    delete q;
 }
 
-void insert_thread(Ext *extension, size_t n, gsl_rng *rng) {
+void insert_thread(Ext *extension, size_t start, std::vector<int64_t> *records) {
     size_t reccnt = 0;
     Rec r;
-    for (size_t i=0; i<n; i++) {
-        r.key = gsl_rng_uniform_int(rng, n);
-        r.value = gsl_rng_uniform_int(rng, n);
+    for (size_t i=start; i<records->size(); i++) {
+        r.key = (*records)[i];
+        r.value = i;
 
         while (!extension->insert(r)) {
             usleep(1);
@@ -65,22 +64,27 @@ void insert_thread(Ext *extension, size_t n, gsl_rng *rng) {
 
 int main(int argc, char **argv) {
 
-    if (argc < 3) {
-        fprintf(stderr, "insert_query_tput reccnt query_threads\n");
+    if (argc < 5) {
+        fprintf(stderr, "insert_query_tput reccnt query_threads datafile queryfile\n");
         exit(EXIT_FAILURE);
     }
 
     size_t n = atol(argv[1]);
     size_t qthread_cnt = atol(argv[2]);
+    std::string d_fname = std::string(argv[3]);
+    std::string q_fname = std::string(argv[4]);
 
     auto extension = new Ext(1000, 12000, 8, 0, 64);
     gsl_rng * rng = gsl_rng_alloc(gsl_rng_mt19937);
+    
+    auto data = read_sosd_file(d_fname, n);
+    auto queries = read_range_queries<QP>(q_fname, .001);
 
     /* warmup structure w/ 10% of records */
     size_t warmup = .1 * n;
     Rec r;
     for (size_t i=0; i<warmup; i++) {
-        r.key = gsl_rng_uniform_int(rng, n);
+        r.key = data[i];
         r.value = gsl_rng_uniform_int(rng, n);
 
         while (!extension->insert(r)) {
@@ -95,9 +99,9 @@ int main(int argc, char **argv) {
     std::vector<std::thread> qthreads(qthread_cnt);
 
     TIMER_START();
-    std::thread i_thrd(insert_thread, extension, n - warmup, rng);
+    std::thread i_thrd(insert_thread, extension, warmup, &data);
     for (size_t i=0; i<qthread_cnt; i++) {
-        qthreads[i] = std::thread(query_thread, extension, n);
+        qthreads[i] = std::thread(query_thread, extension, &queries);
     }
     i_thrd.join();
     TIMER_STOP();