cluster/ec: Fix bad management of lock owners

Since the addition of parallel reads patch for ec, a lock can have
more than one owner at the same time. The list of owners was stored
inside the 'owner_list' field of each fop.

The problem was with fops that required more than one lock (like
rename). In this case the same field was used to add the fop to
more than one list, casing an overwrite of the previous list.

This has been solved moving the 'owner_list' field from ec_fop_data_t
to ec_lock_link_t structure.

Change-Id: I6042129f09082497b80782b5704a52c35c78f44d
BUG: 1276031
Signed-off-by: Xavier Hernandez <xhernandez@datalab.es>
Reviewed-on: http://review.gluster.org/12445
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Jeff Darcy <jdarcy@redhat.com>

3 files changed, 11 insertions, 10 deletions

diff --git a/xlators/cluster/ec/src/ec-common.c b/xlators/cluster/ec/src/ec-common.c
index b65f79faa55..a7d6da4038f 100644
--- a/xlators/cluster/ec/src/ec-common.c
+++ b/xlators/cluster/ec/src/ec-common.c
@@ -905,11 +905,11 @@ ec_prepare_update_cbk (call_frame_t *frame, void *cookie,
 
     LOCK(&lock->loc.inode->lock);
 
-    list_for_each_entry(tmp, &lock->owners, owner_list) {
-        if ((tmp->flags & EC_FLAG_WAITING_SIZE) != 0) {
-            tmp->flags ^= EC_FLAG_WAITING_SIZE;
+    list_for_each_entry(link, &lock->owners, owner_list) {
+        if ((link->fop->flags & EC_FLAG_WAITING_SIZE) != 0) {
+            link->fop->flags ^= EC_FLAG_WAITING_SIZE;
 
-            list_add_tail(&tmp->cbk_list, &list);
+            list_add_tail(&link->fop->cbk_list, &list);
         }
     }
 
@@ -1337,7 +1337,7 @@ ec_lock_wake_shared(ec_lock_t *lock, struct list_head *list)
 
         list_move_tail(&link->wait_list, list);
 
-        list_add_tail(&fop->owner_list, &lock->owners);
+        list_add_tail(&link->owner_list, &lock->owners);
 
         ec_lock_update_fd(lock, fop);
     }
@@ -1523,7 +1523,7 @@ ec_lock_assign_owner(ec_lock_link_t *link)
         }
     }
 
-    list_add_tail(&fop->owner_list, &lock->owners);
+    list_add_tail(&link->owner_list, &lock->owners);
 
     assigned = _gf_true;
 
@@ -1557,8 +1557,8 @@ ec_lock_next_owner(ec_lock_link_t *link, ec_cbk_data_t *cbk,
 
     ec_trace("LOCK_DONE", fop, "lock=%p", lock);
 
-    GF_ASSERT(!list_empty(&fop->owner_list));
-    list_del_init(&fop->owner_list);
+    GF_ASSERT(!list_empty(&link->owner_list));
+    list_del_init(&link->owner_list);
     lock->release |= release;
 
     if ((fop->error == 0) && (cbk != NULL) && (cbk->op_ret >= 0)) {
diff --git a/xlators/cluster/ec/src/ec-data.c b/xlators/cluster/ec/src/ec-data.c
index 721b104245f..ba47eebda3a 100644
--- a/xlators/cluster/ec/src/ec-data.c
+++ b/xlators/cluster/ec/src/ec-data.c
@@ -135,12 +135,13 @@ ec_fop_data_t * ec_fop_data_allocate(call_frame_t * frame, xlator_t * this,
         return NULL;
     }
 
-    INIT_LIST_HEAD(&fop->owner_list);
     INIT_LIST_HEAD(&fop->cbk_list);
     INIT_LIST_HEAD(&fop->healer);
     INIT_LIST_HEAD(&fop->answer_list);
     INIT_LIST_HEAD(&fop->pending_list);
+    INIT_LIST_HEAD(&fop->locks[0].owner_list);
     INIT_LIST_HEAD(&fop->locks[0].wait_list);
+    INIT_LIST_HEAD(&fop->locks[1].owner_list);
     INIT_LIST_HEAD(&fop->locks[1].wait_list);
 
     fop->xl = this;
diff --git a/xlators/cluster/ec/src/ec-data.h b/xlators/cluster/ec/src/ec-data.h
index c37bc95cde8..cc5c05aa838 100644
--- a/xlators/cluster/ec/src/ec-data.h
+++ b/xlators/cluster/ec/src/ec-data.h
@@ -167,6 +167,7 @@ struct _ec_lock_link
 {
     ec_lock_t        *lock;
     ec_fop_data_t    *fop;
+    struct list_head  owner_list;
     struct list_head  wait_list;
     gf_boolean_t      update[2];
     loc_t            *base;
@@ -187,7 +188,6 @@ struct _ec_fop_data
     xlator_t          *xl;
     call_frame_t      *req_frame;    /* frame of the calling xlator */
     call_frame_t      *frame;        /* frame used by this fop */
-    struct list_head   owner_list;   /* member of lock owner list */
     struct list_head   cbk_list;     /* sorted list of groups of answers */
     struct list_head   answer_list;  /* list of answers */
     struct list_head   pending_list; /* member of ec_t.pending_fops */