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 CFS_MODULE_PARM(hash_lqs_cur_bits, "i", int, 0444,
  39                 "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         va_list args;
 104
 105         LASSERT(site->lqs_ops->lqe_debug != NULL);
 106
 107         va_start(args, fmt);
 108         site->lqs_ops->lqe_debug(lqe, site->lqs_parent, msgdata, fmt, args);
 109         va_end(args);
 110 }
 111
 112 struct lqe_iter_data {
 113         unsigned long   lid_inuse;
 114         unsigned long   lid_freed;
 115         bool            lid_free_all;
 116 };
 117
 118 static int lqe_iter_cb(struct cfs_hash *hs, struct cfs_hash_bd *bd,
 119                        struct hlist_node *hnode, void *data)
 120 {
 121         struct lqe_iter_data *d = (struct lqe_iter_data *)data;
 122         struct lquota_entry  *lqe;
 123
 124         lqe = hlist_entry(hnode, struct lquota_entry, lqe_hash);
 125         LASSERT(atomic_read(&lqe->lqe_ref) > 0);
 126
 127         /* Only one reference held by hash table, and nobody else can
 128          * grab the entry at this moment, it's safe to remove it from
 129          * the hash and free it. */
 130         if (atomic_read(&lqe->lqe_ref) == 1) {
 131                 if (!lqe_is_master(lqe)) {
 132                         LASSERT(lqe->lqe_pending_write == 0);
 133                         LASSERT(lqe->lqe_pending_req == 0);
 134                 }
 135                 if (d->lid_free_all || lqe->lqe_enforced) {
 136                         d->lid_freed++;
 137                         cfs_hash_bd_del_locked(hs, bd, hnode);
 138                         return 0;
 139                 }
 140         }
 141         d->lid_inuse++;
 142
 143         if (d->lid_free_all)
 144                 LQUOTA_ERROR(lqe, "Inuse quota entry");
 145         return 0;
 146 }
 147
 148 /**
 149  * Cleanup the entries in the hashtable
 150  *
 151  * \param hash     - hash table which stores quota entries
 152  * \param free_all - free all entries or only free the entries
 153  *                   without quota enforce ?
 154  */
 155 static void lqe_cleanup(struct cfs_hash *hash, bool free_all)
 156 {
 157         struct lqe_iter_data    d;
 158         int                     repeat = 0;
 159         ENTRY;
 160 retry:
 161         memset(&d, 0, sizeof(d));
 162         d.lid_free_all = free_all;
 163
 164         cfs_hash_for_each_safe(hash, lqe_iter_cb, &d);
 165
 166         /* In most case, when this function is called on master or
 167          * slave finalization, there should be no inuse quota entry.
 168          *
 169          * If the per-fs quota updating thread is still holding
 170          * some entries, we just wait for it's finished. */
 171         if (free_all && d.lid_inuse) {
 172                 CDEBUG(D_QUOTA, "Hash:%p has entries inuse: inuse:%lu, "
 173                         "freed:%lu, repeat:%u\n", hash,
 174                         d.lid_inuse, d.lid_freed, repeat);
 175                 repeat++;
 176                 set_current_state(TASK_INTERRUPTIBLE);
 177                 schedule_timeout(cfs_time_seconds(1));
 178                 goto retry;
 179         }
 180         EXIT;
 181 }
 182
 183 /*
 184  * Allocate a new lquota site.
 185  *
 186  * \param env    - the environment passed by the caller
 187  * \param parent - is a pointer to the parent structure, either a qmt_pool_info
 188  *                 structure on the master or a qsd_qtype_info structure on the
 189  *                 slave.
 190  * \param is_master - is set when the site belongs to a QMT.
 191  * \param qtype     - is the quota type managed by this site
 192  * \param ops       - is the quota entry operation vector to be used for quota
 193  *                    entry belonging to this site.
 194  *
 195  * \retval 0     - success
 196  * \retval -ve   - failure
 197  */
 198 struct lquota_site *lquota_site_alloc(const struct lu_env *env, void *parent,
 199                                       bool is_master, short qtype,
 200                                       struct lquota_entry_operations *ops)
 201 {
 202         struct lquota_site      *site;
 203         char                     hashname[15];
 204         ENTRY;
 205
 206         LASSERT(qtype < MAXQUOTAS);
 207
 208         OBD_ALLOC_PTR(site);
 209         if (site == NULL)
 210                 RETURN(ERR_PTR(-ENOMEM));
 211
 212         /* assign parameters */
 213         site->lqs_qtype  = qtype;
 214         site->lqs_parent = parent;
 215         site->lqs_is_mst = is_master;
 216         site->lqs_ops    = ops;
 217
 218         /* allocate hash table */
 219         memset(hashname, 0, sizeof(hashname));
 220         snprintf(hashname, sizeof(hashname), "LQUOTA_HASH%hu", qtype);
 221         site->lqs_hash= cfs_hash_create(hashname, hash_lqs_cur_bits,
 222                                         HASH_LQE_MAX_BITS,
 223                                         min(hash_lqs_cur_bits,
 224                                             HASH_LQE_BKT_BITS),
 225                                         0, CFS_HASH_MIN_THETA,
 226                                         CFS_HASH_MAX_THETA, &lqe64_hash_ops,
 227                                         CFS_HASH_DEFAULT|CFS_HASH_BIGNAME);
 228         if (site->lqs_hash == NULL) {
 229                 OBD_FREE_PTR(site);
 230                 RETURN(ERR_PTR(-ENOMEM));
 231         }
 232
 233         RETURN(site);
 234 }
 235
 236 /*
 237  * Destroy a lquota site.
 238  *
 239  * \param env  - the environment passed by the caller
 240  * \param site - lquota site to be destroyed
 241  *
 242  * \retval 0     - success
 243  * \retval -ve   - failure
 244  */
 245 void lquota_site_free(const struct lu_env *env, struct lquota_site *site)
 246 {
 247         /* cleanup hash table */
 248         lqe_cleanup(site->lqs_hash, true);
 249         cfs_hash_putref(site->lqs_hash);
 250
 251         site->lqs_parent = NULL;
 252         OBD_FREE_PTR(site);
 253 }
 254
 255 /*
 256  * Initialize qsd/qmt-specific fields of quota entry.
 257  *
 258  * \param lqe - is the quota entry to initialize
 259  */
 260 static void lqe_init(struct lquota_entry *lqe)
 261 {
 262         struct lquota_site *site;
 263         ENTRY;
 264
 265         LASSERT(lqe != NULL);
 266         site = lqe->lqe_site;
 267         LASSERT(site != NULL);
 268         LASSERT(site->lqs_ops->lqe_init != NULL);
 269
 270         LQUOTA_DEBUG(lqe, "init");
 271
 272         site->lqs_ops->lqe_init(lqe, site->lqs_parent);
 273 }
 274
 275 /*
 276  * Update a lquota entry. This is done by reading quota settings from the
 277  * on-disk index. The lquota entry must be write locked.
 278  *
 279  * \param env - the environment passed by the caller
 280  * \param lqe - is the quota entry to refresh
 281  */
 282 static int lqe_read(const struct lu_env *env, struct lquota_entry *lqe)
 283 {
 284         struct lquota_site      *site;
 285         int                      rc;
 286         ENTRY;
 287
 288         LASSERT(lqe != NULL);
 289         site = lqe->lqe_site;
 290         LASSERT(site != NULL);
 291         LASSERT(site->lqs_ops->lqe_read != NULL);
 292
 293         LQUOTA_DEBUG(lqe, "read");
 294
 295         rc = site->lqs_ops->lqe_read(env, lqe, site->lqs_parent);
 296         if (rc == 0)
 297                 /* mark the entry as up-to-date */
 298                 lqe->lqe_uptodate = true;
 299
 300         RETURN(rc);
 301 }
 302
 303 /*
 304  * Find or create a quota entry.
 305  *
 306  * \param env  - the environment passed by the caller
 307  * \param site - lquota site which stores quota entries in a hash table
 308  * \param qid  - is the quota ID to be found/created
 309  *
 310  * \retval 0     - success
 311  * \retval -ve   - failure
 312  */
 313 struct lquota_entry *lqe_locate(const struct lu_env *env,
 314                                 struct lquota_site *site, union lquota_id *qid)
 315 {
 316         struct lquota_entry     *lqe, *new = NULL;
 317         int                      rc = 0;
 318         ENTRY;
 319
 320         lqe = cfs_hash_lookup(site->lqs_hash, (void *)&qid->qid_uid);
 321         if (lqe != NULL) {
 322                 LASSERT(lqe->lqe_uptodate);
 323                 RETURN(lqe);
 324         }
 325
 326         OBD_SLAB_ALLOC_PTR_GFP(new, lqe_kmem, GFP_NOFS);
 327         if (new == NULL) {
 328                 CERROR("Fail to allocate lqe for id:"LPU64", "
 329                         "hash:%s\n", qid->qid_uid, site->lqs_hash->hs_name);
 330                 RETURN(ERR_PTR(-ENOMEM));
 331         }
 332
 333         atomic_set(&new->lqe_ref, 1); /* hold 1 for caller */
 334         new->lqe_id     = *qid;
 335         new->lqe_site   = site;
 336         INIT_LIST_HEAD(&new->lqe_link);
 337
 338         /* quota settings need to be updated from disk, that's why
 339          * lqe->lqe_uptodate isn't set yet */
 340         new->lqe_uptodate = false;
 341
 342         /* perform qmt/qsd specific initialization */
 343         lqe_init(new);
 344
 345         /* read quota settings from disk and mark lqe as up-to-date */
 346         rc = lqe_read(env, new);
 347         if (rc)
 348                 GOTO(out, lqe = ERR_PTR(rc));
 349
 350         /* add new entry to hash */
 351         lqe = cfs_hash_findadd_unique(site->lqs_hash, &new->lqe_id.qid_uid,
 352                                       &new->lqe_hash);
 353         if (lqe == new)
 354                 new = NULL;
 355 out:
 356         if (new)
 357                 lqe_putref(new);
 358         RETURN(lqe);
 359 }