LU-6147 lfsck: NOT purge object by OI scrub

Originally, when the OI scrub found some inconsistent FID mapping,
it will repair the FID mapping and ask others to reload the object
by purging such object. Such behavior may cause others to hang.
Because if the object corresponding to the FID has already been
established in RAM, and if some other holds the object's reference,
such as the LFSCK engine will hold the .lustre/lost+found/MDTxxxx,
then purging object will set LU_OBJECT_HEARD_BANSHEE on the object,
then the subsequent object find against such FID will be blocked
until the object's reference become zero and re-establish the object
in RAM. Unfortunately, if it is the object's reference holder tries
to find the same object, it will be blocked by itself for ever.

On the other hand, on the server side, the OI scrub will repair
the bad OI mappping, if the object is established in RAM before
its bad FID mapping repaired, then it must be marked as non-exist,
and should not be cached in RAM after the last reference released.

Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: I651ef5f5e8f4f478f07bcbb5622b345deed7cb31
Reviewed-on: http://review.whamcloud.com/13493
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Andreas Dilger <andreas.dilger@intel.com>
Reviewed-by: Bobi Jam <bobijam@hotmail.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/cl_object.h b/lustre/include/cl_object.h
index 6df788a..a629a9b 100644 (file)
--- a/lustre/include/cl_object.h
+++ b/lustre/include/cl_object.h
@@ -2089,11 +2089,6 @@ static inline struct cl_site *lu2cl_site(const struct lu_site *site)
         return container_of(site, struct cl_site, cs_lu);
 }
 
-static inline int lu_device_is_cl(const struct lu_device *d)
-{
-        return d->ld_type->ldt_tags & LU_DEVICE_CL;
-}
-
 static inline struct cl_device *lu2cl_dev(const struct lu_device *d)
 {
         LASSERT(d == NULL || IS_ERR(d) || lu_device_is_cl(d));

diff --git a/lustre/include/lu_object.h b/lustre/include/lu_object.h
index 6c62b3a..8937a20 100644 (file)
--- a/lustre/include/lu_object.h
+++ b/lustre/include/lu_object.h
@@ -731,8 +731,6 @@ struct lu_object *lu_object_find_at(const struct lu_env *env,
                                     struct lu_device *dev,
                                     const struct lu_fid *f,
                                     const struct lu_object_conf *conf);
-void lu_object_purge(const struct lu_env *env, struct lu_device *dev,
-                    const struct lu_fid *f);
 struct lu_object *lu_object_find_slice(const struct lu_env *env,
                                        struct lu_device *dev,
                                        const struct lu_fid *f,
@@ -1379,5 +1377,15 @@ struct lu_buf *lu_buf_check_and_alloc(struct lu_buf *buf, size_t len);
 extern __u32 lu_context_tags_default;
 extern __u32 lu_session_tags_default;
 
+static inline bool lu_device_is_cl(const struct lu_device *d)
+{
+       return d->ld_type->ldt_tags & LU_DEVICE_CL;
+}
+
+static inline bool lu_object_is_cl(const struct lu_object *o)
+{
+       return lu_device_is_cl(o->lo_dev);
+}
+
 /** @} lu */
 #endif /* __LUSTRE_LU_OBJECT_H */

diff --git a/lustre/lfsck/lfsck_lib.c b/lustre/lfsck/lfsck_lib.c
index 1603099..8b05183 100644 (file)
--- a/lustre/lfsck/lfsck_lib.c
+++ b/lustre/lfsck/lfsck_lib.c
@@ -1329,8 +1329,10 @@ int lfsck_verify_lpf(const struct lu_env *env, struct lfsck_instance *lfsck)
                } else {
                        child1 = lfsck_object_find_by_dev(env, dev,
                                                          &bk->lb_lpf_fid);
-                       if (IS_ERR(child1))
-                               GOTO(put, rc = PTR_ERR(child1));
+                       if (IS_ERR(child1)) {
+                               child1 = NULL;
+                               goto find_child2;
+                       }
 
                        if (unlikely(!dt_object_exists(child1) ||
                                     dt_object_remote(child1)) ||
@@ -1356,6 +1358,7 @@ int lfsck_verify_lpf(const struct lu_env *env, struct lfsck_instance *lfsck)
                }
        }
 
+find_child2:
        snprintf(name, 8, "MDT%04x", node);
        rc = dt_lookup(env, parent, (struct dt_rec *)cfid,
                       (const struct dt_key *)name, BYPASS_CAPA);
@@ -3155,6 +3158,9 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
                        if (IS_ERR(obj))
                                GOTO(out, rc = PTR_ERR(obj));
 
+                       if (unlikely(!dt_try_as_dir(env, obj)))
+                               GOTO(out, rc = -ENOTDIR);
+
                        rc = dt_lookup(env, obj, (struct dt_rec *)fid,
                                (const struct dt_key *)dotlustre, BYPASS_CAPA);
                        if (rc != 0)
@@ -3172,6 +3178,9 @@ int lfsck_register(const struct lu_env *env, struct dt_device *key,
                        if (rc != 0)
                                GOTO(out, rc);
 
+                       if (unlikely(!dt_try_as_dir(env, obj)))
+                               GOTO(out, rc = -ENOTDIR);
+
                        *pfid = *fid;
                        rc = dt_lookup(env, obj, (struct dt_rec *)fid,
                                       (const struct dt_key *)lostfound,

diff --git a/lustre/obdclass/lu_object.c b/lustre/obdclass/lu_object.c
index 2053bd3..3e17c05 100644 (file)
--- a/lustre/obdclass/lu_object.c
+++ b/lustre/obdclass/lu_object.c
@@ -151,7 +151,8 @@ void lu_object_put(const struct lu_env *env, struct lu_object *o)
                         o->lo_ops->loo_object_release(env, o);
         }
 
-        if (!lu_object_is_dying(top)) {
+       if (!lu_object_is_dying(top) &&
+           (lu_object_exists(orig) || lu_object_is_cl(orig))) {
                LASSERT(list_empty(&top->loh_lru));
                list_add_tail(&top->loh_lru, &bkt->lsb_lru);
                 cfs_hash_bd_unlock(site->ls_obj_hash, &bd, 1);
@@ -615,31 +616,6 @@ static struct lu_object *htable_lookup(struct lu_site *s,
        return ERR_PTR(-EAGAIN);
 }
 
-static struct lu_object *htable_lookup_nowait(struct lu_site *s,
-                                             cfs_hash_bd_t *bd,
-                                             const struct lu_fid *f)
-{
-       struct hlist_node       *hnode;
-       struct lu_object_header *h;
-
-       /* cfs_hash_bd_peek_locked is a somehow "internal" function
-        * of cfs_hash, it doesn't add refcount on object. */
-       hnode = cfs_hash_bd_peek_locked(s->ls_obj_hash, bd, (void *)f);
-       if (hnode == NULL) {
-               lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_MISS);
-               return ERR_PTR(-ENOENT);
-       }
-
-       h = container_of0(hnode, struct lu_object_header, loh_hash);
-       if (unlikely(lu_object_is_dying(h)))
-               return ERR_PTR(-ENOENT);
-
-       cfs_hash_get(s->ls_obj_hash, hnode);
-       lprocfs_counter_incr(s->ls_stats, LU_SS_CACHE_HIT);
-       list_del_init(&h->loh_lru);
-       return lu_object_top(h);
-}
-
 /**
  * Search cache for an object with the fid \a f. If such object is found,
  * return it. Otherwise, create new object, insert it into cache and return
@@ -815,30 +791,6 @@ struct lu_object *lu_object_find_at(const struct lu_env *env,
 EXPORT_SYMBOL(lu_object_find_at);
 
 /**
- * Try to find the object in cache without waiting for the dead object
- * to be released nor allocating object if no cached one was found.
- *
- * The found object will be set as LU_OBJECT_HEARD_BANSHEE for purging.
- */
-void lu_object_purge(const struct lu_env *env, struct lu_device *dev,
-                    const struct lu_fid *f)
-{
-       struct lu_site          *s  = dev->ld_site;
-       cfs_hash_t              *hs = s->ls_obj_hash;
-       cfs_hash_bd_t            bd;
-       struct lu_object        *o;
-
-       cfs_hash_bd_get_and_lock(hs, f, &bd, 1);
-       o = htable_lookup_nowait(s, &bd, f);
-       cfs_hash_bd_unlock(hs, &bd, 1);
-       if (!IS_ERR(o)) {
-               set_bit(LU_OBJECT_HEARD_BANSHEE, &o->lo_header->loh_flags);
-               lu_object_put(env, o);
-       }
-}
-EXPORT_SYMBOL(lu_object_purge);
-
-/**
  * Find object with given fid, and return its slice belonging to given device.
  */
 struct lu_object *lu_object_find_slice(const struct lu_env *env,

diff --git a/lustre/osd-ldiskfs/osd_scrub.c b/lustre/osd-ldiskfs/osd_scrub.c
index 91ed16a..a9a191d 100644 (file)
--- a/lustre/osd-ldiskfs/osd_scrub.c
+++ b/lustre/osd-ldiskfs/osd_scrub.c
@@ -582,9 +582,6 @@ iget:
                        sf->sf_items_updated_prior++;
                else
                        sf->sf_items_updated++;
-
-               /* The target has been changed, need to be re-loaded. */
-               lu_object_purge(info->oti_env, osd2lu_dev(dev), fid);
        }
 
        GOTO(out, rc);
@@ -2221,7 +2218,8 @@ static int do_osd_scrub_start(struct osd_device *dev, __u32 flags)
 again:
        if (thread_is_running(thread)) {
                spin_unlock(&scrub->os_lock);
-               if (!scrub->os_partial_scan || flags & SS_AUTO_PARTIAL)
+               if (!(scrub->os_file.sf_flags & SF_AUTO) ||
+                    (flags & (SS_AUTO_FULL | SS_AUTO_PARTIAL)))
                        RETURN(-EALREADY);
 
                osd_scrub_join(dev, flags, false);