(1)reworked last_link handling, using lustre_shrink_reply() to remove unused buffer;

(2) never trust the client provided la_mode except for create

diff --git a/lustre/include/lustre_req_layout.h b/lustre/include/lustre_req_layout.h
index d483764..f838833 100644 (file)
--- a/lustre/include/lustre_req_layout.h
+++ b/lustre/include/lustre_req_layout.h
@@ -97,7 +97,6 @@ extern const struct req_format RQF_MDS_SETXATTR;
 extern const struct req_format RQF_MDS_GETATTR;
 extern const struct req_format RQF_MDS_CLOSE;
 extern const struct req_format RQF_MDS_PIN;
-extern const struct req_format RQF_MDS_CLOSE_LAST;
 extern const struct req_format RQF_MDS_CONNECT;
 extern const struct req_format RQF_MDS_DISCONNECT;
 extern const struct req_format RQF_MDS_READPAGE;
@@ -110,10 +109,8 @@ extern const struct req_format RQF_MDS_REINT;
 extern const struct req_format RQF_MDS_REINT_CREATE;
 extern const struct req_format RQF_MDS_REINT_OPEN;
 extern const struct req_format RQF_MDS_REINT_UNLINK;
-extern const struct req_format RQF_MDS_REINT_UNLINK_LAST;
 extern const struct req_format RQF_MDS_REINT_LINK;
 extern const struct req_format RQF_MDS_REINT_RENAME;
-extern const struct req_format RQF_MDS_REINT_RENAME_LAST;
 extern const struct req_format RQF_MDS_REINT_SETATTR;
 extern const struct req_format RQF_LDLM_ENQUEUE;
 extern const struct req_format RQF_LDLM_INTENT;

diff --git a/lustre/llite/namei.c b/lustre/llite/namei.c
index 962575b..37a3ae9 100644 (file)
--- a/lustre/llite/namei.c
+++ b/lustre/llite/namei.c
@@ -836,7 +836,8 @@ static int ll_unlink_generic(struct inode * dir, struct qstr *name)
                name->len, name->name, dir->i_ino, dir->i_generation, dir);
 
         ll_prepare_md_op_data(&op_data, dir, NULL, name->name,
-                              name->len, 0);
+                              name->len,  
+                              0);
         rc = md_unlink(ll_i2sbi(dir)->ll_md_exp, &op_data, &request);
         if (rc)
                 GOTO(out, rc);

diff --git a/lustre/mdd/mdd_handler.c b/lustre/mdd/mdd_handler.c
index bb0f175..7f2eba5 100644 (file)
--- a/lustre/mdd/mdd_handler.c
+++ b/lustre/mdd/mdd_handler.c
@@ -665,6 +665,12 @@ static int mdd_unlink(const struct lu_context *ctxt, struct md_object *pobj,
         int rc;
         ENTRY;
 
+        rc = mdd_attr_get(ctxt, cobj, &ma->ma_attr);
+        if (rc == 0)
+                ma->ma_valid |= MA_INODE;
+        else
+                RETURN(rc);
+
         /* sanity checks */
         if (dt_try_as_dir(ctxt, dt_cobj)) {
                 if (!S_ISDIR(ma->ma_attr.la_mode))
@@ -674,6 +680,13 @@ static int mdd_unlink(const struct lu_context *ctxt, struct md_object *pobj,
                         RETURN(rc = -ENOTDIR);
         }
 
+        if (S_ISREG(ma->ma_attr.la_mode) && ma &&
+            ma->ma_lmm != 0 && ma->ma_lmm_size > 0) {
+                rc = mdd_get_md(ctxt, cobj, ma->ma_lmm, &ma->ma_lmm_size, 0);
+                if (rc > 0)
+                        ma->ma_valid |= MA_LOV;
+        }
+
         mdd_txn_param_build(ctxt, &MDD_TXN_UNLINK);
         handle = mdd_trans_start(ctxt, mdd);
         if (IS_ERR(handle))

diff --git a/lustre/mdd/mdd_internal.h b/lustre/mdd/mdd_internal.h
index 6ed34f6..c9eb3e1 100644 (file)
--- a/lustre/mdd/mdd_internal.h
+++ b/lustre/mdd/mdd_internal.h
@@ -63,6 +63,10 @@ int mdd_lov_set_md(const struct lu_context *ctxt, struct md_object *pobj,
 int mdd_lov_create(const struct lu_context *ctxt, struct mdd_device *mdd,
                    struct mdd_object *child, struct lov_mds_md **lmm,
                    int *lmm_size);
+int mdd_get_md(const struct lu_context *ctxt, struct md_object *obj,
+               void *md, int *md_size, int lock);
+
+
 struct mdd_thread_info *mdd_ctx_info(const struct lu_context *ctx);
 extern struct lu_device_operations mdd_lu_ops;
 static inline int lu_device_is_mdd(struct lu_device *d)

diff --git a/lustre/mdd/mdd_lov.c b/lustre/mdd/mdd_lov.c
index d478777..495cfa7 100644 (file)
--- a/lustre/mdd/mdd_lov.c
+++ b/lustre/mdd/mdd_lov.c
@@ -243,12 +243,13 @@ lcfg_cleanup:
         RETURN(rc);
 }
 
-static int mdd_get_md(const struct lu_context *ctxt, struct md_object *obj,
-                      void *md, int *md_size, int lock)
+int mdd_get_md(const struct lu_context *ctxt, struct md_object *obj,
+               void *md, int *md_size, int lock)
 {
         struct dt_object *next;
         int rc = 0;
         int lmm_size;
+        ENTRY;
 
         next = mdd_object_child(md2mdd_obj(obj));
         rc = next->do_ops->do_xattr_get(ctxt, next, md, *md_size,

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 6434a83..9be8e73 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -336,9 +336,10 @@ skip_packing:
                 repbody->max_cookiesize = info->mti_mdt->mdt_max_cookiesize;
                 repbody->max_mdsize = info->mti_mdt->mdt_max_mdsize;
                 repbody->valid |= OBD_MD_FLMODEASIZE;
-                CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE to : %d:%d\n",
-                        repbody->max_cookiesize,
-                        repbody->max_mdsize);
+                CDEBUG(D_INODE, "I am going to change the MAX_MD_SIZE & MAX_COOKIE"
+                                "to : %d:%d\n",
+                                repbody->max_cookiesize,
+                                repbody->max_mdsize);
         }
 
         if (rc != 0)
@@ -696,7 +697,7 @@ static long mdt_reint_opcode(struct mdt_thread_info *info,
 static int mdt_reint(struct mdt_thread_info *info)
 {
         long opc;
-        int  rc = 0;
+        int  rc;
 
         static const struct req_format *reint_fmts[REINT_MAX] = {
                 [REINT_SETATTR] = &RQF_MDS_REINT_SETATTR,
@@ -713,8 +714,7 @@ static int mdt_reint(struct mdt_thread_info *info)
         if (opc >= 0) {
                 OBD_FAIL_RETURN(OBD_FAIL_MDS_REINT_NET, 0);
 
-                if (opc != REINT_UNLINK)
-                        rc = req_capsule_pack(&info->mti_pill);
+                rc = req_capsule_pack(&info->mti_pill);
                 if (rc == 0)
                         rc = mdt_reint_internal(info, opc);
         } else
@@ -2816,7 +2816,7 @@ DEF_MDT_HNDL_F(HABEO_CORPUS,              GETXATTR,     mdt_getxattr),
 DEF_MDT_HNDL_F(0           |HABEO_REFERO, STATFS,       mdt_statfs),
 DEF_MDT_HNDL_F(0                        |MUTABOR,
                                           REINT,        mdt_reint),
-DEF_MDT_HNDL_F(HABEO_CORPUS,              CLOSE,        mdt_close),
+DEF_MDT_HNDL_F(HABEO_CORPUS|HABEO_REFERO, CLOSE,        mdt_close),
 DEF_MDT_HNDL_0(0,                         DONE_WRITING, mdt_done_writing),
 DEF_MDT_HNDL_F(0           |HABEO_REFERO, PIN,          mdt_pin),
 DEF_MDT_HNDL_0(0,                         SYNC,         mdt_sync),

diff --git a/lustre/mdt/mdt_internal.h b/lustre/mdt/mdt_internal.h
index d813088..4794d9f 100644 (file)
--- a/lustre/mdt/mdt_internal.h
+++ b/lustre/mdt/mdt_internal.h
@@ -359,8 +359,8 @@ int mdt_mfd_close(const struct lu_context *ctxt,
 int mdt_close(struct mdt_thread_info *info);
 
 int mdt_done_writing(struct mdt_thread_info *info);
-int mdt_handle_last_unlink(struct mdt_thread_info *, struct mdt_object *,
-                           int need_get_attr, const struct req_format *);
+void mdt_shrink_reply(struct mdt_thread_info *info);
+int mdt_handle_last_unlink(struct mdt_thread_info *, struct mdt_object *);
 
 
 /* debug issues helper starts here*/

diff --git a/lustre/mdt/mdt_lib.c b/lustre/mdt/mdt_lib.c
index 0b6ae6c..108f07d 100644 (file)
--- a/lustre/mdt/mdt_lib.c
+++ b/lustre/mdt/mdt_lib.c
@@ -40,6 +40,8 @@
 
 #include "mdt_internal.h"
 
+
+/* copied from lov/lov_ea.c, just for debugging, will be removed later */
 static void mdt_dump_lmm(int level, struct lov_mds_md *lmm)
 {
         struct lov_ost_data_v1 *lod;
@@ -59,10 +61,37 @@ static void mdt_dump_lmm(int level, struct lov_mds_md *lmm)
                        le64_to_cpu(lod->l_object_id));
 }
 
+void mdt_shrink_reply(struct mdt_thread_info *info)
+{
+        struct ptlrpc_request *req = mdt_info_req(info);
+        struct mdt_body *body;
+        struct lov_mds_md *lmm;
+        int cookie_size = 0;
+        int md_size = 0;
+
+        body = req_capsule_server_get(&info->mti_pill, &RMF_MDT_BODY);
+
+        if (body && body->valid & OBD_MD_FLEASIZE) {
+                md_size = body->eadatasize;
+        }
+        if (body && body->valid & OBD_MD_FLCOOKIE) {
+                LASSERT(body->valid & OBD_MD_FLEASIZE);
+                lmm = req_capsule_server_get(&info->mti_pill, &RMF_MDT_MD);
+                cookie_size = le32_to_cpu(lmm->lmm_stripe_count) * 
+                                sizeof(struct llog_cookie);
+        }
+
+        CDEBUG(D_INFO, "Shrink to md_size %d cookie_size %d \n", 
+                       md_size, cookie_size);
+
+        lustre_shrink_reply(req, 1, md_size, 1);
+        lustre_shrink_reply(req, md_size? 2:1, cookie_size, 0);
+}
+
+
 /* if object is dying, pack the lov/llog data,
  * parameter info->mti_attr should be valid at this point! */
-int mdt_handle_last_unlink(struct mdt_thread_info *info, struct mdt_object *mo,
-                           int need_get_attr, const struct req_format *fmt)
+int mdt_handle_last_unlink(struct mdt_thread_info *info, struct mdt_object *mo)
 {
         struct mdt_body *repbody;
         struct md_attr  *ma = &info->mti_attr;
@@ -70,54 +99,26 @@ int mdt_handle_last_unlink(struct mdt_thread_info *info, struct mdt_object *mo,
         int              rc = 0;
         ENTRY;
 
-        /* if last unlinked object reference so client should destroy ost
-         * objects*/
+
+        /* if this is the last unlinked object reference,
+         * so client should destroy ost objects*/
         if (S_ISREG(la->la_mode) &&
             la->la_nlink == 0 && mo->mot_header.loh_ref == 1) {
 
-                CDEBUG(D_INODE, "Last reference is released on "DFID3
-                                "need_get_attr = %d for it?\n",
-                                PFID3(mdt_object_fid(mo)),
-                                need_get_attr);
-                /* reply should contains more data,
-                 * * so we need to extend it */
-                req_capsule_extend(&info->mti_pill, fmt);
-                
-                req_capsule_pack(&info->mti_pill);
-                if (need_get_attr) {
-                        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);
-                        rc = mo_attr_get(info->mti_ctxt, 
-                                         mdt_object_child(mo), la);
-                        if (rc == 0) {
-                                ma->ma_valid |= MA_INODE;
-                                rc = mo_xattr_get(info->mti_ctxt,
-                                                  mdt_object_child(mo),
-                                                  ma->ma_lmm,
-                                                  ma->ma_lmm_size,
-                                                  XATTR_NAME_LOV);
-                                if (rc >= 0) {
-                                        ma->ma_lmm_size = rc;
-                                        rc = 0;
-                                        ma->ma_valid |= MA_LOV;
-                                        mdt_dump_lmm(D_ERROR, ma->ma_lmm);
-                                }
-                        }
-                }
+                CDEBUG(D_INODE, "Last reference is released on "DFID3"\n",
+                                PFID3(mdt_object_fid(mo)));
 
+                CDEBUG(D_INODE, "ma_valid = "LPX64"\n", ma->ma_valid);
                 repbody = req_capsule_server_get(&info->mti_pill, 
                                                  &RMF_MDT_BODY);
                 if (ma->ma_valid & MA_INODE)
                         mdt_pack_attr2body(repbody, la, mdt_object_fid(mo));
                 if (ma->ma_lmm_size && ma->ma_valid & MA_LOV) {
+                        mdt_dump_lmm(D_ERROR, ma->ma_lmm);
                         repbody->eadatasize = ma->ma_lmm_size;
                         repbody->valid |= OBD_MD_FLEASIZE;
                 }
-        } else 
-                req_capsule_pack(&info->mti_pill);
+        }
 
         RETURN(rc);
 }

diff --git a/lustre/mdt/mdt_open.c b/lustre/mdt/mdt_open.c
index e17af2c..16d223c 100644 (file)
--- a/lustre/mdt/mdt_open.c
+++ b/lustre/mdt/mdt_open.c
@@ -351,8 +351,11 @@ int mdt_mfd_close(const struct lu_context *ctxt,
 
 int mdt_close(struct mdt_thread_info *info)
 {
+        struct md_attr         *ma = &info->mti_attr;
+        struct lu_attr         *la = &ma->ma_attr;
         struct mdt_export_data *med;
         struct mdt_file_data   *mfd;
+        struct mdt_object      *o;
         int rc;
         ENTRY;
 
@@ -371,11 +374,31 @@ int mdt_close(struct mdt_thread_info *info)
                 list_del_init(&mfd->mfd_list);
                 spin_unlock(&med->med_open_lock);
 
-                rc = mdt_handle_last_unlink(info, mfd->mfd_object, 1,
-                                            &RQF_MDS_CLOSE_LAST);
+                o = mfd->mfd_object;
+                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);
+                rc = mo_attr_get(info->mti_ctxt, mdt_object_child(o), la);
+                if (rc == 0) {
+                        ma->ma_valid |= MA_INODE;
+                        rc = mo_xattr_get(info->mti_ctxt,
+                                          mdt_object_child(o),
+                                          ma->ma_lmm,
+                                          ma->ma_lmm_size,
+                                          XATTR_NAME_LOV);
+                        if (rc >= 0) {
+                                ma->ma_lmm_size = rc;
+                                rc = 0;
+                                ma->ma_valid |= MA_LOV;
+                        }
+                        if (rc == 0)
+                                rc = mdt_handle_last_unlink(info, o);
+                }
 
                 mdt_mfd_close(info->mti_ctxt, mfd);
         }
+        mdt_shrink_reply(info);
         RETURN(rc);
 }
 

diff --git a/lustre/mdt/mdt_reint.c b/lustre/mdt/mdt_reint.c
index 808a951..3d8e117 100644 (file)
--- a/lustre/mdt/mdt_reint.c
+++ b/lustre/mdt/mdt_reint.c
@@ -223,13 +223,13 @@ static int mdt_reint_unlink(struct mdt_thread_info *info)
 {
         struct mdt_reint_record *rr = &info->mti_rr;
         struct ptlrpc_request   *req = mdt_info_req(info);
+        struct md_attr          *ma = &info->mti_attr;
+        struct lu_fid           *child_fid = &info->mti_tmp_fid1;
         struct mdt_object       *mp;
         struct mdt_object       *mc;
         struct mdt_lock_handle  *lhp;
         struct mdt_lock_handle  *lhc;
-        struct lu_fid           *child_fid = &info->mti_tmp_fid1;
         int                      rc;
-
         ENTRY;
 
         DEBUG_REQ(D_INODE, req, "unlink "DFID3"/%s\n", PFID3(rr->rr_fid1),
@@ -246,8 +246,7 @@ static int mdt_reint_unlink(struct mdt_thread_info *info)
 
         if (strlen(rr->rr_name) == 0) {
                 /* remote partial operation */
-                rc = mo_ref_del(info->mti_ctxt, mdt_object_child(mp),
-                                &info->mti_attr);
+                rc = mo_ref_del(info->mti_ctxt, mdt_object_child(mp), ma);
                 GOTO(out_unlock_parent, rc);
         }
 
@@ -268,19 +267,23 @@ static int mdt_reint_unlink(struct mdt_thread_info *info)
 
         /* step 4: delete it */
         /* cmm will take care if child is local or remote */
-        rc = mdo_unlink(info->mti_ctxt, mdt_object_child(mp),
-                        mdt_object_child(mc), rr->rr_name, &info->mti_attr);
+        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);
 
+        rc = mdo_unlink(info->mti_ctxt, mdt_object_child(mp),
+                        mdt_object_child(mc), rr->rr_name, ma);
         if (rc)
                 GOTO(out_unlock_child, rc);
 
-        rc = mdt_handle_last_unlink(info, mc, 0, &RQF_MDS_REINT_UNLINK_LAST);
-        GOTO(out_unlock_child, rc);
+        rc = mdt_handle_last_unlink(info, mc);
 
+        GOTO(out_unlock_child, rc);
 out_unlock_child:
         mdt_object_unlock_put(info, mc, lhc, rc);
 out_unlock_parent:
         mdt_object_unlock_put(info, mp, lhp, rc);
+        mdt_shrink_reply(info);
         return rc;
 }
 

diff --git a/lustre/ptlrpc/layout.c b/lustre/ptlrpc/layout.c
index e94b453..915bf10 100644 (file)
--- a/lustre/ptlrpc/layout.c
+++ b/lustre/ptlrpc/layout.c
@@ -249,10 +249,6 @@ static const struct req_format *req_formats[] = {
         &RQF_MDS_CLOSE,
         &RQF_MDS_PIN,
         &RQF_MDS_READPAGE,
-        &RQF_MDS_REINT_UNLINK_LAST,
-        &RQF_MDS_CLOSE_LAST,
-        &RQF_MDS_REINT_RENAME_LAST
-
 };
 
 struct req_msg_field {
@@ -485,7 +481,7 @@ EXPORT_SYMBOL(RQF_MDS_REINT_OPEN);
 
 const struct req_format RQF_MDS_REINT_UNLINK =
         DEFINE_REQ_FMT0("MDS_REINT_UNLINK", mds_reint_unlink_client,
-                        mdt_body_only);
+                        mds_last_unlink_server);
 EXPORT_SYMBOL(RQF_MDS_REINT_UNLINK);
 
 const struct req_format RQF_MDS_REINT_LINK =
@@ -544,7 +540,8 @@ EXPORT_SYMBOL(RQF_LDLM_INTENT_UNLINK);
 
 const struct req_format RQF_MDS_CLOSE =
         DEFINE_REQ_FMT0("MDS_CLOSE",
-                        mdt_body_only, mdt_body_only);
+//                        mdt_body_only, mdt_body_only);
+                        mdt_body_only, mds_last_unlink_server);
 EXPORT_SYMBOL(RQF_MDS_CLOSE);
 
 const struct req_format RQF_MDS_PIN =
@@ -557,21 +554,6 @@ const struct req_format RQF_MDS_READPAGE =
                         mdt_body_only, mdt_body_only);
 EXPORT_SYMBOL(RQF_MDS_READPAGE);
 
-/* formats for requests which delete last object */
-const struct req_format RQF_MDS_REINT_UNLINK_LAST =
-        DEFINE_REQ_FMT0("MDS_REINT_UNLINK_LAST", mds_reint_unlink_client,
-                        mds_last_unlink_server);
-EXPORT_SYMBOL(RQF_MDS_REINT_UNLINK_LAST);
-
-const struct req_format RQF_MDS_CLOSE_LAST =
-        DEFINE_REQ_FMT0("MDS_CLOSE_LAST", mdt_body_only, mds_last_unlink_server);
-EXPORT_SYMBOL(RQF_MDS_CLOSE_LAST);
-
-const struct req_format RQF_MDS_REINT_RENAME_LAST =
-        DEFINE_REQ_FMT0("MDS_REINT_RENAME_LAST", mds_reint_rename_client,
-                        mds_last_unlink_server);
-EXPORT_SYMBOL(RQF_MDS_REINT_RENAME_LAST);
-
 #if !defined(__REQ_LAYOUT_USER__)
 
 int req_layout_init(void)