lustre/lod/lod_sub_object.c

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