b=22107 a fix for the warning message in filter_destroy_internal

[fs/lustre-release.git] / lustre / obdfilter / filter.c

diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index 83453db..a498a66 100644 (file)
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -290,7 +290,6 @@ static int filter_export_stats_init(struct obd_device *obd,
 
         RETURN(0);
  clean:
-        lprocfs_exp_cleanup(exp);
         return rc;
 }
 
@@ -881,9 +880,7 @@ static int filter_init_server_data(struct obd_device *obd, struct file * filp)
                 exp->exp_connecting = 0;
                 exp->exp_in_recovery = 0;
                 cfs_spin_unlock(&exp->exp_lock);
-                cfs_spin_lock_bh(&obd->obd_processing_task_lock);
                 obd->obd_max_recoverable_clients++;
-                cfs_spin_unlock_bh(&obd->obd_processing_task_lock);
                 class_export_put(exp);
 
                 if (last_rcvd > le64_to_cpu(lsd->lsd_last_transno))
@@ -1487,8 +1484,8 @@ struct dentry *filter_fid2dentry(struct obd_device *obd,
         if (dir_dentry == NULL)
                 filter_parent_unlock(dparent);
         if (IS_ERR(dchild)) {
-                CERROR("%s: child lookup error %ld\n", obd->obd_name,
-                       PTR_ERR(dchild));
+                CERROR("%s: object "LPU64":"LPU64" lookup error: rc %ld\n",
+                       obd->obd_name, id, group, PTR_ERR(dchild));
                 RETURN(dchild);
         }
 
@@ -1598,7 +1595,10 @@ static int filter_destroy_internal(struct obd_device *obd, obd_id objid,
         struct inode *inode = dchild->d_inode;
         int rc;
 
-        if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 1) {
+        /* There should be 2 references to the inode:
+         *  1) taken by filter_prepare_destroy
+         *  2) taken by filter_destroy */
+        if (inode->i_nlink != 1 || atomic_read(&inode->i_count) != 2) {
                 CERROR("destroying objid %.*s ino %lu nlink %lu count %d\n",
                        dchild->d_name.len, dchild->d_name.name, inode->i_ino,
                        (unsigned long)inode->i_nlink,
@@ -1702,7 +1702,7 @@ static int filter_intent_policy(struct ldlm_namespace *ns,
          * lock, and should not be granted if the lock will be blocked.
          */
 
-        LASSERT(ns == res->lr_namespace);
+        LASSERT(ns == ldlm_res_to_ns(res));
         lock_res(res);
         rc = policy(lock, &tmpflags, 0, &err, &rpc_list);
         check_res_locked(res);
@@ -1800,13 +1800,6 @@ static int filter_intent_policy(struct ldlm_namespace *ns,
 
         LASSERTF(l->l_glimpse_ast != NULL, "l == %p", l);
         rc = l->l_glimpse_ast(l, NULL); /* this will update the LVB */
-        /* Update the LVB from disk if the AST failed (this is a legal race) */
-        /*
-         * XXX nikita: situation when ldlm_server_glimpse_ast() failed before
-         * sending ast is not handled. This can result in lost client writes.
-         */
-        if (rc != 0)
-                ldlm_res_lvbo_update(res, NULL, 1);
 
         lock_res(res);
         *reply_lvb = *res_lvb;
@@ -2042,6 +2035,7 @@ int filter_common_setup(struct obd_device *obd, struct lustre_cfg* lcfg,
         CFS_INIT_LIST_HEAD(&filter->fo_export_list);
         cfs_sema_init(&filter->fo_alloc_lock, 1);
         init_brw_stats(&filter->fo_filter_stats);
+        cfs_spin_lock_init(&filter->fo_flags_lock);
         filter->fo_read_cache = 1; /* enable read-only cache by default */
         filter->fo_writethrough_cache = 1; /* enable writethrough cache */
         filter->fo_readcache_max_filesize = FILTER_MAX_CACHE_SIZE;
@@ -2511,9 +2505,9 @@ static int filter_llog_connect(struct obd_export *exp,
               obd->obd_name, body->lgdc_logid.lgl_oid,
               body->lgdc_logid.lgl_oseq, body->lgdc_logid.lgl_ogen);
 
-        cfs_spin_lock_bh(&obd->obd_processing_task_lock);
+        cfs_spin_lock(&obd->u.filter.fo_flags_lock);
         obd->u.filter.fo_mds_ost_sync = 1;
-        cfs_spin_unlock_bh(&obd->obd_processing_task_lock);
+        cfs_spin_unlock(&obd->u.filter.fo_flags_lock);
         rc = llog_connect(ctxt, &body->lgdc_logid,
                           &body->lgdc_gen, NULL);
         llog_ctxt_put(ctxt);
@@ -2809,7 +2803,6 @@ static int filter_connect(const struct lu_env *env,
 cleanup:
         if (rc) {
                 class_disconnect(lexp);
-                lprocfs_exp_cleanup(lexp);
                 *exp = NULL;
         } else {
                 *exp = lexp;
@@ -3193,6 +3186,7 @@ int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry,
                         *fcc = oa->o_lcookie;
         }
         if (ia_valid & (ATTR_SIZE | ATTR_UID | ATTR_GID)) {
+                unsigned long now = jiffies;
                 DQUOT_INIT(inode);
                 /* Filter truncates and writes are serialized by
                  * i_alloc_sem, see the comment in
@@ -3200,6 +3194,7 @@ int filter_setattr_internal(struct obd_export *exp, struct dentry *dentry,
                 if (ia_valid & ATTR_SIZE)
                         down_write(&inode->i_alloc_sem);
                 LOCK_INODE_MUTEX(inode);
+                fsfilt_check_slow(exp->exp_obd, now, "i_alloc_sem and i_mutex");
                 old_size = i_size_read(inode);
         }
 
@@ -3396,7 +3391,9 @@ int filter_setattr(struct obd_export *exp, struct obd_info *oinfo,
          */
         if (oa->o_valid &
             (OBD_MD_FLMTIME | OBD_MD_FLATIME | OBD_MD_FLCTIME)) {
+                unsigned long now = jiffies;
                 down_write(&dentry->d_inode->i_alloc_sem);
+                fsfilt_check_slow(exp->exp_obd, now, "i_alloc_sem");
                 fmd = filter_fmd_get(exp, oa->o_id, oa->o_seq);
                 if (fmd && fmd->fmd_mactime_xid < oti->oti_xid)
                         fmd->fmd_mactime_xid = oti->oti_xid;
@@ -3542,7 +3539,14 @@ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
                 filter_set_last_id(filter, id, doa.o_seq);
                 rc = filter_update_last_objid(exp->exp_obd, doa.o_seq, 1);
         } else {
-                /* don't reuse orphan object, return last used objid */
+                /*
+                 * We have destroyed orphan objects, but don't want to reuse
+                 * them. Therefore we don't reset last_id to the last created
+                 * objects. Instead, we report back to the MDS the object id
+                 * of the last orphan, so that the MDS can restart allocating
+                 * objects from this id + 1 and thus skip the whole orphan
+                 * object id range
+                 */
                 oa->o_id = last;
                 rc = 0;
         }
@@ -4009,6 +4013,7 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa,
         struct llog_cookie *fcc = NULL;
         int rc, rc2, cleanup_phase = 0, sync = 0;
         struct iattr iattr;
+        unsigned long now;
         ENTRY;
 
         rc = filter_auth_capa(exp, NULL, oa->o_seq,
@@ -4077,8 +4082,10 @@ int filter_destroy(struct obd_export *exp, struct obdo *oa,
          * between page lock, i_mutex & starting new journal handle.
          * (see bug 20321) -johann
          */
+        now = jiffies;
         down_write(&dchild->d_inode->i_alloc_sem);
         LOCK_INODE_MUTEX(dchild->d_inode);
+        fsfilt_check_slow(exp->exp_obd, now, "i_alloc_sem and i_mutex");
 
         /* VBR: version recovery check */
         rc = filter_version_get_check(exp, oti, dchild->d_inode);
@@ -4474,7 +4481,7 @@ int filter_iocontrol(unsigned int cmd, struct obd_export *exp,
         switch (cmd) {
         case OBD_IOC_ABORT_RECOVERY: {
                 LCONSOLE_WARN("%s: Aborting recovery.\n", obd->obd_name);
-                target_recovery_fini(obd);
+                target_stop_recovery_thread(obd);
                 RETURN(0);
         }