From f1316e313de5c5286b279cec6ed320cba3eb506f Mon Sep 17 00:00:00 2001 From: Douglas Rumbaugh Date: Mon, 17 Feb 2025 16:40:11 -0500 Subject: Average version of mixed-workload bench --- benchmarks/tail-latency/mixed_workload.cpp | 18 ++- benchmarks/tail-latency/mixed_workload_average.cpp | 176 +++++++++++++++++++++ 2 files changed, 186 insertions(+), 8 deletions(-) create mode 100644 benchmarks/tail-latency/mixed_workload_average.cpp (limited to 'benchmarks') diff --git a/benchmarks/tail-latency/mixed_workload.cpp b/benchmarks/tail-latency/mixed_workload.cpp index 517ccd8..f4bfdda 100644 --- a/benchmarks/tail-latency/mixed_workload.cpp +++ b/benchmarks/tail-latency/mixed_workload.cpp @@ -43,7 +43,7 @@ size_t query_ratio = 3; std::atomic total_res = 0; size_t reccnt = 0; -size_t thrd_cnt = 0; +size_t g_thrd_cnt = 0; void operation_thread(Ext *extension, std::vector *queries, std::vector *records) { @@ -60,7 +60,7 @@ void operation_thread(Ext *extension, std::vector *queries, auto res = extension->query(std::move(q)).get(); TIMER_STOP(); - fprintf(stdout, "Q\t%ld\t%ld\n", thrd_cnt, TIMER_RESULT()); + fprintf(stdout, "Q\t%ld\t%ld\n", g_thrd_cnt, TIMER_RESULT()); total_res.fetch_add(res); @@ -78,7 +78,7 @@ void operation_thread(Ext *extension, std::vector *queries, } TIMER_STOP(); - fprintf(stdout, "I\t%ld\t%ld\n", thrd_cnt, TIMER_RESULT()); + fprintf(stdout, "I\t%ld\t%ld\n", g_thrd_cnt, TIMER_RESULT()); if (idx.load() == reccnt) { inserts_done.store(true); @@ -117,22 +117,22 @@ int main(int argc, char **argv) { reccnt = n; for (auto pol : policies) { - for (size_t i = 0; i < thread_counts.size(); i++) { + for (auto internal_thread_cnt : thread_counts) { auto policy = get_policy(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 = thread_counts[i]; + config.maximum_threads = internal_thread_cnt; - thrd_cnt = thread_counts[i]; + g_thrd_cnt = internal_thread_cnt; auto extension = new Ext(std::move(config)); /* warmup structure w/ 10% of records */ size_t warmup = .1 * n; - for (size_t j = 0; j < warmup; j++) { - while (!extension->insert(data[j])) { + for (size_t k = 0; k < warmup; k++) { + while (!extension->insert(data[k])) { usleep(1); } } @@ -153,6 +153,8 @@ int main(int argc, char **argv) { } fprintf(stderr, "%ld\n", total_res.load()); + total_res.store(0); + inserts_done.store(false); delete extension; } } diff --git a/benchmarks/tail-latency/mixed_workload_average.cpp b/benchmarks/tail-latency/mixed_workload_average.cpp new file mode 100644 index 0000000..cbf3a82 --- /dev/null +++ b/benchmarks/tail-latency/mixed_workload_average.cpp @@ -0,0 +1,176 @@ +/* + * + */ + +#define ENABLE_TIMER +#define TS_TEST + +#include + +#include "framework/scheduling/SerialScheduler.h" +#include "framework/util/Configuration.h" +#include "util/types.h" +#include "file_util.h" +#include "framework/DynamicExtension.h" +#include "framework/interface/Record.h" +#include "framework/scheduling/FIFOScheduler.h" +#include "query/rangecount.h" +#include "shard/TrieSpline.h" +#include "standard_benchmarks.h" + +#include "framework/reconstruction/FixedShardCountPolicy.h" + +#include + +#include "psu-util/timer.h" + +typedef de::Record Rec; +typedef de::TrieSpline Shard; +typedef de::rc::Query Q; +typedef de::DynamicExtension + Ext; +typedef Q::Parameters QP; +typedef de::DEConfiguration + Conf; + +std::atomic idx; +std::atomic inserts_done = false; + +size_t query_ratio = 3; + +std::atomic total_res = 0; +size_t reccnt = 0; + +size_t g_thrd_cnt = 0; + +std::atomic total_insert_time = 0; +std::atomic total_insert_count = 0; +std::atomic total_query_time = 0; +std::atomic total_query_count = 0; + +void operation_thread(Ext *extension, std::vector *queries, + std::vector *records) { + TIMER_INIT(); + while (!inserts_done.load()) { + auto type = rand() % 10; + + if (type < 8) { + total_query_count.fetch_add(1); + auto q_idx = rand() % queries->size(); + + auto q = (*queries)[q_idx]; + + TIMER_START(); + auto res = extension->query(std::move(q)).get(); + TIMER_STOP(); + + total_query_time.fetch_add(TIMER_RESULT()); + total_res.fetch_add(res); + } else { + TIMER_START(); + for (size_t i = 0; i < 1000; i++) { + auto insert_idx = idx.fetch_add(1); + if (insert_idx >= reccnt) { + inserts_done.store(true); + break; + } + + while (!extension->insert((*records)[insert_idx])) { + usleep(1); + } + + if (idx.load() == reccnt) { + inserts_done.store(true); + } + } + TIMER_STOP(); + total_insert_time.fetch_add(TIMER_RESULT()); + } + } +} + +void usage(char *progname) { + fprintf(stderr, "%s reccnt datafile queryfile\n", progname); +} + +int main(int argc, char **argv) { + + if (argc < 4) { + usage(argv[0]); + exit(EXIT_FAILURE); + } + + size_t n = atol(argv[1]); + std::string d_fname = std::string(argv[2]); + std::string q_fname = std::string(argv[3]); + + auto data = read_sosd_file(d_fname, n); + auto queries = read_range_queries(q_fname, .0001); + + std::vector sfs = {8}; //, 4, 8, 16, 32, 64, 128, 256, 512, 1024}; + size_t buffer_size = 8000; + std::vector policies = { + 5 + }; + + std::vector thread_counts = {1, 2, 4, 8, 16, 32}; + + reccnt = n; + + for (auto pol : policies) { + for (auto internal_thread_cnt : thread_counts) { + auto policy = get_policy(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 = .1 * n; + for (size_t k = 0; k < warmup; k++) { + while (!extension->insert(data[k])) { + usleep(1); + } + } + + extension->await_version(); + + idx.store(warmup); + + size_t thrd_cnt = 8; + std::thread thrds[thrd_cnt]; + + for (size_t i=0; i