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