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(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(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(D_WARNING, lmm);
  84                 return -EINVAL;
  85         }
  86
  87         if (lmm->lmm_stripe_size == 0 ||
  88              (le32_to_cpu(lmm->lmm_stripe_size)&(LOV_MIN_STRIPE_SIZE-1)) != 0) {
  89                 CERROR("bad stripe size %u\n",
  90                        le32_to_cpu(lmm->lmm_stripe_size));
  91                 lov_dump_lmm(D_WARNING, lmm);
  92                 return -EINVAL;
  93         }
  94         return 0;
  95 }
  96
  97 struct lov_stripe_md *lsm_alloc_plain(int stripe_count, int *size)
  98 {
  99         struct lov_stripe_md *lsm;
 100         int i, oinfo_ptrs_size;
 101         struct lov_oinfo *loi;
 102
 103         LASSERT(stripe_count > 0);
 104
 105         oinfo_ptrs_size = sizeof(struct lov_oinfo *) * stripe_count;
 106         *size = sizeof(struct lov_stripe_md) + oinfo_ptrs_size;
 107
 108         OBD_ALLOC(lsm, *size);
 109         if (!lsm)
 110                 return NULL;;
 111
 112         for (i = 0; i < stripe_count; i++) {
 113                 OBD_SLAB_ALLOC_PTR_GFP(loi, lov_oinfo_slab, CFS_ALLOC_IO);
 114                 if (loi == NULL)
 115                         goto err;
 116                 lsm->lsm_oinfo[i] = loi;
 117         }
 118         lsm->lsm_stripe_count = stripe_count;
 119         lsm->lsm_pool_name[0] = '\0';
 120         return lsm;
 121
 122 err:
 123         while (--i >= 0)
 124                 OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab, sizeof(*loi));
 125         OBD_FREE(lsm, *size);
 126         return NULL;
 127 }
 128
 129 void lsm_free_plain(struct lov_stripe_md *lsm)
 130 {
 131         int stripe_count = lsm->lsm_stripe_count;
 132         int i;
 133
 134         for (i = 0; i < stripe_count; i++)
 135                 OBD_SLAB_FREE(lsm->lsm_oinfo[i], lov_oinfo_slab,
 136                               sizeof(struct lov_oinfo));
 137         OBD_FREE(lsm, sizeof(struct lov_stripe_md) +
 138                  stripe_count * sizeof(struct lov_oinfo *));
 139 }
 140
 141 static void lsm_unpackmd_common(struct lov_stripe_md *lsm,
 142                                 struct lov_mds_md *lmm)
 143 {
 144         /*
 145          * This supposes lov_mds_md_v1/v3 first fields are
 146          * are the same
 147          */
 148         lsm->lsm_object_id = le64_to_cpu(lmm->lmm_object_id);
 149         lsm->lsm_object_gr = le64_to_cpu(lmm->lmm_object_gr);
 150         lsm->lsm_stripe_size = le32_to_cpu(lmm->lmm_stripe_size);
 151         lsm->lsm_pattern = le32_to_cpu(lmm->lmm_pattern);
 152         lsm->lsm_pool_name[0] = '\0';
 153 }
 154
 155 static void
 156 lsm_stripe_by_index_plain(struct lov_stripe_md *lsm, int *stripeno,
 157                            obd_off *lov_off, obd_off *swidth)
 158 {
 159         if (swidth)
 160                 *swidth = (obd_off)lsm->lsm_stripe_size * lsm->lsm_stripe_count;
 161 }
 162
 163 static void
 164 lsm_stripe_by_offset_plain(struct lov_stripe_md *lsm, int *stripeno,
 165                            obd_off *lov_off, obd_off *swidth)
 166 {
 167         if (swidth)
 168                 *swidth = (obd_off)lsm->lsm_stripe_size * lsm->lsm_stripe_count;
 169 }
 170
 171 static obd_off
 172 lsm_stripe_offset_by_index_plain(struct lov_stripe_md *lsm,
 173                                   int stripe_index)
 174 {
 175         return 0;
 176 }
 177
 178 static obd_off
 179 lsm_stripe_offset_by_offset_plain(struct lov_stripe_md *lsm,
 180                                   obd_off lov_off)
 181 {
 182         return 0;
 183 }
 184
 185 static int
 186 lsm_stripe_index_by_offset_plain(struct lov_stripe_md *lsm,
 187                                   obd_off lov_off)
 188 {
 189         return 0;
 190 }
 191
 192 static int lsm_revalidate_plain(struct lov_stripe_md *lsm,
 193                                 struct obd_device *obd)
 194 {
 195         return 0;
 196 }
 197
 198 static int lsm_destroy_plain(struct lov_stripe_md *lsm, struct obdo *oa,
 199                              struct obd_export *md_exp)
 200 {
 201         return 0;
 202 }
 203
 204 static int lsm_lmm_verify_v1(struct lov_mds_md_v1 *lmm, int lmm_bytes,
 205                              int *stripe_count)
 206 {
 207         if (lmm_bytes < sizeof(*lmm)) {
 208                 CERROR("lov_mds_md_v1 too small: %d, need at least %d\n",
 209                        lmm_bytes, (int)sizeof(*lmm));
 210                 return -EINVAL;
 211         }
 212
 213         *stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
 214
 215         if (lmm_bytes < lov_mds_md_size(*stripe_count, LOV_MAGIC_V1)) {
 216                 CERROR("LOV EA V1 too small: %d, need %d\n",
 217                        lmm_bytes, lov_mds_md_size(*stripe_count, LOV_MAGIC_V1));
 218                 lov_dump_lmm_v1(D_WARNING, lmm);
 219                 return -EINVAL;
 220         }
 221
 222         return lsm_lmm_verify_common(lmm, lmm_bytes, *stripe_count);
 223 }
 224
 225 int lsm_unpackmd_v1(struct lov_obd *lov, struct lov_stripe_md *lsm,
 226                     struct lov_mds_md_v1 *lmm)
 227 {
 228         struct lov_oinfo *loi;
 229         int i;
 230
 231         lsm_unpackmd_common(lsm, lmm);
 232
 233         for (i = 0; i < lsm->lsm_stripe_count; i++) {
 234                 /* XXX LOV STACKING call down to osc_unpackmd() */
 235                 loi = lsm->lsm_oinfo[i];
 236                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 237                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 238                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 239                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 240                 if (loi->loi_ost_idx >= lov->desc.ld_tgt_count) {
 241                         CERROR("OST index %d more than OST count %d\n",
 242                                loi->loi_ost_idx, lov->desc.ld_tgt_count);
 243                         lov_dump_lmm_v1(D_WARNING, lmm);
 244                         return -EINVAL;
 245                 }
 246                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
 247                         CERROR("OST index %d missing\n", loi->loi_ost_idx);
 248                         lov_dump_lmm_v1(D_WARNING, lmm);
 249                         return -EINVAL;
 250                 }
 251         }
 252
 253         return 0;
 254 }
 255
 256 const struct lsm_operations lsm_v1_ops = {
 257         .lsm_free            = lsm_free_plain,
 258         .lsm_destroy         = lsm_destroy_plain,
 259         .lsm_stripe_by_index    = lsm_stripe_by_index_plain,
 260         .lsm_stripe_by_offset   = lsm_stripe_by_offset_plain,
 261         .lsm_revalidate         = lsm_revalidate_plain,
 262         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_plain,
 263         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_plain,
 264         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_plain,
 265         .lsm_lmm_verify         = lsm_lmm_verify_v1,
 266         .lsm_unpackmd           = lsm_unpackmd_v1,
 267 };
 268
 269 struct lov_extent *lovea_off2le(struct lov_stripe_md *lsm, obd_off lov_off)
 270 {
 271         struct lov_array_info *lai;
 272         struct lov_extent *le;
 273         int i = 0;
 274
 275         LASSERT(lsm->lsm_array != NULL);
 276         lai = lsm->lsm_array;
 277         LASSERT(lai->lai_ext_count > 1);
 278
 279         for (le = lai->lai_ext_array, i = 0;
 280              i < lai->lai_ext_count && le->le_start + le->le_len <= lov_off
 281              && le->le_len != -1;
 282              i ++, le ++) {
 283                ; /* empty loop */
 284         }
 285
 286         CDEBUG(D_INFO, "off "LPU64" idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
 287                lov_off, i, le->le_start, le->le_len, le->le_loi_idx,
 288                le->le_stripe_count);
 289
 290         RETURN(le);
 291 }
 292
 293 struct lov_extent *lovea_idx2le(struct lov_stripe_md *lsm, int stripe_no)
 294 {
 295         struct lov_extent *le;
 296         struct lov_array_info *lai;
 297         int i, stripe_index;
 298
 299         LASSERT(lsm->lsm_array != NULL);
 300         LASSERT(stripe_no >= 0 && stripe_no <= lsm->lsm_stripe_count);
 301         lai = lsm->lsm_array;
 302         LASSERT(lai->lai_ext_count > 1);
 303
 304         for (le = lai->lai_ext_array, i = 0, stripe_index = le->le_stripe_count;
 305              i < lai->lai_ext_count && stripe_index <= stripe_no &&
 306              le->le_len != -1; i ++, le ++,
 307              stripe_index += le->le_stripe_count) {
 308                 ; /* empty loop */
 309         }
 310
 311         CDEBUG(D_INFO, "stripe %d idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
 312                stripe_no, i, le->le_start, le->le_len, le->le_loi_idx,
 313                le->le_stripe_count);
 314         RETURN(le);
 315 }
 316
 317 static void lovea_free_array_info(struct lov_stripe_md *lsm)
 318 {
 319         if (!lsm || !lsm->lsm_array)
 320                 return;
 321
 322         if (lsm->lsm_array->lai_ext_array)
 323                 OBD_FREE(lsm->lsm_array->lai_ext_array,
 324                          lsm->lsm_array->lai_ext_count *
 325                          sizeof(struct lov_extent));
 326
 327         OBD_FREE_PTR(lsm->lsm_array);
 328 }
 329
 330 static void lsm_free_join(struct lov_stripe_md *lsm)
 331 {
 332         lovea_free_array_info(lsm);
 333         lsm_free_plain(lsm);
 334 }
 335
 336 static void
 337 lsm_stripe_by_index_join(struct lov_stripe_md *lsm, int *stripeno,
 338                            obd_off *lov_off, obd_off *swidth)
 339 {
 340         struct lov_extent *le;
 341
 342         LASSERT(stripeno != NULL);
 343
 344         le = lovea_idx2le(lsm, *stripeno);
 345
 346         LASSERT(le != NULL && le->le_stripe_count != 0);
 347
 348         *stripeno -= le->le_loi_idx;
 349
 350         if (swidth)
 351                 *swidth = (obd_off)lsm->lsm_stripe_size * le->le_stripe_count;
 352
 353         if (lov_off) {
 354                 struct lov_extent *lov_le = lovea_off2le(lsm, *lov_off);
 355                 if (lov_le == le) {
 356                         *lov_off = (*lov_off > le->le_start) ?
 357                                    (*lov_off - le->le_start) : 0;
 358                 } else {
 359                         *lov_off = (*lov_off > le->le_start) ?
 360                                    le->le_len : 0;
 361                         LASSERT(*lov_off != -1);
 362                 }
 363         }
 364 }
 365
 366 static void
 367 lsm_stripe_by_offset_join(struct lov_stripe_md *lsm, int *stripeno,
 368                            obd_off *lov_off, obd_off *swidth)
 369 {
 370         struct lov_extent *le;
 371
 372         LASSERT(lov_off != NULL);
 373
 374         le = lovea_off2le(lsm, *lov_off);
 375
 376         LASSERT(le != NULL && le->le_stripe_count != 0);
 377
 378         *lov_off = (*lov_off > le->le_start) ? (*lov_off - le->le_start) : 0;
 379
 380         if (stripeno)
 381                 *stripeno -= le->le_loi_idx;
 382
 383         if (swidth)
 384                 *swidth = (obd_off)lsm->lsm_stripe_size * le->le_stripe_count;
 385 }
 386
 387 static obd_off
 388 lsm_stripe_offset_by_index_join(struct lov_stripe_md *lsm,
 389                                  int stripe_index)
 390 {
 391         struct lov_extent *le;
 392
 393         le = lovea_idx2le(lsm, stripe_index);
 394
 395         return le ? le->le_start : 0;
 396 }
 397
 398 static obd_off
 399 lsm_stripe_offset_by_offset_join(struct lov_stripe_md *lsm,
 400                                  obd_off lov_off)
 401 {
 402         struct lov_extent *le;
 403
 404         le = lovea_off2le(lsm, lov_off);
 405
 406         return le ? le->le_start : 0;
 407 }
 408
 409 static int
 410 lsm_stripe_index_by_offset_join(struct lov_stripe_md *lsm,
 411                                  obd_off lov_off)
 412 {
 413         struct lov_extent *le = NULL;
 414
 415         le = lovea_off2le(lsm, lov_off);
 416
 417         return le ? le->le_loi_idx : 0;
 418 }
 419
 420 static int lovea_unpack_array(struct llog_handle *handle,
 421                               struct llog_rec_hdr *rec, void *data)
 422 {
 423         struct lovea_unpack_args *args = (struct lovea_unpack_args *)data;
 424         struct llog_array_rec *la_rec = (struct llog_array_rec*)rec;
 425         struct mds_extent_desc *med = &la_rec->lmr_med;
 426         struct lov_stripe_md *lsm = args->lsm;
 427         int cursor = args->cursor++;
 428         struct lov_mds_md *lmm;
 429         struct lov_array_info *lai;
 430         struct lov_oinfo * loi;
 431         int i, loi_index;
 432         ENTRY;
 433
 434         /* sanity check */
 435         LASSERT(lsm->lsm_stripe_count != 0);
 436         lmm = &med->med_lmm;
 437         LASSERT(lsm->lsm_array != NULL);
 438
 439         lai = lsm->lsm_array;
 440
 441         if (cursor == 0) {
 442                lai->lai_ext_array[cursor].le_loi_idx = 0;
 443         } else {
 444                int next_loi_index = lai->lai_ext_array[cursor - 1].le_loi_idx +
 445                                  lai->lai_ext_array[cursor - 1].le_stripe_count;
 446                lai->lai_ext_array[cursor].le_loi_idx = next_loi_index;
 447         }
 448         /* insert extent desc into lsm extent array  */
 449         lai->lai_ext_array[cursor].le_start = le64_to_cpu(med->med_start);
 450         lai->lai_ext_array[cursor].le_len   = le64_to_cpu(med->med_len);
 451         lai->lai_ext_array[cursor].le_stripe_count =
 452                                    le32_to_cpu(lmm->lmm_stripe_count);
 453
 454         /* unpack extent's lmm to lov_oinfo array */
 455         loi_index = lai->lai_ext_array[cursor].le_loi_idx;
 456         CDEBUG(D_INFO, "lovea upackmd cursor %d, loi_index %d extent "
 457                         LPU64":"LPU64"\n", cursor, loi_index, med->med_start,
 458                         med->med_len);
 459
 460         for (i = 0; i < le32_to_cpu(lmm->lmm_stripe_count); i ++, loi_index++) {
 461                 /* XXX LOV STACKING call down to osc_unpackmd() */
 462                 loi = lsm->lsm_oinfo[loi_index];
 463                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 464                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 465                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 466                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 467         }
 468
 469         RETURN(0);
 470 }
 471
 472 static int lsm_revalidate_join(struct lov_stripe_md *lsm,
 473                                struct obd_device *obd)
 474 {
 475         struct llog_handle *llh;
 476         struct llog_ctxt *ctxt;
 477         struct lovea_unpack_args args;
 478         int rc, rc2;
 479         ENTRY;
 480
 481         LASSERT(lsm->lsm_array != NULL);
 482
 483         /*Revalidate lsm might be called from client or MDS server.
 484          *So the ctxt might be in different position
 485          */
 486         ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT);
 487         if (!ctxt)
 488                 ctxt = llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT);
 489
 490         LASSERT(ctxt);
 491
 492         if (lsm->lsm_array && lsm->lsm_array->lai_ext_array)
 493                 GOTO(release_ctxt, rc = 0);
 494
 495         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 496                lsm->lsm_array->lai_array_id.lgl_oid,
 497                lsm->lsm_array->lai_array_id.lgl_ogr);
 498         OBD_ALLOC(lsm->lsm_array->lai_ext_array,lsm->lsm_array->lai_ext_count *
 499                                                 sizeof (struct lov_extent));
 500         if (!lsm->lsm_array->lai_ext_array)
 501                 GOTO(release_ctxt, rc = -ENOMEM);
 502
 503         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 504                lsm->lsm_array->lai_array_id.lgl_oid,
 505                lsm->lsm_array->lai_array_id.lgl_ogr);
 506
 507         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id, NULL);
 508         if (rc)
 509                 GOTO(out, rc);
 510
 511         args.lsm = lsm;
 512         args.cursor = 0;
 513         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 514         if (rc == 0)
 515                 rc = llog_process(llh, lovea_unpack_array, &args, NULL);
 516         rc2 = llog_close(llh);
 517         if (rc == 0)
 518                 rc = rc2;
 519 out:
 520         if (rc)
 521                 lovea_free_array_info(lsm);
 522 release_ctxt:
 523         llog_ctxt_put(ctxt);
 524         RETURN(rc);
 525 }
 526
 527 int lsm_destroy_join(struct lov_stripe_md *lsm, struct obdo *oa,
 528                       struct obd_export *md_exp)
 529 {
 530         struct llog_ctxt *ctxt;
 531         struct llog_handle *llh;
 532         int rc = 0;
 533         ENTRY;
 534
 535         LASSERT(md_exp != NULL);
 536         /*for those orphan inode, we should keep array id*/
 537         if (!(oa->o_valid & OBD_MD_FLCOOKIE))
 538                 RETURN(rc);
 539
 540         ctxt = llog_get_context(md_exp->exp_obd, LLOG_LOVEA_REPL_CTXT);
 541         if (!ctxt)
 542                 RETURN(-EINVAL);
 543
 544         LASSERT(lsm->lsm_array != NULL);
 545         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id,
 546                          NULL);
 547         if (rc)
 548                 GOTO(out, rc);
 549
 550         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 551         if (rc == 0) {
 552                 rc = llog_destroy(llh);
 553         }
 554         llog_free_handle(llh);
 555 out:
 556         llog_ctxt_put(ctxt);
 557         RETURN(rc);
 558 }
 559
 560 static int lsm_lmm_verify_join(struct lov_mds_md *lmm, int lmm_bytes,
 561                                int *stripe_count)
 562 {
 563         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join *)lmm;
 564
 565         if (lmm_bytes < sizeof(*lmmj)) {
 566                 CERROR("lov_mds_md too small: %d, need at least %d\n",
 567                        lmm_bytes, (int)sizeof(*lmmj));
 568                 return -EINVAL;
 569         }
 570
 571         if (lmmj->lmmj_array_id.lgl_oid == 0) {
 572                 CERROR("zero array object id\n");
 573                 return -EINVAL;
 574         }
 575
 576         *stripe_count = le32_to_cpu(lmmj->lmmj_md.lmm_stripe_count);
 577
 578         return lsm_lmm_verify_common(&lmmj->lmmj_md, lmm_bytes, *stripe_count);
 579 }
 580
 581 static int lovea_init_array_info(struct lov_stripe_md *lsm,
 582                                  struct llog_logid *logid,
 583                                  __u32 extent_count)
 584 {
 585         struct lov_array_info *lai;
 586         ENTRY;
 587
 588         OBD_ALLOC_PTR(lai);
 589         if (!lai)
 590                 RETURN(-ENOMEM);
 591
 592         lai->lai_array_id.lgl_oid = le64_to_cpu(logid->lgl_oid);
 593         lai->lai_array_id.lgl_ogr = le64_to_cpu(logid->lgl_ogr);
 594         lai->lai_array_id.lgl_ogen = le32_to_cpu(logid->lgl_ogen);
 595         lai->lai_ext_count = le32_to_cpu(extent_count);
 596         lsm->lsm_array = lai;
 597         RETURN(0);
 598 }
 599
 600 static int lsm_unpackmd_join(struct lov_obd *lov, struct lov_stripe_md *lsm,
 601                       struct lov_mds_md *lmm)
 602 {
 603         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join*)lmm;
 604         int    rc;
 605         ENTRY;
 606
 607         lsm_unpackmd_common(lsm, &lmmj->lmmj_md);
 608
 609         rc = lovea_init_array_info(lsm, &lmmj->lmmj_array_id,
 610                                    lmmj->lmmj_extent_count);
 611         if (rc) {
 612                 CERROR("Init joined lsm id"LPU64" arrary error %d",
 613                         lsm->lsm_object_id, rc);
 614                 GOTO(out, rc);
 615         }
 616 out:
 617         RETURN(rc);
 618 }
 619
 620 const struct lsm_operations lsm_join_ops = {
 621         .lsm_free             = lsm_free_join,
 622         .lsm_destroy          = lsm_destroy_join,
 623         .lsm_stripe_by_index  = lsm_stripe_by_index_join,
 624         .lsm_stripe_by_offset = lsm_stripe_by_offset_join,
 625         .lsm_revalidate       = lsm_revalidate_join,
 626         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_join,
 627         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_join,
 628         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_join,
 629         .lsm_lmm_verify         = lsm_lmm_verify_join,
 630         .lsm_unpackmd           = lsm_unpackmd_join,
 631 };
 632
 633
 634 static int lsm_lmm_verify_v3(struct lov_mds_md *lmmv1, int lmm_bytes,
 635                              int *stripe_count)
 636 {
 637         struct lov_mds_md_v3 *lmm;
 638
 639         lmm = (struct lov_mds_md_v3 *)lmmv1;
 640
 641         if (lmm_bytes < sizeof(*lmm)) {
 642                 CERROR("lov_mds_md_v3 too small: %d, need at least %d\n",
 643                        lmm_bytes, (int)sizeof(*lmm));
 644                 return -EINVAL;
 645         }
 646
 647         *stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
 648
 649         if (lmm_bytes < lov_mds_md_size(*stripe_count, LOV_MAGIC_V3)) {
 650                 CERROR("LOV EA V3 too small: %d, need %d\n",
 651                        lmm_bytes, lov_mds_md_size(*stripe_count, LOV_MAGIC_V3));
 652                 lov_dump_lmm_v3(D_WARNING, lmm);
 653                 return -EINVAL;
 654         }
 655
 656         return lsm_lmm_verify_common((struct lov_mds_md_v1 *)lmm, lmm_bytes,
 657                                      *stripe_count);
 658 }
 659
 660 int lsm_unpackmd_v3(struct lov_obd *lov, struct lov_stripe_md *lsm,
 661                     struct lov_mds_md *lmmv1)
 662 {
 663         struct lov_mds_md_v3 *lmm;
 664         struct lov_oinfo *loi;
 665         int i;
 666
 667         lmm = (struct lov_mds_md_v3 *)lmmv1;
 668
 669         lsm_unpackmd_common(lsm, (struct lov_mds_md_v1 *)lmm);
 670         strncpy(lsm->lsm_pool_name, lmm->lmm_pool_name, LOV_MAXPOOLNAME);
 671
 672         for (i = 0; i < lsm->lsm_stripe_count; i++) {
 673                 /* XXX LOV STACKING call down to osc_unpackmd() */
 674                 loi = lsm->lsm_oinfo[i];
 675                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 676                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 677                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 678                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 679                 if (loi->loi_ost_idx >= lov->desc.ld_tgt_count) {
 680                         CERROR("OST index %d more than OST count %d\n",
 681                                loi->loi_ost_idx, lov->desc.ld_tgt_count);
 682                         lov_dump_lmm_v3(D_WARNING, lmm);
 683                         return -EINVAL;
 684                 }
 685                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
 686                         CERROR("OST index %d missing\n", loi->loi_ost_idx);
 687                         lov_dump_lmm_v3(D_WARNING, lmm);
 688                         return -EINVAL;
 689                 }
 690         }
 691
 692         return 0;
 693 }
 694
 695 const struct lsm_operations lsm_v3_ops = {
 696         .lsm_free            = lsm_free_plain,
 697         .lsm_destroy         = lsm_destroy_plain,
 698         .lsm_stripe_by_index    = lsm_stripe_by_index_plain,
 699         .lsm_stripe_by_offset   = lsm_stripe_by_offset_plain,
 700         .lsm_revalidate         = lsm_revalidate_plain,
 701         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_plain,
 702         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_plain,
 703         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_plain,
 704         .lsm_lmm_verify         = lsm_lmm_verify_v3,
 705         .lsm_unpackmd           = lsm_unpackmd_v3,
 706 };
 707