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