cluster/afr: Hold blocking locks for data self-heal.

Data self-heal now holds blocking locks, and instead of locking
on all subvolumes, it only locks on {data-lock-server-count} subvolumes.

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

BUG: 170 (Auto-heal fails on files that are open()-ed/mmap()-ed)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=170

4 files changed, 62 insertions, 35 deletions

diff --git a/xlators/cluster/afr/src/afr-self-heal-data.c b/xlators/cluster/afr/src/afr-self-heal-data.c
index 34fe181ea0..bafb92fd21 100644
--- a/xlators/cluster/afr/src/afr-self-heal-data.c
+++ b/xlators/cluster/afr/src/afr-self-heal-data.c
@@ -917,12 +917,14 @@ afr_sh_data_lookup (call_frame_t *frame, xlator_t *this)
 
 
 int
+afr_sh_data_lock_rec (call_frame_t *frame, xlator_t *this, int child_index);
+
+int
 afr_sh_data_lock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 		      int32_t op_ret, int32_t op_errno)
 {
 	afr_local_t     *local = NULL;
 	afr_self_heal_t *sh = NULL;
-	int              call_count = 0;
 	int              child_index = (long) cookie;
 
 	/* TODO: what if lock fails? */
@@ -933,15 +935,16 @@ afr_sh_data_lock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 	LOCK (&frame->lock);
 	{
 		if (op_ret == -1) {
-			sh->op_failed = 1;
-
                         sh->locked_nodes[child_index] = 0;
+
 			gf_log (this->name, GF_LOG_DEBUG,
 				"locking of %s on child %d failed: %s",
 				local->loc.path, child_index,
 				strerror (op_errno));
 		} else {
                         sh->locked_nodes[child_index] = 1;
+                        sh->lock_count++;
+
 			gf_log (this->name, GF_LOG_TRACE,
 				"inode of %s on child %d locked",
 				local->loc.path, child_index);
@@ -949,67 +952,87 @@ afr_sh_data_lock_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 	}
 	UNLOCK (&frame->lock);
 
-	call_count = afr_frame_return (frame);
-
-	if (call_count == 0) {
-		if (sh->op_failed) {
-			afr_sh_data_finish (frame, this);
-			return 0;
-		}
-
-		afr_sh_data_lookup (frame, this);
-	}
+        afr_sh_data_lock_rec (frame, this, child_index + 1);
 
 	return 0;
 }
 
 
 int
-afr_sh_data_lock (call_frame_t *frame, xlator_t *this)
+afr_sh_data_lock_rec (call_frame_t *frame, xlator_t *this, int child_index)
 {
 	struct flock flock;			
 	int i = 0;				
-	int call_count = 0;		     
 
 	afr_local_t *   local = NULL;
 	afr_private_t * priv  = NULL;
 	afr_self_heal_t * sh  = NULL;
 
-
 	local = frame->local;
 	sh = &local->self_heal;
 	priv = this->private;
 
-	call_count = local->child_count;
-
-	local->call_count = call_count;		
-
 	flock.l_start = 0;
 	flock.l_len   = 0;
 	flock.l_type  = F_WRLCK;			
 
-	for (i = 0; i < priv->child_count; i++) {
-		if (local->child_up[i]) {
-			gf_log (this->name, GF_LOG_TRACE,
-				"locking %s on subvolume %s",
-				local->loc.path, priv->children[i]->name);
+	/* skip over children that are down */
+	while ((child_index < priv->child_count)
+	       && !local->child_up[child_index])
+		child_index++;
 
-			STACK_WIND_COOKIE (frame, afr_sh_data_lock_cbk,
-					   (void *) (long) i,
-					   priv->children[i], 
-					   priv->children[i]->fops->inodelk,
-                                           this->name,
-					   &local->loc, F_SETLK, &flock); 
-			if (!--call_count)
-				break;
-		}
+	if ((child_index == priv->child_count) &&
+	    sh->lock_count == 0) {
+
+		gf_log (this->name, GF_LOG_DEBUG,
+			"unable to lock on even one child");
+
+                afr_sh_data_done (frame, this);
+		return 0;
 	}
 
+        if ((child_index == priv->child_count)
+            || (sh->lock_count == afr_lock_server_count (priv, AFR_DATA_TRANSACTION))) {
+                afr_sh_data_lookup (frame, this);
+                return 0;
+        }
+
+        gf_log (this->name, GF_LOG_TRACE,
+                "locking %s on subvolume %s",
+                local->loc.path, priv->children[i]->name);
+
+        STACK_WIND_COOKIE (frame, afr_sh_data_lock_cbk,
+                           (void *) (long) child_index,
+                           priv->children[i],
+                           priv->children[i]->fops->inodelk,
+                           this->name,
+                           &local->loc, F_SETLKW, &flock);
+
 	return 0;
 }
 
 
 int
+afr_sh_data_lock (call_frame_t *frame, xlator_t *this)
+{
+	afr_local_t *   local = NULL;
+	afr_private_t * priv  = NULL;
+	afr_self_heal_t * sh  = NULL;
+
+        int i = 0;
+
+	local = frame->local;
+	sh    = &local->self_heal;
+	priv  = this->private;
+
+        for (i = 0; i < priv->child_count; i++)
+                sh->locked_nodes[i] = 0;
+
+        return afr_sh_data_lock_rec (frame, this, 0);
+}
+
+
+int
 afr_self_heal_data (call_frame_t *frame, xlator_t *this)
 {
 	afr_local_t   *local = NULL;
diff --git a/xlators/cluster/afr/src/afr-transaction.c b/xlators/cluster/afr/src/afr-transaction.c
index ca13b9f071..a2627b9cac 100644
--- a/xlators/cluster/afr/src/afr-transaction.c
+++ b/xlators/cluster/afr/src/afr-transaction.c
@@ -358,7 +358,7 @@ out:
 }
 
 
-static int
+int
 afr_lock_server_count (afr_private_t *priv, afr_transaction_type type)
 {
 	int ret = 0;
diff --git a/xlators/cluster/afr/src/afr-transaction.h b/xlators/cluster/afr/src/afr-transaction.h
index 4ba7dfd6dc..0d3d4443e3 100644
--- a/xlators/cluster/afr/src/afr-transaction.h
+++ b/xlators/cluster/afr/src/afr-transaction.h
@@ -24,6 +24,9 @@ void
 afr_transaction_fop_failed (call_frame_t *frame, xlator_t *this,
 			    int child_index);
 
+int
+afr_lock_server_count (afr_private_t *priv, afr_transaction_type type);
+
 int32_t
 afr_transaction (call_frame_t *frame, xlator_t *this, afr_transaction_type type);
 
diff --git a/xlators/cluster/afr/src/afr.h b/xlators/cluster/afr/src/afr.h
index 3bd8a458bc..2f57426621 100644
--- a/xlators/cluster/afr/src/afr.h
+++ b/xlators/cluster/afr/src/afr.h
@@ -95,6 +95,7 @@ typedef struct {
 	int active_sinks;
 	int *success;
 	int *locked_nodes;
+        int lock_count;
 
         mode_t impunging_entry_mode;
         const char *linkname;