cluster/ec: add separate versions for data/entry, metadata

Adding 64 bits in "version" key of extended attributes. First 64 bits (Left)
represents Data version. Last 64 bits (right) represents Meta Data version.

Note: 3.7 and 3.6 version ec can't co-exist with this change because xattrop in
3.6 will fail with ERANGE as the buffer passed to it will be '8' bytes where as
the value will be 16 bytes in 3.7. Where as 3.7 version clients can work with
old version files. For upgrades we need to tell users to complete heals and
then upgrade

BUG: 1215265
Change-Id: Ib85114680cb7e75b8371c984d9f7b6401c1ffb93
Signed-off-by: Ashish Pandey <aspandey@redhat.com>
Reviewed-on: http://review.gluster.org/10312
Reviewed-on: http://review.gluster.org/10626
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Pranith Kumar Karampuri <pkarampu@redhat.com>

8 files changed, 104 insertions, 29 deletions

diff --git a/xlators/cluster/ec/src/ec-common.c b/xlators/cluster/ec/src/ec-common.c
index 897dbe830ab..5422944cfef 100644
--- a/xlators/cluster/ec/src/ec-common.c
+++ b/xlators/cluster/ec/src/ec-common.c
@@ -1022,7 +1022,8 @@ int32_t ec_get_size_version_set(call_frame_t * frame, void * cookie,
         if (ctx != NULL) {
             if (ctx->inode_lock != NULL) {
                 lock = ctx->inode_lock;
-                lock->version = fop->answer->version;
+                lock->version[0] = fop->answer->version[0];
+                lock->version[1] = fop->answer->version[1];
 
                 if (buf->ia_type == IA_IFREG) {
                     lock->have_size = 1;
@@ -1031,7 +1032,8 @@ int32_t ec_get_size_version_set(call_frame_t * frame, void * cookie,
             }
             if (ctx->entry_lock != NULL) {
                 lock = ctx->entry_lock;
-                lock->version = fop->answer->version;
+                lock->version[0] = fop->answer->version[0];
+                lock->version[1] = fop->answer->version[1];
             }
         }
 
@@ -1084,7 +1086,7 @@ ec_prepare_update_cbk (call_frame_t *frame, void *cookie,
     ec_fop_data_t *fop = cookie, *parent;
     ec_lock_t *lock = NULL;
     uint64_t size = 0;
-    uint64_t version = 0;
+    uint64_t version[EC_VERSION_SIZE] = {0, 0};
 
     if (op_ret >= 0) {
         parent = fop->parent;
@@ -1106,9 +1108,10 @@ ec_prepare_update_cbk (call_frame_t *frame, void *cookie,
             }
         }
 
-        if (ec_dict_del_number(dict, EC_XATTR_VERSION, &version) != 0) {
-             ec_fop_set_error(fop, EIO);
-             return 0;
+        if (ec_dict_del_array(dict, EC_XATTR_VERSION, version,
+                               EC_VERSION_SIZE) != 0) {
+            ec_fop_set_error(fop, EIO);
+            return 0;
         }
 
         LOCK(&lock->loc.inode->lock);
@@ -1118,7 +1121,8 @@ ec_prepare_update_cbk (call_frame_t *frame, void *cookie,
             fop->parent->pre_size = fop->parent->post_size = size;
             fop->parent->have_size = lock->have_size = 1;
         }
-        lock->version = version;
+        lock->version[0] = version[0];
+        lock->version[1] = version[1];
 
         UNLOCK(&lock->loc.inode->lock);
 
@@ -1143,6 +1147,7 @@ void ec_get_size_version(ec_fop_data_t * fop)
     uid_t uid;
     gid_t gid;
     int32_t error = ENOMEM;
+    uint64_t version[EC_VERSION_SIZE] = {0, 0};
 
     if (fop->have_size)
     {
@@ -1171,7 +1176,8 @@ void ec_get_size_version(ec_fop_data_t * fop)
     {
         goto out;
     }
-    if ((ec_dict_set_number(xdata, EC_XATTR_VERSION, 0) != 0) ||
+    if ((ec_dict_set_array(xdata, EC_XATTR_VERSION,
+                           version, EC_VERSION_SIZE) != 0) ||
         (ec_dict_set_number(xdata, EC_XATTR_SIZE, 0) != 0) ||
         (ec_dict_set_number(xdata, EC_XATTR_CONFIG, 0) != 0) ||
         (ec_dict_set_number(xdata, EC_XATTR_DIRTY, 0) != 0))
@@ -1240,6 +1246,7 @@ void ec_prepare_update(ec_fop_data_t *fop)
     ec_lock_t *lock;
     uid_t uid;
     gid_t gid;
+    uint64_t version[2] = {0, 0};
     int32_t error = ENOMEM;
 
     tmp = fop;
@@ -1266,7 +1273,8 @@ void ec_prepare_update(ec_fop_data_t *fop)
     if (xdata == NULL) {
         goto out;
     }
-    if ((ec_dict_set_number(xdata, EC_XATTR_VERSION, 0) != 0) ||
+    if ((ec_dict_set_array(xdata, EC_XATTR_VERSION,
+                           version, EC_VERSION_SIZE) != 0) ||
         (ec_dict_set_number(xdata, EC_XATTR_SIZE, 0) != 0) ||
         (ec_dict_set_number(xdata, EC_XATTR_CONFIG, 0) != 0) ||
         (ec_dict_set_number(xdata, EC_XATTR_DIRTY, 1) != 0)) {
@@ -1383,7 +1391,7 @@ int32_t ec_update_size_version_done(call_frame_t * frame, void * cookie,
     return 0;
 }
 
-void ec_update_size_version(ec_fop_data_t *fop, loc_t *loc, uint64_t version,
+void ec_update_size_version(ec_fop_data_t *fop, loc_t *loc, uint64_t version[2],
                             uint64_t size, gf_boolean_t dirty, ec_lock_t *lock)
 {
     dict_t * dict;
@@ -1406,8 +1414,9 @@ void ec_update_size_version(ec_fop_data_t *fop, loc_t *loc, uint64_t version,
         goto out;
     }
 
-    if (version != 0) {
-        if (ec_dict_set_number(dict, EC_XATTR_VERSION, version) != 0) {
+    if (version[0] != 0 || version[1] != 0) {
+        if (ec_dict_set_array(dict, EC_XATTR_VERSION,
+                              version, EC_VERSION_SIZE) != 0) {
             goto out;
         }
     }
@@ -1568,7 +1577,7 @@ void ec_unlock(ec_fop_data_t *fop)
 void ec_flush_size_version(ec_fop_data_t * fop)
 {
     ec_lock_t * lock;
-    uint64_t version, delta;
+    uint64_t version[2], delta;
 
     GF_ASSERT(fop->lock_count == 1);
 
@@ -1578,9 +1587,11 @@ void ec_flush_size_version(ec_fop_data_t * fop)
 
     GF_ASSERT(lock->owner == fop);
 
-    version = lock->version_delta;
+    version[0] = lock->version_delta[0];
+    version[1] = lock->version_delta[1];
     delta = lock->size_delta;
-    lock->version_delta = 0;
+    lock->version_delta[0] = 0;
+    lock->version_delta[1] = 0;
     lock->size_delta = 0;
 
     UNLOCK(&lock->loc.inode->lock);
@@ -1615,7 +1626,11 @@ void ec_lock_reuse(ec_fop_data_t *fop)
         if (((fop->locks_update >> i) & 1) != 0) {
             if (fop->error == 0)
             {
-                lock->version_delta++;
+		if (fop->id == GF_FOP_SETXATTR || fop->id == GF_FOP_SETATTR) {
+                    lock->version_delta[1]++;
+		} else {
+                    lock->version_delta[0]++;
+		}
                 lock->size_delta += fop->post_size - fop->pre_size;
                 if (fop->have_size) {
                     lock->size = fop->post_size;
diff --git a/xlators/cluster/ec/src/ec-data.h b/xlators/cluster/ec/src/ec-data.h
index 80936aeaada..85037f62bb4 100644
--- a/xlators/cluster/ec/src/ec-data.h
+++ b/xlators/cluster/ec/src/ec-data.h
@@ -143,8 +143,8 @@ struct _ec_lock
     int32_t            have_size;
     uint64_t           size;
     uint64_t           size_delta;
-    uint64_t           version;
-    uint64_t           version_delta;
+    uint64_t           version[2];
+    uint64_t           version_delta[2];
     ec_fop_data_t     *owner;
     loc_t              loc;
     union
@@ -249,7 +249,7 @@ struct _ec_cbk_data
     int32_t          int32;
     uintptr_t        uintptr[3];
     uint64_t         size;
-    uint64_t         version;
+    uint64_t         version[2];
     inode_t *        inode;
     fd_t *           fd;
     struct statvfs   statvfs;
@@ -282,7 +282,7 @@ struct _ec_heal
     uintptr_t         fixed;
     uint64_t          offset;
     uint64_t          size;
-    uint64_t          version;
+    uint64_t          version[2];
     uint64_t          raw_size;
 };
 
diff --git a/xlators/cluster/ec/src/ec-dir-write.c b/xlators/cluster/ec/src/ec-dir-write.c
index 866f5120bf6..ffc96bf4351 100644
--- a/xlators/cluster/ec/src/ec-dir-write.c
+++ b/xlators/cluster/ec/src/ec-dir-write.c
@@ -103,6 +103,7 @@ int32_t ec_manager_create(ec_fop_data_t * fop, int32_t state)
     ec_t * ec;
     ec_cbk_data_t * cbk;
     ec_fd_t * ctx;
+    uint64_t version[2] = {0, 0};
 
     switch (state)
     {
@@ -151,7 +152,8 @@ int32_t ec_manager_create(ec_fop_data_t * fop, int32_t state)
                 return EC_STATE_REPORT;
             }
 
-            if (ec_dict_set_number(fop->xdata, EC_XATTR_VERSION, 0) != 0) {
+            if (ec_dict_set_array(fop->xdata, EC_XATTR_VERSION,
+                                  version, EC_VERSION_SIZE) != 0) {
                 fop->error = EIO;
 
                 return EC_STATE_REPORT;
@@ -565,7 +567,7 @@ void ec_wind_mkdir(ec_t * ec, ec_fop_data_t * fop, int32_t idx)
 int32_t ec_manager_mkdir(ec_fop_data_t * fop, int32_t state)
 {
     ec_cbk_data_t * cbk;
-
+    uint64_t version[2] = {0, 0};
     switch (state)
     {
         case EC_STATE_INIT:
@@ -578,9 +580,9 @@ int32_t ec_manager_mkdir(ec_fop_data_t * fop, int32_t state)
                 }
             }
 
-            if (ec_dict_set_number(fop->xdata, EC_XATTR_VERSION, 0) != 0) {
+            if (ec_dict_set_array(fop->xdata, EC_XATTR_VERSION,
+                                  version, EC_VERSION_SIZE) != 0) {
                 fop->error = EIO;
-
                 return EC_STATE_REPORT;
             }
 
@@ -764,6 +766,7 @@ int32_t ec_manager_mknod(ec_fop_data_t * fop, int32_t state)
 {
     ec_t *ec;
     ec_cbk_data_t * cbk;
+    uint64_t version[2] = {0, 0};
 
     switch (state)
     {
@@ -794,7 +797,8 @@ int32_t ec_manager_mknod(ec_fop_data_t * fop, int32_t state)
                     return EC_STATE_REPORT;
                 }
 
-                if (ec_dict_set_number(fop->xdata, EC_XATTR_VERSION, 0) != 0) {
+                if (ec_dict_set_array(fop->xdata, EC_XATTR_VERSION,
+                                      version, EC_VERSION_SIZE) != 0) {
                     fop->error = EIO;
 
                     return EC_STATE_REPORT;
diff --git a/xlators/cluster/ec/src/ec-generic.c b/xlators/cluster/ec/src/ec-generic.c
index fdab9ec2ae9..4cf5a50ecbd 100644
--- a/xlators/cluster/ec/src/ec-generic.c
+++ b/xlators/cluster/ec/src/ec-generic.c
@@ -717,7 +717,8 @@ void ec_lookup_rebuild(ec_t * ec, ec_fop_data_t * fop, ec_cbk_data_t * cbk)
         return;
     }
 
-    ec_dict_del_number(cbk->xdata, EC_XATTR_VERSION, &cbk->version);
+    ec_dict_del_array(cbk->xdata, EC_XATTR_VERSION,
+                      cbk->version, EC_VERSION_SIZE);
 
     if (ec_loc_update(fop->xl, &fop->loc[0], cbk->inode, &cbk->iatt[0]) != 0) {
         cbk->op_ret = -1;
@@ -732,7 +733,8 @@ void ec_lookup_rebuild(ec_t * ec, ec_fop_data_t * fop, ec_cbk_data_t * cbk)
     if ((ctx != NULL) && (ctx->inode_lock != NULL))
     {
         lock = ctx->inode_lock;
-        cbk->version = lock->version;
+        cbk->version[0] = lock->version[0];
+        cbk->version[1] = lock->version[1];
         if (lock->have_size)
         {
             size = lock->size;
diff --git a/xlators/cluster/ec/src/ec-heal.c b/xlators/cluster/ec/src/ec-heal.c
index b8560005b94..6148df904a4 100644
--- a/xlators/cluster/ec/src/ec-heal.c
+++ b/xlators/cluster/ec/src/ec-heal.c
@@ -79,7 +79,8 @@ void ec_heal_lookup_resume(ec_fop_data_t * fop)
             if (cbk->op_ret >= 0)
             {
                 heal->iatt = cbk->iatt[0];
-                heal->version = cbk->version;
+                heal->version[0] = cbk->version[0];
+                heal->version[1] = cbk->version[1];
                 heal->raw_size = cbk->size;
                 heal->fop->pre_size = cbk->iatt[0].ia_size;
                 heal->fop->post_size = cbk->iatt[0].ia_size;
@@ -856,7 +857,8 @@ void ec_heal_setxattr_others(ec_heal_t * heal)
         if ((cbk->iatt[0].ia_type == IA_IFREG) ||
             (cbk->iatt[0].ia_type == IA_IFDIR))
         {
-            if (ec_dict_set_number(xdata, EC_XATTR_VERSION, cbk->version) != 0)
+            if (ec_dict_set_array(xdata, EC_XATTR_VERSION,
+                                  cbk->version, EC_VERSION_SIZE) != 0)
             {
                 goto out;
             }
diff --git a/xlators/cluster/ec/src/ec-helpers.c b/xlators/cluster/ec/src/ec-helpers.c
index 65deba63959..e9d842fcfa9 100644
--- a/xlators/cluster/ec/src/ec-helpers.c
+++ b/xlators/cluster/ec/src/ec-helpers.c
@@ -157,6 +157,52 @@ size_t ec_iov_copy_to(void * dst, struct iovec * vector, int32_t count,
     return total;
 }
 
+int32_t ec_dict_set_array(dict_t *dict, char *key, uint64_t value[],
+                          int32_t size)
+{
+    uint64_t   *ptr = NULL;
+    int32_t     vindex;
+    if (value == NULL)
+        return -1;
+    ptr = GF_MALLOC(sizeof(uint64_t) * size, gf_common_mt_char);
+    if (ptr == NULL) {
+        return -1;
+    }
+    for (vindex = 0; vindex < size; vindex++) {
+         ptr[vindex] = hton64(value[vindex]);
+    }
+    return dict_set_bin(dict, key, ptr, sizeof(uint64_t) * size);
+}
+
+
+int32_t ec_dict_del_array(dict_t *dict, char *key, uint64_t value[],
+                          int32_t size)
+{
+    void    *ptr;
+    int32_t len;
+    int32_t vindex;
+
+    if ((dict == NULL) || (dict_get_ptr_and_len(dict, key, &ptr, &len) != 0)) {
+        return -1;
+    }
+
+    if (len > (size * sizeof(uint64_t)) ||
+        (len % sizeof (uint64_t)))
+            return -1;
+
+    memset (value, 0, size * sizeof(uint64_t));
+    /* 3.6 version ec would have stored version in 64 bit. In that case treat
+     * metadata versions as 0*/
+    size = min (size, len/sizeof(uint64_t));
+    for (vindex = 0; vindex < size; vindex++) {
+         value[vindex] = ntoh64(*((uint64_t *)ptr + vindex));
+    }
+    dict_del(dict, key);
+
+    return 0;
+}
+
+
 int32_t ec_dict_set_number(dict_t * dict, char * key, uint64_t value)
 {
     uint64_t * ptr;
diff --git a/xlators/cluster/ec/src/ec-helpers.h b/xlators/cluster/ec/src/ec-helpers.h
index 11d2707b3c0..df4495138fe 100644
--- a/xlators/cluster/ec/src/ec-helpers.h
+++ b/xlators/cluster/ec/src/ec-helpers.h
@@ -22,6 +22,10 @@ int32_t ec_bits_consume(uint64_t * n);
 size_t ec_iov_copy_to(void * dst, struct iovec * vector, int32_t count,
                       off_t offset, size_t size);
 
+int32_t ec_dict_set_array(dict_t *dict, char *key,
+                          uint64_t *value, int32_t size);
+int32_t ec_dict_del_array(dict_t *dict, char *key,
+                          uint64_t *value, int32_t size);
 int32_t ec_dict_set_number(dict_t * dict, char * key, uint64_t value);
 int32_t ec_dict_del_number(dict_t * dict, char * key, uint64_t * value);
 int32_t ec_dict_set_config(dict_t * dict, char * key, ec_config_t * config);
diff --git a/xlators/cluster/ec/src/ec.h b/xlators/cluster/ec/src/ec.h
index 54c13077993..b8f3e288197 100644
--- a/xlators/cluster/ec/src/ec.h
+++ b/xlators/cluster/ec/src/ec.h
@@ -23,6 +23,8 @@
 #define EC_XATTR_HEAL    EC_XATTR_PREFIX"heal"
 #define EC_XATTR_DIRTY   EC_XATTR_PREFIX"dirty"
 
+#define EC_VERSION_SIZE 2
+
 struct _ec
 {
     xlator_t *        xl;