Fixes and cleanups in lmv.

[fs/lustre-release.git] / lustre / mds / handler.c

diff --git a/lustre/mds/handler.c b/lustre/mds/handler.c
index 5ffc3f5..dfdad07 100644 (file)
--- a/lustre/mds/handler.c
+++ b/lustre/mds/handler.c
@@ -793,6 +793,33 @@ static int mds_getattr_pack_msg(struct ptlrpc_request *req, struct inode *inode,
         return(rc);
 }
 
+int mds_check_mds_num(struct obd_device *obd, struct inode* inode,
+                      char *name, int namelen)
+{
+        struct mea *mea = NULL;
+        int mea_size, rc = 0;
+        ENTRY;
+                                                                                                                                                                                                     
+        rc = mds_get_lmv_attr(obd, inode, &mea, &mea_size);
+        if (rc)
+                RETURN(rc);
+        if (mea != NULL) {
+                /* dir is already splitted, check is requested filename
+                 * should live at this MDS or at another one */
+                int i;
+                i = mea_name2idx(mea, name, namelen - 1);
+                if (mea->mea_master != i) {
+                        CERROR("inapropriate MDS(%d) for %s. should be %d\n",
+                                mea->mea_master, name, i);
+                        rc = -ERESTART;
+                }
+        }
+                                                                                                                                                                                                     
+        if (mea)
+                OBD_FREE(mea, mea_size);
+        RETURN(rc);
+}
+
 static int mds_getattr_name(int offset, struct ptlrpc_request *req,
                             struct lustre_handle *child_lockh, int child_part)
 {
@@ -808,7 +835,7 @@ static int mds_getattr_name(int offset, struct ptlrpc_request *req,
         char *name;
         ENTRY;
 
-        LASSERT(!strcmp(obd->obd_type->typ_name, "mds"));
+        LASSERT(!strcmp(obd->obd_type->typ_name, LUSTRE_MDS_NAME));
 
         /* Swab now, before anyone looks inside the request */
 
@@ -921,12 +948,19 @@ static int mds_getattr_name(int offset, struct ptlrpc_request *req,
 
         cleanup_phase = 2; /* dchild, dparent, locks */
 
+        /* let's make sure this name should leave on this mds node */
+        rc = mds_check_mds_num(obd, dparent->d_inode, name, namesize);
+        if (rc)
+                GOTO(cleanup, rc);
+
 fill_inode:
+
         if (!DENTRY_VALID(dchild)) {
                 intent_set_disposition(rep, DISP_LOOKUP_NEG);
                 /* in the intent case, the policy clears this error:
                    the disposition is enough */
-                GOTO(cleanup, rc = -ENOENT);
+                rc = -ENOENT;
+                GOTO(cleanup, rc);
         } else {
                 intent_set_disposition(rep, DISP_LOOKUP_POS);
         }
@@ -1335,7 +1369,7 @@ repeat:
         if (IS_ERR(new)) {
                 CERROR("%s: can't lookup new inode (%s) for mkdir: %d\n",
                        obd->obd_name, fidname, (int) PTR_ERR(new));
-                fsfilt_commit(obd, new->d_inode, handle, 0);
+                fsfilt_commit(obd, mds->mds_sb, new->d_inode, handle, 0);
                 up(&parent_inode->i_sem);
                 RETURN(PTR_ERR(new));
         } else if (new->d_inode) {
@@ -2076,33 +2110,59 @@ static int mds_postsetup(struct obd_device *obd)
 err_cleanup:
         mds_lov_clean(obd);
 err_llog:
-        obd_llog_cleanup(llog_get_context(&obd->obd_llogs, LLOG_CONFIG_ORIG_CTXT));
+        obd_llog_cleanup(llog_get_context(&obd->obd_llogs,
+                                          LLOG_CONFIG_ORIG_CTXT));
         RETURN(rc);
 }
 
-static int mds_postrecov(struct obd_device *obd)
-
+int mds_postrecov(struct obd_device *obd)
 {
+        struct mds_obd *mds = &obd->u.mds;
         struct llog_ctxt *ctxt;
-        int rc, rc2;
+        int rc, item = 0;
         ENTRY;
 
         LASSERT(!obd->obd_recovering);
         ctxt = llog_get_context(&obd->obd_llogs, LLOG_UNLINK_ORIG_CTXT);
         LASSERT(ctxt != NULL);
 
+        /* set nextid first, so we are sure it happens */
+        rc = mds_lov_set_nextid(obd);
+        if (rc) {
+                CERROR("%s: mds_lov_set_nextid failed\n", obd->obd_name);
+                GOTO(out, rc);
+        }
+
+        /* clean PENDING dir */
+        rc = mds_cleanup_orphans(obd);
+        if (rc < 0)
+                GOTO(out, rc);
+        item = rc;
+
         rc = llog_connect(ctxt, obd->u.mds.mds_lov_desc.ld_tgt_count,
                           NULL, NULL, NULL);
-        if (rc != 0) {
-                CERROR("faild at llog_origin_connect: %d\n", rc);
+        if (rc) {
+                CERROR("%s: failed at llog_origin_connect: %d\n", 
+                       obd->obd_name, rc);
+                GOTO(out, rc);
         }
 
-        rc = mds_cleanup_orphans(obd);
+        /* remove the orphaned precreated objects */
+        rc = mds_lov_clearorphans(mds, NULL /* all OSTs */);
+        if (rc)
+                GOTO(err_llog, rc);
 
-        rc2 = mds_lov_set_nextid(obd);
-        if (rc2 == 0)
-                rc2 = rc;
-        RETURN(rc2);
+out:
+        RETURN(rc < 0 ? rc : item);
+
+err_llog:
+        /* cleanup all llogging subsystems */
+        rc = obd_llog_finish(obd, &obd->obd_llogs,
+                             mds->mds_lov_desc.ld_tgt_count);
+        if (rc)
+                CERROR("%s: failed to cleanup llogging subsystems\n",
+                        obd->obd_name);
+        goto out;
 }
 
 int mds_lov_clean(struct obd_device *obd)
@@ -2331,6 +2391,7 @@ static int mds_intent_policy(struct ldlm_namespace *ns,
                 break;
         case IT_LOOKUP:
                 getattr_part = MDS_INODELOCK_LOOKUP;
+        case IT_CHDIR:
         case IT_GETATTR:
                 getattr_part |= MDS_INODELOCK_LOOKUP;
         case IT_READDIR: