4 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
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.
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).
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
24 * Copyright (c) 2012 Intel, Inc.
25 * Use is subject to license terms.
27 * Author: Johann Lombardi <johann.lombardi@intel.com>
28 * Author: Niu Yawei <yawei.niu@intel.com>
32 * The disk API is used by both the QMT and QSD to access/update on-disk index
33 * files. The API consists of the following functions:
35 * - lquota_disk_dir_find_create: look-up quota directory, create it if not
37 * - lquota_disk_glb_find_create: look-up global index file, create it if not
39 * - lquota_disk_slv_find: look-up a slave index file.
40 * - lquota_disk_slv_find_create: look-up a slave index file. Allocate a FID if
41 * required and create the index file on disk if
43 * - lquota_disk_for_each_slv: iterate over all existing slave index files
44 * - lquota_disk_read: read quota settings from a index file
45 * - lquota_disk_declare_write: reserve credits to update a record in an index
47 * - lquota_disk_write: update a record in an index file
48 * - lquota_disk_update_ver: update version of an index file
52 # define EXPORT_SYMTAB
55 #define DEBUG_SUBSYSTEM S_LQUOTA
57 #include "lquota_internal.h"
59 #define LQUOTA_MODE (S_IFREG | S_IRUGO | S_IWUSR)
62 * Helper function looking up & creating if not found an index file with a
65 static struct dt_object *
66 lquota_disk_find_create(const struct lu_env *env, struct dt_device *dev,
67 struct dt_object *parent, struct lu_fid *fid,
68 const struct dt_index_features *idx_feat,
71 struct lquota_thread_info *qti = lquota_info(env);
72 struct dt_object *obj;
73 struct local_oid_storage *los;
77 /* Set up local storage */
78 rc = local_oid_storage_init(env, dev, fid, &los);
82 /* lookup/create slave index file */
83 obj = local_index_find_or_create(env, los, parent, name, LQUOTA_MODE,
88 /* local_oid_storage_fini() will finalize the local storage device,
89 * we have to open the object in another device stack */
90 qti->qti_fid = obj->do_lu.lo_header->loh_fid;
91 lu_object_put_nocache(env, &obj->do_lu);
92 obj = dt_locate(env, dev, &qti->qti_fid);
96 /* install index operation vector */
97 if (obj->do_index_ops == NULL) {
98 rc = obj->do_ops->do_index_try(env, obj, idx_feat);
100 CERROR("%s: fail to setup index operations for "DFID
101 " rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
102 PFID(lu_object_fid(&obj->do_lu)), rc);
103 lu_object_put(env, &obj->do_lu);
108 local_oid_storage_fini(env, los);
113 * helper function to generate the filename associated with a slave index file
115 static inline int lquota_disk_slv_filename(struct lu_fid *glb_fid,
116 struct obd_uuid *uuid,
119 char *name, *uuid_str;
121 /* In most case, the uuid is NULL terminated */
122 if (uuid->uuid[sizeof(*uuid) - 1] != '\0') {
123 OBD_ALLOC(uuid_str, sizeof(*uuid));
124 if (uuid_str == NULL)
126 memcpy(uuid_str, uuid->uuid, sizeof(*uuid) - 1);
128 uuid_str = (char *)uuid->uuid;
131 /* we strip the slave's UUID (in the form of fsname-OST0001_UUID) of
132 * the filesystem name in case this one is changed in the future */
133 name = strrchr(uuid_str, '-');
135 name = strrchr(uuid_str, ':');
137 CERROR("Failed to extract extract filesystem "
138 "name from UUID %s\n", uuid_str);
139 if (uuid_str != uuid->uuid)
140 OBD_FREE(uuid_str, sizeof(*uuid));
146 /* the filename is composed of the most signicant bits of the global
147 * FID, that's to say the oid which encodes the pool id, pool type and
148 * quota type, followed by the export UUID */
149 sprintf(filename, "0x%x-%s", glb_fid->f_oid, name);
151 if (uuid_str != uuid->uuid)
152 OBD_FREE(uuid_str, sizeof(*uuid));
158 * Set up quota directory (either "quota_master" or "quota_slave") for a QMT or
159 * QSD instance. This function is also used to create per-pool directory on
161 * The directory is created with a local sequence if it does not exist already.
162 * This function is called at ->ldo_prepare time when the full device stack is
165 * \param env - is the environment passed by the caller
166 * \param dev - is the dt_device where to create the quota directory
167 * \param parent - is the parent directory. If not specified, the directory
168 * will be created under the root directory
169 * \param name - is the name of quota directory to be created
171 * \retval - pointer to quota root dt_object on success, appropriate error
174 struct dt_object *lquota_disk_dir_find_create(const struct lu_env *env,
175 struct dt_device *dev,
176 struct dt_object *parent,
179 struct lquota_thread_info *qti = lquota_info(env);
180 struct dt_object *qt_dir = NULL;
181 struct local_oid_storage *los = NULL;
185 /* Set up local storage to create the quota directory.
186 * We use the sequence reserved for local named objects */
187 lu_local_name_obj_fid(&qti->qti_fid, 1);
188 rc = local_oid_storage_init(env, dev, &qti->qti_fid, &los);
192 if (parent == NULL) {
193 /* Fetch dt object associated with root directory */
194 rc = dt_root_get(env, dev, &qti->qti_fid);
198 parent = dt_locate_at(env, dev, &qti->qti_fid,
199 dev->dd_lu_dev.ld_site->ls_top_dev);
201 GOTO(out, rc = PTR_ERR(parent));
203 lu_object_get(&parent->do_lu);
206 /* create quota directory to be used for all quota index files */
207 qt_dir = local_file_find_or_create(env, los, parent, name, S_IFDIR |
208 S_IRUGO | S_IWUSR | S_IXUGO);
210 GOTO(out, rc = PTR_ERR(qt_dir));
212 /* local_oid_storage_fini() will finalize the local storage device,
213 * we have to open the object in another device stack */
214 qti->qti_fid = qt_dir->do_lu.lo_header->loh_fid;
215 lu_object_put_nocache(env, &qt_dir->do_lu);
216 qt_dir = dt_locate(env, dev, &qti->qti_fid);
218 GOTO(out, rc = PTR_ERR(qt_dir));
220 if (!dt_try_as_dir(env, qt_dir))
221 GOTO(out, rc = -ENOTDIR);
224 if (parent != NULL && !IS_ERR(parent))
225 lu_object_put(env, &parent->do_lu);
227 local_oid_storage_fini(env, los);
229 if (qt_dir != NULL && !IS_ERR(qt_dir))
230 lu_object_put(env, &qt_dir->do_lu);
231 qt_dir = ERR_PTR(rc);
237 * Look-up/create a global index file.
239 * \param env - is the environment passed by the caller
240 * \parap dev - is the dt_device where to lookup/create the global index file
241 * \param parent - is the parent directory where to create the global index if
243 * \param fid - is the fid of the global index to be looked up/created
244 * \parap local - indicates whether the index should be created with a local
245 * generated fid or with \fid
247 * \retval - pointer to the dt_object of the global index on success,
248 * appropriate error on failure
250 struct dt_object *lquota_disk_glb_find_create(const struct lu_env *env,
251 struct dt_device *dev,
252 struct dt_object *parent,
253 struct lu_fid *fid, bool local)
255 struct lquota_thread_info *qti = lquota_info(env);
256 struct dt_object *glb_idx;
257 const struct dt_index_features *idx_feat;
260 CDEBUG(D_QUOTA, "look-up/create %sglobal idx file ("DFID")\n",
261 local ? "local " : "", PFID(fid));
263 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2,7,50,0)
264 /* we use different index feature for each quota type and target type
265 * for the time being. This is done for on-disk conversion from the old
266 * quota format. Once this is no longer required, we should just be
267 * using dt_quota_glb_features for all global index file */
268 idx_feat = glb_idx_feature(fid);
270 #warning "remove old quota compatibility code"
271 idx_feat = &dt_quota_glb_features;
274 /* the filename is composed of the most signicant bits of the FID,
275 * that's to say the oid which encodes the pool id, pool type and quota
277 sprintf(qti->qti_buf, "0x%x", fid->f_oid);
280 /* We use the sequence reserved for local named objects */
281 lu_local_name_obj_fid(&qti->qti_fid, 1);
282 glb_idx = lquota_disk_find_create(env, dev, parent,
283 &qti->qti_fid, idx_feat,
286 /* look-up/create global index on disk */
287 glb_idx = local_index_find_or_create_with_fid(env, dev, fid,
295 CERROR("%s: failed to look-up/create idx file "DFID" rc:%ld "
296 "local:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
297 PFID(fid), PTR_ERR(glb_idx), local);
303 * Look-up a slave index file.
305 * \param env - is the environment passed by the caller
306 * \param dev - is the backend dt_device where to look-up/create the slave index
307 * \param parent - is the parent directory where to lookup the slave index
308 * \param glb_fid - is the fid of the global index file associated with this
310 * \param uuid - is the uuid of slave which is (re)connecting to the master
313 * \retval - pointer to the dt_object of the slave index on success,
314 * appropriate error on failure
316 struct dt_object *lquota_disk_slv_find(const struct lu_env *env,
317 struct dt_device *dev,
318 struct dt_object *parent,
319 struct lu_fid *glb_fid,
320 struct obd_uuid *uuid)
322 struct lquota_thread_info *qti = lquota_info(env);
323 struct dt_object *slv_idx;
327 LASSERT(uuid != NULL);
329 CDEBUG(D_QUOTA, "lookup slave index file for %s\n",
332 /* generate filename associated with the slave */
333 rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
337 /* lookup slave index file */
338 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
342 /* name is found, get the object */
343 slv_idx = dt_locate(env, dev, &qti->qti_fid);
347 if (slv_idx->do_index_ops == NULL) {
348 rc = slv_idx->do_ops->do_index_try(env, slv_idx,
349 &dt_quota_slv_features);
351 CERROR("%s: failed to setup slave index operations for "
352 "%s, rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
353 obd_uuid2str(uuid), rc);
354 lu_object_put(env, &slv_idx->do_lu);
355 slv_idx = ERR_PTR(rc);
363 * Look-up a slave index file. If the slave index isn't found:
364 * - if local is set to false, we allocate a FID from FID_SEQ_QUOTA sequence and
366 * - otherwise, we create the index file with a local reserved FID (see
369 * \param env - is the environment passed by the caller
370 * \param dev - is the backend dt_device where to look-up/create the slave index
371 * \param parent - is the parent directory where to create the slave index if
372 * it does not exist already
373 * \param glb_fid - is the fid of the global index file associated with this
375 * \param uuid - is the uuid of slave which is (re)connecting to the master
377 * \param local - indicate whether to use local reserved FID (LQUOTA_USR_OID
378 * & LQUOTA_GRP_OID) for the slave index creation or to
379 * allocate a new fid from sequence FID_SEQ_QUOTA
381 * \retval - pointer to the dt_object of the slave index on success,
382 * appropriate error on failure
384 struct dt_object *lquota_disk_slv_find_create(const struct lu_env *env,
385 struct dt_device *dev,
386 struct dt_object *parent,
387 struct lu_fid *glb_fid,
388 struct obd_uuid *uuid,
391 struct lquota_thread_info *qti = lquota_info(env);
392 struct dt_object *slv_idx;
396 LASSERT(uuid != NULL);
398 CDEBUG(D_QUOTA, "lookup/create slave index file for %s\n",
401 /* generate filename associated with the slave */
402 rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
406 /* Slave indexes uses the FID_SEQ_QUOTA sequence since they can be read
407 * through the network */
408 qti->qti_fid.f_seq = FID_SEQ_QUOTA;
409 qti->qti_fid.f_ver = 0;
413 rc = lquota_extract_fid(glb_fid, NULL, NULL, &type);
417 /* use predefined fid in the reserved oid list */
418 qti->qti_fid.f_oid = (type == USRQUOTA) ? LQUOTA_USR_OID
421 slv_idx = local_index_find_or_create_with_fid(env, dev,
426 &dt_quota_slv_features);
428 /* allocate fid dynamically if index does not exist already */
429 qti->qti_fid.f_oid = LQUOTA_GENERATED_OID;
431 /* lookup/create slave index file */
432 slv_idx = lquota_disk_find_create(env, dev, parent,
434 &dt_quota_slv_features,
442 * Iterate over all slave index files associated with global index \glb_fid and
443 * invoke a callback function for each slave index file.
445 * \param env - is the environment passed by the caller
446 * \param parent - is the parent directory where the slave index files are
448 * \param glb_fid - is the fid of the global index file associated with the
449 * slave indexes to scan
450 * \param func - is the callback function to call each time a slave index
452 * \param arg - is an opaq argument passed to the callback function \func
454 int lquota_disk_for_each_slv(const struct lu_env *env, struct dt_object *parent,
455 struct lu_fid *glb_fid, lquota_disk_slv_cb_t func,
458 struct lquota_thread_info *qti = lquota_info(env);
460 const struct dt_it_ops *iops;
465 OBD_ALLOC(name, sizeof("0x00000000-"));
469 /* filename associated with slave index files are prefixed with the most
470 * signicant bits of the global FID */
471 sprintf(name, "0x%x-", glb_fid->f_oid);
473 iops = &parent->do_index_ops->dio_it;
474 it = iops->init(env, parent, 0, BYPASS_CAPA);
476 OBD_FREE(name, sizeof("0x00000000-"));
480 rc = iops->load(env, it, 0);
483 * Iterator didn't find record with exactly the key requested.
485 * It is currently either
487 * - positioned above record with key less than
488 * requested---skip it.
490 * - or not positioned at all (is in IAM_IT_SKEWED
491 * state)---position it on the next item.
493 rc = iops->next(env, it);
501 len = iops->key_size(env, it);
502 /* IAM iterator can return record with zero len. */
503 if (len == 0 || len <= strlen(name) || len >= LQUOTA_NAME_MAX)
506 key = iops->key(env, it);
512 if (strncmp((char *)key, name, strlen(name)) != 0)
515 /* ldiskfs OSD returns filename as stored in directory entry
516 * which does not end up with '\0' */
517 memcpy(&qti->qti_buf, key, len);
518 qti->qti_buf[len] = '\0';
520 /* lookup fid associated with this slave index file */
521 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
525 if (qti->qti_fid.f_seq != FID_SEQ_QUOTA)
528 rc = func(env, glb_fid, (char *)key, &qti->qti_fid, arg);
533 rc = iops->next(env, it);
534 } while (rc == -ESTALE);
539 OBD_FREE(name, sizeof("0x00000000-"));
546 * Retrieve quota settings from disk for a particular identifier.
548 * \param env - is the environment passed by the caller
549 * \param obj - is the on-disk index where quota settings are stored.
550 * \param id - is the key to be updated
551 * \param rec - is the output record where to store quota settings.
553 * \retval - 0 on success, appropriate error on failure
555 int lquota_disk_read(const struct lu_env *env, struct dt_object *obj,
556 union lquota_id *id, struct dt_rec *rec)
561 LASSERT(dt_object_exists(obj));
562 LASSERT(obj->do_index_ops != NULL);
564 /* lookup on-disk record from index file */
565 dt_read_lock(env, obj, 0);
566 rc = dt_lookup(env, obj, rec, (struct dt_key *)&id->qid_uid,
568 dt_read_unlock(env, obj);
574 * Reserve enough credits to update a record in a quota index file.
576 * \param env - is the environment passed by the caller
577 * \param th - is the transaction to use for disk writes
578 * \param obj - is the on-disk index where quota settings are stored.
579 * \param id - is the key to be updated
581 * \retval - 0 on success, appropriate error on failure
583 int lquota_disk_declare_write(const struct lu_env *env, struct thandle *th,
584 struct dt_object *obj, union lquota_id *id)
586 struct lquota_thread_info *qti = lquota_info(env);
587 struct dt_key *key = (struct dt_key *)&id->qid_uid;
591 LASSERT(dt_object_exists(obj));
592 LASSERT(obj->do_index_ops != NULL);
594 /* speculative delete declaration in case there is already an existing
595 * record in the index */
596 rc = dt_declare_delete(env, obj, key, th);
600 /* declare insertion of updated record */
601 rc = dt_declare_insert(env, obj, (struct dt_rec *)&qti->qti_rec, key,
606 /* we might have to update the version of the global index too */
607 rc = dt_declare_version_set(env, obj, th);
613 * Update a record in a quota index file.
615 * \param env - is the environment passed by the caller
616 * \param th - is the transaction to use for disk writes
617 * \param obj - is the on-disk index to be updated.
618 * \param id - is the key to be updated
619 * \param rec - is the input record containing the new quota settings.
620 * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
621 * \param ver - is the new version of the index if LQUOTA_SET_VER is set or is
622 * used to return the new version of the index when
623 * LQUOTA_BUMP_VER is set.
625 * \retval - 0 on success, appropriate error on failure
627 int lquota_disk_write(const struct lu_env *env, struct thandle *th,
628 struct dt_object *obj, union lquota_id *id,
629 struct dt_rec *rec, __u32 flags, __u64 *ver)
631 struct lquota_thread_info *qti = lquota_info(env);
632 struct dt_key *key = (struct dt_key *)&id->qid_uid;
636 LASSERT(dt_object_exists(obj));
637 LASSERT(obj->do_index_ops != NULL);
640 dt_write_lock(env, obj, 0);
642 /* check whether there is already an existing record for this ID */
643 rc = dt_lookup(env, obj, (struct dt_rec *)&qti->qti_rec, key,
646 /* delete existing record in order to replace it */
647 rc = dt_delete(env, obj, key, th, BYPASS_CAPA);
650 } else if (rc == -ENOENT) {
651 /* probably first insert */
658 /* insert record with updated quota settings */
659 rc = dt_insert(env, obj, rec, key, th, BYPASS_CAPA, 1);
661 /* try to insert the old one */
662 rc = dt_insert(env, obj, (struct dt_rec *)&qti->qti_rec,
663 key, th, BYPASS_CAPA, 1);
664 LASSERTF(rc == 0, "failed to insert record in quota "
666 PFID(lu_object_fid(&obj->do_lu)));
673 if (flags & LQUOTA_BUMP_VER) {
674 /* caller wants to bump the version, let's first read
676 *ver = dt_version_get(env, obj);
679 LASSERT(flags & LQUOTA_SET_VER);
681 dt_version_set(env, obj, *ver, th);
686 dt_write_unlock(env, obj);
691 * Update version of an index file
693 * \param env - is the environment passed by the caller
694 * \param dev - is the backend dt device storing the index file
695 * \param obj - is the on-disk index that should be updated
696 * \param ver - is the new version
698 int lquota_disk_update_ver(const struct lu_env *env, struct dt_device *dev,
699 struct dt_object *obj, __u64 ver)
705 th = dt_trans_create(env, dev);
709 rc = dt_declare_version_set(env, obj, th);
713 rc = dt_trans_start_local(env, dev, th);
718 dt_version_set(env, obj, ver, th);
721 dt_trans_stop(env, dev, th);