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 (file)
--- 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 (file)
--- 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 (executable)
--- 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() {