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