Whamcloud - gitweb
LU-15785 tests: do not detect versions for RPC_MODE mode
[fs/lustre-release.git] / lustre / quota / qsd_entry.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; if not, write to the
18  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
19  * Boston, MA 021110-1307, USA
20  *
21  * GPL HEADER END
22  */
23 /*
24  * Copyright (c) 2012, 2017, Intel Corporation.
25  * Use is subject to license terms.
26  *
27  * Author: Johann Lombardi <johann.lombardi@intel.com>
28  * Author: Niu    Yawei    <yawei.niu@intel.com>
29  */
30
31 #define DEBUG_SUBSYSTEM S_LQUOTA
32
33 #include "qsd_internal.h"
34
35 /*
36  * Initialize qsd-specific fields of quota entry.
37  *
38  * \param lqe - is the quota entry to initialize
39  * \param arg - is the pointer to the qsd_qtype_info structure
40  */
41 static void qsd_lqe_init(struct lquota_entry *lqe, void *arg)
42 {
43         LASSERT(!lqe_is_master(lqe));
44
45         /* initialize slave parameters */
46         rwlock_init(&lqe->lqe_lock);
47         memset(&lqe->lqe_lockh, 0, sizeof(lqe->lqe_lockh));
48         lqe->lqe_pending_write = 0;
49         lqe->lqe_pending_req   = 0;
50         init_waitqueue_head(&lqe->lqe_waiters);
51         lqe->lqe_usage    = 0;
52         lqe->lqe_nopreacq = false;
53         mutex_init(&lqe->lqe_glbl_data_lock);
54 }
55
56 /*
57  * Update a slave quota entry. This is done by reading enforcement status from
58  * the copy of the global index and the amount the slave currenly owns
59  * for this user from the slave index copy.
60  *
61  * \param env - the environment passed by the caller
62  * \param lqe - is the quota entry to refresh
63  * \param arg - is the pointer to the qsd_qtype_info structure
64  * \param need_crt - needed to be compat with qmt_lqe_read
65  */
66 static int qsd_lqe_read(const struct lu_env *env, struct lquota_entry *lqe,
67                         void *arg, bool need_crt)
68 {
69         struct qsd_thread_info *qti = qsd_info(env);
70         struct qsd_qtype_info  *qqi = (struct qsd_qtype_info *)arg;
71         int                     rc;
72
73         LASSERT(!lqe_is_master(lqe));
74
75         /* read record from global index copy to know whether quota is
76          * enforced for this user */
77         rc = lquota_disk_read(env, qqi->qqi_glb_obj, &lqe->lqe_id,
78                               (struct dt_rec *)&qti->qti_glb_rec);
79
80         switch(rc) {
81         case -ENOENT:
82                 /* no such entry, assume quota isn't enforced for this user */
83                 lqe->lqe_enforced = false;
84                 break;
85         case 0:
86                 if (lqe->lqe_id.qid_uid == 0) {
87                         qqi->qqi_default_hardlimit =
88                                                 qti->qti_glb_rec.qbr_hardlimit;
89                         qqi->qqi_default_softlimit =
90                                                 qti->qti_glb_rec.qbr_softlimit;
91                         qqi->qqi_default_gracetime =
92                                                 qti->qti_glb_rec.qbr_granted;
93                 }
94
95                 if (lqe->lqe_id.qid_uid != 0 &&
96                     (qti->qti_glb_rec.qbr_hardlimit != 0 ||
97                      qti->qti_glb_rec.qbr_softlimit != 0))
98                         lqe->lqe_enforced = true;
99                 else
100                         lqe->lqe_enforced = false;
101                 break;
102         default:
103                 LQUOTA_ERROR(lqe, "failed to read quota entry from global "
104                              "index copy, rc:%d", rc);
105                 return rc;
106         }
107
108         if (lqe->lqe_id.qid_uid != 0 &&
109             (rc == -ENOENT ||
110              (LQUOTA_FLAG(qti->qti_glb_rec.qbr_time) & LQUOTA_FLAG_DEFAULT &&
111               qti->qti_glb_rec.qbr_hardlimit == 0 &&
112               qti->qti_glb_rec.qbr_softlimit == 0))) {
113                 struct lquota_entry *lqe_def;
114                 union lquota_id qid = { {0} };
115
116                 /* ensure the lqe storing the default quota setting loaded */
117                 lqe_def = lqe_locate(env, qqi->qqi_site, &qid);
118
119                 lqe->lqe_is_default = true;
120
121                 if (qqi->qqi_default_hardlimit != 0 ||
122                     qqi->qqi_default_softlimit != 0) {
123                         LQUOTA_DEBUG(lqe, "enforced by default quota");
124                         lqe->lqe_enforced = true;
125                 }
126
127                 if (!IS_ERR(lqe_def))
128                         lqe_putref(lqe_def);
129         }
130
131         /* read record from slave index copy to find out how much space is
132          * currently owned by this slave */
133         rc = lquota_disk_read(env, qqi->qqi_slv_obj, &lqe->lqe_id,
134                               (struct dt_rec *)&qti->qti_slv_rec);
135         switch(rc) {
136         case -ENOENT:
137                 lqe->lqe_granted = 0;
138                 break;
139         case 0:
140                 lqe->lqe_granted = qti->qti_slv_rec.qsr_granted;
141                 break;
142         default:
143                 LQUOTA_ERROR(lqe, "failed to read quota entry from slave "
144                              "index copy, rc:%d", rc);
145                 return rc;
146         }
147
148         /* don't know what the qunit value is yet */
149         qsd_set_qunit(lqe, 0);
150
151         /* read current disk-usage from disk */
152         rc = qsd_refresh_usage(env, lqe);
153         if (rc)
154                 return rc;
155
156         LQUOTA_DEBUG(lqe, "successfully read from disk");
157         return 0;
158 }
159
160 /*
161  * Print lqe information for debugging.
162  *
163  * \param lqe - is the quota entry to debug
164  * \param arg - is the pointer to the qsd_qtype_info structure
165  * \param msgdata - debug message
166  * \param fmt     - format of debug message
167  */
168 static void qsd_lqe_debug(struct lquota_entry *lqe, void *arg,
169                           struct libcfs_debug_msg_data *msgdata,
170                           struct va_format *vaf)
171 {
172         struct qsd_qtype_info   *qqi = (struct qsd_qtype_info *)arg;
173
174         libcfs_debug_msg(msgdata,
175                          "%pV qsd:%s qtype:%s id:%llu enforced:%d granted: %llu pending:%llu waiting:%llu req:%d usage: %llu qunit:%llu qtune:%llu edquot:%d default:%s\n",
176                          vaf,
177                          qqi->qqi_qsd->qsd_svname, qtype_name(qqi->qqi_qtype),
178                          lqe->lqe_id.qid_uid, lqe->lqe_enforced,
179                          lqe->lqe_granted, lqe->lqe_pending_write,
180                          lqe->lqe_waiting_write, lqe->lqe_pending_req,
181                          lqe->lqe_usage, lqe->lqe_qunit, lqe->lqe_qtune,
182                          lqe->lqe_edquot, lqe->lqe_is_default ? "yes" : "no");
183 }
184
185 /*
186  * Vector of quota entry operations supported on the slave
187  */
188 const struct lquota_entry_operations qsd_lqe_ops = {
189         .lqe_init               = qsd_lqe_init,
190         .lqe_read               = qsd_lqe_read,
191         .lqe_debug              = qsd_lqe_debug,
192 };
193
194 int qsd_write_version(const struct lu_env *env, struct qsd_qtype_info *qqi,
195                       __u64 ver, bool global)
196 {
197         struct qsd_instance *qsd = qqi->qqi_qsd;
198         struct dt_object    *obj = global ? qqi->qqi_glb_obj :
199                                             qqi->qqi_slv_obj;
200         int                  rc;
201         ENTRY;
202
203         rc = lquota_disk_update_ver(env, qsd->qsd_dev, obj, ver);
204         if (rc)
205                 RETURN(rc);
206
207         qsd_bump_version(qqi, ver, global);
208         RETURN(0);
209 }
210
211 /*
212  * Consult current disk space consumed by a given identifier.
213  *
214  * \param env   - the environment passed by the caller
215  * \param qqi   - is the pointer to the qsd_qtype_info structure associated
216  *                with the identifier.
217  * \param lqe   - is the quota entry associated with the identifier
218  */
219 int qsd_refresh_usage(const struct lu_env *env, struct lquota_entry *lqe)
220 {
221         struct qsd_thread_info  *qti = qsd_info(env);
222         struct lquota_acct_rec  *rec = &qti->qti_acct_rec;
223         struct qsd_qtype_info   *qqi = lqe2qqi(lqe);
224         int                      rc = 0;
225         ENTRY;
226
227         LASSERT(qqi->qqi_acct_obj);
228
229         /* read disk usage */
230         rc = lquota_disk_read(env, qqi->qqi_acct_obj, &lqe->lqe_id,
231                               (struct dt_rec *)rec);
232         switch(rc) {
233         case -ENOENT:
234                 lqe->lqe_usage = 0;
235                 rc = 0;
236                 break;
237         case 0:
238                 if (qqi->qqi_qsd->qsd_is_md)
239                         lqe->lqe_usage = rec->ispace;
240                 else
241                         lqe->lqe_usage = toqb(rec->bspace);
242                 break;
243         default:
244                 LQUOTA_ERROR(lqe, "failed to read disk usage, rc:%d", rc);
245                 RETURN(rc);
246         }
247
248         LQUOTA_DEBUG(lqe, "disk usage: %llu", lqe->lqe_usage);
249         RETURN(0);
250 }
251
252 /*
253  * Update slave or global index copy.
254  *
255  * \param env    - the environment passed by the caller
256  * \param qqi    - is the qsd_type_info structure managing the index to be
257  *                 update
258  * \param qid    - is the identifier for which we need to update the quota
259  *                 settings
260  * \param global - is set to true when updating the global index copy and to
261  *                 false for the slave index copy.
262  * \param ver    - is the new version of the index. If equal to 0, the version
263  *                 of the index isn't changed
264  * \param rec    - is the updated record to insert in the index file
265  */
266 int qsd_update_index(const struct lu_env *env, struct qsd_qtype_info *qqi,
267                      union lquota_id *qid, bool global, __u64 ver, void *rec)
268 {
269         struct thandle          *th = NULL;
270         struct dt_object        *obj;
271         __u64                   *new_verp = NULL;
272         int                      flags = 0;
273         int                      rc;
274         ENTRY;
275
276         obj = global ? qqi->qqi_glb_obj : qqi->qqi_slv_obj;
277
278         /* allocate transaction */
279         th = dt_trans_create(env, qqi->qqi_qsd->qsd_dev);
280         if (IS_ERR(th))
281                 RETURN(PTR_ERR(th));
282
283         /* reserve enough credits to update record in index file */
284         rc = lquota_disk_declare_write(env, th, obj, qid);
285         if (rc)
286                 GOTO(out, rc);
287
288         /* start local transaction */
289         rc = dt_trans_start_local(env, qqi->qqi_qsd->qsd_dev, th);
290         if (rc)
291                 GOTO(out, rc);
292
293         if (global) {
294                 /* Update record in global index copy */
295                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
296
297                 CDEBUG(D_QUOTA, "%s: updating global index hardlimit: %llu, "
298                        "softlimit: %llu for id %llu\n",
299                        qqi->qqi_qsd->qsd_svname, glb_rec->qbr_hardlimit,
300                        glb_rec->qbr_softlimit, qid->qid_uid);
301         } else {
302                 /* Update record in slave index copy */
303                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
304
305                 CDEBUG(D_QUOTA, "%s: update granted to %llu for id %llu"
306                        "\n", qqi->qqi_qsd->qsd_svname, slv_rec->qsr_granted,
307                        qid->qid_uid);
308         }
309
310         if (ver != 0) {
311                 new_verp = &ver;
312                 flags = LQUOTA_SET_VER;
313         }
314
315         /* write new record to index file */
316         rc = lquota_disk_write(env, th, obj, qid, (struct dt_rec *)rec, flags,
317                                new_verp);
318         EXIT;
319 out:
320         dt_trans_stop(env, qqi->qqi_qsd->qsd_dev, th);
321         if (rc)
322                 CERROR("%s: failed to update %s index copy for id %llu, : rc = %d\n",
323                        qqi->qqi_qsd->qsd_svname,
324                        global ? "global" : "slave", qid->qid_uid, rc);
325         else if (flags == LQUOTA_SET_VER)
326                 qsd_bump_version(qqi, ver, global);
327         return rc;
328 }
329
330 /*
331  * Update in-memory lquota entry with new quota setting from record \rec.
332  * The record can either be a global record (i.e. lquota_glb_rec) or a slave
333  * index record (i.e. lquota_slv_rec). In the former case, \global should be
334  * set to true.
335  *
336  * \param env    - the environment passed by the caller
337  * \param lqe    - is the quota entry associated with the identifier
338  * \param global - is set to true when updating the record is of type
339  *                 lquota_glb_rec. Otherwise, it is a lquota_slv_rec record.
340  * \param rec    - is the updated record received from the master.
341  */
342 int qsd_update_lqe(const struct lu_env *env, struct lquota_entry *lqe,
343                    bool global, void *rec)
344 {
345         ENTRY;
346
347         LASSERT(lqe != NULL);
348         LASSERT(!lqe_is_master(lqe));
349
350         /* updating lqe is always serialized, no locking needed. */
351         if (global) {
352                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
353
354                 /* doesn't change quota enforcement if the quota entry is still
355                  * using default quota. */
356                 if (LQUOTA_FLAG(glb_rec->qbr_time) & LQUOTA_FLAG_DEFAULT &&
357                     glb_rec->qbr_hardlimit == 0 && glb_rec->qbr_softlimit == 0)
358                         RETURN(0);
359
360                 LQUOTA_DEBUG(lqe, "the ID has been set quota, so clear the"
361                              " default quota flag");
362                 lqe->lqe_is_default = false;
363
364                 /* change enforcement status based on new hard/soft limit */
365                 if (lqe->lqe_id.qid_uid != 0 && (glb_rec->qbr_hardlimit != 0 ||
366                     glb_rec->qbr_softlimit != 0))
367                         lqe->lqe_enforced = true;
368                 else
369                         lqe->lqe_enforced = false;
370
371                 LQUOTA_DEBUG(lqe, "updating global index hardlimit: %llu, "
372                              "softlimit: %llu", glb_rec->qbr_hardlimit,
373                              glb_rec->qbr_softlimit);
374         } else {
375                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
376
377                 lqe->lqe_granted = slv_rec->qsr_granted;
378
379                 LQUOTA_DEBUG(lqe, "updating slave index, granted:%llu",
380                              slv_rec->qsr_granted);
381         }
382
383         RETURN(0);
384 }