distribute,nufa: layout handling changes

changes to make revalidate not fail but instead perform fresh lookup
and swap inode context (layout) safely

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

BUG: 315 (generation number support)
URL: http://bugs.gluster.com/cgi-bin/bugzilla3/show_bug.cgi?id=315

7 files changed, 209 insertions, 136 deletions

diff --git a/xlators/cluster/dht/src/dht-common.c b/xlators/cluster/dht/src/dht-common.c
index b8996d46e..62b864ac9 100644
--- a/xlators/cluster/dht/src/dht-common.c
+++ b/xlators/cluster/dht/src/dht-common.c
@@ -55,12 +55,8 @@ dht_lookup_selfheal_cbk (call_frame_t *frame, void *cookie,
 
 	if (ret == 0) {
 		layout = local->selfheal.layout;
-		ret = inode_ctx_put (local->inode, this, 
-                                     (uint64_t)(long)layout);
+		ret = dht_layout_set (this, local->inode, layout);
 
-		if (ret == 0)
-			local->selfheal.layout = NULL;
-		
 		if (local->st_ino) {
 			local->stbuf.st_ino = local->st_ino;
 		} else {
@@ -160,8 +156,6 @@ unlock:
 		if (local->op_ret == 0) {
 			ret = dht_layout_normalize (this, &local->loc, layout);
 
-			local->layout = NULL;
-
 			if (ret != 0) {
 				gf_log (this->name, GF_LOG_DEBUG,
 					"fixing assignment on %s",
@@ -169,8 +163,7 @@ unlock:
 				goto selfheal;
 			}
 			
-			inode_ctx_put (local->inode, this,
-                                       (uint64_t)(long)layout);
+			dht_layout_set (this, local->inode, layout);
 			
 			if (local->st_ino) {
 				local->stbuf.st_ino = local->st_ino;
@@ -254,7 +247,7 @@ dht_revalidate_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 			goto unlock;
 		}
 
-		layout = dht_layout_get (this, inode);
+		layout = local->layout;
 		
 		is_dir = check_is_dir (inode, stbuf, xattr);
 		is_linkfile = check_is_linkfile (inode, stbuf, xattr);
@@ -338,7 +331,7 @@ dht_lookup_linkfile_create_cbk (call_frame_t *frame, void *cookie,
 	cached_subvol = local->cached_subvol;
 	conf = this->private;
 
-        ret = dht_layout_inode_set (this, local->cached_subvol, inode);
+        ret = dht_layout_preset (this, local->cached_subvol, inode);
         if (ret < 0) {
                 gf_log (this->name, GF_LOG_DEBUG,
                         "failed to set layout for subvolume %s",
@@ -488,8 +481,8 @@ unlock:
                         local->op_ret = 0;
                         local->op_errno = 0;
 
-                        ret = dht_layout_inode_set (frame->this, cached_subvol,
-                                                    local->inode);
+                        ret = dht_layout_preset (frame->this, cached_subvol,
+                                                 local->inode);
                         if (ret < 0) {
                                 gf_log (this->name, GF_LOG_DEBUG,
                                         "failed to set layout for subvol %s",
@@ -560,10 +553,10 @@ dht_lookup_linkfile_cbk (call_frame_t *frame, void *cookie,
 {
         call_frame_t *prev          = NULL;
 	dht_local_t  *local         = NULL;
-	dht_layout_t *layout        = NULL;
 	xlator_t     *subvol        = NULL;
 	loc_t        *loc           = NULL;
 	dht_conf_t   *conf          = NULL;
+        int           ret           = 0;
 
         prev   = cookie;
 	subvol = prev->this;
@@ -600,18 +593,16 @@ dht_lookup_linkfile_cbk (call_frame_t *frame, void *cookie,
         if (local->loc.parent)
                 postparent->st_ino = local->loc.parent->ino;
         
-	layout = dht_layout_for_subvol (this, prev->this);
-	if (!layout) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"no pre-set layout for subvolume %s",
-			prev->this->name);
+	ret = dht_layout_preset (this, prev->this, inode);
+	if (ret < 0) {
+                gf_log (this->name, GF_LOG_DEBUG,
+                        "failed to set layout for subvolume %s",
+                        prev->this->name);
 		op_ret   = -1;
 		op_errno = EINVAL;
 		goto out;
 	}
 
-	inode_ctx_put (inode, this, (uint64_t)(long)layout);
-
 out:
         DHT_STACK_UNWIND (lookup, frame, op_ret, op_errno, inode, stbuf, xattr,
                           postparent);
@@ -663,7 +654,6 @@ dht_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 inode_t *inode, struct stat *stbuf, dict_t *xattr,
                 struct stat *postparent)
 {
-	dht_layout_t *layout        = NULL;
         char          is_linkfile   = 0;
         char          is_dir        = 0;
         xlator_t     *subvol        = NULL;
@@ -671,6 +661,7 @@ dht_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         dht_local_t  *local         = NULL;
         loc_t        *loc           = NULL;
         call_frame_t *prev          = NULL;
+        int           ret           = 0;
 
         conf  = this->private;
 
@@ -710,20 +701,17 @@ dht_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 
 		dht_itransform (this, prev->this, stbuf->st_ino,
 				&stbuf->st_ino);
-
                 postparent->st_ino = loc->parent->ino;
 
-		layout = dht_layout_for_subvol (this, prev->this);
-		if (!layout) {
+		ret = dht_layout_preset (this, prev->this, inode);
+		if (ret < 0) {
 			gf_log (this->name, GF_LOG_DEBUG,
-				"no pre-set layout for subvolume %s",
+				"could not set pre-set layout for subvolume %s",
 				prev->this->name);
 			op_ret   = -1;
 			op_errno = EINVAL;
 			goto out;
 		}
-
-                inode_ctx_put (inode, this, (uint64_t)(long)layout);
                 goto out; 
 	}
 
@@ -812,7 +800,7 @@ dht_lookup (call_frame_t *frame, xlator_t *this,
 	local->hashed_subvol = hashed_subvol;
 
         if (is_revalidate (loc)) {
-		layout = dht_layout_get (this, loc->inode);
+		local->layout = layout = dht_layout_get (this, loc->inode);
 
                 if (!layout) {
                         gf_log (this->name, GF_LOG_DEBUG,
@@ -826,8 +814,10 @@ dht_lookup (call_frame_t *frame, xlator_t *this,
 			gf_log (this->name, GF_LOG_TRACE,
 				"incomplete layout failure for path=%s",
 				loc->path);
-			op_errno = ESTALE;
-			goto err;
+
+                        dht_layout_unref (this, local->layout);
+                        local->layout = NULL;
+			goto do_fresh_lookup;
 		}
 
 		local->inode    = inode_ref (loc->inode);
@@ -853,6 +843,7 @@ dht_lookup (call_frame_t *frame, xlator_t *this,
 				break;
 		}
         } else {
+        do_fresh_lookup:
 		/* TODO: remove the hard-coding */
 		ret = dict_set_uint32 (local->xattr_req, 
 				       "trusted.glusterfs.dht", 4 * 4);
@@ -1003,13 +994,6 @@ dht_stat (call_frame_t *frame, xlator_t *this,
         VALIDATE_OR_GOTO (loc->inode, err);
         VALIDATE_OR_GOTO (loc->path, err);
 
-	layout = dht_layout_get (this, loc->inode);
-	if (!layout) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"no layout for path=%s", loc->path);
-		op_errno = EINVAL;
-		goto err;
-	}
 
 	local = dht_local_init (frame);
 	if (!local) {
@@ -1019,6 +1003,14 @@ dht_stat (call_frame_t *frame, xlator_t *this,
 		goto err;
 	}
 
+	local->layout = layout = dht_layout_get (this, loc->inode);
+	if (!layout) {
+		gf_log (this->name, GF_LOG_DEBUG,
+			"no layout for path=%s", loc->path);
+		op_errno = EINVAL;
+		goto err;
+	}
+
 	local->inode = inode_ref (loc->inode);
 	local->call_cnt = layout->cnt;
 
@@ -1055,14 +1047,6 @@ dht_fstat (call_frame_t *frame, xlator_t *this,
         VALIDATE_OR_GOTO (this, err);
         VALIDATE_OR_GOTO (fd, err);
 
-	layout = dht_layout_get (this, fd->inode);
-	if (!layout) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"no layout for fd=%p", fd);
-		op_errno = EINVAL;
-		goto err;
-	}
-
 	local = dht_local_init (frame);
 	if (!local) {
 		op_errno = ENOMEM;
@@ -1071,6 +1055,14 @@ dht_fstat (call_frame_t *frame, xlator_t *this,
 		goto err;
 	}
 
+	local->layout = layout = dht_layout_get (this, fd->inode);
+	if (!layout) {
+		gf_log (this->name, GF_LOG_DEBUG,
+			"no layout for fd=%p", fd);
+		op_errno = EINVAL;
+		goto err;
+	}
+
 	local->inode    = inode_ref (fd->inode);
 	local->call_cnt = layout->cnt;;
 
@@ -2252,7 +2244,7 @@ int
 dht_readdir (call_frame_t *frame, xlator_t *this, fd_t *fd, size_t size,
              off_t yoff)
 {
-        dht_do_readdir (frame, this, fd, size, yoff, GF_FOP_READDIR);
+        dht_do_readdir (frame, this, fd, size, yoff, GF_FOP_READDIRP);
         return 0;
 }
 
@@ -2344,7 +2336,6 @@ dht_newfile_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  struct stat *postparent)
 {
 	call_frame_t *prev = NULL;
-	dht_layout_t *layout = NULL;
 	int           ret = -1;
         dht_local_t  *local = NULL;
 
@@ -2367,26 +2358,15 @@ dht_newfile_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 postparent->st_ino = local->loc.parent->ino;
         }
 
-	layout = dht_layout_for_subvol (this, prev->this);
-
-	if (!layout) {
+        ret = dht_layout_preset (this, prev->this, inode);
+	if (ret < 0) {
 		gf_log (this->name, GF_LOG_DEBUG,
-			"no pre-set layout for subvolume %s",
+			"could not set pre-set layout for subvolume %s",
 			prev->this->name);
 		op_ret   = -1;
 		op_errno = EINVAL;
 		goto out;
 	}
-
-	ret = inode_ctx_put (inode, this, (uint64_t)(long)layout);
-	if (ret != 0) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"could not set inode context");
-		op_ret   = -1;
-		op_errno = EINVAL;
-		goto out;
-	}
-
 out:
         /* 
          * FIXME: st_size and st_blocks of preparent and postparent do not have 
@@ -2763,7 +2743,6 @@ dht_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                  struct stat *preparent, struct stat *postparent)
 {
 	call_frame_t *prev = NULL;
-	dht_layout_t *layout = NULL;
 	int           ret = -1;
         dht_local_t  *local = NULL;
 
@@ -2785,22 +2764,11 @@ dht_create_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                 postparent->st_ino = local->loc.parent->ino;
         }
 
-
-	layout = dht_layout_for_subvol (this, prev->this);
-
-	if (!layout) {
-		gf_log (this->name, GF_LOG_DEBUG,
-			"no pre-set layout for subvolume %s",
-			prev->this->name);
-		op_ret   = -1;
-		op_errno = EINVAL;
-		goto out;
-	}
-
-	ret = inode_ctx_put (inode, this, (uint64_t)(long)layout);
+        ret = dht_layout_preset (this, prev->this, inode);
 	if (ret != 0) {
 		gf_log (this->name, GF_LOG_DEBUG,
-			"could not set inode context");
+			"could not set preset layout for subvol %s",
+                        prev->this->name);
 		op_ret   = -1;
 		op_errno = EINVAL;
 		goto out;
@@ -2940,8 +2908,7 @@ dht_mkdir_selfheal_cbk (call_frame_t *frame, void *cookie,
 	layout = local->selfheal.layout;
 
 	if (op_ret == 0) {
-		inode_ctx_put (local->inode, this, (uint64_t)(long)layout);
-		local->selfheal.layout = NULL;
+                dht_layout_set (this, local->inode, layout);
 		local->stbuf.st_ino = local->st_ino;
                 if (local->loc.parent) {
                         local->preparent.st_ino = local->loc.parent->ino;
@@ -3000,7 +2967,6 @@ unlock:
 
 	this_call_cnt = dht_frame_return (frame);
 	if (is_last_call (this_call_cnt)) {
-		local->layout = NULL;
 		dht_selfheal_new_directory (frame, dht_mkdir_selfheal_cbk,
 					    layout);
 	}
@@ -3050,7 +3016,6 @@ dht_mkdir_hashed_cbk (call_frame_t *frame, void *cookie,
 	local->call_cnt = conf->subvolume_cnt - 1;
 	
 	if (local->call_cnt == 0) {
-		local->layout = NULL;
 		dht_selfheal_directory (frame, dht_mkdir_selfheal_cbk,
 					&local->loc, layout);
 	}
@@ -3149,7 +3114,6 @@ dht_rmdir_selfheal_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 	dht_local_t  *local = NULL;
 
 	local = frame->local;
-	local->layout = NULL;
 
         if (local->loc.parent) {
                 local->preparent.st_ino = local->loc.parent->ino;
@@ -3168,11 +3132,9 @@ dht_rmdir_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 	       int op_ret, int op_errno, struct stat *preparent,
                struct stat *postparent)
 {
-	uint64_t      tmp_layout = 0;
 	dht_local_t  *local = NULL;
 	int           this_call_cnt = 0;
 	call_frame_t *prev = NULL;
-	dht_layout_t *layout = NULL;
 
 	local = frame->local;
 	prev  = cookie;
@@ -3204,15 +3166,14 @@ unlock:
 	this_call_cnt = dht_frame_return (frame);
 	if (is_last_call (this_call_cnt)) {
 		if (local->need_selfheal) {
-			inode_ctx_get (local->loc.inode, this, 
-				       &tmp_layout);
-			layout = (dht_layout_t *)(long)tmp_layout;
+                        local->layout =
+                                dht_layout_get (this, local->loc.inode);
 
 			/* TODO: neater interface needed below */
 			local->stbuf.st_mode = local->loc.inode->st_mode;
 
 			dht_selfheal_restore (frame, dht_rmdir_selfheal_cbk,
-					      &local->loc, layout);
+					      &local->loc, local->layout);
 		} else {
                         if (local->loc.parent) {
                                 local->preparent.st_ino =
@@ -3742,7 +3703,7 @@ unlock:
 }
 
 
-int32_t
+int
 dht_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
              struct stat *stbuf, int32_t valid)
 {
@@ -3758,8 +3719,15 @@ dht_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
         VALIDATE_OR_GOTO (loc->inode, err);
         VALIDATE_OR_GOTO (loc->path, err);
 
-	layout = dht_layout_get (this, loc->inode);
+	local = dht_local_init (frame);
+	if (!local) {
+		op_errno = ENOMEM;
+		gf_log (this->name, GF_LOG_DEBUG,
+			"memory allocation failed :(");
+		goto err;
+	}
 
+	local->layout = layout = dht_layout_get (this, loc->inode);
 	if (!layout) {
 		gf_log (this->name, GF_LOG_DEBUG,
 			"no layout for path=%s", loc->path);
@@ -3774,14 +3742,6 @@ dht_setattr (call_frame_t *frame, xlator_t *this, loc_t *loc,
 		goto err;
 	}
 
-	local = dht_local_init (frame);
-	if (!local) {
-		op_errno = ENOMEM;
-		gf_log (this->name, GF_LOG_DEBUG,
-			"memory allocation failed :(");
-		goto err;
-	}
-
 	local->inode = inode_ref (loc->inode);
 	local->call_cnt = layout->cnt;
 
@@ -3802,7 +3762,7 @@ err:
 }
 
 
-int32_t
+int
 dht_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd, struct stat *stbuf,
               int32_t valid)
 {
@@ -3816,7 +3776,15 @@ dht_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd, struct stat *stbuf,
         VALIDATE_OR_GOTO (this, err);
         VALIDATE_OR_GOTO (fd, err);
 
-	layout = dht_layout_get (this, fd->inode);
+	local = dht_local_init (frame);
+	if (!local) {
+		op_errno = ENOMEM;
+		gf_log (this->name, GF_LOG_ERROR,
+			"Out of memory");
+		goto err;
+	}
+
+	local->layout = layout = dht_layout_get (this, fd->inode);
 	if (!layout) {
 		gf_log (this->name, GF_LOG_DEBUG,
 			"no layout for fd=%p", fd);
@@ -3831,14 +3799,6 @@ dht_fsetattr (call_frame_t *frame, xlator_t *this, fd_t *fd, struct stat *stbuf,
 		goto err;
 	}
 
-	local = dht_local_init (frame);
-	if (!local) {
-		op_errno = ENOMEM;
-		gf_log (this->name, GF_LOG_ERROR,
-			"Out of memory");
-		goto err;
-	}
-
 	local->inode = inode_ref (fd->inode);
 	local->call_cnt = layout->cnt;
 
@@ -3871,8 +3831,7 @@ dht_forget (xlator_t *this, inode_t *inode)
 		return 0;
 
 	layout = (dht_layout_t *)(long)tmp_layout;
-	if (!layout->preset)
-		FREE (layout);
+        dht_layout_unref (this, layout);
 
 	return 0;
 }
diff --git a/xlators/cluster/dht/src/dht-common.h b/xlators/cluster/dht/src/dht-common.h
index 62e32cfd8..0666e0ea6 100644
--- a/xlators/cluster/dht/src/dht-common.h
+++ b/xlators/cluster/dht/src/dht-common.h
@@ -36,6 +36,7 @@ struct dht_layout {
 	int               preset;
         int               gen;
 	int               type;
+        int               ref;   /* use with dht_conf_t->layout_lock */
         struct {
 		int       err;   /* 0 = normal
 				   -1 = dir exists and no xattr
@@ -129,6 +130,7 @@ struct dht_conf {
         int32_t        refresh_interval;
         gf_boolean_t   unhashed_sticky_bit;
 	struct timeval last_stat_fetch;
+        gf_lock_t      layout_lock;
 };
 typedef struct dht_conf dht_conf_t;
 
@@ -160,18 +162,22 @@ typedef struct dht_disk_layout dht_disk_layout_t;
 
 #define DHT_STACK_UNWIND(fop, frame, params ...) do {           \
 		dht_local_t *__local = NULL;                    \
+                xlator_t *__xl = NULL;                          \
+                __xl = frame->this;                             \
 		__local = frame->local;                         \
 		frame->local = NULL;                            \
 		STACK_UNWIND_STRICT (fop, frame, params);       \
-		dht_local_wipe (__local);                       \
+		dht_local_wipe (__xl, __local);                 \
 	} while (0)
 
 #define DHT_STACK_DESTROY(frame) do {		       \
 		dht_local_t *__local = NULL;           \
+                xlator_t *__xl = NULL;                 \
+                __xl = frame->this;                    \
 		__local = frame->local;                \
 		frame->local = NULL;		       \
 		STACK_DESTROY (frame->root);	       \
-		dht_local_wipe (__local);	       \
+		dht_local_wipe (__xl, __local);	       \
 	} while (0)
 
 dht_layout_t *dht_layout_new (xlator_t *this, int cnt);
@@ -208,7 +214,7 @@ int dht_itransform (xlator_t *this, xlator_t *subvol, uint64_t x, uint64_t *y);
 int dht_deitransform (xlator_t *this, uint64_t y, xlator_t **subvol,
 		      uint64_t *x);
 
-void dht_local_wipe (dht_local_t *local);
+void dht_local_wipe (xlator_t *this, dht_local_t *local);
 dht_local_t *dht_local_init (call_frame_t *frame);
 int dht_stat_merge (xlator_t *this, struct stat *to, struct stat *from,
 		    xlator_t *subvol);
@@ -245,7 +251,10 @@ int dht_is_subvol_filled (xlator_t *this, xlator_t *subvol);
 xlator_t *dht_free_disk_available_subvol (xlator_t *this, xlator_t *subvol);
 int dht_get_du_info_for_subvol (xlator_t *this, int subvol_idx);
 
-int dht_layout_inode_set (xlator_t *this, xlator_t *subvol, inode_t *inode);
+int dht_layout_preset (xlator_t *this, xlator_t *subvol, inode_t *inode);
+int dht_layout_set (xlator_t *this, inode_t *inode, dht_layout_t *layout);
+void dht_layout_unref (xlator_t *this, dht_layout_t *layout);
+dht_layout_t *dht_layout_ref (xlator_t *this, dht_layout_t *layout);
 xlator_t *dht_first_up_subvol (xlator_t *this);
 
 #endif /* _DHT_H */
diff --git a/xlators/cluster/dht/src/dht-helper.c b/xlators/cluster/dht/src/dht-helper.c
index 3ef528ff2..1b49046ce 100644
--- a/xlators/cluster/dht/src/dht-helper.c
+++ b/xlators/cluster/dht/src/dht-helper.c
@@ -108,7 +108,7 @@ dht_deitransform (xlator_t *this, uint64_t y, xlator_t **subvol_p,
 
 
 void
-dht_local_wipe (dht_local_t *local)
+dht_local_wipe (xlator_t *this, dht_local_t *local)
 {
 	if (!local)
 		return;
@@ -122,8 +122,10 @@ dht_local_wipe (dht_local_t *local)
 	if (local->inode)
 		inode_unref (local->inode);
 
-	if (local->layout)
-		FREE (local->layout);
+	if (local->layout) {
+		dht_layout_unref (this, local->layout);
+                local->layout = NULL;
+        }
 
 	loc_wipe (&local->linkfile.loc);
 
@@ -141,6 +143,11 @@ dht_local_wipe (dht_local_t *local)
 	if (local->xattr_req)
 		dict_unref (local->xattr_req);
 
+        if (local->selfheal.layout) {
+                dht_layout_unref (this, local->selfheal.layout);
+                local->selfheal.layout = NULL;
+        }
+
 	FREE (local);
 }
 
@@ -230,6 +237,10 @@ dht_subvol_get_hashed (xlator_t *this, loc_t *loc)
         }
 
 out:
+        if (layout) {
+                dht_layout_unref (this, layout);
+        }
+
         return subvol;
 }
 
@@ -250,6 +261,10 @@ dht_subvol_get_cached (xlator_t *this, inode_t *inode)
 	subvol = layout->list[0].xlator;
 
 out:
+        if (layout) {
+                dht_layout_unref (this, layout);
+        }
+
         return subvol;
 }
 
diff --git a/xlators/cluster/dht/src/dht-layout.c b/xlators/cluster/dht/src/dht-layout.c
index 7984f2556..4b7b44fc4 100644
--- a/xlators/cluster/dht/src/dht-layout.c
+++ b/xlators/cluster/dht/src/dht-layout.c
@@ -54,6 +54,8 @@ dht_layout_new (xlator_t *this, int cnt)
 	layout->cnt = cnt;
         if (conf)
                 layout->gen = conf->gen;
+
+        layout->ref = 1;
 out:
 	return layout;
 }
@@ -62,12 +64,92 @@ out:
 dht_layout_t *
 dht_layout_get (xlator_t *this, inode_t *inode)
 {
-        uint64_t layout = 0;
-        int      ret    = -1;
+        dht_conf_t   *conf = NULL;
+        uint64_t      layout_int = 0;
+        dht_layout_t *layout = NULL;
+        int           ret    = -1;
+
+        conf = this->private;
+        LOCK (&conf->layout_lock);
+        {
+                ret = inode_ctx_get (inode, this, &layout_int);
+                if (ret == 0) {
+                        layout = (dht_layout_t *) (unsigned long) layout_int;
+                        layout->ref++;
+                }
+        }
+        UNLOCK (&conf->layout_lock);
+
+        return layout;
+}
+
+
+int
+dht_layout_set (xlator_t *this, inode_t *inode, dht_layout_t *layout)
+{
+        dht_conf_t   *conf = NULL;
+        int           oldret = -1;
+        int           ret = 0;
+        dht_layout_t *old_layout;
+        uint64_t      old_layout_int;
+
+        conf = this->private;
+        LOCK (&conf->layout_lock);
+        {
+                oldret = inode_ctx_get (inode, this, &old_layout_int);
+
+                layout->ref++;
+                ret = inode_ctx_put (inode, this, (uint64_t) (unsigned long)
+                                     layout);
+        }
+        UNLOCK (&conf->layout_lock);
+
+        if (oldret == 0) {
+                old_layout = (dht_layout_t *) (unsigned long) old_layout_int;
+                dht_layout_unref (this, old_layout);
+        }
+
+        return ret;
+}
+
+
+void
+dht_layout_unref (xlator_t *this, dht_layout_t *layout)
+{
+        dht_conf_t  *conf = NULL;
+        int          ref = 0;
 
-        ret = inode_ctx_get (inode, this, &layout);
+        if (layout->preset)
+                return;
 
-        return (dht_layout_t *)(long)layout;
+        conf = this->private;
+        LOCK (&conf->layout_lock);
+        {
+                ref = --layout->ref;
+        }
+        UNLOCK (&conf->layout_lock);
+
+        if (!ref)
+                FREE (layout);
+}
+
+
+dht_layout_t *
+dht_layout_ref (xlator_t *this, dht_layout_t *layout)
+{
+        dht_conf_t  *conf = NULL;
+
+        if (layout->preset)
+                return layout;
+
+        conf = this->private;
+        LOCK (&conf->layout_lock);
+        {
+                layout->ref++;
+        }
+        UNLOCK (&conf->layout_lock);
+
+        return layout;
 }
 
 
@@ -599,10 +681,13 @@ out:
 
 
 int
-dht_layout_inode_set (xlator_t *this, xlator_t *subvol, inode_t *inode)
+dht_layout_preset (xlator_t *this, xlator_t *subvol, inode_t *inode)
 {
         dht_layout_t *layout = NULL;
-        int ret = -1;
+        int           ret = -1;
+        dht_conf_t   *conf = NULL;
+
+        conf = this->private;
 
 	layout = dht_layout_for_subvol (this, subvol);
 	if (!layout) {
@@ -613,8 +698,12 @@ dht_layout_inode_set (xlator_t *this, xlator_t *subvol, inode_t *inode)
 		goto out;
 	}
 
-	inode_ctx_put (inode, this, (uint64_t)(long)layout);
-        
+        LOCK (&conf->layout_lock);
+        {
+                inode_ctx_put (inode, this, (uint64_t)(long)layout);
+        }
+        UNLOCK (&conf->layout_lock);
+
         ret = 0;
 out:
         return ret;
diff --git a/xlators/cluster/dht/src/dht-selfheal.c b/xlators/cluster/dht/src/dht-selfheal.c
index f3a92a1fe..a254b0ddc 100644
--- a/xlators/cluster/dht/src/dht-selfheal.c
+++ b/xlators/cluster/dht/src/dht-selfheal.c
@@ -439,7 +439,7 @@ dht_selfheal_new_directory (call_frame_t *frame,
 	local = frame->local;
 
 	local->selfheal.dir_cbk = dir_cbk;
-	local->selfheal.layout = layout;
+	local->selfheal.layout = dht_layout_ref (frame->this, layout);
 
 	dht_layout_sort_volname (layout);
 	dht_selfheal_layout_new_directory (frame, &local->loc, layout);	
@@ -478,7 +478,7 @@ dht_selfheal_directory (call_frame_t *frame, dht_selfheal_dir_cbk_t dir_cbk,
 	misc     = local->selfheal.misc;
 
 	local->selfheal.dir_cbk = dir_cbk;
-	local->selfheal.layout = layout;
+	local->selfheal.layout = dht_layout_ref (this, layout);
 
 	if (down) {
 		gf_log (this->name, GF_LOG_DEBUG,
@@ -526,7 +526,7 @@ dht_selfheal_restore (call_frame_t *frame, dht_selfheal_dir_cbk_t dir_cbk,
 	local = frame->local;
 
 	local->selfheal.dir_cbk = dir_cbk;
-	local->selfheal.layout = layout;
+	local->selfheal.layout = dht_layout_ref (frame->this, layout);
 
 	ret = dht_selfheal_dir_mkdir (frame, loc, layout, 1);
 
diff --git a/xlators/cluster/dht/src/dht.c b/xlators/cluster/dht/src/dht.c
index e43766f99..f49dc495c 100644
--- a/xlators/cluster/dht/src/dht.c
+++ b/xlators/cluster/dht/src/dht.c
@@ -312,6 +312,7 @@ init (xlator_t *this)
         }
 
 	LOCK_INIT (&conf->subvolume_lock);
+	LOCK_INIT (&conf->layout_lock);
 
 	conf->gen = 1;
 
diff --git a/xlators/cluster/dht/src/nufa.c b/xlators/cluster/dht/src/nufa.c
index 545017c77..710d5b95f 100644
--- a/xlators/cluster/dht/src/nufa.c
+++ b/xlators/cluster/dht/src/nufa.c
@@ -33,7 +33,6 @@ nufa_local_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
                        inode_t *inode, struct stat *stbuf, dict_t *xattr,
                        struct stat *postparent)
 {
-	dht_layout_t *layout      = NULL;
         xlator_t     *subvol      = NULL;
         char          is_linkfile = 0;
         char          is_dir      = 0;
@@ -43,6 +42,7 @@ nufa_local_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
         int           i           = 0;
         call_frame_t *prev        = NULL;
 	int           call_cnt    = 0;
+        int           ret         = 0;
 
 
         conf  = this->private;
@@ -71,17 +71,16 @@ nufa_local_lookup_cbk (call_frame_t *frame, void *cookie, xlator_t *this,
 		dht_itransform (this, prev->this, stbuf->st_ino,
 				&stbuf->st_ino);
 
-		layout = dht_layout_for_subvol (this, prev->this);
-		if (!layout) {
+		ret = dht_layout_preset (this, prev->this, inode);
+		if (ret < 0) {
 			gf_log (this->name, GF_LOG_DEBUG,
-				"no pre-set layout for subvolume %s",
+				"could not set pre-set layout for subvol %s",
 				prev->this->name);
 			op_ret   = -1;
 			op_errno = EINVAL;
 			goto err;
 		}
 
-                inode_ctx_put (inode, this, (uint64_t)(long)layout);
                 goto out;
         }
 
@@ -205,7 +204,7 @@ nufa_lookup (call_frame_t *frame, xlator_t *this,
 	local->hashed_subvol = hashed_subvol;
 
         if (is_revalidate (loc)) {
-		layout = dht_layout_get (this, loc->inode);
+		local->layout = layout = dht_layout_get (this, loc->inode);
 
                 if (!layout) {
                         gf_log (this->name, GF_LOG_DEBUG,
@@ -219,8 +218,8 @@ nufa_lookup (call_frame_t *frame, xlator_t *this,
 			gf_log (this->name, GF_LOG_DEBUG,
 				"incomplete layout failure for path=%s",
 				loc->path);
-			op_errno = ESTALE;
-			goto err;
+                        dht_layout_unref (this, local->layout);
+			goto do_fresh_lookup;
 		}
 
 		local->inode    = inode_ref (loc->inode);
@@ -246,6 +245,7 @@ nufa_lookup (call_frame_t *frame, xlator_t *this,
 				break;
 		}
 	} else {
+        do_fresh_lookup:
 		ret = dict_set_uint32 (local->xattr_req, 
 				       "trusted.glusterfs.dht", 4 * 4);