Whamcloud - gitweb
LU-1346 libcfs: replace libcfs wrappers with kernel API
[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) 2011, 2012, Intel, Inc.
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 #ifndef EXPORT_SYMTAB
32 # define EXPORT_SYMTAB
33 #endif
34
35 #define DEBUG_SUBSYSTEM S_LQUOTA
36
37 #include "qsd_internal.h"
38
39 /*
40  * Initialize qsd-specific fields of quota entry.
41  *
42  * \param lqe - is the quota entry to initialize
43  * \param arg - is the pointer to the qsd_qtype_info structure
44  */
45 static void qsd_lqe_init(struct lquota_entry *lqe, void *arg)
46 {
47         LASSERT(!lqe_is_master(lqe));
48
49         /* initialize slave parameters */
50         rwlock_init(&lqe->lqe_lock);
51         memset(&lqe->lqe_lockh, 0, sizeof(lqe->lqe_lockh));
52         lqe->lqe_pending_write = 0;
53         lqe->lqe_pending_req   = 0;
54         cfs_waitq_init(&lqe->lqe_waiters);
55         lqe->lqe_usage    = 0;
56         lqe->lqe_nopreacq = false;
57 }
58
59 /*
60  * Update a slave quota entry. This is done by reading enforcement status from
61  * the copy of the global index and then how much is the slave currenly owns
62  * for this user from the slave index copy.
63  *
64  * \param env - the environment passed by the caller
65  * \param lqe - is the quota entry to refresh
66  * \param arg - is the pointer to the qsd_qtype_info structure
67  */
68 static int qsd_lqe_read(const struct lu_env *env, struct lquota_entry *lqe,
69                         void *arg)
70 {
71         struct qsd_thread_info *qti = qsd_info(env);
72         struct qsd_qtype_info  *qqi = (struct qsd_qtype_info *)arg;
73         int                     rc;
74
75         LASSERT(!lqe_is_master(lqe));
76
77         /* read record from global index copy to know whether quota is
78          * enforced for this user */
79         rc = lquota_disk_read(env, qqi->qqi_glb_obj, &lqe->lqe_id,
80                               (struct dt_rec *)&qti->qti_glb_rec);
81
82         switch(rc) {
83         case -ENOENT:
84                 /* no such entry, assume quota isn't enforced for this user */
85                 lqe->lqe_enforced = false;
86                 break;
87         case 0:
88                 if (qti->qti_glb_rec.qbr_hardlimit == 0 &&
89                     qti->qti_glb_rec.qbr_softlimit == 0)
90                         /* quota isn't enforced for this use */
91                         lqe->lqe_enforced = false;
92                 else
93                         lqe->lqe_enforced = true;
94                 break;
95         default:
96                 LQUOTA_ERROR(lqe, "failed to read quota entry from global "
97                              "index copy, rc:%d", rc);
98                 return rc;
99         }
100
101         /* read record from slave index copy to find out how much space is
102          * currently owned by this slave */
103         rc = lquota_disk_read(env, qqi->qqi_slv_obj, &lqe->lqe_id,
104                               (struct dt_rec *)&qti->qti_slv_rec);
105         switch(rc) {
106         case -ENOENT:
107                 lqe->lqe_granted = 0;
108                 break;
109         case 0:
110                 lqe->lqe_granted = qti->qti_slv_rec.qsr_granted;
111                 break;
112         default:
113                 LQUOTA_ERROR(lqe, "failed to read quota entry from slave "
114                              "index copy, rc:%d", rc);
115                 return rc;
116         }
117
118         /* don't know what the qunit value is yet */
119         qsd_set_qunit(lqe, 0);
120
121         /* read current disk-usage from disk */
122         rc = qsd_refresh_usage(env, lqe);
123         if (rc)
124                 return rc;
125
126         LQUOTA_DEBUG(lqe, "successfully read from disk");
127         return 0;
128 }
129
130 /*
131  * Print lqe information for debugging.
132  *
133  * \param lqe - is the quota entry to debug
134  * \param arg - is the pointer to the qsd_qtype_info structure
135  * \param msgdata - debug message
136  * \param fmt     - format of debug message
137  */
138 static void qsd_lqe_debug(struct lquota_entry *lqe, void *arg,
139                           struct libcfs_debug_msg_data *msgdata,
140                           const char *fmt, va_list args)
141 {
142         struct qsd_qtype_info   *qqi = (struct qsd_qtype_info *)arg;
143
144         libcfs_debug_vmsg2(msgdata, fmt, args,
145                            "qsd:%s qtype:%s id:"LPU64" enforced:%d granted:"
146                            LPU64" pending:"LPU64" waiting:"LPU64" req:%d usage:"
147                            LPU64" qunit:"LPU64" qtune:"LPU64" edquot:%d\n",
148                            qqi->qqi_qsd->qsd_svname, QTYPE_NAME(qqi->qqi_qtype),
149                            lqe->lqe_id.qid_uid, lqe->lqe_enforced,
150                            lqe->lqe_granted, lqe->lqe_pending_write,
151                            lqe->lqe_waiting_write, lqe->lqe_pending_req,
152                            lqe->lqe_usage, lqe->lqe_qunit, lqe->lqe_qtune,
153                            lqe->lqe_edquot);
154 }
155
156 /*
157  * Vector of quota entry operations supported on the slave
158  */
159 struct lquota_entry_operations qsd_lqe_ops = {
160         .lqe_init               = qsd_lqe_init,
161         .lqe_read               = qsd_lqe_read,
162         .lqe_debug              = qsd_lqe_debug,
163 };
164
165 int qsd_write_version(const struct lu_env *env, struct qsd_qtype_info *qqi,
166                       __u64 ver, bool global)
167 {
168         struct qsd_instance *qsd = qqi->qqi_qsd;
169         struct dt_object    *obj = global ? qqi->qqi_glb_obj :
170                                             qqi->qqi_slv_obj;
171         int                  rc;
172         ENTRY;
173
174         rc = lquota_disk_update_ver(env, qsd->qsd_dev, obj, ver);
175         if (rc)
176                 RETURN(rc);
177
178         qsd_bump_version(qqi, ver, global);
179         RETURN(0);
180 }
181
182 /*
183  * Consult current disk space consumed by a given identifier.
184  *
185  * \param env   - the environment passed by the caller
186  * \param qqi   - is the pointer to the qsd_qtype_info structure associated
187  *                with the identifier.
188  * \param lqe   - is the quota entry associated with the identifier
189  */
190 int qsd_refresh_usage(const struct lu_env *env, struct lquota_entry *lqe)
191 {
192         struct qsd_thread_info  *qti = qsd_info(env);
193         struct lquota_acct_rec  *rec = &qti->qti_acct_rec;
194         struct qsd_qtype_info   *qqi = lqe2qqi(lqe);
195         int                      rc = 0;
196         ENTRY;
197
198         LASSERT(qqi->qqi_acct_obj);
199
200         /* read disk usage */
201         rc = lquota_disk_read(env, qqi->qqi_acct_obj, &lqe->lqe_id,
202                               (struct dt_rec *)rec);
203         switch(rc) {
204         case -ENOENT:
205                 lqe->lqe_usage = 0;
206                 rc = 0;
207                 break;
208         case 0:
209                 if (qqi->qqi_qsd->qsd_is_md)
210                         lqe->lqe_usage = rec->ispace;
211                 else
212                         lqe->lqe_usage = toqb(rec->bspace);
213                 break;
214         default:
215                 LQUOTA_ERROR(lqe, "failed to read disk usage, rc:%d", rc);
216                 RETURN(rc);
217         }
218
219         LQUOTA_DEBUG(lqe, "disk usage: "LPU64, lqe->lqe_usage);
220         RETURN(0);
221 }
222
223 /*
224  * Update slave or global index copy.
225  *
226  * \param env    - the environment passed by the caller
227  * \param qqi    - is the qsd_type_info structure managing the index to be
228  *                 update
229  * \param qid    - is the identifier for which we need to update the quota
230  *                 settings
231  * \param global - is set to true when updating the global index copy and to
232  *                 false for the slave index copy.
233  * \param ver    - is the new version of the index. If equal to 0, the version
234  *                 of the index isn't changed
235  * \param rec    - is the updated record to insert in the index file
236  */
237 int qsd_update_index(const struct lu_env *env, struct qsd_qtype_info *qqi,
238                      union lquota_id *qid, bool global, __u64 ver, void *rec)
239 {
240         struct thandle          *th = NULL;
241         struct dt_object        *obj;
242         __u64                   *new_verp = NULL;
243         int                      flags = 0;
244         int                      rc;
245         ENTRY;
246
247         obj = global ? qqi->qqi_glb_obj : qqi->qqi_slv_obj;
248
249         /* allocate transaction */
250         th = dt_trans_create(env, qqi->qqi_qsd->qsd_dev);
251         if (IS_ERR(th))
252                 RETURN(PTR_ERR(th));
253
254         /* reserve enough credits to update record in index file */
255         rc = lquota_disk_declare_write(env, th, obj, qid);
256         if (rc)
257                 GOTO(out, rc);
258
259         /* start local transaction */
260         rc = dt_trans_start_local(env, qqi->qqi_qsd->qsd_dev, th);
261         if (rc)
262                 GOTO(out, rc);
263
264         if (global) {
265                 /* Update record in global index copy */
266                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
267
268                 CDEBUG(D_QUOTA, "%s: updating global index hardlimit: "LPU64", "
269                        "softlimit: "LPU64" for id "LPU64"\n",
270                        qqi->qqi_qsd->qsd_svname, glb_rec->qbr_hardlimit,
271                        glb_rec->qbr_softlimit, qid->qid_uid);
272         } else {
273                 /* Update record in slave index copy */
274                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
275
276                 CDEBUG(D_QUOTA, "%s: update granted to "LPU64" for id "LPU64
277                        "\n", qqi->qqi_qsd->qsd_svname, slv_rec->qsr_granted,
278                        qid->qid_uid);
279         }
280
281         if (ver != 0) {
282                 new_verp = &ver;
283                 flags = LQUOTA_SET_VER;
284         }
285
286         /* write new record to index file */
287         rc = lquota_disk_write(env, th, obj, qid, (struct dt_rec *)rec, flags,
288                                new_verp);
289         EXIT;
290 out:
291         dt_trans_stop(env, qqi->qqi_qsd->qsd_dev, th);
292         if (rc)
293                 CERROR("%s: failed to update %s index copy for id "LPU64", rc:"
294                        "%d\n", qqi->qqi_qsd->qsd_svname,
295                        global ? "global" : "slave", qid->qid_uid, rc);
296         else if (flags == LQUOTA_SET_VER)
297                 qsd_bump_version(qqi, ver, global);
298         return rc;
299 }
300
301 /*
302  * Update in-memory lquota entry with new quota setting from record \rec.
303  * The record can either be a global record (i.e. lquota_glb_rec) or a slave
304  * index record (i.e. lquota_slv_rec). In the former case, \global should be
305  * set to true.
306  *
307  * \param env    - the environment passed by the caller
308  * \param lqe    - is the quota entry associated with the identifier
309  * \param global - is set to true when updating the record is of type
310  *                 lquota_glb_rec. Otherwise, it is a lquota_slv_rec record.
311  * \param rec    - is the updated record received from the master.
312  */
313 int qsd_update_lqe(const struct lu_env *env, struct lquota_entry *lqe,
314                    bool global, void *rec)
315 {
316         struct qsd_qtype_info *qqi;
317         ENTRY;
318
319         LASSERT(lqe != NULL);
320         LASSERT(!lqe_is_master(lqe));
321
322         qqi = lqe2qqi(lqe);
323
324         /* updating lqe is always serialized, no locking needed. */
325         if (global) {
326                 struct lquota_glb_rec *glb_rec = (struct lquota_glb_rec *)rec;
327
328                 /* change enforcement status based on new hard/soft limit */
329                 lqe->lqe_enforced = (glb_rec->qbr_hardlimit ||
330                                      glb_rec->qbr_softlimit) ? true : false;
331
332                 LQUOTA_DEBUG(lqe, "updating global index hardlimit: "LPU64", "
333                              "softlimit: "LPU64, glb_rec->qbr_hardlimit,
334                              glb_rec->qbr_softlimit);
335         } else {
336                 struct lquota_slv_rec *slv_rec = (struct lquota_slv_rec *)rec;
337
338                 lqe->lqe_granted = slv_rec->qsr_granted;
339
340                 LQUOTA_DEBUG(lqe, "updating slave index, granted:"LPU64"",
341                              slv_rec->qsr_granted);
342         }
343
344         RETURN(0);
345 }