LU-5099 api: transfer object type via dt_insert API

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

diff --git a/lustre/obdclass/local_storage.c b/lustre/obdclass/local_storage.c
index 75e36a8..89fc5df 100644 (file)
--- a/lustre/obdclass/local_storage.c
+++ b/lustre/obdclass/local_storage.c
@@ -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);
        }
@@ -299,6 +301,7 @@ struct dt_object *__local_file_create(const struct lu_env *env,
 {
        struct dt_thread_info   *dti    = dt_info(env);
        struct lu_object_conf   *conf   = &dti->dti_conf;
+       struct dt_insert_rec    *rec    = &dti->dti_dt_rec;
        struct dt_object        *dto;
        struct thandle          *th;
        int                      rc;
@@ -329,7 +332,10 @@ struct dt_object *__local_file_create(const struct lu_env *env,
                dt_declare_ref_add(env, parent, th);
        }
 
-       rc = dt_declare_insert(env, parent, (void *)fid, (void *)name, th);
+       rec->rec_fid = fid;
+       rec->rec_type = dto->do_lu.lo_header->loh_attr;
+       rc = dt_declare_insert(env, parent, (const struct dt_rec *)rec,
+                              (const struct dt_key *)name, th);
        if (rc)
                GOTO(trans_stop, rc);
 
@@ -351,20 +357,27 @@ struct dt_object *__local_file_create(const struct lu_env *env,
        if (dti->dti_dof.dof_type == DFT_DIR) {
                if (!dt_try_as_dir(env, dto))
                        GOTO(destroy, rc = -ENOTDIR);
+
+               rec->rec_type = S_IFDIR;
+               rec->rec_fid = fid;
                /* Add "." and ".." for newly created dir */
-               rc = dt_insert(env, dto, (void *)fid, (void *)".", th,
-                              BYPASS_CAPA, 1);
-               if (rc)
+               rc = dt_insert(env, dto, (const struct dt_rec *)rec,
+                              (const struct dt_key *)".", th, BYPASS_CAPA, 1);
+               if (rc != 0)
                        GOTO(destroy, rc);
+
                dt_ref_add(env, dto, th);
-               rc = dt_insert(env, dto, (void *)lu_object_fid(&parent->do_lu),
-                              (void *)"..", th, BYPASS_CAPA, 1);
-               if (rc)
+               rec->rec_fid = lu_object_fid(&parent->do_lu);
+               rc = dt_insert(env, dto, (const struct dt_rec *)rec,
+                              (const struct dt_key *)"..", th, BYPASS_CAPA, 1);
+               if (rc != 0)
                        GOTO(destroy, rc);
        }
 
+       rec->rec_fid = fid;
+       rec->rec_type = dto->do_lu.lo_header->loh_attr;
        dt_write_lock(env, parent, 0);
-       rc = dt_insert(env, parent, (const struct dt_rec *)fid,
+       rc = dt_insert(env, parent, (const struct dt_rec *)rec,
                       (const struct dt_key *)name, th, BYPASS_CAPA, 1);
        if (dti->dti_dof.dof_type == DFT_DIR)
                dt_ref_add(env, parent, th);
@@ -627,8 +640,11 @@ int local_object_unlink(const struct lu_env *env, struct dt_device *dt,
 
        rc = dt_ref_del(env, dto, th);
        if (rc < 0) {
-               rc = dt_insert(env, parent,
-                              (const struct dt_rec *)&dti->dti_fid,
+               struct dt_insert_rec *rec = &dti->dti_dt_rec;
+
+               rec->rec_fid = &dti->dti_fid;
+               rec->rec_type = dto->do_lu.lo_header->loh_attr;
+               rc = dt_insert(env, parent, (const struct dt_rec *)rec,
                               (const struct dt_key *)name, th, BYPASS_CAPA, 1);
                GOTO(unlock, rc);
        }
@@ -693,17 +709,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);
 
@@ -729,18 +741,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;
 }
 
@@ -824,7 +836,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);
 
@@ -841,11 +859,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);
@@ -894,19 +907,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;
        }