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 Intel, Inc.
  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 a 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 #ifndef EXPORT_SYMTAB
  52 # define EXPORT_SYMTAB
  53 #endif
  54
  55 #define DEBUG_SUBSYSTEM S_LQUOTA
  56
  57 #include "lquota_internal.h"
  58
  59 #define LQUOTA_MODE (S_IFREG | S_IRUGO | S_IWUSR)
  60
  61 /*
  62  * Helper function looking up & creating if not found an index file with a
  63  * dynamic fid.
  64  */
  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,
  69                         char *name)
  70 {
  71         struct lquota_thread_info       *qti = lquota_info(env);
  72         struct dt_object                *obj;
  73         struct local_oid_storage        *los;
  74         int                              rc;
  75         ENTRY;
  76
  77         /* Set up local storage */
  78         rc = local_oid_storage_init(env, dev, fid, &los);
  79         if (rc)
  80                 RETURN(ERR_PTR(rc));
  81
  82         /* lookup/create slave index file */
  83         obj = local_index_find_or_create(env, los, parent, name, LQUOTA_MODE,
  84                                          idx_feat);
  85         if (IS_ERR(obj))
  86                 GOTO(out, obj);
  87
  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);
  93         if (IS_ERR(obj))
  94                 GOTO(out, obj);
  95 out:
  96         local_oid_storage_fini(env, los);
  97         RETURN(obj);
  98 }
  99
 100 /*
 101  * helper function to generate the filename associated with a slave index file
 102  */
 103 static inline int lquota_disk_slv_filename(struct lu_fid *glb_fid,
 104                                            struct obd_uuid *uuid,
 105                                            char *filename)
 106 {
 107         char    *name, *uuid_str;
 108
 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)
 113                         RETURN(-ENOMEM);
 114                 memcpy(uuid_str, uuid->uuid, sizeof(*uuid) - 1);
 115         } else {
 116                 uuid_str = (char *)uuid->uuid;
 117         }
 118
 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, '-');
 122         if (name == NULL) {
 123                 name = strrchr(uuid_str, ':');
 124                 if (name == NULL) {
 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));
 129                         return -EINVAL;
 130                 }
 131         }
 132         name++;
 133
 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);
 138
 139         if (uuid_str != uuid->uuid)
 140                 OBD_FREE(uuid_str, sizeof(*uuid));
 141
 142         return 0;
 143 }
 144
 145 /*
 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
 148  * the quota master.
 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
 151  * configured.
 152  *
 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
 158  *
 159  * \retval     - pointer to quota root dt_object on success, appropriate error
 160  *               on failure
 161  */
 162 struct dt_object *lquota_disk_dir_find_create(const struct lu_env *env,
 163                                               struct dt_device *dev,
 164                                               struct dt_object *parent,
 165                                               const char *name)
 166 {
 167         struct lquota_thread_info       *qti = lquota_info(env);
 168         struct dt_object                *qt_dir = NULL;
 169         struct local_oid_storage        *los = NULL;
 170         int                              rc;
 171         ENTRY;
 172
 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);
 177         if (rc)
 178                 RETURN(ERR_PTR(rc));
 179
 180         if (parent == NULL) {
 181                 /* Fetch dt object associated with root directory */
 182                 rc = dt_root_get(env, dev, &qti->qti_fid);
 183                 if (rc)
 184                         GOTO(out, rc);
 185
 186                 parent = dt_locate_at(env, dev, &qti->qti_fid,
 187                                       dev->dd_lu_dev.ld_site->ls_top_dev);
 188                 if (IS_ERR(parent))
 189                         GOTO(out, rc = PTR_ERR(parent));
 190         } else {
 191                 lu_object_get(&parent->do_lu);
 192         }
 193
 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);
 197         if (IS_ERR(qt_dir))
 198                 GOTO(out, rc = PTR_ERR(qt_dir));
 199
 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);
 205         if (IS_ERR(qt_dir))
 206                 GOTO(out, rc = PTR_ERR(qt_dir));
 207
 208         if (!dt_try_as_dir(env, qt_dir))
 209                 GOTO(out, rc = -ENOTDIR);
 210         EXIT;
 211 out:
 212         if (parent != NULL && !IS_ERR(parent))
 213                 lu_object_put(env, &parent->do_lu);
 214         if (los != NULL)
 215                 local_oid_storage_fini(env, los);
 216         if (rc) {
 217                 if (qt_dir != NULL && !IS_ERR(qt_dir))
 218                         lu_object_put(env, &qt_dir->do_lu);
 219                 qt_dir = ERR_PTR(rc);
 220         }
 221         return qt_dir;
 222 }
 223
 224 /*
 225  * Look-up/create a global index file.
 226  *
 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
 230  *                 not found
 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
 234  *
 235  * \retval     - pointer to the dt_object of the global index on success,
 236  *               appropriate error on failure
 237  */
 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)
 242 {
 243         struct lquota_thread_info       *qti = lquota_info(env);
 244         struct dt_object                *glb_idx;
 245         const struct dt_index_features  *idx_feat;
 246         ENTRY;
 247
 248         CDEBUG(D_QUOTA, "look-up/create %sglobal idx file ("DFID")\n",
 249                local ? "local " : "", PFID(fid));
 250
 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);
 257 #else
 258 #warning "remove old quota compatibility code"
 259         idx_feat = &dt_quota_glb_features;
 260 #endif
 261
 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
 264          * type */
 265         sprintf(qti->qti_buf, "0x%x", fid->f_oid);
 266
 267         if (local) {
 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,
 272                                                   qti->qti_buf);
 273         } else {
 274                 /* look-up/create global index on disk */
 275                 glb_idx = local_index_find_or_create_with_fid(env, dev, fid,
 276                                                               parent,
 277                                                               qti->qti_buf,
 278                                                               LQUOTA_MODE,
 279                                                               idx_feat);
 280         }
 281
 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);
 286                 RETURN(glb_idx);
 287         }
 288
 289         /* install index operation vector */
 290         if (glb_idx->do_index_ops == NULL) {
 291                 int rc;
 292
 293                 rc = glb_idx->do_ops->do_index_try(env, glb_idx, idx_feat);
 294                 if (rc) {
 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);
 300                 }
 301         }
 302
 303         RETURN(glb_idx);
 304 }
 305
 306 /*
 307  * Look-up a slave index file.
 308  *
 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
 313  *                  slave index.
 314  * \param uuid    - is the uuid of slave which is (re)connecting to the master
 315  *                  target
 316  *
 317  * \retval     - pointer to the dt_object of the slave index on success,
 318  *               appropriate error on failure
 319  */
 320 struct dt_object *lquota_disk_slv_find(const struct lu_env *env,
 321                                        struct dt_device *dev,
 322                                        struct dt_object *parent,
 323                                        struct lu_fid *glb_fid,
 324                                        struct obd_uuid *uuid)
 325 {
 326         struct lquota_thread_info       *qti = lquota_info(env);
 327         struct dt_object                *slv_idx;
 328         int                              rc;
 329         ENTRY;
 330
 331         LASSERT(uuid != NULL);
 332
 333         CDEBUG(D_QUOTA, "lookup slave index file for %s\n",
 334                obd_uuid2str(uuid));
 335
 336         /* generate filename associated with the slave */
 337         rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
 338         if (rc)
 339                 RETURN(ERR_PTR(rc));
 340
 341         /* lookup slave index file */
 342         rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
 343         if (rc)
 344                 RETURN(ERR_PTR(rc));
 345
 346         /* name is found, get the object */
 347         slv_idx = dt_locate(env, dev, &qti->qti_fid);
 348         if (IS_ERR(slv_idx))
 349                 RETURN(slv_idx);
 350
 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);
 354                 if (rc) {
 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);
 360                 }
 361         }
 362
 363         RETURN(slv_idx);
 364 }
 365
 366 /*
 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
 369  *   create the index.
 370  * - otherwise, we create the index file with a local reserved FID (see
 371  *   lquota_local_oid)
 372  *
 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
 378  *                  slave index.
 379  * \param uuid    - is the uuid of slave which is (re)connecting to the master
 380  *                  target
 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
 384  *
 385  * \retval     - pointer to the dt_object of the slave index on success,
 386  *               appropriate error on failure
 387  */
 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,
 393                                               bool local)
 394 {
 395         struct lquota_thread_info       *qti = lquota_info(env);
 396         struct dt_object                *slv_idx;
 397         int                              rc;
 398         ENTRY;
 399
 400         LASSERT(uuid != NULL);
 401
 402         CDEBUG(D_QUOTA, "lookup/create slave index file for %s\n",
 403                obd_uuid2str(uuid));
 404
 405         /* generate filename associated with the slave */
 406         rc = lquota_disk_slv_filename(glb_fid, uuid, qti->qti_buf);
 407         if (rc)
 408                 RETURN(ERR_PTR(rc));
 409
 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;
 414         if (local) {
 415                 int type;
 416
 417                 rc = lquota_extract_fid(glb_fid, NULL, NULL, &type);
 418                 if (rc)
 419                         RETURN(ERR_PTR(rc));
 420
 421                 /* use predefined fid in the reserved oid list */
 422                 qti->qti_fid.f_oid = (type == USRQUOTA) ? LQUOTA_USR_OID
 423                                                         : LQUOTA_GRP_OID;
 424
 425                 slv_idx = local_index_find_or_create_with_fid(env, dev,
 426                                                               &qti->qti_fid,
 427                                                               parent,
 428                                                               qti->qti_buf,
 429                                                               LQUOTA_MODE,
 430                                                         &dt_quota_slv_features);
 431         } else {
 432                 /* allocate fid dynamically if index does not exist already */
 433                 qti->qti_fid.f_oid = LQUOTA_GENERATED_OID;
 434
 435                 /* lookup/create slave index file */
 436                 slv_idx = lquota_disk_find_create(env, dev, parent,
 437                                                   &qti->qti_fid,
 438                                                   &dt_quota_slv_features,
 439                                                   qti->qti_buf);
 440         }
 441
 442         if (IS_ERR(slv_idx))
 443                 RETURN(slv_idx);
 444
 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);
 449                 if (rc) {
 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);
 455                 }
 456         }
 457
 458         RETURN(slv_idx);
 459 }
 460
 461 /*
 462  * Iterate over all slave index files associated with global index \glb_fid and
 463  * invoke a callback function for each slave index file.
 464  *
 465  * \param env     - is the environment passed by the caller
 466  * \param parent  - is the parent directory where the slave index files are
 467  *                  stored
 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
 471  *                  file is found
 472  * \param arg     - is an opaq argument passed to the callback function \func
 473  */
 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,
 476                              void *arg)
 477 {
 478         struct lquota_thread_info       *qti = lquota_info(env);
 479         struct dt_it                    *it;
 480         const struct dt_it_ops          *iops;
 481         char                            *name;
 482         int                              rc;
 483         ENTRY;
 484
 485         OBD_ALLOC(name, sizeof("0x00000000-"));
 486         if (name == NULL)
 487                 RETURN(-ENOMEM);
 488
 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);
 492
 493         iops = &parent->do_index_ops->dio_it;
 494         it = iops->init(env, parent, 0, BYPASS_CAPA);
 495         if (IS_ERR(it)) {
 496                 OBD_FREE(name, sizeof("0x00000000-"));
 497                 RETURN(PTR_ERR(it));
 498         }
 499
 500         rc = iops->load(env, it, 0);
 501         if (rc == 0) {
 502                 /*
 503                  * Iterator didn't find record with exactly the key requested.
 504                  *
 505                  * It is currently either
 506                  *
 507                  *     - positioned above record with key less than
 508                  *     requested---skip it.
 509                  *
 510                  *     - or not positioned at all (is in IAM_IT_SKEWED
 511                  *     state)---position it on the next item.
 512                  */
 513                 rc = iops->next(env, it);
 514         } else if (rc > 0)
 515                 rc = 0;
 516
 517         while (rc == 0) {
 518                 struct dt_key   *key;
 519                 int              len;
 520
 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)
 524                         goto next;
 525
 526                 key = iops->key(env, it);
 527                 if (IS_ERR(key)) {
 528                         rc = PTR_ERR(key);
 529                         break;
 530                 }
 531
 532                 if (strncmp((char *)key, name, strlen(name)) != 0)
 533                         goto next;
 534
 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';
 539
 540                 /* lookup fid associated with this slave index file */
 541                 rc = dt_lookup_dir(env, parent, qti->qti_buf, &qti->qti_fid);
 542                 if (rc)
 543                         break;
 544
 545                 if (qti->qti_fid.f_seq != FID_SEQ_QUOTA)
 546                         goto next;
 547
 548                 rc = func(env, glb_fid, (char *)key, &qti->qti_fid, arg);
 549                 if (rc)
 550                         break;
 551 next:
 552                 do {
 553                         rc = iops->next(env, it);
 554                 } while (rc == -ESTALE);
 555         }
 556
 557         iops->put(env, it);
 558         iops->fini(env, it);
 559         OBD_FREE(name, sizeof("0x00000000-"));
 560         if (rc > 0)
 561                 rc = 0;
 562         RETURN(rc);
 563 }
 564
 565 /*
 566  * Retrieve quota settings from disk for a particular identifier.
 567  *
 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.
 572  *
 573  * \retval    - 0 on success, appropriate error on failure
 574  */
 575 int lquota_disk_read(const struct lu_env *env, struct dt_object *obj,
 576                      union lquota_id *id, struct dt_rec *rec)
 577 {
 578         int     rc;
 579         ENTRY;
 580
 581         LASSERT(dt_object_exists(obj));
 582         LASSERT(obj->do_index_ops != NULL);
 583
 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,
 587                        BYPASS_CAPA);
 588         dt_read_unlock(env, obj);
 589
 590         RETURN(rc);
 591 }
 592
 593 /*
 594  * Reserve enough credits to update a record in a quota index file.
 595  *
 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
 600  *
 601  * \retval    - 0 on success, appropriate error on failure
 602  */
 603 int lquota_disk_declare_write(const struct lu_env *env, struct thandle *th,
 604                               struct dt_object *obj, union lquota_id *id)
 605 {
 606         struct lquota_thread_info       *qti = lquota_info(env);
 607         struct dt_key                   *key = (struct dt_key *)&id->qid_uid;
 608         int                              rc;
 609         ENTRY;
 610
 611         LASSERT(dt_object_exists(obj));
 612         LASSERT(obj->do_index_ops != NULL);
 613
 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);
 617         if (rc)
 618                 RETURN(rc);
 619
 620         /* declare insertion of updated record */
 621         rc = dt_declare_insert(env, obj, (struct dt_rec *)&qti->qti_rec, key,
 622                                th);
 623         if (rc)
 624                 RETURN(rc);
 625
 626         /* we might have to update the version of the global index too */
 627         rc = dt_declare_version_set(env, obj, th);
 628
 629         RETURN(rc);
 630 }
 631
 632 /*
 633  * Update a record in a quota index file.
 634  *
 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.
 644  *
 645  * \retval    - 0 on success, appropriate error on failure
 646  */
 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)
 650 {
 651         struct lquota_thread_info       *qti = lquota_info(env);
 652         struct dt_key                   *key = (struct dt_key *)&id->qid_uid;
 653         int                              rc;
 654         ENTRY;
 655
 656         LASSERT(dt_object_exists(obj));
 657         LASSERT(obj->do_index_ops != NULL);
 658
 659         /* lock index */
 660         dt_write_lock(env, obj, 0);
 661
 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,
 664                        BYPASS_CAPA);
 665         if (rc == 0) {
 666                 /* delete existing record in order to replace it */
 667                 rc = dt_delete(env, obj, key, th, BYPASS_CAPA);
 668                 if (rc)
 669                         GOTO(out, rc);
 670         } else if (rc == -ENOENT) {
 671                 /* probably first insert */
 672                 rc = 0;
 673         } else {
 674                 GOTO(out, rc);
 675         }
 676
 677         if (rec != NULL) {
 678                 /* insert record with updated quota settings */
 679                 rc = dt_insert(env, obj, rec, key, th, BYPASS_CAPA, 1);
 680                 if (rc) {
 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 "
 685                                  "index "DFID,
 686                                  PFID(lu_object_fid(&obj->do_lu)));
 687                         GOTO(out, rc);
 688                 }
 689         }
 690
 691         if (flags != 0) {
 692                 LASSERT(ver);
 693                 if (flags & LQUOTA_BUMP_VER) {
 694                         /* caller wants to bump the version, let's first read
 695                          * it */
 696                         *ver = dt_version_get(env, obj);
 697                         (*ver)++;
 698                 } else {
 699                         LASSERT(flags & LQUOTA_SET_VER);
 700                 }
 701                 dt_version_set(env, obj, *ver, th);
 702         }
 703
 704         EXIT;
 705 out:
 706         dt_write_unlock(env, obj);
 707         return rc;
 708 }
 709
 710 /*
 711  * Update version of an index file
 712  *
 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
 717  */
 718 int lquota_disk_update_ver(const struct lu_env *env, struct dt_device *dev,
 719                            struct dt_object *obj, __u64 ver)
 720 {
 721         struct thandle  *th;
 722         int              rc;
 723         ENTRY;
 724
 725         th = dt_trans_create(env, dev);
 726         if (IS_ERR(th))
 727                 RETURN(PTR_ERR(th));
 728
 729         rc = dt_declare_version_set(env, obj, th);
 730         if (rc)
 731                 GOTO(out, rc);
 732
 733         rc = dt_trans_start_local(env, dev, th);
 734         if (rc)
 735                 GOTO(out, rc);
 736         th->th_sync = 1;
 737
 738         dt_version_set(env, obj, ver, th);
 739         EXIT;
 740 out:
 741         dt_trans_stop(env, dev, th);
 742         return rc;
 743 }