From: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Date: Wed, 9 Jul 2014 13:50:11 +0000 (+0400)
Subject: LU-5285 mdt: reconstruct to handle non-existing objects
X-Git-Tag: 2.6.0-RC1~3
X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=commitdiff_plain;h=13352ee70c2f1b3c382fa1bae1885967fb2cf69b

LU-5285 mdt: reconstruct to handle non-existing objects

it's possible that reconstruction finds the object doesn't exist
because the original transaction was committed and the rep-ack
lock got released letting another clients to destroy the object.
we should handle this propely - return attributes to indicate the
object has been unlinked (nlink=0

Signed-off-by: Alex Zhuravlev <alexey.zhuravlev@intel.com>
Change-Id: I3bc18ff0cdb3b7f6d28e47604cf3e142e3f5bc42
Reviewed-on: http://review.whamcloud.com/11025
Tested-by: Jenkins
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>
---

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 66b0eb4..2e4e8c0 100644
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -683,13 +683,17 @@ int mdt_attr_get_complex(struct mdt_thread_info *info,
 	const struct lu_env *env = info->mti_env;
 	struct md_object    *next = mdt_object_child(o);
 	struct lu_buf       *buf = &info->mti_buf;
-	u32                  mode = lu_object_attr(&next->mo_lu);
 	int                  need = ma->ma_need;
 	int                  rc = 0, rc2;
+	u32                  mode;
 	ENTRY;
 
 	ma->ma_valid = 0;
 
+	if (mdt_object_exists(o) == 0)
+		GOTO(out, rc = -ENOENT);
+	mode = lu_object_attr(&next->mo_lu);
+
 	if (need & MA_INODE) {
 		ma->ma_need = MA_INODE;
 		rc = mo_attr_get(env, next, ma);
diff --git a/lustre/mdt/mdt_recovery.c b/lustre/mdt/mdt_recovery.c
index f3e6884..f5a49da 100644
--- a/lustre/mdt/mdt_recovery.c
+++ b/lustre/mdt/mdt_recovery.c
@@ -234,6 +234,26 @@ void mdt_reconstruct_generic(struct mdt_thread_info *mti,
         return mdt_req_from_lcd(req, ted->ted_lcd);
 }
 
+/**
+ * Generate fake attributes for a non-existing object
+ *
+ * While the client was waiting for the reply, the original transaction
+ * got committed and corresponding rep-ack lock got released, then another
+ * client was able to destroy the object. But we still need to send some
+ * attributes back. So we fake them and set nlink=0, so the client will
+ * be able to detect a non-existing object and drop it from the cache
+ * immediately.
+ *
+ * \param[out] ma	attributes to fill
+ */
+static void mdt_fake_ma(struct md_attr *ma)
+{
+	ma->ma_valid = MA_INODE;
+	memset(&ma->ma_attr, 0, sizeof(ma->ma_attr));
+	ma->ma_attr.la_valid = LA_NLINK;
+	ma->ma_attr.la_mode = S_IFREG;
+}
+
 static void mdt_reconstruct_create(struct mdt_thread_info *mti,
                                    struct mdt_lock_handle *lhc)
 {
@@ -266,7 +286,9 @@ static void mdt_reconstruct_create(struct mdt_thread_info *mti,
         mti->mti_attr.ma_need = MA_INODE;
         mti->mti_attr.ma_valid = 0;
 	rc = mdt_attr_get_complex(mti, child, &mti->mti_attr);
-	if (rc == -EREMOTE) {
+	if (rc == -ENOENT) {
+		mdt_fake_ma(&mti->mti_attr);
+	} else if (rc == -EREMOTE) {
 		/* object was created on remote server */
 		if (!mdt_is_dne_client(exp))
 			/* Return -EIO for old client */
@@ -310,7 +332,10 @@ static void mdt_reconstruct_setattr(struct mdt_thread_info *mti,
 
         mti->mti_attr.ma_need = MA_INODE;
         mti->mti_attr.ma_valid = 0;
-	mdt_attr_get_complex(mti, obj, &mti->mti_attr);
+
+	rc = mdt_attr_get_complex(mti, obj, &mti->mti_attr);
+	if (rc == -ENOENT)
+		mdt_fake_ma(&mti->mti_attr);
         mdt_pack_attr2body(mti, body, &mti->mti_attr.ma_attr,
                            mdt_object_fid(obj));
         if (mti->mti_ioepoch && (mti->mti_ioepoch->flags & MF_EPOCH_OPEN)) {
diff --git a/lustre/tests/replay-dual.sh b/lustre/tests/replay-dual.sh
index 7b73099..7ea4846 100755
--- a/lustre/tests/replay-dual.sh
+++ b/lustre/tests/replay-dual.sh
@@ -861,6 +861,23 @@ test_23d () {
 }
 run_test 23d "c1 rmdir d1, M0 drop update reply and fail M0/M1, c2 mkdir d1"
 
+test_24 () {
+	touch $MOUNT/$tfile
+	stat $MOUNT/$tfile >&/dev/null
+# OBD_FAIL_MDS_REINT_NET_REP
+	do_facet $SINGLEMDS $LCTL set_param fail_loc=0x119
+	$TRUNCATE $MOUNT/$tfile 100 &
+	PID=$!
+	sleep 1
+	do_facet $SINGLEMDS lctl set_param fail_loc=0
+	# sync to release rep-ack lock quickly
+	do_nodes $(comma_list $(mdts_nodes)) \
+	    "lctl set_param -n osd*.*MDT*.force_sync 1"
+	rm $MOUNT2/$tfile
+	wait
+}
+run_test 24 "reconstruct on non-existing object"
+
 # end commit on sharing tests 
 
 test_24() {