cluster/afr: Do self-heal on unopened fds.

This patch completes the previous patch for self-heal of
open fds in replicate.

If an fd was never opened on a subvolume, we remember that
and do the open after we've done self-heal on that fd.

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

1 files changed, 163 insertions, 22 deletions

diff --git a/xlators/cluster/afr/src/afr-open.c b/xlators/cluster/afr/src/afr-open.c
index 1918dee8167..539e258a1d7 100644
--- a/xlators/cluster/afr/src/afr-open.c
+++ b/xlators/cluster/afr/src/afr-open.c
@@ -62,14 +62,6 @@ afr_open_ftruncate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                         struct stat *postbuf)
 {
 	afr_local_t * local = frame->local;
-        int ret = 0;
-
-        ret = afr_fd_ctx_set (this, local->fd);
-
-        if (ret < 0) {
-                local->op_ret   =   -1;
-                local->op_errno = -ret;
-        }
 
 	AFR_STACK_UNWIND (open, frame, local->op_ret, local->op_errno,
 			  local->fd);
@@ -85,6 +77,11 @@ afr_open_cbk (call_frame_t *frame, void *cookie,
 	afr_local_t *  local = NULL;
 	afr_private_t * priv = NULL;
 
+        int child_index = (long) cookie;
+
+        uint64_t      ctx;
+        afr_fd_ctx_t *fd_ctx;
+
         int ret = 0;
 
 	int call_count = -1;
@@ -101,6 +98,32 @@ afr_open_cbk (call_frame_t *frame, void *cookie,
 		if (op_ret >= 0) {
 			local->op_ret = op_ret;
                         local->success_count++;
+
+                        ret = afr_fd_ctx_set (this, fd);
+
+                        if (ret < 0) {
+                                gf_log (this->name, GF_LOG_DEBUG,
+                                        "could not set fd ctx for fd=%p",
+                                        fd);
+
+                                local->op_ret   = -1;
+                                local->op_errno = -ret;
+                        }
+
+                        ret = fd_ctx_get (fd, this, &ctx);
+
+                        if (ret < 0) {
+                                gf_log (this->name, GF_LOG_DEBUG,
+                                        "could not get fd ctx for fd=%p", fd);
+                                local->op_ret   = -1;
+                                local->op_errno = -ret;
+                        }
+
+                        fd_ctx = (afr_fd_ctx_t *)(long) ctx;
+
+                        fd_ctx->opened_on[child_index] = 1;
+                        fd_ctx->flags                  = local->cont.open.flags;
+                        fd_ctx->wbflags                = local->cont.open.wbflags;
 		}
 	}
 	UNLOCK (&frame->lock);
@@ -114,17 +137,6 @@ afr_open_cbk (call_frame_t *frame, void *cookie,
 				    this, this->fops->ftruncate,
 				    fd, 0);
 		} else {
-                        ret = afr_fd_ctx_set (this, fd);
-
-                        if (ret < 0) {
-                                gf_log (this->name, GF_LOG_DEBUG,
-                                        "could not set fd ctx for fd=%p",
-                                        fd);
-
-                                local->op_ret   = -1;
-                                local->op_errno = -ret;
-                        }
-
                         AFR_STACK_UNWIND (open, frame, local->op_ret,
                                           local->op_errno, local->fd);
 		}
@@ -173,7 +185,11 @@ afr_open (call_frame_t *frame, xlator_t *this, loc_t *loc, int32_t flags,
 	frame->local = local;
 	call_count   = local->call_count;
 
-	local->cont.open.flags = flags;
+        loc_copy (&local->loc, loc);
+
+	local->cont.open.flags   = flags;
+        local->cont.open.wbflags = wbflags;
+
 	local->fd = fd_ref (fd);
 
 	for (i = 0; i < priv->child_count; i++) {
@@ -199,13 +215,133 @@ out:
 
 
 int
+afr_up_down_flush_open_cbk (call_frame_t *frame, void *cookie,
+                            xlator_t *this, int32_t op_ret, int32_t op_errno,
+                            fd_t *fd)
+{
+        afr_local_t *local  = NULL;
+        afr_private_t *priv = NULL;
+
+        int ret = 0;
+
+        uint64_t      ctx;
+        afr_fd_ctx_t *fd_ctx;
+
+        int call_count  = 0;
+        int child_index = (long) cookie;
+
+        priv  = this->private;
+        local = frame->local;
+
+        LOCK (&frame->lock);
+        {
+                if (op_ret >= 0) {
+                        ret = fd_ctx_get (fd, this, &ctx);
+
+                        if (ret < 0) {
+                                goto out;
+                        }
+
+                        fd_ctx = (afr_fd_ctx_t *)(long) ctx;
+
+                        fd_ctx->opened_on[child_index] = 1;
+
+                        gf_log (this->name, GF_LOG_TRACE,
+                                "fd for %s opened successfully on subvolume %s",
+                                local->loc.path, priv->children[child_index]->name);
+                }
+        }
+out:
+        UNLOCK (&frame->lock);
+
+        call_count = afr_frame_return (frame);
+
+        if (call_count == 0) {
+                local->transaction.post_post_op (frame, this);
+        }
+
+        return 0;
+}
+
+
+static int
+__unopened_count (int child_count, unsigned char *opened_on, unsigned char *child_up)
+{
+        int i;
+        int count = 0;
+
+        for (i = 0; i < child_count; i++) {
+                if (!opened_on[i] && child_up[i])
+                        count++;
+        }
+
+        return count;
+}
+
+
+int
 afr_up_down_flush_sh_unwind (call_frame_t *frame, xlator_t *this)
 {
-        afr_local_t *local = NULL;
+        afr_local_t *local  = NULL;
+        afr_private_t *priv = NULL;
+
+        uint64_t      ctx;
+        afr_fd_ctx_t *fd_ctx;
 
+        int abandon = 0;
+        int ret = 0;
+        int i;
+        int call_count = 0;
+
+        priv  = this->private;
         local = frame->local;
 
-        local->transaction.post_post_op (frame, this);
+        /*
+         * Some subvolumes might have come up on which we never
+         * opened this fd in the first place. Re-open fd's on those
+         * subvolumes now.
+         */
+
+        ret = fd_ctx_get (local->fd, this, &ctx);
+
+        if (ret < 0) {
+                abandon = 1;
+                goto out;
+        }
+
+        fd_ctx = (afr_fd_ctx_t *)(long) ctx;
+
+        call_count = __unopened_count (priv->child_count, fd_ctx->opened_on,
+                                       local->child_up);
+
+        if (call_count == 0) {
+                abandon = 1;
+                goto out;
+        }
+
+        local->call_count = call_count;
+
+        for (i = 0; i < priv->child_count; i++) {
+                if (!fd_ctx->opened_on[i] && local->child_up[i]) {
+                        gf_log (this->name, GF_LOG_TRACE,
+                                "opening fd for %s on subvolume %s",
+                                local->loc.path, priv->children[i]->name);
+
+                        STACK_WIND_COOKIE (frame, afr_up_down_flush_open_cbk,
+                                           (void *)(long) i,
+                                           priv->children[i],
+                                           priv->children[i]->fops->open,
+                                           &local->loc, fd_ctx->flags, local->fd,
+                                           fd_ctx->wbflags);
+
+                        if (!--call_count)
+                                break;
+                }
+        }
+
+out:
+        if (abandon)
+                local->transaction.post_post_op (frame, this);
 
         return 0;
 }
@@ -227,6 +363,11 @@ afr_up_down_flush_post_post_op (call_frame_t *frame, xlator_t *this)
         local->loc.inode  = inode_ref (local->fd->inode);
         local->loc.parent = inode_parent (local->fd->inode, 0, NULL);
 
+        /* forcibly trigger missing-entries self-heal */
+
+        local->success_count    = 1;
+        local->enoent_count     = 1;
+
         sh->data_lock_held      = _gf_true;
         sh->need_data_self_heal = _gf_true;
         sh->mode                = local->fd->inode->st_mode;