Whamcloud - gitweb
LU-15785 tests: do not detect versions for RPC_MODE mode
[fs/lustre-release.git] / lustre / quota / lquota_entry.c
1 /* GPL HEADER START
2  *
3  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
4  *
5  * This program is free software; you can redistribute it and/or modify
6  * it under the terms of the GNU General Public License version 2 only,
7  * as published by the Free Software Foundation.
8  *
9  * This program is distributed in the hope that it will be useful, but
10  * WITHOUT ANY WARRANTY; without even the implied warranty of
11  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
12  * General Public License version 2 for more details (a copy is included
13  * in the LICENSE file that accompanied this code).
14  *
15  * You should have received a copy of the GNU General Public License
16  * version 2 along with this program; if not, write to the
17  * Free Software Foundation, Inc., 59 Temple Place - Suite 330,
18  * Boston, MA 021110-1307, USA
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2012, 2015, Intel Corporation.
24  * Use is subject to license terms.
25  *
26  * Author: Johann Lombardi <johann.lombardi@intel.com>
27  * Author: Niu    Yawei    <yawei.niu@intel.com>
28  */
29
30 #define DEBUG_SUBSYSTEM S_LQUOTA
31
32 #include <linux/module.h>
33 #include <linux/slab.h>
34 #include <obd_class.h>
35 #include "lquota_internal.h"
36
37 static int hash_lqs_cur_bits = HASH_LQE_CUR_BITS;
38 module_param(hash_lqs_cur_bits, int, 0444);
39 MODULE_PARM_DESC(hash_lqs_cur_bits, "the current bits of lqe hash");
40
41 static unsigned
42 lqe64_hash_hash(struct cfs_hash *hs, const void *key, unsigned mask)
43 {
44         return cfs_hash_u64_hash(*((__u64 *)key), mask);
45 }
46
47 static void *lqe64_hash_key(struct hlist_node *hnode)
48 {
49         struct lquota_entry *lqe;
50         lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
51         return &lqe->lqe_id.qid_uid;
52 }
53
54 static int lqe64_hash_keycmp(const void *key, struct hlist_node *hnode)
55 {
56         struct lquota_entry *lqe;
57         lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
58         return (lqe->lqe_id.qid_uid == *((__u64*)key));
59 }
60
61 static void *lqe_hash_object(struct hlist_node *hnode)
62 {
63         return hlist_entry(hnode, struct lquota_entry, lqe_hash);
64 }
65
66 static void lqe_hash_get(struct cfs_hash *hs, struct hlist_node *hnode)
67 {
68         struct lquota_entry *lqe;
69         lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
70         lqe_getref(lqe);
71 }
72
73 static void lqe_hash_put_locked(struct cfs_hash *hs, struct hlist_node *hnode)
74 {
75         struct lquota_entry *lqe;
76         lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
77         lqe_putref(lqe);
78 }
79
80 static void lqe_hash_exit(struct cfs_hash *hs, struct hlist_node *hnode)
81 {
82         CERROR("Should not have any item left!\n");
83 }
84
85 /* lqe hash methods for 64-bit uid/gid, new hash functions would have to be
86  * defined for per-directory quota relying on a 128-bit FID */
87 static struct cfs_hash_ops lqe64_hash_ops = {
88         .hs_hash       = lqe64_hash_hash,
89         .hs_key        = lqe64_hash_key,
90         .hs_keycmp     = lqe64_hash_keycmp,
91         .hs_object     = lqe_hash_object,
92         .hs_get        = lqe_hash_get,
93         .hs_put_locked = lqe_hash_put_locked,
94         .hs_exit       = lqe_hash_exit
95 };
96
97 /* Logging helper function */
98 void lquota_lqe_debug0(struct lquota_entry *lqe,
99                        struct libcfs_debug_msg_data *msgdata,
100                        const char *fmt, ...)
101 {
102         struct lquota_site *site = lqe->lqe_site;
103         struct va_format vaf;
104         va_list args;
105
106         LASSERT(site->lqs_ops->lqe_debug != NULL);
107
108         va_start(args, fmt);
109         vaf.fmt = fmt;
110         vaf.va = &args;
111         site->lqs_ops->lqe_debug(lqe, site->lqs_parent, msgdata, &vaf);
112         va_end(args);
113 }
114
115 struct lqe_iter_data {
116         unsigned long   lid_inuse;
117         unsigned long   lid_freed;
118         bool            lid_free_all;
119 };
120
121 static int lqe_iter_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
122                        struct hlist_node *hnode, void *data)
123 {
124         struct lqe_iter_data *d = (struct lqe_iter_data *)data;
125         struct lquota_entry  *lqe;
126
127         lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
128         LASSERT(atomic_read(&lqe->lqe_ref) > 0);
129
130         /* Only one reference held by hash table, and nobody else can
131          * grab the entry at this moment, it's safe to remove it from
132          * the hash and free it. */
133         if (atomic_read(&lqe->lqe_ref) == 1) {
134                 if (!lqe_is_master(lqe)) {
135                         LASSERT(lqe->lqe_pending_write == 0);
136                         LASSERT(lqe->lqe_pending_req == 0);
137                 }
138                 if (d->lid_free_all || lqe->lqe_enforced) {
139                         d->lid_freed++;
140                         cfs_hash_bd_del_locked(hs, bd, hnode);
141                         return 0;
142                 }
143         }
144         d->lid_inuse++;
145
146         if (d->lid_free_all)
147                 LQUOTA_ERROR(lqe, "Inuse quota entry");
148         return 0;
149 }
150
151 /**
152  * Cleanup the entries in the hashtable
153  *
154  * \param hash     - hash table which stores quota entries
155  * \param free_all - free all entries or only free the entries
156  *                   without quota enforce ?
157  */
158 static void lqe_cleanup(struct cfs_hash *hash, bool free_all)
159 {
160         struct lqe_iter_data    d;
161         int                     repeat = 0;
162         ENTRY;
163 retry:
164         memset(&d, 0, sizeof(d));
165         d.lid_free_all = free_all;
166
167         cfs_hash_for_each_safe(hash, lqe_iter_cb, &d);
168
169         /* In most case, when this function is called on master or
170          * slave finalization, there should be no inuse quota entry.
171          *
172          * If the per-fs quota updating thread is still holding
173          * some entries, we just wait for it's finished. */
174         if (free_all && d.lid_inuse) {
175                 CDEBUG(D_QUOTA, "Hash:%p has entries inuse: inuse:%lu, "
176                         "freed:%lu, repeat:%u\n", hash,
177                         d.lid_inuse, d.lid_freed, repeat);
178                 repeat++;
179                 schedule_timeout_interruptible(cfs_time_seconds(1));
180                 goto retry;
181         }
182         EXIT;
183 }
184
185 /*
186  * Allocate a new lquota site.
187  *
188  * \param env    - the environment passed by the caller
189  * \param parent - is a pointer to the parent structure, either a qmt_pool_info
190  *                 structure on the master or a qsd_qtype_info structure on the
191  *                 slave.
192  * \param is_master - is set when the site belongs to a QMT.
193  * \param qtype     - is the quota type managed by this site
194  * \param ops       - is the quota entry operation vector to be used for quota
195  *                    entry belonging to this site.
196  *
197  * \retval 0     - success
198  * \retval -ve   - failure
199  */
200 struct lquota_site *lquota_site_alloc(const struct lu_env *env, void *parent,
201                                       bool is_master, short qtype,
202                                       const struct lquota_entry_operations *ops)
203 {
204         struct lquota_site      *site;
205         char                     hashname[15];
206         ENTRY;
207
208         if (qtype >= LL_MAXQUOTAS)
209                 RETURN(ERR_PTR(-ENOTSUPP));
210
211         OBD_ALLOC_PTR(site);
212         if (site == NULL)
213                 RETURN(ERR_PTR(-ENOMEM));
214
215         /* assign parameters */
216         site->lqs_qtype  = qtype;
217         site->lqs_parent = parent;
218         site->lqs_is_mst = is_master;
219         site->lqs_ops    = ops;
220
221         /* allocate hash table */
222         memset(hashname, 0, sizeof(hashname));
223         snprintf(hashname, sizeof(hashname), "LQUOTA_HASH%hu", qtype);
224         site->lqs_hash= cfs_hash_create(hashname, hash_lqs_cur_bits,
225                                         HASH_LQE_MAX_BITS,
226                                         min(hash_lqs_cur_bits,
227                                             HASH_LQE_BKT_BITS),
228                                         0, CFS_HASH_MIN_THETA,
229                                         CFS_HASH_MAX_THETA, &lqe64_hash_ops,
230                                         CFS_HASH_RW_SEM_BKTLOCK |
231                                         CFS_HASH_COUNTER |
232                                         CFS_HASH_REHASH |
233                                         CFS_HASH_BIGNAME);
234
235         if (site->lqs_hash == NULL) {
236                 OBD_FREE_PTR(site);
237                 RETURN(ERR_PTR(-ENOMEM));
238         }
239
240         RETURN(site);
241 }
242
243 /*
244  * Destroy a lquota site.
245  *
246  * \param env  - the environment passed by the caller
247  * \param site - lquota site to be destroyed
248  *
249  * \retval 0     - success
250  * \retval -ve   - failure
251  */
252 void lquota_site_free(const struct lu_env *env, struct lquota_site *site)
253 {
254         /* cleanup hash table */
255         lqe_cleanup(site->lqs_hash, true);
256         cfs_hash_putref(site->lqs_hash);
257
258         site->lqs_parent = NULL;
259         OBD_FREE_PTR(site);
260 }
261
262 /*
263  * Initialize qsd/qmt-specific fields of quota entry.
264  *
265  * \param lqe - is the quota entry to initialize
266  */
267 static void lqe_init(struct lquota_entry *lqe)
268 {
269         struct lquota_site *site;
270         ENTRY;
271
272         LASSERT(lqe != NULL);
273         site = lqe->lqe_site;
274         LASSERT(site != NULL);
275         LASSERT(site->lqs_ops->lqe_init != NULL);
276
277         LQUOTA_DEBUG(lqe, "init");
278
279         site->lqs_ops->lqe_init(lqe, site->lqs_parent);
280 }
281
282 /*
283  * Update a lquota entry. This is done by reading quota settings from the
284  * on-disk index. The lquota entry must be write locked.
285  *
286  * \param env - the environment passed by the caller
287  * \param lqe - is the quota entry to refresh
288  * \param find - don't create entry on disk if true
289  */
290 static int lqe_read(const struct lu_env *env,
291                     struct lquota_entry *lqe, bool find)
292 {
293         struct lquota_site      *site;
294         int                      rc;
295         ENTRY;
296
297         LASSERT(lqe != NULL);
298         site = lqe->lqe_site;
299         LASSERT(site != NULL);
300         LASSERT(site->lqs_ops->lqe_read != NULL);
301
302         LQUOTA_DEBUG(lqe, "read");
303
304         rc = site->lqs_ops->lqe_read(env, lqe, site->lqs_parent, find);
305         if (rc == 0)
306                 /* mark the entry as up-to-date */
307                 lqe->lqe_uptodate = true;
308
309         RETURN(rc);
310 }
311
312 /*
313  * Find or create a quota entry.
314  *
315  * \param env  - the environment passed by the caller
316  * \param site - lquota site which stores quota entries in a hash table
317  * \param qid  - is the quota ID to be found/created
318  * \param find - don't create lqe on disk in case of ENOENT if true
319  *
320  * \retval 0     - success
321  * \retval -ve   - failure
322  */
323 struct lquota_entry *lqe_locate_find(const struct lu_env *env,
324                                      struct lquota_site *site,
325                                      union lquota_id *qid,
326                                      bool find)
327 {
328         struct lquota_entry     *lqe, *new = NULL;
329         int                      rc = 0;
330         ENTRY;
331
332         lqe = cfs_hash_lookup(site->lqs_hash, (void *)&qid->qid_uid);
333         if (lqe != NULL) {
334                 LASSERT(lqe->lqe_uptodate);
335                 RETURN(lqe);
336         }
337
338         OBD_SLAB_ALLOC_PTR_GFP(new, lqe_kmem, GFP_NOFS);
339         if (new == NULL) {
340                 CERROR("Fail to allocate lqe for id:%llu, "
341                         "hash:%s\n", qid->qid_uid, site->lqs_hash->hs_name);
342                 RETURN(ERR_PTR(-ENOMEM));
343         }
344
345         atomic_set(&new->lqe_ref, 1); /* hold 1 for caller */
346         new->lqe_id     = *qid;
347         new->lqe_site   = site;
348         INIT_LIST_HEAD(&new->lqe_link);
349
350         /* quota settings need to be updated from disk, that's why
351          * lqe->lqe_uptodate isn't set yet */
352         new->lqe_uptodate = false;
353
354         /* perform qmt/qsd specific initialization */
355         lqe_init(new);
356
357         /* read quota settings from disk and mark lqe as up-to-date */
358         rc = lqe_read(env, new, find);
359         if (rc)
360                 GOTO(out, lqe = ERR_PTR(rc));
361
362         /* add new entry to hash */
363         lqe = cfs_hash_findadd_unique(site->lqs_hash, &new->lqe_id.qid_uid,
364                                       &new->lqe_hash);
365         if (lqe == new)
366                 new = NULL;
367 out:
368         if (new)
369                 lqe_putref(new);
370         RETURN(lqe);
371 }