summaryrefslogtreecommitdiffstats
path: root/xlators
diff options
context:
space:
mode:
authorRaghavendra G <raghavendra@gluster.com>2011-09-27 01:20:21 +0530
committerVijay Bellur <vijay@gluster.com>2012-03-28 10:40:41 -0700
commit243f02a5ef77d69fa91c18092d5da8578bb8f042 (patch)
tree077989a974e9d49ac8466bde4e13e2124b61e5e2 /xlators
parentbfb6366bd2a6de4cb5322dda72fbc1e6ad1edc41 (diff)
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>
Diffstat (limited to 'xlators')
-rw-r--r--xlators/performance/quick-read/src/quick-read-mem-types.h1
-rw-r--r--xlators/performance/quick-read/src/quick-read.c327
-rw-r--r--xlators/performance/quick-read/src/quick-read.h32
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;