lustre/quota/lquota_disk.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   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.
   9  *
  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).
  15  *
  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
  20  *
  21  * GPL HEADER END
  22  */
  23 /*
  24  * Copyright (c) 2012, 2014, Intel Corporation.
  25  * Use is subject to license terms.
  26  *
  27  * Author: Johann Lombardi <johann.lombardi@intel.com>
  28  * Author: Niu    Yawei    <yawei.niu@intel.com>
  29  */
  30
  31 /*
  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:
  34  *
  35  * - lquota_disk_dir_find_create: look-up quota directory, create it if not
  36  *                                found.
  37  * - lquota_disk_glb_find_create: look-up global index file, create it if not
  38  *                                found.
  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
  42  *                                it does not exist.
  43  * - lquota_disk_for_each_slv:    iterate over all existing slave index files
  44  * - lquota_disk_read:            read quota settings from an index file
  45  * - lquota_disk_declare_write:   reserve credits to update a record in an index
  46  *                                file
  47  * - lquota_disk_write:           update a record in an index file
  48  * - lquota_disk_update_ver:      update version of an index file
  49  */
  50
  51 #define DEBUG_SUBSYSTEM S_LQUOTA
  52
  53 #include "lquota_internal.h"
  54
  55 #define LQUOTA_MODE (S_IFREG | S_IRUGO | S_IWUSR)
  56
  57 /*
  58  * Helper function looking up & creating if not found an index file with a
  59  * dynamic fid.
  60  */
  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,
  65                         char *name)
  66 {
  67         struct lquota_thread_info       *qti = lquota_info(env);
  68         struct dt_object                *obj;
  69         struct local_oid_storage        *los;
  70         int                              rc;
  71         ENTRY;
  72
  73         /* Set up local storage */
  74         rc = local_oid_storage_init(env, dev, fid, &los);
  75         if (rc)
  76                 RETURN(ERR_PTR(rc));
  77
  78         /* lookup/create slave index file */
  79         obj = local_index_find_or_create(env, los, parent, name, LQUOTA_MODE,
  80                                          idx_feat);
  81         if (IS_ERR(obj))
  82                 GOTO(out, obj);
  83
  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);
  89         if (IS_ERR(obj))
  90                 GOTO(out, obj);
  91 out:
  92         local_oid_storage_fini(env, los);
  93         RETURN(obj);
  94 }
  95
  96 /*
  97  * helper function to generate the filename associated with a slave index file
  98  */
  99 static inline int lquota_disk_slv_filename(const struct lu_fid *glb_fid,
 100                                            struct obd_uuid *uuid,
 101                                            char *filename)
 102 {
 103         char    *name, *uuid_str;
 104
 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)
 109                         RETURN(-ENOMEM);
 110                 memcpy(uuid_str, uuid->uuid, sizeof(*uuid) - 1);
 111         } else {
 112                 uuid_str = (char *)uuid->uuid;
 113         }
 114
 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, '-');
 118         if (name == NULL) {
 119                 name = strrchr(uuid_str, ':');
 120                 if (name == NULL) {
 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));
 125                         return -EINVAL;
 126                 }
 127         }
 128         name++;
 129
 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);
 134
 135         if (uuid_str != uuid->uuid)
 136                 OBD_FREE(uuid_str, sizeof(*uuid));
 137
 138         return 0;
 139 }
 140
 141 /*
 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
 144  * the quota master.
 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
 147  * configured.
 148  *
 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
 154  *
 155  * \retval     - pointer to quota root dt_object on success, appropriate error
 156  *               on failure
 157  */
 158 struct dt_object *lquota_disk_dir_find_create(const struct lu_env *env,
 159                                               struct dt_device *dev,
 160                                               struct dt_object *parent,
 161                                               const char *name)
 162 {
 163         struct lquota_thread_info       *qti = lquota_info(env);
 164         struct dt_object                *qt_dir = NULL;
 165         struct local_oid_storage        *los = NULL;
 166         int                              rc;
 167         ENTRY;
 168
 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);
 173         if (rc)
 174                 RETURN(ERR_PTR(rc));
 175
 176         if (parent == NULL) {
 177                 /* Fetch dt object associated with root directory */
 178                 rc = dt_root_get(env, dev, &qti->qti_fid);
 179                 if (rc)
 180                         GOTO(out, rc);
 181
 182                 parent = dt_locate_at(env, dev, &qti->qti_fid,
 183                                       dev->dd_lu_dev.ld_site->ls_top_dev, NULL);
 184                 if (IS_ERR(parent))
 185                         GOTO(out, rc = PTR_ERR(parent));
 186         } else {
 187                 lu_object_get(&parent->do_lu);
 188         }
 189
 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);
 193         if (IS_ERR(qt_dir))
 194                 GOTO(out, rc = PTR_ERR(qt_dir));
 195
 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);
 201         if (IS_ERR(qt_dir))
 202                 GOTO(out, rc = PTR_ERR(qt_dir));
 203
 204         if (!dt_try_as_dir(env, qt_dir))
 205                 GOTO(out, rc = -ENOTDIR);
 206         EXIT;
 207 out:
 208         if (parent != NULL && !IS_ERR(parent))
 209                 lu_object_put(env, &parent->do_lu);
 210         if (los != NULL)
 211                 local_oid_storage_fini(env, los);
 212         if (rc) {
 213                 if (qt_dir != NULL && !IS_ERR(qt_dir))
 214                         lu_object_put(env, &qt_dir->do_lu);
 215                 qt_dir = ERR_PTR(rc);
 216         }
 217         return qt_dir;
 218 }
 219
 220 /*
 221  * Look-up/create a global index file.
 222  *
 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
 226  *                 not found
 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
 230  *
 231  * \retval     - pointer to the dt_object of the global index on success,
 232  *               appropriate error on failure
 233  */
 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)
 238 {
 239         struct lquota_thread_info       *qti = lquota_info(env);
 240         struct dt_object                *glb_idx;
 241         const struct dt_index_features  *idx_feat;
 242         ENTRY;
 243
 244         CDEBUG(D_QUOTA, "look-up/create %sglobal idx file ("DFID")\n",
 245                local ? "local " : "", PFID(fid));
 246
 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);
 253 #else
 254         idx_feat = &dt_quota_glb_features;
 255 #endif
 256
 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
 259          * type */
 260         sprintf(qti->qti_buf, "0x%x", fid->f_oid);
 261
 262         if (local) {
 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,
 267                                                   qti->qti_buf);
 268         } else {
 269                 /* look-up/create global index on disk */
 270                 glb_idx = local_index_find_or_create_with_fid(env, dev, fid,
 271                                                               parent,
 272                                                               qti->qti_buf,
 273                                                               LQUOTA_MODE,
 274                                                               idx_feat);
 275         }
 276
 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);
 281                 RETURN(glb_idx);
 282         }
 283
 284         /* install index operation vector */
 285         if (glb_idx->do_index_ops == NULL) {
 286                 int rc;
 287
 288                 rc = glb_idx->do_ops->do_index_try(env, glb_idx, idx_feat);
 289                 if (rc) {
 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);
 295                 }
 296         }
 297
 298         RETURN(glb_idx);
 299 }
 300
 301 /*
 302  * Look-up a slave index file.
 303  *
 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
 308  *                  slave index.
 309  * \param uuid    - is the uuid of slave which is (re)connecting to the master
 310  *                  target
 311  *
 312  * \retval     - pointer to the dt_object of the slave index on success,
 313  *               appropriate error on failure
 314  */
 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)
 320 {
 321         struct lquota_thread_info       *qti = lquota_info(env);
 322         struct dt_object                *slv_idx;
 323         int                              rc;
 324         ENTRY;
 325
 326         LASSERT(uuid != NULL);
 327
 328         CDEBUG(D_QUOTA, "lookup slave index file for %s\n",
 329                obd_uuid2str(uuid));
 330
 331         /* generate filename associated with the slave */
 332         rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
 333         if (rc)
 334                 RETURN(ERR_PTR(rc));
 335
 336         /* lookup slave index file */
 337         rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
 338         if (rc)
 339                 RETURN(ERR_PTR(rc));
 340
 341         /* name is found, get the object */
 342         slv_idx = dt_locate(env, dev, &qti->qti_fid);
 343         if (IS_ERR(slv_idx))
 344                 RETURN(slv_idx);
 345
 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);
 349                 if (rc) {
 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);
 355                 }
 356         }
 357
 358         RETURN(slv_idx);
 359 }
 360
 361 /*
 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
 364  *   create the index.
 365  * - otherwise, we create the index file with a local reserved FID (see
 366  *   lquota_local_oid)
 367  *
 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
 373  *                  slave index.
 374  * \param uuid    - is the uuid of slave which is (re)connecting to the master
 375  *                  target
 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
 379  *
 380  * \retval     - pointer to the dt_object of the slave index on success,
 381  *               appropriate error on failure
 382  */
 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,
 388                                               bool local)
 389 {
 390         struct lquota_thread_info       *qti = lquota_info(env);
 391         struct dt_object                *slv_idx;
 392         int                              rc;
 393         ENTRY;
 394
 395         LASSERT(uuid != NULL);
 396
 397         CDEBUG(D_QUOTA, "lookup/create slave index file for %s\n",
 398                obd_uuid2str(uuid));
 399
 400         /* generate filename associated with the slave */
 401         rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
 402         if (rc)
 403                 RETURN(ERR_PTR(rc));
 404
 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;
 409         if (local) {
 410                 int type;
 411
 412                 rc = lquota_extract_fid(glb_fid, NULL, NULL, &type);
 413                 if (rc)
 414                         RETURN(ERR_PTR(rc));
 415
 416                 /* use predefined fid in the reserved oid list */
 417                 qti->qti_fid.f_oid = (type == USRQUOTA) ? LQUOTA_USR_OID
 418                                                         : LQUOTA_GRP_OID;
 419
 420                 slv_idx = local_index_find_or_create_with_fid(env, dev,
 421                                                               &qti->qti_fid,
 422                                                               parent,
 423                                                               qti->qti_buf,
 424                                                               LQUOTA_MODE,
 425                                                         &dt_quota_slv_features);
 426         } else {
 427                 /* allocate fid dynamically if index does not exist already */
 428                 qti->qti_fid.f_oid = LQUOTA_GENERATED_OID;
 429
 430                 /* lookup/create slave index file */
 431                 slv_idx = lquota_disk_find_create(env, dev, parent,
 432                                                   &qti->qti_fid,
 433                                                   &dt_quota_slv_features,
 434                                                   qti->qti_buf);
 435         }
 436
 437         if (IS_ERR(slv_idx))
 438                 RETURN(slv_idx);
 439
 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);
 444                 if (rc) {
 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);
 450                 }
 451         }
 452
 453         RETURN(slv_idx);
 454 }
 455
 456 /*
 457  * Iterate over all slave index files associated with global index \glb_fid and
 458  * invoke a callback function for each slave index file.
 459  *
 460  * \param env     - is the environment passed by the caller
 461  * \param parent  - is the parent directory where the slave index files are
 462  *                  stored
 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
 466  *                  file is found
 467  * \param arg     - is an opaq argument passed to the callback function \func
 468  */
 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,
 471                              void *arg)
 472 {
 473         struct lquota_thread_info       *qti = lquota_info(env);
 474         struct dt_it                    *it;
 475         const struct dt_it_ops          *iops;
 476         char                            *name;
 477         int                              rc;
 478         ENTRY;
 479
 480         OBD_ALLOC(name, LQUOTA_NAME_MAX);
 481         if (name == NULL)
 482                 RETURN(-ENOMEM);
 483
 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);
 487
 488         iops = &parent->do_index_ops->dio_it;
 489         it = iops->init(env, parent, 0);
 490         if (IS_ERR(it)) {
 491                 OBD_FREE(name, LQUOTA_NAME_MAX);
 492                 RETURN(PTR_ERR(it));
 493         }
 494
 495         rc = iops->load(env, it, 0);
 496         if (rc == 0) {
 497                 /*
 498                  * Iterator didn't find record with exactly the key requested.
 499                  *
 500                  * It is currently either
 501                  *
 502                  *     - positioned above record with key less than
 503                  *     requested---skip it.
 504                  *
 505                  *     - or not positioned at all (is in IAM_IT_SKEWED
 506                  *     state)---position it on the next item.
 507                  */
 508                 rc = iops->next(env, it);
 509         } else if (rc > 0)
 510                 rc = 0;
 511
 512         while (rc == 0) {
 513                 struct dt_key   *key;
 514                 int              len;
 515
 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)
 519                         goto next;
 520
 521                 key = iops->key(env, it);
 522                 if (IS_ERR(key)) {
 523                         rc = PTR_ERR(key);
 524                         break;
 525                 }
 526
 527                 if (strncmp((char *)key, name, strlen(name)) != 0)
 528                         goto next;
 529
 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';
 534
 535                 /* lookup fid associated with this slave index file */
 536                 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
 537                 if (rc)
 538                         break;
 539
 540                 if (qti->qti_fid.f_seq != FID_SEQ_QUOTA)
 541                         goto next;
 542
 543                 rc = func(env, glb_fid, (char *)key, &qti->qti_fid, arg);
 544                 if (rc)
 545                         break;
 546 next:
 547                 do {
 548                         rc = iops->next(env, it);
 549                 } while (rc == -ESTALE);
 550         }
 551
 552         iops->put(env, it);
 553         iops->fini(env, it);
 554         OBD_FREE(name, LQUOTA_NAME_MAX);
 555         if (rc > 0)
 556                 rc = 0;
 557         RETURN(rc);
 558 }
 559
 560 /*
 561  * Retrieve quota settings from disk for a particular identifier.
 562  *
 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.
 567  *
 568  * \retval    - 0 on success, appropriate error on failure
 569  */
 570 int lquota_disk_read(const struct lu_env *env, struct dt_object *obj,
 571                      union lquota_id *id, struct dt_rec *rec)
 572 {
 573         int     rc;
 574         ENTRY;
 575
 576         LASSERT(dt_object_exists(obj));
 577         LASSERT(obj->do_index_ops != NULL);
 578
 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);
 582         dt_read_unlock(env, obj);
 583
 584         RETURN(rc);
 585 }
 586
 587 /*
 588  * Reserve enough credits to update a record in a quota index file.
 589  *
 590  * \param env - is the environment passed by the caller
 591  * \param th  - is the transaction to use for disk writes
 592  * \param obj - is the on-disk index where quota settings are stored.
 593  * \param id  - is the key to be updated
 594  *
 595  * \retval    - 0 on success, appropriate error on failure
 596  */
 597 int lquota_disk_declare_write(const struct lu_env *env, struct thandle *th,
 598                               struct dt_object *obj, union lquota_id *id)
 599 {
 600         struct lquota_thread_info       *qti = lquota_info(env);
 601         struct dt_key                   *key = (struct dt_key *)&id->qid_uid;
 602         int                              rc;
 603         ENTRY;
 604
 605         LASSERT(dt_object_exists(obj));
 606         LASSERT(obj->do_index_ops != NULL);
 607
 608         /* speculative delete declaration in case there is already an existing
 609          * record in the index */
 610         rc = dt_declare_delete(env, obj, key, th);
 611         if (rc)
 612                 RETURN(rc);
 613
 614         /* declare insertion of updated record */
 615         rc = dt_declare_insert(env, obj, (struct dt_rec *)&qti->qti_rec, key,
 616                                th);
 617         if (rc)
 618                 RETURN(rc);
 619
 620         /* we might have to update the version of the global index too */
 621         rc = dt_declare_version_set(env, obj, th);
 622
 623         RETURN(rc);
 624 }
 625
 626 /*
 627  * Update a record in a quota index file.
 628  *
 629  * \param env - is the environment passed by the caller
 630  * \param th  - is the transaction to use for disk writes
 631  * \param obj - is the on-disk index to be updated.
 632  * \param id  - is the key to be updated
 633  * \param rec - is the input record containing the new quota settings.
 634  * \param flags - can be LQUOTA_BUMP_VER or LQUOTA_SET_VER.
 635  * \param ver   - is the new version of the index if LQUOTA_SET_VER is set or is
 636  *                used to return the new version of the index when
 637  *                LQUOTA_BUMP_VER is set.
 638  *
 639  * \retval    - 0 on success, appropriate error on failure
 640  */
 641 int lquota_disk_write(const struct lu_env *env, struct thandle *th,
 642                       struct dt_object *obj, union lquota_id *id,
 643                       struct dt_rec *rec, __u32 flags, __u64 *ver)
 644 {
 645         struct lquota_thread_info       *qti = lquota_info(env);
 646         struct dt_key                   *key = (struct dt_key *)&id->qid_uid;
 647         int                              rc;
 648         ENTRY;
 649
 650         LASSERT(dt_object_exists(obj));
 651         LASSERT(obj->do_index_ops != NULL);
 652
 653         /* lock index */
 654         dt_write_lock(env, obj, 0);
 655
 656         /* check whether there is already an existing record for this ID */
 657         rc = dt_lookup(env, obj, (struct dt_rec *)&qti->qti_rec, key);
 658         if (rc == 0) {
 659                 /* delete existing record in order to replace it */
 660                 rc = dt_delete(env, obj, key, th);
 661                 if (rc)
 662                         GOTO(out, rc);
 663         } else if (rc == -ENOENT) {
 664                 /* probably first insert */
 665                 rc = 0;
 666         } else {
 667                 GOTO(out, rc);
 668         }
 669
 670         if (rec != NULL) {
 671                 /* insert record with updated quota settings */
 672                 rc = dt_insert(env, obj, rec, key, th, 1);
 673                 if (rc) {
 674                         /* try to insert the old one */
 675                         rc = dt_insert(env, obj, (struct dt_rec *)&qti->qti_rec,
 676                                        key, th, 1);
 677                         LASSERTF(rc == 0, "failed to insert record in quota "
 678                                  "index "DFID"\n",
 679                                  PFID(lu_object_fid(&obj->do_lu)));
 680                         GOTO(out, rc);
 681                 }
 682         }
 683
 684         if (flags != 0) {
 685                 LASSERT(ver);
 686                 if (flags & LQUOTA_BUMP_VER) {
 687                         /* caller wants to bump the version, let's first read
 688                          * it */
 689                         *ver = dt_version_get(env, obj);
 690                         (*ver)++;
 691                 } else {
 692                         LASSERT(flags & LQUOTA_SET_VER);
 693                 }
 694                 dt_version_set(env, obj, *ver, th);
 695         }
 696
 697         EXIT;
 698 out:
 699         dt_write_unlock(env, obj);
 700         return rc;
 701 }
 702
 703 /*
 704  * Update version of an index file
 705  *
 706  * \param env - is the environment passed by the caller
 707  * \param dev - is the backend dt device storing the index file
 708  * \param obj - is the on-disk index that should be updated
 709  * \param ver - is the new version
 710  */
 711 int lquota_disk_update_ver(const struct lu_env *env, struct dt_device *dev,
 712                            struct dt_object *obj, __u64 ver)
 713 {
 714         struct thandle  *th;
 715         int              rc;
 716         ENTRY;
 717
 718         th = dt_trans_create(env, dev);
 719         if (IS_ERR(th))
 720                 RETURN(PTR_ERR(th));
 721
 722         rc = dt_declare_version_set(env, obj, th);
 723         if (rc)
 724                 GOTO(out, rc);
 725
 726         rc = dt_trans_start_local(env, dev, th);
 727         if (rc)
 728                 GOTO(out, rc);
 729         th->th_sync = 1;
 730
 731         dt_version_set(env, obj, ver, th);
 732         EXIT;
 733 out:
 734         dt_trans_stop(env, dev, th);
 735         return rc;
 736 }
 737
 738 /*
 739  * Write a global record
 740  *
 741  * \param env - is the environment passed by the caller
 742  * \param obj - is the on-disk global index to be updated
 743  * \param id  - index to be updated
 744  * \param rec - record to be written
 745  */
 746 int lquota_disk_write_glb(const struct lu_env *env, struct dt_object *obj,
 747                           __u64 id, struct lquota_glb_rec *rec)
 748 {
 749         struct dt_device        *dev = lu2dt_dev(obj->do_lu.lo_dev);
 750         struct thandle          *th;
 751         struct dt_key           *key = (struct dt_key *)&id;
 752         int                      rc;
 753         ENTRY;
 754
 755         th = dt_trans_create(env, dev);
 756         if (IS_ERR(th))
 757                 RETURN(PTR_ERR(th));
 758
 759         /* the entry with 0 key can always be found in IAM file. */
 760         if (id == 0) {
 761                 rc = dt_declare_delete(env, obj, key, th);
 762                 if (rc)
 763                         GOTO(out, rc);
 764         }
 765
 766         rc = dt_declare_insert(env, obj, (struct dt_rec *)rec, key, th);
 767         if (rc)
 768                 GOTO(out, rc);
 769
 770         rc = dt_trans_start_local(env, dev, th);
 771         if (rc)
 772                 GOTO(out, rc);
 773
 774         dt_write_lock(env, obj, 0);
 775
 776         if (id == 0) {
 777                 struct lquota_glb_rec *tmp;
 778
 779                 OBD_ALLOC_PTR(tmp);
 780                 if (tmp == NULL)
 781                         GOTO(out_lock, rc = -ENOMEM);
 782
 783                 rc = dt_lookup(env, obj, (struct dt_rec *)tmp, key);
 784
 785                 OBD_FREE_PTR(tmp);
 786                 if (rc == 0) {
 787                         rc = dt_delete(env, obj, key, th);
 788                         if (rc)
 789                                 GOTO(out_lock, rc);
 790                 }
 791                 rc = 0;
 792         }
 793
 794         rc = dt_insert(env, obj, (struct dt_rec *)rec, key, th, 1);
 795 out_lock:
 796         dt_write_unlock(env, obj);
 797 out:
 798         dt_trans_stop(env, dev, th);
 799         RETURN(rc);
 800 }
 801 EXPORT_SYMBOL(lquota_disk_write_glb);