lustre/lod/lod_sub_object.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,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13  * GNU General Public License version 2 for more details.  A copy is
  14  * included in the COPYING file that accompanied this code.
  15
  16  * You should have received a copy of the GNU General Public License
  17  * along with this program; if not, write to the Free Software
  18  * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
  19  *
  20  * GPL HEADER END
  21  */
  22 /*
  23  * Copyright (c) 2014, Intel Corporation.
  24  */
  25 /*
  26  * lustre/lod/lod_sub_object.c
  27  *
  28  * LOD sub object methods
  29  *
  30  * This file implements sub-object methods for LOD.
  31  *
  32  * LOD is Logic volume layer in the MDS stack, which will handle striping
  33  * and distribute the update to different OSP/OSD. After directing the updates
  34  * to one specific OSD/OSP, it also needs to do some thing before calling
  35  * OSD/OSP API, for example recording updates for cross-MDT operation, get
  36  * the next level transaction etc.
  37  *
  38  * Author: Di Wang <di.wang@intel.com>
  39  */
  40
  41 #define DEBUG_SUBSYSTEM S_MDS
  42
  43 #include <obd.h>
  44 #include <obd_class.h>
  45 #include <lustre_ver.h>
  46 #include <obd_support.h>
  47 #include <lprocfs_status.h>
  48
  49 #include <lustre_fid.h>
  50 #include <lustre_param.h>
  51 #include <md_object.h>
  52 #include <lustre_linkea.h>
  53 #include <lustre_log.h>
  54
  55 #include "lod_internal.h"
  56
  57 struct thandle *lod_sub_get_thandle(const struct lu_env *env,
  58                                     struct thandle *th,
  59                                     const struct dt_object *sub_obj,
  60                                     bool *record_update)
  61 {
  62         struct lod_device       *lod = dt2lod_dev(th->th_dev);
  63         struct top_thandle      *tth;
  64         struct thandle          *sub_th;
  65         int                     type = LU_SEQ_RANGE_ANY;
  66         __u32                   mdt_index;
  67         int                     rc;
  68         ENTRY;
  69
  70         if (record_update != NULL)
  71                 *record_update = false;
  72
  73         if (th->th_top == NULL)
  74                 RETURN(th);
  75
  76         tth = container_of(th, struct top_thandle, tt_super);
  77
  78         /* local object must be mdt object, Note: during ost object
  79          * creation, FID is not assigned until osp_object_create(),
  80          * so if the FID of sub_obj is zero, it means OST object. */
  81         if (!dt_object_remote(sub_obj) ||
  82             fid_is_zero(lu_object_fid(&sub_obj->do_lu))) {
  83                 /* local MDT object */
  84                 if (fid_is_sane(lu_object_fid(&sub_obj->do_lu)) &&
  85                     tth->tt_multiple_thandle != NULL &&
  86                     record_update != NULL &&
  87                     th->th_result == 0)
  88                         *record_update = true;
  89
  90                 RETURN(tth->tt_master_sub_thandle);
  91         }
  92
  93         rc = lod_fld_lookup(env, lod, lu_object_fid(&sub_obj->do_lu),
  94                             &mdt_index, &type);
  95         if (rc < 0)
  96                 RETURN(ERR_PTR(rc));
  97
  98         if (type == LU_SEQ_RANGE_OST)
  99                 RETURN(tth->tt_master_sub_thandle);
 100
 101         sub_th = thandle_get_sub(env, th, sub_obj);
 102         if (IS_ERR(sub_th))
 103                 RETURN(sub_th);
 104
 105         if (tth->tt_multiple_thandle != NULL && record_update != NULL &&
 106             th->th_result == 0)
 107                 *record_update = true;
 108
 109         RETURN(sub_th);
 110 }
 111
 112 /**
 113  * Declare sub-object creation.
 114  *
 115  * Get transaction of next layer and declare the creation of the object.
 116  *
 117  * \param[in] env       execution environment
 118  * \param[in] dt        the object being created
 119  * \param[in] attr      the attributes of the object being created
 120  * \param[in] hint      the hint of the creation
 121  * \param[in] dof       the object format of the creation
 122  * \param[th] th        the transaction handle
 123  *
 124  * \retval              0 if the declaration succeeds
 125  * \retval              negative errno if the declaration fails.
 126  */
 127 int lod_sub_object_declare_create(const struct lu_env *env,
 128                                   struct dt_object *dt,
 129                                   struct lu_attr *attr,
 130                                   struct dt_allocation_hint *hint,
 131                                   struct dt_object_format *dof,
 132                                   struct thandle *th)
 133 {
 134         struct thandle *sub_th;
 135         bool record_update;
 136
 137         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 138         if (IS_ERR(sub_th))
 139                 return PTR_ERR(sub_th);
 140
 141         if (record_update)
 142                 update_record_size(env, create, th, lu_object_fid(&dt->do_lu),
 143                                    attr, hint, dof);
 144
 145         return dt_declare_create(env, dt, attr, hint, dof, sub_th);
 146 }
 147
 148 /**
 149  * Create sub-object.
 150  *
 151  * Get transaction of next layer, record updates if it belongs to cross-MDT
 152  * operation, and create the object.
 153  *
 154  * \param[in] env       execution environment
 155  * \param[in] dt        the object being created
 156  * \param[in] attr      the attributes of the object being created
 157  * \param[in] hint      the hint of the creation
 158  * \param[in] dof       the object format of the creation
 159  * \param[th] th        the transaction handle
 160  *
 161  * \retval              0 if the creation succeeds
 162  * \retval              negative errno if the creation fails.
 163  */
 164 int lod_sub_object_create(const struct lu_env *env, struct dt_object *dt,
 165                           struct lu_attr *attr,
 166                           struct dt_allocation_hint *hint,
 167                           struct dt_object_format *dof,
 168                           struct thandle *th)
 169 {
 170         struct thandle     *sub_th;
 171         bool               record_update;
 172         int                 rc;
 173         ENTRY;
 174
 175         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 176         if (IS_ERR(sub_th))
 177                 RETURN(PTR_ERR(sub_th));
 178
 179         if (record_update) {
 180                 rc = update_record_pack(create, th,
 181                                         lu_object_fid(&dt->do_lu),
 182                                         attr, hint, dof);
 183                 if (rc < 0)
 184                         RETURN(rc);
 185         }
 186
 187         rc = dt_create(env, dt, attr, hint, dof, sub_th);
 188
 189         RETURN(rc);
 190 }
 191
 192 /**
 193  * Declare adding reference for the sub-object
 194  *
 195  * Get transaction of next layer and declare the reference adding.
 196  *
 197  * \param[in] env       execution environment
 198  * \param[in] dt        dt object to add reference
 199  * \param[in] th        transaction handle
 200  *
 201  * \retval              0 if the declaration succeeds.
 202  * \retval              negative errno if the declaration fails.
 203  */
 204 int lod_sub_object_declare_ref_add(const struct lu_env *env,
 205                                    struct dt_object *dt,
 206                                    struct thandle *th)
 207 {
 208         struct thandle  *sub_th;
 209         bool            record_update;
 210         int             rc;
 211         ENTRY;
 212
 213         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 214         if (IS_ERR(sub_th))
 215                 RETURN(PTR_ERR(sub_th));
 216
 217         if (record_update)
 218                 update_record_size(env, ref_add, th, lu_object_fid(&dt->do_lu));
 219
 220         rc = dt_declare_ref_add(env, dt, sub_th);
 221
 222         RETURN(rc);
 223 }
 224
 225 /**
 226  * Add reference for the sub-object
 227  *
 228  * Get transaction of next layer, record updates if it belongs to cross-MDT
 229  * operation and add reference of the object.
 230  *
 231  * \param[in] env       execution environment
 232  * \param[in] dt        dt object to add reference
 233  * \param[in] th        transaction handle
 234  *
 235  * \retval              0 if it succeeds.
 236  * \retval              negative errno if it fails.
 237  */
 238 int lod_sub_object_ref_add(const struct lu_env *env, struct dt_object *dt,
 239                            struct thandle *th)
 240 {
 241         struct thandle  *sub_th;
 242         bool            record_update;
 243         int             rc;
 244         ENTRY;
 245
 246         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 247         if (IS_ERR(sub_th))
 248                 RETURN(PTR_ERR(sub_th));
 249
 250         if (record_update) {
 251                 rc = update_record_pack(ref_add, th,
 252                                         lu_object_fid(&dt->do_lu));
 253                 if (rc < 0)
 254                         RETURN(rc);
 255         }
 256
 257         rc = dt_ref_add(env, dt, sub_th);
 258
 259         RETURN(rc);
 260 }
 261
 262 /**
 263  * Declare deleting reference for the sub-object
 264  *
 265  * Get transaction of next layer and declare the reference deleting.
 266  *
 267  * \param[in] env       execution environment
 268  * \param[in] dt        dt object to delete reference
 269  * \param[in] th        transaction handle
 270  *
 271  * \retval              0 if the declaration succeeds.
 272  * \retval              negative errno if the declaration fails.
 273  */
 274 int lod_sub_object_declare_ref_del(const struct lu_env *env,
 275                                    struct dt_object *dt,
 276                                    struct thandle *th)
 277 {
 278         struct thandle  *sub_th;
 279         bool            record_update;
 280         int             rc;
 281         ENTRY;
 282
 283         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 284         if (IS_ERR(sub_th))
 285                 RETURN(PTR_ERR(sub_th));
 286
 287         if (record_update)
 288                 update_record_size(env, ref_del, th, lu_object_fid(&dt->do_lu));
 289
 290         rc = dt_declare_ref_del(env, dt, sub_th);
 291
 292         RETURN(rc);
 293 }
 294
 295 /**
 296  * Delete reference for the sub-object
 297  *
 298  * Get transaction of next layer, record updates if it belongs to cross-MDT
 299  * operation and delete reference of the object.
 300  *
 301  * \param[in] env       execution environment
 302  * \param[in] dt        dt object to delete reference
 303  * \param[in] th        transaction handle
 304  *
 305  * \retval              0 if it succeeds.
 306  * \retval              negative errno if it fails.
 307  */
 308 int lod_sub_object_ref_del(const struct lu_env *env, struct dt_object *dt,
 309                            struct thandle *th)
 310 {
 311         struct thandle  *sub_th;
 312         bool            record_update;
 313         int             rc;
 314         ENTRY;
 315
 316         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 317         if (IS_ERR(sub_th))
 318                 RETURN(PTR_ERR(sub_th));
 319
 320         if (record_update) {
 321                 rc = update_record_pack(ref_del, th,
 322                                         lu_object_fid(&dt->do_lu));
 323                 if (rc < 0)
 324                         RETURN(rc);
 325         }
 326
 327         rc = dt_ref_del(env, dt, sub_th);
 328
 329         RETURN(rc);
 330 }
 331
 332 /**
 333  * Declare destroying sub-object
 334  *
 335  * Get transaction of next layer and declare the sub-object destroy.
 336  *
 337  * \param[in] env       execution environment
 338  * \param[in] dt        dt object to be destroyed
 339  * \param[in] th        transaction handle
 340  *
 341  * \retval              0 if the declaration succeeds.
 342  * \retval              negative errno if the declaration fails.
 343  */
 344 int lod_sub_object_declare_destroy(const struct lu_env *env,
 345                                    struct dt_object *dt,
 346                                    struct thandle *th)
 347 {
 348         struct thandle  *sub_th;
 349         bool            record_update;
 350         int             rc;
 351         ENTRY;
 352
 353         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 354         if (IS_ERR(sub_th))
 355                 RETURN(PTR_ERR(sub_th));
 356
 357         if (record_update)
 358                 update_record_size(env, object_destroy, th,
 359                                    lu_object_fid(&dt->do_lu));
 360
 361         rc = dt_declare_destroy(env, dt, sub_th);
 362
 363         RETURN(rc);
 364 }
 365
 366 /**
 367  * Destroy sub-object
 368  *
 369  * Get transaction of next layer, record updates if it belongs to cross-MDT
 370  * operation and destroy the object.
 371  *
 372  * \param[in] env       execution environment
 373  * \param[in] dt        dt object to be destroyed
 374  * \param[in] th        transaction handle
 375  *
 376  * \retval              0 if the destroy succeeds.
 377  * \retval              negative errno if the destroy fails.
 378  */
 379 int lod_sub_object_destroy(const struct lu_env *env, struct dt_object *dt,
 380                            struct thandle *th)
 381 {
 382         struct thandle  *sub_th;
 383         bool            record_update;
 384         int             rc;
 385         ENTRY;
 386
 387         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 388         if (IS_ERR(sub_th))
 389                 RETURN(PTR_ERR(sub_th));
 390
 391         if (record_update) {
 392                 rc = update_record_pack(object_destroy, th,
 393                                         lu_object_fid(&dt->do_lu));
 394                 if (rc < 0)
 395                         RETURN(rc);
 396         }
 397
 398         rc = dt_destroy(env, dt, sub_th);
 399
 400         RETURN(rc);
 401 }
 402
 403 /**
 404  * Declare sub-object index insert
 405  *
 406  * Get transaction of next layer and declare index insert.
 407  *
 408  * \param[in] env       execution environment
 409  * \param[in] dt        object for which to insert index
 410  * \param[in] rec       record of the index which will be inserted
 411  * \param[in] key       key of the index which will be inserted
 412  * \param[in] th        the transaction handle
 413  *
 414  * \retval              0 if the declaration succeeds.
 415  * \retval              negative errno if the declaration fails.
 416  */
 417 int lod_sub_object_declare_insert(const struct lu_env *env,
 418                                   struct dt_object *dt,
 419                                   const struct dt_rec *rec,
 420                                   const struct dt_key *key,
 421                                   struct thandle *th)
 422 {
 423         struct thandle *sub_th;
 424         bool            record_update;
 425
 426         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 427         if (IS_ERR(sub_th))
 428                 return PTR_ERR(sub_th);
 429
 430         if (record_update)
 431                 update_record_size(env, index_insert, th,
 432                                    lu_object_fid(&dt->do_lu), rec, key);
 433
 434         return dt_declare_insert(env, dt, rec, key, sub_th);
 435 }
 436
 437 /**
 438  * Insert index of sub object
 439  *
 440  * Get transaction of next layer, record updates if it belongs to cross-MDT
 441  * operation, and insert the index.
 442  *
 443  * \param[in] env       execution environment
 444  * \param[in] dt        object for which to insert index
 445  * \param[in] rec       record of the index to be inserted
 446  * \param[in] key       key of the index to be inserted
 447  * \param[in] th        the transaction handle
 448  * \param[in] ign       whether ignore quota
 449  *
 450  * \retval              0 if the insertion succeeds.
 451  * \retval              negative errno if the insertion fails.
 452  */
 453 int lod_sub_object_index_insert(const struct lu_env *env, struct dt_object *dt,
 454                                 const struct dt_rec *rec,
 455                                 const struct dt_key *key, struct thandle *th,
 456                                 int ign)
 457 {
 458         struct thandle *sub_th;
 459         int             rc;
 460         bool            record_update;
 461
 462         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 463         if (IS_ERR(sub_th))
 464                 return PTR_ERR(sub_th);
 465
 466         if (record_update) {
 467                 rc = update_record_pack(index_insert, th,
 468                                         lu_object_fid(&dt->do_lu), rec, key);
 469                 if (rc < 0)
 470                         return rc;
 471         }
 472
 473         return dt_insert(env, dt, rec, key, sub_th, ign);
 474 }
 475
 476 /**
 477  * Declare sub-object index delete
 478  *
 479  * Get transaction of next layer and declare index deletion.
 480  *
 481  * \param[in] env       execution environment
 482  * \param[in] dt        object for which to delete index
 483  * \param[in] key       key of the index which will be deleted
 484  * \param[in] th        the transaction handle
 485  *
 486  * \retval              0 if the declaration succeeds.
 487  * \retval              negative errno if the declaration fails.
 488  */
 489 int lod_sub_object_declare_delete(const struct lu_env *env,
 490                                   struct dt_object *dt,
 491                                   const struct dt_key *key,
 492                                   struct thandle *th)
 493 {
 494         struct thandle *sub_th;
 495         bool            record_update;
 496
 497         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 498         if (IS_ERR(sub_th))
 499                 return PTR_ERR(sub_th);
 500
 501         if (record_update)
 502                 update_record_size(env, index_delete, th,
 503                                    lu_object_fid(&dt->do_lu), key);
 504
 505         return dt_declare_delete(env, dt, key, sub_th);
 506 }
 507
 508 /**
 509  * Delete index of sub object
 510  *
 511  * Get transaction of next layer, record updates if it belongs to cross-MDT
 512  * operation, and delete the index.
 513  *
 514  * \param[in] env       execution environment
 515  * \param[in] dt        object for which to delete index
 516  * \param[in] key       key of the index to be deleted
 517  * \param[in] th        the transaction handle
 518  *
 519  * \retval              0 if the deletion succeeds.
 520  * \retval              negative errno if the deletion fails.
 521  */
 522 int lod_sub_object_delete(const struct lu_env *env, struct dt_object *dt,
 523                           const struct dt_key *name, struct thandle *th)
 524 {
 525         struct thandle  *sub_th;
 526         bool            record_update;
 527         int             rc;
 528         ENTRY;
 529
 530         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 531         if (IS_ERR(sub_th))
 532                 RETURN(PTR_ERR(sub_th));
 533
 534         if (record_update) {
 535                 rc = update_record_pack(index_delete, th,
 536                                         lu_object_fid(&dt->do_lu), name);
 537                 if (rc < 0)
 538                         RETURN(rc);
 539         }
 540
 541         rc = dt_delete(env, dt, name, sub_th);
 542         RETURN(rc);
 543 }
 544
 545 /**
 546  * Declare xattr_set
 547  *
 548  * Get transaction of next layer, and declare xattr set.
 549  *
 550  * \param[in] env       execution environment
 551  * \param[in] dt        object on which to set xattr
 552  * \param[in] buf       xattr to be set
 553  * \param[in] name      name of the xattr
 554  * \param[in] fl        flag for setting xattr
 555  *
 556  * \retval              0 if the declaration succeeds.
 557  * \retval              negative errno if the declaration fails.
 558  */
 559 int lod_sub_object_declare_xattr_set(const struct lu_env *env,
 560                                      struct dt_object *dt,
 561                                      const struct lu_buf *buf,
 562                                      const char *name, int fl,
 563                                      struct thandle *th)
 564 {
 565         struct thandle  *sub_th;
 566         bool            record_update;
 567         int             rc;
 568         ENTRY;
 569
 570         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 571         if (IS_ERR(sub_th))
 572                 RETURN(PTR_ERR(sub_th));
 573
 574         if (record_update)
 575                 update_record_size(env, xattr_set, th,
 576                                    lu_object_fid(&dt->do_lu),
 577                                    buf, name, fl);
 578
 579         rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
 580
 581         RETURN(rc);
 582 }
 583
 584 /**
 585  * Set xattr
 586  *
 587  * Get transaction of next layer, record updates if it belongs to cross-MDT
 588  * operation, and set xattr to the object.
 589  *
 590  * \param[in] env       execution environment
 591  * \param[in] dt        object on which to set xattr
 592  * \param[in] buf       xattr to be set
 593  * \param[in] name      name of the xattr
 594  * \param[in] fl        flag for setting xattr
 595  * \param[in] th        transaction handle
 596  *
 597  * \retval              0 if the xattr setting succeeds.
 598  * \retval              negative errno if xattr setting fails.
 599  */
 600 int lod_sub_object_xattr_set(const struct lu_env *env, struct dt_object *dt,
 601                              const struct lu_buf *buf, const char *name, int fl,
 602                              struct thandle *th)
 603 {
 604         struct thandle  *sub_th;
 605         bool            record_update;
 606         int             rc;
 607         ENTRY;
 608
 609         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 610         if (IS_ERR(sub_th))
 611                 RETURN(PTR_ERR(sub_th));
 612
 613         if (record_update) {
 614                 rc = update_record_pack(xattr_set, th,
 615                                         lu_object_fid(&dt->do_lu),
 616                                         buf, name, fl);
 617                 if (rc < 0)
 618                         RETURN(rc);
 619         }
 620
 621         rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
 622
 623         RETURN(rc);
 624 }
 625
 626 /**
 627  * Declare attr_set
 628  *
 629  * Get transaction of next layer, and declare attr set.
 630  *
 631  * \param[in] env       execution environment
 632  * \param[in] dt        object on which to set attr
 633  * \param[in] attr      attributes to be set
 634  * \param[in] th        transaction handle
 635  *
 636  * \retval              0 if the declaration succeeds.
 637  * \retval              negative errno if the declaration fails.
 638  */
 639 int lod_sub_object_declare_attr_set(const struct lu_env *env,
 640                                     struct dt_object *dt,
 641                                     const struct lu_attr *attr,
 642                                     struct thandle *th)
 643 {
 644         struct thandle  *sub_th;
 645         bool            record_update;
 646         int             rc;
 647         ENTRY;
 648
 649         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 650         if (IS_ERR(sub_th))
 651                 RETURN(PTR_ERR(sub_th));
 652
 653         if (record_update)
 654                 update_record_size(env, attr_set, th,
 655                                    lu_object_fid(&dt->do_lu), attr);
 656
 657         rc = dt_declare_attr_set(env, dt, attr, sub_th);
 658
 659         RETURN(rc);
 660 }
 661
 662 /**
 663  * attributes set
 664  *
 665  * Get transaction of next layer, record updates if it belongs to cross-MDT
 666  * operation, and set attributes to the object.
 667  *
 668  * \param[in] env       execution environment
 669  * \param[in] dt        object on which to set attr
 670  * \param[in] attr      attrbutes to be set
 671  * \param[in] th        transaction handle
 672  *
 673  * \retval              0 if attributes setting succeeds.
 674  * \retval              negative errno if the attributes setting fails.
 675  */
 676 int lod_sub_object_attr_set(const struct lu_env *env,
 677                             struct dt_object *dt,
 678                             const struct lu_attr *attr,
 679                             struct thandle *th)
 680 {
 681         bool               record_update;
 682         struct thandle     *sub_th;
 683         int                 rc;
 684         ENTRY;
 685
 686         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 687         if (IS_ERR(sub_th))
 688                 RETURN(PTR_ERR(sub_th));
 689
 690         if (record_update) {
 691                 rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
 692                                         attr);
 693                 if (rc < 0)
 694                         RETURN(rc);
 695         }
 696
 697         rc = dt_attr_set(env, dt, attr, sub_th);
 698
 699         RETURN(rc);
 700 }
 701
 702 /**
 703  * Declare xattr_del
 704  *
 705  * Get transaction of next layer, and declare xattr deletion.
 706  *
 707  * \param[in] env       execution environment
 708  * \param[in] dt        object on which to delete xattr
 709  * \param[in] name      name of the xattr to be deleted
 710  * \param[in] th        transaction handle
 711  *
 712  * \retval              0 if the declaration succeeds.
 713  * \retval              negative errno if the declaration fails.
 714  */
 715 int lod_sub_object_declare_xattr_del(const struct lu_env *env,
 716                                      struct dt_object *dt,
 717                                      const char *name,
 718                                      struct thandle *th)
 719 {
 720         struct thandle  *sub_th;
 721         bool            record_update;
 722         int             rc;
 723         ENTRY;
 724
 725         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 726         if (IS_ERR(sub_th))
 727                 RETURN(PTR_ERR(sub_th));
 728
 729         if (record_update)
 730                 update_record_size(env, xattr_del, th,
 731                                    lu_object_fid(&dt->do_lu),
 732                                    name);
 733
 734         rc = dt_declare_xattr_del(env, dt, name, sub_th);
 735
 736         RETURN(rc);
 737 }
 738
 739 /**
 740  * xattribute deletion
 741  *
 742  * Get transaction of next layer, record update if it belongs to cross-MDT
 743  * operation and delete xattr.
 744  *
 745  * \param[in] env       execution environment
 746  * \param[in] dt        object on which to delete xattr
 747  * \param[in] name      name of the xattr to be deleted
 748  * \param[in] th        transaction handle
 749  *
 750  * \retval              0 if the deletion succeeds.
 751  * \retval              negative errno if the deletion fails.
 752  */
 753 int lod_sub_object_xattr_del(const struct lu_env *env,
 754                              struct dt_object *dt,
 755                              const char *name,
 756                              struct thandle *th)
 757 {
 758         struct thandle  *sub_th;
 759         bool            record_update;
 760         int             rc;
 761         ENTRY;
 762
 763         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 764         if (IS_ERR(sub_th))
 765                 RETURN(PTR_ERR(sub_th));
 766
 767         if (record_update) {
 768                 rc = update_record_pack(xattr_del, th,
 769                                         lu_object_fid(&dt->do_lu), name);
 770                 if (rc < 0)
 771                         RETURN(rc);
 772         }
 773
 774         rc = dt_xattr_del(env, dt, name, sub_th);
 775
 776         RETURN(rc);
 777 }
 778
 779 /**
 780  * Declare buffer write
 781  *
 782  * Get transaction of next layer and declare buffer write.
 783  *
 784  * \param[in] env       execution environment
 785  * \param[in] dt        object to be written
 786  * \param[in] buf       buffer to write which includes an embedded size field
 787  * \param[in] pos       offet in the object to start writing at
 788  * \param[in] th        transaction handle
 789  *
 790  * \retval              0 if the insertion succeeds.
 791  * \retval              negative errno if the insertion fails.
 792  */
 793 int lod_sub_object_declare_write(const struct lu_env *env,
 794                                  struct dt_object *dt,
 795                                  const struct lu_buf *buf, loff_t pos,
 796                                  struct thandle *th)
 797 {
 798         struct thandle  *sub_th;
 799         bool            record_update;
 800         int             rc;
 801         ENTRY;
 802
 803         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 804         if (IS_ERR(sub_th))
 805                 RETURN(PTR_ERR(sub_th));
 806
 807         if (record_update)
 808                 update_record_size(env, write, th,
 809                                    lu_object_fid(&dt->do_lu),
 810                                    buf, pos);
 811
 812         rc = dt_declare_write(env, dt, buf, pos, sub_th);
 813
 814         RETURN(rc);
 815 }
 816
 817 /**
 818  * Write buffer to sub object
 819  *
 820  * Get transaction of next layer, records buffer write if it belongs to
 821  * Cross-MDT operation, and write buffer.
 822  *
 823  * \param[in] env       execution environment
 824  * \param[in] dt        object to be written
 825  * \param[in] buf       buffer to write which includes an embedded size field
 826  * \param[in] pos       offet in the object to start writing at
 827  * \param[in] th        transaction handle
 828  * \param[in] rq        enforcement for this write
 829  *
 830  * \retval              the buffer size in bytes if it succeeds.
 831  * \retval              negative errno if it fails.
 832  */
 833 ssize_t lod_sub_object_write(const struct lu_env *env, struct dt_object *dt,
 834                              const struct lu_buf *buf, loff_t *pos,
 835                              struct thandle *th, int rq)
 836 {
 837         struct thandle  *sub_th;
 838         bool            record_update;
 839         ssize_t         rc;
 840         ENTRY;
 841
 842         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 843         if (IS_ERR(sub_th))
 844                 RETURN(PTR_ERR(sub_th));
 845
 846         if (record_update) {
 847                 rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
 848                                         buf, *pos);
 849                 if (rc < 0)
 850                         RETURN(rc);
 851         }
 852
 853         rc = dt_write(env, dt, buf, pos, sub_th, rq);
 854         RETURN(rc);
 855 }
 856
 857 /**
 858  * Declare punch
 859  *
 860  * Get transaction of next layer and declare punch.
 861  *
 862  * \param[in] env       execution environment
 863  * \param[in] dt        object to be written
 864  * \param[in] start     start offset of punch
 865  * \param[in] end       end offet of punch
 866  * \param[in] th        transaction handle
 867  *
 868  * \retval              0 if the insertion succeeds.
 869  * \retval              negative errno if the insertion fails.
 870  */
 871 int lod_sub_object_declare_punch(const struct lu_env *env,
 872                                  struct dt_object *dt,
 873                                  __u64 start, __u64 end,
 874                                  struct thandle *th)
 875 {
 876         struct thandle  *sub_th;
 877         bool            record_update;
 878         int             rc;
 879         ENTRY;
 880
 881         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 882         if (IS_ERR(sub_th))
 883                 RETURN(PTR_ERR(sub_th));
 884
 885         if (record_update)
 886                 update_record_size(env, punch, th,
 887                                    lu_object_fid(&dt->do_lu),
 888                                    start, end);
 889
 890         rc = dt_declare_punch(env, dt, start, end, sub_th);
 891
 892         RETURN(rc);
 893 }
 894
 895 /**
 896  * Punch to sub object
 897  *
 898  * Get transaction of next layer, records buffer write if it belongs to
 899  * Cross-MDT operation, and punch object.
 900  *
 901  * \param[in] env       execution environment
 902  * \param[in] dt        object to be written
 903  * \param[in] start     start offset of punch
 904  * \param[in] end       end offset of punch
 905  * \param[in] th        transaction handle
 906  * \param[in] capa      capability of the write
 907  *
 908  * \retval              the buffer size in bytes if it succeeds.
 909  * \retval              negative errno if it fails.
 910  */
 911 int lod_sub_object_punch(const struct lu_env *env, struct dt_object *dt,
 912                          __u64 start, __u64 end, struct thandle *th)
 913 {
 914         struct thandle  *sub_th;
 915         bool            record_update;
 916         int             rc;
 917         ENTRY;
 918
 919         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 920         if (IS_ERR(sub_th))
 921                 RETURN(PTR_ERR(sub_th));
 922
 923         if (record_update) {
 924                 rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
 925                                         start, end);
 926                 if (rc < 0)
 927                         RETURN(rc);
 928         }
 929
 930         rc = dt_punch(env, dt, start, end, sub_th);
 931
 932         RETURN(rc);
 933 }
 934
 935 int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
 936                       struct dt_device *dt, int index)
 937 {
 938         struct lod_thread_info  *lti = lod_env_info(env);
 939         struct llog_ctxt        *ctxt;
 940         struct llog_handle      *lgh;
 941         struct llog_catid       *cid = &lti->lti_cid;
 942         struct lu_fid           *fid = &lti->lti_fid;
 943         struct obd_device       *obd;
 944         int                     rc;
 945         bool                    need_put = false;
 946         ENTRY;
 947
 948         lu_update_log_fid(fid, index);
 949
 950         rc = lodname2mdt_index(lod2obd(lod)->obd_name, (__u32 *)&index);
 951         if (rc < 0)
 952                 RETURN(rc);
 953
 954         rc = llog_osd_get_cat_list(env, dt, index, 1, cid, fid);
 955         if (rc != 0) {
 956                 CERROR("%s: can't get id from catalogs: rc = %d\n",
 957                        lod2obd(lod)->obd_name, rc);
 958                 RETURN(rc);
 959         }
 960
 961         obd = dt->dd_lu_dev.ld_obd;
 962         ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
 963         LASSERT(ctxt != NULL);
 964         ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
 965         ctxt->loc_chunk_size = LLOG_MIN_CHUNK_SIZE * 4;
 966         if (likely(logid_id(&cid->lci_logid) != 0)) {
 967                 rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
 968                                LLOG_OPEN_EXISTS);
 969
 970                 /* re-create llog if it is missing */
 971                 if (rc == -ENOENT)
 972                         logid_set_id(&cid->lci_logid, 0);
 973                 else if (rc < 0)
 974                         GOTO(out_put, rc);
 975         }
 976
 977         if (unlikely(logid_id(&cid->lci_logid) == 0)) {
 978                 rc = llog_open_create(env, ctxt, &lgh, NULL, NULL);
 979                 if (rc < 0)
 980                         GOTO(out_put, rc);
 981                 cid->lci_logid = lgh->lgh_id;
 982                 need_put = true;
 983         }
 984
 985         LASSERT(lgh != NULL);
 986
 987         rc = llog_cat_init_and_process(env, lgh);
 988         if (rc != 0)
 989                 GOTO(out_close, rc);
 990
 991         if (need_put) {
 992                 rc = llog_osd_put_cat_list(env, dt, index, 1, cid, fid);
 993                 if (rc != 0)
 994                         GOTO(out_close, rc);
 995         }
 996
 997         ctxt->loc_handle = lgh;
 998
 999         CDEBUG(D_INFO, "%s: Init llog for %d - catid "DOSTID":%x\n",
1000                obd->obd_name, index, POSTID(&cid->lci_logid.lgl_oi),
1001                cid->lci_logid.lgl_ogen);
1002 out_close:
1003         if (rc != 0)
1004                 llog_cat_close(env, lgh);
1005 out_put:
1006         llog_ctxt_put(ctxt);
1007         RETURN(rc);
1008 }