LU-617 recovery: setattr from open breaks recovery

The setattr from open(open(O_TRUNC)) is now serialized with
'cl_setattr_lock' on client and goes to a dedicate portal, which is
different with other reint operations, consequently, setattr RPC
can be parallel with other reint RPCs, and that result in the race of
updating last_transno/last_xid on server.

This patch removed the 'cl_setattr_lock' stuff to make all the reint
operations serialized by 'cl_rpc_lock', and the code on server side
which assumes client is holding DLM lock when setattr from open is also
removed, since it's not true.

The MDS_SETATTR_PORTAL service is preserved to keep the compatibility
with old client, and the MDS_SETATTR_FROM_OPEN is also preserved, since
we are using this flag to check write access for open(O_TRUNC), and
it probably can be used for some optimization purpose in future.

Signed-off-by: Niu Yawei <niu@whamcloud.com>
Change-Id: I45f83f8f05022ff0d31f8e7784381821c835785d
Reviewed-on: http://review.whamcloud.com/1654
Tested-by: Hudson
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Oleg Drokin <green@whamcloud.com>
Reviewed-by: Mikhail Pershin <tappro@whamcloud.com>

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 2728674..c6e27ea 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -465,7 +465,6 @@ struct client_obd {
         cfs_waitq_t              cl_destroy_waitq;
 
         struct mdc_rpc_lock     *cl_rpc_lock;
-        struct mdc_rpc_lock     *cl_setattr_lock;
         struct mdc_rpc_lock     *cl_close_lock;
         struct osc_creator       cl_oscc;
 

diff --git a/lustre/mdc/mdc_reint.c b/lustre/mdc/mdc_reint.c
index 89cd5be..296f030 100644 (file)
--- a/lustre/mdc/mdc_reint.c
+++ b/lustre/mdc/mdc_reint.c
@@ -108,12 +108,6 @@ static int mdc_prep_elc_req(struct obd_export *exp, struct ptlrpc_request *req,
                                  0, cancels, count);
 }
 
-/* If mdc_setattr is called with an 'iattr', then it is a normal RPC that
- * should take the normal semaphore and go to the normal portal.
- *
- * If it is called with iattr->ia_valid & ATTR_FROM_OPEN, then it is a
- * magic open-path setattr that should take the setattr semaphore and
- * go to the setattr portal. */
 int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
                 void *ea, int ealen, void *ea2, int ea2len,
                 struct ptlrpc_request **request, struct md_open_data **mod)
@@ -156,13 +150,7 @@ int mdc_setattr(struct obd_export *exp, struct md_op_data *op_data,
                 RETURN(rc);
         }
 
-        if (op_data->op_attr.ia_valid & ATTR_FROM_OPEN) {
-                req->rq_request_portal = MDS_SETATTR_PORTAL;
-                ptlrpc_at_set_req_timeout(req);
-                rpc_lock = obd->u.cli.cl_setattr_lock;
-        } else {
-                rpc_lock = obd->u.cli.cl_rpc_lock;
-        }
+        rpc_lock = obd->u.cli.cl_rpc_lock;
 
         if (op_data->op_attr.ia_valid & (ATTR_MTIME | ATTR_CTIME))
                 CDEBUG(D_INODE, "setting mtime "CFS_TIME_T

diff --git a/lustre/mdc/mdc_request.c b/lustre/mdc/mdc_request.c
index 2f05091..be881f7 100644 (file)
--- a/lustre/mdc/mdc_request.c
+++ b/lustre/mdc/mdc_request.c
@@ -2007,14 +2007,9 @@ static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg)
 
         ptlrpcd_addref();
 
-        OBD_ALLOC(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
-        if (!cli->cl_setattr_lock)
-                GOTO(err_rpc_lock, rc = -ENOMEM);
-        mdc_init_rpc_lock(cli->cl_setattr_lock);
-
         OBD_ALLOC(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
         if (!cli->cl_close_lock)
-                GOTO(err_setattr_lock, rc = -ENOMEM);
+                GOTO(err_rpc_lock, rc = -ENOMEM);
         mdc_init_rpc_lock(cli->cl_close_lock);
 
         rc = client_obd_setup(obd, cfg);
@@ -2037,8 +2032,6 @@ static int mdc_setup(struct obd_device *obd, struct lustre_cfg *cfg)
 
 err_close_lock:
         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
-err_setattr_lock:
-        OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
 err_rpc_lock:
         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
         ptlrpcd_decref();
@@ -2098,7 +2091,6 @@ static int mdc_cleanup(struct obd_device *obd)
         struct client_obd *cli = &obd->u.cli;
 
         OBD_FREE(cli->cl_rpc_lock, sizeof (*cli->cl_rpc_lock));
-        OBD_FREE(cli->cl_setattr_lock, sizeof (*cli->cl_setattr_lock));
         OBD_FREE(cli->cl_close_lock, sizeof (*cli->cl_close_lock));
 
         ptlrpcd_decref();

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 8a57ab0..0a98f73 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -3912,6 +3912,10 @@ static int mdt_start_ptlrpc_service(struct mdt_device *m)
 
         /*
          * setattr service configuration.
+         *
+         * XXX To keep the compatibility with old client(< 2.2), we need to
+         * preserve this portal for a certain time, it should be removed
+         * eventually. LU-617.
          */
         conf = (typeof(conf)) {
                 .psc_nbufs           = MDS_NBUFS,

diff --git a/lustre/mdt/mdt_internal.h b/lustre/mdt/mdt_internal.h
index 1317643..aee5a99 100644 (file)
--- a/lustre/mdt/mdt_internal.h
+++ b/lustre/mdt/mdt_internal.h
@@ -258,7 +258,7 @@ struct mdt_reint_record {
 };
 
 enum mdt_reint_flag {
-        MRF_SETATTR_LOCKED = 1 << 0,
+        MRF_OPEN_TRUNC = 1 << 0,
 };
 
 /*

diff --git a/lustre/mdt/mdt_lib.c b/lustre/mdt/mdt_lib.c
index 9e64d59..0dea2bb 100644 (file)
--- a/lustre/mdt/mdt_lib.c
+++ b/lustre/mdt/mdt_lib.c
@@ -728,7 +728,7 @@ static __u64 mdt_attr_valid_xlate(__u64 in, struct mdt_reint_record *rr,
                 out |= LA_BLOCKS;
 
         if (in & ATTR_FROM_OPEN)
-                rr->rr_flags |= MRF_SETATTR_LOCKED;
+                rr->rr_flags |= MRF_OPEN_TRUNC;
 
         if (in & ATTR_ATIME_SET)
                 out |= LA_ATIME;

diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index 0c7e6da..3c4c011 100644 (file)
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -414,16 +414,12 @@ static int mdt_md_mkobj(struct mdt_thread_info *info)
         RETURN(rc);
 }
 
-/* In the raw-setattr case, we lock the child inode.
- * In the write-back case or if being called from open,
- *               the client holds a lock already.
- * We use the ATTR_FROM_OPEN (translated into MRF_SETATTR_LOCKED by
- * mdt_setattr_unpack()) flag to tell these cases apart. */
 int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
                  struct md_attr *ma, int flags)
 {
         struct mdt_lock_handle  *lh;
         int do_vbr = ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID|LA_FLAGS);
+        __u64 lockpart = MDS_INODELOCK_UPDATE;
         int rc;
         ENTRY;
 
@@ -433,15 +429,12 @@ int mdt_attr_set(struct mdt_thread_info *info, struct mdt_object *mo,
         lh = &info->mti_lh[MDT_LH_PARENT];
         mdt_lock_reg_init(lh, LCK_PW);
 
-        if (!(flags & MRF_SETATTR_LOCKED)) {
-                __u64 lockpart = MDS_INODELOCK_UPDATE;
-                if (ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID))
-                        lockpart |= MDS_INODELOCK_LOOKUP;
+        if (ma->ma_attr.la_valid & (LA_MODE|LA_UID|LA_GID))
+                lockpart |= MDS_INODELOCK_LOOKUP;
 
-                rc = mdt_object_lock(info, mo, lh, lockpart, MDT_LOCAL_LOCK);
-                if (rc != 0)
-                        RETURN(rc);
-        }
+        rc = mdt_object_lock(info, mo, lh, lockpart, MDT_LOCAL_LOCK);
+        if (rc != 0)
+                RETURN(rc);
 
         if (mdt_object_exists(mo) == 0)
                 GOTO(out_unlock, rc = -ENOENT);
@@ -502,7 +495,7 @@ static int mdt_reint_setattr(struct mdt_thread_info *info,
         /* start a log jounal handle if needed */
         if (!(mdt_conn_flags(info) & OBD_CONNECT_SOM)) {
                 if ((ma->ma_attr.la_valid & LA_SIZE) ||
-                    (rr->rr_flags & MRF_SETATTR_LOCKED)) {
+                    (rr->rr_flags & MRF_OPEN_TRUNC)) {
                         /* Check write access for the O_TRUNC case */
                         if (mdt_write_read(mo) < 0)
                                 GOTO(out_put, rc = -ETXTBSY);