Whamcloud - gitweb
LU-12678 lnet: Fix some out-of-date comments.
[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 }
54
55 /*
56  * Update a slave quota entry. This is done by reading enforcement status from
57  * the copy of the global index and then how much is the slave currenly owns
58  * for this user from the slave index copy.
59  *
60  * \param env - the environment passed by the caller
61  * \param lqe - is the quota entry to refresh
62  * \param arg - is the pointer to the qsd_qtype_info structure
63  * \param need_crt - needed to be compat with qmt_lqe_read
64  */
65 static int qsd_lqe_read(const struct lu_env *env, struct lquota_entry *lqe,
66                         void *arg, bool need_crt)
67 {
68         struct qsd_thread_info *qti = qsd_info(env);
69         struct qsd_qtype_info  *qqi = (struct qsd_qtype_info *)arg;
70         int                     rc;
71
72         LASSERT(!lqe_is_master(lqe));
73
74         /* read record from global index copy to know whether quota is
75          * enforced for this user */
76         rc = lquota_disk_read(env, qqi->qqi_glb_obj, &lqe->lqe_id,
77                               (struct dt_rec *)&qti->qti_glb_rec);
78
79         switch(rc) {
80         case -ENOENT:
81                 /* no such entry, assume quota isn't enforced for this user */
82                 lqe->lqe_enforced = false;
83                 break;
84         case 0:
85                 if (lqe->lqe_id.qid_uid == 0) {
86                         qqi->qqi_default_hardlimit =
87                                                 qti->qti_glb_rec.qbr_hardlimit;
88                         qqi->qqi_default_softlimit =
89                                                 qti->qti_glb_rec.qbr_softlimit;
90                         qqi->qqi_default_gracetime =
91                                                 qti->qti_glb_rec.qbr_granted;
92                 }
93
94                 if (lqe->lqe_id.qid_uid != 0 &&
95                     (qti->qti_glb_rec.qbr_hardlimit != 0 ||
96                      qti->qti_glb_rec.qbr_softlimit != 0))
97                         lqe->lqe_enforced = true;
98                 else
99                         lqe->lqe_enforced = false;
100                 break;
101         default:
102                 LQUOTA_ERROR(lqe, "failed to read quota entry from global "
103                              "index copy, rc:%d", rc);
104                 return rc;
105         }
106
107         if (lqe->lqe_id.qid_uid != 0 &&
108             (rc == -ENOENT ||
109              (LQUOTA_FLAG(qti->qti_glb_rec.qbr_time) & LQUOTA_FLAG_DEFAULT &&
110               qti->qti_glb_rec.qbr_hardlimit == 0 &&
111               qti->qti_glb_rec.qbr_softlimit == 0))) {
112                 struct lquota_entry *lqe_def;
113                 union lquota_id qid = { {0} };
114
115                 /* ensure the lqe storing the default quota setting loaded */
116                 lqe_def = lqe_locate(env, qqi->qqi_site, &qid);
117
118                 lqe->lqe_is_default = true;
119
120                 if (qqi->qqi_default_hardlimit != 0 ||
121                     qqi->qqi_default_softlimit != 0) {
122                         LQUOTA_DEBUG(lqe, "enforced by default quota");
123                         lqe->lqe_enforced = true;
124                 }
125
126                 if (!IS_ERR(lqe_def))
127                         lqe_putref(lqe_def);
128         }
129
130         /* read record from slave index copy to find out how much space is
131          * currently owned by this slave */
132         rc = lquota_disk_read(env, qqi->qqi_slv_obj, &lqe->lqe_id,
133                               (struct dt_rec *)&qti->qti_slv_rec);
134         switch(rc) {
135         case -ENOENT:
136                 lqe->lqe_granted = 0;
137                 break;
138         case 0:
139                 lqe->lqe_granted = qti->qti_slv_rec.qsr_granted;
140                 break;
141         default:
142                 LQUOTA_ERROR(lqe, "failed to read quota entry from slave "
143                              "index copy, rc:%d", rc);
144                 return rc;
145         }
146
147         /* don't know what the qunit value is yet */
148         qsd_set_qunit(lqe, 0);
149
150         /* read current disk-usage from disk */
151         rc = qsd_refresh_usage(env, lqe);
152         if (rc)
153                 return rc;
154
155         LQUOTA_DEBUG(lqe, "successfully read from disk");
156         return 0;
157 }
158
159 /*
160  * Print lqe information for debugging.
161  *
162  * \param lqe - is the quota entry to debug
163  * \param arg - is the pointer to the qsd_qtype_info structure
164  * \param msgdata - debug message
165  * \param fmt     - format of debug message
166  */
167 static void qsd_lqe_debug(struct lquota_entry *lqe, void *arg,
168                           struct libcfs_debug_msg_data *msgdata,
169                           struct va_format *vaf)
170 {
171         struct qsd_qtype_info   *qqi = (struct qsd_qtype_info *)arg;
172
173         libcfs_debug_msg(msgdata,
174                          "%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",
175                          vaf,
176                          qqi->qqi_qsd->qsd_svname, qtype_name(qqi->qqi_qtype),
177                          lqe->lqe_id.qid_uid, lqe->lqe_enforced,
178                          lqe->lqe_granted, lqe->lqe_pending_write,
179                          lqe->lqe_waiting_write, lqe->lqe_pending_req,
180                          lqe->lqe_usage, lqe->lqe_qunit, lqe->lqe_qtune,
181                          lqe->lqe_edquot, lqe->lqe_is_default ? "yes" : "no");
182 }
183
184 /*
185  * Vector of quota entry operations supported on the slave
186  */
187 struct lquota_entry_operations qsd_lqe_ops = {
188         .lqe_init               = qsd_lqe_init,
189         .lqe_read               = qsd_lqe_read,
190         .lqe_debug              = qsd_lqe_debug,
191 };
192
193 int qsd_write_version(const struct lu_env *env, struct qsd_qtype_info *qqi,
194                       __u64 ver, bool global)
195 {
196         struct qsd_instance *qsd = qqi->qqi_qsd;
197         struct dt_object    *obj = global ? qqi->qqi_glb_obj :
198                                             qqi->qqi_slv_obj;
199         int                  rc;
200         ENTRY;
201
202         rc = lquota_disk_update_ver(env, qsd->qsd_dev, obj, ver);
203         if (rc)
204                 RETURN(rc);
205
206         qsd_bump_version(qqi, ver, global);
207         RETURN(0);
208 }
209
210 /*
211  * Consult current disk space consumed by a given identifier.
212  *
213  * \param env   - the environment passed by the caller
214  * \param qqi   - is the pointer to the qsd_qtype_info structure associated
215  *                with the identifier.
216  * \param lqe   - is the quota entry associated with the identifier
217  */
218 int qsd_refresh_usage(const struct lu_env *env, struct lquota_entry *lqe)
219 {
220         struct qsd_thread_info  *qti = qsd_info(env);
221         struct lquota_acct_rec  *rec = &qti->qti_acct_rec;
222         struct qsd_qtype_info   *qqi = lqe2qqi(lqe);
223         int                      rc = 0;
224         ENTRY;
225
226         LASSERT(qqi->qqi_acct_obj);
227
228         /* read disk usage */
229         rc = lquota_disk_read(env, qqi->qqi_acct_obj, &lqe->lqe_id,
230                               (struct dt_rec *)rec);
231         switch(rc) {
232         case -ENOENT:
233                 lqe->lqe_usage = 0;
234                 rc = 0;
235                 break;
236         case 0:
237                 if (qqi->qqi_qsd->qsd_is_md)
238                         lqe->lqe_usage = rec->ispace;
239                 else
240                         lqe->lqe_usage = toqb(rec->bspace);
241                 break;
242         default:
243                 LQUOTA_ERROR(lqe, "failed to read disk usage, rc:%d", rc);
244                 RETURN(rc);
245         }
246
247         LQUOTA_DEBUG(lqe, "disk usage: %llu", lqe->lqe_usage);
248         RETURN(0);
249 }
250
251 /*
252  * Update slave or global index copy.
253  *
254  * \param env    - the environment passed by the caller
255  * \param qqi    - is the qsd_type_info structure managing the index to be
256  *                 update
257  * \param qid    - is the identifier for which we need to update the quota
258  *                 settings
259  * \param global - is set to true when updating the global index copy and to
260  *                 false for the slave index copy.
261  * \param ver    - is the new version of the index. If equal to 0, the version
262  *                 of the index isn't changed
263  * \param rec    - is the updated record to insert in the index file
264  */
265 int qsd_update_index(const struct lu_env *env, struct qsd_qtype_info *qqi,
266                      union lquota_id *qid, bool global, __u64 ver, void *rec)
267 {
268         struct thandle          *th = NULL;
269         struct dt_object        *obj;
270         __u64                   *new_verp = NULL;
271         int                      flags = 0;
272         int                      rc;
273         ENTRY;
274
275         obj = global ? qqi->qqi_glb_obj : qqi->qqi_slv_obj;
276
277         /* allocate transaction */
278         th = dt_trans_create(env, qqi->qqi_qsd->qsd_dev);
279         if (IS_ERR(th))
280                 RETURN(PTR_ERR(th));
281
282         /* reserve enough credits to update record in index file */
283         rc = lquota_disk_declare_write(env, th, obj, qid);
284         if (rc)
285                 GOTO(out, rc);
286
287         /* start local transaction */
288         rc = dt_trans_start_local(env, qqi->qqi_qsd->qsd_dev, th);
289         if (rc)
290                 GOTO(out, rc);
291
292         if (global) {
293                 /* Update record in global index copy */
294                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
295
296                 CDEBUG(D_QUOTA, "%s: updating global index hardlimit: %llu, "
297                        "softlimit: %llu for id %llu\n",
298                        qqi->qqi_qsd->qsd_svname, glb_rec->qbr_hardlimit,
299                        glb_rec->qbr_softlimit, qid->qid_uid);
300         } else {
301                 /* Update record in slave index copy */
302                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
303
304                 CDEBUG(D_QUOTA, "%s: update granted to %llu for id %llu"
305                        "\n", qqi->qqi_qsd->qsd_svname, slv_rec->qsr_granted,
306                        qid->qid_uid);
307         }
308
309         if (ver != 0) {
310                 new_verp = &ver;
311                 flags = LQUOTA_SET_VER;
312         }
313
314         /* write new record to index file */
315         rc = lquota_disk_write(env, th, obj, qid, (struct dt_rec *)rec, flags,
316                                new_verp);
317         EXIT;
318 out:
319         dt_trans_stop(env, qqi->qqi_qsd->qsd_dev, th);
320         if (rc)
321                 CERROR("%s: failed to update %s index copy for id %llu, : rc = %d\n",
322                        qqi->qqi_qsd->qsd_svname,
323                        global ? "global" : "slave", qid->qid_uid, rc);
324         else if (flags == LQUOTA_SET_VER)
325                 qsd_bump_version(qqi, ver, global);
326         return rc;
327 }
328
329 /*
330  * Update in-memory lquota entry with new quota setting from record \rec.
331  * The record can either be a global record (i.e. lquota_glb_rec) or a slave
332  * index record (i.e. lquota_slv_rec). In the former case, \global should be
333  * set to true.
334  *
335  * \param env    - the environment passed by the caller
336  * \param lqe    - is the quota entry associated with the identifier
337  * \param global - is set to true when updating the record is of type
338  *                 lquota_glb_rec. Otherwise, it is a lquota_slv_rec record.
339  * \param rec    - is the updated record received from the master.
340  */
341 int qsd_update_lqe(const struct lu_env *env, struct lquota_entry *lqe,
342                    bool global, void *rec)
343 {
344         ENTRY;
345
346         LASSERT(lqe != NULL);
347         LASSERT(!lqe_is_master(lqe));
348
349         /* updating lqe is always serialized, no locking needed. */
350         if (global) {
351                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
352
353                 /* doesn't change quota enforcement if the quota entry is still
354                  * using default quota. */
355                 if (LQUOTA_FLAG(glb_rec->qbr_time) & LQUOTA_FLAG_DEFAULT &&
356                     glb_rec->qbr_hardlimit == 0 && glb_rec->qbr_softlimit == 0)
357                         RETURN(0);
358
359                 LQUOTA_DEBUG(lqe, "the ID has been set quota, so clear the"
360                              " default quota flag");
361                 lqe->lqe_is_default = false;
362
363                 /* change enforcement status based on new hard/soft limit */
364                 if (lqe->lqe_id.qid_uid != 0 && (glb_rec->qbr_hardlimit != 0 ||
365                     glb_rec->qbr_softlimit != 0))
366                         lqe->lqe_enforced = true;
367                 else
368                         lqe->lqe_enforced = false;
369
370                 LQUOTA_DEBUG(lqe, "updating global index hardlimit: %llu, "
371                              "softlimit: %llu", glb_rec->qbr_hardlimit,
372                              glb_rec->qbr_softlimit);
373         } else {
374                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
375
376                 lqe->lqe_granted = slv_rec->qsr_granted;
377
378                 LQUOTA_DEBUG(lqe, "updating slave index, granted:%llu",
379                              slv_rec->qsr_granted);
380         }
381
382         RETURN(0);
383 }