diff options
Diffstat (limited to 'xlators/features/marker/src/marker-quota.c')
| -rw-r--r-- | xlators/features/marker/src/marker-quota.c | 1110 |
1 files changed, 701 insertions, 409 deletions
diff --git a/xlators/features/marker/src/marker-quota.c b/xlators/features/marker/src/marker-quota.c index 4978cd776..6f9af6e13 100644 --- a/xlators/features/marker/src/marker-quota.c +++ b/xlators/features/marker/src/marker-quota.c @@ -1,21 +1,12 @@ -/*Copyright (c) 2008-2010 Gluster, Inc. <http://www.gluster.com> - This file is part of GlusterFS. - - GlusterFS is free software; you can redistribute it and/or modify - it under the terms of the GNU General Public License as published - by the Free Software Foundation; either version 3 of the License, - or (at your option) any later version. - - GlusterFS is distributed in the hope that it will be useful, but - WITHOUT ANY WARRANTY; without even the implied warranty of - MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU - General Public License for more details. - - You should have received a copy of the GNU General Public License - along with this program. If not, see - <http://www.gnu.org/licenses/>. +/* + Copyright (c) 2008-2012 Red Hat, Inc. <http://www.redhat.com> + This file is part of GlusterFS. + + This file is licensed to you under your choice of the GNU Lesser + General Public License, version 3 or any later version (LGPLv3 or + later), or the GNU General Public License, version 2 (GPLv2), in all + cases as published by the Free Software Foundation. */ - #ifndef _CONFIG_H #define _CONFIG_H #include "config.h" @@ -30,6 +21,110 @@ #include "marker-quota.h" #include "marker-quota-helper.h" +int +mq_loc_copy (loc_t *dst, loc_t *src) +{ + int ret = -1; + + GF_VALIDATE_OR_GOTO ("marker", dst, out); + GF_VALIDATE_OR_GOTO ("marker", src, out); + + if (src->inode == NULL || + src->path == NULL) { + gf_log ("marker", GF_LOG_WARNING, + "src loc is not valid"); + goto out; + } + + ret = loc_copy (dst, src); +out: + return ret; +} + +int32_t +mq_get_local_err (quota_local_t *local, + int32_t *val) +{ + int32_t ret = -1; + + GF_VALIDATE_OR_GOTO ("marker", local, out); + GF_VALIDATE_OR_GOTO ("marker", val, out); + + LOCK (&local->lock); + { + *val = local->err; + } + UNLOCK (&local->lock); + + ret = 0; +out: + return ret; +} + +int32_t +mq_get_ctx_updation_status (quota_inode_ctx_t *ctx, + gf_boolean_t *status) +{ + int32_t ret = -1; + + GF_VALIDATE_OR_GOTO ("marker", ctx, out); + GF_VALIDATE_OR_GOTO ("marker", status, out); + + LOCK (&ctx->lock); + { + *status = ctx->updation_status; + } + UNLOCK (&ctx->lock); + + ret = 0; +out: + return ret; +} + + +int32_t +mq_set_ctx_updation_status (quota_inode_ctx_t *ctx, + gf_boolean_t status) +{ + int32_t ret = -1; + + if (ctx == NULL) + goto out; + + LOCK (&ctx->lock); + { + ctx->updation_status = status; + } + UNLOCK (&ctx->lock); + + ret = 0; +out: + return ret; +} + +int32_t +mq_test_and_set_ctx_updation_status (quota_inode_ctx_t *ctx, + gf_boolean_t *status) +{ + int32_t ret = -1; + gf_boolean_t temp = _gf_false; + + GF_VALIDATE_OR_GOTO ("marker", ctx, out); + GF_VALIDATE_OR_GOTO ("marker", status, out); + + LOCK (&ctx->lock); + { + temp = *status; + *status = ctx->updation_status; + ctx->updation_status = temp; + } + UNLOCK (&ctx->lock); + + ret = 0; +out: + return ret; +} + void mq_assign_lk_owner (xlator_t *this, call_frame_t *frame) { @@ -48,15 +143,15 @@ mq_assign_lk_owner (xlator_t *this, call_frame_t *frame) } UNLOCK (&conf->lock); - frame->root->lk_owner = lk_owner; + set_lk_owner_from_uint64 (&frame->root->lk_owner, lk_owner); return; } int32_t -loc_fill_from_name (xlator_t *this, loc_t *newloc, loc_t *oldloc, - uint64_t ino, char *name) +mq_loc_fill_from_name (xlator_t *this, loc_t *newloc, loc_t *oldloc, + uint64_t ino, char *name) { int32_t ret = -1; int32_t len = 0; @@ -67,8 +162,6 @@ loc_fill_from_name (xlator_t *this, loc_t *newloc, loc_t *oldloc, GF_VALIDATE_OR_GOTO ("marker", oldloc, out); GF_VALIDATE_OR_GOTO ("marker", name, out); - newloc->ino = ino; - newloc->inode = inode_new (oldloc->inode->table); if (!newloc->inode) { @@ -77,6 +170,7 @@ loc_fill_from_name (xlator_t *this, loc_t *newloc, loc_t *oldloc, } newloc->parent = inode_ref (oldloc->inode); + uuid_copy (newloc->pargfid, oldloc->inode->gfid); len = strlen (oldloc->path); @@ -104,8 +198,8 @@ out: } int32_t -dirty_inode_updation_done (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno) +mq_dirty_inode_updation_done (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { QUOTA_STACK_DESTROY (frame, this); @@ -113,18 +207,20 @@ dirty_inode_updation_done (call_frame_t *frame, void *cookie, xlator_t *this, } int32_t -release_lock_on_dirty_inode (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno) +mq_release_lock_on_dirty_inode (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { struct gf_flock lock = {0, }; quota_local_t *local = NULL; + loc_t loc = {0, }; + int ret = -1; local = frame->local; if (op_ret == -1) { local->err = -1; - dirty_inode_updation_done (frame, NULL, this, 0, 0); + mq_dirty_inode_updation_done (frame, NULL, this, 0, 0, NULL); return 0; } @@ -138,32 +234,50 @@ release_lock_on_dirty_inode (call_frame_t *frame, void *cookie, xlator_t *this, lock.l_len = 0; lock.l_pid = 0; + ret = loc_copy (&loc, &local->loc); + if (ret == -1) { + local->err = -1; + frame->local = NULL; + mq_dirty_inode_updation_done (frame, NULL, this, 0, 0, NULL); + return 0; + } + + if (local->loc.inode == NULL) { + gf_log (this->name, GF_LOG_WARNING, + "Inode is NULL, so can't stackwind."); + goto out; + } + STACK_WIND (frame, - dirty_inode_updation_done, + mq_dirty_inode_updation_done, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, - this->name, &local->loc, F_SETLKW, &lock); + this->name, &loc, F_SETLKW, &lock, NULL); + + loc_wipe (&loc); + + return 0; +out: + mq_dirty_inode_updation_done (frame, NULL, this, -1, 0, NULL); return 0; } int32_t -mark_inode_undirty (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno, dict_t *dict) +mq_mark_inode_undirty (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *dict, + dict_t *xdata) { int32_t ret = -1; int64_t *size = NULL; dict_t *newdict = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; local = (quota_local_t *) frame->local; if (op_ret == -1) goto err; - priv = (marker_conf_t *) this->private; - if (!dict) goto wind; @@ -186,17 +300,21 @@ wind: if (ret) goto err; - STACK_WIND (frame, release_lock_on_dirty_inode, + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, local->loc.inode->gfid); + + GF_UUID_ASSERT (local->loc.gfid); + STACK_WIND (frame, mq_release_lock_on_dirty_inode, FIRST_CHILD(this), FIRST_CHILD(this)->fops->setxattr, - &local->loc, newdict, 0); + &local->loc, newdict, 0, NULL); ret = 0; err: if (op_ret == -1 || ret == -1) { local->err = -1; - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); } if (newdict) @@ -206,24 +324,21 @@ err: } int32_t -update_size_xattr (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno, inode_t *inode, - struct iatt *buf, dict_t *dict, struct iatt *postparent) +mq_update_size_xattr (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, inode_t *inode, + struct iatt *buf, dict_t *dict, struct iatt *postparent) { int32_t ret = -1; dict_t *new_dict = NULL; int64_t *size = NULL; int64_t *delta = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; local = frame->local; if (op_ret == -1) goto err; - priv = this->private; - if (dict == NULL) { gf_log (this->name, GF_LOG_WARNING, "Dict is null while updating the size xattr %s", @@ -255,9 +370,14 @@ update_size_xattr (call_frame_t *frame, void *cookie, xlator_t *this, if (ret) goto err; - STACK_WIND (frame, mark_inode_undirty, FIRST_CHILD(this), + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, buf->ia_gfid); + + GF_UUID_ASSERT (local->loc.gfid); + + STACK_WIND (frame, mq_mark_inode_undirty, FIRST_CHILD(this), FIRST_CHILD(this)->fops->xattrop, &local->loc, - GF_XATTROP_ADD_ARRAY64, new_dict); + GF_XATTROP_ADD_ARRAY64, new_dict, NULL); ret = 0; @@ -265,7 +385,7 @@ err: if (op_ret == -1 || ret == -1) { local->err = -1; - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); } if (new_dict) @@ -275,17 +395,37 @@ err: } int32_t -get_dirty_inode_size (call_frame_t *frame, xlator_t *this) +mq_test_and_set_local_err(quota_local_t *local, + int32_t *val) +{ + int tmp = 0; + int32_t ret = -1; + + GF_VALIDATE_OR_GOTO ("marker", local, out); + GF_VALIDATE_OR_GOTO ("marker", val, out); + + LOCK (&local->lock); + { + tmp = local->err; + local->err = *val; + *val = tmp; + } + UNLOCK (&local->lock); + + ret = 0; +out: + return ret; +} + +int32_t +mq_get_dirty_inode_size (call_frame_t *frame, xlator_t *this) { int32_t ret = -1; dict_t *dict = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; local = (quota_local_t *) frame->local; - priv = (marker_conf_t *) this->private; - dict = dict_new (); if (!dict) { ret = -1; @@ -296,7 +436,12 @@ get_dirty_inode_size (call_frame_t *frame, xlator_t *this) if (ret) goto err; - STACK_WIND (frame, update_size_xattr, FIRST_CHILD(this), + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, local->loc.inode->gfid); + + GF_UUID_ASSERT (local->loc.gfid); + + STACK_WIND (frame, mq_update_size_xattr, FIRST_CHILD(this), FIRST_CHILD(this)->fops->lookup, &local->loc, dict); ret =0; @@ -304,7 +449,7 @@ err: if (ret) { local->err = -1; - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); } if (dict) @@ -314,15 +459,15 @@ err: } int32_t -get_child_contribution (call_frame_t *frame, - void *cookie, - xlator_t *this, - int32_t op_ret, - int32_t op_errno, - inode_t *inode, - struct iatt *buf, - dict_t *dict, - struct iatt *postparent) +mq_get_child_contribution (call_frame_t *frame, + void *cookie, + xlator_t *this, + int32_t op_ret, + int32_t op_errno, + inode_t *inode, + struct iatt *buf, + dict_t *dict, + struct iatt *postparent) { int32_t ret = -1; int32_t val = 0; @@ -337,17 +482,20 @@ get_child_contribution (call_frame_t *frame, QUOTA_STACK_DESTROY (frame, this); if (op_ret == -1) { - gf_log (this->name, GF_LOG_ERROR, "%s", strerror (op_errno)); - - local->err = -2; - - release_lock_on_dirty_inode (local->frame, NULL, this, 0, 0); + gf_log (this->name, GF_LOG_ERROR, "%s", + strerror (op_errno)); + val = -2; + if (!mq_test_and_set_local_err (local, &val) && + val != -2) + mq_release_lock_on_dirty_inode (local->frame, NULL, + this, 0, 0, NULL); - goto out; + goto exit; } - if (local->err) - goto out; + ret = mq_get_local_err (local, &val); + if (!ret && val == -2) + goto exit; GET_CONTRI_KEY (contri_key, local->loc.inode->gfid, ret); if (ret < 0) @@ -366,29 +514,29 @@ out: } UNLOCK (&local->lock); - if (val== 0) { - if (local->err) { - QUOTA_SAFE_DECREMENT (&local->lock, local->ref, val); - - quota_local_unref (this, local); - } else - quota_dirty_inode_readdir (local->frame, NULL, this, - 0, 0, NULL); + if (val == 0) { + mq_dirty_inode_readdir (local->frame, NULL, this, + 0, 0, NULL, NULL); } + mq_local_unref (this, local); return 0; +exit: + mq_local_unref (this, local); + return 0; } int32_t -quota_readdir_cbk (call_frame_t *frame, - void *cookie, - xlator_t *this, - int32_t op_ret, - int32_t op_errno, - gf_dirent_t *entries) +mq_readdir_cbk (call_frame_t *frame, + void *cookie, + xlator_t *this, + int32_t op_ret, + int32_t op_errno, + gf_dirent_t *entries, dict_t *xdata) { char contri_key [512] = {0, }; int32_t ret = 0; + int32_t val = 0; off_t offset = 0; int32_t count = 0; dict_t *dict = NULL; @@ -397,20 +545,20 @@ quota_readdir_cbk (call_frame_t *frame, call_frame_t *newframe = NULL; loc_t loc = {0, }; - local = frame->local; + local = mq_local_ref (frame->local); if (op_ret == -1) { gf_log (this->name, GF_LOG_DEBUG, "readdir failed %s", strerror (op_errno)); local->err = -1; - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); - return 0; + goto end; } else if (op_ret == 0) { - get_dirty_inode_size (frame, this); + mq_get_dirty_inode_size (frame, this); - return 0; + goto end; } local->dentry_child_count = 0; @@ -429,16 +577,20 @@ quota_readdir_cbk (call_frame_t *frame, count++; } + if (count == 0) { + mq_get_dirty_inode_size (frame, this); + goto end; + + } + local->frame = frame; - if (count > 0) { - LOCK (&local->lock); - { - local->dentry_child_count = count; - local->d_off = offset; - } - UNLOCK (&local->lock); + LOCK (&local->lock); + { + local->dentry_child_count = count; + local->d_off = offset; } + UNLOCK (&local->lock); list_for_each_entry (entry, (&entries->list), list) { @@ -451,18 +603,29 @@ quota_readdir_cbk (call_frame_t *frame, continue; } - ret = loc_fill_from_name (this, &loc, &local->loc, - entry->d_ino, entry->d_name); + ret = mq_loc_fill_from_name (this, &loc, &local->loc, + entry->d_ino, entry->d_name); if (ret < 0) goto out; - newframe = copy_frame (frame); - if (!newframe) { - ret = -1; - goto out; + ret = 0; + + LOCK (&local->lock); + { + if (local->err != -2) { + newframe = copy_frame (frame); + if (!newframe) { + ret = -1; + } + } else + ret = -1; } + UNLOCK (&local->lock); + + if (ret == -1) + goto out; - newframe->local = local; + newframe->local = mq_local_ref (local); dict = dict_new (); if (!dict) { @@ -479,7 +642,7 @@ quota_readdir_cbk (call_frame_t *frame, goto out; STACK_WIND (newframe, - get_child_contribution, + mq_get_child_contribution, FIRST_CHILD(this), FIRST_CHILD(this)->fops->lookup, &loc, dict); @@ -488,6 +651,8 @@ quota_readdir_cbk (call_frame_t *frame, loc_wipe (&loc); + newframe = NULL; + out: if (dict) { dict_unref (dict); @@ -495,18 +660,12 @@ quota_readdir_cbk (call_frame_t *frame, } if (ret) { - LOCK (&local->lock); - { - if (local->dentry_child_count == 0) - local->err = -1; - else - local->err = -2; - } - UNLOCK (&local->lock); + val = -2; + mq_test_and_set_local_err (local, &val); if (newframe) { newframe->local = NULL; - + mq_local_unref(this, local); QUOTA_STACK_DESTROY (newframe, this); } @@ -514,22 +673,22 @@ quota_readdir_cbk (call_frame_t *frame, } } - if (ret) { - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); - } else if (count == 0 ) { - get_dirty_inode_size (frame, this); + if (ret && val != -2) { + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); } +end: + mq_local_unref (this, local); return 0; } int32_t -quota_dirty_inode_readdir (call_frame_t *frame, - void *cookie, - xlator_t *this, - int32_t op_ret, - int32_t op_errno, - fd_t *fd) +mq_dirty_inode_readdir (call_frame_t *frame, + void *cookie, + xlator_t *this, + int32_t op_ret, + int32_t op_errno, + fd_t *fd, dict_t *xdata) { quota_local_t *local = NULL; @@ -537,7 +696,7 @@ quota_dirty_inode_readdir (call_frame_t *frame, if (op_ret == -1) { local->err = -1; - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); return 0; } @@ -545,30 +704,29 @@ quota_dirty_inode_readdir (call_frame_t *frame, local->fd = fd_ref (fd); STACK_WIND (frame, - quota_readdir_cbk, + mq_readdir_cbk, FIRST_CHILD(this), FIRST_CHILD(this)->fops->readdir, - local->fd, READDIR_BUF, local->d_off); + local->fd, READDIR_BUF, local->d_off, xdata); return 0; } int32_t -check_if_still_dirty (call_frame_t *frame, - void *cookie, - xlator_t *this, - int32_t op_ret, - int32_t op_errno, - inode_t *inode, - struct iatt *buf, - dict_t *dict, - struct iatt *postparent) +mq_check_if_still_dirty (call_frame_t *frame, + void *cookie, + xlator_t *this, + int32_t op_ret, + int32_t op_errno, + inode_t *inode, + struct iatt *buf, + dict_t *dict, + struct iatt *postparent) { int8_t dirty = -1; int32_t ret = -1; fd_t *fd = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; local = frame->local; @@ -578,8 +736,6 @@ check_if_still_dirty (call_frame_t *frame, goto err; } - priv = this->private; - if (!dict) { ret = -1; goto err; @@ -591,7 +747,7 @@ check_if_still_dirty (call_frame_t *frame, //the inode is not dirty anymore if (dirty == 0) { - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); return 0; } @@ -600,18 +756,22 @@ check_if_still_dirty (call_frame_t *frame, local->d_off = 0; + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, buf->ia_gfid); + + GF_UUID_ASSERT (local->loc.gfid); STACK_WIND(frame, - quota_dirty_inode_readdir, + mq_dirty_inode_readdir, FIRST_CHILD(this), FIRST_CHILD(this)->fops->opendir, - &local->loc, fd); + &local->loc, fd, NULL); ret = 0; err: if (op_ret == -1 || ret == -1) { local->err = -1; - release_lock_on_dirty_inode (frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode (frame, NULL, this, 0, 0, NULL); } if (fd != NULL) { @@ -622,21 +782,18 @@ err: } int32_t -get_dirty_xattr (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, int32_t op_errno) +mq_get_dirty_xattr (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { int32_t ret = -1; dict_t *xattr_req = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; if (op_ret == -1) { - dirty_inode_updation_done (frame, NULL, this, 0, 0); + mq_dirty_inode_updation_done (frame, NULL, this, 0, 0, NULL); return 0; } - priv = (marker_conf_t *) this->private; - local = frame->local; xattr_req = dict_new (); @@ -649,8 +806,13 @@ get_dirty_xattr (call_frame_t *frame, void *cookie, if (ret) goto err; + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, local->loc.inode->gfid); + + GF_UUID_ASSERT (local->loc.gfid); + STACK_WIND (frame, - check_if_still_dirty, + mq_check_if_still_dirty, FIRST_CHILD(this), FIRST_CHILD(this)->fops->lookup, &local->loc, @@ -660,7 +822,7 @@ get_dirty_xattr (call_frame_t *frame, void *cookie, err: if (ret) { local->err = -1; - release_lock_on_dirty_inode(frame, NULL, this, 0, 0); + mq_release_lock_on_dirty_inode(frame, NULL, this, 0, 0, NULL); } if (xattr_req) @@ -669,17 +831,27 @@ err: return 0; } +/* return 1 when dirty updation started + * 0 other wise + */ int32_t -update_dirty_inode (xlator_t *this, - loc_t *loc, - quota_inode_ctx_t *ctx, - inode_contribution_t *contribution) +mq_update_dirty_inode (xlator_t *this, + loc_t *loc, + quota_inode_ctx_t *ctx, + inode_contribution_t *contribution) { int32_t ret = -1; quota_local_t *local = NULL; + gf_boolean_t status = _gf_false; struct gf_flock lock = {0, }; call_frame_t *frame = NULL; + ret = mq_get_ctx_updation_status (ctx, &status); + if (ret == -1 || status == _gf_true) { + ret = 0; + goto out; + } + frame = create_frame (this, this->ctx->pool); if (frame == NULL) { ret = -1; @@ -688,13 +860,12 @@ update_dirty_inode (xlator_t *this, mq_assign_lk_owner (this, frame); - local = quota_local_new (); + local = mq_local_new (); if (local == NULL) goto fr_destroy; frame->local = local; - - ret = loc_copy (&local->loc, loc); + ret = mq_loc_copy (&local->loc, loc); if (ret < 0) goto fr_destroy; @@ -707,12 +878,19 @@ update_dirty_inode (xlator_t *this, lock.l_start = 0; lock.l_len = 0; + if (local->loc.inode == NULL) { + ret = -1; + gf_log (this->name, GF_LOG_WARNING, + "Inode is NULL, so can't stackwind."); + goto fr_destroy; + } + STACK_WIND (frame, - get_dirty_xattr, + mq_get_dirty_xattr, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, - this->name, &local->loc, F_SETLKW, &lock); - return 0; + this->name, &local->loc, F_SETLKW, &lock, NULL); + return 1; fr_destroy: QUOTA_STACK_DESTROY (frame, this); @@ -723,12 +901,20 @@ out: int32_t -quota_inode_creation_done (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno) +mq_inode_creation_done (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { + quota_local_t *local = NULL; + if (frame == NULL) return 0; + local = frame->local; + + if (local != NULL) { + mq_initiate_quota_txn (this, &local->loc); + } + QUOTA_STACK_DESTROY (frame, this); return 0; @@ -736,9 +922,9 @@ quota_inode_creation_done (call_frame_t *frame, void *cookie, xlator_t *this, int32_t -quota_xattr_creation_release_lock (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, - int32_t op_errno) +mq_xattr_creation_release_lock (call_frame_t *frame, void *cookie, + xlator_t *this, int32_t op_ret, + int32_t op_errno, dict_t *xdata) { struct gf_flock lock = {0, }; quota_local_t *local = NULL; @@ -752,24 +938,24 @@ quota_xattr_creation_release_lock (call_frame_t *frame, void *cookie, lock.l_pid = 0; STACK_WIND (frame, - quota_inode_creation_done, + mq_inode_creation_done, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, this->name, &local->loc, - F_SETLKW, &lock); + F_SETLKW, &lock, NULL); return 0; } int32_t -create_dirty_xattr (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno, dict_t *dict) +mq_create_dirty_xattr (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *dict, + dict_t *xdata) { int32_t ret = -1; dict_t *newdict = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; if (op_ret < 0) { goto err; @@ -777,8 +963,6 @@ create_dirty_xattr (call_frame_t *frame, void *cookie, xlator_t *this, local = frame->local; - priv = (marker_conf_t *) this->private; - if (local->loc.inode->ia_type == IA_IFDIR) { newdict = dict_new (); if (!newdict) { @@ -790,19 +974,22 @@ create_dirty_xattr (call_frame_t *frame, void *cookie, xlator_t *this, goto err; } - STACK_WIND (frame, quota_xattr_creation_release_lock, + uuid_copy (local->loc.gfid, local->loc.inode->gfid); + GF_UUID_ASSERT (local->loc.gfid); + + STACK_WIND (frame, mq_xattr_creation_release_lock, FIRST_CHILD(this), FIRST_CHILD(this)->fops->setxattr, - &local->loc, newdict, 0); + &local->loc, newdict, 0, NULL); } else { - quota_xattr_creation_release_lock (frame, NULL, this, 0, 0); + mq_xattr_creation_release_lock (frame, NULL, this, 0, 0, NULL); } ret = 0; err: if (ret < 0) { - quota_xattr_creation_release_lock (frame, NULL, this, 0, 0); + mq_xattr_creation_release_lock (frame, NULL, this, 0, 0, NULL); } if (newdict != NULL) @@ -813,7 +1000,7 @@ err: int32_t -quota_create_xattr (xlator_t *this, call_frame_t *frame) +mq_create_xattr (xlator_t *this, call_frame_t *frame) { int32_t ret = 0; int64_t *value = NULL; @@ -821,7 +1008,6 @@ quota_create_xattr (xlator_t *this, call_frame_t *frame) dict_t *dict = NULL; char key[512] = {0, }; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; quota_inode_ctx_t *ctx = NULL; inode_contribution_t *contri = NULL; @@ -830,14 +1016,12 @@ quota_create_xattr (xlator_t *this, call_frame_t *frame) local = frame->local; - priv = (marker_conf_t *) this->private; - - ret = quota_inode_ctx_get (local->loc.inode, this, &ctx); + ret = mq_inode_ctx_get (local->loc.inode, this, &ctx); if (ret < 0) { - ctx = quota_inode_ctx_new (local->loc.inode, this); + ctx = mq_inode_ctx_new (local->loc.inode, this); if (ctx == NULL) { gf_log (this->name, GF_LOG_WARNING, - "quota_inode_ctx_new failed"); + "mq_inode_ctx_new failed"); ret = -1; goto out; } @@ -855,7 +1039,7 @@ quota_create_xattr (xlator_t *this, call_frame_t *frame) } if (strcmp (local->loc.path, "/") != 0) { - contri = add_new_contribution_node (this, ctx, &local->loc); + contri = mq_add_new_contribution_node (this, ctx, &local->loc); if (contri == NULL) goto err; @@ -867,9 +1051,11 @@ quota_create_xattr (xlator_t *this, call_frame_t *frame) goto free_value; } - STACK_WIND (frame, create_dirty_xattr, FIRST_CHILD(this), + GF_UUID_ASSERT (local->loc.gfid); + + STACK_WIND (frame, mq_create_dirty_xattr, FIRST_CHILD(this), FIRST_CHILD(this)->fops->xattrop, &local->loc, - GF_XATTROP_ADD_ARRAY64, dict); + GF_XATTROP_ADD_ARRAY64, dict, NULL); ret = 0; free_size: @@ -887,7 +1073,7 @@ err: out: if (ret < 0) { - quota_xattr_creation_release_lock (frame, NULL, this, 0, 0); + mq_xattr_creation_release_lock (frame, NULL, this, 0, 0, NULL); } return 0; @@ -895,15 +1081,14 @@ out: int32_t -quota_check_n_set_inode_xattr (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, int32_t op_errno, - inode_t *inode, struct iatt *buf, dict_t *dict, - struct iatt *postparent) +mq_check_n_set_inode_xattr (call_frame_t *frame, void *cookie, + xlator_t *this, int32_t op_ret, int32_t op_errno, + inode_t *inode, struct iatt *buf, dict_t *dict, + struct iatt *postparent) { quota_local_t *local = NULL; int64_t *size = NULL, *contri = NULL; int8_t dirty = 0; - marker_conf_t *priv = NULL; int32_t ret = 0; char contri_key[512] = {0, }; @@ -912,7 +1097,6 @@ quota_check_n_set_inode_xattr (call_frame_t *frame, void *cookie, } local = frame->local; - priv = this->private; ret = dict_get_bin (dict, QUOTA_SIZE_KEY, (void **) &size); if (ret < 0) @@ -934,18 +1118,22 @@ quota_check_n_set_inode_xattr (call_frame_t *frame, void *cookie, } out: - quota_xattr_creation_release_lock (frame, NULL, this, 0, 0); + mq_xattr_creation_release_lock (frame, NULL, this, 0, 0, NULL); return 0; create_xattr: - quota_create_xattr (this, frame); + if (uuid_is_null (local->loc.gfid)) { + uuid_copy (local->loc.gfid, buf->ia_gfid); + } + + mq_create_xattr (this, frame); return 0; } int32_t -quota_get_xattr (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno) +mq_get_xattr (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { dict_t *xattr_req = NULL; quota_local_t *local = NULL; @@ -962,13 +1150,18 @@ quota_get_xattr (call_frame_t *frame, void *cookie, xlator_t *this, goto err; } - ret = quota_req_xattr (this, &local->loc, xattr_req); + ret = mq_req_xattr (this, &local->loc, xattr_req); if (ret < 0) { gf_log (this->name, GF_LOG_WARNING, "cannot request xattr"); goto err; } - STACK_WIND (frame, quota_check_n_set_inode_xattr, FIRST_CHILD(this), + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, local->loc.inode->gfid); + + GF_UUID_ASSERT (local->loc.gfid); + + STACK_WIND (frame, mq_check_n_set_inode_xattr, FIRST_CHILD(this), FIRST_CHILD(this)->fops->lookup, &local->loc, xattr_req); dict_unref (xattr_req); @@ -976,20 +1169,20 @@ quota_get_xattr (call_frame_t *frame, void *cookie, xlator_t *this, return 0; err: - quota_xattr_creation_release_lock (frame, NULL, this, 0, 0); + mq_xattr_creation_release_lock (frame, NULL, this, 0, 0, NULL); if (xattr_req) dict_unref (xattr_req); return 0; lock_err: - quota_inode_creation_done (frame, NULL, this, 0, 0); + mq_inode_creation_done (frame, NULL, this, 0, 0, NULL); return 0; } int32_t -quota_set_inode_xattr (xlator_t *this, loc_t *loc) +mq_set_inode_xattr (xlator_t *this, loc_t *loc) { struct gf_flock lock = {0, }; quota_local_t *local = NULL; @@ -1002,7 +1195,7 @@ quota_set_inode_xattr (xlator_t *this, loc_t *loc) goto err; } - local = quota_local_new (); + local = mq_local_new (); if (local == NULL) { goto err; } @@ -1022,10 +1215,10 @@ quota_set_inode_xattr (xlator_t *this, loc_t *loc) lock.l_whence = SEEK_SET; STACK_WIND (frame, - quota_get_xattr, + mq_get_xattr, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, - this->name, &local->loc, F_SETLKW, &lock); + this->name, &local->loc, F_SETLKW, &lock, NULL); return 0; @@ -1037,38 +1230,68 @@ err: int32_t -get_parent_inode_local (xlator_t *this, quota_local_t *local) +mq_get_parent_inode_local (xlator_t *this, quota_local_t *local) { - int32_t ret; + int32_t ret = -1; quota_inode_ctx_t *ctx = NULL; + GF_VALIDATE_OR_GOTO ("marker", this, out); + GF_VALIDATE_OR_GOTO ("marker", local, out); + + local->contri = NULL; + loc_wipe (&local->loc); - loc_copy (&local->loc, &local->parent_loc); + ret = mq_loc_copy (&local->loc, &local->parent_loc); + if (ret < 0) { + gf_log_callingfn (this->name, GF_LOG_WARNING, + "loc copy failed"); + goto out; + } loc_wipe (&local->parent_loc); - quota_inode_loc_fill (NULL, local->loc.parent, &local->parent_loc); + ret = mq_inode_loc_fill (NULL, local->loc.parent, + &local->parent_loc); + if (ret < 0) { + gf_log_callingfn (this->name, GF_LOG_WARNING, + "failed to build parent loc of %s", + local->loc.path); + goto out; + } - ret = quota_inode_ctx_get (local->loc.inode, this, &ctx); - if (ret < 0) - return -1; + ret = mq_inode_ctx_get (local->loc.inode, this, &ctx); + if (ret < 0) { + gf_log_callingfn (this->name, GF_LOG_WARNING, + "inode ctx get failed"); + goto out; + } local->ctx = ctx; + if (list_empty (&ctx->contribution_head)) { + gf_log_callingfn (this->name, GF_LOG_WARNING, + "contribution node list is empty which " + "is an error"); + ret = -1; + goto out; + } + local->contri = (inode_contribution_t *) ctx->contribution_head.next; - return 0; + ret = 0; +out: + return ret; } int32_t -xattr_updation_done (call_frame_t *frame, - void *cookie, - xlator_t *this, - int32_t op_ret, - int32_t op_errno, - dict_t *dict) +mq_xattr_updation_done (call_frame_t *frame, + void *cookie, + xlator_t *this, + int32_t op_ret, + int32_t op_errno, + dict_t *dict, dict_t *xdata) { QUOTA_STACK_DESTROY (frame, this); return 0; @@ -1076,11 +1299,12 @@ xattr_updation_done (call_frame_t *frame, int32_t -quota_inodelk_cbk (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, int32_t op_errno) +mq_inodelk_cbk (call_frame_t *frame, void *cookie, + xlator_t *this, int32_t op_ret, int32_t op_errno, dict_t *xdata) { - int32_t ret = 0; - quota_local_t *local = NULL; + int32_t ret = 0; + gf_boolean_t status = _gf_false; + quota_local_t *local = NULL; local = frame->local; @@ -1090,8 +1314,7 @@ quota_inodelk_cbk (call_frame_t *frame, void *cookie, "unlocking failed on path (%s)(%s)", local->parent_loc.path, strerror (op_errno)); } - - xattr_updation_done (frame, NULL, this, 0, 0, NULL); + mq_xattr_updation_done (frame, NULL, this, 0, 0, NULL, NULL); return 0; } @@ -1101,15 +1324,23 @@ quota_inodelk_cbk (call_frame_t *frame, void *cookie, if ((strcmp (local->parent_loc.path, "/") == 0) || (local->delta == 0)) { - xattr_updation_done (frame, NULL, this, 0, 0, NULL); + mq_xattr_updation_done (frame, NULL, this, 0, 0, NULL, NULL); } else { - ret = get_parent_inode_local (this, local); + ret = mq_get_parent_inode_local (this, local); if (ret < 0) { - xattr_updation_done (frame, NULL, this, 0, 0, NULL); + mq_xattr_updation_done (frame, NULL, this, 0, 0, NULL, + NULL); goto out; } - - get_lock_on_parent (frame, this); + status = _gf_true; + + ret = mq_test_and_set_ctx_updation_status (local->ctx, &status); + if (ret == 0 && status == _gf_false) { + mq_get_lock_on_parent (frame, this); + } else { + mq_xattr_updation_done (frame, NULL, this, 0, 0, NULL, + NULL); + } } out: return 0; @@ -1118,9 +1349,9 @@ out: //now release lock on the parent inode int32_t -quota_release_parent_lock (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, - int32_t op_errno) +mq_release_parent_lock (call_frame_t *frame, void *cookie, + xlator_t *this, int32_t op_ret, + int32_t op_errno, dict_t *xdata) { int32_t ret = 0; quota_local_t *local = NULL; @@ -1138,7 +1369,7 @@ quota_release_parent_lock (call_frame_t *frame, void *cookie, strerror (local->err)); } - ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx); + ret = mq_inode_ctx_get (local->parent_loc.inode, this, &ctx); if (ret < 0) goto wind; @@ -1148,6 +1379,12 @@ quota_release_parent_lock (call_frame_t *frame, void *cookie, } UNLOCK (&ctx->lock); + if (local->parent_loc.inode == NULL) { + gf_log (this->name, GF_LOG_DEBUG, + "Invalid parent inode."); + goto err; + } + wind: lock.l_type = F_UNLCK; lock.l_whence = SEEK_SET; @@ -1156,30 +1393,33 @@ wind: lock.l_pid = 0; STACK_WIND (frame, - quota_inodelk_cbk, + mq_inodelk_cbk, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, this->name, &local->parent_loc, - F_SETLKW, &lock); + F_SETLKW, &lock, NULL); return 0; +err: + mq_xattr_updation_done (frame, NULL, this, + 0, 0 , NULL, NULL); + return 0; } int32_t -quota_mark_undirty (call_frame_t *frame, - void *cookie, - xlator_t *this, - int32_t op_ret, - int32_t op_errno, - dict_t *dict) +mq_mark_undirty (call_frame_t *frame, + void *cookie, + xlator_t *this, + int32_t op_ret, + int32_t op_errno, + dict_t *dict, dict_t *xdata) { int32_t ret = -1; int64_t *size = NULL; dict_t *newdict = NULL; quota_local_t *local = NULL; quota_inode_ctx_t *ctx = NULL; - marker_conf_t *priv = NULL; local = frame->local; @@ -1191,11 +1431,9 @@ quota_mark_undirty (call_frame_t *frame, goto err; } - priv = this->private; - //update the size of the parent inode if (dict != NULL) { - ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx); + ret = mq_inode_ctx_get (local->parent_loc.inode, this, &ctx); if (ret < 0) { op_errno = EINVAL; goto err; @@ -1230,17 +1468,20 @@ quota_mark_undirty (call_frame_t *frame, goto err; } - STACK_WIND (frame, quota_release_parent_lock, + uuid_copy (local->parent_loc.gfid, local->parent_loc.inode->gfid); + GF_UUID_ASSERT (local->parent_loc.gfid); + + STACK_WIND (frame, mq_release_parent_lock, FIRST_CHILD(this), FIRST_CHILD(this)->fops->setxattr, - &local->parent_loc, newdict, 0); + &local->parent_loc, newdict, 0, NULL); ret = 0; err: if (op_ret == -1 || ret == -1) { local->err = op_errno; - quota_release_parent_lock (frame, NULL, this, 0, 0); + mq_release_parent_lock (frame, NULL, this, 0, 0, NULL); } if (newdict) @@ -1251,17 +1492,16 @@ err: int32_t -quota_update_parent_size (call_frame_t *frame, - void *cookie, - xlator_t *this, - int32_t op_ret, - int32_t op_errno, - dict_t *dict) +mq_update_parent_size (call_frame_t *frame, + void *cookie, + xlator_t *this, + int32_t op_ret, + int32_t op_errno, + dict_t *dict, dict_t *xdata) { int64_t *size = NULL; int32_t ret = -1; dict_t *newdict = NULL; - marker_conf_t *priv = NULL; quota_local_t *local = NULL; quota_inode_ctx_t *ctx = NULL; @@ -1285,14 +1525,12 @@ quota_update_parent_size (call_frame_t *frame, local->loc.path, local->ctx->size, local->contri->contribution); - priv = this->private; - if (dict == NULL) { op_errno = EINVAL; goto err; } - ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx); + ret = mq_inode_ctx_get (local->parent_loc.inode, this, &ctx); if (ret < 0) { op_errno = EINVAL; goto err; @@ -1315,18 +1553,23 @@ quota_update_parent_size (call_frame_t *frame, goto err; } + if (uuid_is_null (local->parent_loc.gfid)) + uuid_copy (local->parent_loc.gfid, + local->parent_loc.inode->gfid); + GF_UUID_ASSERT (local->parent_loc.gfid); + STACK_WIND (frame, - quota_mark_undirty, + mq_mark_undirty, FIRST_CHILD(this), FIRST_CHILD(this)->fops->xattrop, &local->parent_loc, GF_XATTROP_ADD_ARRAY64, - newdict); + newdict, NULL); ret = 0; err: if (op_ret == -1 || ret < 0) { local->err = op_errno; - quota_release_parent_lock (frame, NULL, this, 0, 0); + mq_release_parent_lock (frame, NULL, this, 0, 0, NULL); } if (newdict) @@ -1336,11 +1579,11 @@ err: } int32_t -quota_update_inode_contribution (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, - int32_t op_errno, inode_t *inode, - struct iatt *buf, dict_t *dict, - struct iatt *postparent) +mq_update_inode_contribution (call_frame_t *frame, void *cookie, + xlator_t *this, int32_t op_ret, + int32_t op_errno, inode_t *inode, + struct iatt *buf, dict_t *dict, + struct iatt *postparent) { int32_t ret = -1; int64_t *size = NULL, size_int = 0, contri_int = 0; @@ -1350,7 +1593,6 @@ quota_update_inode_contribution (call_frame_t *frame, void *cookie, dict_t *newdict = NULL; quota_local_t *local = NULL; quota_inode_ctx_t *ctx = NULL; - marker_conf_t *priv = NULL; inode_contribution_t *contribution = NULL; local = frame->local; @@ -1363,8 +1605,6 @@ quota_update_inode_contribution (call_frame_t *frame, void *cookie, goto err; } - priv = this->private; - ctx = local->ctx; contribution = local->contri; @@ -1417,7 +1657,7 @@ unlock: local->delta = size_int - contri_int; if (local->delta == 0) { - quota_mark_undirty (frame, NULL, this, 0, 0, NULL); + mq_mark_undirty (frame, NULL, this, 0, 0, NULL, NULL); return 0; } @@ -1439,20 +1679,25 @@ unlock: goto err; } + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, buf->ia_gfid); + + GF_UUID_ASSERT (local->loc.gfid); + STACK_WIND (frame, - quota_update_parent_size, + mq_update_parent_size, FIRST_CHILD(this), FIRST_CHILD(this)->fops->xattrop, &local->loc, GF_XATTROP_ADD_ARRAY64, - newdict); + newdict, NULL); ret = 0; err: if (op_ret == -1 || ret < 0) { local->err = op_errno; - quota_release_parent_lock (frame, NULL, this, 0, 0); + mq_release_parent_lock (frame, NULL, this, 0, 0, NULL); } if (newdict) @@ -1462,15 +1707,14 @@ err: } int32_t -quota_fetch_child_size_and_contri (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, - int32_t op_errno) +mq_fetch_child_size_and_contri (call_frame_t *frame, void *cookie, + xlator_t *this, int32_t op_ret, + int32_t op_errno, dict_t *xdata) { int32_t ret = -1; char contri_key [512] = {0, }; dict_t *newdict = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; quota_inode_ctx_t *ctx = NULL; local = frame->local; @@ -1488,10 +1732,8 @@ quota_fetch_child_size_and_contri (call_frame_t *frame, void *cookie, gf_log (this->name, GF_LOG_DEBUG, "%s marked dirty", local->parent_loc.path); - priv = this->private; - //update parent ctx - ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx); + ret = mq_inode_ctx_get (local->parent_loc.inode, this, &ctx); if (ret == -1) { op_errno = EINVAL; goto err; @@ -1511,6 +1753,11 @@ quota_fetch_child_size_and_contri (call_frame_t *frame, void *cookie, if (local->loc.inode->ia_type == IA_IFDIR) { ret = dict_set_int64 (newdict, QUOTA_SIZE_KEY, 0); + if (ret < 0) { + gf_log (this->name, GF_LOG_WARNING, + "dict_set failed."); + goto err; + } } GET_CONTRI_KEY (contri_key, local->contri->gfid, ret); @@ -1520,17 +1767,31 @@ quota_fetch_child_size_and_contri (call_frame_t *frame, void *cookie, } ret = dict_set_int64 (newdict, contri_key, 0); + if (ret < 0) { + gf_log (this->name, GF_LOG_WARNING, + "dict_set failed."); + goto err; + } - STACK_WIND (frame, quota_update_inode_contribution, FIRST_CHILD(this), + mq_set_ctx_updation_status (local->ctx, _gf_false); + + if (uuid_is_null (local->loc.gfid)) + uuid_copy (local->loc.gfid, local->loc.inode->gfid); + + GF_UUID_ASSERT (local->loc.gfid); + + STACK_WIND (frame, mq_update_inode_contribution, FIRST_CHILD(this), FIRST_CHILD(this)->fops->lookup, &local->loc, newdict); ret = 0; err: - if ((op_ret == -1) || (ret == -1)) { + if ((op_ret == -1) || (ret < 0)) { local->err = op_errno; - quota_release_parent_lock (frame, NULL, this, 0, 0); + mq_set_ctx_updation_status (local->ctx, _gf_false); + + mq_release_parent_lock (frame, NULL, this, 0, 0, NULL); } if (newdict) @@ -1540,13 +1801,12 @@ err: } int32_t -quota_markdirty (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, int32_t op_errno) +mq_markdirty (call_frame_t *frame, void *cookie, + xlator_t *this, int32_t op_ret, int32_t op_errno, dict_t *xdata) { int32_t ret = -1; dict_t *dict = NULL; quota_local_t *local = NULL; - marker_conf_t *priv = NULL; local = frame->local; @@ -1557,7 +1817,9 @@ quota_markdirty (call_frame_t *frame, void *cookie, local->err = op_errno; - quota_inodelk_cbk (frame, NULL, this, 0, 0); + mq_set_ctx_updation_status (local->ctx, _gf_false); + + mq_inodelk_cbk (frame, NULL, this, 0, 0, NULL); return 0; } @@ -1565,8 +1827,6 @@ quota_markdirty (call_frame_t *frame, void *cookie, gf_log (this->name, GF_LOG_TRACE, "inodelk succeeded on %s", local->parent_loc.path); - priv = this->private; - dict = dict_new (); if (!dict) { ret = -1; @@ -1577,17 +1837,23 @@ quota_markdirty (call_frame_t *frame, void *cookie, if (ret == -1) goto err; - STACK_WIND (frame, quota_fetch_child_size_and_contri, + uuid_copy (local->parent_loc.gfid, + local->parent_loc.inode->gfid); + GF_UUID_ASSERT (local->parent_loc.gfid); + + STACK_WIND (frame, mq_fetch_child_size_and_contri, FIRST_CHILD(this), FIRST_CHILD(this)->fops->setxattr, - &local->parent_loc, dict, 0); + &local->parent_loc, dict, 0, NULL); ret = 0; err: if (ret == -1) { local->err = 1; - quota_release_parent_lock (frame, NULL, this, 0, 0); + mq_set_ctx_updation_status (local->ctx, _gf_false); + + mq_release_parent_lock (frame, NULL, this, 0, 0, NULL); } if (dict) @@ -1598,7 +1864,7 @@ err: int32_t -get_lock_on_parent (call_frame_t *frame, xlator_t *this) +mq_get_lock_on_parent (call_frame_t *frame, xlator_t *this) { struct gf_flock lock = {0, }; quota_local_t *local = NULL; @@ -1609,30 +1875,37 @@ get_lock_on_parent (call_frame_t *frame, xlator_t *this) gf_log (this->name, GF_LOG_DEBUG, "taking lock on %s", local->parent_loc.path); + if (local->parent_loc.inode == NULL) { + gf_log (this->name, GF_LOG_DEBUG, + "parent inode is not valid, aborting " + "transaction."); + goto fr_destroy; + } + lock.l_len = 0; lock.l_start = 0; lock.l_type = F_WRLCK; lock.l_whence = SEEK_SET; STACK_WIND (frame, - quota_markdirty, + mq_markdirty, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, - this->name, &local->parent_loc, F_SETLKW, &lock); + this->name, &local->parent_loc, F_SETLKW, &lock, NULL); return 0; fr_destroy: QUOTA_STACK_DESTROY (frame, this); - return 0; + return -1; } int -start_quota_txn (xlator_t *this, loc_t *loc, - quota_inode_ctx_t *ctx, - inode_contribution_t *contri) +mq_start_quota_txn (xlator_t *this, loc_t *loc, + quota_inode_ctx_t *ctx, + inode_contribution_t *contri) { int32_t ret = -1; call_frame_t *frame = NULL; @@ -1644,46 +1917,52 @@ start_quota_txn (xlator_t *this, loc_t *loc, mq_assign_lk_owner (this, frame); - local = quota_local_new (); + local = mq_local_new (); if (local == NULL) goto fr_destroy; frame->local = local; - ret = loc_copy (&local->loc, loc); + ret = mq_loc_copy (&local->loc, loc); if (ret < 0) goto fr_destroy; - ret = quota_inode_loc_fill (NULL, local->loc.parent, - &local->parent_loc); + ret = mq_inode_loc_fill (NULL, local->loc.parent, + &local->parent_loc); if (ret < 0) goto fr_destroy; local->ctx = ctx; local->contri = contri; - get_lock_on_parent (frame, this); + ret = mq_get_lock_on_parent (frame, this); + if (ret == -1) + goto err; return 0; fr_destroy: QUOTA_STACK_DESTROY (frame, this); - err: + mq_set_ctx_updation_status (ctx, _gf_false); + return -1; } int -initiate_quota_txn (xlator_t *this, loc_t *loc) +mq_initiate_quota_txn (xlator_t *this, loc_t *loc) { int32_t ret = -1; + gf_boolean_t status = _gf_false; quota_inode_ctx_t *ctx = NULL; inode_contribution_t *contribution = NULL; - VALIDATE_OR_GOTO (loc, out); + GF_VALIDATE_OR_GOTO ("marker", this, out); + GF_VALIDATE_OR_GOTO ("marker", loc, out); + GF_VALIDATE_OR_GOTO ("marker", loc->inode, out); - ret = quota_inode_ctx_get (loc->inode, this, &ctx); + ret = mq_inode_ctx_get (loc->inode, this, &ctx); if (ret == -1) { gf_log (this->name, GF_LOG_WARNING, "inode ctx get failed, aborting quota txn"); @@ -1691,13 +1970,26 @@ initiate_quota_txn (xlator_t *this, loc_t *loc) goto out; } - contribution = get_contribution_node (loc->parent, ctx); + contribution = mq_get_contribution_node (loc->parent, ctx); if (contribution == NULL) goto out; - start_quota_txn (this, loc, ctx, contribution); + /* To improve performance, donot start another transaction + * if one is already in progress for same inode + */ + status = _gf_true; + + ret = mq_test_and_set_ctx_updation_status (ctx, &status); + if (ret < 0) + goto out; + + if (status == _gf_false) { + mq_start_quota_txn (this, loc, ctx, contribution); + } + + ret = 0; out: - return 0; + return ret; } @@ -1706,7 +1998,7 @@ out: /* int64_t contribution) */ /* { */ /* if (size != contribution) { */ -/* initiate_quota_txn (this, loc); */ +/* mq_initiate_quota_txn (this, loc); */ /* } */ /* return 0; */ @@ -1714,31 +2006,39 @@ out: int32_t -inspect_directory_xattr (xlator_t *this, - loc_t *loc, - dict_t *dict, - struct iatt buf) +mq_inspect_directory_xattr (xlator_t *this, + loc_t *loc, + dict_t *dict, + struct iatt buf) { int32_t ret = 0; int8_t dirty = -1; int64_t *size = NULL, size_int = 0; int64_t *contri = NULL, contri_int = 0; char contri_key [512] = {0, }; - marker_conf_t *priv = NULL; gf_boolean_t not_root = _gf_false; quota_inode_ctx_t *ctx = NULL; inode_contribution_t *contribution = NULL; - priv = this->private; - - ret = quota_inode_ctx_get (loc->inode, this, &ctx); + ret = mq_inode_ctx_get (loc->inode, this, &ctx); if (ret < 0) { - ctx = quota_inode_ctx_new (loc->inode, this); + ctx = mq_inode_ctx_new (loc->inode, this); if (ctx == NULL) { gf_log (this->name, GF_LOG_WARNING, - "quota_inode_ctx_new failed"); + "mq_inode_ctx_new failed"); ret = -1; - goto out; + goto err; + } + } + + if (strcmp (loc->path, "/") != 0) { + contribution = mq_add_new_contribution_node (this, ctx, loc); + if (contribution == NULL) { + if (!uuid_is_null (loc->inode->gfid)) + gf_log (this->name, GF_LOG_WARNING, + "cannot add a new contribution node"); + ret = -1; + goto err; } } @@ -1753,13 +2053,6 @@ inspect_directory_xattr (xlator_t *this, if (strcmp (loc->path, "/") != 0) { not_root = _gf_true; - contribution = add_new_contribution_node (this, ctx, loc); - if (contribution == NULL) { - gf_log (this->name, GF_LOG_DEBUG, - "cannot add a new contributio node"); - goto out; - } - GET_CONTRI_KEY (contri_key, contribution->gfid, ret); if (ret < 0) goto out; @@ -1788,47 +2081,47 @@ inspect_directory_xattr (xlator_t *this, " contri=%"PRId64, size_int, contri_int); if (dirty) { - update_dirty_inode (this, loc, ctx, contribution); - } else if ((not_root == _gf_true) && (size_int != contri_int)) { - initiate_quota_txn (this, loc); + ret = mq_update_dirty_inode (this, loc, ctx, contribution); + } + + if ((!dirty || ret == 0) && (not_root == _gf_true) && + (size_int != contri_int)) { + mq_initiate_quota_txn (this, loc); } ret = 0; out: if (ret) - quota_set_inode_xattr (this, loc); - - return 0; + mq_set_inode_xattr (this, loc); +err: + return ret; } int32_t -inspect_file_xattr (xlator_t *this, - loc_t *loc, - dict_t *dict, - struct iatt buf) +mq_inspect_file_xattr (xlator_t *this, + loc_t *loc, + dict_t *dict, + struct iatt buf) { int32_t ret = -1; uint64_t contri_int = 0, size = 0; int64_t *contri_ptr = NULL; char contri_key [512] = {0, }; - marker_conf_t *priv = NULL; quota_inode_ctx_t *ctx = NULL; inode_contribution_t *contribution = NULL; - priv = this->private; - - ret = quota_inode_ctx_get (loc->inode, this, &ctx); + ret = mq_inode_ctx_get (loc->inode, this, &ctx); if (ret < 0) { - ctx = quota_inode_ctx_new (loc->inode, this); + ctx = mq_inode_ctx_new (loc->inode, this); if (ctx == NULL) { gf_log (this->name, GF_LOG_WARNING, - "quota_inode_ctx_new failed"); + "mq_inode_ctx_new failed"); ret = -1; goto out; } } - contribution = add_new_contribution_node (this, ctx, loc); + contribution = mq_add_new_contribution_node (this, ctx, loc); if (contribution == NULL) goto out; @@ -1860,10 +2153,10 @@ inspect_file_xattr (xlator_t *this, "size=%"PRId64 " contri=%"PRId64, size, contri_int); if (size != contri_int) { - initiate_quota_txn (this, loc); + mq_initiate_quota_txn (this, loc); } } else - initiate_quota_txn (this, loc); + mq_initiate_quota_txn (this, loc); } out: @@ -1871,39 +2164,38 @@ out: } int32_t -quota_xattr_state (xlator_t *this, - loc_t *loc, - dict_t *dict, - struct iatt buf) +mq_xattr_state (xlator_t *this, + loc_t *loc, + dict_t *dict, + struct iatt buf) { if (buf.ia_type == IA_IFREG || buf.ia_type == IA_IFLNK) { - inspect_file_xattr (this, loc, dict, buf); + mq_inspect_file_xattr (this, loc, dict, buf); } else if (buf.ia_type == IA_IFDIR) - inspect_directory_xattr (this, loc, dict, buf); + mq_inspect_directory_xattr (this, loc, dict, buf); return 0; } int32_t -quota_req_xattr (xlator_t *this, - loc_t *loc, - dict_t *dict) +mq_req_xattr (xlator_t *this, + loc_t *loc, + dict_t *dict) { int32_t ret = -1; - marker_conf_t *priv = NULL; GF_VALIDATE_OR_GOTO ("marker", this, out); - GF_VALIDATE_OR_GOTO ("marker", loc, out); GF_VALIDATE_OR_GOTO ("marker", dict, out); - priv = this->private; + if (!loc) + goto set_size; //if not "/" then request contribution if (strcmp (loc->path, "/") == 0) goto set_size; - ret = dict_set_contribution (this, dict, loc); + ret = mq_dict_set_contribution (this, dict, loc); if (ret == -1) goto out; @@ -1928,8 +2220,8 @@ out: int32_t -quota_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno) +mq_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { QUOTA_STACK_DESTROY (frame, this); @@ -1937,8 +2229,8 @@ quota_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this, } int32_t -quota_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno) +_mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { int32_t ret = 0; char contri_key [512] = {0, }; @@ -1947,7 +2239,7 @@ quota_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this, local = (quota_local_t *) frame->local; if (op_ret == -1 || local->err == -1) { - quota_removexattr_cbk (frame, NULL, this, -1, 0); + mq_removexattr_cbk (frame, NULL, this, -1, 0, NULL); return 0; } @@ -1956,30 +2248,31 @@ quota_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this, if (local->hl_count > 1) { GET_CONTRI_KEY (contri_key, local->contri->gfid, ret); - STACK_WIND (frame, quota_removexattr_cbk, FIRST_CHILD(this), + STACK_WIND (frame, mq_removexattr_cbk, FIRST_CHILD(this), FIRST_CHILD(this)->fops->removexattr, - &local->loc, contri_key); + &local->loc, contri_key, NULL); ret = 0; } else { - quota_removexattr_cbk (frame, NULL, this, 0, 0); + mq_removexattr_cbk (frame, NULL, this, 0, 0, NULL); } if (strcmp (local->parent_loc.path, "/") != 0) { - get_parent_inode_local (this, local); + ret = mq_get_parent_inode_local (this, local); + if (ret < 0) + goto out; - start_quota_txn (this, &local->loc, local->ctx, local->contri); + mq_start_quota_txn (this, &local->loc, local->ctx, local->contri); } - - /* TODO: free local in quota_local_unref only*/ - quota_local_unref (this, local); - GF_FREE (local); +out: + mq_local_unref (this, local); return 0; } int32_t mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this, - int32_t op_ret, int32_t op_errno, dict_t *dict) + int32_t op_ret, int32_t op_errno, dict_t *dict, + dict_t *xdata) { int32_t ret = -1; struct gf_flock lock = {0, }; @@ -1991,7 +2284,7 @@ mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this, if (op_ret == -1) local->err = -1; - ret = quota_inode_ctx_get (local->parent_loc.inode, this, &ctx); + ret = mq_inode_ctx_get (local->parent_loc.inode, this, &ctx); LOCK (&local->contri->lock); { @@ -2022,24 +2315,22 @@ mq_inode_remove_done (call_frame_t *frame, void *cookie, xlator_t *this, lock.l_pid = 0; STACK_WIND (frame, - quota_inode_remove_done, + _mq_inode_remove_done, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, this->name, &local->parent_loc, - F_SETLKW, &lock); + F_SETLKW, &lock, NULL); return 0; } int32_t -mq_reduce_parent_size_xattr (call_frame_t *frame, void *cookie, - xlator_t *this, int32_t op_ret, int32_t op_errno) +mq_reduce_parent_size_xattr (call_frame_t *frame, void *cookie, xlator_t *this, + int32_t op_ret, int32_t op_errno, dict_t *xdata) { int32_t ret = -1; int64_t *size = NULL; dict_t *dict = NULL; - marker_conf_t *priv = NULL; quota_local_t *local = NULL; - inode_contribution_t *contribution = NULL; local = frame->local; if (op_ret == -1) { @@ -2051,10 +2342,6 @@ mq_reduce_parent_size_xattr (call_frame_t *frame, void *cookie, VALIDATE_OR_GOTO (local->contri, err); - priv = this->private; - - contribution = local->contri; - dict = dict_new (); if (dict == NULL) { ret = -1; @@ -2069,28 +2356,30 @@ mq_reduce_parent_size_xattr (call_frame_t *frame, void *cookie, if (ret < 0) goto err; + uuid_copy (local->parent_loc.gfid, + local->parent_loc.inode->gfid); + GF_UUID_ASSERT (local->parent_loc.gfid); STACK_WIND (frame, mq_inode_remove_done, FIRST_CHILD(this), FIRST_CHILD(this)->fops->xattrop, &local->parent_loc, - GF_XATTROP_ADD_ARRAY64, dict); + GF_XATTROP_ADD_ARRAY64, dict, NULL); dict_unref (dict); return 0; err: local->err = 1; - mq_inode_remove_done (frame, NULL, this, -1, 0, NULL); + mq_inode_remove_done (frame, NULL, this, -1, 0, NULL, NULL); if (dict) dict_unref (dict); return 0; } int32_t -reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri) +mq_reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri) { int32_t ret = -1; struct gf_flock lock = {0,}; call_frame_t *frame = NULL; - marker_conf_t *priv = NULL; quota_local_t *local = NULL; quota_inode_ctx_t *ctx = NULL; inode_contribution_t *contribution = NULL; @@ -2098,17 +2387,15 @@ reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri) GF_VALIDATE_OR_GOTO ("marker", this, out); GF_VALIDATE_OR_GOTO ("marker", loc, out); - priv = this->private; - - ret = quota_inode_ctx_get (loc->inode, this, &ctx); + ret = mq_inode_ctx_get (loc->inode, this, &ctx); if (ret < 0) goto out; - contribution = get_contribution_node (loc->parent, ctx); + contribution = mq_get_contribution_node (loc->parent, ctx); if (contribution == NULL) goto out; - local = quota_local_new (); + local = mq_local_new (); if (local == NULL) { ret = -1; goto out; @@ -2129,14 +2416,14 @@ reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri) goto out; } - ret = loc_copy (&local->loc, loc); + ret = mq_loc_copy (&local->loc, loc); if (ret < 0) goto out; local->ctx = ctx; local->contri = contribution; - ret = quota_inode_loc_fill (NULL, loc->parent, &local->parent_loc); + ret = mq_inode_loc_fill (NULL, loc->parent, &local->parent_loc); if (ret < 0) goto out; @@ -2155,19 +2442,24 @@ reduce_parent_size (xlator_t *this, loc_t *loc, int64_t contri) lock.l_type = F_WRLCK; lock.l_whence = SEEK_SET; + if (local->parent_loc.inode == NULL) { + ret = -1; + gf_log (this->name, GF_LOG_DEBUG, + "Inode is NULL, so can't stackwind."); + goto out; + } + STACK_WIND (frame, mq_reduce_parent_size_xattr, FIRST_CHILD(this), FIRST_CHILD(this)->fops->inodelk, - this->name, &local->parent_loc, F_SETLKW, &lock); + this->name, &local->parent_loc, F_SETLKW, &lock, NULL); local = NULL; ret = 0; out: - if (local != NULL) { - quota_local_unref (this, local); - GF_FREE (local); - } + if (local != NULL) + mq_local_unref (this, local); return ret; } @@ -2181,7 +2473,7 @@ init_quota_priv (xlator_t *this) int32_t -quota_rename_update_newpath (xlator_t *this, loc_t *loc) +mq_rename_update_newpath (xlator_t *this, loc_t *loc) { int32_t ret = -1; quota_inode_ctx_t *ctx = NULL; @@ -2191,23 +2483,23 @@ quota_rename_update_newpath (xlator_t *this, loc_t *loc) GF_VALIDATE_OR_GOTO ("marker", loc, out); GF_VALIDATE_OR_GOTO ("marker", loc->inode, out); - ret = quota_inode_ctx_get (loc->inode, this, &ctx); + ret = mq_inode_ctx_get (loc->inode, this, &ctx); if (ret < 0) goto out; - contribution = add_new_contribution_node (this, ctx, loc); + contribution = mq_add_new_contribution_node (this, ctx, loc); if (contribution == NULL) { ret = -1; goto out; } - initiate_quota_txn (this, loc); + mq_initiate_quota_txn (this, loc); out: return ret; } int32_t -quota_forget (xlator_t *this, quota_inode_ctx_t *ctx) +mq_forget (xlator_t *this, quota_inode_ctx_t *ctx) { inode_contribution_t *contri = NULL; inode_contribution_t *next = NULL; |