cluster/dht: Send removexattr() to all subvolumes.

Signed-off-by: Vikas Gorur <vikas@gluster.com>
Signed-off-by: Anand V. Avati <avati@dev.gluster.com>

BUG: 2426 (removexattr is not sent to all subvolumes)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=2426

1 files changed, 52 insertions, 4 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index 4b574bb4504..37ec5c4123b 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -2129,13 +2129,50 @@ err:
 
 
 int
+dht_removexattr_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
+		     int op_ret, int op_errno)
+{
+	dht_local_t  *local = NULL;
+	int           this_call_cnt = 0;
+	call_frame_t *prev = NULL;
+
+	local = frame->local;
+	prev = cookie;
+
+	LOCK (&frame->lock);
+	{
+		if (op_ret == -1) {
+			local->op_errno = op_errno;
+			gf_log (this->name, GF_LOG_DEBUG,
+				"subvolume %s returned -1 (%s)",
+				prev->this->name, strerror (op_errno));
+			goto unlock;
+		}
+
+		local->op_ret = 0;
+	}
+unlock:
+	UNLOCK (&frame->lock);
+
+	this_call_cnt = dht_frame_return (frame);
+	if (is_last_call (this_call_cnt)) {
+		DHT_STACK_UNWIND (removexattr, frame, local->op_ret, local->op_errno);
+        }
+
+        return 0;
+}
+
+
+int
 dht_removexattr (call_frame_t *frame, xlator_t *this,
 		 loc_t *loc, const char *key)
 {
 	xlator_t     *subvol = NULL;
         int           op_errno = -1;
 	dht_local_t  *local = NULL;
+	dht_layout_t *layout = NULL;
 
+	int i;
 
         VALIDATE_OR_GOTO (frame, err);
         VALIDATE_OR_GOTO (this, err);
@@ -2159,11 +2196,22 @@ dht_removexattr (call_frame_t *frame, xlator_t *this,
 		goto err;
 	}
 
-	local->call_cnt = 1;
+	local->layout = layout = dht_layout_get (this, loc->inode);
+	if (!local->layout) {
+		gf_log (this->name, GF_LOG_DEBUG,
+			"no layout for path=%s", loc->path);
+		op_errno = EINVAL;
+		goto err;
+	}
 
-	STACK_WIND (frame, dht_err_cbk,
-		    subvol, subvol->fops->removexattr,
-		    loc, key);
+	local->call_cnt = layout->cnt;
+
+	for (i = 0; i < layout->cnt; i++) {
+		STACK_WIND (frame, dht_removexattr_cbk,
+			    layout->list[i].xlator,
+			    layout->list[i].xlator->fops->removexattr,
+			    loc, key);
+	}
 
 	return 0;