ec: Fix return errors when not enough bricks

Changes introduced by this patch:

* Fix an incorrect error propagation when the state of the life
  cycle of a fop returns an error.

* Fix incorrect unlocking of failed locks.

* Return ENOTCONN if there aren't enough bricks online.

* In readdir(p) check that the fd has been successfully open by
  a previous opendir.

This is a backport of http://review.gluster.org/9098/

Change-Id: Ib44f25a1297849ebcbab839332f3b6359f275ebe
BUG: 1161066
Signed-off-by: Xavier Hernandez <xhernandez@datalab.es>
Reviewed-on: http://review.gluster.org/9107
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Dan Lambright <dlambrig@redhat.com>
Reviewed-by: Raghavendra Bhat <raghavendra@redhat.com>
Tested-by: Raghavendra Bhat <raghavendra@redhat.com>

1 files changed, 23 insertions, 14 deletions

diff --git a/xlators/cluster/ec/src/ec-common.c b/xlators/cluster/ec/src/ec-common.c
index 2d69ac0f384..7da9e25f2c8 100644
--- a/xlators/cluster/ec/src/ec-common.c
+++ b/xlators/cluster/ec/src/ec-common.c
@@ -306,6 +306,7 @@ void ec_resume_parent(ec_fop_data_t * fop, int32_t error)
     parent = fop->parent;
     if (parent != NULL)
     {
+        ec_trace("RESUME_PARENT", fop, "error=%u", error);
         fop->parent = NULL;
         ec_resume(parent, error);
     }
@@ -1137,7 +1138,7 @@ int32_t ec_unlocked(call_frame_t *frame, void *cookie, xlator_t *this,
 
 void ec_unlock_lock(ec_fop_data_t *fop, ec_lock_t *lock)
 {
-    if (lock->mask != 0) {
+    if ((lock->mask != 0) && lock->acquired) {
         ec_owner_set(fop->frame, lock);
 
         switch (lock->kind) {
@@ -1316,9 +1317,7 @@ void ec_unlock_timer_add(ec_lock_link_t *link)
         lock->refs--;
 
         UNLOCK(&lock->loc.inode->lock);
-    } else {
-        ec_trace("UNLOCK_DELAY", fop, "lock=%p", lock);
-
+    } else if (lock->acquired) {
         delay.tv_sec = 1;
         delay.tv_nsec = 0;
 
@@ -1336,6 +1335,8 @@ void ec_unlock_timer_add(ec_lock_link_t *link)
 
             *lock->plock = NULL;
             refs = 0;
+        } else {
+            ec_trace("UNLOCK_DELAY", fop, "lock=%p", lock);
         }
 
         UNLOCK(&lock->loc.inode->lock);
@@ -1343,6 +1344,12 @@ void ec_unlock_timer_add(ec_lock_link_t *link)
         if (refs == 0) {
             ec_unlock_now(fop, lock);
         }
+    } else {
+        *lock->plock = NULL;
+
+        UNLOCK(&lock->loc.inode->lock);
+
+        ec_lock_destroy(lock);
     }
 }
 
@@ -1443,23 +1450,25 @@ void ec_lock_reuse(ec_fop_data_t *fop)
 
 void __ec_manager(ec_fop_data_t * fop, int32_t error)
 {
-    do
-    {
-        ec_trace("MANAGER", fop, "error=%d", error);
+    ec_t *ec = fop->xl->private;
 
-        if (fop->state == EC_STATE_END)
-        {
-            ec_fop_data_release(fop);
+    do {
+        ec_trace("MANAGER", fop, "error=%d", error);
 
-            break;
+        if (ec->xl_up_count < ec->fragments) {
+            error = ENOTCONN;
         }
-
-        if (error != 0)
-        {
+        if (error != 0) {
             fop->error = error;
             fop->state = -fop->state;
         }
 
+        if ((fop->state == EC_STATE_END) || (fop->state == -EC_STATE_END)) {
+            ec_fop_data_release(fop);
+
+            break;
+        }
+
         fop->state = fop->handler(fop, fop->state);
 
         error = ec_check_complete(fop, __ec_manager);