LU-3951 lfsck: OST-object inconsistency self detect/repair

[fs/lustre-release.git] / lustre / obdclass / local_storage.c

diff --git a/lustre/obdclass/local_storage.c b/lustre/obdclass/local_storage.c
index 7f41304..f0856c2 100644 (file)
--- a/lustre/obdclass/local_storage.c
+++ b/lustre/obdclass/local_storage.c
@@ -110,7 +110,7 @@ static struct ls_device *__ls_find_dev(struct dt_device *dev)
 
        cfs_list_for_each_entry(ls, &ls_list_head, ls_linkage) {
                if (ls->ls_osd == dev) {
-                       cfs_atomic_inc(&ls->ls_refcount);
+                       atomic_inc(&ls->ls_refcount);
                        ret = ls;
                        break;
                }
@@ -155,7 +155,7 @@ struct ls_device *ls_device_get(struct dt_device *dev)
        if (ls == NULL)
                GOTO(out_ls, ls = ERR_PTR(-ENOMEM));
 
-       cfs_atomic_set(&ls->ls_refcount, 1);
+       atomic_set(&ls->ls_refcount, 1);
        CFS_INIT_LIST_HEAD(&ls->ls_los_list);
        mutex_init(&ls->ls_los_mutex);
 
@@ -176,11 +176,11 @@ out_ls:
 void ls_device_put(const struct lu_env *env, struct ls_device *ls)
 {
        LASSERT(env);
-       if (!cfs_atomic_dec_and_test(&ls->ls_refcount))
+       if (!atomic_dec_and_test(&ls->ls_refcount))
                return;
 
        mutex_lock(&ls_list_mutex);
-       if (cfs_atomic_read(&ls->ls_refcount) == 0) {
+       if (atomic_read(&ls->ls_refcount) == 0) {
                LASSERT(cfs_list_empty(&ls->ls_los_list));
                cfs_list_del(&ls->ls_linkage);
                lu_site_purge(env, ls->ls_top_dev.dd_lu_dev.ld_site, ~0);
@@ -229,8 +229,10 @@ int local_object_declare_create(const struct lu_env *env,
        /* update fid generation file */
        if (los != NULL) {
                LASSERT(dt_object_exists(los->los_obj));
+               dti->dti_lb.lb_buf = NULL;
+               dti->dti_lb.lb_len = sizeof(struct los_ondisk);
                rc = dt_declare_record_write(env, los->los_obj,
-                                            sizeof(struct los_ondisk), 0, th);
+                                            &dti->dti_lb, 0, th);
                if (rc)
                        RETURN(rc);
        }
@@ -297,12 +299,18 @@ struct dt_object *__local_file_create(const struct lu_env *env,
                                      const char *name, struct lu_attr *attr,
                                      struct dt_object_format *dof)
 {
-       struct dt_thread_info   *dti = dt_info(env);
+       struct dt_thread_info   *dti    = dt_info(env);
+       struct lu_object_conf   *conf   = &dti->dti_conf;
        struct dt_object        *dto;
        struct thandle          *th;
        int                      rc;
 
-       dto = ls_locate(env, ls, fid);
+       /* We know that the target object does not exist, to be created,
+        * then give some hints - LOC_F_NEW to help low layer to handle
+        * that efficiently and properly. */
+       memset(conf, 0, sizeof(*conf));
+       conf->loc_flags = LOC_F_NEW;
+       dto = ls_locate(env, ls, fid, conf);
        if (unlikely(IS_ERR(dto)))
                RETURN(dto);
 
@@ -398,7 +406,8 @@ struct dt_object *local_file_find_or_create(const struct lu_env *env,
        rc = dt_lookup_dir(env, parent, name, &dti->dti_fid);
        if (rc == 0)
                /* name is found, get the object */
-               dto = ls_locate(env, dt2ls_dev(los->los_dev), &dti->dti_fid);
+               dto = ls_locate(env, dt2ls_dev(los->los_dev),
+                               &dti->dti_fid, NULL);
        else if (rc != -ENOENT)
                dto = ERR_PTR(rc);
        else {
@@ -485,7 +494,8 @@ struct dt_object *local_index_find_or_create(const struct lu_env *env,
        rc = dt_lookup_dir(env, parent, name, &dti->dti_fid);
        if (rc == 0) {
                /* name is found, get the object */
-               dto = ls_locate(env, dt2ls_dev(los->los_dev), &dti->dti_fid);
+               dto = ls_locate(env, dt2ls_dev(los->los_dev),
+                               &dti->dti_fid, NULL);
        } else if (rc != -ENOENT) {
                dto = ERR_PTR(rc);
        } else {
@@ -642,7 +652,7 @@ struct local_oid_storage *dt_los_find(struct ls_device *ls, __u64 seq)
 
        cfs_list_for_each_entry(los, &ls->ls_los_list, los_list) {
                if (los->los_seq == seq) {
-                       cfs_atomic_inc(&los->los_refcount);
+                       atomic_inc(&los->los_refcount);
                        ret = los;
                        break;
                }
@@ -652,7 +662,7 @@ struct local_oid_storage *dt_los_find(struct ls_device *ls, __u64 seq)
 
 void dt_los_put(struct local_oid_storage *los)
 {
-       if (cfs_atomic_dec_and_test(&los->los_refcount))
+       if (atomic_dec_and_test(&los->los_refcount))
                /* should never happen, only local_oid_storage_fini should
                 * drop refcount to zero */
                LBUG();
@@ -675,7 +685,7 @@ int lastid_compat_check(const struct lu_env *env, struct dt_device *dev,
        if (rc)
                return rc;
 
-       root = ls_locate(env, ls, &dti->dti_fid);
+       root = ls_locate(env, ls, &dti->dti_fid, NULL);
        if (IS_ERR(root))
                return PTR_ERR(root);
 
@@ -685,13 +695,17 @@ 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;
-               o = ls_locate(env, ls, &dti->dti_fid);
+               memset(conf, 0, sizeof(*conf));
+               conf->loc_flags = LOC_F_NEW;
+               o = ls_locate(env, ls, &dti->dti_fid, conf);
                if (IS_ERR(o))
                        return PTR_ERR(o);
 
@@ -705,7 +719,7 @@ int lastid_compat_check(const struct lu_env *env, struct dt_device *dev,
        } else {
                CDEBUG(D_INFO, "Found old lastid file for sequence "LPX64"\n",
                       lastid_seq);
-               o = ls_locate(env, ls, &dti->dti_fid);
+               o = ls_locate(env, ls, &dti->dti_fid, NULL);
                if (IS_ERR(o))
                        return PTR_ERR(o);
        }
@@ -777,10 +791,10 @@ int local_oid_storage_init(const struct lu_env *env, struct dt_device *dev,
        if (*los == NULL)
                GOTO(out, rc = -ENOMEM);
 
-       cfs_atomic_set(&(*los)->los_refcount, 1);
+       atomic_set(&(*los)->los_refcount, 1);
        mutex_init(&(*los)->los_id_lock);
        (*los)->los_dev = &ls->ls_top_dev;
-       cfs_atomic_inc(&ls->ls_refcount);
+       atomic_inc(&ls->ls_refcount);
        cfs_list_add(&(*los)->los_list, &ls->ls_los_list);
 
        /* Use {seq, 0, 0} to create the LAST_ID file for every
@@ -789,7 +803,7 @@ int local_oid_storage_init(const struct lu_env *env, struct dt_device *dev,
        dti->dti_fid.f_seq = fid_seq(first_fid);
        dti->dti_fid.f_oid = LUSTRE_FID_LASTID_OID;
        dti->dti_fid.f_ver = 0;
-       o = ls_locate(env, ls, &dti->dti_fid);
+       o = ls_locate(env, ls, &dti->dti_fid, NULL);
        if (IS_ERR(o))
                GOTO(out_los, rc = PTR_ERR(o));
 
@@ -812,7 +826,13 @@ int local_oid_storage_init(const struct lu_env *env, struct dt_device *dev,
                if (rc)
                        GOTO(out_trans, rc);
 
-               rc = dt_declare_record_write(env, o, sizeof(lastid), 0, th);
+               lastid = cpu_to_le64(first_oid);
+
+               dti->dti_off = 0;
+               dti->dti_lb.lb_buf = &lastid;
+               dti->dti_lb.lb_len = sizeof(lastid);
+               rc = dt_declare_record_write(env, o, &dti->dti_lb, dti->dti_off,
+                                            th);
                if (rc)
                        GOTO(out_trans, rc);
 
@@ -829,11 +849,6 @@ int local_oid_storage_init(const struct lu_env *env, struct dt_device *dev,
                if (rc)
                        GOTO(out_lock, rc);
 
-               lastid = cpu_to_le64(first_oid);
-
-               dti->dti_off = 0;
-               dti->dti_lb.lb_buf = &lastid;
-               dti->dti_lb.lb_len = sizeof(lastid);
                rc = dt_record_write(env, o, &dti->dti_lb, &dti->dti_off, th);
                if (rc)
                        GOTO(out_lock, rc);
@@ -858,7 +873,7 @@ out_trans:
 out_los:
        if (rc != 0) {
                cfs_list_del(&(*los)->los_list);
-               cfs_atomic_dec(&ls->ls_refcount);
+               atomic_dec(&ls->ls_refcount);
                OBD_FREE_PTR(*los);
                *los = NULL;
                if (o != NULL && !IS_ERR(o))
@@ -886,7 +901,7 @@ void local_oid_storage_fini(const struct lu_env *env,
 {
        struct ls_device *ls;
 
-       if (!cfs_atomic_dec_and_test(&los->los_refcount))
+       if (!atomic_dec_and_test(&los->los_refcount))
                return;
 
        LASSERT(env);
@@ -894,12 +909,15 @@ void local_oid_storage_fini(const struct lu_env *env,
        ls = dt2ls_dev(los->los_dev);
 
        mutex_lock(&ls->ls_los_mutex);
-       if (cfs_atomic_read(&los->los_refcount) == 0) {
-               if (los->los_obj)
-                       lu_object_put_nocache(env, &los->los_obj->do_lu);
-               cfs_list_del(&los->los_list);
-               OBD_FREE_PTR(los);
+       if (atomic_read(&los->los_refcount) > 0) {
+               mutex_unlock(&ls->ls_los_mutex);
+               return;
        }
+
+       if (los->los_obj)
+               lu_object_put_nocache(env, &los->los_obj->do_lu);
+       cfs_list_del(&los->los_list);
+       OBD_FREE_PTR(los);
        mutex_unlock(&ls->ls_los_mutex);
        ls_device_put(env, ls);
 }