Branch HEAD

b=20082
i=pravin.shelar
i=robert.read

1) for clear orphan case when MDS recovery, we should not care whether
   "precreate" succeed or not, we can just ignore such error (either
   ENOSPC or ETIMEDOUT), and let the succedent create operation to
   process the real no space case.
2) code cleanup

diff --git a/lustre/llite/statahead.c b/lustre/llite/statahead.c
index 417afc2..0d36a3b 100644 (file)
--- a/lustre/llite/statahead.c
+++ b/lustre/llite/statahead.c
@@ -176,7 +176,7 @@ static struct ll_statahead_info *ll_sai_alloc(void)
         return sai;
 }
 
-static inline 
+static inline
 struct ll_statahead_info *ll_sai_get(struct ll_statahead_info *sai)
 {
         LASSERT(sai);
@@ -275,7 +275,7 @@ static void ll_sai_entry_fini(struct ll_statahead_info *sai)
         struct ll_inode_info *lli = ll_i2info(sai->sai_inode);
         struct ll_sai_entry  *entry;
         ENTRY;
-        
+
         spin_lock(&lli->lli_lock);
         sai->sai_index_next++;
         if (likely(!list_empty(&sai->sai_entries_stated))) {
@@ -462,9 +462,7 @@ static int do_statahead_interpret(struct ll_statahead_info *sai)
                 spin_lock(&dcache_lock);
                 lock_dentry(dentry);
                 __d_drop(dentry);
-#ifdef DCACHE_LUSTRE_INVALID
                 dentry->d_flags &= ~DCACHE_LUSTRE_INVALID;
-#endif
                 unlock_dentry(dentry);
                 d_rehash_cond(dentry, 0);
                 spin_unlock(&dcache_lock);
@@ -690,11 +688,7 @@ static int ll_statahead_one(struct dentry *parent, const char* entry_name,
         int                       rc;
         ENTRY;
 
-#ifdef DCACHE_LUSTRE_INVALID
         if (parent->d_flags & DCACHE_LUSTRE_INVALID) {
-#else
-        if (d_unhashed(parent)) {
-#endif
                 CDEBUG(D_READA, "parent dentry@%p %.*s is "
                        "invalid, skip statahead\n",
                        parent, parent->d_name.len, parent->d_name.name);
@@ -1018,11 +1012,14 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry)
                                 continue;
                         }
 
-                        if (target->len == namelen &&
-                            memcmp(target->name, name, namelen) == 0)
-                                rc = LS_FIRST_DE + dot_de;
-                        else
+                        if (target->len != namelen ||
+                            memcmp(target->name, name, namelen) != 0)
                                 rc = LS_NONE_FIRST_DE;
+                        else if (!dot_de)
+                                rc = LS_FIRST_DE;
+                        else
+                                rc = LS_FIRST_DOT_DE;
+
                         ll_put_page(page);
                         GOTO(out, rc);
                 }
@@ -1035,7 +1032,7 @@ static int is_first_dirent(struct inode *dir, struct dentry *dentry)
                         break;
                 } else if (1) {
                         /*
-                         * chain is exhausted 
+                         * chain is exhausted
                          * Normal case: continue to the next page.
                          */
                         page = ll_get_dir_page(dir, pos, 1, &chain);
@@ -1143,7 +1140,7 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp, int lookup)
                 RETURN(rc);
         }
 
-        /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */ 
+        /* I am the "lli_opendir_pid" owner, only me can set "lli_sai". */
         rc = is_first_dirent(dir, *dentryp);
         if (rc == LS_NONE_FIRST_DE)
                 /* It is not "ls -{a}l" operation, no need statahead for it. */
@@ -1189,7 +1186,7 @@ int do_statahead_enter(struct inode *dir, struct dentry **dentryp, int lookup)
                 RETURN(-EAGAIN);
         }
 
-        l_wait_event(sai->sai_thread.t_ctl_waitq, 
+        l_wait_event(sai->sai_thread.t_ctl_waitq,
                      sa_is_running(sai) || sa_is_stopped(sai),
                      &lwi);
 

diff --git a/lustre/osc/osc_create.c b/lustre/osc/osc_create.c
index 48af0a1..9046d7d 100644 (file)
--- a/lustre/osc/osc_create.c
+++ b/lustre/osc/osc_create.c
@@ -513,7 +513,7 @@ int osc_create(struct obd_export *exp, struct obdo *oa,
         struct osc_creator *oscc = &exp->exp_obd->u.cli.cl_oscc;
         struct obd_import  *imp  = exp->exp_obd->u.cli.cl_import;
         struct lov_stripe_md *lsm;
-        int rc = 0;
+        int del_orphan = 0, rc = 0;
         ENTRY;
 
         LASSERT(oa);
@@ -529,7 +529,7 @@ int osc_create(struct obd_export *exp, struct obdo *oa,
                 RETURN(osc_real_create(exp, oa, ea, oti));
 
         /* this is the special case where create removes orphans */
-        if ((oa->o_valid & OBD_MD_FLFLAGS) &&
+        if (oa->o_valid & OBD_MD_FLFLAGS &&
             oa->o_flags == OBD_FL_DELORPHAN) {
                 spin_lock(&oscc->oscc_lock);
                 if (oscc->oscc_flags & OSCC_FLAG_SYNC_IN_PROGRESS) {
@@ -549,6 +549,8 @@ int osc_create(struct obd_export *exp, struct obdo *oa,
                 CDEBUG(D_HA, "%s: oscc recovery started - delete to "LPU64"\n",
                        oscc->oscc_obd->obd_name, oscc->oscc_next_id - 1);
 
+                del_orphan = 1;
+
                 /* delete from next_id on up */
                 oa->o_valid |= OBD_MD_FLID;
                 oa->o_id = oscc->oscc_next_id - 1;
@@ -645,11 +647,16 @@ int osc_create(struct obd_export *exp, struct obdo *oa,
                 spin_unlock(&oscc->oscc_lock);
         }
 
-        if (rc == 0)
+        if (rc == 0) {
                 CDEBUG(D_INFO, "%s: returning objid "LPU64"\n",
                        obd2cli_tgt(oscc->oscc_obd), lsm->lsm_object_id);
-        else if (*ea == NULL)
-                obd_free_memmd(exp, &lsm);
+        } else {
+                if (*ea == NULL)
+                        obd_free_memmd(exp, &lsm);
+                if (del_orphan != 0 && rc != -EIO)
+                        /* Ignore non-IO precreate error for clear orphan */
+                        rc = 0;
+        }
         RETURN(rc);
 }