6 files changed, 82 insertions, 33 deletions

diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h
index da2945a..91bed98 100644
--- a/include/framework/DynamicExtension.h
+++ b/include/framework/DynamicExtension.h
@@ -87,6 +87,8 @@ public:
                                           m_config.maximum_threads);
     m_active_version.store(
         std::make_shared<VersionType>(INITIAL_VERSION, std::make_unique<StructureType>(), m_buffer.get(), 0));
+
+    m_version_counter = INITIAL_VERSION;
   }
 
   /**
@@ -226,7 +228,7 @@ public:
    *  @return The number of levels within the index
    */
   size_t get_height() {
-    return get_active_version()->get_structure()->get_height();
+    return get_active_version()->get_structure()->get_height() - 1;
   }
 
   /**
@@ -381,8 +383,6 @@ private:
   size_t m_core_cnt;
   std::atomic<int> m_next_core;
 
-  ReconPolicyType const *m_recon_policy;
-
   /* versioning + concurrency variables */
   std::atomic<size_t> m_version_counter;
   std::atomic<std::shared_ptr<VersionType>> m_active_version;
@@ -390,6 +390,8 @@ private:
   std::condition_variable m_version_advance_cv;
   std::mutex m_version_advance_mtx;
 
+  std::atomic<bool> m_flush_in_progress = false;
+
   alignas(64) std::atomic<bool> m_scheduling_reconstruction;
 
   void enforce_delete_invariant(VersionType *version) {
@@ -420,6 +422,8 @@ private:
     }
   }
 
+  size_t m_flush_cnt = 0;
+
   static void reconstruction(void *arguments) {
     auto args = (ReconstructionArgs<ShardType, QueryType> *)arguments;
     auto extension = (DynamicExtension *)args->extension;
@@ -437,11 +441,14 @@ private:
      * this code will be bypassed in that case.
      */
     if (args->priority == ReconstructionPriority::FLUSH) {
+      assert(extension->m_flush_in_progress.load());
       /* we first construct a shard from the buffer */
       auto buffview = args->version->get_buffer();
-      args->version->set_next_buffer_head(buffview.get_tail());
+      assert(buffview.get_tail() != buffview.get_head());
+
       auto new_shard = std::make_shared<ShardType>(std::move(buffview));
 
+
       /*
        * Flushes already know their version id. To avoid needing to
        * do any update reconciliation between structures, they wait
@@ -449,14 +456,22 @@ private:
        * and only then take a copy of the structure.
        */
       extension->await_version(args->version->get_id() - 1);
-      StructureType *structure =
-          extension->get_active_version()->get_structure()->copy();
 
-      /* add the newly created shard to the structure copy */
-      structure->append_l0(new_shard);
+      if (extension->m_config.recon_maint_disabled) {
+        assert(args->version->get_mutable_structure());
+        args->version->get_mutable_structure()->append_l0(std::move(new_shard));
+      } else {
+        assert(!args->version->get_mutable_structure());
+        auto structure = std::unique_ptr<StructureType>(extension->get_active_version()->get_structure()->copy());
 
-      /* set this version's structure to the newly created one */
-      args->version->set_structure(std::unique_ptr<StructureType>(structure));
+        /* add the newly created shard to the structure copy */
+        structure->append_l0(std::move(new_shard));
+
+        /* set this version's structure to the newly created one */
+        args->version->set_structure(std::move(structure));
+      }
+
+      args->version->advance_buffer_head();
     }
 
     /* perform all of the reconstructions */
@@ -476,6 +491,10 @@ private:
     /* advance the index to the newly finished version */
     extension->install_new_version(args->version);
 
+    if (args->priority == ReconstructionPriority::FLUSH) {
+      extension->m_flush_in_progress.store(false);
+    }
+
     /* manually delete the argument object */
     delete args;
   }
@@ -549,11 +568,11 @@ private:
    * at the time it is activated, but the version number must be claimed
    * early to minimize activation blocking.
    */
-  version_ptr create_version() {
+  version_ptr create_version_flush(std::unique_ptr<StructureType> structure) {
     size_t version_id = m_version_counter.fetch_add(1);
     auto active_version = get_active_version();
     std::shared_ptr<VersionType> new_version =
-        std::make_shared<VersionType>(version_id, nullptr, m_buffer.get(), active_version->get_buffer().get_head());
+        std::make_shared<VersionType>(version_id, std::move(structure), m_buffer.get(), active_version->get_buffer().get_head());
 
     return new_version;
   }
@@ -565,7 +584,7 @@ private:
    * no version number is claimed until the version is activated, to
    * prevent blocking buffer flushes.
    */
-  version_ptr create_version(std::unique_ptr<StructureType> structure) {
+  version_ptr create_version_maint(std::unique_ptr<StructureType> structure) {
     auto active_version = get_active_version();
     version_ptr new_version =
         std::make_shared<VersionType>(INVALID_VERSION, std::move(structure), m_buffer.get(), active_version->get_buffer().get_head());
@@ -622,11 +641,34 @@ private:
 
   void schedule_flush() {
     begin_reconstruction_scheduling();
-    auto new_version = create_version();
+
+    bool old = m_flush_in_progress.load();
+    if (old || !m_flush_in_progress.compare_exchange_strong(old, true)) {
+      end_reconstruction_scheduling();
+      return;
+    }
+
+    /*
+     * for "legacy" policies, without background reconstruction, we need
+     * a valid structure object as part of the version prior to determining
+     * the flush operations. Otherwise, the flush operation should only ever
+     * do the flush itself, and so no structure is needed at this point
+     *
+     * delaying obtaining the structure until later, when maintenance
+     * reconstructions are enabled, removes the need for flushes to reconcile
+     * differences in structure between when they are scheduled and when they
+     * finish
+     */
+    std::unique_ptr<StructureType> structure = nullptr;
+    if (m_config.recon_maint_disabled) {
+      structure = std::unique_ptr<StructureType>(m_active_version.load()->get_structure()->copy());
+    }
+      
+    auto new_version = create_version_flush(std::move(structure));
 
     auto *args = new ReconstructionArgs<ShardType, QueryType>();
     args->version = new_version;
-    args->tasks = m_recon_policy->get_flush_tasks(args->version.get());
+    args->tasks = m_config.recon_policy->get_flush_tasks(args->version.get());
     args->extension = this;
     args->priority = ReconstructionPriority::FLUSH;
 
@@ -646,16 +688,20 @@ private:
 
   void schedule_maint_reconstruction(bool take_reconstruction_lock = true) {
 
+    if (m_config.recon_maint_disabled) {
+      return;
+    }
+
     if (take_reconstruction_lock) {
       begin_reconstruction_scheduling();
     }
 
     // FIXME: memory management issue here?
-    auto new_version = create_version(std::unique_ptr<StructureType>(m_active_version.load()->get_structure()->copy()));
+    auto new_version = create_version_maint(std::unique_ptr<StructureType>(m_active_version.load()->get_structure()->copy()));
 
     auto *args = new ReconstructionArgs<ShardType, QueryType>();
     args->version = new_version;
-    args->tasks = m_recon_policy->get_reconstruction_tasks(args->version.get());
+    args->tasks = m_config.recon_policy->get_reconstruction_tasks(args->version.get());
     args->extension = this;
     args->priority = ReconstructionPriority::MAINT;
 
diff --git a/include/framework/reconstruction/LevelingPolicy.h b/include/framework/reconstruction/LevelingPolicy.h
index 1523e74..3a0b73e 100644
--- a/include/framework/reconstruction/LevelingPolicy.h
+++ b/include/framework/reconstruction/LevelingPolicy.h
@@ -36,8 +36,7 @@ public:
     auto levels = version->get_structure()->get_level_vector();
 
     level_index target_level = find_reconstruction_target(levels);
-    assert(target_level != -1);
-    level_index source_level = 1;
+    level_index source_level = 0;
 
     if (target_level == invalid_level_idx) {
       /* grow */
@@ -47,7 +46,9 @@ 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;
+      size_t total_reccnt =
+          (i == 0) ? m_buffer_size + target_reccnt
+                   : levels[i - 1]->get_record_count() + target_reccnt;
 
       reconstructions.add_reconstruction(i - 1, i, total_reccnt,
                                          ReconstructionType::Merge);
@@ -61,7 +62,7 @@ private:
     level_index target_level = invalid_level_idx;
     size_t incoming_records = m_buffer_size;
 
-    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_record_count() + incoming_records < capacity(i)) {
         target_level = i;
         break;
@@ -74,7 +75,7 @@ private:
   }
 
   inline size_t capacity(level_index level) const {
-    return m_buffer_size * pow(m_scale_factor, level + 1);
+    return m_buffer_size * pow(m_scale_factor, level);
   }
 
   size_t m_scale_factor;
diff --git a/include/framework/scheduling/FIFOScheduler.h b/include/framework/scheduling/FIFOScheduler.h
index 7cb6d20..4c1db8d 100644
--- a/include/framework/scheduling/FIFOScheduler.h
+++ b/include/framework/scheduling/FIFOScheduler.h
@@ -82,6 +82,7 @@ private:
   std::atomic<size_t> m_counter;
   std::mutex m_cv_lock;
   std::condition_variable m_cv;
+  std::mutex m_queue_lock;
 
   std::thread m_sched_thrd;
   std::thread m_sched_wakeup_thrd;
@@ -102,6 +103,7 @@ private:
   }
 
   void schedule_next() {
+    auto lk = std::unique_lock<std::mutex>(m_queue_lock);
     assert(m_task_queue.size() > 0);
     auto t = m_task_queue.pop();
     m_stats.job_scheduled(t.m_timestamp);
diff --git a/include/framework/scheduling/Version.h b/include/framework/scheduling/Version.h
index 8d3d038..9c62ea1 100644
--- a/include/framework/scheduling/Version.h
+++ b/include/framework/scheduling/Version.h
@@ -84,13 +84,10 @@ public:
     return version;
   }
 
-  void set_next_buffer_head(size_t new_head) {
-    m_pending_buffer_head = new_head;
-  }
-
   bool advance_buffer_head() {
-    m_buffer_head = m_pending_buffer_head;
-    return m_buffer->advance_head(m_buffer_head);
+    auto new_head = m_buffer->get_buffer_view().get_tail();
+    m_buffer_head = new_head;
+    return m_buffer->advance_head(new_head);
   }
 
 private:
diff --git a/include/framework/structure/ExtensionStructure.h b/include/framework/structure/ExtensionStructure.h
index 62c27f5..60fb6c7 100644
--- a/include/framework/structure/ExtensionStructure.h
+++ b/include/framework/structure/ExtensionStructure.h
@@ -27,7 +27,10 @@ class ExtensionStructure {
   typedef std::vector<std::shared_ptr<InternalLevel<ShardType, QueryType>>>
       LevelVector;
 public:
-  ExtensionStructure() = default;
+  ExtensionStructure() {
+    m_levels.emplace_back(std::make_shared<InternalLevel<ShardType, QueryType>>(0));
+  }
+  
   ~ExtensionStructure() = default;
 
   /*
@@ -217,7 +220,6 @@ public:
   }
 
   void append_l0(std::shared_ptr<ShardType> shard) {
-    // FIXME: ensure that there's always a level 0 in the version
     m_levels[0]->append(shard);
   }
 
diff --git a/include/framework/util/Configuration.h b/include/framework/util/Configuration.h
index a751a29..81698d2 100644
--- a/include/framework/util/Configuration.h
+++ b/include/framework/util/Configuration.h
@@ -20,9 +20,9 @@ DeletePolicy D, SchedulerInterface SchedType>
 class DEConfiguration {
   public:
   DEConfiguration(std::unique_ptr<ReconstructionPolicy<ShardType, QueryType>> recon_policy) 
-   : m_recon_policy(std::move(recon_policy)) {}
+   : recon_policy(std::move(recon_policy)) {}
 
-    std::unique_ptr<ReconstructionPolicy<ShardType, QueryType>> m_recon_policy;
+    std::unique_ptr<ReconstructionPolicy<ShardType, QueryType>> recon_policy;
 
     /* buffer parameters */
     size_t buffer_count = 1;
@@ -31,8 +31,9 @@ class DEConfiguration {
 
     /* reconstruction triggers */
     bool recon_enable_seek_trigger = false;
-    bool recon_enable_maint_on_flush = true;
+    bool recon_enable_maint_on_flush = false;
     bool recon_enable_delete_cmpct = false;
+    bool recon_maint_disabled = true;
 
     size_t recon_l0_capacity = 0; /* 0 for unbounded */
     double maximum_delete_proportion = 1;