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