implement MDT_FAIL_WRITE(), set the missed fail checks in mdt code

diff --git a/lustre/include/dt_object.h b/lustre/include/dt_object.h
index f1e913d..1c93b6c 100644 (file)
--- a/lustre/include/dt_object.h
+++ b/lustre/include/dt_object.h
@@ -90,6 +90,11 @@ struct dt_device_operations {
         void  (*dt_conf_get)(const struct lu_context *ctx,
                              const struct dt_device *dev,
                              struct dt_device_param *param);
+        /*
+         * used for test purposes to set device RO
+         */
+        void  (*dt_ro_set)(const struct lu_context *ctx,
+                           const struct dt_device *dev);
 };
 
 struct dt_index_features {

diff --git a/lustre/mdd/mdd_handler.c b/lustre/mdd/mdd_handler.c
index 43658e3..ad7a901 100644 (file)
--- a/lustre/mdd/mdd_handler.c
+++ b/lustre/mdd/mdd_handler.c
@@ -905,7 +905,8 @@ int mdd_xattr_set_txn(const struct lu_context *ctxt, struct mdd_object *obj,
 }
 
 static int mdd_xattr_set(const struct lu_context *ctxt, struct md_object *obj,
-                         const void *buf, int buf_len, const char *name, int fl)
+                         const void *buf, int buf_len, const char *name,
+                         int fl)
 {
         struct mdd_device *mdd = mdo2mdd(obj);
         struct thandle *handle;

diff --git a/lustre/mdd/mdd_lov.c b/lustre/mdd/mdd_lov.c
index 799ef93..0ba0347 100644 (file)
--- a/lustre/mdd/mdd_lov.c
+++ b/lustre/mdd/mdd_lov.c
@@ -348,6 +348,8 @@ int mdd_lov_create(const struct lu_context *ctxt, struct mdd_device *mdd,
                         !(create_flags & FMODE_WRITE))
                 RETURN(0);
 
+        OBD_FAIL_RETURN((OBD_FAIL_MDS_ALLOC_OBDO), -ENOMEM);
+
         oa = obdo_alloc();
         if (oa == NULL)
                 RETURN(-ENOMEM);

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 63c7bb7..15e00aa 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -155,6 +155,11 @@ static int mdt_statfs(struct mdt_thread_info *info)
 
         ENTRY;
 
+        /* This will trigger a watchdog timeout */
+        OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_STATFS_LCW_SLEEP,
+                         (MDT_SERVICE_WATCHDOG_TIMEOUT / 1000) + 1);
+        
+
         if (MDT_FAIL_CHECK(OBD_FAIL_MDS_STATFS_PACK)) {
                 result = -ENOMEM;
         } else {
@@ -223,9 +228,9 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
         int                     rc;
         ENTRY;
 
-        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK))
                 RETURN(-ENOMEM);
-        }
+        
         repbody = req_capsule_server_get(pill, &RMF_MDT_BODY);
         repbody->eadatasize = 0;
         repbody->aclsize = 0;
@@ -328,11 +333,7 @@ static int mdt_getattr(struct mdt_thread_info *info)
         if (result)
                 RETURN(result);
 
-        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_GETATTR_PACK)) {
-                result = -ENOMEM;
-        } else {
-                result = mdt_getattr_internal(info, obj);
-        }
+        result = mdt_getattr_internal(info, obj);
         mdt_shrink_reply(info, REPLY_REC_OFF + 1);
         RETURN(result);
 }
@@ -612,6 +613,9 @@ free_rdpg:
                 if (rdpg->rp_pages[i] != NULL)
                         __free_pages(rdpg->rp_pages[i], 0);
         OBD_FREE(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
+
+        MDT_FAIL_RETURN(OBD_FAIL_MDS_SENDPAGE, 0);
+
         return rc;
 }
 
@@ -620,7 +624,8 @@ static int mdt_reint_internal(struct mdt_thread_info *info, __u32 op)
         int rc;
         ENTRY;
 
-        OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_UNPACK, -EFAULT);
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNPACK))
+                RETURN(-EFAULT);
 
         rc = mdt_reint_unpack(info, op);
         if (rc == 0) {
@@ -680,11 +685,11 @@ static int mdt_reint(struct mdt_thread_info *info)
 
         opc = mdt_reint_opcode(info, reint_fmts);
         if (opc >= 0) {
-                OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
-
                 rc = mdt_reint_internal(info, opc);
         } else
                 rc = opc;
+        
+        info->mti_fail_id = OBD_FAIL_MDS_REINT_NET_REP;
         RETURN(rc);
 }
 
@@ -1140,7 +1145,7 @@ static int mdt_req_handle(struct mdt_thread_info *info,
         DEBUG_REQ(D_INODE, req, "%s", h->mh_name);
 
         if (h->mh_fail_id != 0)
-                OBD_FAIL_RETURN(h->mh_fail_id, 0);
+                MDT_FAIL_RETURN(h->mh_fail_id, 0);
 
         result = 0;
         flags = h->mh_flags;
@@ -1356,7 +1361,7 @@ static int mdt_handle0(struct ptlrpc_request *req,
 
         ENTRY;
 
-        OBD_FAIL_RETURN(OBD_FAIL_MDS_ALL_REQUEST_NET | OBD_FAIL_ONCE, 0);
+        MDT_FAIL_RETURN(OBD_FAIL_MDS_ALL_REQUEST_NET | OBD_FAIL_ONCE, 0);
 
         LASSERT(current->journal_info == NULL);
 

diff --git a/lustre/mdt/mdt_internal.h b/lustre/mdt/mdt_internal.h
index 9a2602f..2380428 100644 (file)
--- a/lustre/mdt/mdt_internal.h
+++ b/lustre/mdt/mdt_internal.h
@@ -397,6 +397,18 @@ void mdt_dump_lmm(int level, struct lov_mds_md *lmm);
 
 extern struct lu_context_key       mdt_thread_key;
 /* debug issues helper starts here*/
+static inline void MDT_FAIL_WRITE(const struct lu_context *ctx,
+                                  const struct dt_device *dd, int id)
+{
+        if (OBD_FAIL_CHECK(id)) {
+                CERROR(LUSTRE_MDT0_NAME": obd_fail_loc=%x, fail write ops\n",
+                       id);
+                dd->dd_ops->dt_ro_set(ctx, dd);
+                /* We set FAIL_ONCE because we never "un-fail" a device */
+                obd_fail_loc |= OBD_FAILED | OBD_FAIL_ONCE;
+        }
+}
+
 #define MDT_FAIL_CHECK(id)                                              \
 ({                                                                      \
         if (OBD_FAIL_CHECK(id))                                         \
@@ -404,5 +416,24 @@ extern struct lu_context_key       mdt_thread_key;
         OBD_FAIL_CHECK(id);                                             \
 })
 
+#define MDT_FAIL_CHECK_ONCE(id)                                              \
+({      int _ret_ = 0;                                                       \
+        if (OBD_FAIL_CHECK(id)) {                                            \
+                CERROR(LUSTRE_MDT0_NAME": *** obd_fail_loc=%x ***\n", id);   \
+                obd_fail_loc |= OBD_FAILED;                                  \
+                if ((id) & OBD_FAIL_ONCE)                                    \
+                        obd_fail_loc |= OBD_FAIL_ONCE;                       \
+                _ret_ = 1;                                                   \
+        }                                                                    \
+        _ret_;                                                               \
+})
+
+#define MDT_FAIL_RETURN(id, ret)                                             \
+do {                                                                         \
+        if (MDT_FAIL_CHECK_ONCE(id)) {                                       \
+                RETURN(ret);                                                 \
+        }                                                                    \
+} while(0)
+
 #endif /* __KERNEL__ */
 #endif /* _MDT_H */

diff --git a/lustre/mdt/mdt_open.c b/lustre/mdt/mdt_open.c
index 3f342b2..0e0f0de 100644 (file)
--- a/lustre/mdt/mdt_open.c
+++ b/lustre/mdt/mdt_open.c
@@ -228,6 +228,11 @@ static int mdt_mfd_open(struct mdt_thread_info *info,
         } else if (flags & MDS_OPEN_DIRECTORY)
                 RETURN(-ENOTDIR);
 
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_OPEN_CREATE)) {
+                obd_fail_loc = OBD_FAIL_LDLM_REPLY | OBD_FAIL_ONCE;
+                RETURN(-EAGAIN);
+        }
+
         if (isreg && !(ma->ma_valid & MA_LOV)) {
                 /*No EA, check whether it is will set regEA and dirEA
                  *since in above attr get, these size might be zero,
@@ -415,6 +420,9 @@ int mdt_open(struct mdt_thread_info *info)
         int                     created = 0;
         ENTRY;
 
+        OBD_FAIL_TIMEOUT(OBD_FAIL_MDS_PAUSE_OPEN | OBD_FAIL_ONCE,
+                         (obd_timeout + 1) / 4);
+
         req_capsule_set_size(&info->mti_pill, &RMF_MDT_MD, RCL_SERVER,
                              mdt->mdt_max_mdsize);
 
@@ -457,6 +465,9 @@ int mdt_open(struct mdt_thread_info *info)
                 }
         }
 
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_OPEN_PACK))
+                RETURN(-ENOMEM);
+
         intent_set_disposition(ldlm_rep, DISP_LOOKUP_EXECD);
         if (rr->rr_name[0] == 0) {
                 /* this is cross-ref open */
@@ -631,6 +642,10 @@ int mdt_close(struct mdt_thread_info *info)
                 mdt_object_put(info->mti_ctxt, o);
         }
         mdt_shrink_reply(info, REPLY_REC_OFF + 1);
+
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_CLOSE_PACK))
+                RETURN(-ENOMEM);
+        
         RETURN(rc);
 }
 

diff --git a/lustre/mdt/mdt_recovery.c b/lustre/mdt/mdt_recovery.c
index da2d648..4263f21 100644 (file)
--- a/lustre/mdt/mdt_recovery.c
+++ b/lustre/mdt/mdt_recovery.c
@@ -389,7 +389,7 @@ int mdt_client_add(const struct lu_context *ctx,
                 cl_idx = find_first_zero_bit(bitmap, LR_MAX_CLIENTS);
         repeat:
                 if (cl_idx >= LR_MAX_CLIENTS ||
-                    OBD_FAIL_CHECK_ONCE(OBD_FAIL_MDS_CLIENT_ADD)) {
+                    MDT_FAIL_CHECK_ONCE(OBD_FAIL_MDS_CLIENT_ADD)) {
                         CERROR("no room for clients - fix LR_MAX_CLIENTS\n");
                         return -EOVERFLOW;
                 }

diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index 89e4831..558db93 100644 (file)
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -64,6 +64,9 @@ static int mdt_md_create(struct mdt_thread_info *info)
                 struct md_object *next = mdt_object_child(parent);
 
                 ma->ma_need = MA_INODE;
+                MDT_FAIL_WRITE(info->mti_ctxt, info->mti_mdt->mdt_bottom,
+                               OBD_FAIL_MDS_REINT_CREATE_WRITE);
+
                 rc = mdo_create(info->mti_ctxt, next, rr->rr_name,
                                 mdt_object_child(child), &info->mti_spec,
                                 ma);
@@ -160,6 +163,9 @@ static int mdt_reint_setattr(struct mdt_thread_info *info)
                 GOTO(out_unlock, rc = -ENOENT);
 
         /* all attrs are packed into mti_attr in unpack_setattr */
+        MDT_FAIL_WRITE(info->mti_ctxt, info->mti_mdt->mdt_bottom,
+                       OBD_FAIL_MDS_REINT_SETATTR_WRITE);
+
         rc = mo_attr_set(info->mti_ctxt, next, &info->mti_attr);
         if (rc != 0)
                 GOTO(out_unlock, rc);
@@ -195,6 +201,9 @@ static int mdt_reint_create(struct mdt_thread_info *info)
         rc = req_capsule_pack(&info->mti_pill);
         if (rc)
                 RETURN(rc);
+        
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
+                RETURN(-ESTALE);
 
         switch (info->mti_attr.ma_attr.la_mode & S_IFMT) {
         case S_IFREG:
@@ -211,7 +220,6 @@ static int mdt_reint_create(struct mdt_thread_info *info)
         case S_IFSOCK:{
                 /* special file should stay on the same node as parent */
                 LASSERT(strlen(info->mti_rr.rr_name) > 0);
-
                 rc = mdt_md_create(info);
                 break;
         }
@@ -246,7 +254,8 @@ static int mdt_reint_unlink(struct mdt_thread_info *info)
         if (rc)
                 RETURN(rc);
 
-        /* MDS_CHECK_RESENT here */
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_REINT_UNLINK))
+                RETURN(-ENOENT);
 
         /* step 1: lock the parent */
         lhp = &info->mti_lh[MDT_LH_PARENT];
@@ -277,10 +286,11 @@ static int mdt_reint_unlink(struct mdt_thread_info *info)
 
         /*step 3:  do some checking ...*/
 
+        /* step 4: delete it */
 
+        MDT_FAIL_WRITE(info->mti_ctxt, info->mti_mdt->mdt_bottom,
+                       OBD_FAIL_MDS_REINT_UNLINK_WRITE);
 
-        /* step 4: delete it */
-        /* cmm will take care if child is local or remote */
         ma->ma_lmm = req_capsule_server_get(&info->mti_pill, &RMF_MDT_MD);
         ma->ma_lmm_size = req_capsule_get_size(&info->mti_pill,
                                                &RMF_MDT_MD, RCL_SERVER);
@@ -327,11 +337,13 @@ static int mdt_reint_link(struct mdt_thread_info *info)
         DEBUG_REQ(D_INODE, req, "link original "DFID" to "DFID" %s",
                   PFID(rr->rr_fid1), PFID(rr->rr_fid2), rr->rr_name);
 
-        /* MDS_CHECK_RESENT here */
-
         rc = req_capsule_pack(&info->mti_pill);
         if (rc)
                 RETURN(rc);
+
+        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_REINT_LINK))
+                RETURN(-ENOENT);
+
         /* step 1: lock the source */
         lhs = &info->mti_lh[MDT_LH_PARENT];
         lhs->mlh_mode = LCK_EX;
@@ -354,6 +366,10 @@ static int mdt_reint_link(struct mdt_thread_info *info)
                 GOTO(out_unlock_source, rc = PTR_ERR(mp));
 
         /* step 4: link it */
+
+        MDT_FAIL_WRITE(info->mti_ctxt, info->mti_mdt->mdt_bottom,
+                       OBD_FAIL_MDS_REINT_LINK_WRITE);
+
         rc = mdo_link(info->mti_ctxt, mdt_object_child(mp),
                       mdt_object_child(ms), rr->rr_name);
         GOTO(out_unlock_target, rc);
@@ -546,6 +562,10 @@ static int mdt_reint_rename(struct mdt_thread_info *info)
                 GOTO(out_unlock_new, rc = -EINVAL);
 
         ma->ma_need = MA_INODE | MA_LOV | MA_COOKIE;
+
+        MDT_FAIL_WRITE(info->mti_ctxt, info->mti_mdt->mdt_bottom,
+                       OBD_FAIL_MDS_REINT_RENAME_WRITE);
+
         rc = mdo_rename(info->mti_ctxt, mdt_object_child(msrcdir),
                         mdt_object_child(mtgtdir), old_fid,
                         rr->rr_name, mnew ? mdt_object_child(mnew): NULL,

diff --git a/lustre/mdt/mdt_xattr.c b/lustre/mdt/mdt_xattr.c
index 627a0f7..388d8e9 100644 (file)
--- a/lustre/mdt/mdt_xattr.c
+++ b/lustre/mdt/mdt_xattr.c
@@ -113,9 +113,6 @@ int mdt_getxattr(struct mdt_thread_info *info)
         CDEBUG(D_INODE, "getxattr "DFID"\n",
                         PFID(&info->mti_body->fid1));
 
-        if (MDT_FAIL_CHECK(OBD_FAIL_MDS_GETXATTR_PACK))
-                RETURN(-ENOMEM);
-
         next = mdt_object_child(info->mti_object);
 
         rc = mdt_getxattr_pack_reply(info);
@@ -228,6 +225,8 @@ int mdt_setxattr(struct mdt_thread_info *info)
 
                         if (body->flags & XATTR_CREATE)
                                 flags |= LU_XATTR_CREATE;
+                        MDT_FAIL_WRITE(ctx, info->mti_mdt->mdt_bottom,
+                                       OBD_FAIL_MDS_SETXATTR_WRITE);
 
                         rc = mo_xattr_set(ctx, child, xattr,
                                           xattr_len, xattr_name, flags);

diff --git a/lustre/osd/osd_handler.c b/lustre/osd/osd_handler.c
index 0989261..3ec5457 100644 (file)
--- a/lustre/osd/osd_handler.c
+++ b/lustre/osd/osd_handler.c
@@ -442,6 +442,13 @@ static void osd_conf_get(const struct lu_context *ctx,
         param->ddp_block_shift   = osd_sb(osd_dt_dev(dev))->s_blocksize_bits;
 }
 
+static void osd_ro_set(const struct lu_context *ctx,
+                       const struct dt_device *d)
+{
+        struct osd_device *osd = osd_dt_dev(d);
+        lvfs_set_rdonly(lvfs_sbdev(osd_sb(osd)));
+}
+
 /*
  * Journal
  */
@@ -568,7 +575,8 @@ static struct dt_device_operations osd_dt_ops = {
         .dt_statfs      = osd_statfs,
         .dt_trans_start = osd_trans_start,
         .dt_trans_stop  = osd_trans_stop,
-        .dt_conf_get    = osd_conf_get
+        .dt_conf_get    = osd_conf_get,
+        .dt_ro_set      = osd_ro_set
 };
 
 static void osd_object_read_lock(const struct lu_context *ctx,