Progress: began adding parallel merging and locking of levels

7 files changed, 35 insertions, 37 deletions

diff --git a/include/framework/reconstruction/BSMPolicy.h b/include/framework/reconstruction/BSMPolicy.h
index 9ddd150..6d55a12 100644
--- a/include/framework/reconstruction/BSMPolicy.h
+++ b/include/framework/reconstruction/BSMPolicy.h
@@ -24,10 +24,9 @@ public:
   BSMPolicy(size_t buffer_size)
       : m_scale_factor(2), m_buffer_size(buffer_size) {}
 
-  ReconstructionVector
-  get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const override {
-    ReconstructionVector reconstructions;
-    return reconstructions;
+  std::vector<ReconstructionVector>
+  get_reconstruction_tasks(const Version<ShardType, QueryType> *version, LockManager &lock_mngr) const override {
+    return {};
   }
 
   ReconstructionVector
diff --git a/include/framework/reconstruction/BackgroundTieringPolicy.h b/include/framework/reconstruction/BackgroundTieringPolicy.h
index 4c266fd..5a82695 100644
--- a/include/framework/reconstruction/BackgroundTieringPolicy.h
+++ b/include/framework/reconstruction/BackgroundTieringPolicy.h
@@ -24,14 +24,14 @@ public:
   BackgroundTieringPolicy(size_t scale_factor, size_t buffer_size)
       : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {}
 
-  ReconstructionVector get_reconstruction_tasks(
-      const Version<ShardType, QueryType> *version) const override {
-    ReconstructionVector reconstructions;
+  std::vector<ReconstructionVector> get_reconstruction_tasks(
+      const Version<ShardType, QueryType> *version, LockManager &lock_mngr) const override {
+    std::vector<ReconstructionVector> reconstructions;
 
     auto levels = version->get_structure()->get_level_vector();
 
-    if (levels[0]->get_shard_count() < m_scale_factor) {
-      return reconstructions;
+    if (levels[0]->get_shard_count() == 0) {
+      return {};
     }
 
     level_index target_level = find_reconstruction_target(levels);
@@ -44,16 +44,19 @@ public:
     }
 
     for (level_index i = target_level; i > source_level; i--) {
-      size_t target_reccnt =
-          (i < (level_index)levels.size()) ? levels[i]->get_record_count() : 0;
-      size_t total_reccnt = levels[i - 1]->get_record_count() + target_reccnt;
-
-      std::vector<ShardID> shards;
-      for (ssize_t j=0; j<(ssize_t)levels[i-1]->get_shard_count(); j++) {
-        shards.push_back({i-1, j});
+      if (lock_mngr.take_lock(i-1)) {
+        ReconstructionVector recon;
+        size_t target_reccnt =
+            (i < (level_index)levels.size()) ? levels[i]->get_record_count() : 0;
+        size_t total_reccnt = levels[i - 1]->get_record_count() + target_reccnt;
+        std::vector<ShardID> shards;
+        for (ssize_t j=0; j<(ssize_t)levels[i-1]->get_shard_count(); j++) {
+          shards.push_back({i-1, j});
+        }
+
+        recon.add_reconstruction(shards, i, total_reccnt, ReconstructionType::Compact);
+        reconstructions.push_back(recon);
       }
-
-      reconstructions.add_reconstruction(shards, i, total_reccnt, ReconstructionType::Compact);
     }
     
     return reconstructions;
@@ -70,7 +73,7 @@ private:
   level_index find_reconstruction_target(LevelVector &levels) const {
     level_index target_level = invalid_level_idx;
 
-    for (level_index i = 0; i < (level_index)levels.size(); i++) {
+    for (level_index i = 1; i < (level_index)levels.size(); i++) {
       if (levels[i]->get_shard_count() + 1 <= capacity()) {
         target_level = i;
         break;
diff --git a/include/framework/reconstruction/FixedShardCountPolicy.h b/include/framework/reconstruction/FixedShardCountPolicy.h
index a181052..cc8dce4 100644
--- a/include/framework/reconstruction/FixedShardCountPolicy.h
+++ b/include/framework/reconstruction/FixedShardCountPolicy.h
@@ -24,11 +24,9 @@ public:
   FixedShardCountPolicy(size_t buffer_size, size_t shard_count, size_t max_record_count) 
   : m_buffer_size(buffer_size), m_shard_count(shard_count), m_max_reccnt(max_record_count) {}
 
-  ReconstructionVector
-  get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const override {
-    ReconstructionVector reconstructions;
-    return reconstructions;
-
+  std::vector<ReconstructionVector>
+  get_reconstruction_tasks(const Version<ShardType, QueryType> *version, LockManager &lock_mngr) const override {
+    return {};
   }
 
   ReconstructionVector
diff --git a/include/framework/reconstruction/FloodL0Policy.h b/include/framework/reconstruction/FloodL0Policy.h
index b4f453b..c0d29fe 100644
--- a/include/framework/reconstruction/FloodL0Policy.h
+++ b/include/framework/reconstruction/FloodL0Policy.h
@@ -23,10 +23,9 @@ class FloodL0Policy : public ReconstructionPolicy<ShardType, QueryType> {
 public:
   FloodL0Policy(size_t buffer_size) : m_buffer_size(buffer_size) {}
 
-  ReconstructionVector
-  get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const override {
-    ReconstructionVector reconstructions;
-    return reconstructions;
+  std::vector<ReconstructionVector>
+  get_reconstruction_tasks(const Version<ShardType, QueryType> *version, LockManager &lock_mngr) const override {
+    return {};
   }
 
   ReconstructionVector
diff --git a/include/framework/reconstruction/LevelingPolicy.h b/include/framework/reconstruction/LevelingPolicy.h
index 3a0b73e..f0feb53 100644
--- a/include/framework/reconstruction/LevelingPolicy.h
+++ b/include/framework/reconstruction/LevelingPolicy.h
@@ -24,10 +24,9 @@ public:
   LevelingPolicy(size_t scale_factor, size_t buffer_size)
       : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {}
 
-  ReconstructionVector
-  get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const override { 
-    ReconstructionVector reconstructions;
-    return reconstructions;
+  std::vector<ReconstructionVector>
+  get_reconstruction_tasks(const Version<ShardType, QueryType> *version, LockManager &lock_mngr) const override { 
+    return {};
   }
 
   ReconstructionVector
diff --git a/include/framework/reconstruction/ReconstructionPolicy.h b/include/framework/reconstruction/ReconstructionPolicy.h
index 6f99b32..41a2092 100644
--- a/include/framework/reconstruction/ReconstructionPolicy.h
+++ b/include/framework/reconstruction/ReconstructionPolicy.h
@@ -15,6 +15,7 @@
 #include "util/types.h"
 #include "framework/structure/ExtensionStructure.h"
 #include "framework/scheduling/Version.h"
+#include "framework/scheduling/LockManager.h"
 
 namespace de {
 template<ShardInterface ShardType, QueryInterface<ShardType> QueryType>
@@ -23,7 +24,7 @@ class ReconstructionPolicy {
 
 public:
   ReconstructionPolicy() {}
-  virtual ReconstructionVector get_reconstruction_tasks(const Version<ShardType, QueryType> *version) const = 0;
+  virtual std::vector<ReconstructionVector> get_reconstruction_tasks(const Version<ShardType, QueryType> *version, LockManager &lock_mngr) const = 0;
   virtual ReconstructionVector get_flush_tasks(const Version<ShardType, QueryType> *version) const = 0;
   };
 }
diff --git a/include/framework/reconstruction/TieringPolicy.h b/include/framework/reconstruction/TieringPolicy.h
index aa94f7a..d8769f7 100644
--- a/include/framework/reconstruction/TieringPolicy.h
+++ b/include/framework/reconstruction/TieringPolicy.h
@@ -24,10 +24,9 @@ public:
   TieringPolicy(size_t scale_factor, size_t buffer_size)
       : m_scale_factor(scale_factor), m_buffer_size(buffer_size) {}
 
-  ReconstructionVector get_reconstruction_tasks(
-      const Version<ShardType, QueryType> *version) const override {
-    ReconstructionVector reconstructions;
-    return reconstructions;
+  std::vector<ReconstructionVector> get_reconstruction_tasks(
+      const Version<ShardType, QueryType> *version, LockManager &lock_mngr) const override {
+    return {};
   }
 
   ReconstructionVector