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