LU-2730 mdt: fix erroneous LASSERT in mdt_reint_opcode

[fs/lustre-release.git] / lustre / mdt / mdt_handler.c

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 3d1febd..02dcdb8 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -686,6 +686,10 @@ static int mdt_getattr_internal(struct mdt_thread_info *info,
 
        if (mdt_object_remote(o)) {
                /* This object is located on remote node.*/
+               /* Return -EIO for old client */
+               if (!mdt_is_dne_client(req->rq_export))
+                       GOTO(out, rc = -EIO);
+
                repbody->fid1 = *mdt_object_fid(o);
                repbody->valid = OBD_MD_FLID | OBD_MD_MDS;
                GOTO(out, rc = 0);
@@ -1042,7 +1046,7 @@ int mdt_swap_layouts(struct mdt_thread_info *info)
        if (IS_ERR(o))
                GOTO(out, rc = PTR_ERR(o));
 
-       if (mdt_object_exists(o) < 0) /* remote object */
+       if (mdt_object_remote(o) || !mdt_object_exists(o)) /* remote object */
                GOTO(put, rc = -ENOENT);
 
        rc = lu_fid_cmp(&info->mti_body->fid1, &info->mti_body->fid2);
@@ -1605,10 +1609,10 @@ static int mdt_sendpage(struct mdt_thread_info *info,
         int                      rc;
         ENTRY;
 
-        desc = ptlrpc_prep_bulk_exp(req, rdpg->rp_npages, BULK_PUT_SOURCE,
-                                    MDS_BULK_PORTAL);
-        if (desc == NULL)
-                RETURN(-ENOMEM);
+       desc = ptlrpc_prep_bulk_exp(req, rdpg->rp_npages, 1, BULK_PUT_SOURCE,
+                                   MDS_BULK_PORTAL);
+       if (desc == NULL)
+               RETURN(-ENOMEM);
 
        if (!(exp_connect_flags(exp) & OBD_CONNECT_BRW_SIZE))
                /* old client requires reply size in it's PAGE_SIZE,
@@ -1661,7 +1665,7 @@ int mdt_readpage(struct mdt_thread_info *info)
        if (exp_connect_flags(info->mti_exp) & OBD_CONNECT_64BITHASH)
                rdpg->rp_attrs |= LUDA_64BITHASH;
        rdpg->rp_count  = min_t(unsigned int, reqbody->nlink,
-                               exp_brw_size(info->mti_exp));
+                               exp_max_brw_size(info->mti_exp));
         rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE - 1) >>
                           CFS_PAGE_SHIFT;
         OBD_ALLOC(rdpg->rp_pages, rdpg->rp_npages * sizeof rdpg->rp_pages[0]);
@@ -1766,24 +1770,28 @@ out_shrink:
 }
 
 static long mdt_reint_opcode(struct mdt_thread_info *info,
-                             const struct req_format **fmt)
-{
-        struct mdt_rec_reint *rec;
-        long opc;
-
-        opc = err_serious(-EFAULT);
-        rec = req_capsule_client_get(info->mti_pill, &RMF_REC_REINT);
-        if (rec != NULL) {
-                opc = rec->rr_opcode;
-                DEBUG_REQ(D_INODE, mdt_info_req(info), "reint opt = %ld", opc);
-                if (opc < REINT_MAX && fmt[opc] != NULL)
-                        req_capsule_extend(info->mti_pill, fmt[opc]);
-                else {
-                        CERROR("Unsupported opc: %ld\n", opc);
-                        opc = err_serious(opc);
-                }
-        }
-        return opc;
+                            const struct req_format **fmt)
+{
+       struct mdt_rec_reint *rec;
+       long opc;
+
+       rec = req_capsule_client_get(info->mti_pill, &RMF_REC_REINT);
+       if (rec != NULL) {
+               opc = rec->rr_opcode;
+               DEBUG_REQ(D_INODE, mdt_info_req(info), "reint opt = %ld", opc);
+               if (opc < REINT_MAX && fmt[opc] != NULL)
+                       req_capsule_extend(info->mti_pill, fmt[opc]);
+               else {
+                       CERROR("%s: Unsupported opcode '%ld' from client '%s': "
+                              "rc = %d\n", mdt_obd_name(info->mti_mdt), opc,
+                              info->mti_mdt->mdt_ldlm_client->cli_name,
+                              -EFAULT);
+                       opc = err_serious(-EFAULT);
+               }
+       } else {
+               opc = err_serious(-EFAULT);
+       }
+       return opc;
 }
 
 int mdt_reint(struct mdt_thread_info *info)
@@ -2095,7 +2103,7 @@ int mdt_obd_idx_read(struct mdt_thread_info *info)
        if (req_ii->ii_count <= 0)
                GOTO(out, rc = -EFAULT);
        rdpg->rp_count = min_t(unsigned int, req_ii->ii_count << LU_PAGE_SHIFT,
-                              exp_brw_size(info->mti_exp));
+                              exp_max_brw_size(info->mti_exp));
        rdpg->rp_npages = (rdpg->rp_count + CFS_PAGE_SIZE -1) >> CFS_PAGE_SHIFT;
 
        /* allocate pages to store the containers */
@@ -3080,6 +3088,7 @@ static void mdt_thread_info_init(struct ptlrpc_request *req,
 
         /* To not check for split by default. */
         info->mti_spec.no_create = 0;
+       info->mti_spec.sp_rm_entry = 0;
 }
 
 static void mdt_thread_info_fini(struct mdt_thread_info *info)
@@ -4678,6 +4687,7 @@ static int mdt_init0(const struct lu_env *env, struct mdt_device *m,
        spin_lock_init(&m->mdt_osfs_lock);
        m->mdt_osfs_age = cfs_time_shift_64(-1000);
        m->mdt_enable_remote_dir = 0;
+       m->mdt_enable_remote_dir_gid = 0;
 
         m->mdt_md_dev.md_lu_dev.ld_ops = &mdt_lu_ops;
         m->mdt_md_dev.md_lu_dev.ld_obd = obd;
@@ -5518,12 +5528,12 @@ static int mdt_fid2path(const struct lu_env *env, struct mdt_device *mdt,
                RETURN(-EINVAL);
        }
 
-       rc = lu_object_exists(&obj->mot_obj.mo_lu);
-       if (rc <= 0) {
-               if (rc == -1)
-                       rc = -EREMOTE;
-               else
-                       rc = -ENOENT;
+       if (mdt_object_remote(obj))
+               rc = -EREMOTE;
+       else if (!mdt_object_exists(obj))
+               rc = -ENOENT;
+
+       if (rc < 0) {
                mdt_object_put(env, obj);
                CDEBUG(D_IOCTL, "nonlocal object "DFID": %d\n",
                        PFID(&fp->gf_fid), rc);