performance/quick-read: fix race-conditions in qr_unlink.

The list of fds on which open needs to be done as part of unlink,
was being modified at different places using different locks.
This resulted in a race-condition where open was marked as in-transit,
but fdctx was removed from the list of fds on which open was being
sent even before open was done. Because of this, open_in_transit would
be set forever (as an open was never actually sent, there would be no
open_cbk called and hence we could not reset the variable), blocking
all the future fd based fops on this fd.

Change-Id: Ie84a55bee578869a9a060a094ba28480e7643ae8
BUG: 819490
Signed-off-by: Raghavendra G <raghavendra@gluster.com>
Reviewed-on: http://review.gluster.com/3371
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vijay@gluster.com>

3 files changed, 80 insertions, 46 deletions

diff --git a/xlators/performance/quick-read/src/quick-read-mem-types.h b/xlators/performance/quick-read/src/quick-read-mem-types.h
index 9a6be76b0e6..73c87c819d0 100644
--- a/xlators/performance/quick-read/src/quick-read-mem-types.h
+++ b/xlators/performance/quick-read/src/quick-read-mem-types.h
@@ -20,7 +20,7 @@ enum gf_qr_mem_types_ {
         gf_qr_mt_qr_conf_t,
         gf_qr_mt_qr_priority_t,
         gf_qr_mt_qr_private_t,
-        gf_qr_mt_qr_dentry_t,
+        gf_qr_mt_qr_unlink_ctx_t,
         gf_qr_mt_end
 };
 #endif
diff --git a/xlators/performance/quick-read/src/quick-read.c b/xlators/performance/quick-read/src/quick-read.c
index ef980a5898a..fed57601577 100644
--- a/xlators/performance/quick-read/src/quick-read.c
+++ b/xlators/performance/quick-read/src/quick-read.c
@@ -51,7 +51,7 @@ qr_local_new (xlator_t *this)
         }
 
         LOCK_INIT (&local->lock);
-        INIT_LIST_HEAD (&local->fd_list);
+        INIT_LIST_HEAD (&local->list);
 out:
         return local;
 }
@@ -710,7 +710,6 @@ qr_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         LOCK_INIT (&qr_fd_ctx->lock);
         INIT_LIST_HEAD (&qr_fd_ctx->waiting_ops);
         INIT_LIST_HEAD (&qr_fd_ctx->inode_list);
-        INIT_LIST_HEAD (&qr_fd_ctx->tmp_list);
         qr_fd_ctx->fd = fd;
 
         qr_fd_ctx->path = gf_strdup (loc->path);
@@ -3175,9 +3174,9 @@ int32_t
 qr_unlink_helper (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                   dict_t *xdata)
 {
-        qr_local_t  *local      = NULL;
-        uint32_t     open_count = 0;
-        qr_fd_ctx_t *fdctx      = NULL, *tmp = NULL;
+        qr_local_t      *local      = NULL;
+        uint32_t         open_count = 0;
+        qr_unlink_ctx_t *unlink_ctx = NULL, *tmp = NULL;
 
         local = frame->local;
 
@@ -3191,33 +3190,56 @@ qr_unlink_helper (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                 goto out;
         }
 
-        list_for_each_entry_safe (fdctx, tmp, &local->fd_list, tmp_list) {
-                fd_unref (fdctx->fd);
+        list_for_each_entry_safe (unlink_ctx, tmp, &local->list, list) {
+                fd_unref (unlink_ctx->fdctx->fd);
+                list_del_init (&unlink_ctx->list);
+                GF_FREE (unlink_ctx);
         }
 
-        STACK_WIND (frame, qr_unlink_cbk, FIRST_CHILD(this),
-                    FIRST_CHILD(this)->fops->unlink, loc, xflag, xdata);
+        if (local->op_ret < 0) {
+                /* unwind even if we couldn't open one fd */
+                QR_STACK_UNWIND (unlink, frame, -1, local->op_errno, NULL, NULL,
+                                 NULL);
+        } else {
+                STACK_WIND (frame, qr_unlink_cbk, FIRST_CHILD(this),
+                            FIRST_CHILD(this)->fops->unlink, loc, xflag, xdata);
+        }
 
 out:
         return 0;
 }
 
 
+qr_unlink_ctx_t *
+qr_unlink_ctx_new ()
+{
+        qr_unlink_ctx_t *ctx = NULL;
+
+        ctx = GF_CALLOC (1, sizeof (*ctx), gf_qr_mt_qr_unlink_ctx_t);
+        if (ctx == NULL) {
+                goto out;
+        }
+
+        INIT_LIST_HEAD (&ctx->list);
+out:
+        return ctx;
+}
+
+
 int32_t
 qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
            dict_t *xdata)
 {
         int32_t           op_errno   = -1, ret = -1, op_ret = -1;
         uint64_t          value      = 0;
-        struct list_head  fd_list    = {0, };
         char              need_open  = 0;
         qr_local_t       *local      = NULL;
-        qr_fd_ctx_t      *fdctx      = NULL, *tmp = NULL;
+        qr_fd_ctx_t      *fdctx      = NULL;
         call_frame_t     *open_frame = NULL;
         call_stub_t      *stub       = NULL;
         qr_inode_t       *qr_inode   = NULL;
         uint32_t          open_count = 0;
-        char              ignore     = 0;
+        qr_unlink_ctx_t  *unlink_ctx = NULL;
 
         ret = inode_ctx_get (loc->inode, this, &value);
         if (ret == 0) {
@@ -3228,30 +3250,18 @@ qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                 goto wind;
         }
 
-        INIT_LIST_HEAD (&fd_list);
-
         local = qr_local_new (this);
         GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, unwind, op_errno,
                                         ENOMEM);
 
         frame->local = local;
 
-        LOCK (&loc->inode->lock);
-        {
-                list_for_each_entry (fdctx, &qr_inode->fd_list, inode_list) {
-                        __fd_ref (fdctx->fd);
-                        list_add_tail (&fdctx->tmp_list, &fd_list);
-                }
-        }
-        UNLOCK (&loc->inode->lock);
-
         op_ret = 0;
 
-        LOCK (&local->lock);
+        LOCK (&loc->inode->lock);
         {
-                list_for_each_entry_safe (fdctx, tmp, &fd_list, tmp_list) {
+                list_for_each_entry (fdctx, &qr_inode->fd_list, inode_list) {
                         need_open = 0;
-                        ignore = 0;
 
                         LOCK (&fdctx->lock);
                         {
@@ -3261,9 +3271,6 @@ qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
 
                                 if ((fdctx->opened)
                                     || (strcmp (loc->path, fdctx->path) != 0)) {
-                                        list_del (&fdctx->tmp_list);
-                                        __fd_unref (fdctx->fd);
-                                        ignore = 1;
                                         goto unlock;
                                 }
 
@@ -3274,6 +3281,14 @@ qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                                 }
 
                                 if (!fdctx->opened) {
+                                        unlink_ctx = qr_unlink_ctx_new ();
+                                        if (unlink_ctx == NULL) {
+                                                op_ret = -1;
+                                                op_errno = ENOMEM;
+                                                fdctx->open_in_transit = 0;
+                                                goto unlock;
+                                        }
+
                                         stub = fop_unlink_stub (frame,
                                                                 qr_unlink_helper,
                                                                 loc, xflag,
@@ -3282,14 +3297,21 @@ qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                                                 op_ret = -1;
                                                 op_errno = ENOMEM;
                                                 fdctx->open_in_transit = 0;
+                                                GF_FREE (unlink_ctx);
                                                 goto unlock;
                                         }
 
                                         list_add_tail (&stub->list,
                                                        &fdctx->waiting_ops);
-                                }
 
-                                local->open_count++;
+                                        local->open_count++;
+
+                                        unlink_ctx->need_open = need_open;
+                                        __fd_ref (fdctx->fd);
+                                        unlink_ctx->fdctx = fdctx;
+                                        list_add_tail (&unlink_ctx->list,
+                                                       &local->list);
+                                }
                         }
                 unlock:
                         UNLOCK (&fdctx->lock);
@@ -3297,16 +3319,11 @@ qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                         if (op_ret == -1) {
                                 break;
                         }
-
-                        if (!need_open && !ignore) {
-                                list_move_tail (&fdctx->tmp_list,
-                                                &local->fd_list);
-                        }
                 }
 
                 open_count = local->open_count;
         }
-        UNLOCK (&local->lock);
+        UNLOCK (&loc->inode->lock);
 
         if (op_ret == -1) {
                 goto unwind;
@@ -3316,13 +3333,17 @@ qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
                 goto wind;
         }
 
-        list_for_each_entry_safe (fdctx, tmp, &fd_list, tmp_list) {
-                LOCK (&local->lock);
-                {
-                        list_move_tail (&fdctx->tmp_list, &local->fd_list);
+        /* no need to hold local->lock, since we are gaurded by condition
+         * local->open_count cannot be zero till we send open on
+         * all the required fds. qr_unlink_helper will not modify
+         * local->list till local->open_count becomes 0.
+         */
+        list_for_each_entry (unlink_ctx, &local->list, list) {
+                if (!unlink_ctx->need_open) {
+                        continue;
                 }
-                UNLOCK (&local->lock);
 
+                fdctx = unlink_ctx->fdctx;
                 open_frame = create_frame (this, this->ctx->pool);
                 if (open_frame == NULL) {
                         qr_resume_pending_ops (fdctx, -1, ENOMEM);
@@ -3337,7 +3358,14 @@ qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc, int xflag,
         return 0;
 
 unwind:
-        QR_STACK_UNWIND (unlink, frame, -1, op_errno, NULL, NULL, NULL);
+        if (local && !list_empty (&local->list)) {
+                list_for_each_entry (unlink_ctx, &local->list, list) {
+                        qr_resume_pending_ops (unlink_ctx->fdctx, -1, op_errno);
+                }
+        } else {
+                QR_STACK_UNWIND (unlink, frame, -1, op_errno, NULL, NULL, NULL);
+        }
+
         return 0;
 
 wind:
diff --git a/xlators/performance/quick-read/src/quick-read.h b/xlators/performance/quick-read/src/quick-read.h
index 28f43a8bcb6..10a04e79c14 100644
--- a/xlators/performance/quick-read/src/quick-read.h
+++ b/xlators/performance/quick-read/src/quick-read.h
@@ -44,7 +44,6 @@ struct qr_fd_ctx {
         struct list_head  waiting_ops;
         gf_lock_t         lock;
         struct list_head  inode_list;
-        struct list_head  tmp_list;
         fd_t             *fd;
         dict_t           *xdata;
 };
@@ -60,7 +59,7 @@ struct qr_local {
         int32_t           op_errno;
         uint32_t          open_count;
         call_stub_t      *stub;
-        struct list_head  fd_list;
+        struct list_head  list;
         gf_lock_t         lock;
 };
 typedef struct qr_local qr_local_t;
@@ -106,6 +105,13 @@ struct qr_private {
 };
 typedef struct qr_private qr_private_t;
 
+struct qr_unlink_ctx {
+        struct list_head  list;
+        qr_fd_ctx_t      *fdctx;
+        char              need_open;
+};
+typedef struct qr_unlink_ctx qr_unlink_ctx_t;
+
 void qr_local_free (qr_local_t *local);
 
 #define QR_STACK_UNWIND(op, frame, params ...) do {             \