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
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 local_oid_storage_fini(env, los);
101 * helper function to generate the filename associated with a slave index file
103 static inline int lquota_disk_slv_filename(const struct lu_fid *glb_fid,
104 struct obd_uuid *uuid,
107 char *name, *uuid_str;
109 /* In most case, the uuid is NULL terminated */
110 if (uuid->uuid[sizeof(*uuid) - 1] != '\0') {
111 OBD_ALLOC(uuid_str, sizeof(*uuid));
112 if (uuid_str == NULL)
114 memcpy(uuid_str, uuid->uuid, sizeof(*uuid) - 1);
116 uuid_str = (char *)uuid->uuid;
119 /* we strip the slave's UUID (in the form of fsname-OST0001_UUID) of
120 * the filesystem name in case this one is changed in the future */
121 name = strrchr(uuid_str, '-');
123 name = strrchr(uuid_str, ':');
125 CERROR("Failed to extract extract filesystem "
126 "name from UUID %s\n", uuid_str);
127 if (uuid_str != uuid->uuid)
128 OBD_FREE(uuid_str, sizeof(*uuid));
134 /* the filename is composed of the most signicant bits of the global
135 * FID, that's to say the oid which encodes the pool id, pool type and
136 * quota type, followed by the export UUID */
137 sprintf(filename, "0x%x-%s", glb_fid->f_oid, name);
139 if (uuid_str != uuid->uuid)
140 OBD_FREE(uuid_str, sizeof(*uuid));
146 * Set up quota directory (either "quota_master" or "quota_slave") for a QMT or
147 * QSD instance. This function is also used to create per-pool directory on
149 * The directory is created with a local sequence if it does not exist already.
150 * This function is called at ->ldo_prepare time when the full device stack is
153 * \param env - is the environment passed by the caller
154 * \param dev - is the dt_device where to create the quota directory
155 * \param parent - is the parent directory. If not specified, the directory
156 * will be created under the root directory
157 * \param name - is the name of quota directory to be created
159 * \retval - pointer to quota root dt_object on success, appropriate error
162 struct dt_object *lquota_disk_dir_find_create(const struct lu_env *env,
163 struct dt_device *dev,
164 struct dt_object *parent,
167 struct lquota_thread_info *qti = lquota_info(env);
168 struct dt_object *qt_dir = NULL;
169 struct local_oid_storage *los = NULL;
173 /* Set up local storage to create the quota directory.
174 * We use the sequence reserved for local named objects */
175 lu_local_name_obj_fid(&qti->qti_fid, 1);
176 rc = local_oid_storage_init(env, dev, &qti->qti_fid, &los);
180 if (parent == NULL) {
181 /* Fetch dt object associated with root directory */
182 rc = dt_root_get(env, dev, &qti->qti_fid);
186 parent = dt_locate_at(env, dev, &qti->qti_fid,
187 dev->dd_lu_dev.ld_site->ls_top_dev);
189 GOTO(out, rc = PTR_ERR(parent));
191 lu_object_get(&parent->do_lu);
194 /* create quota directory to be used for all quota index files */
195 qt_dir = local_file_find_or_create(env, los, parent, name, S_IFDIR |
196 S_IRUGO | S_IWUSR | S_IXUGO);
198 GOTO(out, rc = PTR_ERR(qt_dir));
200 /* local_oid_storage_fini() will finalize the local storage device,
201 * we have to open the object in another device stack */
202 qti->qti_fid = qt_dir->do_lu.lo_header->loh_fid;
203 lu_object_put_nocache(env, &qt_dir->do_lu);
204 qt_dir = dt_locate(env, dev, &qti->qti_fid);
206 GOTO(out, rc = PTR_ERR(qt_dir));
208 if (!dt_try_as_dir(env, qt_dir))
209 GOTO(out, rc = -ENOTDIR);
212 if (parent != NULL && !IS_ERR(parent))
213 lu_object_put(env, &parent->do_lu);
215 local_oid_storage_fini(env, los);
217 if (qt_dir != NULL && !IS_ERR(qt_dir))
218 lu_object_put(env, &qt_dir->do_lu);
219 qt_dir = ERR_PTR(rc);
225 * Look-up/create a global index file.
227 * \param env - is the environment passed by the caller
228 * \parap dev - is the dt_device where to lookup/create the global index file
229 * \param parent - is the parent directory where to create the global index if
231 * \param fid - is the fid of the global index to be looked up/created
232 * \parap local - indicates whether the index should be created with a local
233 * generated fid or with \fid
235 * \retval - pointer to the dt_object of the global index on success,
236 * appropriate error on failure
238 struct dt_object *lquota_disk_glb_find_create(const struct lu_env *env,
239 struct dt_device *dev,
240 struct dt_object *parent,
241 struct lu_fid *fid, bool local)
243 struct lquota_thread_info *qti = lquota_info(env);
244 struct dt_object *glb_idx;
245 const struct dt_index_features *idx_feat;
248 CDEBUG(D_QUOTA, "look-up/create %sglobal idx file ("DFID")\n",
249 local ? "local " : "", PFID(fid));
251 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(2,7,50,0)
252 /* we use different index feature for each quota type and target type
253 * for the time being. This is done for on-disk conversion from the old
254 * quota format. Once this is no longer required, we should just be
255 * using dt_quota_glb_features for all global index file */
256 idx_feat = glb_idx_feature(fid);
258 #warning "remove old quota compatibility code"
259 idx_feat = &dt_quota_glb_features;
262 /* the filename is composed of the most signicant bits of the FID,
263 * that's to say the oid which encodes the pool id, pool type and quota
265 sprintf(qti->qti_buf, "0x%x", fid->f_oid);
268 /* We use the sequence reserved for local named objects */
269 lu_local_name_obj_fid(&qti->qti_fid, 1);
270 glb_idx = lquota_disk_find_create(env, dev, parent,
271 &qti->qti_fid, idx_feat,
274 /* look-up/create global index on disk */
275 glb_idx = local_index_find_or_create_with_fid(env, dev, fid,
282 if (IS_ERR(glb_idx)) {
283 CERROR("%s: failed to look-up/create idx file "DFID" rc:%ld "
284 "local:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
285 PFID(fid), PTR_ERR(glb_idx), local);
289 /* install index operation vector */
290 if (glb_idx->do_index_ops == NULL) {
293 rc = glb_idx->do_ops->do_index_try(env, glb_idx, idx_feat);
295 CERROR("%s: failed to setup index operations for "DFID
296 " rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
297 PFID(lu_object_fid(&glb_idx->do_lu)), rc);
298 lu_object_put(env, &glb_idx->do_lu);
299 glb_idx = ERR_PTR(rc);
307 * Look-up a slave index file.
309 * \param env - is the environment passed by the caller
310 * \param dev - is the backend dt_device where to look-up/create the slave index
311 * \param parent - is the parent directory where to lookup the slave index
312 * \param glb_fid - is the fid of the global index file associated with this
314 * \param uuid - is the uuid of slave which is (re)connecting to the master
317 * \retval - pointer to the dt_object of the slave index on success,
318 * appropriate error on failure
320 struct dt_object *lquota_disk_slv_find(const struct lu_env *env,
321 struct dt_device *dev,
322 struct dt_object *parent,
323 const struct lu_fid *glb_fid,
324 struct obd_uuid *uuid)
326 struct lquota_thread_info *qti = lquota_info(env);
327 struct dt_object *slv_idx;
331 LASSERT(uuid != NULL);
333 CDEBUG(D_QUOTA, "lookup slave index file for %s\n",
336 /* generate filename associated with the slave */
337 rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
341 /* lookup slave index file */
342 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
346 /* name is found, get the object */
347 slv_idx = dt_locate(env, dev, &qti->qti_fid);
351 if (slv_idx->do_index_ops == NULL) {
352 rc = slv_idx->do_ops->do_index_try(env, slv_idx,
353 &dt_quota_slv_features);
355 CERROR("%s: failed to setup slave index operations for "
356 "%s, rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
357 obd_uuid2str(uuid), rc);
358 lu_object_put(env, &slv_idx->do_lu);
359 slv_idx = ERR_PTR(rc);
367 * Look-up a slave index file. If the slave index isn't found:
368 * - if local is set to false, we allocate a FID from FID_SEQ_QUOTA sequence and
370 * - otherwise, we create the index file with a local reserved FID (see
373 * \param env - is the environment passed by the caller
374 * \param dev - is the backend dt_device where to look-up/create the slave index
375 * \param parent - is the parent directory where to create the slave index if
376 * it does not exist already
377 * \param glb_fid - is the fid of the global index file associated with this
379 * \param uuid - is the uuid of slave which is (re)connecting to the master
381 * \param local - indicate whether to use local reserved FID (LQUOTA_USR_OID
382 * & LQUOTA_GRP_OID) for the slave index creation or to
383 * allocate a new fid from sequence FID_SEQ_QUOTA
385 * \retval - pointer to the dt_object of the slave index on success,
386 * appropriate error on failure
388 struct dt_object *lquota_disk_slv_find_create(const struct lu_env *env,
389 struct dt_device *dev,
390 struct dt_object *parent,
391 struct lu_fid *glb_fid,
392 struct obd_uuid *uuid,
395 struct lquota_thread_info *qti = lquota_info(env);
396 struct dt_object *slv_idx;
400 LASSERT(uuid != NULL);
402 CDEBUG(D_QUOTA, "lookup/create slave index file for %s\n",
405 /* generate filename associated with the slave */
406 rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
410 /* Slave indexes uses the FID_SEQ_QUOTA sequence since they can be read
411 * through the network */
412 qti->qti_fid.f_seq = FID_SEQ_QUOTA;
413 qti->qti_fid.f_ver = 0;
417 rc = lquota_extract_fid(glb_fid, NULL, NULL, &type);
421 /* use predefined fid in the reserved oid list */
422 qti->qti_fid.f_oid = (type == USRQUOTA) ? LQUOTA_USR_OID
425 slv_idx = local_index_find_or_create_with_fid(env, dev,
430 &dt_quota_slv_features);
432 /* allocate fid dynamically if index does not exist already */
433 qti->qti_fid.f_oid = LQUOTA_GENERATED_OID;
435 /* lookup/create slave index file */
436 slv_idx = lquota_disk_find_create(env, dev, parent,
438 &dt_quota_slv_features,
445 /* install index operation vector */
446 if (slv_idx->do_index_ops == NULL) {
447 rc = slv_idx->do_ops->do_index_try(env, slv_idx,
448 &dt_quota_slv_features);
450 CERROR("%s: failed to setup index operations for "DFID
451 " rc:%d\n", dev->dd_lu_dev.ld_obd->obd_name,
452 PFID(lu_object_fid(&slv_idx->do_lu)), rc);
453 lu_object_put(env, &slv_idx->do_lu);
454 slv_idx = ERR_PTR(rc);
462 * Iterate over all slave index files associated with global index \glb_fid and
463 * invoke a callback function for each slave index file.
465 * \param env - is the environment passed by the caller
466 * \param parent - is the parent directory where the slave index files are
468 * \param glb_fid - is the fid of the global index file associated with the
469 * slave indexes to scan
470 * \param func - is the callback function to call each time a slave index
472 * \param arg - is an opaq argument passed to the callback function \func
474 int lquota_disk_for_each_slv(const struct lu_env *env, struct dt_object *parent,
475 struct lu_fid *glb_fid, lquota_disk_slv_cb_t func,
478 struct lquota_thread_info *qti = lquota_info(env);
480 const struct dt_it_ops *iops;
485 OBD_ALLOC(name, LQUOTA_NAME_MAX);
489 /* filename associated with slave index files are prefixed with the most
490 * signicant bits of the global FID */
491 sprintf(name, "0x%x-", glb_fid->f_oid);
493 iops = &parent->do_index_ops->dio_it;
494 it = iops->init(env, parent, 0, BYPASS_CAPA);
496 OBD_FREE(name, LQUOTA_NAME_MAX);
500 rc = iops->load(env, it, 0);
503 * Iterator didn't find record with exactly the key requested.
505 * It is currently either
507 * - positioned above record with key less than
508 * requested---skip it.
510 * - or not positioned at all (is in IAM_IT_SKEWED
511 * state)---position it on the next item.
513 rc = iops->next(env, it);
521 len = iops->key_size(env, it);
522 /* IAM iterator can return record with zero len. */
523 if (len == 0 || len <= strlen(name) || len >= LQUOTA_NAME_MAX)
526 key = iops->key(env, it);
532 if (strncmp((char *)key, name, strlen(name)) != 0)
535 /* ldiskfs OSD returns filename as stored in directory entry
536 * which does not end up with '\0' */
537 memcpy(&qti->qti_buf, key, len);
538 qti->qti_buf[len] = '\0';
540 /* lookup fid associated with this slave index file */
541 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
545 if (qti->qti_fid.f_seq != FID_SEQ_QUOTA)
548 rc = func(env, glb_fid, (char *)key, &qti->qti_fid, arg);
553 rc = iops->next(env, it);
554 } while (rc == -ESTALE);
559 OBD_FREE(name, LQUOTA_NAME_MAX);
566 * Retrieve quota settings from disk for a particular identifier.
568 * \param env - is the environment passed by the caller
569 * \param obj - is the on-disk index where quota settings are stored.
570 * \param id - is the key to be updated
571 * \param rec - is the output record where to store quota settings.
573 * \retval - 0 on success, appropriate error on failure
575 int lquota_disk_read(const struct lu_env *env, struct dt_object *obj,
576 union lquota_id *id, struct dt_rec *rec)
581 LASSERT(dt_object_exists(obj));
582 LASSERT(obj->do_index_ops != NULL);
584 /* lookup on-disk record from index file */
585 dt_read_lock(env, obj, 0);
586 rc = dt_lookup(env, obj, rec, (struct dt_key *)&id->qid_uid,
588 dt_read_unlock(env, obj);
594 * Reserve enough credits to update a record in a quota index file.
596 * \param env - is the environment passed by the caller
597 * \param th - is the transaction to use for disk writes
598 * \param obj - is the on-disk index where quota settings are stored.
599 * \param id - is the key to be updated
601 * \retval - 0 on success, appropriate error on failure
603 int lquota_disk_declare_write(const struct lu_env *env, struct thandle *th,
604 struct dt_object *obj, union lquota_id *id)
606 struct lquota_thread_info *qti = lquota_info(env);
607 struct dt_key *key = (struct dt_key *)&id->qid_uid;
611 LASSERT(dt_object_exists(obj));
612 LASSERT(obj->do_index_ops != NULL);
614 /* speculative delete declaration in case there is already an existing
615 * record in the index */
616 rc = dt_declare_delete(env, obj, key, th);
620 /* declare insertion of updated record */
621 rc = dt_declare_insert(env, obj, (struct dt_rec *)&qti->qti_rec, key,
626 /* we might have to update the version of the global index too */
627 rc = dt_declare_version_set(env, obj, th);
633 * Update a record in a quota index file.
635 * \param env - is the environment passed by the caller
636 * \param th - is the transaction to use for disk writes
637 * \param obj - is the on-disk index to be updated.
638 * \param id - is the key to be updated
639 * \param rec - is the input record containing the new quota settings.
640 * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
641 * \param ver - is the new version of the index if LQUOTA_SET_VER is set or is
642 * used to return the new version of the index when
643 * LQUOTA_BUMP_VER is set.
645 * \retval - 0 on success, appropriate error on failure
647 int lquota_disk_write(const struct lu_env *env, struct thandle *th,
648 struct dt_object *obj, union lquota_id *id,
649 struct dt_rec *rec, __u32 flags, __u64 *ver)
651 struct lquota_thread_info *qti = lquota_info(env);
652 struct dt_key *key = (struct dt_key *)&id->qid_uid;
656 LASSERT(dt_object_exists(obj));
657 LASSERT(obj->do_index_ops != NULL);
660 dt_write_lock(env, obj, 0);
662 /* check whether there is already an existing record for this ID */
663 rc = dt_lookup(env, obj, (struct dt_rec *)&qti->qti_rec, key,
666 /* delete existing record in order to replace it */
667 rc = dt_delete(env, obj, key, th, BYPASS_CAPA);
670 } else if (rc == -ENOENT) {
671 /* probably first insert */
678 /* insert record with updated quota settings */
679 rc = dt_insert(env, obj, rec, key, th, BYPASS_CAPA, 1);
681 /* try to insert the old one */
682 rc = dt_insert(env, obj, (struct dt_rec *)&qti->qti_rec,
683 key, th, BYPASS_CAPA, 1);
684 LASSERTF(rc == 0, "failed to insert record in quota "
686 PFID(lu_object_fid(&obj->do_lu)));
693 if (flags & LQUOTA_BUMP_VER) {
694 /* caller wants to bump the version, let's first read
696 *ver = dt_version_get(env, obj);
699 LASSERT(flags & LQUOTA_SET_VER);
701 dt_version_set(env, obj, *ver, th);
706 dt_write_unlock(env, obj);
711 * Update version of an index file
713 * \param env - is the environment passed by the caller
714 * \param dev - is the backend dt device storing the index file
715 * \param obj - is the on-disk index that should be updated
716 * \param ver - is the new version
718 int lquota_disk_update_ver(const struct lu_env *env, struct dt_device *dev,
719 struct dt_object *obj, __u64 ver)
725 th = dt_trans_create(env, dev);
729 rc = dt_declare_version_set(env, obj, th);
733 rc = dt_trans_start_local(env, dev, th);
738 dt_version_set(env, obj, ver, th);
741 dt_trans_stop(env, dev, th);
746 * Write a global record
748 * \param env - is the environment passed by the caller
749 * \param obj - is the on-disk global index to be updated
750 * \param id - index to be updated
751 * \param rec - record to be written
753 int lquota_disk_write_glb(const struct lu_env *env, struct dt_object *obj,
754 __u64 id, struct lquota_glb_rec *rec)
756 struct dt_device *dev = lu2dt_dev(obj->do_lu.lo_dev);
758 struct dt_key *key = (struct dt_key *)&id;
762 th = dt_trans_create(env, dev);
766 /* the entry with 0 key can always be found in IAM file. */
768 rc = dt_declare_delete(env, obj, key, th);
773 rc = dt_declare_insert(env, obj, (struct dt_rec *)rec, key, th);
777 rc = dt_trans_start_local(env, dev, th);
781 dt_write_lock(env, obj, 0);
784 struct lquota_glb_rec *tmp;
788 GOTO(out_lock, rc = -ENOMEM);
790 rc = dt_lookup(env, obj, (struct dt_rec *)tmp, key,
795 rc = dt_delete(env, obj, key, th, BYPASS_CAPA);
802 rc = dt_insert(env, obj, (struct dt_rec *)rec, key, th, BYPASS_CAPA, 1);
804 dt_write_unlock(env, obj);
806 dt_trans_stop(env, dev, th);
809 EXPORT_SYMBOL(lquota_disk_write_glb);