summaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--xlators/cluster/dht/src/dht-common.h3
-rw-r--r--xlators/cluster/dht/src/dht-helper.c3
-rw-r--r--xlators/cluster/dht/src/dht-mem-types.h1
-rw-r--r--xlators/cluster/dht/src/dht-rename.c61
4 files changed, 61 insertions, 7 deletions
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index be9d2ce5e4e..6030af7c41d 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -303,6 +303,9 @@ struct dht_local {
call_stub_t *stub;
int32_t parent_disk_layout[4];
+
+ /* rename rollback */
+ int *ret_cache ;
};
typedef struct dht_local dht_local_t;
diff --git a/xlators/cluster/dht/src/dht-helper.c b/xlators/cluster/dht/src/dht-helper.c
index ad031b6216c..81d1dffa0af 100644
--- a/xlators/cluster/dht/src/dht-helper.c
+++ b/xlators/cluster/dht/src/dht-helper.c
@@ -638,6 +638,9 @@ dht_local_wipe (xlator_t *this, dht_local_t *local)
local->stub = NULL;
}
+ if (local->ret_cache)
+ GF_FREE (local->ret_cache);
+
mem_put (local);
}
diff --git a/xlators/cluster/dht/src/dht-mem-types.h b/xlators/cluster/dht/src/dht-mem-types.h
index 5de5d1838ad..3554f3f9c2d 100644
--- a/xlators/cluster/dht/src/dht-mem-types.h
+++ b/xlators/cluster/dht/src/dht-mem-types.h
@@ -38,6 +38,7 @@ enum gf_dht_mem_types_ {
gf_tier_mt_ipc_ctr_params_t,
gf_dht_mt_fd_ctx_t,
gf_tier_mt_qfile_array_t,
+ gf_dht_ret_cache_t,
gf_dht_mt_end
};
#endif
diff --git a/xlators/cluster/dht/src/dht-rename.c b/xlators/cluster/dht/src/dht-rename.c
index 7e7e7151af7..53c61f8a714 100644
--- a/xlators/cluster/dht/src/dht-rename.c
+++ b/xlators/cluster/dht/src/dht-rename.c
@@ -25,17 +25,22 @@ dht_rename_dir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
struct iatt *prenewparent, struct iatt *postnewparent,
dict_t *xdata)
{
- dht_local_t *local = NULL;
- int this_call_cnt = 0;
- xlator_t *prev = NULL;
- char gfid[GF_UUID_BUF_SIZE] = {0};
+ dht_conf_t *conf = NULL;
+ dht_local_t *local = NULL;
+ int this_call_cnt = 0;
+ xlator_t *prev = NULL;
+ int i = 0;
+ char gfid[GF_UUID_BUF_SIZE] = {0};
+ int subvol_cnt = -1;
+ conf = this->private;
local = frame->local;
prev = cookie;
+ subvol_cnt = dht_subvol_cnt (this, prev);
+ local->ret_cache[subvol_cnt] = op_ret;
if (op_ret == -1) {
- /* TODO: undo the damage */
gf_uuid_unparse(local->loc.inode->gfid, gfid);
gf_msg (this->name, GF_LOG_INFO, op_errno,
@@ -63,6 +68,42 @@ dht_rename_dir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
unwind:
this_call_cnt = dht_frame_return (frame);
if (is_last_call (this_call_cnt)) {
+ /* We get here with local->call_cnt == 0. Which means
+ * we are the only one executing this code, there is
+ * no contention. Therefore it's safe to manipulate or
+ * deref local->call_cnt directly (without locking).
+ */
+ if (local->ret_cache[conf->subvolume_cnt] == 0) {
+ /* count errant subvols in last field of ret_cache */
+ for (i = 0; i < conf->subvolume_cnt; i++) {
+ if (local->ret_cache[i] != 0)
+ ++local->ret_cache[conf->subvolume_cnt];
+ }
+ if (local->ret_cache[conf->subvolume_cnt]) {
+ /* undoing the damage:
+ * for all subvolumes, where rename
+ * succeeded, we perform the reverse operation
+ */
+ for (i = 0; i < conf->subvolume_cnt; i++) {
+ if (local->ret_cache[i] == 0)
+ ++local->call_cnt;
+ }
+ for (i = 0; i < conf->subvolume_cnt; i++) {
+ if (local->ret_cache[i])
+ continue;
+
+ STACK_WIND (frame,
+ dht_rename_dir_cbk,
+ conf->subvolumes[i],
+ conf->subvolumes[i]->fops->rename,
+ &local->loc2, &local->loc,
+ NULL);
+ }
+
+ return 0;
+ }
+ }
+
WIPE (&local->preoldparent);
WIPE (&local->postoldparent);
WIPE (&local->preparent);
@@ -96,8 +137,6 @@ dht_rename_hashed_dir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
if (op_ret == -1) {
- /* TODO: undo the damage */
-
gf_uuid_unparse(local->loc.inode->gfid, gfid);
gf_msg (this->name, GF_LOG_INFO, op_errno,
@@ -324,6 +363,14 @@ dht_rename_dir (call_frame_t *frame, xlator_t *this)
conf = frame->this->private;
local = frame->local;
+ local->ret_cache = GF_CALLOC (conf->subvolume_cnt + 1, sizeof (int),
+ gf_dht_ret_cache_t);
+
+ if (local->ret_cache == NULL) {
+ op_errno = ENOMEM;
+ goto err;
+ }
+
/* We must take a lock on all the subvols with src gfid.
* Along with this if dst exists we must take lock on
* any one subvol with dst gfid.