1 files changed, 209 insertions, 74 deletions

diff --git a/xlators/cluster/ec/src/ec-common.c b/xlators/cluster/ec/src/ec-common.c
index fcae083aa84..92f7657cd75 100644
--- a/xlators/cluster/ec/src/ec-common.c
+++ b/xlators/cluster/ec/src/ec-common.c
@@ -936,78 +936,77 @@ void ec_lock(ec_fop_data_t * fop)
     }
 }
 
+gf_boolean_t
+ec_config_check (ec_fop_data_t *fop, dict_t *xdata)
+{
+    ec_t *ec;
+
+    if (ec_dict_del_config(xdata, EC_XATTR_CONFIG, &fop->config) < 0) {
+        gf_log(fop->xl->name, GF_LOG_ERROR, "Failed to get a valid "
+                                            "config");
+
+        ec_fop_set_error(fop, EIO);
+
+        return _gf_false;
+    }
+
+    ec = fop->xl->private;
+    if ((fop->config.version != EC_CONFIG_VERSION) ||
+        (fop->config.algorithm != EC_CONFIG_ALGORITHM) ||
+        (fop->config.gf_word_size != EC_GF_BITS) ||
+        (fop->config.bricks != ec->nodes) ||
+        (fop->config.redundancy != ec->redundancy) ||
+        (fop->config.chunk_size != EC_METHOD_CHUNK_SIZE)) {
+        uint32_t data_bricks;
+
+        /* This combination of version/algorithm requires the following
+           values. Incorrect values for these fields are a sign of
+           corruption:
+
+             redundancy > 0
+             redundancy * 2 < bricks
+             gf_word_size must be a power of 2
+             chunk_size (in bits) must be a multiple of gf_word_size *
+                 (bricks - redundancy) */
+
+        data_bricks = fop->config.bricks - fop->config.redundancy;
+        if ((fop->config.redundancy < 1) ||
+            (fop->config.redundancy * 2 >= fop->config.bricks) ||
+            !ec_is_power_of_2(fop->config.gf_word_size) ||
+            ((fop->config.chunk_size * 8) % (fop->config.gf_word_size *
+                                             data_bricks) != 0)) {
+            gf_log(fop->xl->name, GF_LOG_ERROR, "Invalid or corrupted config");
+        } else {
+            gf_log(fop->xl->name, GF_LOG_ERROR, "Unsupported config "
+                                                "(V=%u, A=%u, W=%u, "
+                                                "N=%u, R=%u, S=%u)",
+                   fop->config.version, fop->config.algorithm,
+                   fop->config.gf_word_size, fop->config.bricks,
+                   fop->config.redundancy, fop->config.chunk_size);
+        }
+
+        ec_fop_set_error(fop, EIO);
+
+        return _gf_false;
+    }
+
+    return _gf_true;
+}
+
 int32_t ec_get_size_version_set(call_frame_t * frame, void * cookie,
                                 xlator_t * this, int32_t op_ret,
                                 int32_t op_errno, inode_t * inode,
                                 struct iatt * buf, dict_t * xdata,
                                 struct iatt * postparent)
 {
-    ec_t * ec;
     ec_fop_data_t * fop = cookie;
     ec_inode_t * ctx;
     ec_lock_t *lock = NULL;
 
     if (op_ret >= 0)
     {
-        if (buf->ia_type == IA_IFREG)
-        {
-            if (ec_dict_del_config(xdata, EC_XATTR_CONFIG, &fop->config) < 0)
-            {
-                gf_log(this->name, GF_LOG_ERROR, "Failed to get a valid "
-                                                 "config");
-
-                ec_fop_set_error(fop, EIO);
-
-                return 0;
-            }
-            ec = this->private;
-            if ((fop->config.version != EC_CONFIG_VERSION) ||
-                (fop->config.algorithm != EC_CONFIG_ALGORITHM) ||
-                (fop->config.gf_word_size != EC_GF_BITS) ||
-                (fop->config.bricks != ec->nodes) ||
-                (fop->config.redundancy != ec->redundancy) ||
-                (fop->config.chunk_size != EC_METHOD_CHUNK_SIZE))
-            {
-                uint32_t data_bricks;
-
-                // This combination of version/algorithm requires the following
-                // values. Incorrect values for these fields are a sign of
-                // corruption:
-                //
-                //   redundancy > 0
-                //   redundancy * 2 < bricks
-                //   gf_word_size must be a power of 2
-                //   chunk_size (in bits) must be a multiple of gf_word_size *
-                //       (bricks - redundancy)
-
-                data_bricks = fop->config.bricks - fop->config.redundancy;
-                if ((fop->config.redundancy < 1) ||
-                    (fop->config.redundancy * 2 >= fop->config.bricks) ||
-                    !ec_is_power_of_2(fop->config.gf_word_size) ||
-                    ((fop->config.chunk_size * 8) % (fop->config.gf_word_size *
-                                                     data_bricks) != 0))
-                {
-                    gf_log(this->name, GF_LOG_ERROR, "Invalid or corrupted "
-                                                     "config (V=%u, A=%u, "
-                                                     "W=%u, N=%u, R=%u, S=%u)",
-                           fop->config.version, fop->config.algorithm,
-                           fop->config.gf_word_size, fop->config.bricks,
-                           fop->config.redundancy, fop->config.chunk_size);
-                }
-                else
-                {
-                    gf_log(this->name, GF_LOG_ERROR, "Unsupported config "
-                                                     "(V=%u, A=%u, W=%u, "
-                                                     "N=%u, R=%u, S=%u)",
-                           fop->config.version, fop->config.algorithm,
-                           fop->config.gf_word_size, fop->config.bricks,
-                           fop->config.redundancy, fop->config.chunk_size);
-                }
-
-                ec_fop_set_error(fop, EIO);
-
-                return 0;
-            }
+        if ((buf->ia_type == IA_IFREG) && !ec_config_check(fop, xdata)) {
+            return 0;
         }
 
         LOCK(&inode->lock);
@@ -1053,6 +1052,58 @@ int32_t ec_get_size_version_set(call_frame_t * frame, void * cookie,
     return 0;
 }
 
+int32_t ec_prepare_update_cbk(call_frame_t *frame, void *cookie,
+                              xlator_t *this, int32_t op_ret, int32_t op_errno,
+                              dict_t *dict, dict_t *xdata)
+{
+    ec_fop_data_t *fop = cookie, *parent;
+    ec_lock_t *lock = NULL;
+
+    if (op_ret >= 0) {
+        parent = fop->parent;
+        while ((parent != NULL) && (parent->locks[0].lock == NULL)) {
+            parent = parent->parent;
+        }
+        if (parent == NULL) {
+            return 0;
+        }
+
+        lock = parent->locks[0].lock;
+        lock->is_dirty = _gf_true;
+
+        if (!ec_config_check(fop, dict)) {
+            return 0;
+        }
+
+        LOCK(&lock->loc.inode->lock);
+
+        if ((ec_dict_del_number(dict, EC_XATTR_VERSION, &lock->version) != 0) ||
+            (ec_dict_del_number(dict, EC_XATTR_SIZE, &lock->size) != 0)) {
+            UNLOCK(&lock->loc.inode->lock);
+
+            ec_fop_set_error(fop, EIO);
+
+            return 0;
+        }
+
+        lock->have_size = 1;
+
+        UNLOCK(&lock->loc.inode->lock);
+
+        fop->parent->mask &= fop->good;
+
+        fop->parent->pre_size = fop->parent->post_size = lock->size;
+        fop->parent->have_size = 1;
+    } else {
+        gf_log(this->name, GF_LOG_WARNING,
+               "Failed to get size and version (error %d: %s)", op_errno,
+               strerror (op_errno));
+        ec_fop_set_error(fop, op_errno);
+    }
+
+    return 0;
+}
+
 void ec_get_size_version(ec_fop_data_t * fop)
 {
     loc_t loc;
@@ -1085,7 +1136,8 @@ void ec_get_size_version(ec_fop_data_t * fop)
     }
     if ((dict_set_uint64(xdata, EC_XATTR_VERSION, 0) != 0) ||
         (dict_set_uint64(xdata, EC_XATTR_SIZE, 0) != 0) ||
-        (dict_set_uint64(xdata, EC_XATTR_CONFIG, 0) != 0))
+        (dict_set_uint64(xdata, EC_XATTR_CONFIG, 0) != 0) ||
+        (dict_set_uint64(xdata, EC_XATTR_DIRTY, 0) != 0))
     {
         goto out;
     }
@@ -1141,7 +1193,84 @@ out:
         dict_unref(xdata);
     }
 
-    ec_fop_set_error(fop, error);
+    if (error != 0) {
+        ec_fop_set_error(fop, error);
+    }
+}
+
+void ec_prepare_update(ec_fop_data_t *fop)
+{
+    loc_t loc;
+    dict_t *xdata;
+    ec_fop_data_t *tmp;
+    ec_lock_t *lock;
+    uid_t uid;
+    gid_t gid;
+    int32_t error = ENOMEM;
+
+    tmp = fop;
+    while ((tmp != NULL) && (tmp->locks[0].lock == NULL)) {
+        tmp = tmp->parent;
+    }
+    if ((tmp != NULL) && tmp->locks[0].lock->is_dirty) {
+        lock = tmp->locks[0].lock;
+
+        fop->pre_size = fop->post_size = lock->size;
+        fop->have_size = 1;
+
+        return;
+    }
+
+    memset(&loc, 0, sizeof(loc));
+
+    xdata = dict_new();
+    if (xdata == NULL) {
+        goto out;
+    }
+    if ((ec_dict_set_number(xdata, EC_XATTR_VERSION, 0) != 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)) {
+        goto out;
+    }
+
+    uid = fop->frame->root->uid;
+    gid = fop->frame->root->gid;
+
+    fop->frame->root->uid = 0;
+    fop->frame->root->gid = 0;
+
+    error = EIO;
+
+    if (!fop->use_fd) {
+        if (ec_loc_from_loc(fop->xl, &loc, &fop->loc[0]) != 0) {
+            goto out;
+        }
+
+        ec_xattrop(fop->frame, fop->xl, fop->mask, fop->minimum,
+                   ec_prepare_update_cbk, NULL, &loc, GF_XATTROP_ADD_ARRAY64,
+                   xdata, NULL);
+    } else {
+        ec_fxattrop(fop->frame, fop->xl, fop->mask, fop->minimum,
+                   ec_prepare_update_cbk, NULL, fop->fd,
+                   GF_XATTROP_ADD_ARRAY64, xdata, NULL);
+    }
+
+    fop->frame->root->uid = uid;
+    fop->frame->root->gid = gid;
+
+    error = 0;
+
+out:
+    loc_wipe(&loc);
+
+    if (xdata != NULL) {
+        dict_unref(xdata);
+    }
+
+    if (error != 0) {
+        ec_fop_set_error(fop, error);
+    }
 }
 
 int32_t ec_unlocked(call_frame_t *frame, void *cookie, xlator_t *this,
@@ -1221,7 +1350,7 @@ int32_t ec_update_size_version_done(call_frame_t * frame, void * cookie,
 }
 
 void ec_update_size_version(ec_fop_data_t *fop, loc_t *loc, uint64_t version,
-                            uint64_t size, ec_lock_t *lock)
+                            uint64_t size, gf_boolean_t dirty, ec_lock_t *lock)
 {
     dict_t * dict;
     uid_t uid;
@@ -1234,7 +1363,8 @@ void ec_update_size_version(ec_fop_data_t *fop, loc_t *loc, uint64_t version,
         return;
     }
 
-    ec_trace("UPDATE", fop, "version=%ld, size=%ld", version, size);
+    ec_trace("UPDATE", fop, "version=%ld, size=%ld, dirty=%u", version, size,
+             dirty);
 
     dict = dict_new();
     if (dict == NULL)
@@ -1242,14 +1372,18 @@ void ec_update_size_version(ec_fop_data_t *fop, loc_t *loc, uint64_t version,
         goto out;
     }
 
-    if (ec_dict_set_number(dict, EC_XATTR_VERSION, version) != 0)
-    {
-        goto out;
+    if (version != 0) {
+        if (ec_dict_set_number(dict, EC_XATTR_VERSION, version) != 0) {
+            goto out;
+        }
     }
-    if (size != 0)
-    {
-        if (ec_dict_set_number(dict, EC_XATTR_SIZE, size) != 0)
-        {
+    if (size != 0) {
+        if (ec_dict_set_number(dict, EC_XATTR_SIZE, size) != 0) {
+            goto out;
+        }
+    }
+    if (dirty) {
+        if (ec_dict_set_number(dict, EC_XATTR_DIRTY, -1) != 0) {
             goto out;
         }
     }
@@ -1286,9 +1420,9 @@ void ec_unlock_now(ec_fop_data_t *fop, ec_lock_t *lock)
 {
     ec_trace("UNLOCK_NOW", fop, "lock=%p", lock);
 
-    if (lock->version_delta != 0) {
+    if ((lock->version_delta != 0) || lock->is_dirty) {
         ec_update_size_version(fop, &lock->loc, lock->version_delta,
-                               lock->size_delta, lock);
+                               lock->size_delta, lock->is_dirty, lock);
     } else {
         ec_unlock_lock(fop, lock);
     }
@@ -1413,7 +1547,8 @@ void ec_flush_size_version(ec_fop_data_t * fop)
 
     if (version > 0)
     {
-        ec_update_size_version(fop, &lock->loc, version, delta, NULL);
+        ec_update_size_version(fop, &lock->loc, version, delta, _gf_false,
+                               NULL);
     }
 }