From 7c03d771475421c1d5a2bbc135242536af1a371c Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 25 Sep 2023 10:49:36 -0400
Subject: Re-structuring Project + scheduling updates

This is a big one--probably should have split it apart, but I'm feeling
lazy this morning.

* Organized the mess of header files in include/framework by splitting
them out into their own subdirectories, and renaming a few files to
remove redundancies introduced by the directory structure.

* Introduced a new framework/ShardRequirements.h header file for simpler
shard development. This header simply contains the necessary includes
from framework/* for creating shard files. This should help to remove
structural dependencies from the framework file structure and shards,
as well as centralizing the necessary framework files to make shard
development easier.

* Created a (currently dummy) SchedulerInterface, and make the scheduler
implementation a template parameter of the dynamic extension for easier
testing of various scheduling policies. There's still more work to be
done to fully integrate the scheduler (queries, multiple buffers), but
some more of the necessary framework code for this has been added as well.

* Adjusted the Task interface setup for the scheduler. The task structures
have been removed from ExtensionStructure and placed in their own header
file. Additionally, I started experimenting with using std::variant,
as opposed to inheritence, to implement subtype polymorphism on the
Merge and Query tasks. The scheduler now has a general task queue that
contains both, and std::variant, std::visit, and std::get are used to
manipulate them without virtual functions.

* Removed Alex.h, as it can't build anyway. There's a branch out there
containing the Alex implementation stripped of the C++20 stuff. So
there's no need to keep it here.
---
 include/framework/structure/MutableBuffer.h | 242 ++++++++++++++++++++++++++++
 1 file changed, 242 insertions(+)
 create mode 100644 include/framework/structure/MutableBuffer.h

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
new file mode 100644
index 0000000..9f12175
--- /dev/null
+++ b/include/framework/structure/MutableBuffer.h
@@ -0,0 +1,242 @@
+/*
+ * include/framework/MutableBuffer.h
+ *
+ * Copyright (C) 2023 Douglas Rumbaugh <drumbaugh@psu.edu> 
+ *                    Dong Xie <dongx@psu.edu>
+ *
+ * All rights reserved. Published under the Modified BSD License.
+ *
+ */
+#pragma once
+
+#include <cstdlib>
+#include <atomic>
+#include <condition_variable>
+#include <cassert>
+#include <numeric>
+#include <algorithm>
+#include <type_traits>
+
+#include "psu-util/alignment.h"
+#include "util/bf_config.h"
+#include "psu-ds/BloomFilter.h"
+#include "psu-ds/Alias.h"
+#include "psu-util/timer.h"
+#include "framework/interface/Record.h"
+
+using psudb::CACHELINE_SIZE;
+
+namespace de {
+
+template <RecordInterface R>
+class MutableBuffer {
+public:
+    MutableBuffer(size_t capacity, size_t max_tombstone_cap)
+    : m_cap(capacity), m_tombstone_cap(max_tombstone_cap), m_reccnt(0)
+    , m_tombstonecnt(0), m_weight(0), m_max_weight(0) {
+        m_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
+        m_merge_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
+        m_tombstone_filter = nullptr;
+        if (max_tombstone_cap > 0) {
+            m_tombstone_filter = new psudb::BloomFilter<R>(BF_FPR, max_tombstone_cap, BF_HASH_FUNCS);
+        }
+
+        m_refcnt.store(0);
+        m_deferred_truncate.store(false);
+        m_merging.store(false);
+    }
+
+    ~MutableBuffer() {
+        assert(m_refcnt.load() == 0);
+        assert(m_merging.load() == false);
+
+        if (m_data) free(m_data);
+        if (m_tombstone_filter) delete m_tombstone_filter;
+        if (m_merge_data) free(m_merge_data);
+    }
+
+    template <typename R_ = R>
+    int append(const R &rec, bool tombstone=false) {
+        if (tombstone && m_tombstonecnt + 1 > m_tombstone_cap) return 0;
+
+        int32_t pos = 0;
+        if ((pos = try_advance_tail()) == -1) return 0;
+
+        Wrapped<R> wrec;
+        wrec.rec = rec;
+        wrec.header = 0;
+        if (tombstone) wrec.set_tombstone();
+
+        m_data[pos] = wrec;
+        m_data[pos].header |= (pos << 2);
+
+        if (tombstone) {
+            m_tombstonecnt.fetch_add(1);
+            if (m_tombstone_filter) m_tombstone_filter->insert(rec);
+        }
+
+        if constexpr (WeightedRecordInterface<R_>) {
+            m_weight.fetch_add(rec.weight);
+            double old = m_max_weight.load();
+            while (old < rec.weight) {
+                m_max_weight.compare_exchange_strong(old, rec.weight);
+                old = m_max_weight.load();
+            }
+        } else {
+            m_weight.fetch_add(1);
+        }
+
+        return 1;     
+    }
+
+    bool truncate() {
+        m_tombstonecnt.store(0);
+        m_reccnt.store(0);
+        m_weight.store(0);
+        m_max_weight.store(0);
+        if (m_tombstone_filter) m_tombstone_filter->clear();
+
+        return true;
+    }
+
+    size_t get_record_count() {
+        return m_reccnt;
+    }
+    
+    size_t get_capacity() {
+        return m_cap;
+    }
+
+    bool is_full() {
+        return m_reccnt == m_cap;
+    }
+
+    size_t get_tombstone_count() {
+        return m_tombstonecnt.load();
+    }
+
+    bool delete_record(const R& rec) {
+        auto offset = 0;
+        while (offset < m_reccnt.load()) {
+            if (m_data[offset].rec == rec) {
+                m_data[offset].set_delete();
+                return true;
+            }
+            offset++;
+        }
+
+        return false;
+    }
+
+    bool check_tombstone(const R& rec) {
+        if (m_tombstone_filter && !m_tombstone_filter->lookup(rec)) return false;
+
+        auto offset = 0;
+        while (offset < m_reccnt.load()) {
+            if (m_data[offset].rec == rec && m_data[offset].is_tombstone()) {
+                return true;
+            }
+            offset++;;
+        }
+        return false;
+    }
+
+    size_t get_memory_usage() {
+        return m_cap * sizeof(R);
+    }
+
+    size_t get_aux_memory_usage() {
+        return m_tombstone_filter->get_memory_usage();
+    }
+
+    size_t get_tombstone_capacity() {
+        return m_tombstone_cap;
+    }
+
+    double get_total_weight() {
+        return m_weight.load();
+    }
+
+    Wrapped<R> *get_data() {
+        return m_data;
+    }
+
+    double get_max_weight() {
+        return m_max_weight;
+    }
+
+    bool start_merge() {
+        if (m_merge_lock.try_lock()) {
+            /* there cannot already been an active merge */
+            if (m_merging.load()) {
+                m_merge_lock.unlock();
+                return false;
+            } 
+
+            m_merging.store(true);
+            memcpy(m_merge_data, m_data, sizeof(Wrapped<R>) * m_reccnt.load());
+            return true;
+        }
+
+        /* lock could not be obtained */
+        return false;
+    }
+
+    bool finish_merge() {
+        m_merge_lock.unlock();
+        return true;
+    }
+
+    /*
+     * Concurrency-related operations
+     */
+    bool take_reference() {
+        m_refcnt.fetch_add(1);
+        return true;
+    }
+
+    bool release_reference() {
+        m_refcnt.fetch_add(-1);
+
+        if (m_refcnt.load() == 0 && m_deferred_truncate.load()) {
+            assert(this->truncate());
+        }
+
+        return true;
+    }
+
+    bool active_merge() {
+        return m_merging.load();
+    }
+
+private:
+    int32_t try_advance_tail() {
+        size_t new_tail = m_reccnt.fetch_add(1);
+
+        if (new_tail < m_cap) return new_tail;
+        else return -1;
+    }
+
+    size_t m_cap;
+    size_t m_tombstone_cap;
+    
+    Wrapped<R>* m_data;
+    Wrapped<R>* m_merge_data;
+
+    psudb::BloomFilter<R>* m_tombstone_filter;
+
+    alignas(64) std::atomic<size_t> m_tombstonecnt;
+    alignas(64) std::atomic<uint32_t> m_reccnt;
+    alignas(64) std::atomic<double> m_weight;
+    alignas(64) std::atomic<double> m_max_weight;
+    alignas(64) std::atomic<bool> m_merging;
+    alignas(64) std::atomic<bool> m_deferred_truncate;
+    alignas(64) std::atomic<size_t> m_refcnt;
+
+    alignas(64) std::mutex m_merge_lock;
+    alignas(64) std::mutex m_trunc_lock;
+    alignas(64) std::condition_variable m_trunc_signal;
+
+};
+
+}
-- 
cgit v1.2.3


From 7ecfb22c32b7986ed1a2439c1abbeed298e4153a Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Fri, 20 Oct 2023 17:00:42 -0400
Subject: Initial pass w/ new scheduler setup

currently there's a race condition of some type to sort out.
---
 include/framework/structure/MutableBuffer.h | 11 +++++++++++
 1 file changed, 11 insertions(+)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 9f12175..804ca5e 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -90,12 +90,23 @@ public:
     }
 
     bool truncate() {
+
+        while (active_merge() || m_refcnt.load() > 0)
+            ;
+
+        m_merge_lock.lock();
+
+        while (m_refcnt > 0) 
+            ;
+
         m_tombstonecnt.store(0);
         m_reccnt.store(0);
         m_weight.store(0);
         m_max_weight.store(0);
         if (m_tombstone_filter) m_tombstone_filter->clear();
 
+        m_merge_lock.unlock();
+
         return true;
     }
 
-- 
cgit v1.2.3


From b72103cb11347f0dd108bd2321f29b0d6ab05106 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 23 Oct 2023 13:18:30 -0400
Subject: Bugfixes

---
 include/framework/structure/MutableBuffer.h | 1 +
 1 file changed, 1 insertion(+)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 804ca5e..4e0b5c2 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -195,6 +195,7 @@ public:
 
     bool finish_merge() {
         m_merge_lock.unlock();
+        m_merging.store(false);
         return true;
     }
 
-- 
cgit v1.2.3


From 3afacb7702e6d8fa67749a2a41dc776d315e02a9 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 23 Oct 2023 17:43:22 -0400
Subject: Began moving to an explicit epoch-based system

I started moving over to an explicit Epoch based system, which has
necessitated a ton of changes throughout the code base. This will
ultimately allow for a much cleaner set of abstractions for managing
concurrency.
---
 include/framework/structure/MutableBuffer.h | 4 ++++
 1 file changed, 4 insertions(+)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 4e0b5c2..974dc28 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -217,6 +217,10 @@ public:
         return true;
     }
 
+    size_t get_reference_count() {
+        return m_refcnt.load();
+    }
+
     bool active_merge() {
         return m_merging.load();
     }
-- 
cgit v1.2.3


From d2279e1b96d352a0af1d425dcaaf93e8a26a8d52 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 30 Oct 2023 17:15:05 -0400
Subject: General Comment + Consistency updates

---
 include/framework/structure/MutableBuffer.h | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 974dc28..e0a6962 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -1,7 +1,7 @@
 /*
- * include/framework/MutableBuffer.h
+ * include/framework/structure/MutableBuffer.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.
-- 
cgit v1.2.3


From ca729108869b4143f1eea31f6dde9195decfec9c Mon Sep 17 00:00:00 2001
From: "Douglas B. Rumbaugh" <doug@douglasrumbaugh.com>
Date: Tue, 31 Oct 2023 12:14:57 -0400
Subject: MutableBuffer: removed most concurrency control stuff

The buffer isn't responsible for a lot of CC anymore (just the append
operation), so this code was no longer necessary. Also removed the only
calls to some of these CC operations within the rest of the framework.
---
 include/framework/structure/MutableBuffer.h | 59 +++++------------------------
 1 file changed, 10 insertions(+), 49 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index e0a6962..a70b86b 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -42,13 +42,10 @@ public:
         }
 
         m_refcnt.store(0);
-        m_deferred_truncate.store(false);
-        m_merging.store(false);
     }
 
     ~MutableBuffer() {
         assert(m_refcnt.load() == 0);
-        assert(m_merging.load() == false);
 
         if (m_data) free(m_data);
         if (m_tombstone_filter) delete m_tombstone_filter;
@@ -90,23 +87,12 @@ public:
     }
 
     bool truncate() {
-
-        while (active_merge() || m_refcnt.load() > 0)
-            ;
-
-        m_merge_lock.lock();
-
-        while (m_refcnt > 0) 
-            ;
-
         m_tombstonecnt.store(0);
         m_reccnt.store(0);
         m_weight.store(0);
         m_max_weight.store(0);
         if (m_tombstone_filter) m_tombstone_filter->clear();
 
-        m_merge_lock.unlock();
-
         return true;
     }
 
@@ -176,26 +162,15 @@ public:
         return m_max_weight;
     }
 
+    /*
+     * This operation assumes that no other threads have write access
+     * to the buffer. This will be the case in normal operation, at
+     * present, but may change (in which case this approach will need
+     * to be adjusted). Other threads having read access is perfectly
+     * acceptable, however.
+     */
     bool start_merge() {
-        if (m_merge_lock.try_lock()) {
-            /* there cannot already been an active merge */
-            if (m_merging.load()) {
-                m_merge_lock.unlock();
-                return false;
-            } 
-
-            m_merging.store(true);
-            memcpy(m_merge_data, m_data, sizeof(Wrapped<R>) * m_reccnt.load());
-            return true;
-        }
-
-        /* lock could not be obtained */
-        return false;
-    }
-
-    bool finish_merge() {
-        m_merge_lock.unlock();
-        m_merging.store(false);
+        memcpy(m_merge_data, m_data, sizeof(Wrapped<R>) * m_reccnt.load());
         return true;
     }
 
@@ -208,12 +183,8 @@ public:
     }
 
     bool release_reference() {
+        assert(m_refcnt > 0);
         m_refcnt.fetch_add(-1);
-
-        if (m_refcnt.load() == 0 && m_deferred_truncate.load()) {
-            assert(this->truncate());
-        }
-
         return true;
     }
 
@@ -221,10 +192,6 @@ public:
         return m_refcnt.load();
     }
 
-    bool active_merge() {
-        return m_merging.load();
-    }
-
 private:
     int32_t try_advance_tail() {
         size_t new_tail = m_reccnt.fetch_add(1);
@@ -245,14 +212,8 @@ private:
     alignas(64) std::atomic<uint32_t> m_reccnt;
     alignas(64) std::atomic<double> m_weight;
     alignas(64) std::atomic<double> m_max_weight;
-    alignas(64) std::atomic<bool> m_merging;
-    alignas(64) std::atomic<bool> m_deferred_truncate;
-    alignas(64) std::atomic<size_t> m_refcnt;
-
-    alignas(64) std::mutex m_merge_lock;
-    alignas(64) std::mutex m_trunc_lock;
-    alignas(64) std::condition_variable m_trunc_signal;
 
+    alignas(64) std::atomic<size_t> m_refcnt;
 };
 
 }
-- 
cgit v1.2.3


From 7249af78a3f39bd2852c3f81fe92dc5b647161fb Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 6 Nov 2023 11:33:17 -0500
Subject: MutableBuffer: added explicit tail variable

Use an explicit m_tail variable for insertion, rather than using m_reccnt.
This ensures that the record count doesn't increase despite new records
being inserted, and allows for the m_tail variable to be decremented on
failure without causing the record count to momentarily change.
---
 include/framework/structure/MutableBuffer.h | 17 ++++++++++++-----
 1 file changed, 12 insertions(+), 5 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index a70b86b..ba25cc3 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -33,7 +33,7 @@ class MutableBuffer {
 public:
     MutableBuffer(size_t capacity, size_t max_tombstone_cap)
     : m_cap(capacity), m_tombstone_cap(max_tombstone_cap), m_reccnt(0)
-    , m_tombstonecnt(0), m_weight(0), m_max_weight(0) {
+    , m_tombstonecnt(0), m_weight(0), m_max_weight(0), m_tail(0) {
         m_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
         m_merge_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
         m_tombstone_filter = nullptr;
@@ -83,6 +83,7 @@ public:
             m_weight.fetch_add(1);
         }
 
+        m_reccnt.fetch_add(1);
         return 1;     
     }
 
@@ -91,6 +92,7 @@ public:
         m_reccnt.store(0);
         m_weight.store(0);
         m_max_weight.store(0);
+        m_tail.store(0);
         if (m_tombstone_filter) m_tombstone_filter->clear();
 
         return true;
@@ -193,11 +195,15 @@ public:
     }
 
 private:
-    int32_t try_advance_tail() {
-        size_t new_tail = m_reccnt.fetch_add(1);
+    int64_t try_advance_tail() {
+        int64_t new_tail = m_tail.fetch_add(1);
 
-        if (new_tail < m_cap) return new_tail;
-        else return -1;
+        if (new_tail < m_cap) {
+            return new_tail;
+        } 
+
+        m_tail.fetch_add(-1);
+        return -1;
     }
 
     size_t m_cap;
@@ -210,6 +216,7 @@ private:
 
     alignas(64) std::atomic<size_t> m_tombstonecnt;
     alignas(64) std::atomic<uint32_t> m_reccnt;
+    alignas(64) std::atomic<int64_t> m_tail;
     alignas(64) std::atomic<double> m_weight;
     alignas(64) std::atomic<double> m_max_weight;
 
-- 
cgit v1.2.3


From 357cab549c2ed33970562b84ff6f83923742343d Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Tue, 7 Nov 2023 15:34:24 -0500
Subject: Comment and License updates

---
 include/framework/structure/MutableBuffer.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index ba25cc3..671824f 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -4,7 +4,7 @@
  * 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
-- 
cgit v1.2.3


From 90bb0614fc1d8f1a185a778e31aaf9027c01aeb8 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 13 Nov 2023 11:44:09 -0500
Subject: Tombstone Compaction: re-enabled tombstone compaction

Currently, proactive buffer tombstone compaction is disabled by forcing
the buffer tombstone capacity to match its record capacity. It isn't
clear how to best handle proactive buffer compactions in an environment
where new buffers are spawned anyway.
---
 include/framework/structure/MutableBuffer.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 671824f..8b17091 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -32,7 +32,7 @@ template <RecordInterface R>
 class MutableBuffer {
 public:
     MutableBuffer(size_t capacity, size_t max_tombstone_cap)
-    : m_cap(capacity), m_tombstone_cap(max_tombstone_cap), m_reccnt(0)
+    : m_cap(capacity), m_tombstone_cap(capacity), m_reccnt(0)
     , m_tombstonecnt(0), m_weight(0), m_max_weight(0), m_tail(0) {
         m_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
         m_merge_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
-- 
cgit v1.2.3


From 3c127eda69295cb306739bdd3c5ddccff6026a8d Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 13 Dec 2023 12:39:54 -0500
Subject: Refactoring: corrected a number of names and added more comments

---
 include/framework/structure/MutableBuffer.h | 18 ++++++++++++------
 1 file changed, 12 insertions(+), 6 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 8b17091..58b5fb4 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -1,11 +1,17 @@
 /*
  * include/framework/structure/MutableBuffer.h
  *
- * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu> 
+ * Copyright (C) 2023 Douglas B. Rumbaugh <drumbaugh@psu.edu>
  *                    Dong Xie <dongx@psu.edu>
  *
  * Distributed under the Modified BSD License.
  *
+ * FIXME: currently, the buffer itself is responsible for managing a
+ * secondary buffer for storing sorted records used during buffer flushes. It
+ * probably makes more sense to make the shard being flushed into responsible
+ * for this instead. This would also facilitate simultaneous flushes of multiple
+ * buffers more easily.
+ *
  */
 #pragma once
 
@@ -35,7 +41,7 @@ public:
     : m_cap(capacity), m_tombstone_cap(capacity), m_reccnt(0)
     , m_tombstonecnt(0), m_weight(0), m_max_weight(0), m_tail(0) {
         m_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
-        m_merge_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
+        m_sorted_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
         m_tombstone_filter = nullptr;
         if (max_tombstone_cap > 0) {
             m_tombstone_filter = new psudb::BloomFilter<R>(BF_FPR, max_tombstone_cap, BF_HASH_FUNCS);
@@ -49,7 +55,7 @@ public:
 
         if (m_data) free(m_data);
         if (m_tombstone_filter) delete m_tombstone_filter;
-        if (m_merge_data) free(m_merge_data);
+        if (m_sorted_data) free(m_sorted_data);
     }
 
     template <typename R_ = R>
@@ -171,8 +177,8 @@ public:
      * to be adjusted). Other threads having read access is perfectly
      * acceptable, however.
      */
-    bool start_merge() {
-        memcpy(m_merge_data, m_data, sizeof(Wrapped<R>) * m_reccnt.load());
+    bool start_flush() {
+        memcpy(m_sorted_data, m_data, sizeof(Wrapped<R>) * m_reccnt.load());
         return true;
     }
 
@@ -210,7 +216,7 @@ private:
     size_t m_tombstone_cap;
     
     Wrapped<R>* m_data;
-    Wrapped<R>* m_merge_data;
+    Wrapped<R>* m_sorted_data;
 
     psudb::BloomFilter<R>* m_tombstone_filter;
 
-- 
cgit v1.2.3


From 3c2e6b3b456867d7155b158432b891b84e4e1dd6 Mon Sep 17 00:00:00 2001
From: "Douglas B. Rumbaugh" <doug@douglasrumbaugh.com>
Date: Tue, 9 Jan 2024 14:47:48 -0500
Subject: Initial update of buffer to new specifications

There are a few minor issues that this introduces, however. Global
tracking of a lot of secondary information, such as weights for WIRS/WSS,
or the exact number of tombstones will need to be approached differently
than they have been historically with this new approach.

I've also removed most of the tombstone capacity related code. We had
decided not to bother enforcing this at the buffer level anyway, and it
would greatly increase the complexity of the problem of predicting when
the next compaction will be.

On the whole this new approach seems like it'll simplify a lot. This
commit actually removes significantly more code than it adds.

One minor issue: the currently implementation will have problems
in the circular array indexes once more than 2^64 records have been
inserted. This doesn't seem like a realistic problem at the moment.
---
 include/framework/structure/MutableBuffer.h | 116 +++++++++-------------------
 1 file changed, 37 insertions(+), 79 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 58b5fb4..7bec219 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -6,12 +6,6 @@
  *
  * Distributed under the Modified BSD License.
  *
- * FIXME: currently, the buffer itself is responsible for managing a
- * secondary buffer for storing sorted records used during buffer flushes. It
- * probably makes more sense to make the shard being flushed into responsible
- * for this instead. This would also facilitate simultaneous flushes of multiple
- * buffers more easily.
- *
  */
 #pragma once
 
@@ -29,6 +23,7 @@
 #include "psu-ds/Alias.h"
 #include "psu-util/timer.h"
 #include "framework/interface/Record.h"
+#include "framework/structure/BufferView.h"
 
 using psudb::CACHELINE_SIZE;
 
@@ -36,18 +31,22 @@ namespace de {
 
 template <RecordInterface R>
 class MutableBuffer {
+    friend class BufferView<R>;
 public:
-    MutableBuffer(size_t capacity, size_t max_tombstone_cap)
-    : m_cap(capacity), m_tombstone_cap(capacity), m_reccnt(0)
-    , m_tombstonecnt(0), m_weight(0), m_max_weight(0), m_tail(0) {
-        m_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
-        m_sorted_data = (Wrapped<R>*) psudb::sf_aligned_alloc(CACHELINE_SIZE, capacity*sizeof(Wrapped<R>));
-        m_tombstone_filter = nullptr;
-        if (max_tombstone_cap > 0) {
-            m_tombstone_filter = new psudb::BloomFilter<R>(BF_FPR, max_tombstone_cap, BF_HASH_FUNCS);
+    MutableBuffer(size_t low_watermark, size_t high_watermark, size_t capacity=0) 
+    : m_lwm(low_watermark), m_hwm(high_watermark), m_cap(capacity), m_head(0), m_tail(0) {
+        /* 
+         * default capacity is twice the high water mark, to account for the worst-case
+         * memory requirements.
+         */
+        if (m_cap == 0) {
+            m_cap = m_hwm * 2;
         }
 
-        m_refcnt.store(0);
+        m_data = (Wrapped<R> *) psudb::sf_aligned_alloc(CACHELINE_SIZE, m_cap * sizeof(Wrapped<R>));
+        
+        // FIXME: need to figure out how to detail with tombstones at some point...
+        m_tombstone_filter = new psudb::BloomFilter<R>(BF_FPR, m_hwm, BF_HASH_FUNCS);
     }
 
     ~MutableBuffer() {
@@ -55,13 +54,10 @@ public:
 
         if (m_data) free(m_data);
         if (m_tombstone_filter) delete m_tombstone_filter;
-        if (m_sorted_data) free(m_sorted_data);
     }
 
     template <typename R_ = R>
     int append(const R &rec, bool tombstone=false) {
-        if (tombstone && m_tombstonecnt + 1 > m_tombstone_cap) return 0;
-
         int32_t pos = 0;
         if ((pos = try_advance_tail()) == -1) return 0;
 
@@ -78,26 +74,11 @@ public:
             if (m_tombstone_filter) m_tombstone_filter->insert(rec);
         }
 
-        if constexpr (WeightedRecordInterface<R_>) {
-            m_weight.fetch_add(rec.weight);
-            double old = m_max_weight.load();
-            while (old < rec.weight) {
-                m_max_weight.compare_exchange_strong(old, rec.weight);
-                old = m_max_weight.load();
-            }
-        } else {
-            m_weight.fetch_add(1);
-        }
-
-        m_reccnt.fetch_add(1);
         return 1;     
     }
 
     bool truncate() {
         m_tombstonecnt.store(0);
-        m_reccnt.store(0);
-        m_weight.store(0);
-        m_max_weight.store(0);
         m_tail.store(0);
         if (m_tombstone_filter) m_tombstone_filter->clear();
 
@@ -105,7 +86,7 @@ public:
     }
 
     size_t get_record_count() {
-        return m_reccnt;
+        return (m_tail - m_head) % m_cap;
     }
     
     size_t get_capacity() {
@@ -113,7 +94,7 @@ public:
     }
 
     bool is_full() {
-        return m_reccnt == m_cap;
+        return (m_tail % m_cap) >= m_hwm;
     }
 
     size_t get_tombstone_count() {
@@ -121,13 +102,11 @@ public:
     }
 
     bool delete_record(const R& rec) {
-        auto offset = 0;
-        while (offset < m_reccnt.load()) {
-            if (m_data[offset].rec == rec) {
-                m_data[offset].set_delete();
+        for (size_t i=0; i<get_record_count(); i++) {
+            if (m_data[to_idx(i)].rec == rec) {
+                m_data[to_idx(i)].set_delete();
                 return true;
             }
-            offset++;
         }
 
         return false;
@@ -136,13 +115,12 @@ public:
     bool check_tombstone(const R& rec) {
         if (m_tombstone_filter && !m_tombstone_filter->lookup(rec)) return false;
 
-        auto offset = 0;
-        while (offset < m_reccnt.load()) {
-            if (m_data[offset].rec == rec && m_data[offset].is_tombstone()) {
+        for (size_t i=0; i<get_record_count(); i++) {
+            if (m_data[to_idx(i)].rec == rec && m_data[to_idx(i)].is_tombstone()) {
                 return true;
             }
-            offset++;;
         }
+
         return false;
     }
 
@@ -155,31 +133,8 @@ public:
     }
 
     size_t get_tombstone_capacity() {
-        return m_tombstone_cap;
-    }
-
-    double get_total_weight() {
-        return m_weight.load();
-    }
-
-    Wrapped<R> *get_data() {
-        return m_data;
-    }
-
-    double get_max_weight() {
-        return m_max_weight;
-    }
-
-    /*
-     * This operation assumes that no other threads have write access
-     * to the buffer. This will be the case in normal operation, at
-     * present, but may change (in which case this approach will need
-     * to be adjusted). Other threads having read access is perfectly
-     * acceptable, however.
-     */
-    bool start_flush() {
-        memcpy(m_sorted_data, m_data, sizeof(Wrapped<R>) * m_reccnt.load());
-        return true;
+        // FIXME: tombstone capacity needs figured out again
+        return m_cap;
     }
 
     /*
@@ -202,30 +157,33 @@ public:
 
 private:
     int64_t try_advance_tail() {
-        int64_t new_tail = m_tail.fetch_add(1);
+        int64_t new_tail = m_tail.fetch_add(1) % m_cap;
 
-        if (new_tail < m_cap) {
+        if (new_tail < m_hwm) {
             return new_tail;
-        } 
+        }
 
         m_tail.fetch_add(-1);
         return -1;
     }
 
+    size_t to_idx(size_t i) {
+        return (m_head + i) % m_cap;
+    }
+
     size_t m_cap;
-    size_t m_tombstone_cap;
+
+    size_t m_lwm;
+    size_t m_hwm;
+    
+    alignas(64) std::atomic<size_t> m_tail;
+    alignas(64) std::atomic<size_t> m_head;
     
     Wrapped<R>* m_data;
-    Wrapped<R>* m_sorted_data;
 
     psudb::BloomFilter<R>* m_tombstone_filter;
 
     alignas(64) std::atomic<size_t> m_tombstonecnt;
-    alignas(64) std::atomic<uint32_t> m_reccnt;
-    alignas(64) std::atomic<int64_t> m_tail;
-    alignas(64) std::atomic<double> m_weight;
-    alignas(64) std::atomic<double> m_max_weight;
-
     alignas(64) std::atomic<size_t> m_refcnt;
 };
 
-- 
cgit v1.2.3


From 53879a0d69f5e578710b7125e9b41e516c2371d4 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 10 Jan 2024 17:39:28 -0500
Subject: MutableBuffer+View: Implementation with unit tests

---
 include/framework/structure/MutableBuffer.h | 133 ++++++++++++++++++++--------
 1 file changed, 98 insertions(+), 35 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 7bec219..d57ad6e 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -11,17 +11,11 @@
 
 #include <cstdlib>
 #include <atomic>
-#include <condition_variable>
 #include <cassert>
-#include <numeric>
-#include <algorithm>
-#include <type_traits>
 
 #include "psu-util/alignment.h"
 #include "util/bf_config.h"
 #include "psu-ds/BloomFilter.h"
-#include "psu-ds/Alias.h"
-#include "psu-util/timer.h"
 #include "framework/interface/Record.h"
 #include "framework/structure/BufferView.h"
 
@@ -34,7 +28,8 @@ class MutableBuffer {
     friend class BufferView<R>;
 public:
     MutableBuffer(size_t low_watermark, size_t high_watermark, size_t capacity=0) 
-    : m_lwm(low_watermark), m_hwm(high_watermark), m_cap(capacity), m_head(0), m_tail(0) {
+    : m_lwm(low_watermark), m_hwm(high_watermark), m_cap(capacity), m_head(0), m_tail(0),
+    m_old_head(0), m_head_refcnt(0), m_old_head_refcnt(0) {
         /* 
          * default capacity is twice the high water mark, to account for the worst-case
          * memory requirements.
@@ -50,10 +45,8 @@ public:
     }
 
     ~MutableBuffer() {
-        assert(m_refcnt.load() == 0);
-
-        if (m_data) free(m_data);
-        if (m_tombstone_filter) delete m_tombstone_filter;
+        free(m_data);
+        delete m_tombstone_filter;
     }
 
     template <typename R_ = R>
@@ -94,7 +87,11 @@ public:
     }
 
     bool is_full() {
-        return (m_tail % m_cap) >= m_hwm;
+        return get_record_count() >= m_hwm;
+    }
+
+    bool is_at_low_watermark() {
+        return (m_tail % m_cap) > m_lwm;
     }
 
     size_t get_tombstone_count() {
@@ -125,66 +122,132 @@ public:
     }
 
     size_t get_memory_usage() {
-        return m_cap * sizeof(R);
+        return m_cap * sizeof(Wrapped<R>);
     }
 
     size_t get_aux_memory_usage() {
         return m_tombstone_filter->get_memory_usage();
     }
 
-    size_t get_tombstone_capacity() {
-        // FIXME: tombstone capacity needs figured out again
-        return m_cap;
+    BufferView<R> get_buffer_view() {
+        m_head_refcnt.fetch_add(1);
+        return BufferView(m_data, m_head, m_tail.load(), m_tombstone_filter, (void*) this, release_head_reference);
     }
 
     /*
-     * Concurrency-related operations
+     * Advance the buffer following a reconstruction. Move current
+     * head and head_refcnt into old_head and old_head_refcnt, then
+     * assign new_head to old_head.
      */
-    bool take_reference() {
-        m_refcnt.fetch_add(1);
-        return true;
+    void advance_head(size_t new_head) {
+        assert(new_head > m_head.load());
+        assert(new_head <= m_tail.load());
+        assert(m_old_head_refcnt == 0);
+
+        /*
+         * the order here is very important. We first store zero to the 
+         * old_refcnt (should be zero anyway). Then we move the current 
+         * head to old head. At this point, any new buffer views should
+         * increment the old head refcnt, so no new references to the 
+         * current head will be taken. Then we add the current head 
+         * refcnt to this. This is to ensure that no references get 
+         * dropped. Only after this do we change to the new head
+         */
+        m_old_head_refcnt.store(0);
+        m_old_head.store(m_head.load());
+        m_old_head_refcnt.fetch_add(m_head_refcnt);
+
+        m_head_refcnt.store(0);
+        m_head.store(new_head);
     }
 
-    bool release_reference() {
-        assert(m_refcnt > 0);
-        m_refcnt.fetch_add(-1);
-        return true;
+    void set_low_watermark(size_t lwm) {
+        assert(lwm < m_hwm);
+        m_lwm = lwm;
     }
 
-    size_t get_reference_count() {
-        return m_refcnt.load();
+    size_t get_low_watermark() {
+        return m_lwm;
+    }
+
+    void set_high_watermark(size_t hwm) {
+        assert(hwm > m_lwm);
+        assert(hwm < m_cap);
+        m_hwm = hwm;
+    }
+
+    size_t get_high_watermark() {
+        return m_hwm;
+    }
+
+    size_t get_tail() {
+        return m_tail.load();
+    }
+
+    /*
+     * Note: this returns the available physical storage capacity,
+     * *not* now many more records can be inserted before the
+     * HWM is reached.
+     */
+    size_t get_available_capacity() {
+        return m_cap - (m_tail.load() - m_old_head.load());
     }
 
 private:
     int64_t try_advance_tail() {
-        int64_t new_tail = m_tail.fetch_add(1) % m_cap;
+        size_t old_value = m_tail.load();
+
+        /* if full, fail to advance the tail */
+        if (old_value >= m_hwm) {
+            return -1;
+        }
 
-        if (new_tail < m_hwm) {
-            return new_tail;
+        while (!m_tail.compare_exchange_strong(old_value, old_value+1)) {
+            /* if full, stop trying and fail to advance the tail */
+            if (m_tail.load() >= m_hwm) {
+                return -1;
+            }
         }
 
-        m_tail.fetch_add(-1);
-        return -1;
+        return old_value;
     }
 
     size_t to_idx(size_t i) {
         return (m_head + i) % m_cap;
     }
 
-    size_t m_cap;
+    static void release_head_reference(void *buff, size_t head) {
+        MutableBuffer<R> *buffer = (MutableBuffer<R> *) buff;
+
+        if (head == buffer->m_head.load()) {
+            buffer->m_head_refcnt.fetch_sub(1);
+        } else if (head == buffer->m_old_head.load()) {
+            buffer->m_old_head_refcnt.fetch_sub(1);
+            /* 
+             * if the old head refcnt drops to 0, free
+             * the records by setting old_head = head
+             */
+            if (buffer->m_old_head_refcnt.load() == 0) {
+                buffer->m_old_head.store(buffer->m_head);   
+            }
+        }
+    }
 
     size_t m_lwm;
     size_t m_hwm;
+    size_t m_cap;
     
     alignas(64) std::atomic<size_t> m_tail;
+
     alignas(64) std::atomic<size_t> m_head;
+    alignas(64) std::atomic<size_t> m_head_refcnt;
+
+    alignas(64) std::atomic<size_t> m_old_head;
+    alignas(64) std::atomic<size_t> m_old_head_refcnt;
     
     Wrapped<R>* m_data;
-
     psudb::BloomFilter<R>* m_tombstone_filter;
-
     alignas(64) std::atomic<size_t> m_tombstonecnt;
-    alignas(64) std::atomic<size_t> m_refcnt;
 };
 
 }
-- 
cgit v1.2.3


From eb19677340be6f0befe9da2199e5832af51eea0d Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 10 Jan 2024 18:01:30 -0500
Subject: MutableBuffer: multithreaded insert test + bugfixes

---
 include/framework/structure/MutableBuffer.h | 13 +++++++++----
 1 file changed, 9 insertions(+), 4 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index d57ad6e..a065154 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -49,7 +49,6 @@ public:
         delete m_tombstone_filter;
     }
 
-    template <typename R_ = R>
     int append(const R &rec, bool tombstone=false) {
         int32_t pos = 0;
         if ((pos = try_advance_tail()) == -1) return 0;
@@ -91,7 +90,7 @@ public:
     }
 
     bool is_at_low_watermark() {
-        return (m_tail % m_cap) > m_lwm;
+        return get_record_count() >= m_lwm;
     }
 
     size_t get_tombstone_count() {
@@ -139,10 +138,14 @@ public:
      * head and head_refcnt into old_head and old_head_refcnt, then
      * assign new_head to old_head.
      */
-    void advance_head(size_t new_head) {
+    bool advance_head(size_t new_head) {
         assert(new_head > m_head.load());
         assert(new_head <= m_tail.load());
-        assert(m_old_head_refcnt == 0);
+
+        /* refuse to advance head while there is an old with one references */
+        if (m_old_head_refcnt > 0) {
+            return false;
+        }
 
         /*
          * the order here is very important. We first store zero to the 
@@ -159,6 +162,8 @@ public:
 
         m_head_refcnt.store(0);
         m_head.store(new_head);
+
+        return true;
     }
 
     void set_low_watermark(size_t lwm) {
-- 
cgit v1.2.3


From 5db0f96e9f3d2505b5f751abc133cbf7e13b5129 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Thu, 11 Jan 2024 11:31:33 -0500
Subject: Fixed some potential buffer-related concurrency bugs

---
 include/framework/structure/MutableBuffer.h | 106 +++++++++++++++++-----------
 1 file changed, 63 insertions(+), 43 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index a065154..3a06f0d 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -6,12 +6,22 @@
  *
  * Distributed under the Modified BSD License.
  *
+ * NOTE: Concerning the tombstone count. One possible approach
+ * would be to track the number of tombstones below and above the
+ * low water mark--this would be straightforward to do. Then, if we
+ * *require* that the head only advance up to the LWM, we can get a
+ * correct view on the number of tombstones in the active buffer at
+ * any point in time, and the BufferView will have a pretty good
+ * approximation as well (potentially with a few extra if new inserts
+ * happen between when the tail pointer and tombstone count are fetched)
+ *
  */
 #pragma once
 
 #include <cstdlib>
 #include <atomic>
 #include <cassert>
+#include <immintrin.h>
 
 #include "psu-util/alignment.h"
 #include "util/bf_config.h"
@@ -28,20 +38,22 @@ class MutableBuffer {
     friend class BufferView<R>;
 public:
     MutableBuffer(size_t low_watermark, size_t high_watermark, size_t capacity=0) 
-    : m_lwm(low_watermark), m_hwm(high_watermark), m_cap(capacity), m_head(0), m_tail(0),
-    m_old_head(0), m_head_refcnt(0), m_old_head_refcnt(0) {
-        /* 
-         * default capacity is twice the high water mark, to account for the worst-case
-         * memory requirements.
-         */
-        if (m_cap == 0) {
-            m_cap = m_hwm * 2;
-        }
-
-        m_data = (Wrapped<R> *) psudb::sf_aligned_alloc(CACHELINE_SIZE, m_cap * sizeof(Wrapped<R>));
-        
-        // FIXME: need to figure out how to detail with tombstones at some point...
-        m_tombstone_filter = new psudb::BloomFilter<R>(BF_FPR, m_hwm, BF_HASH_FUNCS);
+        : m_lwm(low_watermark)
+        , m_hwm(high_watermark)
+        , m_cap((capacity == 0) ? 2 * high_watermark : capacity)
+        , m_tail(0)
+        , m_head(0)
+        , m_head_refcnt(0)
+        , m_old_head(0)
+        , m_old_head_refcnt(0)
+        , m_data((Wrapped<R> *) psudb::sf_aligned_alloc(CACHELINE_SIZE, m_cap * sizeof(Wrapped<R>)))
+        , m_tombstone_filter(new psudb::BloomFilter<R>(BF_FPR, m_hwm, BF_HASH_FUNCS))
+        , m_tscnt(0)
+        , m_old_tscnt(0)
+        , m_active_head_advance(false) 
+    {
+        assert(m_cap > m_hwm);
+        assert(m_hwm > m_lwm);
     }
 
     ~MutableBuffer() {
@@ -62,7 +74,7 @@ public:
         m_data[pos].header |= (pos << 2);
 
         if (tombstone) {
-            m_tombstonecnt.fetch_add(1);
+            m_tscnt.fetch_add(1);
             if (m_tombstone_filter) m_tombstone_filter->insert(rec);
         }
 
@@ -70,7 +82,7 @@ public:
     }
 
     bool truncate() {
-        m_tombstonecnt.store(0);
+        m_tscnt.store(0);
         m_tail.store(0);
         if (m_tombstone_filter) m_tombstone_filter->clear();
 
@@ -78,7 +90,7 @@ public:
     }
 
     size_t get_record_count() {
-        return (m_tail - m_head) % m_cap;
+        return m_tail - m_head;
     }
     
     size_t get_capacity() {
@@ -94,30 +106,15 @@ public:
     }
 
     size_t get_tombstone_count() {
-        return m_tombstonecnt.load();
+        return m_tscnt.load();
     }
 
     bool delete_record(const R& rec) {
-        for (size_t i=0; i<get_record_count(); i++) {
-            if (m_data[to_idx(i)].rec == rec) {
-                m_data[to_idx(i)].set_delete();
-                return true;
-            }
-        }
-
-        return false;
-    }
+        return get_buffer_view().delete_record(rec);
+   }
 
     bool check_tombstone(const R& rec) {
-        if (m_tombstone_filter && !m_tombstone_filter->lookup(rec)) return false;
-
-        for (size_t i=0; i<get_record_count(); i++) {
-            if (m_data[to_idx(i)].rec == rec && m_data[to_idx(i)].is_tombstone()) {
-                return true;
-            }
-        }
-
-        return false;
+        return get_buffer_view().check_tombstone(rec);
     }
 
     size_t get_memory_usage() {
@@ -130,7 +127,8 @@ public:
 
     BufferView<R> get_buffer_view() {
         m_head_refcnt.fetch_add(1);
-        return BufferView(m_data, m_head, m_tail.load(), m_tombstone_filter, (void*) this, release_head_reference);
+        auto f = std::bind(release_head_reference, (void *) this, m_head.load());
+        return BufferView<R>(m_data, m_cap, m_head.load(), m_tail.load(), m_tscnt.load(), m_tombstone_filter, f);
     }
 
     /*
@@ -147,6 +145,8 @@ public:
             return false;
         }
 
+        m_active_head_advance.store(true);
+
         /*
          * the order here is very important. We first store zero to the 
          * old_refcnt (should be zero anyway). Then we move the current 
@@ -157,12 +157,14 @@ public:
          * dropped. Only after this do we change to the new head
          */
         m_old_head_refcnt.store(0);
+
         m_old_head.store(m_head.load());
         m_old_head_refcnt.fetch_add(m_head_refcnt);
 
         m_head_refcnt.store(0);
         m_head.store(new_head);
 
+        m_active_head_advance.store(false);
         return true;
     }
 
@@ -212,29 +214,44 @@ private:
             if (m_tail.load() >= m_hwm) {
                 return -1;
             }
+
+            _mm_pause();
         }
 
         return old_value;
     }
 
-    size_t to_idx(size_t i) {
-        return (m_head + i) % m_cap;
+    size_t to_idx(size_t i, size_t head) {
+        return (head + i) % m_cap;
     }
 
     static void release_head_reference(void *buff, size_t head) {
         MutableBuffer<R> *buffer = (MutableBuffer<R> *) buff;
 
-        if (head == buffer->m_head.load()) {
-            buffer->m_head_refcnt.fetch_sub(1);
-        } else if (head == buffer->m_old_head.load()) {
+        /* 
+         * check old head first. During a head transition, the head being 
+         * retired will first be assigned to *both* head and old_head. As
+         * a result, any refcnt updates during this time should be applied
+         * to old_head, even if the current head and the head being released
+         * also match.
+         */
+        if (head == buffer->m_old_head.load()) {
             buffer->m_old_head_refcnt.fetch_sub(1);
             /* 
              * if the old head refcnt drops to 0, free
              * the records by setting old_head = head
+             * before this, spin while the two heads are equal to
+             * avoid 
              */
+            while (buffer->m_active_head_advance.load()) {
+                _mm_pause();
+            }
+
             if (buffer->m_old_head_refcnt.load() == 0) {
                 buffer->m_old_head.store(buffer->m_head);   
             }
+        } else if (head == buffer->m_head.load()) {
+            buffer->m_head_refcnt.fetch_sub(1);
         }
     }
 
@@ -252,7 +269,10 @@ private:
     
     Wrapped<R>* m_data;
     psudb::BloomFilter<R>* m_tombstone_filter;
-    alignas(64) std::atomic<size_t> m_tombstonecnt;
+    alignas(64) std::atomic<size_t> m_tscnt;
+    size_t m_old_tscnt;
+
+    alignas(64) std::atomic<bool> m_active_head_advance;
 };
 
 }
-- 
cgit v1.2.3


From cf178ae74a76b780b655a447531d2114f9f81d98 Mon Sep 17 00:00:00 2001
From: "Douglas B. Rumbaugh" <doug@douglasrumbaugh.com>
Date: Mon, 15 Jan 2024 14:01:36 -0500
Subject: Various single-threaded bug fixes

---
 include/framework/structure/MutableBuffer.h | 26 ++++++++++++++++++++++----
 1 file changed, 22 insertions(+), 4 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 3a06f0d..5b655fc 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -44,7 +44,7 @@ public:
         , m_tail(0)
         , m_head(0)
         , m_head_refcnt(0)
-        , m_old_head(0)
+        , m_old_head(high_watermark)
         , m_old_head_refcnt(0)
         , m_data((Wrapped<R> *) psudb::sf_aligned_alloc(CACHELINE_SIZE, m_cap * sizeof(Wrapped<R>)))
         , m_tombstone_filter(new psudb::BloomFilter<R>(BF_FPR, m_hwm, BF_HASH_FUNCS))
@@ -62,14 +62,18 @@ public:
     }
 
     int append(const R &rec, bool tombstone=false) {
-        int32_t pos = 0;
-        if ((pos = try_advance_tail()) == -1) return 0;
+        int32_t tail = 0;
+        if ((tail = try_advance_tail()) == -1) {
+            return 0;
+        }
 
         Wrapped<R> wrec;
         wrec.rec = rec;
         wrec.header = 0;
         if (tombstone) wrec.set_tombstone();
 
+        size_t pos = tail % m_cap;
+
         m_data[pos] = wrec;
         m_data[pos].header |= (pos << 2);
 
@@ -131,6 +135,13 @@ public:
         return BufferView<R>(m_data, m_cap, m_head.load(), m_tail.load(), m_tscnt.load(), m_tombstone_filter, f);
     }
 
+    BufferView<R> get_flush_buffer_view() {
+        m_head_refcnt.fetch_add(1);
+        auto f = std::bind(release_head_reference, (void *) this, m_head.load());
+        return BufferView<R>(m_data, m_cap, m_head.load(), m_head.load() + m_lwm, m_tscnt.load(), m_tombstone_filter, f);
+
+    }
+
     /*
      * Advance the buffer following a reconstruction. Move current
      * head and head_refcnt into old_head and old_head_refcnt, then
@@ -142,6 +153,7 @@ public:
 
         /* refuse to advance head while there is an old with one references */
         if (m_old_head_refcnt > 0) {
+            fprintf(stderr, "[W]: Refusing to advance head due to remaining reference counts");
             return false;
         }
 
@@ -195,6 +207,10 @@ public:
      * Note: this returns the available physical storage capacity,
      * *not* now many more records can be inserted before the
      * HWM is reached.
+     *
+     * FIXME: this logic is incorrect for the buffer prior to the
+     * first call to advance_head, and will under-report the available
+     * space.
      */
     size_t get_available_capacity() {
         return m_cap - (m_tail.load() - m_old_head.load());
@@ -205,7 +221,7 @@ private:
         size_t old_value = m_tail.load();
 
         /* if full, fail to advance the tail */
-        if (old_value >= m_hwm) {
+        if (old_value - m_head.load() >= m_hwm) {
             return -1;
         }
 
@@ -236,6 +252,7 @@ private:
          * also match.
          */
         if (head == buffer->m_old_head.load()) {
+            assert(buffer->m_old_head_refcnt > 0);
             buffer->m_old_head_refcnt.fetch_sub(1);
             /* 
              * if the old head refcnt drops to 0, free
@@ -251,6 +268,7 @@ private:
                 buffer->m_old_head.store(buffer->m_head);   
             }
         } else if (head == buffer->m_head.load()) {
+            assert(buffer->m_head_refcnt > 0);
             buffer->m_head_refcnt.fetch_sub(1);
         }
     }
-- 
cgit v1.2.3


From b485685968c7ab626d98cc2a84a122d7ca3c68ce Mon Sep 17 00:00:00 2001
From: "Douglas B. Rumbaugh" <doug@douglasrumbaugh.com>
Date: Mon, 15 Jan 2024 15:16:20 -0500
Subject: Use 16-byte CAS to control buffer head

---
 include/framework/structure/MutableBuffer.h | 120 ++++++++++++++--------------
 1 file changed, 59 insertions(+), 61 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 5b655fc..eeb3dc9 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -36,16 +36,20 @@ namespace de {
 template <RecordInterface R>
 class MutableBuffer {
     friend class BufferView<R>;
+
+    struct buffer_head {
+        size_t head_idx;
+        size_t refcnt;
+    };
+    
 public:
     MutableBuffer(size_t low_watermark, size_t high_watermark, size_t capacity=0) 
         : m_lwm(low_watermark)
         , m_hwm(high_watermark)
         , m_cap((capacity == 0) ? 2 * high_watermark : capacity)
         , m_tail(0)
-        , m_head(0)
-        , m_head_refcnt(0)
-        , m_old_head(high_watermark)
-        , m_old_head_refcnt(0)
+        , m_head({0, 0})
+        , m_old_head({high_watermark, 0})
         , m_data((Wrapped<R> *) psudb::sf_aligned_alloc(CACHELINE_SIZE, m_cap * sizeof(Wrapped<R>)))
         , m_tombstone_filter(new psudb::BloomFilter<R>(BF_FPR, m_hwm, BF_HASH_FUNCS))
         , m_tscnt(0)
@@ -94,7 +98,7 @@ public:
     }
 
     size_t get_record_count() {
-        return m_tail - m_head;
+        return m_tail.load() - m_head.load().head_idx;
     }
     
     size_t get_capacity() {
@@ -130,16 +134,17 @@ public:
     }
 
     BufferView<R> get_buffer_view() {
-        m_head_refcnt.fetch_add(1);
-        auto f = std::bind(release_head_reference, (void *) this, m_head.load());
-        return BufferView<R>(m_data, m_cap, m_head.load(), m_tail.load(), m_tscnt.load(), m_tombstone_filter, f);
+        size_t head = get_head();
+        auto f = std::bind(release_head_reference, (void *) this, head);
+
+        return BufferView<R>(m_data, m_cap, head, m_tail.load(), m_tscnt.load(), m_tombstone_filter, f);
     }
 
     BufferView<R> get_flush_buffer_view() {
-        m_head_refcnt.fetch_add(1);
-        auto f = std::bind(release_head_reference, (void *) this, m_head.load());
-        return BufferView<R>(m_data, m_cap, m_head.load(), m_head.load() + m_lwm, m_tscnt.load(), m_tombstone_filter, f);
+        size_t head = get_head();
+        auto f = std::bind(release_head_reference, (void *) this, head);
 
+        return BufferView<R>(m_data, m_cap, head, head + m_lwm, m_tscnt.load(), m_tombstone_filter, f);
     }
 
     /*
@@ -148,38 +153,41 @@ public:
      * assign new_head to old_head.
      */
     bool advance_head(size_t new_head) {
-        assert(new_head > m_head.load());
+        assert(new_head > m_head.load().head_idx);
         assert(new_head <= m_tail.load());
 
         /* refuse to advance head while there is an old with one references */
-        if (m_old_head_refcnt > 0) {
+        if (m_old_head.load().refcnt > 0) {
             fprintf(stderr, "[W]: Refusing to advance head due to remaining reference counts");
             return false;
         }
 
         m_active_head_advance.store(true);
 
-        /*
-         * the order here is very important. We first store zero to the 
-         * old_refcnt (should be zero anyway). Then we move the current 
-         * head to old head. At this point, any new buffer views should
-         * increment the old head refcnt, so no new references to the 
-         * current head will be taken. Then we add the current head 
-         * refcnt to this. This is to ensure that no references get 
-         * dropped. Only after this do we change to the new head
-         */
-        m_old_head_refcnt.store(0);
-
-        m_old_head.store(m_head.load());
-        m_old_head_refcnt.fetch_add(m_head_refcnt);
+        buffer_head new_hd = {new_head, 0};
+        buffer_head cur_hd;
 
-        m_head_refcnt.store(0);
-        m_head.store(new_head);
+        /* move the current head into the old head */
+        do {
+            buffer_head cur_hd = m_head.load();
+            m_old_head.store(cur_hd);
+        } while(!m_head.compare_exchange_strong(cur_hd, new_hd));
 
         m_active_head_advance.store(false);
         return true;
     }
 
+    size_t get_head() {
+        buffer_head cur_hd, new_hd;
+
+        do {
+            cur_hd = m_head.load();
+            new_hd = {cur_hd.head_idx, cur_hd.refcnt + 1};
+        } while(!m_head.compare_exchange_strong(cur_hd, new_hd));
+
+        return new_hd.head_idx;
+    }
+
     void set_low_watermark(size_t lwm) {
         assert(lwm < m_hwm);
         m_lwm = lwm;
@@ -213,7 +221,7 @@ public:
      * space.
      */
     size_t get_available_capacity() {
-        return m_cap - (m_tail.load() - m_old_head.load());
+        return m_cap - (m_tail.load() - m_old_head.load().head_idx);
     }
 
 private:
@@ -221,7 +229,7 @@ private:
         size_t old_value = m_tail.load();
 
         /* if full, fail to advance the tail */
-        if (old_value - m_head.load() >= m_hwm) {
+        if (old_value - m_head.load().head_idx >= m_hwm) {
             return -1;
         }
 
@@ -244,33 +252,26 @@ private:
     static void release_head_reference(void *buff, size_t head) {
         MutableBuffer<R> *buffer = (MutableBuffer<R> *) buff;
 
-        /* 
-         * check old head first. During a head transition, the head being 
-         * retired will first be assigned to *both* head and old_head. As
-         * a result, any refcnt updates during this time should be applied
-         * to old_head, even if the current head and the head being released
-         * also match.
-         */
-        if (head == buffer->m_old_head.load()) {
-            assert(buffer->m_old_head_refcnt > 0);
-            buffer->m_old_head_refcnt.fetch_sub(1);
-            /* 
-             * if the old head refcnt drops to 0, free
-             * the records by setting old_head = head
-             * before this, spin while the two heads are equal to
-             * avoid 
-             */
-            while (buffer->m_active_head_advance.load()) {
-                _mm_pause();
-            }
-
-            if (buffer->m_old_head_refcnt.load() == 0) {
-                buffer->m_old_head.store(buffer->m_head);   
+        buffer_head cur_hd, new_hd;
+        do {
+            if (buffer->m_head.load().head_idx == head) {
+                cur_hd = buffer->m_head;
+                assert(cur_hd.refcnt > 0);
+                new_hd = {cur_hd.head_idx, cur_hd.refcnt-1};
+
+                if (buffer->m_head.compare_exchange_strong(cur_hd, new_hd)) {
+                    break;
+                }
+            } else {
+                cur_hd = buffer->m_old_head;
+                assert(cur_hd.refcnt > 0);
+                new_hd = {cur_hd.head_idx, cur_hd.refcnt-1};
+                if (buffer->m_old_head.compare_exchange_strong(cur_hd, new_hd)) {
+                    break;
+                }
             }
-        } else if (head == buffer->m_head.load()) {
-            assert(buffer->m_head_refcnt > 0);
-            buffer->m_head_refcnt.fetch_sub(1);
-        }
+            _mm_pause();
+        } while(true);
     }
 
     size_t m_lwm;
@@ -279,11 +280,8 @@ private:
     
     alignas(64) std::atomic<size_t> m_tail;
 
-    alignas(64) std::atomic<size_t> m_head;
-    alignas(64) std::atomic<size_t> m_head_refcnt;
-
-    alignas(64) std::atomic<size_t> m_old_head;
-    alignas(64) std::atomic<size_t> m_old_head_refcnt;
+    alignas(64) std::atomic<buffer_head> m_head;
+    alignas(64) std::atomic<buffer_head> m_old_head;
     
     Wrapped<R>* m_data;
     psudb::BloomFilter<R>* m_tombstone_filter;
-- 
cgit v1.2.3


From 138c793b0a58577713d98c98bb140cf1d9c79bee Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 17 Jan 2024 18:22:00 -0500
Subject: Multiple concurrency bug fixes

A poorly organized commit with fixes for a variety of bugs that were
causing missing records. The core problems all appear to be fixed,
though there is an outstanding problem with tombstones not being
completely canceled. A very small number are appearing in the wrong
order during the static structure test.
---
 include/framework/structure/MutableBuffer.h | 58 ++++++++++++++++++-----------
 1 file changed, 37 insertions(+), 21 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index eeb3dc9..7edde2f 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -133,18 +133,18 @@ public:
         return m_tombstone_filter->get_memory_usage();
     }
 
-    BufferView<R> get_buffer_view() {
-        size_t head = get_head();
+    BufferView<R> get_buffer_view(size_t target_head) {
+        size_t head = get_head(target_head);
         auto f = std::bind(release_head_reference, (void *) this, head);
 
         return BufferView<R>(m_data, m_cap, head, m_tail.load(), m_tscnt.load(), m_tombstone_filter, f);
     }
 
-    BufferView<R> get_flush_buffer_view() {
-        size_t head = get_head();
+    BufferView<R> get_buffer_view() {
+        size_t head = get_head(m_head.load().head_idx);
         auto f = std::bind(release_head_reference, (void *) this, head);
 
-        return BufferView<R>(m_data, m_cap, head, head + m_lwm, m_tscnt.load(), m_tombstone_filter, f);
+        return BufferView<R>(m_data, m_cap, head, m_tail.load(), m_tscnt.load(), m_tombstone_filter, f);
     }
 
     /*
@@ -167,23 +167,39 @@ public:
         buffer_head new_hd = {new_head, 0};
         buffer_head cur_hd;
 
-        /* move the current head into the old head */
+        /* replace current head with new head */
         do {
-            buffer_head cur_hd = m_head.load();
-            m_old_head.store(cur_hd);
+            cur_hd = m_head.load();
         } while(!m_head.compare_exchange_strong(cur_hd, new_hd));
 
+        /* move the current head into the old head */
+        m_old_head.store(cur_hd);
+
         m_active_head_advance.store(false);
         return true;
     }
 
-    size_t get_head() {
+    /*
+     * FIXME: If target_head does not match *either* the old_head or the
+     * current_head, this routine will loop infinitely.
+     */
+    size_t get_head(size_t target_head) {
         buffer_head cur_hd, new_hd;
+        bool head_acquired = false;
 
         do {
-            cur_hd = m_head.load();
-            new_hd = {cur_hd.head_idx, cur_hd.refcnt + 1};
-        } while(!m_head.compare_exchange_strong(cur_hd, new_hd));
+            if (m_old_head.load().head_idx == target_head) {
+                cur_hd = m_old_head.load();
+                cur_hd.head_idx = target_head;
+                new_hd = {cur_hd.head_idx, cur_hd.refcnt + 1};
+                head_acquired = m_old_head.compare_exchange_strong(cur_hd, new_hd);
+            } else if (m_head.load().head_idx == target_head){
+                cur_hd = m_head.load();
+                cur_hd.head_idx = target_head;
+                new_hd = {cur_hd.head_idx, cur_hd.refcnt + 1};
+                head_acquired = m_head.compare_exchange_strong(cur_hd, new_hd);
+            }
+        } while(!head_acquired);
 
         return new_hd.head_idx;
     }
@@ -254,22 +270,22 @@ private:
 
         buffer_head cur_hd, new_hd;
         do {
-            if (buffer->m_head.load().head_idx == head) {
-                cur_hd = buffer->m_head;
-                assert(cur_hd.refcnt > 0);
+            if (buffer->m_old_head.load().head_idx == head) {
+                cur_hd = buffer->m_old_head;
+                if (cur_hd.refcnt == 0) continue;
                 new_hd = {cur_hd.head_idx, cur_hd.refcnt-1};
-
-                if (buffer->m_head.compare_exchange_strong(cur_hd, new_hd)) {
+                if (buffer->m_old_head.compare_exchange_strong(cur_hd, new_hd)) {
                     break;
                 }
             } else {
-                cur_hd = buffer->m_old_head;
-                assert(cur_hd.refcnt > 0);
+                cur_hd = buffer->m_head;
+                if (cur_hd.refcnt == 0) continue;
                 new_hd = {cur_hd.head_idx, cur_hd.refcnt-1};
-                if (buffer->m_old_head.compare_exchange_strong(cur_hd, new_hd)) {
+
+                if (buffer->m_head.compare_exchange_strong(cur_hd, new_hd)) {
                     break;
                 }
-            }
+            } 
             _mm_pause();
         } while(true);
     }
-- 
cgit v1.2.3


From 27d36dd9a68e4cf454be2ca7877ece0a34c3e929 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Wed, 31 Jan 2024 15:48:21 -0500
Subject: Insert throughput benchmark

---
 include/framework/structure/MutableBuffer.h | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 7edde2f..94a9c41 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -158,7 +158,7 @@ public:
 
         /* refuse to advance head while there is an old with one references */
         if (m_old_head.load().refcnt > 0) {
-            fprintf(stderr, "[W]: Refusing to advance head due to remaining reference counts");
+            //fprintf(stderr, "[W]: Refusing to advance head due to remaining reference counts\n");
             return false;
         }
 
-- 
cgit v1.2.3


From 0ff3cedf5df9c27bccd3053ce6339e317f87ff76 Mon Sep 17 00:00:00 2001
From: Douglas Rumbaugh <dbr4@psu.edu>
Date: Mon, 5 Feb 2024 15:18:33 -0500
Subject: BufferView: Adjusted BV to avoid repeated modulus operations

---
 include/framework/structure/MutableBuffer.h | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

(limited to 'include/framework/structure/MutableBuffer.h')

diff --git a/include/framework/structure/MutableBuffer.h b/include/framework/structure/MutableBuffer.h
index 94a9c41..415c95a 100644
--- a/include/framework/structure/MutableBuffer.h
+++ b/include/framework/structure/MutableBuffer.h
@@ -230,13 +230,16 @@ public:
     /*
      * Note: this returns the available physical storage capacity,
      * *not* now many more records can be inserted before the
-     * HWM is reached.
-     *
-     * FIXME: this logic is incorrect for the buffer prior to the
-     * first call to advance_head, and will under-report the available
-     * space.
+     * HWM is reached. It considers the old_head to be "free"
+     * when it has no remaining references. This should be true,
+     * but a buggy framework implementation may violate the
+     * assumption.
      */
     size_t get_available_capacity() {
+        if (m_old_head.load().refcnt == 0) {
+            return m_cap - (m_tail.load() - m_head.load().head_idx);
+        }
+
         return m_cap - (m_tail.load() - m_old_head.load().head_idx);
     }
 
-- 
cgit v1.2.3