+static inline
+void qmt_grant_lqes(const struct lu_env *env, __u64 *slv, __u64 cnt)
+{
+ int i;
+
+ for (i = 0; i < qti_lqes_cnt(env); i++)
+ qti_lqe_granted(env, i) += cnt;
+
+ *slv += cnt;
+}
+
+static inline bool qmt_lqes_can_rel(const struct lu_env *env, __u64 cnt)
+{
+ bool can_release = true;
+ int i;
+
+ for (i = 0; i < qti_lqes_cnt(env); i++) {
+ if (cnt > qti_lqe_granted(env, i)) {
+ LQUOTA_ERROR(qti_lqes(env)[i],
+ "Can't release %llu that is larger than lqe_granted.\n",
+ cnt);
+ can_release = false;
+ }
+ }
+ return can_release;
+}
+
+static inline void qmt_rel_lqes(const struct lu_env *env, __u64 *slv, __u64 cnt)
+{
+ int i;
+
+ for (i = 0; i < qti_lqes_cnt(env); i++)
+ qti_lqe_granted(env, i) -= cnt;
+
+ *slv -= cnt;
+}
+
+static inline bool qmt_lqes_cannot_grant(const struct lu_env *env, __u64 cnt)
+{
+ bool cannot_grant = false;
+ int i;
+
+ for (i = 0; i < qti_lqes_cnt(env); i++) {
+ if (qti_lqe_hard(env, i) != 0 &&
+ qti_lqe_granted(env, i) + cnt > qti_lqe_hard(env, i)) {
+ cannot_grant = true;
+ break;
+ }
+ }
+ return cannot_grant;
+}
+
+static inline __u64 qmt_lqes_grant_some_quota(const struct lu_env *env)
+{
+ __u64 min_count, tmp;
+ bool flag = false;
+ int i;
+
+ for (i = 0, min_count = 0; i < qti_lqes_cnt(env); i++) {
+ if (!qti_lqes(env)[i]->lqe_enforced &&
+ !qti_lqes(env)[i]->lqe_is_global)
+ continue;
+
+ tmp = qti_lqe_hard(env, i) - qti_lqe_granted(env, i);
+ if (flag) {
+ min_count = tmp < min_count ? tmp : min_count;
+ } else {
+ flag = true;
+ min_count = tmp;
+ }
+ }
+ return min_count;
+}
+
+static inline __u64 qmt_lqes_alloc_expand(const struct lu_env *env,
+ __u64 slv_granted, __u64 spare)
+{
+ __u64 min_count, tmp;
+ bool flag = false;
+ int i;
+
+ for (i = 0, min_count = 0; i < qti_lqes_cnt(env); i++) {
+ /* Don't take into account not enforced lqes that belong
+ * to non global pool. These lqes present in array to
+ * support actual lqe_granted even for lqes without limits. */
+ if (!qti_lqes(env)[i]->lqe_enforced &&
+ !qti_lqes(env)[i]->lqe_is_global)
+ continue;
+
+ tmp = qmt_alloc_expand(qti_lqes(env)[i], slv_granted, spare);
+ if (flag) {
+ min_count = tmp < min_count ? tmp : min_count;
+ } else {
+ flag = true;
+ min_count = tmp;
+ }
+ }
+ return min_count;
+}
+
+static inline void qmt_lqes_tune_grace(const struct lu_env *env, __u64 now)
+{
+ int i;
+
+ for (i = 0; i < qti_lqes_cnt(env); i++) {
+ struct lquota_entry *lqe;
+
+ lqe = qti_lqes(env)[i];
+ if (lqe->lqe_softlimit != 0) {
+ if (lqe->lqe_granted > lqe->lqe_softlimit &&
+ lqe->lqe_gracetime == 0) {
+ /* First time over soft limit, let's start grace
+ * timer */
+ lqe->lqe_gracetime = now + qmt_lqe_grace(lqe);
+ } else if (lqe->lqe_granted <= lqe->lqe_softlimit &&
+ lqe->lqe_gracetime != 0) {
+ /* Clear grace timer */
+ lqe->lqe_gracetime = 0;
+ }
+ }
+ }
+}
+
+/*
+ * Helper function to handle quota request from slave.
+ *
+ * \param env - is the environment passed by the caller
+ * \param qmt - is the master device
+ * \param uuid - is the uuid associated with the slave
+ * \param qb_flags - are the quota request flags as packed in the quota_body
+ * \param qb_count - is the amount of quota space the slave wants to
+ * acquire/release
+ * \param qb_usage - is the current space usage on the slave
+ * \param repbody - is the quota_body of reply
+ *
+ * \retval 0 : success
+ * \retval -EDQUOT : out of quota
+ * -EINPROGRESS : inform client to retry write/create
+ * -ve : other appropriate errors
+ */
+int qmt_dqacq0(const struct lu_env *env, struct qmt_device *qmt,
+ struct obd_uuid *uuid, __u32 qb_flags, __u64 qb_count,
+ __u64 qb_usage, struct quota_body *repbody)
+{
+ __u64 now, count;
+ struct dt_object *slv_obj = NULL;
+ __u64 slv_granted, slv_granted_bck;
+ struct thandle *th = NULL;
+ int rc, ret;
+ struct lquota_entry *lqe = qti_lqes_glbl(env);
+ ENTRY;
+
+ LASSERT(uuid != NULL);
+
+ /* initialize reply */
+ memset(repbody, 0, sizeof(*repbody));
+ memcpy(&repbody->qb_id, &lqe->lqe_id, sizeof(repbody->qb_id));
+
+ if (OBD_FAIL_CHECK(OBD_FAIL_QUOTA_RECOVERABLE_ERR))
+ RETURN(-cfs_fail_val);
+
+ if (qti_lqes_restore_init(env))
+ RETURN(-ENOMEM);
+
+ /* look-up index file associated with acquiring slave */
+ slv_obj = lquota_disk_slv_find(env, qmt->qmt_child, LQE_ROOT(lqe),
+ lu_object_fid(&LQE_GLB_OBJ(lqe)->do_lu),
+ uuid);
+ if (IS_ERR(slv_obj))
+ GOTO(out, rc = PTR_ERR(slv_obj));
+
+ /* pack slave fid in reply just for sanity check */
+ memcpy(&repbody->qb_slv_fid, lu_object_fid(&slv_obj->do_lu),
+ sizeof(struct lu_fid));
+
+ /* allocate & start transaction with enough credits to update
+ * global & slave indexes */
+ th = qmt_trans_start_with_slv(env, NULL, slv_obj, false);
+ if (IS_ERR(th))
+ GOTO(out, rc = PTR_ERR(th));
+
+ qti_lqes_write_lock(env);
+
+ LQUOTA_DEBUG_LQES(env, "dqacq starts uuid:%s flags:0x%x wanted:%llu"
+ " usage:%llu", obd_uuid2str(uuid), qb_flags, qb_count,
+ qb_usage);
+
+ /* Legal race, limits have been removed on master, but slave didn't
+ * receive the change yet. Just return EINPROGRESS until the slave gets
+ * notified. */
+ if (!lqe->lqe_enforced && !req_is_rel(qb_flags))
+ GOTO(out_locked, rc = -ESRCH);
+
+ /* recompute qunit in case it was never initialized */
+ qmt_revalidate_lqes(env, qmt, qb_flags);
+
+ /* slave just wants to acquire per-ID lock */
+ if (req_is_acq(qb_flags) && qb_count == 0)
+ GOTO(out_locked, rc = 0);
+
+ /* fetch how much quota space is already granted to this slave */
+ rc = qmt_slv_read(env, &lqe->lqe_id, slv_obj, &slv_granted);
+ if (rc) {
+ LQUOTA_ERROR(lqe, "Failed to get granted for slave %s, rc=%d",
+ obd_uuid2str(uuid), rc);
+ GOTO(out_locked, rc);
+ }
+ /* recall how much space this slave currently owns in order to restore
+ * it in case of failure */
+ slv_granted_bck = slv_granted;
+
+ /* record current time for soft limit & grace time management */
+ now = ktime_get_real_seconds();
+
+ if (req_is_rel(qb_flags)) {
+ /* Slave would like to release quota space */
+ if (slv_granted < qb_count ||
+ !qmt_lqes_can_rel(env, qb_count)) {
+ /* can't release more than granted */
+ LQUOTA_ERROR_LQES(env,
+ "Release too much! uuid:%s release: %llu granted:%llu, total:%llu",
+ obd_uuid2str(uuid), qb_count,
+ slv_granted, lqe->lqe_granted);
+ GOTO(out_locked, rc = -EINVAL);
+ }
+
+ repbody->qb_count = qb_count;
+ /* put released space back to global pool */
+ qmt_rel_lqes(env, &slv_granted, qb_count);
+ GOTO(out_write, rc = 0);
+ }
+
+ if (req_has_rep(qb_flags) && slv_granted < qb_usage) {
+ /* Slave is reporting space usage in quota request and it turns
+ * out to be using more quota space than owned, so we adjust
+ * granted space regardless of the current state of affairs */
+ repbody->qb_count = qb_usage - slv_granted;
+ qmt_grant_lqes(env, &slv_granted, repbody->qb_count);
+ }
+
+ if (!req_is_acq(qb_flags) && !req_is_preacq(qb_flags))
+ GOTO(out_write, rc = 0);
+
+ qmt_adjust_edquot_notify(env, qmt, now, qb_flags);
+ if (qti_lqes_edquot(env))
+ /* no hope to claim further space back */
+ GOTO(out_write, rc = -EDQUOT);
+
+ if (qmt_space_exhausted_lqes(env, now)) {
+ /* might have some free space once rebalancing is completed */
+ rc = req_is_acq(qb_flags) ? -EINPROGRESS : -EDQUOT;
+ GOTO(out_write, rc);
+ }
+
+ if (req_is_preacq(qb_flags)) {
+ /* slave would like to pre-acquire quota space. To do so, it
+ * reports in qb_count how much spare quota space it owns and we
+ * can grant back quota space which is consistent with qunit
+ * value. */
+ if (qb_count >= qti_lqes_min_qunit(env))
+ /* slave already own the maximum it should */
+ GOTO(out_write, rc = 0);
+
+ count = qmt_lqes_alloc_expand(env, slv_granted, qb_count);
+ if (count == 0)
+ GOTO(out_write, rc = -EDQUOT);
+
+ repbody->qb_count += count;
+ qmt_grant_lqes(env, &slv_granted, count);
+ GOTO(out_write, rc = 0);
+ }
+
+ /* processing acquire request with clients waiting */
+ if (qmt_lqes_cannot_grant(env, qb_count)) {
+ /* cannot grant as much as asked, but can still afford to grant
+ * some quota space back */
+ count = qmt_lqes_grant_some_quota(env);
+ repbody->qb_count += count;
+ qmt_grant_lqes(env, &slv_granted, count);
+ GOTO(out_write, rc = 0);
+ }
+
+ /* Whouhou! we can satisfy the slave request! */
+ repbody->qb_count += qb_count;
+ qmt_grant_lqes(env, &slv_granted, qb_count);
+
+ /* Try to expand the acquired count for DQACQ */
+ count = qmt_lqes_alloc_expand(env, slv_granted, 0);
+ if (count != 0) {
+ /* can even grant more than asked, it is like xmas ... */
+ repbody->qb_count += count;
+ qmt_grant_lqes(env, &slv_granted, count);
+ GOTO(out_write, rc = 0);
+ }
+
+ GOTO(out_write, rc = 0);
+out_write:
+ if (repbody->qb_count == 0)
+ GOTO(out_locked, rc);
+
+ /* start/stop grace timer if required */
+ qmt_lqes_tune_grace(env, now);
+
+ /* Update slave index first since it is easier to roll back */
+ ret = qmt_slv_write(env, th, lqe, slv_obj, LQUOTA_BUMP_VER,
+ &repbody->qb_slv_ver, slv_granted);
+ if (ret) {
+ /* restore initial quota settings */
+ qmt_restore_lqes(env);
+ /* reset qb_count */
+ repbody->qb_count = 0;
+ GOTO(out_locked, rc = ret);
+ }
+
+ /* Update global index, no version bump needed */
+ ret = qmt_glb_write_lqes(env, th, 0, NULL);
+ if (ret) {
+ rc = ret;
+ /* restore initial quota settings */
+ qmt_restore_lqes(env);
+ /* reset qb_count */
+ repbody->qb_count = 0;
+
+ /* restore previous granted value */
+ ret = qmt_slv_write(env, th, lqe, slv_obj, 0, NULL,
+ slv_granted_bck);
+ if (ret) {
+ LQUOTA_ERROR(lqe, "failed to restore initial slave "
+ "value rc:%d ret%d", rc, ret);
+ LBUG();
+ }
+ qmt_adjust_edquot_notify(env, qmt, now, qb_flags);
+ GOTO(out_locked, rc);
+ }
+
+ /* Total granted has been changed, let's try to adjust the qunit
+ * size according to the total granted & limits. */
+
+ /* clear/set edquot flag and notify slaves via glimpse if needed */
+ qmt_adjust_and_notify(env, qmt, now, qb_flags);
+out_locked:
+ LQUOTA_DEBUG_LQES(env, "dqacq ends count:%llu ver:%llu rc:%d",
+ repbody->qb_count, repbody->qb_slv_ver, rc);
+ qti_lqes_write_unlock(env);
+out:
+ qti_lqes_restore_fini(env);
+
+ if (th != NULL && !IS_ERR(th))
+ dt_trans_stop(env, qmt->qmt_child, th);
+
+ if (slv_obj != NULL && !IS_ERR(slv_obj))
+ dt_object_put(env, slv_obj);
+
+ if ((req_is_acq(qb_flags) || req_is_preacq(qb_flags)) &&
+ OBD_FAIL_CHECK(OBD_FAIL_QUOTA_EDQUOT)) {
+ /* introduce inconsistency between granted value in slave index
+ * and slave index copy of slave */
+ repbody->qb_count = 0;
+ rc = -EDQUOT;
+ }
+
+ RETURN(rc);
+}
+
+/*
+ * Extract index from uuid or quota index file name.
+ *
+ * \param[in] uuid uuid or quota index name(0x1020000-OST0001_UUID)
+ * \param[out] idx pointer to save index
+ *
+ * \retval slave type(QMT_STYPE_MDT or QMT_STYPE_OST)
+ * \retval -EINVAL wrong uuid
+ */
+int qmt_uuid2idx(struct obd_uuid *uuid, int *idx)
+{
+ char *uuid_str, *name, *dash;
+ int rc = -EINVAL;
+
+ uuid_str = (char *)uuid->uuid;
+
+ if (strnlen(uuid_str, UUID_MAX) >= UUID_MAX) {
+ CERROR("quota: UUID '%.*s' missing trailing NUL: rc = %d\n",
+ UUID_MAX, uuid_str, rc);
+ return rc;
+ }
+
+ dash = strrchr(uuid_str, '-');
+ name = dash + 1;
+ /* Going to get index from MDTXXXX/OSTXXXX. Thus uuid should
+ * have at least 8 bytes after '-': 3 for MDT/OST, 4 for index
+ * and 1 byte for null character. */
+ if (*dash != '-' || ((uuid_str + UUID_MAX - name) < 8)) {
+ CERROR("quota: wrong UUID format '%s': rc = %d\n",
+ uuid_str, rc);
+ return rc;
+ }
+
+ rc = target_name2index(name, idx, NULL);
+ switch (rc) {
+ case LDD_F_SV_TYPE_MDT:
+ rc = QMT_STYPE_MDT;
+ break;
+ case LDD_F_SV_TYPE_OST:
+ rc = QMT_STYPE_OST;
+ break;
+ default:
+ CERROR("quota: wrong UUID type '%s': rc = %d\n", uuid_str, rc);
+ rc = -EINVAL;
+ }
+
+ RETURN(rc);
+}
+