diff options
| -rw-r--r-- | CMakeLists.txt | 5 | ||||
| -rw-r--r-- | benchmarks/insert_query_tput.cpp | 87 | ||||
| -rw-r--r-- | benchmarks/insert_query_tput_serial.cpp | 104 | ||||
| -rw-r--r-- | include/framework/DynamicExtension.h | 7 |
4 files changed, 162 insertions, 41 deletions
diff --git a/CMakeLists.txt b/CMakeLists.txt index 1dbacca..3f6ea4c 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -135,6 +135,11 @@ if (bench) target_include_directories(insert_query_tput PRIVATE include external external/m-tree/cpp external/PGM-index/include external/PLEX/include bench/include external/psudb-common/cpp/include) target_link_options(insert_query_tput PUBLIC -mcx16) + add_executable(insert_query_tput_serial ${CMAKE_CURRENT_SOURCE_DIR}/benchmarks/insert_query_tput_serial.cpp) + target_link_libraries(insert_query_tput_serial PUBLIC gsl pthread gomp atomic) + target_include_directories(insert_query_tput_serial PRIVATE include external external/m-tree/cpp external/PGM-index/include external/PLEX/include bench/include external/psudb-common/cpp/include) + target_link_options(insert_query_tput_serial PUBLIC -mcx16) + add_executable(watermark_testing ${CMAKE_CURRENT_SOURCE_DIR}/benchmarks/watermark_testing.cpp) target_link_libraries(watermark_testing PUBLIC gsl pthread gomp atomic) target_include_directories(watermark_testing PRIVATE include external external/m-tree/cpp external/PGM-index/include external/PLEX/include bench/include external/psudb-common/cpp/include) diff --git a/benchmarks/insert_query_tput.cpp b/benchmarks/insert_query_tput.cpp index 865e82c..8844d04 100644 --- a/benchmarks/insert_query_tput.cpp +++ b/benchmarks/insert_query_tput.cpp @@ -23,71 +23,82 @@ typedef de::DynamicExtension<Rec, ISAM, Q> Ext; std::atomic<bool> inserts_done = false; -void insert_thread(Ext *extension, size_t n, size_t k) { + +void query_thread(Ext *extension, double selectivity, size_t k, gsl_rng *rng) { TIMER_INIT(); - for (int64_t i=0; i<n; i+=k) { - TIMER_START(); - for (int64_t j=0; j<k; j++) { - Rec r = {i+j, i+j}; - while (!extension->insert(r)) { - _mm_pause(); - } - } - TIMER_STOP(); - auto insert_lat = TIMER_RESULT(); - fprintf(stdout, "I\t%ld\t%ld\t%ld\n", extension->get_record_count(), insert_lat, k); - } + size_t reccnt = extension->get_record_count(); - inserts_done.store(true); -} + size_t range = reccnt * selectivity; -void query_thread(Ext *extension, double selectivity, size_t k, gsl_rng *rng) { - TIMER_INIT(); + auto q = new de::rc::Parms<Rec>(); - while (!inserts_done.load()) { - size_t reccnt = extension->get_record_count(); - if (reccnt == 0) { - continue; // don't start querying until there is data - } + TIMER_START(); + for (int64_t i=0; i<k; i++) { + size_t start = gsl_rng_uniform_int(rng, reccnt - range); - size_t range = reccnt * selectivity; + q->lower_bound = start; + q->upper_bound = start + range; + auto res = extension->query(q); + auto r = res.get(); + } + TIMER_STOP(); + auto query_lat = TIMER_RESULT(); + fprintf(stdout, "Q\t%ld\t%ld\t%ld\n", reccnt, query_lat, k); + delete q; +} + +void insert_thread(Ext *extension, size_t n, size_t k, gsl_rng *rng) { + TIMER_INIT(); - auto q = new de::rc::Parms<Rec>(); + size_t reccnt = 0; + Rec r; + while (reccnt < n) { + auto old_reccnt = reccnt; TIMER_START(); - for (int64_t i=0; i<k; i++) { - size_t start = gsl_rng_uniform_int(rng, reccnt - range); + for (size_t i=0; i<k; i++) { + r.key = reccnt; + r.value = reccnt; - q->lower_bound = start; - q->upper_bound = start + range; - auto res = extension->query(q); - auto r = res.get(); + if (extension->insert(r)) { + reccnt++; + } } TIMER_STOP(); - auto query_lat = TIMER_RESULT(); - fprintf(stdout, "Q\t%ld\t%ld\t%ld\n", reccnt, query_lat, k); - delete q; + auto insert_lat = TIMER_RESULT(); + + fprintf(stdout, "I\t%ld\t%ld\t%ld\n", reccnt, insert_lat, reccnt - old_reccnt); + + if (reccnt % 100000 == 0 && reccnt != n) { + auto a = std::thread(query_thread, extension, .01, 20, rng); + a.detach(); + } } } int main(int argc, char **argv) { /* the closeout routine takes _forever_ ... so we'll just leak the memory */ - auto extension = new Ext(1000, 10000, 2); + auto extension = new Ext(1000, 12000, 8); size_t n = 10000000; size_t per_trial = 1000; double selectivity = .001; - gsl_rng * rng = gsl_rng_alloc(gsl_rng_mt19937); + TIMER_INIT(); - std::thread i_thrd(insert_thread, extension, n, per_trial); - std::thread q_thrd(query_thread, extension, selectivity, 1, rng); + gsl_rng * rng = gsl_rng_alloc(gsl_rng_mt19937); - q_thrd.join(); + TIMER_START(); + std::thread i_thrd(insert_thread, extension, n, per_trial, rng); i_thrd.join(); + TIMER_STOP(); + + auto total_latency = TIMER_RESULT(); + fprintf(stdout, "T\t%ld\n", total_latency); gsl_rng_free(rng); + delete extension; fflush(stderr); } diff --git a/benchmarks/insert_query_tput_serial.cpp b/benchmarks/insert_query_tput_serial.cpp new file mode 100644 index 0000000..ff9dc40 --- /dev/null +++ b/benchmarks/insert_query_tput_serial.cpp @@ -0,0 +1,104 @@ +/* + * + */ + +#define ENABLE_TIMER + +#include <thread> + +#include "framework/DynamicExtension.h" +#include "shard/ISAMTree.h" +#include "query/rangecount.h" +#include "framework/interface/Record.h" + +#include <gsl/gsl_rng.h> + +#include "psu-util/timer.h" + + +typedef de::Record<int64_t, int64_t> Rec; +typedef de::ISAMTree<Rec> ISAM; +typedef de::rc::Query<ISAM, Rec> Q; +typedef de::DynamicExtension<Rec, ISAM, Q, de::LayoutPolicy::TEIRING, de::DeletePolicy::TAGGING, de::SerialScheduler> Ext; + +std::atomic<bool> inserts_done = false; + + +void query_thread(Ext *extension, double selectivity, size_t k, gsl_rng *rng) { + TIMER_INIT(); + + size_t reccnt = extension->get_record_count(); + + size_t range = reccnt * selectivity; + + auto q = new de::rc::Parms<Rec>(); + + TIMER_START(); + for (int64_t i=0; i<k; i++) { + size_t start = gsl_rng_uniform_int(rng, reccnt - range); + + q->lower_bound = start; + q->upper_bound = start + range; + auto res = extension->query(q); + auto r = res.get(); + } + TIMER_STOP(); + auto query_lat = TIMER_RESULT(); + fprintf(stdout, "Q\t%ld\t%ld\t%ld\n", reccnt, query_lat, k); + delete q; +} + +void insert_thread(Ext *extension, size_t n, size_t k, gsl_rng *rng) { + TIMER_INIT(); + + size_t reccnt = 0; + Rec r; + while (reccnt < n) { + auto old_reccnt = reccnt; + + TIMER_START(); + for (size_t i=0; i<k; i++) { + r.key = reccnt; + r.value = reccnt; + + if (extension->insert(r)) { + reccnt++; + } + } + TIMER_STOP(); + auto insert_lat = TIMER_RESULT(); + + fprintf(stdout, "I\t%ld\t%ld\t%ld\n", reccnt, insert_lat, reccnt - old_reccnt); + + if (reccnt % 100000 == 0 && reccnt != n) { + auto a = std::thread(query_thread, extension, .01, 20, rng); + a.join(); + } + } +} + +int main(int argc, char **argv) { + + /* the closeout routine takes _forever_ ... so we'll just leak the memory */ + auto extension = new Ext(1000, 12000, 8); + size_t n = 10000000; + size_t per_trial = 1000; + double selectivity = .001; + + TIMER_INIT(); + + gsl_rng * rng = gsl_rng_alloc(gsl_rng_mt19937); + + TIMER_START(); + std::thread i_thrd(insert_thread, extension, n, per_trial, rng); + i_thrd.join(); + TIMER_STOP(); + + auto total_latency = TIMER_RESULT(); + fprintf(stdout, "T\t%ld\n", total_latency); + + gsl_rng_free(rng); + delete extension; + fflush(stderr); +} + diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h index 3203945..a10831e 100644 --- a/include/framework/DynamicExtension.h +++ b/include/framework/DynamicExtension.h @@ -328,14 +328,15 @@ private: */ enforce_delete_invariant(new_epoch); - // FIXME: this may currently fail because there isn't any + // FIXME: this may currently block because there isn't any // query preemption yet. At this point, we'd need to either // 1) wait for all queries on the old_head to finish // 2) kill all queries on the old_head // 3) somehow migrate all queries on the old_head to the new // version - auto res = new_epoch->advance_buffer_head(buffer_head); - assert(res); + while (!new_epoch->advance_buffer_head(buffer_head)) { + _mm_pause(); + } m_current_epoch.fetch_add(1); old_epoch->set_inactive(); |