LU-4721 obdclass: handle local storage init/fini properly

1) In local_oid_storage_fini(), take the mutex on ls_device
   before decreasing the 'los' reference to avoid others to
   obtain the mutex earlier and freed the 'los' by race.

2) When llog init the local stroage for FID_SEQ_LLOG and
   FID_SEQ_LLOG_NAME, it should record the handlers which
   can be used to fini them to avoid releasing the handler
   which is in using by others.

3) NOT forget the llog_ctxt_put() if something wrong during
   the llog_osd_setup().

4) NOT put the object in lastid_compat_check() until all the
   usages on such object have been done.

Test-Parameters: envdefinitions=SLOW=yes,ENABLE_QUOTA=yes testlist=sanity-scrub,sanity-scrub,sanity-scrub,sanity-scrub
Signed-off-by: Fan Yong <fan.yong@intel.com>
Change-Id: Ibf139cf9783d06276a6c0ed764c40dec6e3f70cb
Reviewed-on: http://review.whamcloud.com/9573
Tested-by: Jenkins
Tested-by: Maloo <hpdd-maloo@intel.com>
Reviewed-by: Mike Pershin <mike.pershin@intel.com>
Reviewed-by: Jian Yu <jian.yu@intel.com>
Reviewed-by: Oleg Drokin <oleg.drokin@intel.com>

diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index bb69048..6f490eb 100644 (file)
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -349,6 +349,8 @@ struct llog_ctxt {
        atomic_t                 loc_refcount;
        long                     loc_flags; /* flags, see above defines */
        struct dt_object        *loc_dir;
+       struct local_oid_storage *loc_los_nameless;
+       struct local_oid_storage *loc_los_named;
 };
 
 #define LLOG_PROC_BREAK 0x0001

diff --git a/lustre/mdd/mdd_device.c b/lustre/mdd/mdd_device.c
index 4e863c1..dce2249 100644 (file)
--- a/lustre/mdd/mdd_device.c
+++ b/lustre/mdd/mdd_device.c
@@ -863,8 +863,10 @@ static void mdd_device_shutdown(const struct lu_env *env, struct mdd_device *m,
        mdd_changelog_fini(env, m);
        orph_index_fini(env, m);
        mdd_dot_lustre_cleanup(env, m);
-       if (m->mdd_los != NULL)
+       if (m->mdd_los != NULL) {
                local_oid_storage_fini(env, m->mdd_los);
+               m->mdd_los = NULL;
+       }
        lu_site_purge(env, mdd2lu_dev(m)->ld_site, ~0);
 
        if (m->mdd_child_exp)

diff --git a/lustre/obdclass/llog_osd.c b/lustre/obdclass/llog_osd.c
index ec0c0d6..9bff08e 100644 (file)
--- a/lustre/obdclass/llog_osd.c
+++ b/lustre/obdclass/llog_osd.c
@@ -1096,11 +1096,9 @@ static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
                          struct obd_llog_group *olg, int ctxt_idx,
                          struct obd_device *disk_obd)
 {
-       struct local_oid_storage        *los;
        struct llog_thread_info         *lgi = llog_info(env);
        struct llog_ctxt                *ctxt;
        int                              rc = 0;
-
        ENTRY;
 
        LASSERT(obd);
@@ -1115,44 +1113,41 @@ static int llog_osd_setup(const struct lu_env *env, struct obd_device *obd,
        lgi->lgi_fid.f_oid = 1;
        lgi->lgi_fid.f_ver = 0;
        rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
-                                   &lgi->lgi_fid, &los);
-       if (rc < 0)
-               return rc;
+                                   &lgi->lgi_fid,
+                                   &ctxt->loc_los_nameless);
+       if (rc != 0)
+               GOTO(out, rc);
 
        lgi->lgi_fid.f_seq = FID_SEQ_LLOG_NAME;
        lgi->lgi_fid.f_oid = 1;
        lgi->lgi_fid.f_ver = 0;
        rc = local_oid_storage_init(env, disk_obd->obd_lvfs_ctxt.dt,
-                                   &lgi->lgi_fid, &los);
+                                   &lgi->lgi_fid,
+                                   &ctxt->loc_los_named);
+       if (rc != 0) {
+               local_oid_storage_fini(env, ctxt->loc_los_nameless);
+               ctxt->loc_los_nameless = NULL;
+       }
+
+       GOTO(out, rc);
+
+out:
        llog_ctxt_put(ctxt);
        return rc;
 }
 
 static int llog_osd_cleanup(const struct lu_env *env, struct llog_ctxt *ctxt)
 {
-       struct dt_device                *dt;
-       struct ls_device                *ls;
-       struct local_oid_storage        *los, *nlos;
-
-       LASSERT(ctxt->loc_exp->exp_obd);
-       dt = ctxt->loc_exp->exp_obd->obd_lvfs_ctxt.dt;
-       ls = ls_device_get(dt);
-       if (IS_ERR(ls))
-               RETURN(PTR_ERR(ls));
-
-       mutex_lock(&ls->ls_los_mutex);
-       los = dt_los_find(ls, FID_SEQ_LLOG);
-       nlos = dt_los_find(ls, FID_SEQ_LLOG_NAME);
-       mutex_unlock(&ls->ls_los_mutex);
-       if (los != NULL) {
-               dt_los_put(los);
-               local_oid_storage_fini(env, los);
+       if (ctxt->loc_los_nameless != NULL) {
+               local_oid_storage_fini(env, ctxt->loc_los_nameless);
+               ctxt->loc_los_nameless = NULL;
        }
-       if (nlos != NULL) {
-               dt_los_put(nlos);
-               local_oid_storage_fini(env, nlos);
+
+       if (ctxt->loc_los_named != NULL) {
+               local_oid_storage_fini(env, ctxt->loc_los_named);
+               ctxt->loc_los_named = NULL;
        }
-       ls_device_put(env, ls);
+
        return 0;
 }
 

diff --git a/lustre/obdclass/local_storage.c b/lustre/obdclass/local_storage.c
index f0856c2..fd9f352 100644 (file)
--- a/lustre/obdclass/local_storage.c
+++ b/lustre/obdclass/local_storage.c
@@ -695,17 +695,13 @@ int lastid_compat_check(const struct lu_env *env, struct dt_device *dev,
        rc = dt_lookup_dir(env, root, dti->dti_buf, &dti->dti_fid);
        lu_object_put_nocache(env, &root->do_lu);
        if (rc == -ENOENT) {
-               struct lu_object_conf *conf = &dti->dti_conf;
-
                /* old llog lastid accessed by FID only */
                if (lastid_seq != FID_SEQ_LLOG)
                        return 0;
                dti->dti_fid.f_seq = FID_SEQ_LLOG;
                dti->dti_fid.f_oid = 1;
                dti->dti_fid.f_ver = 0;
-               memset(conf, 0, sizeof(*conf));
-               conf->loc_flags = LOC_F_NEW;
-               o = ls_locate(env, ls, &dti->dti_fid, conf);
+               o = ls_locate(env, ls, &dti->dti_fid, NULL);
                if (IS_ERR(o))
                        return PTR_ERR(o);
 
@@ -731,18 +727,18 @@ int lastid_compat_check(const struct lu_env *env, struct dt_device *dev,
        dt_read_lock(env, o, 0);
        rc = dt_record_read(env, o, &dti->dti_lb, &dti->dti_off);
        dt_read_unlock(env, o);
-       lu_object_put_nocache(env, &o->do_lu);
        if (rc == 0 && le32_to_cpu(losd.lso_magic) != LOS_MAGIC) {
                CERROR("%s: wrong content of seq-"LPX64"-lastid file, magic %x\n",
                       o->do_lu.lo_dev->ld_obd->obd_name, lastid_seq,
                       le32_to_cpu(losd.lso_magic));
-               return -EINVAL;
+               rc = -EINVAL;
        } else if (rc < 0) {
                CERROR("%s: failed to read seq-"LPX64"-lastid: rc = %d\n",
                       o->do_lu.lo_dev->ld_obd->obd_name, lastid_seq, rc);
-               return rc;
        }
-       *first_oid = le32_to_cpu(losd.lso_next_oid);
+       lu_object_put_nocache(env, &o->do_lu);
+       if (rc == 0)
+               *first_oid = le32_to_cpu(losd.lso_next_oid);
        return rc;
 }
 
@@ -897,19 +893,18 @@ out:
 EXPORT_SYMBOL(local_oid_storage_init);
 
 void local_oid_storage_fini(const struct lu_env *env,
-                            struct local_oid_storage *los)
+                           struct local_oid_storage *los)
 {
        struct ls_device *ls;
 
-       if (!atomic_dec_and_test(&los->los_refcount))
-               return;
-
        LASSERT(env);
        LASSERT(los->los_dev);
        ls = dt2ls_dev(los->los_dev);
 
+       /* Take the mutex before decreasing the reference to avoid race
+        * conditions as described in LU-4721. */
        mutex_lock(&ls->ls_los_mutex);
-       if (atomic_read(&los->los_refcount) > 0) {
+       if (!atomic_dec_and_test(&los->los_refcount)) {
                mutex_unlock(&ls->ls_los_mutex);
                return;
        }