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 #else
  34 #include <liblustre.h>
  35 #endif
  36
  37 #include <linux/obd_class.h>
  38 #include <linux/obd_lov.h>
  39 #include <linux/lustre_idl.h>
  40 #include <linux/lustre_log.h>
  41
  42 #include "lov_internal.h"
  43
  44 struct lovea_unpack_args {
  45         struct lov_stripe_md *lsm;
  46         int                   cursor;
  47 };
  48
  49 static int lsm_lmm_verify_common(struct lov_mds_md *lmm, int lmm_bytes,
  50                                  int stripe_count)
  51 {
  52
  53         if (stripe_count == 0) {
  54                 CERROR("bad stripe count %d\n", stripe_count);
  55                 lov_dump_lmm_v1(D_WARNING, lmm);
  56                 return -EINVAL;
  57         }
  58
  59         if (lmm->lmm_object_id == 0) {
  60                 CERROR("zero object id\n");
  61                 lov_dump_lmm_v1(D_WARNING, lmm);
  62                 return -EINVAL;
  63         }
  64
  65         if (lmm->lmm_pattern != cpu_to_le32(LOV_PATTERN_RAID0)) {
  66                 CERROR("bad striping pattern\n");
  67                 lov_dump_lmm_v1(D_WARNING, lmm);
  68                 return -EINVAL;
  69         }
  70
  71         if (lmm->lmm_stripe_size == 0 ||
  72             (__u64)le32_to_cpu(lmm->lmm_stripe_size) * stripe_count > ~0UL) {
  73                 CERROR("bad stripe size %u\n",
  74                        le32_to_cpu(lmm->lmm_stripe_size));
  75                 lov_dump_lmm_v1(D_WARNING, lmm);
  76                 return -EINVAL;
  77         }
  78         return 0;
  79 }
  80
  81 static void lsm_unpackmd_common(struct lov_stripe_md *lsm,
  82                                 struct lov_mds_md *lmm)
  83 {
  84         lsm->lsm_object_id = le64_to_cpu(lmm->lmm_object_id);
  85         lsm->lsm_object_gr = le64_to_cpu(lmm->lmm_object_gr);
  86         lsm->lsm_stripe_size = le32_to_cpu(lmm->lmm_stripe_size);
  87         lsm->lsm_pattern = le32_to_cpu(lmm->lmm_pattern);
  88         lsm->lsm_xfersize = lsm->lsm_stripe_size * lsm->lsm_stripe_count;
  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                 loi++;
 180         }
 181
 182         return 0;
 183 }
 184
 185 struct lsm_operations lsm_plain_ops = {
 186         .lsm_free            = lsm_free_plain,
 187         .lsm_destroy         = lsm_destroy_plain,
 188         .lsm_stripe_by_index    = lsm_stripe_by_index_plain,
 189         .lsm_stripe_by_offset   = lsm_stripe_by_offset_plain,
 190         .lsm_revalidate         = lsm_revalidate_plain,
 191         .lsm_stripe_offset_by_index = lsm_stripe_offset_by_index_plain,
 192         .lsm_stripe_index_by_offset = lsm_stripe_index_by_offset_plain,
 193         .lsm_lmm_verify         = lsm_lmm_verify_plain,
 194         .lsm_unpackmd           = lsm_unpackmd_plain,
 195 };
 196
 197 struct lov_extent *lovea_off2le(struct lov_stripe_md *lsm, obd_off lov_off)
 198 {
 199         struct lov_array_info *lai;
 200         struct lov_extent *le;
 201         int i = 0;
 202
 203         LASSERT(lsm->lsm_array != NULL);
 204         lai = lsm->lsm_array;
 205         LASSERT(lai->lai_ext_count > 1);
 206
 207         for (le = lai->lai_ext_array, i = 0;
 208              i < lai->lai_ext_count && le->le_start + le->le_len <= lov_off
 209              && le->le_len != -1;
 210              i ++, le ++) {
 211                ; /* empty loop */
 212         }
 213
 214         CDEBUG(D_INFO, "off "LPU64" idx%d, ext"LPU64":"LPU64"idx%d sc%d\n",
 215                lov_off, i, le->le_start, le->le_len, le->le_loi_idx,
 216                le->le_stripe_count);
 217
 218         RETURN(le);
 219 }
 220
 221 struct lov_extent *lovea_idx2le(struct lov_stripe_md *lsm, int stripe_no)
 222 {
 223         struct lov_extent *le;
 224         struct lov_array_info *lai;
 225         int i, stripe_index;
 226
 227         LASSERT(lsm->lsm_array != NULL);
 228         LASSERT(stripe_no >= 0 && stripe_no <= lsm->lsm_stripe_count);
 229         lai = lsm->lsm_array;
 230         LASSERT(lai->lai_ext_count > 1);
 231
 232         for (le = lai->lai_ext_array, i = 0, stripe_index = le->le_stripe_count;
 233              i < lai->lai_ext_count && stripe_index <= stripe_no &&
 234              le->le_len != -1; i ++, le ++,
 235              stripe_index += le->le_stripe_count) {
 236                 ; /* empty loop */
 237         }
 238
 239         CDEBUG(D_INFO, "stripe %d idx%d, ext"LPU64":"LPU64"idx %d scount%d\n",
 240                stripe_no, i, le->le_start, le->le_len, le->le_loi_idx,
 241                le->le_stripe_count);
 242         RETURN(le);
 243 }
 244
 245
 246 static void lovea_free_array_info(struct lov_stripe_md *lsm)
 247 {
 248         if (!lsm || !lsm->lsm_array)
 249                 return;
 250
 251         if (lsm->lsm_array->lai_ext_array)
 252                 OBD_FREE(lsm->lsm_array->lai_ext_array,
 253                          lsm->lsm_array->lai_ext_count *
 254                          sizeof(struct lov_extent));
 255
 256         OBD_FREE_PTR(lsm->lsm_array);
 257 }
 258
 259 static void lsm_free_join(struct lov_stripe_md *lsm)
 260 {
 261         lovea_free_array_info(lsm);
 262         OBD_FREE(lsm, lov_stripe_md_size(lsm->lsm_stripe_count));
 263 }
 264
 265 static void
 266 lsm_stripe_by_index_join(struct lov_stripe_md *lsm, int *stripeno,
 267                            obd_off *lov_off, unsigned long *swidth)
 268 {
 269         struct lov_extent *le;
 270
 271         LASSERT(stripeno != NULL);
 272
 273         le = lovea_idx2le(lsm, *stripeno);
 274
 275         LASSERT(le != NULL && le->le_stripe_count != 0);
 276
 277         *stripeno -= le->le_loi_idx;
 278
 279         if (swidth)
 280                 *swidth = lsm->lsm_stripe_size * le->le_stripe_count;
 281
 282         if (lov_off) {
 283                 struct lov_extent *lov_le = lovea_off2le(lsm, *lov_off);
 284                 if (lov_le == le) {
 285                         *lov_off = (*lov_off > le->le_start) ?
 286                                    (*lov_off - le->le_start) : 0;
 287                 } else {
 288                         *lov_off = (*lov_off > le->le_start) ?
 289                                    le->le_len : 0;
 290                         LASSERT(*lov_off != -1);
 291                 }
 292         }
 293 }
 294
 295 static void
 296 lsm_stripe_by_offset_join(struct lov_stripe_md *lsm, int *stripeno,
 297                            obd_off *lov_off, unsigned long *swidth)
 298 {
 299         struct lov_extent *le;
 300
 301         LASSERT(lov_off != NULL);
 302
 303         le = lovea_off2le(lsm, *lov_off);
 304
 305         LASSERT(le != NULL && le->le_stripe_count != 0);
 306
 307         *lov_off = (*lov_off > le->le_start) ? (*lov_off - le->le_start) : 0;
 308
 309         if (stripeno)
 310                 *stripeno -= le->le_loi_idx;
 311
 312         if (swidth)
 313                 *swidth = lsm->lsm_stripe_size * le->le_stripe_count;
 314 }
 315
 316 static obd_off
 317 lsm_stripe_offset_by_index_join(struct lov_stripe_md *lsm,
 318                                  int stripe_index)
 319 {
 320         struct lov_extent *le;
 321
 322         le = lovea_idx2le(lsm, stripe_index);
 323
 324         return le ? le->le_start : 0;
 325 }
 326
 327 static int
 328 lsm_stripe_index_by_offset_join(struct lov_stripe_md *lsm,
 329                                  obd_off lov_off)
 330 {
 331         struct lov_extent *le = NULL;
 332
 333         le = lovea_off2le(lsm, lov_off);
 334
 335         return le ? le->le_loi_idx : 0;
 336 }
 337
 338 static int lovea_unpack_array(struct llog_handle *handle,
 339                               struct llog_rec_hdr *rec, void *data)
 340 {
 341         struct lovea_unpack_args *args = (struct lovea_unpack_args *)data;
 342         struct llog_array_rec *la_rec = (struct llog_array_rec*)rec;
 343         struct mds_extent_desc *med = &la_rec->lmr_med;
 344         struct lov_stripe_md *lsm = args->lsm;
 345         int cursor = args->cursor++;
 346         struct lov_mds_md *lmm;
 347         struct lov_array_info *lai;
 348         struct lov_oinfo * loi;
 349         int i, loi_index;
 350         ENTRY;
 351
 352         /* sanity check */
 353         LASSERT(lsm->lsm_stripe_count != 0);
 354         lmm = &med->med_lmm;
 355         LASSERT(lsm->lsm_array != NULL);
 356
 357         lai = lsm->lsm_array;
 358
 359         if (cursor == 0) {
 360                lai->lai_ext_array[cursor].le_loi_idx = 0;
 361         } else {
 362                int next_loi_index = lai->lai_ext_array[cursor - 1].le_loi_idx +
 363                                  lai->lai_ext_array[cursor - 1].le_stripe_count;
 364                lai->lai_ext_array[cursor].le_loi_idx = next_loi_index;
 365         }
 366         /* insert extent desc into lsm extent array  */
 367         lai->lai_ext_array[cursor].le_start = le64_to_cpu(med->med_start);
 368         lai->lai_ext_array[cursor].le_len   = le64_to_cpu(med->med_len);
 369         lai->lai_ext_array[cursor].le_stripe_count = lmm->lmm_stripe_count;
 370
 371         /* unpack extent's lmm to lov_oinfo array */
 372         loi_index = lai->lai_ext_array[cursor].le_loi_idx;
 373         loi = &lsm->lsm_oinfo[loi_index];
 374         CDEBUG(D_INFO, "lovea upackmd cursor %d, loi_index %d extent "
 375                         LPU64":"LPU64"\n", cursor, loi_index, med->med_start,
 376                         med->med_len);
 377
 378         for (i = 0; i < lmm->lmm_stripe_count; i ++) {
 379                 /* XXX LOV STACKING call down to osc_unpackmd() */
 380                 loi->loi_id = le64_to_cpu(lmm->lmm_objects[i].l_object_id);
 381                 loi->loi_gr = le64_to_cpu(lmm->lmm_objects[i].l_object_gr);
 382                 loi->loi_ost_idx = le32_to_cpu(lmm->lmm_objects[i].l_ost_idx);
 383                 loi->loi_ost_gen = le32_to_cpu(lmm->lmm_objects[i].l_ost_gen);
 384                 loi++;
 385         }
 386
 387         RETURN(0);
 388 }
 389
 390 static int lsm_revalidate_join(struct lov_stripe_md *lsm,
 391                                struct obd_device *obd)
 392 {
 393         struct llog_handle *llh;
 394         struct llog_ctxt *ctxt;
 395         struct lovea_unpack_args args;
 396         int rc, rc2;
 397         ENTRY;
 398
 399         LASSERT(lsm->lsm_array != NULL);
 400
 401         /*Revalidate lsm might be called from client or MDS server.
 402          *So the ctxt might be in different position
 403          */
 404         ctxt = llog_get_context(obd, LLOG_LOVEA_REPL_CTXT);
 405         if (!ctxt)
 406                 ctxt = llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT);
 407
 408         LASSERT(ctxt);
 409
 410         if (lsm->lsm_array && lsm->lsm_array->lai_ext_array)
 411                 RETURN(0);
 412
 413         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 414                lsm->lsm_array->lai_array_id.lgl_oid,
 415                lsm->lsm_array->lai_array_id.lgl_ogr);
 416         OBD_ALLOC(lsm->lsm_array->lai_ext_array,lsm->lsm_array->lai_ext_count *
 417                                                 sizeof (struct lov_extent));
 418         if (!lsm->lsm_array->lai_ext_array)
 419                 RETURN(-ENOMEM);
 420
 421         CDEBUG(D_INFO, "get lsm logid: "LPU64":"LPU64"\n",
 422                lsm->lsm_array->lai_array_id.lgl_oid,
 423                lsm->lsm_array->lai_array_id.lgl_ogr);
 424
 425         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id, NULL);
 426         if (rc)
 427                 GOTO(out, rc);
 428
 429         args.lsm = lsm;
 430         args.cursor = 0;
 431         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 432         if (rc == 0)
 433                 rc = llog_process(llh, lovea_unpack_array, &args, NULL);
 434         rc2 = llog_close(llh);
 435         if (rc == 0)
 436                 rc = rc2;
 437 out:
 438         if (rc)
 439                 lovea_free_array_info(lsm);
 440         RETURN(rc);
 441 }
 442
 443 int lsm_destroy_join(struct lov_stripe_md *lsm, struct obdo *oa,
 444                       struct obd_export *md_exp)
 445 {
 446         struct llog_ctxt *ctxt;
 447         struct llog_handle *llh;
 448         int rc = 0;
 449         ENTRY;
 450
 451         LASSERT(md_exp != NULL);
 452         ctxt = llog_get_context(md_exp->exp_obd, LLOG_LOVEA_REPL_CTXT);
 453         if (!ctxt)
 454                 GOTO(out, rc = -EINVAL);
 455
 456         LASSERT(lsm->lsm_array != NULL);
 457         /*for those orphan inode, we should keep array id*/
 458         if (!(oa->o_valid & OBD_MD_FLCOOKIE))
 459                 RETURN(0);
 460
 461         LASSERT(ctxt != NULL);
 462         rc = llog_create(ctxt, &llh, &lsm->lsm_array->lai_array_id,
 463                          NULL);
 464         if (rc)
 465                 GOTO(out, rc);
 466
 467         rc = llog_init_handle(llh, LLOG_F_IS_PLAIN, NULL);
 468         if (rc == 0) {
 469                 rc = llog_destroy(llh);
 470         }
 471         llog_free_handle(llh);
 472 out:
 473         RETURN(rc);
 474 }
 475
 476 static int lsm_lmm_verify_join(struct lov_mds_md *lmm, int lmm_bytes,
 477                                int *stripe_count)
 478 {
 479         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join *)lmm;
 480
 481         if (lmm_bytes < sizeof(*lmmj)) {
 482                 CERROR("lov_mds_md too small: %d, need at least %d\n",
 483                        lmm_bytes, (int)sizeof(*lmmj));
 484                 return -EINVAL;
 485         }
 486
 487         if (lmmj->lmmj_array_id.lgl_oid == 0) {
 488                 CERROR("zero array object id\n");
 489                 return -EINVAL;
 490         }
 491
 492         *stripe_count = le32_to_cpu(lmmj->lmmj_md.lmm_stripe_count);
 493
 494         return lsm_lmm_verify_common(&lmmj->lmmj_md, lmm_bytes, *stripe_count);
 495 }
 496
 497 static int lovea_init_array_info(struct lov_stripe_md *lsm,
 498                                  struct llog_logid *logid,
 499                                  __u32 extent_count)
 500 {
 501         struct lov_array_info *lai;
 502         ENTRY;
 503
 504         OBD_ALLOC_PTR(lai);
 505         if (!lai)
 506                 RETURN(-ENOMEM);
 507
 508         lai->lai_array_id = *logid;
 509         lai->lai_ext_count = extent_count;
 510         lsm->lsm_array = lai;
 511         RETURN(0);
 512 }
 513
 514 static int lsm_unpackmd_join(struct lov_obd *lov, struct lov_stripe_md *lsm,
 515                       struct lov_mds_md *lmm)
 516 {
 517         struct lov_mds_md_join *lmmj = (struct lov_mds_md_join*)lmm;
 518         int    rc;
 519         ENTRY;
 520
 521         lsm_unpackmd_common(lsm, &lmmj->lmmj_md);
 522
 523         rc = lovea_init_array_info(lsm, &lmmj->lmmj_array_id,
 524                                    lmmj->lmmj_extent_count);
 525         if (rc) {
 526                 CERROR("Init joined lsm id"LPU64" arrary error %d",
 527                         lsm->lsm_object_id, rc);
 528                 GOTO(out, rc);
 529         }
 530 out:
 531         RETURN(rc);
 532 }
 533
 534 struct lsm_operations lsm_join_ops = {
 535         .lsm_free             = lsm_free_join,
 536         .lsm_destroy          = lsm_destroy_join,
 537         .lsm_stripe_by_index  = lsm_stripe_by_index_join,
 538         .lsm_stripe_by_offset = lsm_stripe_by_offset_join,
 539         .lsm_revalidate       = lsm_revalidate_join,
 540         .lsm_stripe_offset_by_index = lsm_stripe_offset_by_index_join,
 541         .lsm_stripe_index_by_offset = lsm_stripe_index_by_offset_join,
 542         .lsm_lmm_verify         = lsm_lmm_verify_join,
 543         .lsm_unpackmd           = lsm_unpackmd_join,
 544 };
 545
 546