Add rollback op to the OSD code

osd_exec_secondary.cpp

@@ -25,9 +25,10 @@ void osd_t::exec_secondary(osd_op_t *cur_op)
         : (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_WRITE ? BS_OP_WRITE
         : (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_SYNC ? BS_OP_SYNC
         : (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE ? BS_OP_STABLE
+        : (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_ROLLBACK ? BS_OP_ROLLBACK
         : (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_DELETE ? BS_OP_DELETE
         : (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_LIST ? BS_OP_LIST
-        : -1))))));
+        : -1)))))));
     if (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_READ ||
         cur_op->op.hdr.opcode == OSD_OP_SECONDARY_WRITE)
     {
@@ -42,7 +43,8 @@ void osd_t::exec_secondary(osd_op_t *cur_op)
         cur_op->bs_op.oid = cur_op->op.sec_del.oid;
         cur_op->bs_op.version = cur_op->op.sec_del.version;
     }
-    else if (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE)
+    else if (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE ||
+        cur_op->op.hdr.opcode == OSD_OP_SECONDARY_ROLLBACK)
     {
         cur_op->bs_op.len = cur_op->op.sec_stab.len/sizeof(obj_ver_id);
         cur_op->bs_op.buf = cur_op->buf;

osd_ops.h

@@ -13,13 +13,14 @@
 #define OSD_OP_SECONDARY_WRITE      2
 #define OSD_OP_SECONDARY_SYNC       3
 #define OSD_OP_SECONDARY_STABILIZE  4
-#define OSD_OP_SECONDARY_DELETE     5
-#define OSD_OP_TEST_SYNC_STAB_ALL   6
-#define OSD_OP_SECONDARY_LIST       7
-#define OSD_OP_SHOW_CONFIG          8
-#define OSD_OP_READ                 9
-#define OSD_OP_WRITE                10
-#define OSD_OP_MAX                  10
+#define OSD_OP_SECONDARY_ROLLBACK   5
+#define OSD_OP_SECONDARY_DELETE     6
+#define OSD_OP_TEST_SYNC_STAB_ALL   7
+#define OSD_OP_SECONDARY_LIST       8
+#define OSD_OP_SHOW_CONFIG          9
+#define OSD_OP_READ                 10
+#define OSD_OP_WRITE                11
+#define OSD_OP_MAX                  11
 // Alignment & limit for read/write operations
 #define OSD_RW_ALIGN                512
 #define OSD_RW_MAX                  64*1024*1024
@@ -95,18 +96,20 @@ struct __attribute__((__packed__)) osd_reply_secondary_sync_t
     osd_reply_header_t header;
 };
 
-// stabilize objects on the secondary OSD
+// stabilize or rollback objects on the secondary OSD
 struct __attribute__((__packed__)) osd_op_secondary_stabilize_t
 {
     osd_op_header_t header;
     // obj_ver_id array length in bytes
     uint32_t len;
 };
+typedef osd_op_secondary_stabilize_t osd_op_secondary_rollback_t;
 
 struct __attribute__((__packed__)) osd_reply_secondary_stabilize_t
 {
     osd_reply_header_t header;
 };
+typedef osd_reply_secondary_stabilize_t osd_reply_secondary_rollback_t;
 
 // show configuration
 struct __attribute__((__packed__)) osd_op_show_config_t

osd_read.cpp

@@ -102,7 +102,8 @@ void osd_t::handle_read_op(osd_client_t *cl)
     osd_op_t *cur_op = cl->read_op;
     if (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_READ ||
         cur_op->op.hdr.opcode == OSD_OP_SECONDARY_WRITE ||
-        cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE)
+        cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE ||
+        cur_op->op.hdr.opcode == OSD_OP_SECONDARY_ROLLBACK)
     {
         // Allocate a buffer
         cur_op->buf = memalign(512, cur_op->op.sec_rw.len);
@@ -114,6 +115,7 @@ void osd_t::handle_read_op(osd_client_t *cl)
     }
     if (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_WRITE ||
         cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE ||
+        cur_op->op.hdr.opcode == OSD_OP_SECONDARY_ROLLBACK ||
         cur_op->op.hdr.opcode == OSD_OP_WRITE)
     {
         // Read data

osd_send.cpp

@@ -115,7 +115,8 @@ void osd_t::handle_send(ring_data_t *data, int peer_fd)
                             cl.write_remaining = cur_op->op.sec_rw.len;
                             cl.write_state = CL_WRITE_DATA;
                         }
-                        else if (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE)
+                        else if (cur_op->op.hdr.opcode == OSD_OP_SECONDARY_STABILIZE ||
+                            cur_op->op.hdr.opcode == OSD_OP_SECONDARY_ROLLBACK)
                         {
                             cl.write_buf = cur_op->buf;
                             cl.write_remaining = cur_op->op.sec_stab.len;