LU-15880 quota: fix issues in reserving quota

[fs/lustre-release.git] / lustre / osd-zfs / osd_handler.c

diff --git a/lustre/osd-zfs/osd_handler.c b/lustre/osd-zfs/osd_handler.c
index bd6d8df..7d1b579 100644 (file)
--- a/lustre/osd-zfs/osd_handler.c
+++ b/lustre/osd-zfs/osd_handler.c
@@ -701,21 +701,20 @@ static int osd_ro(const struct lu_env *env, struct dt_device *d)
 /* reserve or free quota for some operation */
 static int osd_reserve_or_free_quota(const struct lu_env *env,
                                     struct dt_device *dev,
-                                    enum quota_type type, __u64 uid,
-                                    __u64 gid, __s64 count, bool is_md)
+                                    struct lquota_id_info *qi)
 {
-       int rc;
        struct osd_device       *osd = osd_dt_dev(dev);
-       struct osd_thread_info  *info = osd_oti_get(env);
-       struct lquota_id_info   *qi = &info->oti_qi;
        struct qsd_instance     *qsd = NULL;
+       int rc;
 
-       if (is_md)
-               qsd = osd->od_quota_slave_md;
-       else
+       ENTRY;
+
+       if (qi->lqi_is_blk)
                qsd = osd->od_quota_slave_dt;
+       else
+               qsd = osd->od_quota_slave_md;
 
-       rc = quota_reserve_or_free(env, qsd, qi, type, uid, gid, count, is_md);
+       rc = qsd_reserve_or_free_quota(env, qsd, qi);
        RETURN(rc);
 }
 
@@ -1191,6 +1190,10 @@ static int osd_mount(const struct lu_env *env,
        if (opts && strstr(opts, "resetoi"))
                resetoi = true;
 
+       rc = lprocfs_init_brw_stats(&o->od_brw_stats);
+       if (rc)
+               GOTO(err, rc);
+
        o->od_in_init = 1;
        rc = osd_scrub_setup(env, o, interval, resetoi);
        o->od_in_init = 0;
@@ -1204,8 +1207,8 @@ static int osd_mount(const struct lu_env *env,
        /* currently it's no need to prepare qsd_instance_md for OST */
        if (!o->od_is_ost) {
                o->od_quota_slave_md = qsd_init(env, o->od_svname,
-                                               &o->od_dt_dev,
-                                               o->od_proc_entry, true);
+                                               &o->od_dt_dev, o->od_proc_entry,
+                                               true, false);
                if (IS_ERR(o->od_quota_slave_md)) {
                        rc = PTR_ERR(o->od_quota_slave_md);
                        o->od_quota_slave_md = NULL;
@@ -1214,7 +1217,7 @@ static int osd_mount(const struct lu_env *env,
        }
 
        o->od_quota_slave_dt = qsd_init(env, o->od_svname, &o->od_dt_dev,
-                                    o->od_proc_entry, false);
+                                       o->od_proc_entry, false, false);
 
        if (IS_ERR(o->od_quota_slave_dt)) {
                if (o->od_quota_slave_md != NULL) {
@@ -1239,12 +1242,10 @@ static int osd_mount(const struct lu_env *env,
                o->od_posix_acl = 1;
 
        osd_unlinked_drain(env, o);
-err:
-       if (rc && o->od_os) {
-               osd_dmu_objset_disown(o->od_os, B_TRUE, o);
-               o->od_os = NULL;
-       }
 
+       RETURN(0);
+
+err:
        RETURN(rc);
 }
 
@@ -1323,7 +1324,59 @@ out:
 }
 
 static struct lu_device *osd_device_fini(const struct lu_env *env,
-                                        struct lu_device *dev);
+                                        struct lu_device *d)
+{
+       struct osd_device *o = osd_dev(d);
+       int                rc;
+
+       ENTRY;
+       osd_index_backup(env, o, false);
+       if (o->od_os) {
+               osd_objset_unregister_callbacks(o);
+               if (!o->od_dt_dev.dd_rdonly) {
+                       osd_sync(env, lu2dt_dev(d));
+                       txg_wait_callbacks(
+                                       spa_get_dsl(dmu_objset_spa(o->od_os)));
+               }
+       }
+
+       /* now with all the callbacks completed we can cleanup the remainings */
+       osd_shutdown(env, o);
+       osd_scrub_cleanup(env, o);
+
+       rc = osd_procfs_fini(o);
+       if (rc) {
+               CERROR("proc fini error %d\n", rc);
+               RETURN(ERR_PTR(rc));
+       }
+
+       if (o->od_os)
+               osd_umount(env, o);
+
+       RETURN(NULL);
+}
+
+
+static struct lu_device *osd_device_free(const struct lu_env *env,
+                                        struct lu_device *d)
+{
+       struct osd_device *o = osd_dev(d);
+
+       ENTRY;
+       /* XXX: make osd top device in order to release reference */
+       if (d->ld_site) {
+               d->ld_site->ls_top_dev = d;
+               lu_site_purge(env, d->ld_site, -1);
+               lu_site_print(env, d->ld_site, &d->ld_site->ls_obj_hash.nelems,
+                             D_ERROR, lu_cdebug_printer);
+       }
+       if (o->od_site.ls_bottom_dev)
+               lu_site_fini(&o->od_site);
+       dt_device_fini(&o->od_dt_dev);
+       OBD_FREE_PTR(o);
+
+       RETURN(NULL);
+}
 
 static struct lu_device *osd_device_alloc(const struct lu_env *env,
                                          struct lu_device_type *type,
@@ -1351,70 +1404,22 @@ static struct lu_device *osd_device_alloc(const struct lu_env *env,
                rc = osd_device_init0(env, dev, cfg);
                if (rc == 0) {
                        rc = osd_mount(env, dev, cfg);
-                       if (rc)
+                       if (rc) {
                                osd_device_fini(env, osd2lu_dev(dev));
-               }
-               if (rc)
+                               osd_device_free(env, osd2lu_dev(dev));
+                               dev = NULL;
+                       }
+               } else {
                        dt_device_fini(&dev->od_dt_dev);
+               }
        }
 
-       if (unlikely(rc != 0))
+       if (unlikely(rc != 0) && dev)
                OBD_FREE_PTR(dev);
 
        return rc == 0 ? osd2lu_dev(dev) : ERR_PTR(rc);
 }
 
-static struct lu_device *osd_device_free(const struct lu_env *env,
-                                        struct lu_device *d)
-{
-       struct osd_device *o = osd_dev(d);
-       ENTRY;
-
-       /* XXX: make osd top device in order to release reference */
-       d->ld_site->ls_top_dev = d;
-       lu_site_purge(env, d->ld_site, -1);
-       lu_site_print(env, d->ld_site, &d->ld_site->ls_obj_hash.nelems,
-                     D_ERROR, lu_cdebug_printer);
-       lu_site_fini(&o->od_site);
-       dt_device_fini(&o->od_dt_dev);
-       OBD_FREE_PTR(o);
-
-       RETURN (NULL);
-}
-
-static struct lu_device *osd_device_fini(const struct lu_env *env,
-                                        struct lu_device *d)
-{
-       struct osd_device *o = osd_dev(d);
-       int                rc;
-       ENTRY;
-
-       osd_index_backup(env, o, false);
-       if (o->od_os) {
-               osd_objset_unregister_callbacks(o);
-               if (!o->od_dt_dev.dd_rdonly) {
-                       osd_sync(env, lu2dt_dev(d));
-                       txg_wait_callbacks(
-                                       spa_get_dsl(dmu_objset_spa(o->od_os)));
-               }
-       }
-
-       /* now with all the callbacks completed we can cleanup the remainings */
-       osd_shutdown(env, o);
-       osd_scrub_cleanup(env, o);
-
-       rc = osd_procfs_fini(o);
-       if (rc) {
-               CERROR("proc fini error %d\n", rc);
-               RETURN(ERR_PTR(rc));
-       }
-
-       if (o->od_os)
-               osd_umount(env, o);
-
-       RETURN(NULL);
-}
-
 static int osd_device_init(const struct lu_env *env, struct lu_device *d,
                            const char *name, struct lu_device *next)
 {