Montage as an instance

Makefile

@@ -118,7 +118,7 @@ ARCHIVEDIR:=./lib
 # -since we do pattern matching between this list and the
 # source files, the file path specified must be the same
 # type (absolute or relative)
-EXECUTABLES:= ./src/main.cpp ./unit_test/dcss.cpp
+EXECUTABLES:= ./src/main.cpp #./unit_test/dcss.cpp
 
 # A list of source files contained in the
 # source directory to exclude from the build

ext/threadcached/include/memcached.h

@@ -398,7 +398,7 @@ extern unsigned stats_id;
  * Structure for storing items within memcached.
  */
 #ifdef MONTAGE
-#include "persist_struct_api.hpp"
+#include "montage_global_api.hpp"
 
 using namespace pds;
 struct item : public PBlk{

ext/ycsb-tcd/ycsbc.cc

@@ -23,7 +23,7 @@ using namespace std;
 
 #ifdef MONTAGE
 #include "TestConfig.hpp"
-#include "persist_struct_api.hpp"
+#include "montage_global_api.hpp"
 #endif /* MONTAGE */
 
 void UsageMessage(const char *command);
@@ -131,7 +131,7 @@ int main(const int argc, char *argv[]) {
   hwloc_get_type_depth(gtc.topology, HWLOC_OBJ_PU));
   std::cout<<"initial affinity built"<<std::endl;
   gtc.buildAffinity(gtc.affinities);
-  pds::init(&gtc);
+  // pds::init(&gtc);
 #endif
   if (do_cache_test_flag){
     do_cache_test();

src/Rideable.hpp

@@ -10,9 +10,11 @@
 
 
 class GlobalTestConfig;
+class LocalTestConfig;
 
 class Rideable{
 public:
+	virtual void init_thread(GlobalTestConfig* gtc, LocalTestConfig* ltc){};
 	virtual ~Rideable(){};
 };
 

src/main.cpp

@@ -71,7 +71,6 @@
 #include "MapTest.hpp"
 #include "MapChurnTest.hpp"
 #ifndef MNEMOSYNE
-#include "EpochSysVerifyTest.hpp"
 #include "RecoverVerifyTest.hpp"
 #include "GraphRecoveryTest.hpp"
 #include "TGraphConstructionTest.hpp"

src/persist/EpochAdvancers.cpp

@@ -0,0 +1,85 @@
+#include "EpochSys.hpp"
+#include "EpochAdvancers.hpp"
+
+using namespace pds;
+
+SingleThreadEpochAdvancer::SingleThreadEpochAdvancer(GlobalTestConfig* gtc){
+    trans_cnts = new padded<uint64_t>[gtc->task_num];
+    for (int i = 0; i < gtc->task_num; i++){
+        trans_cnts[i].ui = 0;
+    }
+}
+void SingleThreadEpochAdvancer::set_epoch_freq(int epoch_power){
+    epoch_threshold = 0x1ULL << epoch_power;
+}
+void SingleThreadEpochAdvancer::set_help_freq(int help_power){
+    help_threshold = 0x1ULL << help_power;
+}
+void SingleThreadEpochAdvancer::on_end_transaction(EpochSys* esys, uint64_t c){
+    assert(EpochSys::tid != -1);
+    trans_cnts[EpochSys::tid].ui++;
+    if (EpochSys::tid == 0){
+        // only a single thread can advance epochs.
+        if (trans_cnts[EpochSys::tid].ui % epoch_threshold == 0){
+            esys->advance_epoch(c);
+        } 
+    }
+}
+
+
+void GlobalCounterEpochAdvancer::set_epoch_freq(int epoch_power){
+    epoch_threshold = 0x1ULL << epoch_power;
+}
+void GlobalCounterEpochAdvancer::set_help_freq(int help_power){
+    help_threshold = 0x1ULL << help_power;
+}
+void GlobalCounterEpochAdvancer::on_end_transaction(EpochSys* esys, uint64_t c){
+    uint64_t curr_cnt = trans_cnt.fetch_add(1, std::memory_order_acq_rel);
+    if (curr_cnt % epoch_threshold == 0){
+        esys->advance_epoch(c);
+    } else if (curr_cnt % help_threshold == 0){
+        esys->help_local();
+    }
+}
+
+
+DedicatedEpochAdvancer::DedicatedEpochAdvancer(GlobalTestConfig* gtc, EpochSys* es):esys(es){
+    if (gtc->checkEnv("EpochLength")){
+        epoch_length = stoi(gtc->getEnv("EpochLength"));
+    } else {
+        epoch_length = 100*1000;
+    }
+    if (gtc->checkEnv("EpochLengthUnit")){
+        std::string env_unit = gtc->getEnv("EpochLengthUnit");
+        if (env_unit == "Second"){
+            epoch_length *= 1000000;
+        } else if (env_unit == "Millisecond"){
+            epoch_length *= 1000;
+        } else if (env_unit == "Microsecond"){
+            // do nothing.
+        } else {
+            errexit("time unit not supported.");
+        }
+    }
+    started.store(false);
+    advancer_thread = std::move(std::thread(&DedicatedEpochAdvancer::advancer, this));
+    started.store(true);
+}
+
+void DedicatedEpochAdvancer::advancer(){
+    while(!started.load()){}
+    while(started.load()){
+        esys->advance_epoch_dedicated();
+        std::this_thread::sleep_for(std::chrono::microseconds(epoch_length));
+    }
+    // std::cout<<"advancer_thread terminating..."<<std::endl;
+}
+
+DedicatedEpochAdvancer::~DedicatedEpochAdvancer(){
+    // std::cout<<"terminating advancer_thread"<<std::endl;
+    started.store(false);
+    if (advancer_thread.joinable()){
+        advancer_thread.join();
+    }
+    // std::cout<<"terminated advancer_thread"<<std::endl;
+}

src/persist/EpochAdvancers.hpp

@@ -0,0 +1,75 @@
+#ifndef EPOCHADVANCERS_HPP
+#define EPOCHADVANCERS_HPP
+
+#include <atomic>
+#include <thread>
+#include "TestConfig.hpp"
+#include "ConcurrentPrimitives.hpp"
+
+namespace pds{
+
+class EpochSys;
+
+/////////////////////
+// Epoch Advancers //
+/////////////////////
+
+class EpochAdvancer{
+public:
+    virtual void set_epoch_freq(int epoch_freq) = 0;
+    virtual void set_help_freq(int help_freq) = 0;
+    virtual void on_end_transaction(EpochSys* esys, uint64_t c) = 0;
+    virtual ~EpochAdvancer(){}
+};
+
+class SingleThreadEpochAdvancer : public EpochAdvancer{
+    // uint64_t trans_cnt;
+    padded<uint64_t>* trans_cnts;
+    uint64_t epoch_threshold = 0x1ULL << 19;
+    uint64_t help_threshold = 0x1ULL << 6;
+public:
+    SingleThreadEpochAdvancer(GlobalTestConfig* gtc);
+    void set_epoch_freq(int epoch_power);
+    void set_help_freq(int help_power);
+    void on_end_transaction(EpochSys* esys, uint64_t c);
+};
+
+class GlobalCounterEpochAdvancer : public EpochAdvancer{
+    std::atomic<uint64_t> trans_cnt;
+    uint64_t epoch_threshold = 0x1ULL << 14;
+    uint64_t help_threshold = 0x1ULL;
+public:
+    // GlobalCounterEpochAdvancer();
+    void set_epoch_freq(int epoch_power);
+    void set_help_freq(int help_power);
+    void on_end_transaction(EpochSys* esys, uint64_t c);
+};
+
+class DedicatedEpochAdvancer : public EpochAdvancer{
+    EpochSys* esys;
+    std::thread advancer_thread;
+    std::atomic<bool> started;
+    uint64_t epoch_length = 100*1000;
+    void advancer();
+public:
+    DedicatedEpochAdvancer(GlobalTestConfig* gtc, EpochSys* es);
+    ~DedicatedEpochAdvancer();
+    void set_epoch_freq(int epoch_interval){}
+    void set_help_freq(int help_interval){}
+    void on_end_transaction(EpochSys* esys, uint64_t c){
+        // do nothing here.
+    }
+};
+
+class NoEpochAdvancer : public EpochAdvancer{
+    // an epoch advancer that does absolutely nothing.
+public:
+    // GlobalCounterEpochAdvancer();
+    void set_epoch_freq(int epoch_power){}
+    void set_help_freq(int help_power){}
+    void on_end_transaction(EpochSys* esys, uint64_t c){}
+};
+
+}
+
+#endif

src/persist/EpochSys.cpp

@@ -1,20 +1,10 @@
 #include "EpochSys.hpp"
-#include "DCSS.hpp"
 
 #include <omp.h>
 #include <atomic>
 namespace pds{
 
-    __thread int _tid = -1;
-    EpochSys* esys = nullptr;
-    padded<uint64_t>* epochs = nullptr;
-    SysMode sys_mode = ONLINE;
-    // std::atomic<size_t> abort_cnt(0);
-    // std::atomic<size_t> total_cnt(0);
-
-    UIDGenerator PBlk::uid_generator;
-
-    padded<sc_desc_t>* local_descs = nullptr;
+    thread_local int EpochSys::tid = -1;
 
     void EpochSys::parse_env(){
         if (to_be_persisted){
@@ -136,8 +126,8 @@ namespace pds{
         do{
             ret = global_epoch->load(std::memory_order_seq_cst);
         } while(!trans_tracker->consistent_register_active(ret, ret));
-        //to_be_freed->free_on_new_epoch(ret);
         to_be_freed->free_on_new_epoch(ret);
+
         return ret;
     }