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         LASSERT(tth->tt_magic == TOP_THANDLE_MAGIC);
  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_update_records != NULL &&
  86                     record_update != NULL)
  87                         *record_update = true;
  88
  89                 RETURN(tth->tt_master_sub_thandle);
  90         }
  91
  92         rc = lod_fld_lookup(env, lod, lu_object_fid(&sub_obj->do_lu),
  93                             &mdt_index, &type);
  94         if (rc < 0)
  95                 RETURN(ERR_PTR(rc));
  96
  97         if (type == LU_SEQ_RANGE_OST)
  98                 RETURN(tth->tt_master_sub_thandle);
  99
 100         if (tth->tt_update_records != NULL && record_update != NULL)
 101                 *record_update = true;
 102
 103         sub_th = thandle_get_sub(env, th, sub_obj);
 104
 105         RETURN(sub_th);
 106 }
 107
 108 /**
 109  * Declare sub-object creation.
 110  *
 111  * Get transaction of next layer and declare the creation of the object.
 112  *
 113  * \param[in] env       execution environment
 114  * \param[in] dt        the object being created
 115  * \param[in] attr      the attributes of the object being created
 116  * \param[in] hint      the hint of the creation
 117  * \param[in] dof       the object format of the creation
 118  * \param[th] th        the transaction handle
 119  *
 120  * \retval              0 if the declaration succeeds
 121  * \retval              negative errno if the declaration fails.
 122  */
 123 int lod_sub_object_declare_create(const struct lu_env *env,
 124                                   struct dt_object *dt,
 125                                   struct lu_attr *attr,
 126                                   struct dt_allocation_hint *hint,
 127                                   struct dt_object_format *dof,
 128                                   struct thandle *th)
 129 {
 130         struct thandle *sub_th;
 131
 132         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 133         if (IS_ERR(sub_th))
 134                 return PTR_ERR(sub_th);
 135
 136         return dt_declare_create(env, dt, attr, hint, dof, sub_th);
 137 }
 138
 139 /**
 140  * Create sub-object.
 141  *
 142  * Get transaction of next layer, record updates if it belongs to cross-MDT
 143  * operation, and create the object.
 144  *
 145  * \param[in] env       execution environment
 146  * \param[in] dt        the object being created
 147  * \param[in] attr      the attributes of the object being created
 148  * \param[in] hint      the hint of the creation
 149  * \param[in] dof       the object format of the creation
 150  * \param[th] th        the transaction handle
 151  *
 152  * \retval              0 if the creation succeeds
 153  * \retval              negative errno if the creation fails.
 154  */
 155 int lod_sub_object_create(const struct lu_env *env, struct dt_object *dt,
 156                           struct lu_attr *attr,
 157                           struct dt_allocation_hint *hint,
 158                           struct dt_object_format *dof,
 159                           struct thandle *th)
 160 {
 161         struct thandle     *sub_th;
 162         bool               record_update;
 163         int                 rc;
 164         ENTRY;
 165
 166         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 167         if (IS_ERR(sub_th))
 168                 RETURN(PTR_ERR(sub_th));
 169
 170         if (record_update) {
 171                 rc = update_record_pack(create, th,
 172                                         lu_object_fid(&dt->do_lu),
 173                                         attr, hint, dof);
 174                 if (rc < 0)
 175                         RETURN(rc);
 176         }
 177
 178         rc = dt_create(env, dt, attr, hint, dof, sub_th);
 179
 180         RETURN(rc);
 181 }
 182
 183 /**
 184  * Declare adding reference for the sub-object
 185  *
 186  * Get transaction of next layer and declare the reference adding.
 187  *
 188  * \param[in] env       execution environment
 189  * \param[in] dt        dt object to add reference
 190  * \param[in] th        transaction handle
 191  *
 192  * \retval              0 if the declaration succeeds.
 193  * \retval              negative errno if the declaration fails.
 194  */
 195 int lod_sub_object_declare_ref_add(const struct lu_env *env,
 196                                    struct dt_object *dt,
 197                                    struct thandle *th)
 198 {
 199         struct thandle  *sub_th;
 200         int             rc;
 201         ENTRY;
 202
 203         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 204         if (IS_ERR(sub_th))
 205                 RETURN(PTR_ERR(sub_th));
 206
 207         rc = dt_declare_ref_add(env, dt, sub_th);
 208
 209         RETURN(rc);
 210 }
 211
 212 /**
 213  * Add reference for the sub-object
 214  *
 215  * Get transaction of next layer, record updates if it belongs to cross-MDT
 216  * operation and add reference of the object.
 217  *
 218  * \param[in] env       execution environment
 219  * \param[in] dt        dt object to add reference
 220  * \param[in] th        transaction handle
 221  *
 222  * \retval              0 if it succeeds.
 223  * \retval              negative errno if it fails.
 224  */
 225 int lod_sub_object_ref_add(const struct lu_env *env, struct dt_object *dt,
 226                            struct thandle *th)
 227 {
 228         struct thandle  *sub_th;
 229         bool            record_update;
 230         int             rc;
 231         ENTRY;
 232
 233         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 234         if (IS_ERR(sub_th))
 235                 RETURN(PTR_ERR(sub_th));
 236
 237         if (record_update) {
 238                 rc = update_record_pack(ref_add, th,
 239                                         lu_object_fid(&dt->do_lu));
 240                 if (rc < 0)
 241                         RETURN(rc);
 242         }
 243
 244         rc = dt_ref_add(env, dt, sub_th);
 245
 246         RETURN(rc);
 247 }
 248
 249 /**
 250  * Declare deleting reference for the sub-object
 251  *
 252  * Get transaction of next layer and declare the reference deleting.
 253  *
 254  * \param[in] env       execution environment
 255  * \param[in] dt        dt object to delete reference
 256  * \param[in] th        transaction handle
 257  *
 258  * \retval              0 if the declaration succeeds.
 259  * \retval              negative errno if the declaration fails.
 260  */
 261 int lod_sub_object_declare_ref_del(const struct lu_env *env,
 262                                    struct dt_object *dt,
 263                                    struct thandle *th)
 264 {
 265         struct thandle  *sub_th;
 266         int             rc;
 267         ENTRY;
 268
 269         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 270         if (IS_ERR(sub_th))
 271                 RETURN(PTR_ERR(sub_th));
 272
 273         rc = dt_declare_ref_del(env, dt, sub_th);
 274
 275         RETURN(rc);
 276 }
 277
 278 /**
 279  * Delete reference for the sub-object
 280  *
 281  * Get transaction of next layer, record updates if it belongs to cross-MDT
 282  * operation and delete reference of the object.
 283  *
 284  * \param[in] env       execution environment
 285  * \param[in] dt        dt object to delete reference
 286  * \param[in] th        transaction handle
 287  *
 288  * \retval              0 if it succeeds.
 289  * \retval              negative errno if it fails.
 290  */
 291 int lod_sub_object_ref_del(const struct lu_env *env, struct dt_object *dt,
 292                            struct thandle *th)
 293 {
 294         struct thandle  *sub_th;
 295         bool            record_update;
 296         int             rc;
 297         ENTRY;
 298
 299         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 300         if (IS_ERR(sub_th))
 301                 RETURN(PTR_ERR(sub_th));
 302
 303         if (record_update) {
 304                 rc = update_record_pack(ref_del, th,
 305                                         lu_object_fid(&dt->do_lu));
 306                 if (rc < 0)
 307                         RETURN(rc);
 308         }
 309
 310         rc = dt_ref_del(env, dt, sub_th);
 311
 312         RETURN(rc);
 313 }
 314
 315 /**
 316  * Declare destroying sub-object
 317  *
 318  * Get transaction of next layer and declare the sub-object destroy.
 319  *
 320  * \param[in] env       execution environment
 321  * \param[in] dt        dt object to be destroyed
 322  * \param[in] th        transaction handle
 323  *
 324  * \retval              0 if the declaration succeeds.
 325  * \retval              negative errno if the declaration fails.
 326  */
 327 int lod_sub_object_declare_destroy(const struct lu_env *env,
 328                                    struct dt_object *dt,
 329                                    struct thandle *th)
 330 {
 331         struct thandle  *sub_th;
 332         int             rc;
 333         ENTRY;
 334
 335         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 336         if (IS_ERR(sub_th))
 337                 RETURN(PTR_ERR(sub_th));
 338
 339         rc = dt_declare_destroy(env, dt, sub_th);
 340
 341         RETURN(rc);
 342 }
 343
 344 /**
 345  * Destroy sub-object
 346  *
 347  * Get transaction of next layer, record updates if it belongs to cross-MDT
 348  * operation and destroy the object.
 349  *
 350  * \param[in] env       execution environment
 351  * \param[in] dt        dt object to be destroyed
 352  * \param[in] th        transaction handle
 353  *
 354  * \retval              0 if the destroy succeeds.
 355  * \retval              negative errno if the destroy fails.
 356  */
 357 int lod_sub_object_destroy(const struct lu_env *env, struct dt_object *dt,
 358                            struct thandle *th)
 359 {
 360         struct thandle  *sub_th;
 361         bool            record_update;
 362         int             rc;
 363         ENTRY;
 364
 365         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 366         if (IS_ERR(sub_th))
 367                 RETURN(PTR_ERR(sub_th));
 368
 369         if (record_update) {
 370                 rc = update_record_pack(object_destroy, th,
 371                                         lu_object_fid(&dt->do_lu));
 372                 if (rc < 0)
 373                         RETURN(rc);
 374         }
 375
 376         rc = dt_destroy(env, dt, sub_th);
 377
 378         RETURN(rc);
 379 }
 380
 381 /**
 382  * Declare sub-object index insert
 383  *
 384  * Get transaction of next layer and declare index insert.
 385  *
 386  * \param[in] env       execution environment
 387  * \param[in] dt        object for which to insert index
 388  * \param[in] rec       record of the index which will be inserted
 389  * \param[in] key       key of the index which will be inserted
 390  * \param[in] th        the transaction handle
 391  *
 392  * \retval              0 if the declaration succeeds.
 393  * \retval              negative errno if the declaration fails.
 394  */
 395 int lod_sub_object_declare_insert(const struct lu_env *env,
 396                                   struct dt_object *dt,
 397                                   const struct dt_rec *rec,
 398                                   const struct dt_key *key,
 399                                   struct thandle *th)
 400 {
 401         struct thandle *sub_th;
 402
 403         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 404         if (IS_ERR(sub_th))
 405                 return PTR_ERR(sub_th);
 406
 407         return dt_declare_insert(env, dt, rec, key, sub_th);
 408 }
 409
 410 /**
 411  * Insert index of sub object
 412  *
 413  * Get transaction of next layer, record updates if it belongs to cross-MDT
 414  * operation, and insert the index.
 415  *
 416  * \param[in] env       execution environment
 417  * \param[in] dt        object for which to insert index
 418  * \param[in] rec       record of the index to be inserted
 419  * \param[in] key       key of the index to be inserted
 420  * \param[in] th        the transaction handle
 421  * \param[in] ign       whether ignore quota
 422  *
 423  * \retval              0 if the insertion succeeds.
 424  * \retval              negative errno if the insertion fails.
 425  */
 426 int lod_sub_object_index_insert(const struct lu_env *env, struct dt_object *dt,
 427                                 const struct dt_rec *rec,
 428                                 const struct dt_key *key, struct thandle *th,
 429                                 int ign)
 430 {
 431         struct thandle *sub_th;
 432         int             rc;
 433         bool            record_update;
 434
 435         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 436         if (IS_ERR(sub_th))
 437                 return PTR_ERR(sub_th);
 438
 439         if (record_update) {
 440                 rc = update_record_pack(index_insert, th,
 441                                         lu_object_fid(&dt->do_lu), rec, key);
 442                 if (rc < 0)
 443                         return rc;
 444         }
 445
 446         return dt_insert(env, dt, rec, key, sub_th, ign);
 447 }
 448
 449 /**
 450  * Declare sub-object index delete
 451  *
 452  * Get transaction of next layer and declare index deletion.
 453  *
 454  * \param[in] env       execution environment
 455  * \param[in] dt        object for which to delete index
 456  * \param[in] key       key of the index which will be deleted
 457  * \param[in] th        the transaction handle
 458  *
 459  * \retval              0 if the declaration succeeds.
 460  * \retval              negative errno if the declaration fails.
 461  */
 462 int lod_sub_object_declare_delete(const struct lu_env *env,
 463                                   struct dt_object *dt,
 464                                   const struct dt_key *key,
 465                                   struct thandle *th)
 466 {
 467         struct thandle *sub_th;
 468
 469         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 470         if (IS_ERR(sub_th))
 471                 return PTR_ERR(sub_th);
 472
 473         return dt_declare_delete(env, dt, key, sub_th);
 474 }
 475
 476 /**
 477  * Delete index of sub object
 478  *
 479  * Get transaction of next layer, record updates if it belongs to cross-MDT
 480  * operation, and delete the index.
 481  *
 482  * \param[in] env       execution environment
 483  * \param[in] dt        object for which to delete index
 484  * \param[in] key       key of the index to be deleted
 485  * \param[in] th        the transaction handle
 486  *
 487  * \retval              0 if the deletion succeeds.
 488  * \retval              negative errno if the deletion fails.
 489  */
 490 int lod_sub_object_delete(const struct lu_env *env, struct dt_object *dt,
 491                           const struct dt_key *name, struct thandle *th)
 492 {
 493         struct thandle  *sub_th;
 494         bool            record_update;
 495         int             rc;
 496         ENTRY;
 497
 498         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 499         if (IS_ERR(sub_th))
 500                 RETURN(PTR_ERR(sub_th));
 501
 502         if (record_update) {
 503                 rc = update_record_pack(index_delete, th,
 504                                         lu_object_fid(&dt->do_lu), name);
 505                 if (rc < 0)
 506                         RETURN(rc);
 507         }
 508
 509         rc = dt_delete(env, dt, name, sub_th);
 510         RETURN(rc);
 511 }
 512
 513 /**
 514  * Declare xattr_set
 515  *
 516  * Get transaction of next layer, and declare xattr set.
 517  *
 518  * \param[in] env       execution environment
 519  * \param[in] dt        object on which to set xattr
 520  * \param[in] buf       xattr to be set
 521  * \param[in] name      name of the xattr
 522  * \param[in] fl        flag for setting xattr
 523  *
 524  * \retval              0 if the declaration succeeds.
 525  * \retval              negative errno if the declaration fails.
 526  */
 527 int lod_sub_object_declare_xattr_set(const struct lu_env *env,
 528                                      struct dt_object *dt,
 529                                      const struct lu_buf *buf,
 530                                      const char *name, int fl,
 531                                      struct thandle *th)
 532 {
 533         struct thandle  *sub_th;
 534         int             rc;
 535         ENTRY;
 536
 537         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 538         if (IS_ERR(sub_th))
 539                 RETURN(PTR_ERR(sub_th));
 540
 541         rc = dt_declare_xattr_set(env, dt, buf, name, fl, sub_th);
 542
 543         RETURN(rc);
 544 }
 545
 546 /**
 547  * Set xattr
 548  *
 549  * Get transaction of next layer, record updates if it belongs to cross-MDT
 550  * operation, and set xattr to the object.
 551  *
 552  * \param[in] env       execution environment
 553  * \param[in] dt        object on which to set xattr
 554  * \param[in] buf       xattr to be set
 555  * \param[in] name      name of the xattr
 556  * \param[in] fl        flag for setting xattr
 557  * \param[in] th        transaction handle
 558  *
 559  * \retval              0 if the xattr setting succeeds.
 560  * \retval              negative errno if xattr setting fails.
 561  */
 562 int lod_sub_object_xattr_set(const struct lu_env *env, struct dt_object *dt,
 563                              const struct lu_buf *buf, const char *name, int fl,
 564                              struct thandle *th)
 565 {
 566         struct thandle  *sub_th;
 567         bool            record_update;
 568         int             rc;
 569         ENTRY;
 570
 571         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 572         if (IS_ERR(sub_th))
 573                 RETURN(PTR_ERR(sub_th));
 574
 575         if (record_update) {
 576                 rc = update_record_pack(xattr_set, th,
 577                                         lu_object_fid(&dt->do_lu),
 578                                         buf, name, fl);
 579                 if (rc < 0)
 580                         RETURN(rc);
 581         }
 582
 583         rc = dt_xattr_set(env, dt, buf, name, fl, sub_th);
 584
 585         RETURN(rc);
 586 }
 587
 588 /**
 589  * Declare attr_set
 590  *
 591  * Get transaction of next layer, and declare attr set.
 592  *
 593  * \param[in] env       execution environment
 594  * \param[in] dt        object on which to set attr
 595  * \param[in] attr      attributes to be set
 596  * \param[in] th        transaction handle
 597  *
 598  * \retval              0 if the declaration succeeds.
 599  * \retval              negative errno if the declaration fails.
 600  */
 601 int lod_sub_object_declare_attr_set(const struct lu_env *env,
 602                                     struct dt_object *dt,
 603                                     const struct lu_attr *attr,
 604                                     struct thandle *th)
 605 {
 606         struct thandle  *sub_th;
 607         int             rc;
 608         ENTRY;
 609
 610         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 611         if (IS_ERR(sub_th))
 612                 RETURN(PTR_ERR(sub_th));
 613
 614         rc = dt_declare_attr_set(env, dt, attr, sub_th);
 615
 616         RETURN(rc);
 617 }
 618
 619 /**
 620  * attributes set
 621  *
 622  * Get transaction of next layer, record updates if it belongs to cross-MDT
 623  * operation, and set attributes to the object.
 624  *
 625  * \param[in] env       execution environment
 626  * \param[in] dt        object on which to set attr
 627  * \param[in] attr      attrbutes to be set
 628  * \param[in] th        transaction handle
 629  *
 630  * \retval              0 if attributes setting succeeds.
 631  * \retval              negative errno if the attributes setting fails.
 632  */
 633 int lod_sub_object_attr_set(const struct lu_env *env,
 634                             struct dt_object *dt,
 635                             const struct lu_attr *attr,
 636                             struct thandle *th)
 637 {
 638         bool               record_update;
 639         struct thandle     *sub_th;
 640         int                 rc;
 641         ENTRY;
 642
 643         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 644         if (IS_ERR(sub_th))
 645                 RETURN(PTR_ERR(sub_th));
 646
 647         if (record_update) {
 648                 rc = update_record_pack(attr_set, th, lu_object_fid(&dt->do_lu),
 649                                         attr);
 650                 if (rc < 0)
 651                         RETURN(rc);
 652         }
 653
 654         rc = dt_attr_set(env, dt, attr, sub_th);
 655
 656         RETURN(rc);
 657 }
 658
 659 /**
 660  * Declare xattr_del
 661  *
 662  * Get transaction of next layer, and declare xattr deletion.
 663  *
 664  * \param[in] env       execution environment
 665  * \param[in] dt        object on which to delete xattr
 666  * \param[in] name      name of the xattr to be deleted
 667  * \param[in] th        transaction handle
 668  *
 669  * \retval              0 if the declaration succeeds.
 670  * \retval              negative errno if the declaration fails.
 671  */
 672 int lod_sub_object_declare_xattr_del(const struct lu_env *env,
 673                                      struct dt_object *dt,
 674                                      const char *name,
 675                                      struct thandle *th)
 676 {
 677         struct thandle  *sub_th;
 678         int             rc;
 679         ENTRY;
 680
 681         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 682         if (IS_ERR(sub_th))
 683                 RETURN(PTR_ERR(sub_th));
 684
 685         rc = dt_declare_xattr_del(env, dt, name, sub_th);
 686
 687         RETURN(rc);
 688 }
 689
 690 /**
 691  * xattribute deletion
 692  *
 693  * Get transaction of next layer, record update if it belongs to cross-MDT
 694  * operation and delete xattr.
 695  *
 696  * \param[in] env       execution environment
 697  * \param[in] dt        object on which to delete xattr
 698  * \param[in] name      name of the xattr to be deleted
 699  * \param[in] th        transaction handle
 700  *
 701  * \retval              0 if the deletion succeeds.
 702  * \retval              negative errno if the deletion fails.
 703  */
 704 int lod_sub_object_xattr_del(const struct lu_env *env,
 705                              struct dt_object *dt,
 706                              const char *name,
 707                              struct thandle *th)
 708 {
 709         struct thandle  *sub_th;
 710         bool            record_update;
 711         int             rc;
 712         ENTRY;
 713
 714         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 715         if (IS_ERR(sub_th))
 716                 RETURN(PTR_ERR(sub_th));
 717
 718         if (record_update) {
 719                 rc = update_record_pack(xattr_del, th,
 720                                         lu_object_fid(&dt->do_lu), name);
 721                 if (rc < 0)
 722                         RETURN(rc);
 723         }
 724
 725         rc = dt_xattr_del(env, dt, name, sub_th);
 726
 727         RETURN(rc);
 728 }
 729
 730 /**
 731  * Declare buffer write
 732  *
 733  * Get transaction of next layer and declare buffer write.
 734  *
 735  * \param[in] env       execution environment
 736  * \param[in] dt        object to be written
 737  * \param[in] buf       buffer to write which includes an embedded size field
 738  * \param[in] pos       offet in the object to start writing at
 739  * \param[in] th        transaction handle
 740  *
 741  * \retval              0 if the insertion succeeds.
 742  * \retval              negative errno if the insertion fails.
 743  */
 744 int lod_sub_object_declare_write(const struct lu_env *env,
 745                                  struct dt_object *dt,
 746                                  const struct lu_buf *buf, loff_t pos,
 747                                  struct thandle *th)
 748 {
 749         struct thandle  *sub_th;
 750         int             rc;
 751         ENTRY;
 752
 753         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 754         if (IS_ERR(sub_th))
 755                 RETURN(PTR_ERR(sub_th));
 756
 757         rc = dt_declare_write(env, dt, buf, pos, sub_th);
 758
 759         RETURN(rc);
 760 }
 761
 762 /**
 763  * Write buffer to sub object
 764  *
 765  * Get transaction of next layer, records buffer write if it belongs to
 766  * Cross-MDT operation, and write buffer.
 767  *
 768  * \param[in] env       execution environment
 769  * \param[in] dt        object to be written
 770  * \param[in] buf       buffer to write which includes an embedded size field
 771  * \param[in] pos       offet in the object to start writing at
 772  * \param[in] th        transaction handle
 773  * \param[in] rq        enforcement for this write
 774  *
 775  * \retval              the buffer size in bytes if it succeeds.
 776  * \retval              negative errno if it fails.
 777  */
 778 ssize_t lod_sub_object_write(const struct lu_env *env, struct dt_object *dt,
 779                              const struct lu_buf *buf, loff_t *pos,
 780                              struct thandle *th, int rq)
 781 {
 782         struct thandle  *sub_th;
 783         bool            record_update;
 784         ssize_t         rc;
 785         ENTRY;
 786
 787         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 788         if (IS_ERR(sub_th))
 789                 RETURN(PTR_ERR(sub_th));
 790
 791         if (record_update) {
 792                 rc = update_record_pack(write, th, lu_object_fid(&dt->do_lu),
 793                                         buf, *pos);
 794                 if (rc < 0)
 795                         RETURN(rc);
 796         }
 797
 798         rc = dt_write(env, dt, buf, pos, sub_th, rq);
 799         RETURN(rc);
 800 }
 801
 802 /**
 803  * Declare punch
 804  *
 805  * Get transaction of next layer and declare punch.
 806  *
 807  * \param[in] env       execution environment
 808  * \param[in] dt        object to be written
 809  * \param[in] start     start offset of punch
 810  * \param[in] end       end offet of punch
 811  * \param[in] th        transaction handle
 812  *
 813  * \retval              0 if the insertion succeeds.
 814  * \retval              negative errno if the insertion fails.
 815  */
 816 int lod_sub_object_declare_punch(const struct lu_env *env,
 817                                  struct dt_object *dt,
 818                                  __u64 start, __u64 end,
 819                                  struct thandle *th)
 820 {
 821         struct thandle  *sub_th;
 822         int             rc;
 823         ENTRY;
 824
 825         sub_th = lod_sub_get_thandle(env, th, dt, NULL);
 826         if (IS_ERR(sub_th))
 827                 RETURN(PTR_ERR(sub_th));
 828
 829         rc = dt_declare_punch(env, dt, start, end, sub_th);
 830
 831         RETURN(rc);
 832 }
 833
 834 /**
 835  * Punch to sub object
 836  *
 837  * Get transaction of next layer, records buffer write if it belongs to
 838  * Cross-MDT operation, and punch object.
 839  *
 840  * \param[in] env       execution environment
 841  * \param[in] dt        object to be written
 842  * \param[in] start     start offset of punch
 843  * \param[in] end       end offset of punch
 844  * \param[in] th        transaction handle
 845  * \param[in] capa      capability of the write
 846  *
 847  * \retval              the buffer size in bytes if it succeeds.
 848  * \retval              negative errno if it fails.
 849  */
 850 int lod_sub_object_punch(const struct lu_env *env, struct dt_object *dt,
 851                          __u64 start, __u64 end, struct thandle *th)
 852 {
 853         struct thandle  *sub_th;
 854         bool            record_update;
 855         int             rc;
 856         ENTRY;
 857
 858         sub_th = lod_sub_get_thandle(env, th, dt, &record_update);
 859         if (IS_ERR(sub_th))
 860                 RETURN(PTR_ERR(sub_th));
 861
 862         if (record_update) {
 863                 rc = update_record_pack(punch, th, lu_object_fid(&dt->do_lu),
 864                                         start, end);
 865                 if (rc < 0)
 866                         RETURN(rc);
 867         }
 868
 869         rc = dt_punch(env, dt, start, end, sub_th);
 870
 871         RETURN(rc);
 872 }
 873
 874 int lod_sub_prep_llog(const struct lu_env *env, struct lod_device *lod,
 875                       struct dt_device *dt, int index)
 876 {
 877         struct lod_thread_info  *lti = lod_env_info(env);
 878         struct llog_ctxt        *ctxt;
 879         struct llog_handle      *lgh;
 880         struct llog_catid       *cid = &lti->lti_cid;
 881         struct lu_fid           *fid = &lti->lti_fid;
 882         struct obd_device       *obd;
 883         int                     rc;
 884         ENTRY;
 885
 886         lu_update_log_fid(fid, index);
 887         fid_to_logid(fid, &cid->lci_logid);
 888
 889         obd = dt->dd_lu_dev.ld_obd;
 890         ctxt = llog_get_context(obd, LLOG_UPDATELOG_ORIG_CTXT);
 891         LASSERT(ctxt != NULL);
 892         ctxt->loc_flags |= LLOG_CTXT_FLAG_NORMAL_FID;
 893
 894         rc = llog_open(env, ctxt, &lgh, &cid->lci_logid, NULL,
 895                        LLOG_OPEN_EXISTS);
 896         if (rc < 0) {
 897                 llog_ctxt_put(ctxt);
 898                 RETURN(rc);
 899         }
 900
 901         LASSERT(lgh != NULL);
 902         ctxt->loc_handle = lgh;
 903
 904         rc = llog_cat_init_and_process(env, lgh);
 905         if (rc != 0) {
 906                 llog_cat_close(env, ctxt->loc_handle);
 907                 ctxt->loc_handle = NULL;
 908         }
 909
 910         llog_ctxt_put(ctxt);
 911
 912         RETURN(rc);
 913 }