cluster/ec: wind readlink on good subvol(s)

BUG: 1232172
Change-Id: I3a56e487840d86147dd85bf5fbe79b165eae289f
Signed-off-by: Pranith Kumar K <pkarampu@redhat.com>
Reviewed-on: http://review.gluster.org/11589
Tested-by: NetBSD Build System <jenkins@build.gluster.org>
Reviewed-by: Xavier Hernandez <xhernandez@datalab.es>
Tested-by: Gluster Build System <jenkins@build.gluster.com>

6 files changed, 90 insertions, 48 deletions

diff --git a/xlators/cluster/ec/src/ec-common.c b/xlators/cluster/ec/src/ec-common.c
index 1c49eba7735..c54e043f5bb 100644
--- a/xlators/cluster/ec/src/ec-common.c
+++ b/xlators/cluster/ec/src/ec-common.c
@@ -596,13 +596,16 @@ void ec_dispatch_one(ec_fop_data_t * fop)
     }
 }
 
-int32_t ec_dispatch_one_retry(ec_fop_data_t *fop, int32_t idx, int32_t op_ret)
+gf_boolean_t
+ec_dispatch_one_retry(ec_fop_data_t *fop, ec_cbk_data_t *cbk)
 {
-    if (op_ret < 0) {
-        return (ec_dispatch_next(fop, idx) >= 0);
-    }
-
-    return 0;
+        if ((cbk->op_ret < 0) && ec_is_recoverable_error (cbk->op_errno)) {
+                GF_ASSERT (fop->mask & (1ULL<<cbk->idx));
+                fop->mask ^= (1ULL << cbk->idx);
+                if (fop->mask)
+                        return _gf_true;
+        }
+        return _gf_false;
 }
 
 void ec_dispatch_inc(ec_fop_data_t * fop)
@@ -1952,6 +1955,7 @@ void __ec_manager(ec_fop_data_t * fop, int32_t error)
         fop->jobs = 1;
 
         fop->state = fop->handler(fop, fop->state);
+        GF_ASSERT (fop->state >= 0);
 
         error = ec_check_complete(fop, __ec_manager);
     } while (error >= 0);
diff --git a/xlators/cluster/ec/src/ec-common.h b/xlators/cluster/ec/src/ec-common.h
index 4c7fe0c820b..1d78f132a94 100644
--- a/xlators/cluster/ec/src/ec-common.h
+++ b/xlators/cluster/ec/src/ec-common.h
@@ -75,7 +75,7 @@ typedef enum {
 #define EC_STATE_HEAL_POST_INODELK_UNLOCK   217
 #define EC_STATE_HEAL_DISPATCH              218
 
-int32_t ec_dispatch_one_retry (ec_fop_data_t *fop, int32_t idx, int32_t op_ret);
+gf_boolean_t ec_dispatch_one_retry (ec_fop_data_t *fop, ec_cbk_data_t *cbk);
 int32_t ec_dispatch_next(ec_fop_data_t * fop, int32_t idx);
 
 void ec_complete(ec_fop_data_t * fop);
diff --git a/xlators/cluster/ec/src/ec-data.c b/xlators/cluster/ec/src/ec-data.c
index 78c505cc704..765686579d7 100644
--- a/xlators/cluster/ec/src/ec-data.c
+++ b/xlators/cluster/ec/src/ec-data.c
@@ -98,6 +98,7 @@ void ec_cbk_data_destroy(ec_cbk_data_t * cbk)
     }
     GF_FREE(cbk->vector);
     gf_dirent_free (&cbk->entries);
+    GF_FREE (cbk->str);
 
     mem_put(cbk);
 }
diff --git a/xlators/cluster/ec/src/ec-data.h b/xlators/cluster/ec/src/ec-data.h
index ec470e9e8c1..18da06cea04 100644
--- a/xlators/cluster/ec/src/ec-data.h
+++ b/xlators/cluster/ec/src/ec-data.h
@@ -253,6 +253,7 @@ struct _ec_cbk_data
     int32_t          op_errno;
     int32_t          count;
     uintptr_t        mask;
+    uint64_t         dirty[2];
 
     dict_t *         xdata;
     dict_t *         dict;
@@ -267,8 +268,8 @@ struct _ec_cbk_data
     struct gf_flock  flock;
     struct iovec *   vector;
     struct iobref *  buffers;
-    uint64_t         dirty[2];
     gf_dirent_t      entries;
+    char            *str;
 };
 
 struct _ec_heal
diff --git a/xlators/cluster/ec/src/ec-dir-read.c b/xlators/cluster/ec/src/ec-dir-read.c
index 5469d62bc3f..09010d5d108 100644
--- a/xlators/cluster/ec/src/ec-dir-read.c
+++ b/xlators/cluster/ec/src/ec-dir-read.c
@@ -438,14 +438,9 @@ int32_t ec_manager_readdir(ec_fop_data_t * fop, int32_t state)
         case EC_STATE_PREPARE_ANSWER:
             cbk = fop->answer;
             if (cbk) {
-                if ((cbk->op_ret < 0) &&
-                    ec_is_recoverable_error (cbk->op_errno)) {
-                    GF_ASSERT (fop->mask & (1ULL<<cbk->idx));
-                    fop->mask ^= (1ULL << cbk->idx);
-                    if (fop->mask == 0)
-                            return EC_STATE_REPORT;
-                    return EC_STATE_DISPATCH;
-                }
+                if (ec_dispatch_one_retry (fop, cbk))
+                        return EC_STATE_DISPATCH;
+
                 if ((cbk->op_ret > 0) && (fop->id == GF_FOP_READDIRP)) {
                     ec_adjust_readdirp (fop->xl->private, cbk->idx,
                                         &cbk->entries);
diff --git a/xlators/cluster/ec/src/ec-inode-read.c b/xlators/cluster/ec/src/ec-inode-read.c
index 29e61bae8fd..2cc374bbfc9 100644
--- a/xlators/cluster/ec/src/ec-inode-read.c
+++ b/xlators/cluster/ec/src/ec-inode-read.c
@@ -84,13 +84,8 @@ ec_manager_access(ec_fop_data_t *fop, int32_t state)
         case EC_STATE_PREPARE_ANSWER:
             cbk = fop->answer;
             if (cbk) {
-                if ((cbk->op_ret < 0) && ec_is_recoverable_error (cbk->op_errno)) {
-                    GF_ASSERT (fop->mask & (1ULL<<cbk->idx));
-                    fop->mask ^= (1ULL << cbk->idx);
-                    if (fop->mask == 0)
-                            return EC_STATE_REPORT;
-                    return EC_STATE_DISPATCH;
-                }
+                if (ec_dispatch_one_retry (fop, cbk))
+                        return EC_STATE_DISPATCH;
             } else {
                 ec_fop_set_error(fop, EIO);
             }
@@ -1029,12 +1024,14 @@ int32_t ec_combine_readlink(ec_fop_data_t * fop, ec_cbk_data_t * dst,
     return 1;
 }
 
-int32_t ec_readlink_cbk(call_frame_t * frame, void * cookie, xlator_t * this,
-                        int32_t op_ret, int32_t op_errno, const char * path,
-                        struct iatt * buf, dict_t * xdata)
+int32_t
+ec_readlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+                 int32_t op_ret, int32_t op_errno, const char *path,
+                 struct iatt *buf, dict_t *xdata)
 {
-    ec_fop_data_t * fop = NULL;
-    int32_t idx = (int32_t)(uintptr_t)cookie;
+    ec_fop_data_t   *fop = NULL;
+    ec_cbk_data_t   *cbk = NULL;
+    int32_t         idx  = (int32_t)(uintptr_t)cookie;
 
     VALIDATE_OR_GOTO(this, out);
     GF_VALIDATE_OR_GOTO(this->name, frame, out);
@@ -1046,25 +1043,26 @@ int32_t ec_readlink_cbk(call_frame_t * frame, void * cookie, xlator_t * this,
     ec_trace("CBK", fop, "idx=%d, frame=%p, op_ret=%d, op_errno=%d", idx,
              frame, op_ret, op_errno);
 
-    if (op_ret > 0)
-    {
-        ec_iatt_rebuild(fop->xl->private, buf, 1, 1);
-    }
-
-    if (!ec_dispatch_one_retry(fop, idx, op_ret))
-    {
-        if (fop->cbks.readlink != NULL)
-        {
-            fop->cbks.readlink(fop->req_frame, fop, this, op_ret, op_errno,
-                               path, buf, xdata);
-        }
+    cbk = ec_cbk_data_allocate (frame, this, fop, fop->id,
+                                idx, op_ret, op_errno);
+    if (cbk) {
+            if (xdata)
+                    cbk->xdata = dict_ref (xdata);
+
+            if (cbk->op_ret >= 0) {
+                    cbk->iatt[0] = *buf;
+                    cbk->str = gf_strdup (path);
+                    if (!cbk->str) {
+                            cbk->op_ret = -1;
+                            cbk->op_errno = ENOMEM;
+                    }
+            }
+            ec_combine (cbk, NULL);
     }
 
 out:
     if (fop != NULL)
-    {
         ec_complete(fop);
-    }
 
     return 0;
 }
@@ -1080,25 +1078,68 @@ void ec_wind_readlink(ec_t * ec, ec_fop_data_t * fop, int32_t idx)
 
 int32_t ec_manager_readlink(ec_fop_data_t * fop, int32_t state)
 {
+    ec_cbk_data_t *cbk = NULL;
+
     switch (state)
     {
         case EC_STATE_INIT:
+        case EC_STATE_LOCK:
+            ec_lock_prepare_inode (fop, &fop->loc[0], EC_QUERY_INFO);
+            ec_lock (fop);
+            return EC_STATE_DISPATCH;
+
         case EC_STATE_DISPATCH:
-            ec_dispatch_one(fop);
+            ec_dispatch_one (fop);
+
+            return EC_STATE_PREPARE_ANSWER;
+
+        case EC_STATE_PREPARE_ANSWER:
+            cbk = fop->answer;
+            if (cbk) {
+                if (ec_dispatch_one_retry (fop, cbk)) {
+                        return EC_STATE_DISPATCH;
+                } else if (cbk->op_ret >= 0) {
+                        ec_iatt_rebuild(fop->xl->private, &cbk->iatt[0], 1, 1);
+                }
+            } else {
+                    ec_fop_set_error(fop, EIO);
+            }
 
             return EC_STATE_REPORT;
 
+        case EC_STATE_REPORT:
+            cbk = fop->answer;
+            GF_ASSERT (cbk);
+            if (fop->cbks.readlink != NULL) {
+                fop->cbks.readlink (fop->req_frame, fop, fop->xl, cbk->op_ret,
+                                    cbk->op_errno, cbk->str, &cbk->iatt[0],
+                                    cbk->xdata);
+            }
+
+            return EC_STATE_LOCK_REUSE;
+
         case -EC_STATE_INIT:
-            if (fop->cbks.readlink != NULL)
-            {
+        case -EC_STATE_LOCK:
+        case -EC_STATE_DISPATCH:
+        case -EC_STATE_PREPARE_ANSWER:
+        case -EC_STATE_REPORT:
+            if (fop->cbks.readlink != NULL) {
                 fop->cbks.readlink(fop->req_frame, fop, fop->xl, -1,
                                    fop->error, NULL, NULL, NULL);
             }
+            return EC_STATE_LOCK_REUSE;
 
-        case EC_STATE_REPORT:
-        case -EC_STATE_REPORT:
-            return EC_STATE_END;
+        case -EC_STATE_LOCK_REUSE:
+        case EC_STATE_LOCK_REUSE:
+            ec_lock_reuse(fop);
 
+            return EC_STATE_UNLOCK;
+
+        case -EC_STATE_UNLOCK:
+        case EC_STATE_UNLOCK:
+            ec_unlock(fop);
+
+            return EC_STATE_END;
         default:
             gf_msg (fop->xl->name, GF_LOG_ERROR, 0,
                     EC_MSG_UNHANDLED_STATE, "Unhandled state %d for %s",