lustre/lod/lod_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  2009 Sun Microsystems, Inc. All rights reserved
  24  * Use is subject to license terms.
  25  *
  26  * Copyright (c) 2012, Intel Corporation.
  27  */
  28 /*
  29  * lustre/lod/lod_object.c
  30  *
  31  * Author: Alex Zhuravlev <alexey.zhuravlev@intel.com>
  32  */
  33
  34 #ifndef EXPORT_SYMTAB
  35 # define EXPORT_SYMTAB
  36 #endif
  37 #define DEBUG_SUBSYSTEM S_MDS
  38
  39 #include <obd.h>
  40 #include <obd_class.h>
  41 #include <lustre_ver.h>
  42 #include <obd_support.h>
  43 #include <lprocfs_status.h>
  44
  45 #include <lustre_fid.h>
  46 #include <lustre_param.h>
  47 #include <lustre_fid.h>
  48 #include <obd_lov.h>
  49
  50 #include "lod_internal.h"
  51
  52 extern cfs_mem_cache_t *lod_object_kmem;
  53 static const struct dt_body_operations lod_body_lnk_ops;
  54
  55 static int lod_index_lookup(const struct lu_env *env, struct dt_object *dt,
  56                             struct dt_rec *rec, const struct dt_key *key,
  57                             struct lustre_capa *capa)
  58 {
  59         struct dt_object *next = dt_object_child(dt);
  60         return next->do_index_ops->dio_lookup(env, next, rec, key, capa);
  61 }
  62
  63 static int lod_declare_index_insert(const struct lu_env *env,
  64                                     struct dt_object *dt,
  65                                     const struct dt_rec *rec,
  66                                     const struct dt_key *key,
  67                                     struct thandle *handle)
  68 {
  69         return dt_declare_insert(env, dt_object_child(dt), rec, key, handle);
  70 }
  71
  72 static int lod_index_insert(const struct lu_env *env,
  73                             struct dt_object *dt,
  74                             const struct dt_rec *rec,
  75                             const struct dt_key *key,
  76                             struct thandle *th,
  77                             struct lustre_capa *capa,
  78                             int ign)
  79 {
  80         return dt_insert(env, dt_object_child(dt), rec, key, th, capa, ign);
  81 }
  82
  83 static int lod_declare_index_delete(const struct lu_env *env,
  84                                     struct dt_object *dt,
  85                                     const struct dt_key *key,
  86                                     struct thandle *th)
  87 {
  88         return dt_declare_delete(env, dt_object_child(dt), key, th);
  89 }
  90
  91 static int lod_index_delete(const struct lu_env *env,
  92                             struct dt_object *dt,
  93                             const struct dt_key *key,
  94                             struct thandle *th,
  95                             struct lustre_capa *capa)
  96 {
  97         return dt_delete(env, dt_object_child(dt), key, th, capa);
  98 }
  99
 100 static struct dt_it *lod_it_init(const struct lu_env *env,
 101                                  struct dt_object *dt, __u32 attr,
 102                                  struct lustre_capa *capa)
 103 {
 104         struct dt_object        *next = dt_object_child(dt);
 105         struct lod_it           *it = &lod_env_info(env)->lti_it;
 106         struct dt_it            *it_next;
 107
 108
 109         it_next = next->do_index_ops->dio_it.init(env, next, attr, capa);
 110         if (IS_ERR(it_next))
 111                 return it_next;
 112
 113         /* currently we do not use more than one iterator per thread
 114          * so we store it in thread info. if at some point we need
 115          * more active iterators in a single thread, we can allocate
 116          * additional ones */
 117         LASSERT(it->lit_obj == NULL);
 118
 119         it->lit_it = it_next;
 120         it->lit_obj = next;
 121
 122         return (struct dt_it *)it;
 123 }
 124
 125 #define LOD_CHECK_IT(env, it)                                   \
 126 {                                                               \
 127         LASSERT((it)->lit_obj != NULL);                         \
 128         LASSERT((it)->lit_it != NULL);                          \
 129 } while(0)
 130
 131 void lod_it_fini(const struct lu_env *env, struct dt_it *di)
 132 {
 133         struct lod_it *it = (struct lod_it *)di;
 134
 135         LOD_CHECK_IT(env, it);
 136         it->lit_obj->do_index_ops->dio_it.fini(env, it->lit_it);
 137
 138         /* the iterator not in use any more */
 139         it->lit_obj = NULL;
 140         it->lit_it = NULL;
 141 }
 142
 143 int lod_it_get(const struct lu_env *env, struct dt_it *di,
 144                const struct dt_key *key)
 145 {
 146         const struct lod_it *it = (const struct lod_it *)di;
 147
 148         LOD_CHECK_IT(env, it);
 149         return it->lit_obj->do_index_ops->dio_it.get(env, it->lit_it, key);
 150 }
 151
 152 void lod_it_put(const struct lu_env *env, struct dt_it *di)
 153 {
 154         struct lod_it *it = (struct lod_it *)di;
 155
 156         LOD_CHECK_IT(env, it);
 157         return it->lit_obj->do_index_ops->dio_it.put(env, it->lit_it);
 158 }
 159
 160 int lod_it_next(const struct lu_env *env, struct dt_it *di)
 161 {
 162         struct lod_it *it = (struct lod_it *)di;
 163
 164         LOD_CHECK_IT(env, it);
 165         return it->lit_obj->do_index_ops->dio_it.next(env, it->lit_it);
 166 }
 167
 168 struct dt_key *lod_it_key(const struct lu_env *env, const struct dt_it *di)
 169 {
 170         const struct lod_it *it = (const struct lod_it *)di;
 171
 172         LOD_CHECK_IT(env, it);
 173         return it->lit_obj->do_index_ops->dio_it.key(env, it->lit_it);
 174 }
 175
 176 int lod_it_key_size(const struct lu_env *env, const struct dt_it *di)
 177 {
 178         struct lod_it *it = (struct lod_it *)di;
 179
 180         LOD_CHECK_IT(env, it);
 181         return it->lit_obj->do_index_ops->dio_it.key_size(env, it->lit_it);
 182 }
 183
 184 int lod_it_rec(const struct lu_env *env, const struct dt_it *di,
 185                struct dt_rec *rec, __u32 attr)
 186 {
 187         const struct lod_it *it = (const struct lod_it *)di;
 188
 189         LOD_CHECK_IT(env, it);
 190         return it->lit_obj->do_index_ops->dio_it.rec(env, it->lit_it, rec, attr);
 191 }
 192
 193 __u64 lod_it_store(const struct lu_env *env, const struct dt_it *di)
 194 {
 195         const struct lod_it *it = (const struct lod_it *)di;
 196
 197         LOD_CHECK_IT(env, it);
 198         return it->lit_obj->do_index_ops->dio_it.store(env, it->lit_it);
 199 }
 200
 201 int lod_it_load(const struct lu_env *env, const struct dt_it *di, __u64 hash)
 202 {
 203         const struct lod_it *it = (const struct lod_it *)di;
 204
 205         LOD_CHECK_IT(env, it);
 206         return it->lit_obj->do_index_ops->dio_it.load(env, it->lit_it, hash);
 207 }
 208
 209 int lod_it_key_rec(const struct lu_env *env, const struct dt_it *di,
 210                    void* key_rec)
 211 {
 212         const struct lod_it *it = (const struct lod_it *)di;
 213
 214         LOD_CHECK_IT(env, it);
 215         return it->lit_obj->do_index_ops->dio_it.key_rec(env, it->lit_it, key_rec);
 216 }
 217
 218 static struct dt_index_operations lod_index_ops = {
 219         .dio_lookup             = lod_index_lookup,
 220         .dio_declare_insert     = lod_declare_index_insert,
 221         .dio_insert             = lod_index_insert,
 222         .dio_declare_delete     = lod_declare_index_delete,
 223         .dio_delete             = lod_index_delete,
 224         .dio_it = {
 225                 .init           = lod_it_init,
 226                 .fini           = lod_it_fini,
 227                 .get            = lod_it_get,
 228                 .put            = lod_it_put,
 229                 .next           = lod_it_next,
 230                 .key            = lod_it_key,
 231                 .key_size       = lod_it_key_size,
 232                 .rec            = lod_it_rec,
 233                 .store          = lod_it_store,
 234                 .load           = lod_it_load,
 235                 .key_rec        = lod_it_key_rec,
 236         }
 237 };
 238
 239 static void lod_object_read_lock(const struct lu_env *env,
 240                                  struct dt_object *dt, unsigned role)
 241 {
 242         dt_read_lock(env, dt_object_child(dt), role);
 243 }
 244
 245 static void lod_object_write_lock(const struct lu_env *env,
 246                                   struct dt_object *dt, unsigned role)
 247 {
 248         dt_write_lock(env, dt_object_child(dt), role);
 249 }
 250
 251 static void lod_object_read_unlock(const struct lu_env *env,
 252                                    struct dt_object *dt)
 253 {
 254         dt_read_unlock(env, dt_object_child(dt));
 255 }
 256
 257 static void lod_object_write_unlock(const struct lu_env *env,
 258                                     struct dt_object *dt)
 259 {
 260         dt_write_unlock(env, dt_object_child(dt));
 261 }
 262
 263 static int lod_object_write_locked(const struct lu_env *env,
 264                                    struct dt_object *dt)
 265 {
 266         return dt_write_locked(env, dt_object_child(dt));
 267 }
 268
 269 static int lod_attr_get(const struct lu_env *env,
 270                         struct dt_object *dt,
 271                         struct lu_attr *attr,
 272                         struct lustre_capa *capa)
 273 {
 274         return dt_attr_get(env, dt_object_child(dt), attr, capa);
 275 }
 276
 277 static int lod_declare_attr_set(const struct lu_env *env,
 278                                 struct dt_object *dt,
 279                                 const struct lu_attr *attr,
 280                                 struct thandle *handle)
 281 {
 282         struct dt_object  *next = dt_object_child(dt);
 283         struct lod_object *lo = lod_dt_obj(dt);
 284         int                rc, i;
 285         ENTRY;
 286
 287         /*
 288          * declare setattr on the local object
 289          */
 290         rc = dt_declare_attr_set(env, next, attr, handle);
 291         if (rc)
 292                 RETURN(rc);
 293
 294         /* osp_declare_attr_set() ignores all attributes other than
 295          * UID, GID, and size, and osp_attr_set() ignores all but UID
 296          * and GID.  Declaration of size attr setting happens through
 297          * lod_declare_init_size(), and not through this function.
 298          * Therefore we need not load striping unless ownership is
 299          * changing.  This should save memory and (we hope) speed up
 300          * rename(). */
 301         if (!(attr->la_valid & (LA_UID | LA_GID)))
 302                 RETURN(rc);
 303
 304         /*
 305          * load striping information, notice we don't do this when object
 306          * is being initialized as we don't need this information till
 307          * few specific cases like destroy, chown
 308          */
 309         rc = lod_load_striping(env, lo);
 310         if (rc)
 311                 RETURN(rc);
 312
 313         /*
 314          * if object is striped declare changes on the stripes
 315          */
 316         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
 317         for (i = 0; i < lo->ldo_stripenr; i++) {
 318                 LASSERT(lo->ldo_stripe[i]);
 319                 rc = dt_declare_attr_set(env, lo->ldo_stripe[i], attr, handle);
 320                 if (rc) {
 321                         CERROR("failed declaration: %d\n", rc);
 322                         break;
 323                 }
 324         }
 325
 326         RETURN(rc);
 327 }
 328
 329 static int lod_attr_set(const struct lu_env *env,
 330                         struct dt_object *dt,
 331                         const struct lu_attr *attr,
 332                         struct thandle *handle,
 333                         struct lustre_capa *capa)
 334 {
 335         struct dt_object  *next = dt_object_child(dt);
 336         struct lod_object *lo = lod_dt_obj(dt);
 337         int                rc, i;
 338         ENTRY;
 339
 340         /*
 341          * apply changes to the local object
 342          */
 343         rc = dt_attr_set(env, next, attr, handle, capa);
 344         if (rc)
 345                 RETURN(rc);
 346
 347         if (!(attr->la_valid & (LA_UID | LA_GID)))
 348                 RETURN(rc);
 349
 350         /*
 351          * if object is striped, apply changes to all the stripes
 352          */
 353         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
 354         for (i = 0; i < lo->ldo_stripenr; i++) {
 355                 LASSERT(lo->ldo_stripe[i]);
 356                 rc = dt_attr_set(env, lo->ldo_stripe[i], attr, handle, capa);
 357                 if (rc) {
 358                         CERROR("failed declaration: %d\n", rc);
 359                         break;
 360                 }
 361         }
 362
 363         RETURN(rc);
 364 }
 365
 366 static int lod_xattr_get(const struct lu_env *env, struct dt_object *dt,
 367                          struct lu_buf *buf, const char *name,
 368                          struct lustre_capa *capa)
 369 {
 370         struct lod_thread_info  *info = lod_env_info(env);
 371         struct lod_device       *dev = lu2lod_dev(dt->do_lu.lo_dev);
 372         int                      rc, is_root;
 373         ENTRY;
 374
 375         rc = dt_xattr_get(env, dt_object_child(dt), buf, name, capa);
 376         if (rc != -ENODATA || !S_ISDIR(dt->do_lu.lo_header->loh_attr & S_IFMT))
 377                 RETURN(rc);
 378
 379         /*
 380          * lod returns default striping on the real root of the device
 381          * this is like the root stores default striping for the whole
 382          * filesystem. historically we've been using a different approach
 383          * and store it in the config.
 384          */
 385         dt_root_get(env, dev->lod_child, &info->lti_fid);
 386         is_root = lu_fid_eq(&info->lti_fid, lu_object_fid(&dt->do_lu));
 387
 388         if (is_root && strcmp(XATTR_NAME_LOV, name) == 0) {
 389                 struct lov_user_md *lum = buf->lb_buf;
 390                 struct lov_desc    *desc = &dev->lod_desc;
 391
 392                 if (buf->lb_buf == NULL) {
 393                         rc = sizeof(struct lov_user_md_v1);
 394                 } else if (buf->lb_len >= sizeof(struct lov_user_md_v1)) {
 395                         lum->lmm_magic = LOV_USER_MAGIC_V1;
 396                         lmm_oi_set_seq(&lum->lmm_oi, FID_SEQ_LOV_DEFAULT);
 397                         lum->lmm_pattern = desc->ld_pattern;
 398                         lum->lmm_stripe_size = desc->ld_default_stripe_size;
 399                         lum->lmm_stripe_count = desc->ld_default_stripe_count;
 400                         lum->lmm_stripe_offset = desc->ld_default_stripe_offset;
 401                         rc = sizeof(struct lov_user_md_v1);
 402                 } else {
 403                         rc = -ERANGE;
 404                 }
 405         }
 406
 407         RETURN(rc);
 408 }
 409
 410 /*
 411  * LOV xattr is a storage for striping, and LOD owns this xattr.
 412  * but LOD allows others to control striping to some extent
 413  * - to reset strping
 414  * - to set new defined striping
 415  * - to set new semi-defined striping
 416  *   - number of stripes is defined
 417  *   - number of stripes + osts are defined
 418  *   - ??
 419  */
 420 static int lod_declare_xattr_set(const struct lu_env *env,
 421                                  struct dt_object *dt,
 422                                  const struct lu_buf *buf,
 423                                  const char *name, int fl,
 424                                  struct thandle *th)
 425 {
 426         struct dt_object *next = dt_object_child(dt);
 427         struct lu_attr   *attr = &lod_env_info(env)->lti_attr;
 428         __u32             mode;
 429         int               rc;
 430         ENTRY;
 431
 432         /*
 433          * allow to declare predefined striping on a new (!mode) object
 434          * which is supposed to be replay of regular file creation
 435          * (when LOV setting is declared)
 436          * LU_XATTR_REPLACE is set to indicate a layout swap
 437          */
 438         mode = dt->do_lu.lo_header->loh_attr & S_IFMT;
 439         if ((S_ISREG(mode) || !mode) && !strcmp(name, XATTR_NAME_LOV) &&
 440              !(fl & LU_XATTR_REPLACE)) {
 441                 /*
 442                  * this is a request to manipulate object's striping
 443                  */
 444                 if (dt_object_exists(dt)) {
 445                         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
 446                         if (rc)
 447                                 RETURN(rc);
 448                 } else {
 449                         memset(attr, 0, sizeof(*attr));
 450                         attr->la_valid = LA_TYPE | LA_MODE;
 451                         attr->la_mode = S_IFREG;
 452                 }
 453                 rc = lod_declare_striped_object(env, dt, attr, buf, th);
 454                 if (rc)
 455                         RETURN(rc);
 456         }
 457
 458         rc = dt_declare_xattr_set(env, next, buf, name, fl, th);
 459
 460         RETURN(rc);
 461 }
 462
 463 static int lod_xattr_set_lov_on_dir(const struct lu_env *env,
 464                                     struct dt_object *dt,
 465                                     const struct lu_buf *buf,
 466                                     const char *name, int fl,
 467                                     struct thandle *th,
 468                                     struct lustre_capa *capa)
 469 {
 470         struct lod_device       *d = lu2lod_dev(dt->do_lu.lo_dev);
 471         struct dt_object        *next = dt_object_child(dt);
 472         struct lod_object       *l = lod_dt_obj(dt);
 473         struct lov_user_md_v1   *lum;
 474         struct lov_user_md_v3   *v3 = NULL;
 475         int                      rc;
 476         ENTRY;
 477
 478         LASSERT(l->ldo_stripe == NULL);
 479         l->ldo_striping_cached = 0;
 480         l->ldo_def_striping_set = 0;
 481         lod_object_set_pool(l, NULL);
 482         l->ldo_def_stripe_size = 0;
 483         l->ldo_def_stripenr = 0;
 484
 485         LASSERT(buf);
 486         LASSERT(buf->lb_buf);
 487         lum = buf->lb_buf;
 488
 489         rc = lod_verify_striping(d, buf, 0);
 490         if (rc)
 491                 RETURN(rc);
 492
 493         if (lum->lmm_magic == LOV_USER_MAGIC_V3)
 494                 v3 = buf->lb_buf;
 495
 496         /* if { size, offset, count } = { 0, -1, 0 } and no pool
 497          * (i.e. all default values specified) then delete default
 498          * striping from dir. */
 499         CDEBUG(D_OTHER,
 500                 "set default striping: sz %u # %u offset %d %s %s\n",
 501                 (unsigned)lum->lmm_stripe_size,
 502                 (unsigned)lum->lmm_stripe_count,
 503                 (int)lum->lmm_stripe_offset,
 504                 v3 ? "from" : "", v3 ? v3->lmm_pool_name : "");
 505
 506         if (LOVEA_DELETE_VALUES((lum->lmm_stripe_size),
 507                                 (lum->lmm_stripe_count),
 508                                 (lum->lmm_stripe_offset)) &&
 509                         lum->lmm_magic == LOV_USER_MAGIC_V1) {
 510                 rc = dt_xattr_del(env, next, name, th, capa);
 511                 if (rc == -ENODATA)
 512                         rc = 0;
 513         } else {
 514                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 515         }
 516
 517         RETURN(rc);
 518 }
 519
 520 static int lod_xattr_set(const struct lu_env *env,
 521                          struct dt_object *dt, const struct lu_buf *buf,
 522                          const char *name, int fl, struct thandle *th,
 523                          struct lustre_capa *capa)
 524 {
 525         struct dt_object        *next = dt_object_child(dt);
 526         __u32                    attr;
 527         int                      rc;
 528         ENTRY;
 529
 530         attr = dt->do_lu.lo_header->loh_attr & S_IFMT;
 531         if (S_ISDIR(attr)) {
 532                 if (strncmp(name, XATTR_NAME_LOV, strlen(XATTR_NAME_LOV)) == 0)
 533                         rc = lod_xattr_set_lov_on_dir(env, dt, buf, name,
 534                                                       fl, th, capa);
 535                 else
 536                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 537
 538         } else if (S_ISREG(attr) && !strcmp(name, XATTR_NAME_LOV)) {
 539                 /* in case of lov EA swap, just set it
 540                  * if not, it is a replay so check striping match what we
 541                  * already have during req replay, declare_xattr_set()
 542                  * defines striping, then create() does the work
 543                 */
 544                 if (fl & LU_XATTR_REPLACE) {
 545                         /* free stripes, then update disk */
 546                         lod_object_free_striping(env, lod_dt_obj(dt));
 547                         rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 548                 } else {
 549                         rc = lod_striping_create(env, dt, NULL, NULL, th);
 550                 }
 551                 RETURN(rc);
 552         } else {
 553                 /*
 554                  * behave transparantly for all other EAs
 555                  */
 556                 rc = dt_xattr_set(env, next, buf, name, fl, th, capa);
 557         }
 558
 559         RETURN(rc);
 560 }
 561
 562 static int lod_declare_xattr_del(const struct lu_env *env,
 563                                  struct dt_object *dt, const char *name,
 564                                  struct thandle *th)
 565 {
 566         return dt_declare_xattr_del(env, dt_object_child(dt), name, th);
 567 }
 568
 569 static int lod_xattr_del(const struct lu_env *env, struct dt_object *dt,
 570                          const char *name, struct thandle *th,
 571                          struct lustre_capa *capa)
 572 {
 573         return dt_xattr_del(env, dt_object_child(dt), name, th, capa);
 574 }
 575
 576 static int lod_xattr_list(const struct lu_env *env,
 577                           struct dt_object *dt, struct lu_buf *buf,
 578                           struct lustre_capa *capa)
 579 {
 580         return dt_xattr_list(env, dt_object_child(dt), buf, capa);
 581 }
 582
 583 int lod_object_set_pool(struct lod_object *o, char *pool)
 584 {
 585         int len;
 586
 587         if (o->ldo_pool) {
 588                 len = strlen(o->ldo_pool);
 589                 OBD_FREE(o->ldo_pool, len + 1);
 590                 o->ldo_pool = NULL;
 591         }
 592         if (pool) {
 593                 len = strlen(pool);
 594                 OBD_ALLOC(o->ldo_pool, len + 1);
 595                 if (o->ldo_pool == NULL)
 596                         return -ENOMEM;
 597                 strcpy(o->ldo_pool, pool);
 598         }
 599         return 0;
 600 }
 601
 602 static inline int lod_object_will_be_striped(int is_reg, const struct lu_fid *fid)
 603 {
 604         return (is_reg && fid_seq(fid) != FID_SEQ_LOCAL_FILE);
 605 }
 606
 607 static int lod_cache_parent_striping(const struct lu_env *env,
 608                                      struct lod_object *lp)
 609 {
 610         struct lov_user_md_v1   *v1 = NULL;
 611         struct lov_user_md_v3   *v3 = NULL;
 612         int                      rc;
 613         ENTRY;
 614
 615         /* dt_ah_init() is called from MDD without parent being write locked
 616          * lock it here */
 617         dt_write_lock(env, dt_object_child(&lp->ldo_obj), 0);
 618         if (lp->ldo_striping_cached)
 619                 GOTO(unlock, rc = 0);
 620
 621         rc = lod_get_lov_ea(env, lp);
 622         if (rc < 0)
 623                 GOTO(unlock, rc);
 624
 625         if (rc < sizeof(struct lov_user_md)) {
 626                 /* don't lookup for non-existing or invalid striping */
 627                 lp->ldo_def_striping_set = 0;
 628                 lp->ldo_striping_cached = 1;
 629                 lp->ldo_def_stripe_size = 0;
 630                 lp->ldo_def_stripenr = 0;
 631                 lp->ldo_def_stripe_offset = (typeof(v1->lmm_stripe_offset))(-1);
 632                 GOTO(unlock, rc = 0);
 633         }
 634
 635         v1 = (struct lov_user_md_v1 *)lod_env_info(env)->lti_ea_store;
 636         if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V1))
 637                 lustre_swab_lov_user_md_v1(v1);
 638         else if (v1->lmm_magic == __swab32(LOV_USER_MAGIC_V3))
 639                 lustre_swab_lov_user_md_v3(v3);
 640
 641         if (v1->lmm_magic != LOV_MAGIC_V3 && v1->lmm_magic != LOV_MAGIC_V1)
 642                 GOTO(unlock, rc = 0);
 643
 644         if (v1->lmm_pattern != LOV_PATTERN_RAID0 && v1->lmm_pattern != 0)
 645                 GOTO(unlock, rc = 0);
 646
 647         lp->ldo_def_stripenr = v1->lmm_stripe_count;
 648         lp->ldo_def_stripe_size = v1->lmm_stripe_size;
 649         lp->ldo_def_stripe_offset = v1->lmm_stripe_offset;
 650         lp->ldo_striping_cached = 1;
 651         lp->ldo_def_striping_set = 1;
 652
 653         if (v1->lmm_magic == LOV_USER_MAGIC_V3) {
 654                 /* XXX: sanity check here */
 655                 v3 = (struct lov_user_md_v3 *) v1;
 656                 if (v3->lmm_pool_name[0])
 657                         lod_object_set_pool(lp, v3->lmm_pool_name);
 658         }
 659
 660         CDEBUG(D_OTHER, "def. striping: # %d, sz %d, off %d %s%s on "DFID"\n",
 661                lp->ldo_def_stripenr, lp->ldo_def_stripe_size,
 662                lp->ldo_def_stripe_offset, v3 ? "from " : "",
 663                v3 ? lp->ldo_pool : "", PFID(lu_object_fid(&lp->ldo_obj.do_lu)));
 664
 665         EXIT;
 666 unlock:
 667         dt_write_unlock(env, dt_object_child(&lp->ldo_obj));
 668         return rc;
 669 }
 670
 671 /**
 672  * used to transfer default striping data to the object being created
 673  */
 674 static void lod_ah_init(const struct lu_env *env,
 675                         struct dt_allocation_hint *ah,
 676                         struct dt_object *parent,
 677                         struct dt_object *child,
 678                         cfs_umode_t child_mode)
 679 {
 680         struct lod_device *d = lu2lod_dev(child->do_lu.lo_dev);
 681         struct dt_object  *nextp = NULL;
 682         struct dt_object  *nextc;
 683         struct lod_object *lp = NULL;
 684         struct lod_object *lc;
 685         struct lov_desc   *desc;
 686         ENTRY;
 687
 688         LASSERT(child);
 689
 690         if (likely(parent)) {
 691                 nextp = dt_object_child(parent);
 692                 lp = lod_dt_obj(parent);
 693         }
 694
 695         nextc = dt_object_child(child);
 696         lc = lod_dt_obj(child);
 697
 698         LASSERT(lc->ldo_stripenr == 0);
 699         LASSERT(lc->ldo_stripe == NULL);
 700
 701         /*
 702          * local object may want some hints
 703          * in case of late striping creation, ->ah_init()
 704          * can be called with local object existing
 705          */
 706         if (!dt_object_exists(nextc) || dt_object_remote(nextc))
 707                 nextc->do_ops->do_ah_init(env, ah, nextp, nextc, child_mode);
 708
 709         if (S_ISDIR(child_mode)) {
 710                 if (lp->ldo_striping_cached == 0) {
 711                         /* we haven't tried to get default striping for
 712                          * the directory yet, let's cache it in the object */
 713                         lod_cache_parent_striping(env, lp);
 714                 }
 715                 /* transfer defaults to new directory */
 716                 if (lp->ldo_striping_cached) {
 717                         if (lp->ldo_pool)
 718                                 lod_object_set_pool(lc, lp->ldo_pool);
 719                         lc->ldo_def_stripenr = lp->ldo_def_stripenr;
 720                         lc->ldo_def_stripe_size = lp->ldo_def_stripe_size;
 721                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
 722                         lc->ldo_striping_cached = 1;
 723                         lc->ldo_def_striping_set = 1;
 724                         CDEBUG(D_OTHER, "inherite EA sz:%d off:%d nr:%d\n",
 725                                (int)lc->ldo_def_stripenr,
 726                                (int)lc->ldo_def_stripe_size,
 727                                (int)lc->ldo_def_stripe_offset);
 728                 }
 729                 return;
 730         }
 731
 732         /*
 733          * if object is going to be striped over OSTs, transfer default
 734          * striping information to the child, so that we can use it
 735          * during declaration and creation
 736          */
 737         if (!lod_object_will_be_striped(S_ISREG(child_mode),
 738                                         lu_object_fid(&child->do_lu)))
 739                 return;
 740
 741         /*
 742          * try from the parent
 743          */
 744         if (likely(parent)) {
 745                 if (lp->ldo_striping_cached == 0) {
 746                         /* we haven't tried to get default striping for
 747                          * the directory yet, let's cache it in the object */
 748                         lod_cache_parent_striping(env, lp);
 749                 }
 750
 751                 lc->ldo_def_stripe_offset = (__u16) -1;
 752
 753                 if (lp->ldo_def_striping_set) {
 754                         if (lp->ldo_pool)
 755                                 lod_object_set_pool(lc, lp->ldo_pool);
 756                         lc->ldo_stripenr = lp->ldo_def_stripenr;
 757                         lc->ldo_stripe_size = lp->ldo_def_stripe_size;
 758                         lc->ldo_def_stripe_offset = lp->ldo_def_stripe_offset;
 759                         CDEBUG(D_OTHER, "striping from parent: #%d, sz %d %s\n",
 760                                lc->ldo_stripenr, lc->ldo_stripe_size,
 761                                lp->ldo_pool ? lp->ldo_pool : "");
 762                 }
 763         }
 764
 765         /*
 766          * if the parent doesn't provide with specific pattern, grab fs-wide one
 767          */
 768         desc = &d->lod_desc;
 769         if (lc->ldo_stripenr == 0)
 770                 lc->ldo_stripenr = desc->ld_default_stripe_count;
 771         if (lc->ldo_stripe_size == 0)
 772                 lc->ldo_stripe_size = desc->ld_default_stripe_size;
 773         CDEBUG(D_OTHER, "final striping: # %d stripes, sz %d from %s\n",
 774                lc->ldo_stripenr, lc->ldo_stripe_size,
 775                lc->ldo_pool ? lc->ldo_pool : "");
 776
 777         EXIT;
 778 }
 779
 780 #define ll_do_div64(aaa,bbb)    do_div((aaa), (bbb))
 781 /*
 782  * this function handles a special case when truncate was done
 783  * on a stripeless object and now striping is being created
 784  * we can't lose that size, so we have to propagate it to newly
 785  * created object
 786  */
 787 static int lod_declare_init_size(const struct lu_env *env,
 788                                  struct dt_object *dt, struct thandle *th)
 789 {
 790         struct dt_object   *next = dt_object_child(dt);
 791         struct lod_object  *lo = lod_dt_obj(dt);
 792         struct lu_attr     *attr = &lod_env_info(env)->lti_attr;
 793         uint64_t            size, offs;
 794         int                 rc, stripe;
 795         ENTRY;
 796
 797         /* XXX: we support the simplest (RAID0) striping so far */
 798         LASSERT(lo->ldo_stripe || lo->ldo_stripenr == 0);
 799         LASSERT(lo->ldo_stripe_size > 0);
 800
 801         rc = dt_attr_get(env, next, attr, BYPASS_CAPA);
 802         LASSERT(attr->la_valid & LA_SIZE);
 803         if (rc)
 804                 RETURN(rc);
 805
 806         size = attr->la_size;
 807         if (size == 0)
 808                 RETURN(0);
 809
 810         /* ll_do_div64(a, b) returns a % b, and a = a / b */
 811         ll_do_div64(size, (__u64) lo->ldo_stripe_size);
 812         stripe = ll_do_div64(size, (__u64) lo->ldo_stripenr);
 813
 814         size = size * lo->ldo_stripe_size;
 815         offs = attr->la_size;
 816         size += ll_do_div64(offs, lo->ldo_stripe_size);
 817
 818         attr->la_valid = LA_SIZE;
 819         attr->la_size = size;
 820
 821         rc = dt_declare_attr_set(env, lo->ldo_stripe[stripe], attr, th);
 822
 823         RETURN(rc);
 824 }
 825
 826
 827 /**
 828  * Create declaration of striped object
 829  */
 830 int lod_declare_striped_object(const struct lu_env *env, struct dt_object *dt,
 831                                struct lu_attr *attr,
 832                                const struct lu_buf *lovea, struct thandle *th)
 833 {
 834         struct lod_thread_info  *info = lod_env_info(env);
 835         struct dt_object        *next = dt_object_child(dt);
 836         struct lod_object       *lo = lod_dt_obj(dt);
 837         int                      rc;
 838         ENTRY;
 839
 840         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_ALLOC_OBDO)) {
 841                 /* failed to create striping, let's reset
 842                  * config so that others don't get confused */
 843                 lod_object_free_striping(env, lo);
 844                 GOTO(out, rc = -ENOMEM);
 845         }
 846
 847         /* choose OST and generate appropriate objects */
 848         rc = lod_qos_prep_create(env, lo, attr, lovea, th);
 849         if (rc) {
 850                 /* failed to create striping, let's reset
 851                  * config so that others don't get confused */
 852                 lod_object_free_striping(env, lo);
 853                 GOTO(out, rc);
 854         }
 855
 856         /*
 857          * declare storage for striping data
 858          */
 859         info->lti_buf.lb_len = lov_mds_md_size(lo->ldo_stripenr,
 860                                 lo->ldo_pool ?  LOV_MAGIC_V3 : LOV_MAGIC_V1);
 861         rc = dt_declare_xattr_set(env, next, &info->lti_buf, XATTR_NAME_LOV,
 862                                   0, th);
 863         if (rc)
 864                 GOTO(out, rc);
 865
 866         /*
 867          * if striping is created with local object's size > 0,
 868          * we have to propagate this size to specific object
 869          * the case is possible only when local object was created previously
 870          */
 871         if (dt_object_exists(next))
 872                 rc = lod_declare_init_size(env, dt, th);
 873
 874 out:
 875         RETURN(rc);
 876 }
 877
 878 static int lod_declare_object_create(const struct lu_env *env,
 879                                      struct dt_object *dt,
 880                                      struct lu_attr *attr,
 881                                      struct dt_allocation_hint *hint,
 882                                      struct dt_object_format *dof,
 883                                      struct thandle *th)
 884 {
 885         struct dt_object   *next = dt_object_child(dt);
 886         struct lod_object  *lo = lod_dt_obj(dt);
 887         int                 rc;
 888         ENTRY;
 889
 890         LASSERT(dof);
 891         LASSERT(attr);
 892         LASSERT(th);
 893
 894         /*
 895          * first of all, we declare creation of local object
 896          */
 897         rc = dt_declare_create(env, next, attr, hint, dof, th);
 898         if (rc)
 899                 GOTO(out, rc);
 900
 901         if (dof->dof_type == DFT_SYM)
 902                 dt->do_body_ops = &lod_body_lnk_ops;
 903
 904         /*
 905          * it's lod_ah_init() who has decided the object will striped
 906          */
 907         if (dof->dof_type == DFT_REGULAR) {
 908                 /* callers don't want stripes */
 909                 /* XXX: all tricky interactions with ->ah_make_hint() decided
 910                  * to use striping, then ->declare_create() behaving differently
 911                  * should be cleaned */
 912                 if (dof->u.dof_reg.striped == 0)
 913                         lo->ldo_stripenr = 0;
 914                 if (lo->ldo_stripenr > 0)
 915                         rc = lod_declare_striped_object(env, dt, attr,
 916                                                         NULL, th);
 917         } else if (dof->dof_type == DFT_DIR && lo->ldo_striping_cached) {
 918                 struct lod_thread_info *info = lod_env_info(env);
 919
 920                 struct lov_user_md_v3 *v3;
 921
 922                 if (LOVEA_DELETE_VALUES(lo->ldo_def_stripe_size,
 923                                         lo->ldo_def_stripenr,
 924                                         lo->ldo_def_stripe_offset))
 925                         RETURN(0);
 926
 927                 OBD_ALLOC_PTR(v3);
 928                 if (v3 == NULL)
 929                         RETURN(-ENOMEM);
 930
 931                 v3->lmm_magic = cpu_to_le32(LOV_MAGIC_V3);
 932                 v3->lmm_pattern = cpu_to_le32(LOV_PATTERN_RAID0);
 933                 fid_to_lmm_oi(lu_object_fid(&dt->do_lu), &v3->lmm_oi);
 934                 lmm_oi_cpu_to_le(&v3->lmm_oi, &v3->lmm_oi);
 935                 v3->lmm_stripe_size = cpu_to_le32(lo->ldo_def_stripe_size);
 936                 v3->lmm_stripe_count = cpu_to_le32(lo->ldo_def_stripenr);
 937                 v3->lmm_stripe_offset = cpu_to_le16(lo->ldo_def_stripe_offset);
 938                 if (lo->ldo_pool)
 939                         strncpy(v3->lmm_pool_name, lo->ldo_pool,
 940                                 LOV_MAXPOOLNAME);
 941
 942                 info->lti_buf.lb_buf = v3;
 943                 info->lti_buf.lb_len = sizeof(*v3);
 944
 945                 /* to transfer default striping from the parent */
 946                 rc = dt_declare_xattr_set(env, next, &info->lti_buf,
 947                                           XATTR_NAME_LOV, 0, th);
 948                 OBD_FREE_PTR(v3);
 949         }
 950
 951 out:
 952         RETURN(rc);
 953 }
 954
 955 int lod_striping_create(const struct lu_env *env, struct dt_object *dt,
 956                         struct lu_attr *attr, struct dt_object_format *dof,
 957                         struct thandle *th)
 958 {
 959         struct lod_object *lo = lod_dt_obj(dt);
 960         int                rc = 0, i;
 961         ENTRY;
 962
 963         LASSERT(lo->ldo_stripe);
 964         LASSERT(lo->ldo_stripenr > 0);
 965         LASSERT(lo->ldo_striping_cached == 0);
 966
 967         /* create all underlying objects */
 968         for (i = 0; i < lo->ldo_stripenr; i++) {
 969                 LASSERT(lo->ldo_stripe[i]);
 970                 rc = dt_create(env, lo->ldo_stripe[i], attr, NULL, dof, th);
 971
 972                 if (rc)
 973                         break;
 974         }
 975         if (rc == 0)
 976                 rc = lod_generate_and_set_lovea(env, lo, th);
 977
 978         RETURN(rc);
 979 }
 980
 981 static int lod_object_create(const struct lu_env *env, struct dt_object *dt,
 982                              struct lu_attr *attr,
 983                              struct dt_allocation_hint *hint,
 984                              struct dt_object_format *dof, struct thandle *th)
 985 {
 986         struct dt_object   *next = dt_object_child(dt);
 987         struct lod_object  *lo = lod_dt_obj(dt);
 988         int                 rc;
 989         ENTRY;
 990
 991         /* create local object */
 992         rc = dt_create(env, next, attr, hint, dof, th);
 993
 994         if (rc == 0) {
 995                 if (S_ISDIR(dt->do_lu.lo_header->loh_attr))
 996                         rc = lod_store_def_striping(env, dt, th);
 997                 else if (lo->ldo_stripe)
 998                         rc = lod_striping_create(env, dt, attr, dof, th);
 999         }
1000
1001         RETURN(rc);
1002 }
1003
1004 static int lod_declare_object_destroy(const struct lu_env *env,
1005                                       struct dt_object *dt,
1006                                       struct thandle *th)
1007 {
1008         struct dt_object   *next = dt_object_child(dt);
1009         struct lod_object  *lo = lod_dt_obj(dt);
1010         int                 rc, i;
1011         ENTRY;
1012
1013         /*
1014          * we declare destroy for the local object
1015          */
1016         rc = dt_declare_destroy(env, next, th);
1017         if (rc)
1018                 RETURN(rc);
1019
1020         /*
1021          * load striping information, notice we don't do this when object
1022          * is being initialized as we don't need this information till
1023          * few specific cases like destroy, chown
1024          */
1025         rc = lod_load_striping(env, lo);
1026         if (rc)
1027                 RETURN(rc);
1028
1029         /* declare destroy for all underlying objects */
1030         for (i = 0; i < lo->ldo_stripenr; i++) {
1031                 LASSERT(lo->ldo_stripe[i]);
1032                 rc = dt_declare_destroy(env, lo->ldo_stripe[i], th);
1033
1034                 if (rc)
1035                         break;
1036         }
1037
1038         RETURN(rc);
1039 }
1040
1041 static int lod_object_destroy(const struct lu_env *env,
1042                 struct dt_object *dt, struct thandle *th)
1043 {
1044         struct dt_object  *next = dt_object_child(dt);
1045         struct lod_object *lo = lod_dt_obj(dt);
1046         int                rc, i;
1047         ENTRY;
1048
1049         /* destroy local object */
1050         rc = dt_destroy(env, next, th);
1051         if (rc)
1052                 RETURN(rc);
1053
1054         /* destroy all underlying objects */
1055         for (i = 0; i < lo->ldo_stripenr; i++) {
1056                 LASSERT(lo->ldo_stripe[i]);
1057                 rc = dt_destroy(env, lo->ldo_stripe[i], th);
1058                 if (rc)
1059                         break;
1060         }
1061
1062         RETURN(rc);
1063 }
1064
1065 static int lod_index_try(const struct lu_env *env, struct dt_object *dt,
1066                          const struct dt_index_features *feat)
1067 {
1068         struct dt_object *next = dt_object_child(dt);
1069         int               rc;
1070         ENTRY;
1071
1072         LASSERT(next->do_ops);
1073         LASSERT(next->do_ops->do_index_try);
1074
1075         rc = next->do_ops->do_index_try(env, next, feat);
1076         if (next->do_index_ops && dt->do_index_ops == NULL)
1077                 dt->do_index_ops = &lod_index_ops;
1078
1079         RETURN(rc);
1080 }
1081
1082 static int lod_declare_ref_add(const struct lu_env *env,
1083                                struct dt_object *dt, struct thandle *th)
1084 {
1085         return dt_declare_ref_add(env, dt_object_child(dt), th);
1086 }
1087
1088 static int lod_ref_add(const struct lu_env *env,
1089                        struct dt_object *dt, struct thandle *th)
1090 {
1091         return dt_ref_add(env, dt_object_child(dt), th);
1092 }
1093
1094 static int lod_declare_ref_del(const struct lu_env *env,
1095                                struct dt_object *dt, struct thandle *th)
1096 {
1097         return dt_declare_ref_del(env, dt_object_child(dt), th);
1098 }
1099
1100 static int lod_ref_del(const struct lu_env *env,
1101                        struct dt_object *dt, struct thandle *th)
1102 {
1103         return dt_ref_del(env, dt_object_child(dt), th);
1104 }
1105
1106 static struct obd_capa *lod_capa_get(const struct lu_env *env,
1107                                      struct dt_object *dt,
1108                                      struct lustre_capa *old, __u64 opc)
1109 {
1110         return dt_capa_get(env, dt_object_child(dt), old, opc);
1111 }
1112
1113 static int lod_object_sync(const struct lu_env *env, struct dt_object *dt)
1114 {
1115         return dt_object_sync(env, dt_object_child(dt));
1116 }
1117
1118 static int lod_object_lock(const struct lu_env *env,
1119                            struct dt_object *dt, struct lustre_handle *lh,
1120                            struct ldlm_enqueue_info *einfo,
1121                            void *policy)
1122 {
1123         struct dt_object   *next = dt_object_child(dt);
1124         int              rc;
1125         ENTRY;
1126
1127         /*
1128          * declare setattr on the local object
1129          */
1130         rc = dt_object_lock(env, next, lh, einfo, policy);
1131
1132         RETURN(rc);
1133 }
1134
1135 struct dt_object_operations lod_obj_ops = {
1136         .do_read_lock           = lod_object_read_lock,
1137         .do_write_lock          = lod_object_write_lock,
1138         .do_read_unlock         = lod_object_read_unlock,
1139         .do_write_unlock        = lod_object_write_unlock,
1140         .do_write_locked        = lod_object_write_locked,
1141         .do_attr_get            = lod_attr_get,
1142         .do_declare_attr_set    = lod_declare_attr_set,
1143         .do_attr_set            = lod_attr_set,
1144         .do_xattr_get           = lod_xattr_get,
1145         .do_declare_xattr_set   = lod_declare_xattr_set,
1146         .do_xattr_set           = lod_xattr_set,
1147         .do_declare_xattr_del   = lod_declare_xattr_del,
1148         .do_xattr_del           = lod_xattr_del,
1149         .do_xattr_list          = lod_xattr_list,
1150         .do_ah_init             = lod_ah_init,
1151         .do_declare_create      = lod_declare_object_create,
1152         .do_create              = lod_object_create,
1153         .do_declare_destroy     = lod_declare_object_destroy,
1154         .do_destroy             = lod_object_destroy,
1155         .do_index_try           = lod_index_try,
1156         .do_declare_ref_add     = lod_declare_ref_add,
1157         .do_ref_add             = lod_ref_add,
1158         .do_declare_ref_del     = lod_declare_ref_del,
1159         .do_ref_del             = lod_ref_del,
1160         .do_capa_get            = lod_capa_get,
1161         .do_object_sync         = lod_object_sync,
1162         .do_object_lock         = lod_object_lock,
1163 };
1164
1165 static ssize_t lod_read(const struct lu_env *env, struct dt_object *dt,
1166                         struct lu_buf *buf, loff_t *pos,
1167                         struct lustre_capa *capa)
1168 {
1169         struct dt_object *next = dt_object_child(dt);
1170         return next->do_body_ops->dbo_read(env, next, buf, pos, capa);
1171 }
1172
1173 static ssize_t lod_declare_write(const struct lu_env *env,
1174                                  struct dt_object *dt,
1175                                  const loff_t size, loff_t pos,
1176                                  struct thandle *th)
1177 {
1178         return dt_declare_record_write(env, dt_object_child(dt),
1179                                        size, pos, th);
1180 }
1181
1182 static ssize_t lod_write(const struct lu_env *env, struct dt_object *dt,
1183                          const struct lu_buf *buf, loff_t *pos,
1184                          struct thandle *th, struct lustre_capa *capa, int iq)
1185 {
1186         struct dt_object *next = dt_object_child(dt);
1187         LASSERT(next);
1188         return next->do_body_ops->dbo_write(env, next, buf, pos, th, capa, iq);
1189 }
1190
1191 static const struct dt_body_operations lod_body_lnk_ops = {
1192         .dbo_read               = lod_read,
1193         .dbo_declare_write      = lod_declare_write,
1194         .dbo_write              = lod_write
1195 };
1196
1197 static int lod_object_init(const struct lu_env *env, struct lu_object *o,
1198                            const struct lu_object_conf *conf)
1199 {
1200         struct lod_device *d = lu2lod_dev(o->lo_dev);
1201         struct lu_object  *below;
1202         struct lu_device  *under;
1203         ENTRY;
1204
1205         /*
1206          * create local object
1207          */
1208         under = &d->lod_child->dd_lu_dev;
1209         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
1210         if (below == NULL)
1211                 RETURN(-ENOMEM);
1212
1213         lu_object_add(o, below);
1214
1215         RETURN(0);
1216 }
1217
1218 void lod_object_free_striping(const struct lu_env *env, struct lod_object *lo)
1219 {
1220         int i;
1221
1222         if (lo->ldo_stripe) {
1223                 LASSERT(lo->ldo_stripes_allocated > 0);
1224
1225                 for (i = 0; i < lo->ldo_stripenr; i++) {
1226                         if (lo->ldo_stripe[i])
1227                                 lu_object_put(env, &lo->ldo_stripe[i]->do_lu);
1228                 }
1229
1230                 i = sizeof(struct dt_object *) * lo->ldo_stripes_allocated;
1231                 OBD_FREE(lo->ldo_stripe, i);
1232                 lo->ldo_stripe = NULL;
1233                 lo->ldo_stripes_allocated = 0;
1234         }
1235         lo->ldo_stripenr = 0;
1236 }
1237
1238 /*
1239  * ->start is called once all slices are initialized, including header's
1240  * cache for mode (object type). using the type we can initialize ops
1241  */
1242 static int lod_object_start(const struct lu_env *env, struct lu_object *o)
1243 {
1244         if (S_ISLNK(o->lo_header->loh_attr & S_IFMT))
1245                 lu2lod_obj(o)->ldo_obj.do_body_ops = &lod_body_lnk_ops;
1246         return 0;
1247 }
1248
1249 static void lod_object_free(const struct lu_env *env, struct lu_object *o)
1250 {
1251         struct lod_object *mo = lu2lod_obj(o);
1252
1253         /*
1254          * release all underlying object pinned
1255          */
1256
1257         lod_object_free_striping(env, mo);
1258
1259         lod_object_set_pool(mo, NULL);
1260
1261         lu_object_fini(o);
1262         OBD_SLAB_FREE_PTR(mo, lod_object_kmem);
1263 }
1264
1265 static void lod_object_release(const struct lu_env *env, struct lu_object *o)
1266 {
1267         /* XXX: shouldn't we release everything here in case if object
1268          * creation failed before? */
1269 }
1270
1271 static int lod_object_print(const struct lu_env *env, void *cookie,
1272                             lu_printer_t p, const struct lu_object *l)
1273 {
1274         struct lod_object *o = lu2lod_obj((struct lu_object *) l);
1275
1276         return (*p)(env, cookie, LUSTRE_LOD_NAME"-object@%p", o);
1277 }
1278
1279 struct lu_object_operations lod_lu_obj_ops = {
1280         .loo_object_init        = lod_object_init,
1281         .loo_object_start       = lod_object_start,
1282         .loo_object_free        = lod_object_free,
1283         .loo_object_release     = lod_object_release,
1284         .loo_object_print       = lod_object_print,
1285 };
1286
1287 /**
1288  * Init remote lod object
1289  */
1290 static int lod_robject_init(const struct lu_env *env, struct lu_object *lo,
1291                             const struct lu_object_conf *conf)
1292 {
1293         struct lod_device *lod = lu2lod_dev(lo->lo_dev);
1294         struct lod_tgt_descs *ltd = &lod->lod_mdt_descs;
1295         struct lu_device  *c_dev = NULL;
1296         struct lu_object  *c_obj;
1297         int i;
1298         ENTRY;
1299
1300         lod_getref(ltd);
1301         if (ltd->ltd_tgts_size > 0) {
1302                 cfs_foreach_bit(ltd->ltd_tgt_bitmap, i) {
1303                         struct lod_tgt_desc *tgt;
1304                         tgt = LTD_TGT(ltd, i);
1305                         LASSERT(tgt && tgt->ltd_tgt);
1306                         if (tgt->ltd_index ==
1307                             lu2lod_obj(lo)->ldo_mds_num) {
1308                                 c_dev = &(tgt->ltd_tgt->dd_lu_dev);
1309                                 break;
1310                         }
1311                 }
1312         }
1313         lod_putref(lod, ltd);
1314
1315         if (unlikely(c_dev == NULL))
1316                 RETURN(-ENOENT);
1317
1318         c_obj = c_dev->ld_ops->ldo_object_alloc(env, lo->lo_header, c_dev);
1319         if (unlikely(c_obj == NULL))
1320                 RETURN(-ENOMEM);
1321
1322         lu_object_add(lo, c_obj);
1323
1324         RETURN(0);
1325 }
1326
1327 struct lu_object_operations lod_lu_robj_ops = {
1328         .loo_object_init      = lod_robject_init,
1329         .loo_object_start     = lod_object_start,
1330         .loo_object_free      = lod_object_free,
1331         .loo_object_release   = lod_object_release,
1332         .loo_object_print     = lod_object_print,
1333 };