From 890335bff62e515817cbfc6f2354b68e012ffdfc Mon Sep 17 00:00:00 2001
From: Vitaliy Filippov <vitalif@yourcmc.ru>
Date: Sun, 10 Nov 2019 14:37:45 +0300
Subject: [PATCH] Begin implementation of the STABLE operation

---
 Makefile              |  3 ++-
 blockstore.cpp        |  4 ++--
 blockstore.h          |  6 +++++
 blockstore_stable.cpp | 55 +++++++++++++++++++++++++++++++++++++++++++
 4 files changed, 65 insertions(+), 3 deletions(-)
 create mode 100644 blockstore_stable.cpp

diff --git a/Makefile b/Makefile
index 11377aa1..55dadc14 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,5 @@
-all: allocator.o blockstore.o blockstore_init.o blockstore_open.o blockstore_read.o blockstore_write.o blockstore_sync.o crc32c.o ringloop.o test
+all: allocator.o blockstore.o blockstore_init.o blockstore_open.o blockstore_read.o \
+	blockstore_write.o blockstore_sync.o blockstore_stable.o crc32c.o ringloop.o test
 clean:
 	rm -f *.o
 crc32c.o: crc32c.c
diff --git a/blockstore.cpp b/blockstore.cpp
index 487a5c68..35a84afa 100644
--- a/blockstore.cpp
+++ b/blockstore.cpp
@@ -81,7 +81,7 @@ void blockstore::handle_event(ring_data_t *data)
         }
         else if ((op->flags & OP_TYPE_MASK) == OP_STABLE)
         {
-            
+            handle_stable_event(data, op);
         }
     }
 }
@@ -169,7 +169,7 @@ void blockstore::loop()
             }
             else if ((op->flags & OP_TYPE_MASK) == OP_STABLE)
             {
-                
+                dequeue_op = dequeue_stable(op);
             }
             if (dequeue_op)
             {
diff --git a/blockstore.h b/blockstore.h
index abfa9167..6c59d4ed 100644
--- a/blockstore.h
+++ b/blockstore.h
@@ -162,6 +162,8 @@ public:
 //   to calculate parity for subsequent writes
 // - Writes may be submitted in any order, because they don't overlap. Each write
 //   goes into a new location - either on the journal device or on the data device
+// - Stable (stabilize) must be submitted after sync of that object is completed
+//   It's even OK to return an error to the caller if that object is not synced yet
 // - Journal trim may be processed only after all versions are moved to
 //   the main storage AND after all read operations for older versions complete
 // - If an operation can not be submitted because the ring is full
@@ -286,6 +288,10 @@ class blockstore
     int continue_sync(blockstore_operation *op);
     int ack_sync(blockstore_operation *op);
 
+    // Stable
+    int dequeue_stable(blockstore_operation *op);
+    void handle_stable_event(ring_data_t *data, blockstore_operation *op);
+
 public:
 
     blockstore(spp::sparse_hash_map<std::string, std::string> & config, ring_loop_t *ringloop);
diff --git a/blockstore_stable.cpp b/blockstore_stable.cpp
new file mode 100644
index 00000000..0015a7f2
--- /dev/null
+++ b/blockstore_stable.cpp
@@ -0,0 +1,55 @@
+#include "blockstore.h"
+
+int blockstore::dequeue_stable(blockstore_operation *op)
+{
+    auto dirty_it = dirty_db.find((obj_ver_id){
+        .oid = op->oid,
+        .version = op->version,
+    });
+    if (dirty_it == dirty_db.end())
+    {
+        auto clean_it = object_db.find(op->oid);
+        if (clean_it == object_db.end() || clean_it->second.version < op->version)
+        {
+            // No such object version
+            op->retval = EINVAL;
+        }
+        else
+        {
+            // Already stable
+            op->retval = 0;
+        }
+        op->callback(op);
+        return 1;
+    }
+    else if (IS_UNSYNCED(dirty_it->second.state))
+    {
+        // Object not synced yet. Caller must sync it first
+        op->retval = EAGAIN;
+        op->callback(op);
+        return 1;
+    }
+    else if (IS_STABLE(dirty_it->second.state))
+    {
+        // Already stable
+        op->retval = 0;
+        op->callback(op);
+        return 1;
+    }
+    return 0;
+}
+
+void blockstore::handle_stable_event(ring_data_t *data, blockstore_operation *op)
+{
+    if (data->res < 0)
+    {
+        // sync error
+        // FIXME: our state becomes corrupted after a write error. maybe do something better than just die
+        throw new std::runtime_error("write operation failed. in-memory state is corrupted. AAAAAAAaaaaaaaaa!!!111");
+    }
+    op->pending_ops--;
+    if (op->pending_ops == 0)
+    {
+        
+    }
+}