LU-12702 quota: wait pending write before acquiring remotely

[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 ba29c08..7a8bc16 100644 (file)
--- a/lustre/osd-zfs/osd_handler.c
+++ b/lustre/osd-zfs/osd_handler.c
@@ -97,8 +97,6 @@ static void arc_prune_func(int64_t bytes, void *private)
        struct lu_env      env;
        int rc;
 
-       LASSERT(site->ls_obj_hash);
-
        rc = lu_env_init(&env, LCT_SHRINKER);
        if (rc) {
                CERROR("%s: can't initialize shrinker env: rc = %d\n",
@@ -175,8 +173,8 @@ static void osd_trans_commit_cb(void *cb_data, int error)
 
 static int osd_trans_cb_add(struct thandle *th, struct dt_txn_commit_cb *dcb)
 {
-       struct osd_thandle *oh = container_of0(th, struct osd_thandle,
-                                              ot_super);
+       struct osd_thandle *oh = container_of(th, struct osd_thandle,
+                                             ot_super);
 
        LASSERT(dcb->dcb_magic == TRANS_COMMIT_CB_MAGIC);
        LASSERT(&dcb->dcb_func != NULL);
@@ -200,7 +198,7 @@ static int osd_trans_start(const struct lu_env *env, struct dt_device *d,
 
        ENTRY;
 
-       oh = container_of0(th, struct osd_thandle, ot_super);
+       oh = container_of(th, struct osd_thandle, ot_super);
        LASSERT(oh);
        LASSERT(oh->ot_tx);
 
@@ -285,7 +283,7 @@ static int osd_trans_stop(const struct lu_env *env, struct dt_device *dt,
        int                      rc;
        ENTRY;
 
-       oh = container_of0(th, struct osd_thandle, ot_super);
+       oh = container_of(th, struct osd_thandle, ot_super);
        list_splice_init(&oh->ot_unlinked_list, &unlinked);
 
        osd_oti_get(env)->oti_ins_cache_depth--;
@@ -542,7 +540,7 @@ static int osd_objset_statfs(struct osd_device *osd, struct obd_statfs *osfs)
 
        if (!spa_writeable(dmu_objset_spa(os)) ||
            osd->od_dev_set_rdonly || osd->od_prop_rdonly)
-               osfs->os_state |= OS_STATE_READONLY;
+               osfs->os_state |= OS_STATFS_READONLY;
 
        return 0;
 }
@@ -568,7 +566,7 @@ int osd_statfs(const struct lu_env *env, struct dt_device *d,
        /* ZFS does not support reporting nonrotional status yet, so return
         * flag only if user has set nonrotational.
         */
-       osfs->os_state |= osd->od_nonrotational ? OS_STATE_NONROT : 0;
+       osfs->os_state |= osd->od_nonrotational ? OS_STATFS_NONROT : 0;
 
        RETURN(0);
 }
@@ -644,6 +642,12 @@ static void osd_conf_get(const struct lu_env *env,
                param->ddp_brw_size = osd->od_max_blksz;
        else
                param->ddp_brw_size = ONE_MB_BRW_SIZE;
+
+#ifdef HAVE_DMU_OFFSET_NEXT
+       param->ddp_has_lseek_data_hole = true;
+#else
+       param->ddp_has_lseek_data_hole = false;
+#endif
 }
 
 /*
@@ -695,6 +699,15 @@ static int osd_ro(const struct lu_env *env, struct dt_device *d)
        RETURN(0);
 }
 
+static void osd_wait_quota_pending(struct dt_device *d)
+{
+       struct osd_device  *o = osd_dt_dev(d);
+
+       if (o->od_quota_slave_md != NULL ||
+           o->od_quota_slave_dt != NULL)
+               txg_wait_callbacks(spa_get_dsl(dmu_objset_spa(o->od_os)));
+}
+
 static struct dt_device_operations osd_dt_ops = {
        .dt_root_get            = osd_root_get,
        .dt_statfs              = osd_statfs,
@@ -706,6 +719,7 @@ static struct dt_device_operations osd_dt_ops = {
        .dt_sync                = osd_sync,
        .dt_commit_async        = osd_commit_async,
        .dt_ro                  = osd_ro,
+       .dt_wait_quota_pending  = osd_wait_quota_pending,
 };
 
 /*
@@ -743,7 +757,7 @@ static void osd_key_fini(const struct lu_context *ctx,
 
        if (idc != NULL) {
                LASSERT(info->oti_ins_cache_size > 0);
-               OBD_FREE(idc, sizeof(*idc) * info->oti_ins_cache_size);
+               OBD_FREE_PTR_ARRAY_LARGE(idc, info->oti_ins_cache_size);
                info->oti_ins_cache = NULL;
                info->oti_ins_cache_size = 0;
        }
@@ -1341,10 +1355,8 @@ static struct lu_device *osd_device_free(const struct lu_env *env,
        /* 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);
-       if (!cfs_hash_is_empty(d->ld_site->ls_obj_hash)) {
-               LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
-               lu_site_print(env, d->ld_site, &msgdata, lu_cdebug_printer);
-       }
+       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);
@@ -1359,7 +1371,7 @@ static struct lu_device *osd_device_fini(const struct lu_env *env,
        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) {
@@ -1563,11 +1575,28 @@ static int osd_prepare(const struct lu_env *env, struct lu_device *pdev,
        RETURN(rc);
 }
 
+/**
+ * Implementation of lu_device_operations::ldo_fid_alloc() for OSD
+ *
+ * Allocate FID.
+ *
+ * see include/lu_object.h for the details.
+ */
+static int osd_fid_alloc(const struct lu_env *env, struct lu_device *d,
+                        struct lu_fid *fid, struct lu_object *parent,
+                        const struct lu_name *name)
+{
+       struct osd_device *osd = osd_dev(d);
+
+       return seq_client_alloc_fid(env, osd->od_cl_seq, fid);
+}
+
 struct lu_device_operations osd_lu_ops = {
        .ldo_object_alloc       = osd_object_alloc,
        .ldo_process_config     = osd_process_config,
        .ldo_recovery_complete  = osd_recovery_complete,
        .ldo_prepare            = osd_prepare,
+       .ldo_fid_alloc          = osd_fid_alloc,
 };
 
 static void osd_type_start(struct lu_device_type *t)
@@ -1578,14 +1607,6 @@ static void osd_type_stop(struct lu_device_type *t)
 {
 }
 
-int osd_fid_alloc(const struct lu_env *env, struct obd_export *exp,
-                 struct lu_fid *fid, struct md_op_data *op_data)
-{
-       struct osd_device *osd = osd_dev(exp->exp_obd->obd_lu_dev);
-
-       return seq_client_alloc_fid(env, osd->od_cl_seq, fid);
-}
-
 static struct lu_device_type_operations osd_device_type_ops = {
        .ldto_init              = osd_type_init,
        .ldto_fini              = osd_type_fini,
@@ -1612,7 +1633,6 @@ static const struct obd_ops osd_obd_device_ops = {
        .o_owner       = THIS_MODULE,
        .o_connect      = osd_obd_connect,
        .o_disconnect   = osd_obd_disconnect,
-       .o_fid_alloc    = osd_fid_alloc
 };
 
 static int __init osd_init(void)