Epoch: Adjusted add empty buffer behavior

Add empty buffer now supports a CAS-like operation, where it will only
add a buffer if the currently active one is still the same as when the
decision to add a buffer was made. This is to support adding new buffers
on insert outside of the merge-lock, so that multiple concurrent threads
cannot add multiple new empty buffers.

2 files changed, 26 insertions, 4 deletions

diff --git a/include/framework/DynamicExtension.h b/include/framework/DynamicExtension.h
index 76722c0..955dbe5 100644
--- a/include/framework/DynamicExtension.h
+++ b/include/framework/DynamicExtension.h
@@ -301,14 +301,22 @@ private:
      * buffer while a new epoch is being created in the background. Returns a
      * pointer to the newly created buffer.
      */
-    Buffer *add_empty_buffer(_Epoch *epoch) {
-        auto new_buffer = new Buffer(m_buffer_capacity, m_buffer_delete_capacity); 
+    Buffer *add_empty_buffer(_Epoch *epoch, Buffer *current_buffer=nullptr) {
+        auto temp_buffer = new Buffer(m_buffer_capacity, m_buffer_delete_capacity); 
 
         std::unique_lock<std::mutex> m_struct_lock;
-        epoch->add_buffer(new_buffer);
+        auto new_buffer = epoch->add_buffer(temp_buffer, current_buffer);
+        /* 
+         * if epoch->add_buffer doesn't add the new buffer, this insert 
+         * won't update the buffer set (duplicate insert)
+         */
         m_buffers.insert(new_buffer);
         m_struct_lock.release();
 
+        if (new_buffer != temp_buffer) {
+            delete temp_buffer;
+        }
+
         return new_buffer;
     }
 
diff --git a/include/framework/scheduling/Epoch.h b/include/framework/scheduling/Epoch.h
index f4aefe9..58fe6cd 100644
--- a/include/framework/scheduling/Epoch.h
+++ b/include/framework/scheduling/Epoch.h
@@ -54,11 +54,25 @@ public:
         }
     }
 
-    void add_buffer(Buffer *buf) {
+    Buffer *add_buffer(Buffer *buf, Buffer *cur_buf=nullptr) {
         assert(buf);
 
+        /* 
+         * if a current buffer is specified, only add the
+         * new buffer if the active buffer is the current,
+         * otherwise just return the active buffer (poor man's
+         * CAS).
+         */
+        if (cur_buf) {
+            auto active_buf = get_active_buffer();
+            if (active_buf != cur_buf) {
+                return active_buf;
+            }
+        }
+
         buf->take_reference();
         m_buffers.push_back(buf);
+        return buf;
     }
 
     void start_job() {