Merge pull request #1 from dbrumbaugh/new-buffer

Initial Concurrency Implementation

1 files changed, 590 insertions, 370 deletions

diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h
index 524024b..7ea5370 100644
--- a/include/framework/DynamicExtension.h
+++ b/include/framework/DynamicExtension.h
@@ -1,302 +1,639 @@
 /*
  * include/framework/DynamicExtension.h
  *
- * Copyright (C) 2023 Douglas Rumbaugh <drumbaugh@psu.edu> 
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
  *                    Dong Xie <dongx@psu.edu>
  *
- * All rights reserved. Published under the Modified BSD License.
+ * Distributed under the Modified BSD License.
  *
  */
 #pragma once
 
 #include <atomic>
-#include <numeric>
 #include <cstdio>
 #include <vector>
 
-#include "framework/MutableBuffer.h"
-#include "framework/InternalLevel.h"
-#include "framework/ShardInterface.h"
-#include "framework/QueryInterface.h"
-#include "framework/RecordInterface.h"
+#include "framework/interface/Scheduler.h"
+#include "framework/scheduling/FIFOScheduler.h"
+#include "framework/scheduling/SerialScheduler.h"
 
-#include "shard/WIRS.h"
-#include "psu-util/timer.h"
-#include "psu-ds/Alias.h"
+#include "framework/structure/MutableBuffer.h"
+#include "framework/interface/Record.h"
+#include "framework/structure/ExtensionStructure.h"
 
-namespace de {
-
-thread_local size_t sampling_attempts = 0;
-thread_local size_t sampling_rejections = 0;
-thread_local size_t deletion_rejections = 0;
-thread_local size_t bounds_rejections = 0;
-thread_local size_t tombstone_rejections = 0;
-thread_local size_t buffer_rejections = 0;
-
-/*
- * thread_local size_t various_sampling_times go here.
- */
-thread_local size_t sample_range_time = 0;
-thread_local size_t alias_time = 0;
-thread_local size_t alias_query_time = 0;
-thread_local size_t rejection_check_time = 0;
-thread_local size_t buffer_sample_time = 0;
-thread_local size_t memlevel_sample_time = 0;
-thread_local size_t disklevel_sample_time = 0;
-thread_local size_t sampling_bailouts = 0;
-
-
-enum class LayoutPolicy {
-    LEVELING,
-    TEIRING
-};
-
-enum class DeletePolicy {
-    TOMBSTONE,
-    TAGGING
-};
+#include "framework/util/Configuration.h"
+#include "framework/scheduling/Epoch.h"
 
-typedef ssize_t level_index;
+namespace de {
 
-template <RecordInterface R, ShardInterface S, QueryInterface Q, LayoutPolicy L=LayoutPolicy::TEIRING, DeletePolicy D=DeletePolicy::TAGGING>
+template <RecordInterface R, ShardInterface<R> S, QueryInterface<R, S> Q, LayoutPolicy L=LayoutPolicy::TEIRING, 
+          DeletePolicy D=DeletePolicy::TAGGING, SchedulerInterface SCHED=FIFOScheduler>
 class DynamicExtension {
-    //typedef typename S<R> Shard;
     typedef S Shard;
     typedef MutableBuffer<R> Buffer;
+    typedef ExtensionStructure<R, S, Q, L> Structure;
+    typedef Epoch<R, S, Q, L> _Epoch;
+    typedef BufferView<R> BufView;
+
+    static constexpr size_t QUERY = 1;
+    static constexpr size_t RECONSTRUCTION = 2;
+
+    struct epoch_ptr {
+        _Epoch *epoch;
+        size_t refcnt;
+    };
 
 public:
-    DynamicExtension(size_t buffer_cap, size_t scale_factor, double max_delete_prop)
-        : m_scale_factor(scale_factor), m_max_delete_prop(max_delete_prop),
-          m_buffer(new Buffer(buffer_cap, buffer_cap * max_delete_prop))
-    { }
+    DynamicExtension(size_t buffer_lwm, size_t buffer_hwm, size_t scale_factor, size_t memory_budget=0, 
+                     size_t thread_cnt=16)
+        : m_scale_factor(scale_factor)
+        , m_max_delete_prop(1)
+        , m_sched(memory_budget, thread_cnt)
+        , m_buffer(new Buffer(buffer_lwm, buffer_hwm))
+        , m_core_cnt(thread_cnt)
+        , m_next_core(0)
+        , m_epoch_cnt(0)
+    {
+        auto vers = new Structure(buffer_hwm, m_scale_factor, m_max_delete_prop);
+        m_current_epoch.store({new _Epoch(0, vers, m_buffer, 0), 0});
+        m_previous_epoch.store({nullptr, 0});
+        m_next_epoch.store({nullptr, 0});
+    }
 
     ~DynamicExtension() {
-        delete m_buffer;
 
-        for (size_t i=0; i<m_levels.size(); i++) {
-            delete m_levels[i];
-        }
+        /* let any in-flight epoch transition finish */
+        await_next_epoch();
+
+        /* shutdown the scheduler */
+        m_sched.shutdown();
+
+        /* delete all held resources */
+        delete m_next_epoch.load().epoch;
+        delete m_current_epoch.load().epoch;
+        delete m_previous_epoch.load().epoch;
+
+        delete m_buffer;
     }
 
+    /*
+     * Insert the record `rec` into the index. If the buffer is full and
+     * the framework is blocking on an epoch transition, this call may fail
+     * and return 0. In this case, retry the call again later. If
+     * successful, 1 will be returned. The record will be immediately
+     * visible in the buffer upon the successful return of this function.
+     */
     int insert(const R &rec) {
         return internal_append(rec, false);
     }
 
+    /*
+     * Erase the record `rec` from the index. It is assumed that `rec`
+     * currently exists--no special checks are made for correctness here.
+     * The behavior if this function will differ depending on if tombstone
+     * or tagged deletes are used.
+     *
+     * Tombstone deletes - inserts a tombstone record for `rec`. This *may*
+     * return 0 and fail if the buffer is full and the framework is
+     * blocking on an epoch transition. In this case, repeat the call
+     * later. 1 will be returned when the tombstone is successfully
+     * inserted.
+     *
+     * Tagging deletes - Does a point lookup for the record across the
+     * entire structure, and sets its delete bit when found. Returns 1 if
+     * the record is found and marked, and 0 if it was not (i.e., if it
+     * isn't present in the index). 
+     */
     int erase(const R &rec) {
-        Buffer *buffer;
-
+        // FIXME: delete tagging will require a lot of extra work to get
+        //        operating "correctly" in a concurrent environment.
+ 
+        /* 
+         * Get a view on the buffer *first*. This will ensure a stronger 
+         * ordering than simply accessing the buffer directly, but is
+         * not *strictly* necessary.
+         */
         if constexpr (D == DeletePolicy::TAGGING) {
-            auto buffer = get_buffer();
+            static_assert(std::same_as<SCHED, SerialScheduler>, "Tagging is only supported in single-threaded operation");
 
-            // Check the levels first. This assumes there aren't 
-            // any undeleted duplicate records.
-            for (auto level : m_levels) {
-                if (level && level->delete_record(rec)) {
-                    return 1;
-                }
+            auto view = m_buffer->get_buffer_view();
+
+            auto epoch = get_active_epoch();
+            if (epoch->get_structure()->tagged_delete(rec)) {
+                end_job(epoch);
+                return 1;
             }
 
-            // the buffer will take the longest amount of time, and 
-            // probably has the lowest probability of having the record,
-            // so we'll check it last.
-            return buffer->delete_record(rec);
+            end_job(epoch);
+
+            /*
+             * the buffer will take the longest amount of time, and 
+             * probably has the lowest probability of having the record,
+             * so we'll check it last.
+             */
+            return view.delete_record(rec);
         }
 
+        /*
+         * If tagging isn't used, then delete using a tombstone
+         */
         return internal_append(rec, true);
     }
 
-    std::vector<R> query(void *parms) {
-        auto buffer = get_buffer();
+    /*
+     * Execute the query with parameters `parms` and return a future. This
+     * future can be used to access a vector containing the results of the
+     * query.
+     *
+     * The behavior of this function is undefined if `parms` is not a
+     * pointer to a valid query parameter object for the query type used as
+     * a template parameter to construct the framework.
+     */
+    std::future<std::vector<R>> query(void *parms) {
+        return schedule_query(parms);
+    }
+
+    /*
+     * Returns the number of records (included tagged records and
+     * tombstones) currently within the framework.
+     */
+    size_t get_record_count() {
+        auto epoch = get_active_epoch();
+        auto t =  epoch->get_buffer().get_record_count() + epoch->get_structure()->get_record_count();
+        end_job(epoch);
+
+        return t;
+    }
+
+    /*
+     * Returns the number of tombstone records currently within the
+     * framework. This function can be called when tagged deletes are used,
+     * but will always return 0 in that case.
+     */
+    size_t get_tombstone_count() {
+        auto epoch = get_active_epoch();
+        auto t = epoch->get_buffer().get_tombstone_count() + epoch->get_structure()->get_tombstone_count();
+        end_job(epoch);
 
-        // Get the buffer query state
-        auto buffer_state = Q::get_buffer_query_state(buffer, parms);
+        return t;
+    }
 
-        // Get the shard query states
-        std::vector<std::pair<ShardID, Shard*>> shards;
-        std::vector<void*> states;
+    /*
+     * Get the number of levels within the framework. This count will
+     * include any empty levels, but will not include the buffer. Note that
+     * this is *not* the same as the number of shards when tiering is used,
+     * as each level can contain multiple shards in that case.
+     */
+    size_t get_height() {
+        auto epoch = get_active_epoch();
+        auto t = epoch->get_structure()->get_height();
+        end_job(epoch);
 
-        for (auto &level : m_levels) {
-            level->get_query_states(shards, states, parms);
-        }
+        return t;
+    }
 
-        Q::process_query_states(parms, states, buffer_state);
+    /*
+     * Get the number of bytes of memory allocated across the framework for
+     * storing records and associated index information (i.e., internal
+     * ISAM tree nodes). This includes memory that is allocated but
+     * currently unused in the buffer, or in shards themselves
+     * (overallocation due to delete cancellation, etc.).
+     */
+    size_t get_memory_usage() {
+        auto epoch = get_active_epoch();
+        auto t= epoch->get_buffer().get_memory_usage() + epoch->get_structure()->get_memory_usage();
+        end_job(epoch);
 
-        std::vector<std::vector<Wrapped<R>>> query_results(shards.size() + 1);
+        return t;
+    }
 
-        // Execute the query for the buffer
-        auto buffer_results = Q::buffer_query(buffer, buffer_state, parms);
-        query_results[0] = std::move(filter_deletes(buffer_results, {-1, -1}, buffer));
-        if constexpr (Q::EARLY_ABORT) {
-            if (query_results[0].size() > 0) {
-                auto result = Q::merge(query_results, parms);
-                for (size_t i=0; i<states.size(); i++) {
-                    Q::delete_query_state(states[i]);
-                }
+    /*
+     * Get the number of bytes of memory allocated across the framework for
+     * auxiliary structures. This can include bloom filters, aux
+     * hashtables, etc. 
+     */
+    size_t get_aux_memory_usage() {
+        auto epoch = get_active_epoch();
+        auto t = epoch->get_buffer().get_aux_memory_usage() + epoch->get_structure()->get_aux_memory_usage();
+        end_job(epoch);
 
-                Q::delete_buffer_query_state(buffer_state);
-                return result;
-            }
+        return t;
+    }
+
+    /*
+     * Returns the maximum physical capacity of the buffer, measured in
+     * records.
+     */
+    size_t get_buffer_capacity() {
+        return m_buffer->get_capacity();
+    }
+    
+    /*
+     * Create a new single Shard object containing all of the records
+     * within the framework (buffer and shards). The optional parameter can
+     * be used to specify whether the Shard should be constructed with the
+     * currently active state of the framework (false), or if shard
+     * construction should wait until any ongoing reconstructions have
+     * finished and use that new version (true).
+     */
+    Shard *create_static_structure(bool await_reconstruction_completion=false) {
+        if (await_reconstruction_completion) {
+            await_next_epoch();
         }
 
-        // Execute the query for each shard
-        for (size_t i=0; i<shards.size(); i++) {
-            auto shard_results = Q::query(shards[i].second, states[i], parms);
-            query_results[i+1] = std::move(filter_deletes(shard_results, shards[i].first, buffer));
-            if constexpr (Q::EARLY_ABORT) {
-                if (query_results[i].size() > 0) {
-                    auto result = Q::merge(query_results, parms);
-                    for (size_t i=0; i<states.size(); i++) {
-                        Q::delete_query_state(states[i]);
-                    }
+        auto epoch = get_active_epoch();
+        auto vers = epoch->get_structure();
+        std::vector<Shard *> shards;
 
-                    Q::delete_buffer_query_state(buffer_state);
 
-                    return result;
+        if (vers->get_levels().size() > 0) {
+            for (int i=vers->get_levels().size() - 1; i>= 0; i--) {
+                if (vers->get_levels()[i] && vers->get_levels()[i]->get_record_count() > 0) {
+                    shards.emplace_back(vers->get_levels()[i]->get_combined_shard());
                 }
             }
         }
-        
-        // Merge the results together
-        auto result = Q::merge(query_results, parms);
 
-        for (size_t i=0; i<states.size(); i++) {
-            Q::delete_query_state(states[i]);
+        /* 
+         * construct a shard from the buffer view. We'll hold the view
+         * for as short a time as possible: once the records are exfiltrated
+         * from the buffer, there's no reason to retain a hold on the view's
+         * head pointer any longer
+         */
+        {
+            auto bv = epoch->get_buffer();
+            if (bv.get_record_count() > 0) {
+                shards.emplace_back(new S(std::move(bv)));
+            }
         }
 
-        Q::delete_buffer_query_state(buffer_state);
+        Shard *flattened = new S(shards);
 
-        return result;
-    }
+        for (auto shard : shards) {
+            delete shard;
+        }
 
-    size_t get_record_count() {
-        size_t cnt = get_buffer()->get_record_count();
+        end_job(epoch);
+        return flattened;
+    }
 
-        for (size_t i=0; i<m_levels.size(); i++) {
-            if (m_levels[i]) cnt += m_levels[i]->get_record_count();
+    /*
+     * If the current epoch is *not* the newest one, then wait for
+     * the newest one to become available. Otherwise, returns immediately.
+     */
+    void await_next_epoch() {
+        while (m_next_epoch.load().epoch != nullptr) {
+            std::unique_lock<std::mutex> lk(m_epoch_cv_lk);
+            m_epoch_cv.wait(lk);
         }
+    }
 
-        return cnt;
+    /*
+     * Mostly exposed for unit-testing purposes. Verifies that the current
+     * active version of the ExtensionStructure doesn't violate the maximum
+     * tombstone proportion invariant. 
+     */
+    bool validate_tombstone_proportion() {
+        auto epoch = get_active_epoch();
+        auto t = epoch->get_structure()->validate_tombstone_proportion();
+        end_job(epoch);
+        return t;
     }
 
-    size_t get_tombstone_cnt() {
-        size_t cnt = get_buffer()->get_tombstone_count();
 
-        for (size_t i=0; i<m_levels.size(); i++) {
-            if (m_levels[i]) cnt += m_levels[i]->get_tombstone_count();
-        }
+    void print_scheduler_statistics() {
+        m_sched.print_statistics();
+    }
+
+private:
+    SCHED m_sched;
 
-        return cnt;
+    Buffer *m_buffer;
+
+    //std::mutex m_struct_lock;
+    //std::set<Structure *> m_versions;
+
+    alignas(64) std::atomic<bool> m_reconstruction_scheduled;
+
+    std::atomic<epoch_ptr> m_next_epoch;
+    std::atomic<epoch_ptr> m_current_epoch;
+    std::atomic<epoch_ptr> m_previous_epoch;
+
+    std::condition_variable m_epoch_cv;
+    std::mutex m_epoch_cv_lk;
+
+    std::atomic<size_t> m_epoch_cnt;
+
+    size_t m_scale_factor;
+    double m_max_delete_prop;
+
+    std::atomic<int> m_next_core;
+    size_t m_core_cnt;
+
+    void enforce_delete_invariant(_Epoch *epoch) {
+        auto structure = epoch->get_structure(); 
+        auto compactions = structure->get_compaction_tasks();
+
+        while (compactions.size() > 0) {
+
+            /* schedule a compaction */
+            ReconstructionArgs<R, S, Q, L> *args = new ReconstructionArgs<R, S, Q, L>();
+            args->epoch = epoch;
+            args->merges = compactions;
+            args->extension = this;
+            args->compaction = true;
+            /* NOTE: args is deleted by the reconstruction job, so shouldn't be freed here */
+
+            auto wait = args->result.get_future();
+
+            m_sched.schedule_job(reconstruction, 0, args, RECONSTRUCTION);
+
+            /* wait for compaction completion */
+            wait.get();
+
+            /* get a new batch of compactions to perform, if needed */
+            compactions = structure->get_compaction_tasks();
+        }
     }
 
-    size_t get_height() {
-        return m_levels.size();
+    _Epoch *get_active_epoch() {
+        epoch_ptr old, new_ptr;
+
+        do {
+            /* 
+             * during an epoch transition, a nullptr will installed in the
+             * current_epoch. At this moment, the "new" current epoch will
+             * soon be installed, but the "current" current epoch has been
+             * moved back to m_previous_epoch.
+             */
+            if (m_current_epoch.load().epoch == nullptr) {
+                old = m_previous_epoch;
+                new_ptr = {old.epoch, old.refcnt+1};
+                if (old.epoch != nullptr && m_previous_epoch.compare_exchange_strong(old, new_ptr)) {
+                    break;
+                }
+            } else {
+                old = m_current_epoch;
+                new_ptr = {old.epoch, old.refcnt+1};
+                if (old.epoch != nullptr && m_current_epoch.compare_exchange_strong(old, new_ptr)) {
+                    break;
+                }
+            }
+        } while (true);
+
+        assert(new_ptr.refcnt > 0);
+
+        return new_ptr.epoch;
     }
 
-    size_t get_memory_usage() {
-        size_t cnt = m_buffer->get_memory_usage();
+    void advance_epoch(size_t buffer_head) {
+
+        retire_epoch(m_previous_epoch.load().epoch);
+
+        epoch_ptr tmp = {nullptr, 0};
+        epoch_ptr cur;
+        do {
+            cur = m_current_epoch;
+        } while(!m_current_epoch.compare_exchange_strong(cur, tmp));
+
+        m_previous_epoch.store(cur);
 
-        for (size_t i=0; i<m_levels.size(); i++) {
-            if (m_levels[i]) cnt += m_levels[i]->get_memory_usage();
+        // 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
+        while (!m_next_epoch.load().epoch->advance_buffer_head(buffer_head)) {
+            _mm_pause();
         }
 
-        return cnt;
+
+        m_current_epoch.store(m_next_epoch);
+        m_next_epoch.store({nullptr, 0});
+
+
+        /* notify any blocking threads that the new epoch is available */
+        m_epoch_cv_lk.lock();
+        m_epoch_cv.notify_all();
+        m_epoch_cv_lk.unlock();
     }
 
-    size_t get_aux_memory_usage() {
-        size_t cnt = m_buffer->get_aux_memory_usage();
+    /*
+     * Creates a new epoch by copying the currently active one. The new epoch's
+     * structure will be a shallow copy of the old one's. 
+     */
+    _Epoch *create_new_epoch() {
+        /*
+         * This epoch access is _not_ protected under the assumption that
+         * only one reconstruction will be able to trigger at a time. If that condition
+         * is violated, it is possible that this code will clone a retired
+         * epoch.
+         */
+        assert(m_next_epoch.load().epoch == nullptr);
+        auto current_epoch = get_active_epoch();
+
+        m_epoch_cnt.fetch_add(1);
+        m_next_epoch.store({current_epoch->clone(m_epoch_cnt.load()), 0});
+
+        end_job(current_epoch);
+
+        return m_next_epoch.load().epoch;
+    }
 
-        for (size_t i=0; i<m_levels.size(); i++) {
-            if (m_levels[i]) {
-                cnt += m_levels[i]->get_aux_memory_usage();
-            }
+    void retire_epoch(_Epoch *epoch) {
+        /*
+         * Epochs with currently active jobs cannot
+         * be retired. By the time retire_epoch is called,
+         * it is assumed that a new epoch is active, meaning
+         * that the epoch to be retired should no longer
+         * accumulate new active jobs. Eventually, this
+         * number will hit zero and the function will
+         * proceed.
+         */
+
+        if (epoch == nullptr) {
+            return;
         }
 
-        return cnt;
-    }
+        epoch_ptr old, new_ptr;
+        new_ptr = {nullptr, 0};
+        do {
+            old = m_previous_epoch.load();
+
+            /*
+             * If running in single threaded mode, the failure to retire
+             * an Epoch will result in the thread of execution blocking
+             * indefinitely. 
+             */
+            if constexpr (std::same_as<SCHED, SerialScheduler>) {
+                if (old.epoch == epoch) assert(old.refcnt == 0);
+            }
 
-    bool validate_tombstone_proportion() {
-        long double ts_prop;
-        for (size_t i=0; i<m_levels.size(); i++) {
-            if (m_levels[i]) {
-                ts_prop = (long double) m_levels[i]->get_tombstone_count() / (long double) calc_level_record_capacity(i);
-                if (ts_prop > (long double) m_max_delete_prop) {
-                    return false;
-                }
+            if (old.epoch == epoch && old.refcnt == 0 &&
+                m_previous_epoch.compare_exchange_strong(old, new_ptr)) {
+                break;
             }
-        }
+            usleep(1);
+	    
+        } while(true);
 
-        return true;
+        delete epoch;
     }
 
-    size_t get_buffer_capacity() {
-        return m_buffer->get_capacity();
+    static void reconstruction(void *arguments) {
+        auto args = (ReconstructionArgs<R, S, Q, L> *) arguments;
+
+        ((DynamicExtension *) args->extension)->SetThreadAffinity();
+        Structure *vers = args->epoch->get_structure();
+
+        for (ssize_t i=0; i<args->merges.size(); i++) {
+            vers->reconstruction(args->merges[i].second, args->merges[i].first);
+        }
+
+        /*
+         * we'll grab the buffer AFTER doing the internal reconstruction, so we
+         * can flush as many records as possible in one go. The reconstruction
+         * was done so as to make room for the full buffer anyway, so there's 
+         * no real benefit to doing this first.
+         */
+        auto buffer_view = args->epoch->get_buffer();
+        size_t new_head = buffer_view.get_tail();
+
+        /*
+         * if performing a compaction, don't flush the buffer, as
+         * there is no guarantee that any necessary reconstructions
+         * will free sufficient space in L0 to support a flush
+         */
+        if (!args->compaction) {
+            vers->flush_buffer(std::move(buffer_view));
+        }
+
+        args->result.set_value(true);
+
+        /*
+         * Compactions occur on an epoch _before_ it becomes active,
+         * and as a result the active epoch should _not_ be advanced as
+         * part of a compaction 
+         */
+        if (!args->compaction) {
+            ((DynamicExtension *) args->extension)->advance_epoch(new_head);
+        }
+
+        ((DynamicExtension *) args->extension)->m_reconstruction_scheduled.store(false);
+        
+        delete args;
     }
     
-    Shard *create_static_structure() {
-        std::vector<Shard *> shards;
+    static void async_query(void *arguments) {
+        QueryArgs<R, S, Q, L> *args = (QueryArgs<R, S, Q, L> *) arguments;
 
-        if (m_levels.size() > 0) {
-            for (int i=m_levels.size() - 1; i>= 0; i--) {
-                if (m_levels[i]) {
-                    shards.emplace_back(m_levels[i]->get_merged_shard());
-                }
-            }
-        }
+        auto epoch = ((DynamicExtension *) args->extension)->get_active_epoch();
+
+        auto ptr1 = ((DynamicExtension *) args->extension)->m_previous_epoch.load().epoch;
+        auto ptr2 = ((DynamicExtension *) args->extension)->m_current_epoch.load().epoch;
+        auto ptr3 = ((DynamicExtension *) args->extension)->m_next_epoch.load().epoch;
+
+
+        auto buffer = epoch->get_buffer();
+        auto vers = epoch->get_structure();
+        void *parms = args->query_parms;
+
+        /* Get the buffer query states */
+        void *buffer_state = Q::get_buffer_query_state(&buffer, parms);
+
+        /* Get the shard query states */
+        std::vector<std::pair<ShardID, Shard*>> shards;
+        std::vector<void *> states = vers->get_query_states(shards, parms);
+
+        Q::process_query_states(parms, states, buffer_state);
 
-        shards.emplace_back(new S(get_buffer()));
+        std::vector<std::vector<Wrapped<R>>> query_results(shards.size() + 1);
+        for (size_t i=0; i<query_results.size(); i++) {
+            std::vector<Wrapped<R>> local_results;
+            ShardID shid;
+
+            if (i == 0) { /* process the buffer first */
+                local_results = Q::buffer_query(buffer_state, parms);
+                shid = INVALID_SHID;
+            } else {
+                local_results = Q::query(shards[i - 1].second, states[i - 1], parms);
+                shid = shards[i - 1].first; 
+            }
 
-        Shard *shards_array[shards.size()];
+            query_results[i] = std::move(filter_deletes(local_results, shid, vers, &buffer)); 
 
-        size_t j = 0;
-        for (size_t i=0; i<shards.size(); i++) {
-            if (shards[i]) {
-                shards_array[j++] = shards[i];
+            if constexpr (Q::EARLY_ABORT) {
+                if (query_results[i].size() > 0) break;
             }
         }
 
-        Shard *flattened = new S(shards_array, j);
+        auto result = Q::merge(query_results, parms);
+        args->result_set.set_value(std::move(result));
 
-        for (auto shard : shards) {
-            delete shard;
+        ((DynamicExtension *) args->extension)->end_job(epoch);
+
+        Q::delete_buffer_query_state(buffer_state);
+        for (size_t i=0; i<states.size(); i++) {
+            Q::delete_query_state(states[i]);
         }
 
-        return flattened;
+        delete args;
     }
 
-private:
-    Buffer *m_buffer;
+    void schedule_reconstruction() {
+        auto epoch = create_new_epoch();
+        /* 
+         * the reconstruction process calls end_job(), 
+         * so we must start one before calling it
+         */
+
+        ReconstructionArgs<R, S, Q, L> *args = new ReconstructionArgs<R, S, Q, L>();
+        args->epoch = epoch;
+        args->merges = epoch->get_structure()->get_reconstruction_tasks(m_buffer->get_high_watermark());
+        args->extension = this;
+        args->compaction = false;
+        /* NOTE: args is deleted by the reconstruction job, so shouldn't be freed here */
+
+        m_sched.schedule_job(reconstruction, 0, args, RECONSTRUCTION);
+    }
 
-    size_t m_scale_factor;
-    double m_max_delete_prop;
+    std::future<std::vector<R>> schedule_query(void *query_parms) {
+        QueryArgs<R, S, Q, L> *args = new QueryArgs<R, S, Q, L>();
+        args->extension = this;
+        args->query_parms = query_parms;
+        auto result = args->result_set.get_future();
 
-    std::vector<InternalLevel<R, S, Q> *> m_levels;
+        m_sched.schedule_job(async_query, 0, args, QUERY);
 
-    Buffer *get_buffer() {
-        return m_buffer;
+        return result;
     }
 
     int internal_append(const R &rec, bool ts) {
-        Buffer *buffer;
-        while (!(buffer = get_buffer()))
-            ;
-        
-        if (buffer->is_full()) {
-            merge_buffer();
+        if (m_buffer->is_at_low_watermark()) {
+            auto old = false;
+
+            if (m_reconstruction_scheduled.compare_exchange_strong(old, true)) {
+                schedule_reconstruction();
+            }
         }
 
-        return buffer->append(rec, ts);
+        /* this will fail if the HWM is reached and return 0 */
+        return m_buffer->append(rec, ts);
     }
 
-    std::vector<Wrapped<R>> filter_deletes(std::vector<Wrapped<R>> &records, ShardID shid, Buffer *buffer) {
-        if constexpr (!Q::SKIP_DELETE_FILTER) {
+    static std::vector<Wrapped<R>> filter_deletes(std::vector<Wrapped<R>> &records, ShardID shid, Structure *vers, BufView *bview) {
+        if constexpr (Q::SKIP_DELETE_FILTER) {
             return records;
         }
 
         std::vector<Wrapped<R>> processed_records;
         processed_records.reserve(records.size());
 
-        // For delete tagging, we just need to check the delete bit on each
-        // record.
+        /* 
+         * For delete tagging, we just need to check the delete bit 
+         * on each record. 
+         */
         if constexpr (D == DeletePolicy::TAGGING) {
             for (auto &rec : records) {
                 if (rec.is_deleted()) {
@@ -309,25 +646,35 @@ private:
             return processed_records;
         }
 
-        // For tombstone deletes, we need to search for the corresponding 
-        // tombstone for each record.
+        /*
+         * For tombstone deletes, we need to search for the corresponding 
+         * tombstone for each record.
+         */
         for (auto &rec : records) {
            if (rec.is_tombstone()) {
                 continue;
             } 
 
-            if (buffer->check_tombstone(rec.rec)) {
-                continue;
+            // FIXME: need to figure out how best to re-enable the buffer tombstone
+            // check in the correct manner.
+            //if (buffview.check_tombstone(rec.rec)) {
+                //continue;
+            //}
+
+            for (size_t i=0; i<bview->get_record_count(); i++) {
+                if (bview->get(i)->is_tombstone() && bview->get(i)->rec == rec.rec) {
+                    continue;
+                }
             }
 
             if (shid != INVALID_SHID) {
                 for (size_t lvl=0; lvl<=shid.level_idx; lvl++) {
-                    if (m_levels[lvl]->check_tombstone(0, rec.rec)) {
+                    if (vers->get_levels()[lvl]->check_tombstone(0, rec.rec)) {
                         continue;
                     }
                 }
 
-                if (m_levels[shid.level_idx]->check_tombstone(shid.shard_idx + 1, rec.rec)) {
+                if (vers->get_levels()[shid.level_idx]->check_tombstone(shid.shard_idx + 1, rec.rec)) {
                     continue;
                 }
             }
@@ -338,197 +685,70 @@ private:
         return processed_records;
     }
 
-    /*
-     * Add a new level to the LSM Tree and return that level's index. Will
-     * automatically determine whether the level should be on memory or on disk,
-     * and act appropriately.
-     */
-    inline level_index grow() {
-        level_index new_idx;
-
-        size_t new_shard_cnt = (L == LayoutPolicy::LEVELING) ? 1 : m_scale_factor;
-        new_idx = m_levels.size();
-        if (new_idx > 0) {
-            assert(m_levels[new_idx - 1]->get_shard(0)->get_tombstone_count() == 0);
-        }
-        m_levels.emplace_back(new InternalLevel<R, Shard, Q>(new_idx, new_shard_cnt));
-
-        return new_idx;
-    }
-
-
-    // Merge the memory table down into the tree, completing any required other
-    // merges to make room for it.
-    inline void merge_buffer() {
-        auto buffer = get_buffer();
-
-        if (!can_merge_with(0, buffer->get_record_count())) {
-            merge_down(0);
+    void SetThreadAffinity() {
+        int core = m_next_core.fetch_add(1) % m_core_cnt;
+        cpu_set_t mask;
+        CPU_ZERO(&mask);
+
+        switch (core % 2) {
+        case 0:
+          // 0 |-> 0
+          // 2 |-> 2
+          // 4 |-> 4
+          core = core;
+          break;
+        case 1:
+          // 1 |-> 28
+          // 3 |-> 30
+          // 5 |-> 32
+          core = (core - 1) + m_core_cnt;
+          break;
         }
-
-        merge_buffer_into_l0(buffer);
-        enforce_delete_maximum(0);
-
-        buffer->truncate();
-        return;
+        CPU_SET(core, &mask);
+        ::sched_setaffinity(0, sizeof(mask), &mask);
     }
 
-    /*
-     * Merge the specified level down into the tree. The level index must be
-     * non-negative (i.e., this function cannot be used to merge the buffer). This
-     * routine will recursively perform any necessary merges to make room for the 
-     * specified level.
-     */
-    inline void merge_down(level_index idx) {
-        level_index merge_base_level = find_mergable_level(idx);
-        if (merge_base_level == -1) {
-            merge_base_level = grow();
-        }
 
-        for (level_index i=merge_base_level; i>idx; i--) {
-            merge_levels(i, i-1);
-            enforce_delete_maximum(i);
-        }
+    void end_job(_Epoch *epoch) {
+        epoch_ptr old, new_ptr;
 
-        return;
-    }
+        do {
+            if (m_previous_epoch.load().epoch == epoch) {
+                old = m_previous_epoch;
+                /* 
+                 * This could happen if we get into the system during a
+                 * transition. In this case, we can just back out and retry
+                 */
+                if (old.epoch == nullptr) {
+                    continue;
+                }
 
-    /*
-     * Find the first level below the level indicated by idx that
-     * is capable of sustaining a merge operation and return its
-     * level index. If no such level exists, returns -1. Also
-     * returns -1 if idx==0, and no such level exists, to simplify
-     * the logic of the first merge.
-     */
-    inline level_index find_mergable_level(level_index idx, Buffer *buffer=nullptr) {
+                assert(old.refcnt > 0);
 
-        if (idx == 0 && m_levels.size() == 0) return -1;
+                new_ptr = {old.epoch, old.refcnt - 1};
+                if (m_previous_epoch.compare_exchange_strong(old, new_ptr)) {
+                    break;
+                }
+            } else {
+                old = m_current_epoch;
+                /* 
+                 * This could happen if we get into the system during a
+                 * transition. In this case, we can just back out and retry
+                 */
+                if (old.epoch == nullptr) {
+                    continue;
+                }
 
-        bool level_found = false;
-        bool disk_level;
-        level_index merge_level_idx;
+                assert(old.refcnt > 0);
 
-        size_t incoming_rec_cnt = get_level_record_count(idx, buffer);
-        for (level_index i=idx+1; i<m_levels.size(); i++) {
-            if (can_merge_with(i, incoming_rec_cnt)) {
-                return i;
+                new_ptr = {old.epoch, old.refcnt - 1};
+                if (m_current_epoch.compare_exchange_strong(old, new_ptr)) {
+                    break;
+                }
             }
-
-            incoming_rec_cnt = get_level_record_count(i);
-        }
-
-        return -1;
-    }
-
-    /*
-     * Merge the level specified by incoming level into the level specified
-     * by base level. The two levels should be sequential--i.e. no levels
-     * are skipped in the merge process--otherwise the tombstone ordering
-     * invariant may be violated by the merge operation.
-     */
-    inline void merge_levels(level_index base_level, level_index incoming_level) {
-        // merging two memory levels
-        if constexpr (L == LayoutPolicy::LEVELING) {
-            auto tmp = m_levels[base_level];
-            m_levels[base_level] = InternalLevel<R, Shard, Q>::merge_levels(m_levels[base_level], m_levels[incoming_level]);
-            mark_as_unused(tmp);
-        } else {
-            m_levels[base_level]->append_merged_shards(m_levels[incoming_level]);
-        }
-
-        mark_as_unused(m_levels[incoming_level]);
-        m_levels[incoming_level] = new InternalLevel<R, Shard, Q>(incoming_level, (L == LayoutPolicy::LEVELING) ? 1 : m_scale_factor);
-    }
-
-
-    inline void merge_buffer_into_l0(Buffer *buffer) {
-        assert(m_levels[0]);
-        if constexpr (L == LayoutPolicy::LEVELING) {
-            // FIXME: Kludgey implementation due to interface constraints.
-            auto old_level = m_levels[0];
-            auto temp_level = new InternalLevel<R, Shard, Q>(0, 1);
-            temp_level->append_buffer(buffer);
-            auto new_level = InternalLevel<R, Shard, Q>::merge_levels(old_level, temp_level);
-
-            m_levels[0] = new_level;
-            delete temp_level;
-            mark_as_unused(old_level);
-        } else {
-            m_levels[0]->append_buffer(buffer);
-        }
-    }
-
-    /*
-     * Mark a given memory level as no-longer in use by the tree. For now this
-     * will just free the level. In future, this will be more complex as the
-     * level may not be able to immediately be deleted, depending upon who
-     * else is using it.
-     */ 
-    inline void mark_as_unused(InternalLevel<R, Shard, Q> *level) {
-        delete level;
-    }
-
-    /*
-     * Check the tombstone proportion for the specified level and
-     * if the limit is exceeded, forcibly merge levels until all
-     * levels below idx are below the limit.
-     */
-    inline void enforce_delete_maximum(level_index idx) {
-        long double ts_prop = (long double) m_levels[idx]->get_tombstone_count() / (long double) calc_level_record_capacity(idx);
-
-        if (ts_prop > (long double) m_max_delete_prop) {
-            merge_down(idx);
-        }
-
-        return;
-    }
-
-    /*
-     * Assume that level "0" should be larger than the buffer. The buffer
-     * itself is index -1, which should return simply the buffer capacity.
-     */
-    inline size_t calc_level_record_capacity(level_index idx) {
-        return get_buffer()->get_capacity() * pow(m_scale_factor, idx+1);
+        } while (true);
     }
 
-    /*
-     * Returns the actual number of records present on a specified level. An
-     * index value of -1 indicates the memory table. Can optionally pass in
-     * a pointer to the memory table to use, if desired. Otherwise, there are
-     * no guarantees about which buffer will be accessed if level_index is -1.
-     */
-    inline size_t get_level_record_count(level_index idx, Buffer *buffer=nullptr) {
-
-        assert(idx >= -1);
-        if (idx == -1) {
-            return (buffer) ? buffer->get_record_count() : get_buffer()->get_record_count();
-        }
-
-        return (m_levels[idx]) ? m_levels[idx]->get_record_count() : 0;
-    }
-
-    /*
-     * Determines if the specific level can merge with another record containing
-     * incoming_rec_cnt number of records. The provided level index should be 
-     * non-negative (i.e., not refer to the buffer) and will be automatically
-     * translated into the appropriate index into either the disk or memory level
-     * vector.
-     */
-    inline bool can_merge_with(level_index idx, size_t incoming_rec_cnt) {
-        if (idx>= m_levels.size() || !m_levels[idx]) {
-            return false;
-        }
-
-        if (L == LayoutPolicy::LEVELING) {
-            return m_levels[idx]->get_record_count() + incoming_rec_cnt <= calc_level_record_capacity(idx);
-        } else {
-            return m_levels[idx]->get_shard_count() < m_scale_factor;
-        }
-
-        // unreachable
-        assert(true);
-    }
 };
-
 }