performance/quick-read: make unlink aware of open-behind behaviour of quick-read.

quick-read should do an open of all the fds that are waiting for to be
opened behind before doing an unlink on a path.

Change-Id: I4c8d65c0511e8a37765ff57e30d99dd89b56c4fc
BUG: 765363
Signed-off-by: Raghavendra G <raghavendra@gluster.com>
Reviewed-on: http://review.gluster.com/506
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Shishir Gowda <shishirng@gluster.com>
Reviewed-by: Vijay Bellur <vijay@gluster.com>

3 files changed, 336 insertions, 24 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 68e893a6fac..fc7ee3f954d 100644
--- a/xlators/performance/quick-read/src/quick-read-mem-types.h
+++ b/xlators/performance/quick-read/src/quick-read-mem-types.h
@@ -29,6 +29,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_end
 };
 #endif
diff --git a/xlators/performance/quick-read/src/quick-read.c b/xlators/performance/quick-read/src/quick-read.c
index fed81446029..dae9f1456c8 100644
--- a/xlators/performance/quick-read/src/quick-read.c
+++ b/xlators/performance/quick-read/src/quick-read.c
@@ -25,6 +25,9 @@
 struct volume_options options[];
 
 void
+_fd_unref (fd_t *fd);
+
+void
 qr_local_free (qr_local_t *local)
 {
         if (local == NULL) {
@@ -46,6 +49,23 @@ out:
 }
 
 
+qr_local_t *
+qr_local_new (xlator_t *this)
+{
+        qr_local_t *local = NULL;
+
+        local = mem_get0 (this->local_pool);
+        if (local == NULL) {
+                goto out;
+        }
+
+        LOCK_INIT (&local->lock);
+        INIT_LIST_HEAD (&local->fd_list);
+out:
+        return local;
+}
+
+
 int32_t
 qr_readv (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
           off_t offset, uint32_t flags, dict_t *xdata);
@@ -150,6 +170,12 @@ qr_fd_ctx_free (qr_fd_ctx_t *qr_fd_ctx)
 
         GF_ASSERT (list_empty (&qr_fd_ctx->waiting_ops));
 
+        LOCK (&qr_fd_ctx->fd->inode->lock);
+        {
+                list_del_init (&qr_fd_ctx->inode_list);
+        }
+        UNLOCK (&qr_fd_ctx->fd->inode->lock);
+
         GF_FREE (qr_fd_ctx->path);
         GF_FREE (qr_fd_ctx);
 
@@ -215,6 +241,8 @@ __qr_inode_alloc (xlator_t *this, char *path, inode_t *inode)
         }
 
         INIT_LIST_HEAD (&qr_inode->lru);
+        INIT_LIST_HEAD (&qr_inode->unlinked_dentries);
+        INIT_LIST_HEAD (&qr_inode->fd_list);
 
         priority = qr_get_priority (&priv->conf, path);
 
@@ -231,6 +259,9 @@ out:
 void
 __qr_inode_free (qr_inode_t *qr_inode)
 {
+        qr_dentry_t *dentry = NULL, *tmp_dentry = NULL;
+        qr_fd_ctx_t *fdctx  = NULL, *tmp_fdctx  = NULL;
+
         GF_VALIDATE_OR_GOTO ("quick-read", qr_inode, out);
 
         if (qr_inode->xattr) {
@@ -239,6 +270,23 @@ __qr_inode_free (qr_inode_t *qr_inode)
 
         list_del (&qr_inode->lru);
 
+        LOCK (&qr_inode->inode->lock);
+        {
+                list_for_each_entry_safe (dentry, tmp_dentry,
+                                          &qr_inode->unlinked_dentries,
+                                          unlink_list) {
+                        list_del_init (&dentry->unlink_list);
+                        GF_FREE (dentry->name);
+                        mem_put (dentry);
+                }
+
+                list_for_each_entry_safe (fdctx, tmp_fdctx, &qr_inode->fd_list,
+                                          inode_list) {
+                        list_del_init (&fdctx->inode_list);
+                }
+        }
+        UNLOCK (&qr_inode->inode->lock);
+
         GF_FREE (qr_inode);
 out:
         return;
@@ -451,7 +499,8 @@ qr_lookup (call_frame_t *frame, xlator_t *this, loc_t *loc,
         }
 
         table = &priv->table;
-        local = mem_get0 (this->local_pool);
+
+        local = qr_local_new (this);
         GF_VALIDATE_OR_GOTO_WITH_ERROR (this->name, local, unwind, op_errno,
                                         ENOMEM);
 
@@ -641,6 +690,7 @@ qr_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         qr_local_t       *local          = NULL;
         qr_private_t     *priv           = NULL;
         qr_inode_table_t *table          = NULL;
+        qr_dentry_t      *dentry         = NULL;
 
         GF_ASSERT (frame);
         GF_VALIDATE_OR_GOTO (frame->this->name, this, unwind);
@@ -660,6 +710,9 @@ 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);
         if (qr_fd_ctx->path == NULL) {
@@ -683,7 +736,7 @@ qr_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
 
         tmp_fd_ctx = NULL;
 
-        local = mem_get0 (this->local_pool);
+        local = qr_local_new (this);
         if (local == NULL) {
                 op_ret = -1;
                 op_errno = ENOMEM;
@@ -735,7 +788,26 @@ qr_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
         } else {
                 op_ret = 0;
                 op_errno = 0;
-                goto unwind;
+
+                LOCK (&fd->inode->lock);
+                {
+                        list_for_each_entry (dentry,
+                                             &qr_inode->unlinked_dentries,
+                                             unlink_list) {
+                                if ((strcmp (dentry->name, loc->name) == 0)
+                                    && (uuid_compare (dentry->pargfid,
+                                                      loc->pargfid) == 0)) {
+                                        op_ret = -1;
+                                        op_errno = ENOENT;
+                                        goto unlock;
+                                }
+                        }
+
+                        list_add_tail (&qr_fd_ctx->inode_list,
+                                       &qr_inode->fd_list);
+                }
+        unlock:
+                UNLOCK (&fd->inode->lock);
         }
 
 unwind:
@@ -917,7 +989,7 @@ qr_validate_cache (call_frame_t *frame, xlator_t *this, fd_t *fd,
         GF_VALIDATE_OR_GOTO (frame->this->name, stub, out);
 
         if (frame->local == NULL) {
-                local = mem_get0 (this->local_pool);
+                local = qr_local_new (this);
                 if (local == NULL) {
                         goto out;
                 }
@@ -1248,7 +1320,8 @@ out:
                                         can_wind = 1;
                                 } else {
                                         if (frame->local == NULL) {
-                                                frame->local = mem_get0 (this->local_pool);
+                                                frame->local
+                                                        = qr_local_new (this);
                                                 if (frame->local == NULL) {
                                                         op_ret = -1;
                                                         op_errno = ENOMEM;
@@ -1435,7 +1508,7 @@ qr_writev (call_frame_t *frame, xlator_t *this, fd_t *fd, struct iovec *vector,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -1588,7 +1661,7 @@ qr_fstat (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -1746,7 +1819,7 @@ qr_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -1906,7 +1979,7 @@ qr_fsetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *dict,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -2069,7 +2142,7 @@ qr_fgetxattr (call_frame_t *frame, xlator_t *this, fd_t *fd, const char *name,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -2210,7 +2283,7 @@ qr_flush (call_frame_t *frame, xlator_t *this, fd_t *fd, dict_t *xdata)
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else if (qr_fd_ctx->open_in_transit) {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -2351,7 +2424,7 @@ qr_fentrylk (call_frame_t *frame, xlator_t *this, const char *volume, fd_t *fd,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -2510,7 +2583,7 @@ qr_finodelk (call_frame_t *frame, xlator_t *this, const char *volume, fd_t *fd,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -2668,7 +2741,7 @@ qr_fsync (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t flags,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -2865,7 +2938,7 @@ qr_ftruncate (call_frame_t *frame, xlator_t *this, fd_t *fd, off_t offset,
                 qr_fd_ctx = (qr_fd_ctx_t *)(long)value;
         }
 
-        local = mem_get0 (this->local_pool);
+        local = qr_local_new (this);
         if (local == NULL) {
                 op_ret = -1;
                 op_errno = ENOMEM;
@@ -3040,7 +3113,7 @@ qr_lk (call_frame_t *frame, xlator_t *this, fd_t *fd, int32_t cmd,
                         if (qr_fd_ctx->opened) {
                                 can_wind = 1;
                         } else {
-                                frame->local = mem_get0 (this->local_pool);
+                                frame->local = qr_local_new (this);
                                 if (frame->local == NULL) {
                                         op_ret = -1;
                                         op_errno = ENOMEM;
@@ -3101,6 +3174,219 @@ ret:
 }
 
 
+static qr_dentry_t *
+qr_dentry_new (xlator_t *this)
+{
+        qr_dentry_t  *dentry = NULL;
+        qr_private_t *priv   = NULL;
+
+        priv = this->private;
+
+        dentry = mem_get0 (priv->dentry_pool);
+        if (dentry == NULL) {
+                goto out;
+        }
+
+        INIT_LIST_HEAD (&dentry->unlink_list);
+
+out:
+        return dentry;
+}
+
+
+int32_t
+qr_unlink_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+               int32_t op_ret, int32_t op_errno, struct iatt *preparent,
+               struct iatt *postparent)
+{
+        QR_STACK_UNWIND (unlink, frame, op_ret, op_errno, preparent,
+                         postparent);
+        return 0;
+}
+
+
+int32_t
+qr_unlink_helper (call_frame_t *frame, xlator_t *this, loc_t *loc)
+{
+        qr_local_t  *local      = NULL;
+        uint32_t     open_count = 0;
+        qr_fd_ctx_t *fdctx      = NULL, *tmp = NULL;
+
+        local = frame->local;
+
+        LOCK (&local->lock);
+        {
+                open_count = --local->open_count;
+        }
+        UNLOCK (&local->lock);
+
+        if (open_count > 0) {
+                goto out;
+        }
+
+        list_for_each_entry_safe (fdctx, tmp, &local->fd_list, tmp_list) {
+                fd_unref (fdctx->fd);
+        }
+
+        STACK_WIND (frame, qr_unlink_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->unlink, loc);
+
+out:
+        return 0;
+}
+
+
+int32_t
+qr_unlink (call_frame_t *frame, xlator_t *this, loc_t *loc)
+{
+        qr_dentry_t      *dentry     = NULL;
+        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;
+        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;
+
+        ret = inode_ctx_get (loc->inode, this, &value);
+        if (ret == 0) {
+                qr_inode = (qr_inode_t *)(unsigned long)value;
+        }
+
+        if (qr_inode == NULL) {
+                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;
+
+        dentry = qr_dentry_new (this);
+        if (dentry == NULL) {
+                op_errno = ENOMEM;
+                goto unwind;
+        }
+
+        dentry->name = gf_strdup (loc->name);
+        uuid_copy (dentry->pargfid, loc->pargfid);
+
+        LOCK (&loc->inode->lock);
+        {
+                list_add_tail (&dentry->unlink_list,
+                               &qr_inode->unlinked_dentries);
+
+                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);
+        {
+                list_for_each_entry_safe (fdctx, tmp, &fd_list, tmp_list) {
+                        need_open = 0;
+                        ignore = 0;
+
+                        LOCK (&fdctx->lock);
+                        {
+                                if ((fdctx->opened)
+                                    || (strcmp (loc->path, fdctx->path) != 0)) {
+                                        list_del (&fdctx->tmp_list);
+                                        __fd_unref (fdctx->fd);
+                                        ignore = 1;
+                                        goto unlock;
+                                }
+
+                                if (!(fdctx->opened
+                                      || fdctx->open_in_transit)) {
+                                        need_open = 1;
+                                        fdctx->open_in_transit = 1;
+                                }
+
+                                if (!fdctx->opened) {
+                                        stub = fop_unlink_stub (frame,
+                                                                qr_unlink_helper,
+                                                                loc);
+                                        if (stub == NULL) {
+                                                op_ret = -1;
+                                                op_errno = ENOMEM;
+                                                fdctx->open_in_transit = 0;
+                                                goto unlock;
+                                        }
+
+                                        list_add_tail (&stub->list,
+                                                       &fdctx->waiting_ops);
+                                }
+
+                                local->open_count++;
+                        }
+                unlock:
+                        UNLOCK (&fdctx->lock);
+
+                        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);
+
+        if (op_ret == -1) {
+                goto unwind;
+        }
+
+        if (open_count == 0) {
+                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);
+                }
+                UNLOCK (&local->lock);
+
+                open_frame = create_frame (this, this->ctx->pool);
+                if (open_frame == NULL) {
+                        qr_resume_pending_ops (fdctx, -1, ENOMEM);
+                        continue;
+                }
+
+                STACK_WIND (open_frame, qr_open_cbk, FIRST_CHILD(this),
+                            FIRST_CHILD(this)->fops->open,
+                            loc, fdctx->flags, fdctx->fd, fdctx->wbflags);
+        }
+
+        return 0;
+
+unwind:
+        QR_STACK_UNWIND (unlink, frame, -1, op_errno, NULL, NULL);
+        return 0;
+
+wind:
+        STACK_WIND (frame, qr_unlink_cbk, FIRST_CHILD(this),
+                    FIRST_CHILD(this)->fops->unlink, loc);
+        return 0;
+}
+
+
 int32_t
 qr_release (xlator_t *this, fd_t *fd)
 {
@@ -3574,6 +3860,14 @@ init (xlator_t *this)
                 goto out;
         }
 
+        priv->dentry_pool = mem_pool_new (qr_dentry_t, 1024);
+        if (priv->dentry_pool == NULL) {
+                ret = -1;
+                gf_log (this->name, GF_LOG_ERROR,
+                        "failed to create dentry memory pool");
+                goto out;
+        }
+
         ret = 0;
 
         this->private = priv;
@@ -3658,6 +3952,7 @@ struct xlator_fops fops = {
         .ftruncate   = qr_ftruncate,
         .lk          = qr_lk,
         .fsetattr    = qr_fsetattr,
+        .unlink      = qr_unlink,
 };
 
 struct xlator_cbks cbks = {
diff --git a/xlators/performance/quick-read/src/quick-read.h b/xlators/performance/quick-read/src/quick-read.h
index 064151b634e..95bd62eda45 100644
--- a/xlators/performance/quick-read/src/quick-read.h
+++ b/xlators/performance/quick-read/src/quick-read.h
@@ -52,21 +52,34 @@ struct qr_fd_ctx {
         int               wbflags;
         struct list_head  waiting_ops;
         gf_lock_t         lock;
+        struct list_head  inode_list;
+        struct list_head  tmp_list;
+        fd_t             *fd;
 };
 typedef struct qr_fd_ctx qr_fd_ctx_t;
 
 struct qr_local {
-        char         is_open;
-        char        *path;
-        char         just_validated;
-        fd_t        *fd;
-        int          open_flags;
-        int32_t      op_ret;
-        int32_t      op_errno;
-        call_stub_t *stub;
+        char              is_open;
+        char             *path;
+        char              just_validated;
+        fd_t             *fd;
+        int               open_flags;
+        int32_t           op_ret;
+        int32_t           op_errno;
+        uint32_t          open_count;
+        call_stub_t      *stub;
+        struct list_head  fd_list;
+        gf_lock_t         lock;
 };
 typedef struct qr_local qr_local_t;
 
+struct qr_dentry {
+        char             *name;
+        uuid_t            pargfid;
+        struct list_head  unlink_list;
+};
+typedef struct qr_dentry qr_dentry_t;
+
 struct qr_inode {
         dict_t           *xattr;
         inode_t          *inode;
@@ -74,6 +87,8 @@ struct qr_inode {
         struct iatt       stbuf;
         struct timeval    tv;
         struct list_head  lru;
+        struct list_head  fd_list;
+        struct list_head  unlinked_dentries;
 };
 typedef struct qr_inode qr_inode_t;
 
@@ -103,6 +118,7 @@ typedef struct qr_inode_table qr_inode_table_t;
 struct qr_private {
         qr_conf_t         conf;
         qr_inode_table_t  table;
+        struct mem_pool  *dentry_pool;
 };
 typedef struct qr_private qr_private_t;