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