lustre/lov/lov_ea.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/lov/lov_ea.c
  37  *
  38  * Author: Wang Di <wangdi@clusterfs.com>
  39  */
  40
  41 #ifndef EXPORT_SYMTAB
  42 # define EXPORT_SYMTAB
  43 #endif
  44 #define DEBUG_SUBSYSTEM S_LOV
  45
  46 #ifdef __KERNEL__
  47 #include <asm/div64.h>
  48 #include <libcfs/libcfs.h>
  49 #else
  50 #include <liblustre.h>
  51 #endif
  52
  53 #include <obd_class.h>
  54 #include <obd_lov.h>
  55 #include <lustre/lustre_idl.h>
  56 #include <lustre_log.h>
  57
  58 #include "lov_internal.h"
  59
  60 struct lovea_unpack_args {
  61         struct lov_stripe_md *lsm;
  62         int                   cursor;
  63 };
  64
  65 static int lsm_lmm_verify_common(struct lov_mds_md *lmm, int lmm_bytes,
  66                                  int stripe_count)
  67 {
  68
  69         if (stripe_count == 0 || stripe_count > LOV_V1_INSANE_STRIPE_COUNT) {
  70                 CERROR("bad stripe count %d\n", stripe_count);
  71                 lov_dump_lmm_v1(D_WARNING, lmm);
  72                 return -EINVAL;
  73         }
  74
  75         if (lmm->lmm_object_id == 0) {
  76                 CERROR("zero object id\n");
  77                 lov_dump_lmm_v1(D_WARNING, lmm);
  78                 return -EINVAL;
  79         }
  80
  81         if (lmm->lmm_pattern != cpu_to_le32(LOV_PATTERN_RAID0)) {
  82                 CERROR("bad striping pattern\n");
  83                 lov_dump_lmm_v1(D_WARNING, lmm);
  84                 return -EINVAL;
  85         }
  86
  87         if (lmm->lmm_stripe_size == 0 ||
  88             (stripe_count != -1 &&
  89              (__u64)le32_to_cpu(lmm->lmm_stripe_size)*stripe_count >
  90              0xffffffff)) {
  91                 CERROR("bad stripe size %u\n",
  92                        le32_to_cpu(lmm->lmm_stripe_size));
  93                 lov_dump_lmm_v1(D_WARNING, lmm);
  94                 return -EINVAL;
  95         }
  96         return 0;
  97 }
  98
  99 struct lov_stripe_md *lsm_alloc_plain(int stripe_count, int *size)
 100 {
 101         struct lov_stripe_md *lsm;
 102         int i, oinfo_ptrs_size;
 103         struct lov_oinfo *loi;
 104
 105         LASSERT(stripe_count > 0);
 106
 107         oinfo_ptrs_size = sizeof(struct lov_oinfo *) * stripe_count;
 108         *size = sizeof(struct lov_stripe_md) + oinfo_ptrs_size;
 109
 110         OBD_ALLOC(lsm, *size);
 111         if (!lsm)
 112                 return NULL;;
 113
 114         for (i = 0; i < stripe_count; i++) {
 115                 OBD_SLAB_ALLOC(loi, lov_oinfo_slab, CFS_ALLOC_IO, sizeof(*loi));
 116                 if (loi == NULL)
 117                         goto err;
 118                 lsm->lsm_oinfo[i] = loi;
 119         }
 120         lsm->lsm_stripe_count = stripe_count;
 121         return lsm;
 122
 123 err:
 124         while (--i >= 0)
 125                 OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab, sizeof(*loi));
 126         OBD_FREE(lsm, *size);
 127         return NULL;
 128 }
 129
 130 void lsm_free_plain(struct lov_stripe_md *lsm)
 131 {
 132         int stripe_count = lsm->lsm_stripe_count;
 133         int i;
 134
 135         for (i = 0; i < stripe_count; i++)
 136                 OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab,
 137                               sizeof(struct lov_oinfo));
 138         OBD_FREE(lsm, sizeof(struct lov_stripe_md) +
 139                  stripe_count * sizeof(struct lov_oinfo *));
 140 }
 141
 142 static void lsm_unpackmd_common(struct lov_stripe_md *lsm,
 143                                 struct lov_mds_md *lmm)
 144 {
 145         lsm->lsm_object_id = le64_to_cpu(lmm->lmm_object_id);
 146         lsm->lsm_object_gr = le64_to_cpu(lmm->lmm_object_gr);
 147         lsm->lsm_stripe_size = le32_to_cpu(lmm->lmm_stripe_size);
 148         lsm->lsm_pattern = le32_to_cpu(lmm->lmm_pattern);
 149 }
 150
 151 static void
 152 lsm_stripe_by_index_plain(struct lov_stripe_md *lsm, int *stripeno,
 153                            obd_off *lov_off, unsigned long *swidth)
 154 {
 155         if (swidth)
 156                 *swidth = (unsigned long)lsm->lsm_stripe_size * lsm->lsm_stripe_count;
 157 }
 158
 159 static void
 160 lsm_stripe_by_offset_plain(struct lov_stripe_md *lsm, int *stripeno,
 161                            obd_off *lov_off, unsigned long *swidth)
 162 {
 163         if (swidth)
 164                 *swidth = (unsigned long)lsm->lsm_stripe_size * lsm->lsm_stripe_count;
 165 }
 166
 167 static obd_off
 168 lsm_stripe_offset_by_index_plain(struct lov_stripe_md *lsm,
 169                                   int stripe_index)
 170 {
 171         return 0;
 172 }
 173
 174 static obd_off
 175 lsm_stripe_offset_by_offset_plain(struct lov_stripe_md *lsm,
 176                                   obd_off lov_off)
 177 {
 178         return 0;
 179 }
 180
 181 static int
 182 lsm_stripe_index_by_offset_plain(struct lov_stripe_md *lsm,
 183                                   obd_off lov_off)
 184 {
 185         return 0;
 186 }
 187
 188 static int lsm_revalidate_plain(struct lov_stripe_md *lsm,
 189                                 struct obd_device *obd)
 190 {
 191         return 0;
 192 }
 193
 194 static int lsm_destroy_plain(struct lov_stripe_md *lsm, struct obdo *oa,
 195                              struct obd_export *md_exp)
 196 {
 197         return 0;
 198 }
 199
 200 static int lsm_lmm_verify_plain(struct lov_mds_md *lmm, int lmm_bytes,
 201                              int *stripe_count)
 202 {
 203         if (lmm_bytes < sizeof(*lmm)) {
 204                 CERROR("lov_mds_md too small: %d, need at least %d\n",
 205                        lmm_bytes, (int)sizeof(*lmm));
 206                 return -EINVAL;
 207         }
 208
 209         *stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
 210
 211         if (lmm_bytes < lov_mds_md_v1_size(*stripe_count)) {
 212                 CERROR("LOV EA too small: %d, need %d\n",
 213                        lmm_bytes, lov_mds_md_v1_size(*stripe_count));
 214                 lov_dump_lmm_v1(D_WARNING, lmm);
 215                 return -EINVAL;
 216         }
 217
 218         return lsm_lmm_verify_common(lmm, lmm_bytes, *stripe_count);
 219 }
 220
 221 int lsm_unpackmd_plain(struct lov_obd *lov, struct lov_stripe_md *lsm,
 222                     struct lov_mds_md_v1 *lmm)
 223 {
 224         struct lov_oinfo *loi;
 225         int i;
 226
 227         lsm_unpackmd_common(lsm, lmm);
 228
 229         for (i = 0; i < lsm->lsm_stripe_count; i++) {
 230                 /* XXX LOV STACKING call down to osc_unpackmd() */
 231                 loi = lsm->lsm_oinfo[i];
 232                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 233                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 234                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 235                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 236                 if (loi->loi_ost_idx >= lov->desc.ld_tgt_count) {
 237                         CERROR("OST index %d more than OST count %d\n",
 238                                loi->loi_ost_idx, lov->desc.ld_tgt_count);
 239                         lov_dump_lmm_v1(D_WARNING, lmm);
 240                         return -EINVAL;
 241                 }
 242                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
 243                         CERROR("OST index %d missing\n", loi->loi_ost_idx);
 244                         lov_dump_lmm_v1(D_WARNING, lmm);
 245                         return -EINVAL;
 246                 }
 247         }
 248
 249         return 0;
 250 }
 251
 252 struct lsm_operations lsm_plain_ops = {
 253         .lsm_free            = lsm_free_plain,
 254         .lsm_destroy         = lsm_destroy_plain,
 255         .lsm_stripe_by_index    = lsm_stripe_by_index_plain,
 256         .lsm_stripe_by_offset   = lsm_stripe_by_offset_plain,
 257         .lsm_revalidate         = lsm_revalidate_plain,
 258         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_plain,
 259         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_plain,
 260         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_plain,
 261         .lsm_lmm_verify         = lsm_lmm_verify_plain,
 262         .lsm_unpackmd           = lsm_unpackmd_plain,
 263 };
 264
 265 struct lov_extent *lovea_off2le(struct lov_stripe_md *lsm, obd_off lov_off)
 266 {
 267         struct lov_array_info *lai;
 268         struct lov_extent *le;
 269         int i = 0;
 270
 271         LASSERT(lsm->lsm_array != NULL);
 272         lai = lsm->lsm_array;
 273         LASSERT(lai->lai_ext_count > 1);
 274
 275         for (le = lai->lai_ext_array, i = 0;
 276              i < lai->lai_ext_count && le->le_start + le->le_len <= lov_off
 277              && le->le_len != -1;
 278              i ++, le ++) {
 279                ; /* empty loop */
 280         }
 281
 282         CDEBUG(D_INFO, "off "LPU64" idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
 283                lov_off, i, le->le_start, le->le_len, le->le_loi_idx,
 284                le->le_stripe_count);
 285
 286         RETURN(le);
 287 }
 288
 289 struct lov_extent *lovea_idx2le(struct lov_stripe_md *lsm, int stripe_no)
 290 {
 291         struct lov_extent *le;
 292         struct lov_array_info *lai;
 293         int i, stripe_index;
 294
 295         LASSERT(lsm->lsm_array != NULL);
 296         LASSERT(stripe_no >= 0 && stripe_no <= lsm->lsm_stripe_count);
 297         lai = lsm->lsm_array;
 298         LASSERT(lai->lai_ext_count > 1);
 299
 300         for (le = lai->lai_ext_array, i = 0, stripe_index = le->le_stripe_count;
 301              i < lai->lai_ext_count && stripe_index <= stripe_no &&
 302              le->le_len != -1; i ++, le ++,
 303              stripe_index += le->le_stripe_count) {
 304                 ; /* empty loop */
 305         }
 306
 307         CDEBUG(D_INFO, "stripe %d idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
 308                stripe_no, i, le->le_start, le->le_len, le->le_loi_idx,
 309                le->le_stripe_count);
 310         RETURN(le);
 311 }
 312
 313 static void lovea_free_array_info(struct lov_stripe_md *lsm)
 314 {
 315         if (!lsm || !lsm->lsm_array)
 316                 return;
 317
 318         if (lsm->lsm_array->lai_ext_array)
 319                 OBD_FREE(lsm->lsm_array->lai_ext_array,
 320                          lsm->lsm_array->lai_ext_count *
 321                          sizeof(struct lov_extent));
 322
 323         OBD_FREE_PTR(lsm->lsm_array);
 324 }
 325
 326 static void lsm_free_join(struct lov_stripe_md *lsm)
 327 {
 328         lovea_free_array_info(lsm);
 329         lsm_free_plain(lsm);
 330 }
 331
 332 static void
 333 lsm_stripe_by_index_join(struct lov_stripe_md *lsm, int *stripeno,
 334                            obd_off *lov_off, unsigned long *swidth)
 335 {
 336         struct lov_extent *le;
 337
 338         LASSERT(stripeno != NULL);
 339
 340         le = lovea_idx2le(lsm, *stripeno);
 341
 342         LASSERT(le != NULL && le->le_stripe_count != 0);
 343
 344         *stripeno -= le->le_loi_idx;
 345
 346         if (swidth)
 347                 *swidth = (unsigned long)lsm->lsm_stripe_size * le->le_stripe_count;
 348
 349         if (lov_off) {
 350                 struct lov_extent *lov_le = lovea_off2le(lsm, *lov_off);
 351                 if (lov_le == le) {
 352                         *lov_off = (*lov_off > le->le_start) ?
 353                                    (*lov_off - le->le_start) : 0;
 354                 } else {
 355                         *lov_off = (*lov_off > le->le_start) ?
 356                                    le->le_len : 0;
 357                         LASSERT(*lov_off != -1);
 358                 }
 359         }
 360 }
 361
 362 static void
 363 lsm_stripe_by_offset_join(struct lov_stripe_md *lsm, int *stripeno,
 364                            obd_off *lov_off, unsigned long *swidth)
 365 {
 366         struct lov_extent *le;
 367
 368         LASSERT(lov_off != NULL);
 369
 370         le = lovea_off2le(lsm, *lov_off);
 371
 372         LASSERT(le != NULL && le->le_stripe_count != 0);
 373
 374         *lov_off = (*lov_off > le->le_start) ? (*lov_off - le->le_start) : 0;
 375
 376         if (stripeno)
 377                 *stripeno -= le->le_loi_idx;
 378
 379         if (swidth)
 380                 *swidth = (unsigned long)lsm->lsm_stripe_size * le->le_stripe_count;
 381 }
 382
 383 static obd_off
 384 lsm_stripe_offset_by_index_join(struct lov_stripe_md *lsm,
 385                                  int stripe_index)
 386 {
 387         struct lov_extent *le;
 388
 389         le = lovea_idx2le(lsm, stripe_index);
 390
 391         return le ? le->le_start : 0;
 392 }
 393
 394 static obd_off
 395 lsm_stripe_offset_by_offset_join(struct lov_stripe_md *lsm,
 396                                  obd_off lov_off)
 397 {
 398         struct lov_extent *le;
 399
 400         le = lovea_off2le(lsm, lov_off);
 401
 402         return le ? le->le_start : 0;
 403 }
 404
 405 static int
 406 lsm_stripe_index_by_offset_join(struct lov_stripe_md *lsm,
 407                                  obd_off lov_off)
 408 {
 409         struct lov_extent *le = NULL;
 410
 411         le = lovea_off2le(lsm, lov_off);
 412
 413         return le ? le->le_loi_idx : 0;
 414 }
 415
 416 static int lovea_unpack_array(struct llog_handle *handle,
 417                               struct llog_rec_hdr *rec, void *data)
 418 {
 419         struct lovea_unpack_args *args = (struct lovea_unpack_args *)data;
 420         struct llog_array_rec *la_rec = (struct llog_array_rec*)rec;
 421         struct mds_extent_desc *med = &la_rec->lmr_med;
 422         struct lov_stripe_md *lsm = args->lsm;
 423         int cursor = args->cursor++;
 424         struct lov_mds_md *lmm;
 425         struct lov_array_info *lai;
 426         struct lov_oinfo * loi;
 427         int i, loi_index;
 428         ENTRY;
 429
 430         /* sanity check */
 431         LASSERT(lsm->lsm_stripe_count != 0);
 432         lmm = &med->med_lmm;
 433         LASSERT(lsm->lsm_array != NULL);
 434
 435         lai = lsm->lsm_array;
 436
 437         if (cursor == 0) {
 438                lai->lai_ext_array[cursor].le_loi_idx = 0;
 439         } else {
 440                int next_loi_index = lai->lai_ext_array[cursor - 1].le_loi_idx +
 441                                  lai->lai_ext_array[cursor - 1].le_stripe_count;
 442                lai->lai_ext_array[cursor].le_loi_idx = next_loi_index;
 443         }
 444         /* insert extent desc into lsm extent array  */
 445         lai->lai_ext_array[cursor].le_start = le64_to_cpu(med->med_start);
 446         lai->lai_ext_array[cursor].le_len   = le64_to_cpu(med->med_len);
 447         lai->lai_ext_array[cursor].le_stripe_count = lmm->lmm_stripe_count;
 448
 449         /* unpack extent's lmm to lov_oinfo array */
 450         loi_index = lai->lai_ext_array[cursor].le_loi_idx;
 451         CDEBUG(D_INFO, "lovea upackmd cursor %d, loi_index %d extent "
 452                         LPU64":"LPU64"\n", cursor, loi_index, med->med_start,
 453                         med->med_len);
 454
 455         for (i = 0; i < le32_to_cpu(lmm->lmm_stripe_count); i ++, loi_index++) {
 456                 /* XXX LOV STACKING call down to osc_unpackmd() */
 457                 loi = lsm->lsm_oinfo[loi_index];
 458                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 459                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 460                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 461                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 462         }
 463
 464         RETURN(0);
 465 }
 466
 467 static int lsm_revalidate_join(struct lov_stripe_md *lsm,
 468                                struct obd_device *obd)
 469 {
 470         struct llog_handle *llh;
 471         struct llog_ctxt *ctxt;
 472         struct lovea_unpack_args args;
 473         int rc, rc2;
 474         ENTRY;
 475
 476         LASSERT(lsm->lsm_array != NULL);
 477
 478         /*Revalidate lsm might be called from client or MDS server.
 479          *So the ctxt might be in different position
 480          */
 481         ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT);
 482         if (!ctxt)
 483                 ctxt = llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT);
 484
 485         LASSERT(ctxt);
 486
 487         if (lsm->lsm_array && lsm->lsm_array->lai_ext_array)
 488                 GOTO(release_ctxt, rc = 0);
 489
 490         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 491                lsm->lsm_array->lai_array_id.lgl_oid,
 492                lsm->lsm_array->lai_array_id.lgl_ogr);
 493         OBD_ALLOC(lsm->lsm_array->lai_ext_array,lsm->lsm_array->lai_ext_count *
 494                                                 sizeof (struct lov_extent));
 495         if (!lsm->lsm_array->lai_ext_array)
 496                 GOTO(release_ctxt, rc = -ENOMEM);
 497
 498         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 499                lsm->lsm_array->lai_array_id.lgl_oid,
 500                lsm->lsm_array->lai_array_id.lgl_ogr);
 501
 502         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id, NULL);
 503         if (rc)
 504                 GOTO(out, rc);
 505
 506         args.lsm = lsm;
 507         args.cursor = 0;
 508         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 509         if (rc == 0)
 510                 rc = llog_process(llh, lovea_unpack_array, &args, NULL);
 511         rc2 = llog_close(llh);
 512         if (rc == 0)
 513                 rc = rc2;
 514 out:
 515         if (rc)
 516                 lovea_free_array_info(lsm);
 517 release_ctxt:
 518         llog_ctxt_put(ctxt);
 519         RETURN(rc);
 520 }
 521
 522 int lsm_destroy_join(struct lov_stripe_md *lsm, struct obdo *oa,
 523                       struct obd_export *md_exp)
 524 {
 525         struct llog_ctxt *ctxt;
 526         struct llog_handle *llh;
 527         int rc = 0;
 528         ENTRY;
 529
 530         LASSERT(md_exp != NULL);
 531         /*for those orphan inode, we should keep array id*/
 532         if (!(oa->o_valid & OBD_MD_FLCOOKIE))
 533                 RETURN(rc);
 534
 535         ctxt = llog_get_context(md_exp->exp_obd, LLOG_LOVEA_REPL_CTXT);
 536         if (!ctxt)
 537                 RETURN(-EINVAL);
 538
 539         LASSERT(lsm->lsm_array != NULL);
 540         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id,
 541                          NULL);
 542         if (rc)
 543                 GOTO(out, rc);
 544
 545         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 546         if (rc == 0) {
 547                 rc = llog_destroy(llh);
 548         }
 549         llog_free_handle(llh);
 550 out:
 551         llog_ctxt_put(ctxt);
 552         RETURN(rc);
 553 }
 554
 555 static int lsm_lmm_verify_join(struct lov_mds_md *lmm, int lmm_bytes,
 556                                int *stripe_count)
 557 {
 558         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join *)lmm;
 559
 560         if (lmm_bytes < sizeof(*lmmj)) {
 561                 CERROR("lov_mds_md too small: %d, need at least %d\n",
 562                        lmm_bytes, (int)sizeof(*lmmj));
 563                 return -EINVAL;
 564         }
 565
 566         if (lmmj->lmmj_array_id.lgl_oid == 0) {
 567                 CERROR("zero array object id\n");
 568                 return -EINVAL;
 569         }
 570
 571         *stripe_count = le32_to_cpu(lmmj->lmmj_md.lmm_stripe_count);
 572
 573         return lsm_lmm_verify_common(&lmmj->lmmj_md, lmm_bytes, *stripe_count);
 574 }
 575
 576 static int lovea_init_array_info(struct lov_stripe_md *lsm,
 577                                  struct llog_logid *logid,
 578                                  __u32 extent_count)
 579 {
 580         struct lov_array_info *lai;
 581         ENTRY;
 582
 583         OBD_ALLOC_PTR(lai);
 584         if (!lai)
 585                 RETURN(-ENOMEM);
 586
 587         lai->lai_array_id = *logid;
 588         lai->lai_ext_count = extent_count;
 589         lsm->lsm_array = lai;
 590         RETURN(0);
 591 }
 592
 593 static int lsm_unpackmd_join(struct lov_obd *lov, struct lov_stripe_md *lsm,
 594                       struct lov_mds_md *lmm)
 595 {
 596         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join*)lmm;
 597         int    rc;
 598         ENTRY;
 599
 600         lsm_unpackmd_common(lsm, &lmmj->lmmj_md);
 601
 602         rc = lovea_init_array_info(lsm, &lmmj->lmmj_array_id,
 603                                    lmmj->lmmj_extent_count);
 604         if (rc) {
 605                 CERROR("Init joined lsm id"LPU64" arrary error %d",
 606                         lsm->lsm_object_id, rc);
 607                 GOTO(out, rc);
 608         }
 609 out:
 610         RETURN(rc);
 611 }
 612
 613 struct lsm_operations lsm_join_ops = {
 614         .lsm_free             = lsm_free_join,
 615         .lsm_destroy          = lsm_destroy_join,
 616         .lsm_stripe_by_index  = lsm_stripe_by_index_join,
 617         .lsm_stripe_by_offset = lsm_stripe_by_offset_join,
 618         .lsm_revalidate       = lsm_revalidate_join,
 619         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_join,
 620         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_join,
 621         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_join,
 622         .lsm_lmm_verify         = lsm_lmm_verify_join,
 623         .lsm_unpackmd           = lsm_unpackmd_join,
 624 };