LU-10769 osd-zfs: fix deadlock on osd_object::oo_guard

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

diff --git a/lustre/osd-zfs/osd_object.c b/lustre/osd-zfs/osd_object.c
index ec4f52d..0734da2 100644 (file)
--- a/lustre/osd-zfs/osd_object.c
+++ b/lustre/osd-zfs/osd_object.c
@@ -186,8 +186,8 @@ osd_object_sa_bulk_update(struct osd_object *obj, sa_bulk_attr_t *attrs,
 /*
  * Retrieve the attributes of a DMU object
  */
-int __osd_object_attr_get(const struct lu_env *env, struct osd_device *o,
-                         struct osd_object *obj, struct lu_attr *la)
+static int __osd_object_attr_get(const struct lu_env *env, struct osd_device *o,
+                                struct osd_object *obj, struct lu_attr *la)
 {
        struct osa_attr *osa = &osd_oti_get(env)->oti_osa;
        sa_bulk_attr_t  *bulk = osd_oti_get(env)->oti_attr_bulk;
@@ -406,7 +406,7 @@ out:
 /*
  * Concurrency: shouldn't matter.
  */
-int osd_object_init0(const struct lu_env *env, struct osd_object *obj)
+static int osd_object_init0(const struct lu_env *env, struct osd_object *obj)
 {
        struct osd_device       *osd = osd_obj2dev(obj);
        const struct lu_fid     *fid = lu_object_fid(&obj->oo_dt.do_lu);
@@ -1037,7 +1037,7 @@ static inline int qsd_transfer(const struct lu_env *env,
                               struct qsd_instance *qsd,
                               struct lquota_trans *trans, int qtype,
                               __u64 orig_id, __u64 new_id, __u64 bspace,
-                              struct lquota_id_info *qi)
+                              struct lquota_id_info *qi, bool ignore_edquot)
 {
        int     rc;
 
@@ -1054,7 +1054,7 @@ static inline int qsd_transfer(const struct lu_env *env,
        qi->lqi_id.qid_uid = new_id;
        qi->lqi_space      = 1;
        rc = qsd_op_begin(env, qsd, trans, qi, NULL);
-       if (rc == -EDQUOT || rc == -EINPROGRESS)
+       if (ignore_edquot && (rc == -EDQUOT || rc == -EINPROGRESS))
                rc = 0;
        if (rc)
                return rc;
@@ -1076,7 +1076,7 @@ static inline int qsd_transfer(const struct lu_env *env,
        qi->lqi_id.qid_uid = new_id;
        qi->lqi_space      = bspace;
        rc = qsd_op_begin(env, qsd, trans, qi, NULL);
-       if (rc == -EDQUOT || rc == -EINPROGRESS)
+       if (ignore_edquot && (rc == -EDQUOT || rc == -EINPROGRESS))
                rc = 0;
        if (rc)
                return rc;
@@ -1146,7 +1146,11 @@ static int osd_declare_attr_set(const struct lu_env *env,
 
        if (attr && (attr->la_valid & (LA_UID | LA_GID | LA_PROJID))) {
                sa_object_size(obj->oo_sa_hdl, &blksize, &bspace);
-               bspace = toqb(bspace * blksize);
+               bspace = toqb(bspace * 512);
+
+               CDEBUG(D_QUOTA, "%s: enforce quota on UID %u, GID %u,"
+                      "the quota space is %lld (%u)\n", osd->od_svname,
+                      attr->la_uid, attr->la_gid, bspace, blksize);
        }
 
        if (attr && attr->la_valid & LA_UID) {
@@ -1155,7 +1159,7 @@ static int osd_declare_attr_set(const struct lu_env *env,
                        rc = qsd_transfer(env, osd->od_quota_slave,
                                          &oh->ot_quota_trans, USRQUOTA,
                                          obj->oo_attr.la_uid, attr->la_uid,
-                                         bspace, &info->oti_qi);
+                                         bspace, &info->oti_qi, true);
                        if (rc)
                                GOTO(out, rc);
                }
@@ -1166,7 +1170,9 @@ static int osd_declare_attr_set(const struct lu_env *env,
                        rc = qsd_transfer(env, osd->od_quota_slave,
                                          &oh->ot_quota_trans, GRPQUOTA,
                                          obj->oo_attr.la_gid, attr->la_gid,
-                                         bspace, &info->oti_qi);
+                                         bspace, &info->oti_qi,
+                                         !(attr->la_flags &
+                                                       LUSTRE_SET_SYNC_FL));
                        if (rc)
                                GOTO(out, rc);
                }
@@ -1197,7 +1203,7 @@ static int osd_declare_attr_set(const struct lu_env *env,
                                          &oh->ot_quota_trans, PRJQUOTA,
                                          obj->oo_attr.la_projid,
                                          attr->la_projid, bspace,
-                                         &info->oti_qi);
+                                         &info->oti_qi, true);
                        if (rc)
                                GOTO(out, rc);
                }
@@ -1258,9 +1264,9 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
                        rc = -zap_update(osd->od_os, zapid, buf, 8,
                                         sizeof(*zde) / 8, zde, oh->ot_tx);
                }
-               up_read(&obj->oo_guard);
-
-               RETURN(rc > 0 ? 0 : rc);
+               if (rc > 0)
+                       rc = 0;
+               GOTO(out, rc);
        }
 
        /* Only allow set size for regular file */
@@ -1282,6 +1288,7 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
        if (valid & LA_FLAGS) {
                struct lustre_mdt_attrs *lma;
                struct lu_buf buf;
+               int size = 0;
 
                if (la->la_flags & LUSTRE_LMA_FL_MASKS) {
                        LASSERT(!obj->oo_pfid_in_lma);
@@ -1289,9 +1296,14 @@ static int osd_attr_set(const struct lu_env *env, struct dt_object *dt,
                        lma = (struct lustre_mdt_attrs *)&info->oti_buf;
                        buf.lb_buf = lma;
                        buf.lb_len = sizeof(info->oti_buf);
-                       rc = osd_xattr_get(env, &obj->oo_dt, &buf,
-                                          XATTR_NAME_LMA);
-                       if (rc > 0) {
+
+                       /* Please do NOT call osd_xattr_get() directly, that
+                        * will cause recursive down_read() on oo_guard. */
+                       rc = osd_xattr_get_internal(env, obj, &buf,
+                                                   XATTR_NAME_LMA, &size);
+                       if (!rc && unlikely(size < sizeof(*lma))) {
+                               rc = -EINVAL;
+                       } else if (!rc) {
                                lma->lma_incompat =
                                        le32_to_cpu(lma->lma_incompat);
                                lma->lma_incompat |=
@@ -1937,20 +1949,18 @@ static int osd_create(const struct lu_env *env, struct dt_object *dt,
 
        zapid = osd_get_name_n_idx(env, osd, fid, buf,
                                   sizeof(info->oti_str), &zdn);
-       if (!CFS_FAIL_CHECK(OBD_FAIL_OSD_NO_OI_ENTRY)) {
-               if (osd->od_is_ost &&
-                   OBD_FAIL_CHECK(OBD_FAIL_OSD_COMPAT_INVALID_ENTRY))
-                       zde->zde_dnode++;
-
-               if (!osd->od_is_ost ||
-                   !OBD_FAIL_CHECK(OBD_FAIL_OSD_COMPAT_NO_ENTRY)) {
-                       rc = osd_zap_add(osd, zapid, zdn, buf, 8, 1,
-                                        zde, oh->ot_tx);
-                       if (rc)
-                               GOTO(out, rc);
-               }
-       }
+       if (CFS_FAIL_CHECK(OBD_FAIL_OSD_NO_OI_ENTRY) ||
+           (osd->od_is_ost && OBD_FAIL_CHECK(OBD_FAIL_OSD_COMPAT_NO_ENTRY)))
+               goto skip_add;
+
+       if (osd->od_is_ost && OBD_FAIL_CHECK(OBD_FAIL_OSD_COMPAT_INVALID_ENTRY))
+               zde->zde_dnode++;
+
+       rc = osd_zap_add(osd, zapid, zdn, buf, 8, 1, zde, oh->ot_tx);
+       if (rc)
+               GOTO(out, rc);
 
+skip_add:
        obj->oo_dn = dn;
        /* Now add in all of the "SA" attributes */
        rc = osd_sa_handle_get(obj);