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 > 0xffffffff){
  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 = (ulong)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 = (ulong)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 obd_off
 115 lsm_stripe_offset_by_offset_plain(struct lov_stripe_md *lsm,
 116                                   obd_off lov_off)
 117 {
 118         return 0;
 119 }
 120
 121 static int
 122 lsm_stripe_index_by_offset_plain(struct lov_stripe_md *lsm,
 123                                   obd_off lov_off)
 124 {
 125         return 0;
 126 }
 127
 128 static void lsm_free_plain(struct lov_stripe_md *lsm)
 129 {
 130         OBD_FREE(lsm, lov_stripe_md_size(lsm->lsm_stripe_count));
 131 }
 132
 133 static int lsm_revalidate_plain(struct lov_stripe_md *lsm,
 134                                 struct obd_device *obd)
 135 {
 136         return 0;
 137 }
 138
 139 static int lsm_destroy_plain(struct lov_stripe_md *lsm, struct obdo *oa,
 140                              struct obd_export *md_exp)
 141 {
 142         return 0;
 143 }
 144
 145 static int lsm_lmm_verify_plain(struct lov_mds_md *lmm, int lmm_bytes,
 146                              int *stripe_count)
 147 {
 148         if (lmm_bytes < sizeof(*lmm)) {
 149                 CERROR("lov_mds_md too small: %d, need at least %d\n",
 150                        lmm_bytes, (int)sizeof(*lmm));
 151                 return -EINVAL;
 152         }
 153
 154         *stripe_count = le32_to_cpu(lmm->lmm_stripe_count);
 155
 156         if (lmm_bytes < lov_mds_md_v1_size(*stripe_count)) {
 157                 CERROR("LOV EA too small: %d, need %d\n",
 158                        lmm_bytes, lov_mds_md_v1_size(*stripe_count));
 159                 lov_dump_lmm_v1(D_WARNING, lmm);
 160                 return -EINVAL;
 161         }
 162
 163         return lsm_lmm_verify_common(lmm, lmm_bytes, *stripe_count);
 164 }
 165
 166 int lsm_unpackmd_plain(struct lov_obd *lov, struct lov_stripe_md *lsm,
 167                     struct lov_mds_md_v1 *lmm)
 168 {
 169         struct lov_oinfo *loi;
 170         int i;
 171
 172         lsm_unpackmd_common(lsm, lmm);
 173
 174         for (i = 0, loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++) {
 175                 /* XXX LOV STACKING call down to osc_unpackmd() */
 176                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 177                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 178                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 179                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 180                 if (loi->loi_ost_idx >= lov->desc.ld_tgt_count) {
 181                         CERROR("OST index %d more than OST count %d\n",
 182                                loi->loi_ost_idx, lov->desc.ld_tgt_count);
 183                         lov_dump_lmm_v1(D_WARNING, lmm);
 184                         return -EINVAL;
 185                 }
 186                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
 187                         CERROR("OST index %d missing\n", loi->loi_ost_idx);
 188                         lov_dump_lmm_v1(D_WARNING, lmm);
 189                         return -EINVAL;
 190                 }
 191                 loi++;
 192         }
 193
 194         return 0;
 195 }
 196
 197 struct lsm_operations lsm_plain_ops = {
 198         .lsm_free            = lsm_free_plain,
 199         .lsm_destroy         = lsm_destroy_plain,
 200         .lsm_stripe_by_index    = lsm_stripe_by_index_plain,
 201         .lsm_stripe_by_offset   = lsm_stripe_by_offset_plain,
 202         .lsm_revalidate         = lsm_revalidate_plain,
 203         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_plain,
 204         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_plain,
 205         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_plain,
 206         .lsm_lmm_verify         = lsm_lmm_verify_plain,
 207         .lsm_unpackmd           = lsm_unpackmd_plain,
 208 };
 209
 210 struct lov_extent *lovea_off2le(struct lov_stripe_md *lsm, obd_off lov_off)
 211 {
 212         struct lov_array_info *lai;
 213         struct lov_extent *le;
 214         int i = 0;
 215
 216         LASSERT(lsm->lsm_array != NULL);
 217         lai = lsm->lsm_array;
 218         LASSERT(lai->lai_ext_count > 1);
 219
 220         for (le = lai->lai_ext_array, i = 0;
 221              i < lai->lai_ext_count && le->le_start + le->le_len <= lov_off
 222              && le->le_len != -1;
 223              i ++, le ++) {
 224                ; /* empty loop */
 225         }
 226
 227         CDEBUG(D_INFO, "off "LPU64" idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
 228                lov_off, i, le->le_start, le->le_len, le->le_loi_idx,
 229                le->le_stripe_count);
 230
 231         RETURN(le);
 232 }
 233
 234 struct lov_extent *lovea_idx2le(struct lov_stripe_md *lsm, int stripe_no)
 235 {
 236         struct lov_extent *le;
 237         struct lov_array_info *lai;
 238         int i, stripe_index;
 239
 240         LASSERT(lsm->lsm_array != NULL);
 241         LASSERT(stripe_no >= 0 && stripe_no <= lsm->lsm_stripe_count);
 242         lai = lsm->lsm_array;
 243         LASSERT(lai->lai_ext_count > 1);
 244
 245         for (le = lai->lai_ext_array, i = 0, stripe_index = le->le_stripe_count;
 246              i < lai->lai_ext_count && stripe_index <= stripe_no &&
 247              le->le_len != -1; i ++, le ++,
 248              stripe_index += le->le_stripe_count) {
 249                 ; /* empty loop */
 250         }
 251
 252         CDEBUG(D_INFO, "stripe %d idx %d, ext "LPU64":"LPU64" idx %d sc %d\n",
 253                stripe_no, i, le->le_start, le->le_len, le->le_loi_idx,
 254                le->le_stripe_count);
 255         RETURN(le);
 256 }
 257
 258 static void lovea_free_array_info(struct lov_stripe_md *lsm)
 259 {
 260         if (!lsm || !lsm->lsm_array)
 261                 return;
 262
 263         if (lsm->lsm_array->lai_ext_array)
 264                 OBD_FREE(lsm->lsm_array->lai_ext_array,
 265                          lsm->lsm_array->lai_ext_count *
 266                          sizeof(struct lov_extent));
 267
 268         OBD_FREE_PTR(lsm->lsm_array);
 269 }
 270
 271 static void lsm_free_join(struct lov_stripe_md *lsm)
 272 {
 273         lovea_free_array_info(lsm);
 274         OBD_FREE(lsm, lov_stripe_md_size(lsm->lsm_stripe_count));
 275 }
 276
 277 static void
 278 lsm_stripe_by_index_join(struct lov_stripe_md *lsm, int *stripeno,
 279                            obd_off *lov_off, unsigned long *swidth)
 280 {
 281         struct lov_extent *le;
 282
 283         LASSERT(stripeno != NULL);
 284
 285         le = lovea_idx2le(lsm, *stripeno);
 286
 287         LASSERT(le != NULL && le->le_stripe_count != 0);
 288
 289         *stripeno -= le->le_loi_idx;
 290
 291         if (swidth)
 292                 *swidth = (ulong)lsm->lsm_stripe_size * le->le_stripe_count;
 293
 294         if (lov_off) {
 295                 struct lov_extent *lov_le = lovea_off2le(lsm, *lov_off);
 296                 if (lov_le == le) {
 297                         *lov_off = (*lov_off > le->le_start) ?
 298                                    (*lov_off - le->le_start) : 0;
 299                 } else {
 300                         *lov_off = (*lov_off > le->le_start) ?
 301                                    le->le_len : 0;
 302                         LASSERT(*lov_off != -1);
 303                 }
 304         }
 305 }
 306
 307 static void
 308 lsm_stripe_by_offset_join(struct lov_stripe_md *lsm, int *stripeno,
 309                            obd_off *lov_off, unsigned long *swidth)
 310 {
 311         struct lov_extent *le;
 312
 313         LASSERT(lov_off != NULL);
 314
 315         le = lovea_off2le(lsm, *lov_off);
 316
 317         LASSERT(le != NULL && le->le_stripe_count != 0);
 318
 319         *lov_off = (*lov_off > le->le_start) ? (*lov_off - le->le_start) : 0;
 320
 321         if (stripeno)
 322                 *stripeno -= le->le_loi_idx;
 323
 324         if (swidth)
 325                 *swidth = (ulong)lsm->lsm_stripe_size * le->le_stripe_count;
 326 }
 327
 328 static obd_off
 329 lsm_stripe_offset_by_index_join(struct lov_stripe_md *lsm,
 330                                  int stripe_index)
 331 {
 332         struct lov_extent *le;
 333
 334         le = lovea_idx2le(lsm, stripe_index);
 335
 336         return le ? le->le_start : 0;
 337 }
 338
 339 static obd_off
 340 lsm_stripe_offset_by_offset_join(struct lov_stripe_md *lsm,
 341                                  obd_off lov_off)
 342 {
 343         struct lov_extent *le;
 344
 345         le = lovea_off2le(lsm, lov_off);
 346
 347         return le ? le->le_start : 0;
 348 }
 349
 350 static int
 351 lsm_stripe_index_by_offset_join(struct lov_stripe_md *lsm,
 352                                  obd_off lov_off)
 353 {
 354         struct lov_extent *le = NULL;
 355
 356         le = lovea_off2le(lsm, lov_off);
 357
 358         return le ? le->le_loi_idx : 0;
 359 }
 360
 361 static int lovea_unpack_array(struct llog_handle *handle,
 362                               struct llog_rec_hdr *rec, void *data)
 363 {
 364         struct lovea_unpack_args *args = (struct lovea_unpack_args *)data;
 365         struct llog_array_rec *la_rec = (struct llog_array_rec*)rec;
 366         struct mds_extent_desc *med = &la_rec->lmr_med;
 367         struct lov_stripe_md *lsm = args->lsm;
 368         int cursor = args->cursor++;
 369         struct lov_mds_md *lmm;
 370         struct lov_array_info *lai;
 371         struct lov_oinfo * loi;
 372         int i, loi_index;
 373         ENTRY;
 374
 375         /* sanity check */
 376         LASSERT(lsm->lsm_stripe_count != 0);
 377         lmm = &med->med_lmm;
 378         LASSERT(lsm->lsm_array != NULL);
 379
 380         lai = lsm->lsm_array;
 381
 382         if (cursor == 0) {
 383                lai->lai_ext_array[cursor].le_loi_idx = 0;
 384         } else {
 385                int next_loi_index = lai->lai_ext_array[cursor - 1].le_loi_idx +
 386                                  lai->lai_ext_array[cursor - 1].le_stripe_count;
 387                lai->lai_ext_array[cursor].le_loi_idx = next_loi_index;
 388         }
 389         /* insert extent desc into lsm extent array  */
 390         lai->lai_ext_array[cursor].le_start = le64_to_cpu(med->med_start);
 391         lai->lai_ext_array[cursor].le_len   = le64_to_cpu(med->med_len);
 392         lai->lai_ext_array[cursor].le_stripe_count = lmm->lmm_stripe_count;
 393
 394         /* unpack extent's lmm to lov_oinfo array */
 395         loi_index = lai->lai_ext_array[cursor].le_loi_idx;
 396         loi = &lsm->lsm_oinfo[loi_index];
 397         CDEBUG(D_INFO, "lovea upackmd cursor %d, loi_index %d extent "
 398                         LPU64":"LPU64"\n", cursor, loi_index, med->med_start,
 399                         med->med_len);
 400
 401         for (i = 0; i < lmm->lmm_stripe_count; i ++) {
 402                 /* XXX LOV STACKING call down to osc_unpackmd() */
 403                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 404                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 405                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 406                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 407                 loi++;
 408         }
 409
 410         RETURN(0);
 411 }
 412
 413 static int lsm_revalidate_join(struct lov_stripe_md *lsm,
 414                                struct obd_device *obd)
 415 {
 416         struct llog_handle *llh;
 417         struct llog_ctxt *ctxt;
 418         struct lovea_unpack_args args;
 419         int rc, rc2;
 420         ENTRY;
 421
 422         LASSERT(lsm->lsm_array != NULL);
 423
 424         /*Revalidate lsm might be called from client or MDS server.
 425          *So the ctxt might be in different position
 426          */
 427         ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT);
 428         if (!ctxt)
 429                 ctxt = llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT);
 430
 431         LASSERT(ctxt);
 432
 433         if (lsm->lsm_array && lsm->lsm_array->lai_ext_array)
 434                 RETURN(0);
 435
 436         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 437                lsm->lsm_array->lai_array_id.lgl_oid,
 438                lsm->lsm_array->lai_array_id.lgl_ogr);
 439         OBD_ALLOC(lsm->lsm_array->lai_ext_array,lsm->lsm_array->lai_ext_count *
 440                                                 sizeof (struct lov_extent));
 441         if (!lsm->lsm_array->lai_ext_array)
 442                 RETURN(-ENOMEM);
 443
 444         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 445                lsm->lsm_array->lai_array_id.lgl_oid,
 446                lsm->lsm_array->lai_array_id.lgl_ogr);
 447
 448         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id, NULL);
 449         if (rc)
 450                 GOTO(out, rc);
 451
 452         args.lsm = lsm;
 453         args.cursor = 0;
 454         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 455         if (rc == 0)
 456                 rc = llog_process(llh, lovea_unpack_array, &args, NULL);
 457         rc2 = llog_close(llh);
 458         if (rc == 0)
 459                 rc = rc2;
 460 out:
 461         if (rc)
 462                 lovea_free_array_info(lsm);
 463         RETURN(rc);
 464 }
 465
 466 int lsm_destroy_join(struct lov_stripe_md *lsm, struct obdo *oa,
 467                       struct obd_export *md_exp)
 468 {
 469         struct llog_ctxt *ctxt;
 470         struct llog_handle *llh;
 471         int rc = 0;
 472         ENTRY;
 473
 474         LASSERT(md_exp != NULL);
 475         ctxt = llog_get_context(md_exp->exp_obd, LLOG_LOVEA_REPL_CTXT);
 476         if (!ctxt)
 477                 GOTO(out, rc = -EINVAL);
 478
 479         LASSERT(lsm->lsm_array != NULL);
 480         /*for those orphan inode, we should keep array id*/
 481         if (!(oa->o_valid & OBD_MD_FLCOOKIE))
 482                 RETURN(0);
 483
 484         LASSERT(ctxt != NULL);
 485         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id,
 486                          NULL);
 487         if (rc)
 488                 GOTO(out, rc);
 489
 490         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 491         if (rc == 0) {
 492                 rc = llog_destroy(llh);
 493         }
 494         llog_free_handle(llh);
 495 out:
 496         RETURN(rc);
 497 }
 498
 499 static int lsm_lmm_verify_join(struct lov_mds_md *lmm, int lmm_bytes,
 500                                int *stripe_count)
 501 {
 502         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join *)lmm;
 503
 504         if (lmm_bytes < sizeof(*lmmj)) {
 505                 CERROR("lov_mds_md too small: %d, need at least %d\n",
 506                        lmm_bytes, (int)sizeof(*lmmj));
 507                 return -EINVAL;
 508         }
 509
 510         if (lmmj->lmmj_array_id.lgl_oid == 0) {
 511                 CERROR("zero array object id\n");
 512                 return -EINVAL;
 513         }
 514
 515         *stripe_count = le32_to_cpu(lmmj->lmmj_md.lmm_stripe_count);
 516
 517         return lsm_lmm_verify_common(&lmmj->lmmj_md, lmm_bytes, *stripe_count);
 518 }
 519
 520 static int lovea_init_array_info(struct lov_stripe_md *lsm,
 521                                  struct llog_logid *logid,
 522                                  __u32 extent_count)
 523 {
 524         struct lov_array_info *lai;
 525         ENTRY;
 526
 527         OBD_ALLOC_PTR(lai);
 528         if (!lai)
 529                 RETURN(-ENOMEM);
 530
 531         lai->lai_array_id = *logid;
 532         lai->lai_ext_count = extent_count;
 533         lsm->lsm_array = lai;
 534         RETURN(0);
 535 }
 536
 537 static int lsm_unpackmd_join(struct lov_obd *lov, struct lov_stripe_md *lsm,
 538                       struct lov_mds_md *lmm)
 539 {
 540         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join*)lmm;
 541         int    rc;
 542         ENTRY;
 543
 544         lsm_unpackmd_common(lsm, &lmmj->lmmj_md);
 545
 546         rc = lovea_init_array_info(lsm, &lmmj->lmmj_array_id,
 547                                    lmmj->lmmj_extent_count);
 548         if (rc) {
 549                 CERROR("Init joined lsm id"LPU64" arrary error %d",
 550                         lsm->lsm_object_id, rc);
 551                 GOTO(out, rc);
 552         }
 553 out:
 554         RETURN(rc);
 555 }
 556
 557 struct lsm_operations lsm_join_ops = {
 558         .lsm_free             = lsm_free_join,
 559         .lsm_destroy          = lsm_destroy_join,
 560         .lsm_stripe_by_index  = lsm_stripe_by_index_join,
 561         .lsm_stripe_by_offset = lsm_stripe_by_offset_join,
 562         .lsm_revalidate       = lsm_revalidate_join,
 563         .lsm_stripe_offset_by_index  = lsm_stripe_offset_by_index_join,
 564         .lsm_stripe_offset_by_offset = lsm_stripe_offset_by_offset_join,
 565         .lsm_stripe_index_by_offset  = lsm_stripe_index_by_offset_join,
 566         .lsm_lmm_verify         = lsm_lmm_verify_join,
 567         .lsm_unpackmd           = lsm_unpackmd_join,
 568 };
 569
 570