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, 2013, Intel Corporation.
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
51 #define DEBUG_SUBSYSTEM S_LQUOTA
53 #include "lquota_internal.h"
55 #define LQUOTA_MODE (S_IFREG | S_IRUGO | S_IWUSR)
58 * Helper function looking up & creating if not found an index file with a
61 static struct dt_object *
62 lquota_disk_find_create(const struct lu_env *env, struct dt_device *dev,
63 struct dt_object *parent, struct lu_fid *fid,
64 const struct dt_index_features *idx_feat,
67 struct lquota_thread_info *qti = lquota_info(env);
68 struct dt_object *obj;
69 struct local_oid_storage *los;
73 /* Set up local storage */
74 rc = local_oid_storage_init(env, dev, fid, &los);
78 /* lookup/create slave index file */
79 obj = local_index_find_or_create(env, los, parent, name, LQUOTA_MODE,
84 /* local_oid_storage_fini() will finalize the local storage device,
85 * we have to open the object in another device stack */
86 qti->qti_fid = obj->do_lu.lo_header->loh_fid;
87 lu_object_put_nocache(env, &obj->do_lu);
88 obj = dt_locate(env, dev, &qti->qti_fid);
92 local_oid_storage_fini(env, los);
97 * helper function to generate the filename associated with a slave index file
99 static inline int lquota_disk_slv_filename(const struct lu_fid *glb_fid,
100 struct obd_uuid *uuid,
103 char *name, *uuid_str;
105 /* In most case, the uuid is NULL terminated */
106 if (uuid->uuid[sizeof(*uuid) - 1] != '\0') {
107 OBD_ALLOC(uuid_str, sizeof(*uuid));
108 if (uuid_str == NULL)
110 memcpy(uuid_str, uuid->uuid, sizeof(*uuid) - 1);
112 uuid_str = (char *)uuid->uuid;
115 /* we strip the slave's UUID (in the form of fsname-OST0001_UUID) of
116 * the filesystem name in case this one is changed in the future */
117 name = strrchr(uuid_str, '-');
119 name = strrchr(uuid_str, ':');
121 CERROR("Failed to extract extract filesystem "
122 "name from UUID %s\n", uuid_str);
123 if (uuid_str != uuid->uuid)
124 OBD_FREE(uuid_str, sizeof(*uuid));
130 /* the filename is composed of the most signicant bits of the global
131 * FID, that's to say the oid which encodes the pool id, pool type and
132 * quota type, followed by the export UUID */
133 sprintf(filename, "0x%x-%s", glb_fid->f_oid, name);
135 if (uuid_str != uuid->uuid)
136 OBD_FREE(uuid_str, sizeof(*uuid));
142 * Set up quota directory (either "quota_master" or "quota_slave") for a QMT or
143 * QSD instance. This function is also used to create per-pool directory on
145 * The directory is created with a local sequence if it does not exist already.
146 * This function is called at ->ldo_prepare time when the full device stack is
149 * \param env - is the environment passed by the caller
150 * \param dev - is the dt_device where to create the quota directory
151 * \param parent - is the parent directory. If not specified, the directory
152 * will be created under the root directory
153 * \param name - is the name of quota directory to be created
155 * \retval - pointer to quota root dt_object on success, appropriate error
158 struct dt_object *lquota_disk_dir_find_create(const struct lu_env *env,
159 struct dt_device *dev,
160 struct dt_object *parent,
163 struct lquota_thread_info *qti = lquota_info(env);
164 struct dt_object *qt_dir = NULL;
165 struct local_oid_storage *los = NULL;
169 /* Set up local storage to create the quota directory.
170 * We use the sequence reserved for local named objects */
171 lu_local_name_obj_fid(&qti->qti_fid, 1);
172 rc = local_oid_storage_init(env, dev, &qti->qti_fid, &los);
176 if (parent == NULL) {
177 /* Fetch dt object associated with root directory */
178 rc = dt_root_get(env, dev, &qti->qti_fid);
182 parent = dt_locate_at(env, dev, &qti->qti_fid,
183 dev->dd_lu_dev.ld_site->ls_top_dev, NULL);
185 GOTO(out, rc = PTR_ERR(parent));
187 lu_object_get(&parent->do_lu);
190 /* create quota directory to be used for all quota index files */
191 qt_dir = local_file_find_or_create(env, los, parent, name, S_IFDIR |
192 S_IRUGO | S_IWUSR | S_IXUGO);
194 GOTO(out, rc = PTR_ERR(qt_dir));
196 /* local_oid_storage_fini() will finalize the local storage device,
197 * we have to open the object in another device stack */
198 qti->qti_fid = qt_dir->do_lu.lo_header->loh_fid;
199 lu_object_put_nocache(env, &qt_dir->do_lu);
200 qt_dir = dt_locate(env, dev, &qti->qti_fid);
202 GOTO(out, rc = PTR_ERR(qt_dir));
204 if (!dt_try_as_dir(env, qt_dir))
205 GOTO(out, rc = -ENOTDIR);
208 if (parent != NULL && !IS_ERR(parent))
209 lu_object_put(env, &parent->do_lu);
211 local_oid_storage_fini(env, los);
213 if (qt_dir != NULL && !IS_ERR(qt_dir))
214 lu_object_put(env, &qt_dir->do_lu);
215 qt_dir = ERR_PTR(rc);
221 * Look-up/create a global index file.
223 * \param env - is the environment passed by the caller
224 * \parap dev - is the dt_device where to lookup/create the global index file
225 * \param parent - is the parent directory where to create the global index if
227 * \param fid - is the fid of the global index to be looked up/created
228 * \parap local - indicates whether the index should be created with a local
229 * generated fid or with \fid
231 * \retval - pointer to the dt_object of the global index on success,
232 * appropriate error on failure
234 struct dt_object *lquota_disk_glb_find_create(const struct lu_env *env,
235 struct dt_device *dev,
236 struct dt_object *parent,
237 struct lu_fid *fid, bool local)
239 struct lquota_thread_info *qti = lquota_info(env);
240 struct dt_object *glb_idx;
241 const struct dt_index_features *idx_feat;
244 CDEBUG(D_QUOTA, "look-up/create %sglobal idx file ("DFID")\n",
245 local ? "local " : "", PFID(fid));
247 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2, 7, 53, 0)
248 /* we use different index feature for each quota type and target type
249 * for the time being. This is done for on-disk conversion from the old
250 * quota format. Once this is no longer required, we should just be
251 * using dt_quota_glb_features for all global index file */
252 idx_feat = glb_idx_feature(fid);
254 idx_feat = &dt_quota_glb_features;
257 /* the filename is composed of the most signicant bits of the FID,
258 * that's to say the oid which encodes the pool id, pool type and quota
260 sprintf(qti->qti_buf, "0x%x", fid->f_oid);
263 /* We use the sequence reserved for local named objects */
264 lu_local_name_obj_fid(&qti->qti_fid, 1);
265 glb_idx = lquota_disk_find_create(env, dev, parent,
266 &qti->qti_fid, idx_feat,
269 /* look-up/create global index on disk */
270 glb_idx = local_index_find_or_create_with_fid(env, dev, fid,
277 if (IS_ERR(glb_idx)) {
278 CERROR("%s: failed to look-up/create idx file "DFID" rc:%ld "
279 "local:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
280 PFID(fid), PTR_ERR(glb_idx), local);
284 /* install index operation vector */
285 if (glb_idx->do_index_ops == NULL) {
288 rc = glb_idx->do_ops->do_index_try(env, glb_idx, idx_feat);
290 CERROR("%s: failed to setup index operations for "DFID
291 " rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
292 PFID(lu_object_fid(&glb_idx->do_lu)), rc);
293 lu_object_put(env, &glb_idx->do_lu);
294 glb_idx = ERR_PTR(rc);
302 * Look-up a slave index file.
304 * \param env - is the environment passed by the caller
305 * \param dev - is the backend dt_device where to look-up/create the slave index
306 * \param parent - is the parent directory where to lookup the slave index
307 * \param glb_fid - is the fid of the global index file associated with this
309 * \param uuid - is the uuid of slave which is (re)connecting to the master
312 * \retval - pointer to the dt_object of the slave index on success,
313 * appropriate error on failure
315 struct dt_object *lquota_disk_slv_find(const struct lu_env *env,
316 struct dt_device *dev,
317 struct dt_object *parent,
318 const struct lu_fid *glb_fid,
319 struct obd_uuid *uuid)
321 struct lquota_thread_info *qti = lquota_info(env);
322 struct dt_object *slv_idx;
326 LASSERT(uuid != NULL);
328 CDEBUG(D_QUOTA, "lookup slave index file for %s\n",
331 /* generate filename associated with the slave */
332 rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
336 /* lookup slave index file */
337 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
341 /* name is found, get the object */
342 slv_idx = dt_locate(env, dev, &qti->qti_fid);
346 if (slv_idx->do_index_ops == NULL) {
347 rc = slv_idx->do_ops->do_index_try(env, slv_idx,
348 &dt_quota_slv_features);
350 CERROR("%s: failed to setup slave index operations for "
351 "%s, rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
352 obd_uuid2str(uuid), rc);
353 lu_object_put(env, &slv_idx->do_lu);
354 slv_idx = ERR_PTR(rc);
362 * Look-up a slave index file. If the slave index isn't found:
363 * - if local is set to false, we allocate a FID from FID_SEQ_QUOTA sequence and
365 * - otherwise, we create the index file with a local reserved FID (see
368 * \param env - is the environment passed by the caller
369 * \param dev - is the backend dt_device where to look-up/create the slave index
370 * \param parent - is the parent directory where to create the slave index if
371 * it does not exist already
372 * \param glb_fid - is the fid of the global index file associated with this
374 * \param uuid - is the uuid of slave which is (re)connecting to the master
376 * \param local - indicate whether to use local reserved FID (LQUOTA_USR_OID
377 * & LQUOTA_GRP_OID) for the slave index creation or to
378 * allocate a new fid from sequence FID_SEQ_QUOTA
380 * \retval - pointer to the dt_object of the slave index on success,
381 * appropriate error on failure
383 struct dt_object *lquota_disk_slv_find_create(const struct lu_env *env,
384 struct dt_device *dev,
385 struct dt_object *parent,
386 struct lu_fid *glb_fid,
387 struct obd_uuid *uuid,
390 struct lquota_thread_info *qti = lquota_info(env);
391 struct dt_object *slv_idx;
395 LASSERT(uuid != NULL);
397 CDEBUG(D_QUOTA, "lookup/create slave index file for %s\n",
400 /* generate filename associated with the slave */
401 rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
405 /* Slave indexes uses the FID_SEQ_QUOTA sequence since they can be read
406 * through the network */
407 qti->qti_fid.f_seq = FID_SEQ_QUOTA;
408 qti->qti_fid.f_ver = 0;
412 rc = lquota_extract_fid(glb_fid, NULL, NULL, &type);
416 /* use predefined fid in the reserved oid list */
417 qti->qti_fid.f_oid = (type == USRQUOTA) ? LQUOTA_USR_OID
420 slv_idx = local_index_find_or_create_with_fid(env, dev,
425 &dt_quota_slv_features);
427 /* allocate fid dynamically if index does not exist already */
428 qti->qti_fid.f_oid = LQUOTA_GENERATED_OID;
430 /* lookup/create slave index file */
431 slv_idx = lquota_disk_find_create(env, dev, parent,
433 &dt_quota_slv_features,
440 /* install index operation vector */
441 if (slv_idx->do_index_ops == NULL) {
442 rc = slv_idx->do_ops->do_index_try(env, slv_idx,
443 &dt_quota_slv_features);
445 CERROR("%s: failed to setup index operations for "DFID
446 " rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
447 PFID(lu_object_fid(&slv_idx->do_lu)), rc);
448 lu_object_put(env, &slv_idx->do_lu);
449 slv_idx = ERR_PTR(rc);
457 * Iterate over all slave index files associated with global index \glb_fid and
458 * invoke a callback function for each slave index file.
460 * \param env - is the environment passed by the caller
461 * \param parent - is the parent directory where the slave index files are
463 * \param glb_fid - is the fid of the global index file associated with the
464 * slave indexes to scan
465 * \param func - is the callback function to call each time a slave index
467 * \param arg - is an opaq argument passed to the callback function \func
469 int lquota_disk_for_each_slv(const struct lu_env *env, struct dt_object *parent,
470 struct lu_fid *glb_fid, lquota_disk_slv_cb_t func,
473 struct lquota_thread_info *qti = lquota_info(env);
475 const struct dt_it_ops *iops;
480 OBD_ALLOC(name, LQUOTA_NAME_MAX);
484 /* filename associated with slave index files are prefixed with the most
485 * signicant bits of the global FID */
486 sprintf(name, "0x%x-", glb_fid->f_oid);
488 iops = &parent->do_index_ops->dio_it;
489 it = iops->init(env, parent, 0, BYPASS_CAPA);
491 OBD_FREE(name, LQUOTA_NAME_MAX);
495 rc = iops->load(env, it, 0);
498 * Iterator didn't find record with exactly the key requested.
500 * It is currently either
502 * - positioned above record with key less than
503 * requested---skip it.
505 * - or not positioned at all (is in IAM_IT_SKEWED
506 * state)---position it on the next item.
508 rc = iops->next(env, it);
516 len = iops->key_size(env, it);
517 /* IAM iterator can return record with zero len. */
518 if (len == 0 || len <= strlen(name) || len >= LQUOTA_NAME_MAX)
521 key = iops->key(env, it);
527 if (strncmp((char *)key, name, strlen(name)) != 0)
530 /* ldiskfs OSD returns filename as stored in directory entry
531 * which does not end up with '\0' */
532 memcpy(&qti->qti_buf, key, len);
533 qti->qti_buf[len] = '\0';
535 /* lookup fid associated with this slave index file */
536 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
540 if (qti->qti_fid.f_seq != FID_SEQ_QUOTA)
543 rc = func(env, glb_fid, (char *)key, &qti->qti_fid, arg);
548 rc = iops->next(env, it);
549 } while (rc == -ESTALE);
554 OBD_FREE(name, LQUOTA_NAME_MAX);
561 * Retrieve quota settings from disk for a particular identifier.
563 * \param env - is the environment passed by the caller
564 * \param obj - is the on-disk index where quota settings are stored.
565 * \param id - is the key to be updated
566 * \param rec - is the output record where to store quota settings.
568 * \retval - 0 on success, appropriate error on failure
570 int lquota_disk_read(const struct lu_env *env, struct dt_object *obj,
571 union lquota_id *id, struct dt_rec *rec)
576 LASSERT(dt_object_exists(obj));
577 LASSERT(obj->do_index_ops != NULL);
579 /* lookup on-disk record from index file */
580 dt_read_lock(env, obj, 0);
581 rc = dt_lookup(env, obj, rec, (struct dt_key *)&id->qid_uid,
583 dt_read_unlock(env, obj);
589 * Reserve enough credits to update a record in a quota index file.
591 * \param env - is the environment passed by the caller
592 * \param th - is the transaction to use for disk writes
593 * \param obj - is the on-disk index where quota settings are stored.
594 * \param id - is the key to be updated
596 * \retval - 0 on success, appropriate error on failure
598 int lquota_disk_declare_write(const struct lu_env *env, struct thandle *th,
599 struct dt_object *obj, union lquota_id *id)
601 struct lquota_thread_info *qti = lquota_info(env);
602 struct dt_key *key = (struct dt_key *)&id->qid_uid;
606 LASSERT(dt_object_exists(obj));
607 LASSERT(obj->do_index_ops != NULL);
609 /* speculative delete declaration in case there is already an existing
610 * record in the index */
611 rc = dt_declare_delete(env, obj, key, th);
615 /* declare insertion of updated record */
616 rc = dt_declare_insert(env, obj, (struct dt_rec *)&qti->qti_rec, key,
621 /* we might have to update the version of the global index too */
622 rc = dt_declare_version_set(env, obj, th);
628 * Update a record in a quota index file.
630 * \param env - is the environment passed by the caller
631 * \param th - is the transaction to use for disk writes
632 * \param obj - is the on-disk index to be updated.
633 * \param id - is the key to be updated
634 * \param rec - is the input record containing the new quota settings.
635 * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
636 * \param ver - is the new version of the index if LQUOTA_SET_VER is set or is
637 * used to return the new version of the index when
638 * LQUOTA_BUMP_VER is set.
640 * \retval - 0 on success, appropriate error on failure
642 int lquota_disk_write(const struct lu_env *env, struct thandle *th,
643 struct dt_object *obj, union lquota_id *id,
644 struct dt_rec *rec, __u32 flags, __u64 *ver)
646 struct lquota_thread_info *qti = lquota_info(env);
647 struct dt_key *key = (struct dt_key *)&id->qid_uid;
651 LASSERT(dt_object_exists(obj));
652 LASSERT(obj->do_index_ops != NULL);
655 dt_write_lock(env, obj, 0);
657 /* check whether there is already an existing record for this ID */
658 rc = dt_lookup(env, obj, (struct dt_rec *)&qti->qti_rec, key,
661 /* delete existing record in order to replace it */
662 rc = dt_delete(env, obj, key, th, BYPASS_CAPA);
665 } else if (rc == -ENOENT) {
666 /* probably first insert */
673 /* insert record with updated quota settings */
674 rc = dt_insert(env, obj, rec, key, th, BYPASS_CAPA, 1);
676 /* try to insert the old one */
677 rc = dt_insert(env, obj, (struct dt_rec *)&qti->qti_rec,
678 key, th, BYPASS_CAPA, 1);
679 LASSERTF(rc == 0, "failed to insert record in quota "
681 PFID(lu_object_fid(&obj->do_lu)));
688 if (flags & LQUOTA_BUMP_VER) {
689 /* caller wants to bump the version, let's first read
691 *ver = dt_version_get(env, obj);
694 LASSERT(flags & LQUOTA_SET_VER);
696 dt_version_set(env, obj, *ver, th);
701 dt_write_unlock(env, obj);
706 * Update version of an index file
708 * \param env - is the environment passed by the caller
709 * \param dev - is the backend dt device storing the index file
710 * \param obj - is the on-disk index that should be updated
711 * \param ver - is the new version
713 int lquota_disk_update_ver(const struct lu_env *env, struct dt_device *dev,
714 struct dt_object *obj, __u64 ver)
720 th = dt_trans_create(env, dev);
724 rc = dt_declare_version_set(env, obj, th);
728 rc = dt_trans_start_local(env, dev, th);
733 dt_version_set(env, obj, ver, th);
736 dt_trans_stop(env, dev, th);
741 * Write a global record
743 * \param env - is the environment passed by the caller
744 * \param obj - is the on-disk global index to be updated
745 * \param id - index to be updated
746 * \param rec - record to be written
748 int lquota_disk_write_glb(const struct lu_env *env, struct dt_object *obj,
749 __u64 id, struct lquota_glb_rec *rec)
751 struct dt_device *dev = lu2dt_dev(obj->do_lu.lo_dev);
753 struct dt_key *key = (struct dt_key *)&id;
757 th = dt_trans_create(env, dev);
761 /* the entry with 0 key can always be found in IAM file. */
763 rc = dt_declare_delete(env, obj, key, th);
768 rc = dt_declare_insert(env, obj, (struct dt_rec *)rec, key, th);
772 rc = dt_trans_start_local(env, dev, th);
776 dt_write_lock(env, obj, 0);
779 struct lquota_glb_rec *tmp;
783 GOTO(out_lock, rc = -ENOMEM);
785 rc = dt_lookup(env, obj, (struct dt_rec *)tmp, key,
790 rc = dt_delete(env, obj, key, th, BYPASS_CAPA);
797 rc = dt_insert(env, obj, (struct dt_rec *)rec, key, th, BYPASS_CAPA, 1);
799 dt_write_unlock(env, obj);
801 dt_trans_stop(env, dev, th);
804 EXPORT_SYMBOL(lquota_disk_write_glb);