lustre/lov/lov_obd.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  *  lov/lov.c
   5  *
   6  * Copyright (C) 2002 Cluster File Systems, Inc.
   7  * Author: Phil Schwan <phil@off.net>
   8  *         Peter Braam <braam@clusterfs.com>
   9  *
  10  * This code is issued under the GNU General Public License.
  11  * See the file COPYING in this distribution
  12  */
  13
  14 #define EXPORT_SYMTAB
  15 #define DEBUG_SUBSYSTEM S_LOV
  16
  17 #include <linux/slab.h>
  18 #include <linux/module.h>
  19 #include <linux/obd_support.h>
  20 #include <linux/lustre_lib.h>
  21 #include <linux/lustre_net.h>
  22 #include <linux/lustre_idl.h>
  23 #include <linux/lustre_mds.h>
  24 #include <linux/obd_class.h>
  25 #include <linux/obd_lov.h>
  26 #include <linux/init.h>
  27
  28 extern struct obd_device obd_dev[MAX_OBD_DEVICES];
  29
  30 /* obd methods */
  31 static int lov_connect(struct lustre_handle *conn, struct obd_device *obd,
  32                        char *cluuid)
  33 {
  34         struct ptlrpc_request *req;
  35         struct lov_obd *lov = &obd->u.lov;
  36         struct lustre_handle mdc_conn;
  37         uuid_t *uuidarray;
  38         int rc, rc2;
  39         int i;
  40
  41         MOD_INC_USE_COUNT;
  42         rc = class_connect(conn, obd, cluuid);
  43         if (rc) {
  44                 MOD_DEC_USE_COUNT;
  45                 RETURN(rc);
  46         }
  47
  48         /* retrieve LOV metadata from MDS */
  49         rc = obd_connect(&mdc_conn, lov->mdcobd, NULL);
  50         if (rc) {
  51                 CERROR("cannot connect to mdc: rc = %d\n", rc);
  52                 GOTO(out, rc = -EINVAL);
  53         }
  54
  55         rc = mdc_getlovinfo(obd, &mdc_conn, &uuidarray, &req);
  56         rc2 = obd_disconnect(&mdc_conn);
  57         if (rc || rc2) {
  58                 CERROR("cannot get lov info or disconnect %d/%d\n", rc, rc2);
  59                 GOTO(out, (rc) ? rc : rc2 );
  60         }
  61
  62         /* sanity... */
  63         if (strcmp(obd->obd_uuid, lov->desc.ld_uuid)) {
  64                 CERROR("lov uuid %s not on mds device (%s)\n",
  65                        obd->obd_uuid, lov->desc.ld_uuid);
  66                 GOTO(out, rc = -EINVAL);
  67         }
  68         if (lov->desc.ld_tgt_count > 1000) {
  69                 CERROR("configuration error: target count > 1000 (%d)\n",
  70                        lov->desc.ld_tgt_count);
  71                 GOTO(out, rc = -EINVAL);
  72         }
  73         if (req->rq_repmsg->bufcount < 2 || req->rq_repmsg->buflens[1] <
  74             sizeof(uuid_t) * lov->desc.ld_tgt_count) {
  75                 CERROR("invalid uuid array returned\n");
  76                 GOTO(out, rc = -EINVAL);
  77         }
  78
  79         lov->bufsize = sizeof(struct lov_tgt_desc) *  lov->desc.ld_tgt_count;
  80         OBD_ALLOC(lov->tgts, lov->bufsize);
  81         if (!lov->tgts) {
  82                 CERROR("Out of memory\n");
  83                 GOTO(out, rc = -ENOMEM);
  84         }
  85
  86         uuidarray = lustre_msg_buf(req->rq_repmsg, 1);
  87         for (i = 0 ; i < lov->desc.ld_tgt_count; i++)
  88                 memcpy(lov->tgts[i].uuid, uuidarray[i], sizeof(uuid_t));
  89
  90         for (i = 0 ; i < lov->desc.ld_tgt_count; i++) {
  91                 struct obd_device *tgt = class_uuid2obd(uuidarray[i]);
  92                 if (!tgt) {
  93                         CERROR("Target %s not attached\n", uuidarray[i]);
  94                         GOTO(out_mem, rc = -EINVAL);
  95                 }
  96                 if (!(tgt->obd_flags & OBD_SET_UP)) {
  97                         CERROR("Target %s not set up\n", uuidarray[i]);
  98                         GOTO(out_mem, rc = -EINVAL);
  99                 }
 100                 rc = obd_connect(&lov->tgts[i].conn, tgt, NULL);
 101                 if (rc) {
 102                         CERROR("Target %s connect error %d\n",
 103                                uuidarray[i], rc);
 104                         GOTO(out_mem, rc);
 105                 }
 106         }
 107
 108  out_mem:
 109         if (rc) {
 110                 for (i = 0 ; i < lov->desc.ld_tgt_count; i++) {
 111                         rc2 = obd_disconnect(&lov->tgts[i].conn);
 112                         if (rc2)
 113                                 CERROR("BAD: Target %s disconnect error %d\n",
 114                                        uuidarray[i], rc2);
 115                 }
 116                 OBD_FREE(lov->tgts, lov->bufsize);
 117         }
 118  out:
 119         if (rc)
 120                 class_disconnect(conn);
 121         ptlrpc_free_req(req);
 122         return rc;
 123 }
 124
 125 static int lov_disconnect(struct lustre_handle *conn)
 126 {
 127         struct obd_device *obd = class_conn2obd(conn);
 128         struct lov_obd *lov = &obd->u.lov;
 129         int rc;
 130         int i;
 131
 132         if (!lov->tgts)
 133                 goto out_local;
 134
 135         for (i = 0 ; i < lov->desc.ld_tgt_count; i++) {
 136                 rc = obd_disconnect(&lov->tgts[i].conn);
 137                 if (rc) {
 138                         CERROR("Target %s disconnect error %d\n",
 139                                lov->tgts[i].uuid, rc);
 140                         RETURN(rc);
 141                 }
 142         }
 143         OBD_FREE(lov->tgts, lov->bufsize);
 144         lov->bufsize = 0;
 145         lov->tgts = NULL;
 146
 147  out_local:
 148         rc = class_disconnect(conn);
 149         if (!rc)
 150                 MOD_DEC_USE_COUNT;
 151         return rc;
 152 }
 153
 154 static int lov_setup(struct obd_device *obd, obd_count len, void *buf)
 155 {
 156         struct obd_ioctl_data* data = buf;
 157         struct lov_obd *lov = &obd->u.lov;
 158         int rc = 0;
 159         ENTRY;
 160
 161         if (data->ioc_inllen1 < 1) {
 162                 CERROR("osc setup requires an MDC UUID\n");
 163                 RETURN(-EINVAL);
 164         }
 165
 166         if (data->ioc_inllen1 > 37) {
 167                 CERROR("mdc UUID must be less than 38 characters\n");
 168                 RETURN(-EINVAL);
 169         }
 170
 171         lov->mdcobd = class_uuid2obd(data->ioc_inlbuf1);
 172         if (!lov->mdcobd) {
 173                 CERROR("LOV %s cannot locate MDC %s\n", obd->obd_uuid,
 174                        data->ioc_inlbuf1);
 175                 rc = -EINVAL;
 176         }
 177         RETURN(rc);
 178 }
 179
 180
 181 static inline int lov_stripe_md_size(struct obd_device *obd)
 182 {
 183         struct lov_obd *lov = &obd->u.lov;
 184         int size;
 185
 186         size = sizeof(struct lov_stripe_md) +
 187                 lov->desc.ld_tgt_count * sizeof(struct lov_oinfo);
 188         return size;
 189 }
 190
 191 static inline int lov_mds_md_size(struct obd_device *obd)
 192 {
 193         struct lov_obd *lov = &obd->u.lov;
 194         int size;
 195
 196         size = sizeof(struct lov_mds_md) +
 197                 lov->desc.ld_tgt_count * sizeof(struct lov_object_id);
 198         return size;
 199 }
 200
 201 /* the LOV counts on oa->o_id to be set as the LOV object id */
 202 static int lov_create(struct lustre_handle *conn, struct obdo *oa, struct lov_stripe_md **ea)
 203 {
 204         int rc = 0, i;
 205         struct obdo tmp;
 206         struct obd_export *export = class_conn2export(conn);
 207         struct lov_obd *lov;
 208         struct lov_stripe_md *md;
 209         ENTRY;
 210
 211         if (!ea) {
 212                 CERROR("lov_create needs EA for striping information\n");
 213                 RETURN(-EINVAL);
 214         }
 215         if (!export)
 216                 RETURN(-EINVAL);
 217         lov = &export->exp_obd->u.lov;
 218
 219         oa->o_easize =  lov_stripe_md_size(export->exp_obd);
 220         if (!*ea) {
 221                 OBD_ALLOC(*ea, oa->o_easize);
 222                 if (! *ea)
 223                         RETURN(-ENOMEM);
 224         }
 225
 226         md = *ea;
 227         md->lmd_easize = lov_mds_md_size(export->exp_obd);
 228         md->lmd_object_id = oa->o_id;
 229         if (!md->lmd_stripe_count) {
 230                 md->lmd_stripe_count = lov->desc.ld_default_stripe_count;
 231         }
 232
 233         if (!md->lmd_stripe_size)
 234                 md->lmd_stripe_size = lov->desc.ld_default_stripe_size;
 235
 236
 237
 238         for (i = 0; i < md->lmd_stripe_count; i++) {
 239                 struct lov_stripe_md obj_md;
 240                 struct lov_stripe_md *obj_mdp = &obj_md;
 241                 /* create data objects with "parent" OA */
 242                 memcpy(&tmp, oa, sizeof(tmp));
 243                 tmp.o_easize = sizeof(struct lov_stripe_md);
 244                 rc = obd_create(&lov->tgts[i].conn, &tmp, &obj_mdp);
 245                 if (rc)
 246                         GOTO(out_cleanup, rc);
 247                 md->lmd_oinfo[i].loi_id = tmp.o_id;
 248                 md->lmd_oinfo[i].loi_size = tmp.o_size;
 249         }
 250
 251  out_cleanup:
 252         if (rc) {
 253                 int i2, rc2;
 254                 for (i2 = 0 ; i2 < i ; i2++) {
 255                         /* destroy already created objects here */
 256                         tmp.o_id = md->lmd_oinfo[i].loi_id;
 257                         rc2 = obd_destroy(&lov->tgts[i].conn, &tmp, NULL);
 258                         if (rc2) {
 259                                 CERROR("Failed to remove object from target %d\n",
 260                                        i2);
 261                         }
 262                 }
 263         }
 264         return rc;
 265 }
 266
 267 static int lov_destroy(struct lustre_handle *conn, struct obdo *oa,
 268                        struct lov_stripe_md *md)
 269 {
 270         int rc = 0, i;
 271         struct obdo tmp;
 272         struct obd_export *export = class_conn2export(conn);
 273         struct lov_obd *lov;
 274         ENTRY;
 275
 276         if (!md) {
 277                 CERROR("LOV requires striping ea for desctruction\n");
 278                 RETURN(-EINVAL);
 279         }
 280
 281         if (!export || !export->exp_obd)
 282                 RETURN(-ENODEV);
 283
 284         lov = &export->exp_obd->u.lov;
 285         for (i = 0; i < md->lmd_stripe_count; i++) {
 286                 /* create data objects with "parent" OA */
 287                 memcpy(&tmp, oa, sizeof(tmp));
 288                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 289                 rc = obd_destroy(&lov->tgts[i].conn, &tmp, NULL);
 290                 if (!rc) {
 291                         CERROR("Error destroying object %Ld on %d\n",
 292                                oa->o_id, i);
 293                 }
 294         }
 295         RETURN(rc);
 296 }
 297
 298 static int lov_getattr(struct lustre_handle *conn, struct obdo *oa,
 299                        struct lov_stripe_md *md)
 300 {
 301         int rc = 0, i;
 302         struct obdo tmp;
 303         struct obd_export *export = class_conn2export(conn);
 304         struct lov_obd *lov;
 305         ENTRY;
 306
 307         if (!md) {
 308                 CERROR("LOV requires striping ea for desctruction\n");
 309                 RETURN(-EINVAL);
 310         }
 311
 312         if (!export || !export->exp_obd)
 313                 RETURN(-ENODEV);
 314
 315         lov = &export->exp_obd->u.lov;
 316         oa->o_size = 0;
 317         for (i = 0; i < md->lmd_stripe_count; i++) {
 318                 if (md->lmd_oinfo[i].loi_id == 0)
 319                         continue;
 320                 /* create data objects with "parent" OA */
 321                 memcpy(&tmp, oa, sizeof(tmp));
 322                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 323
 324                 rc = obd_getattr(&lov->tgts[i].conn, &tmp, NULL);
 325                 if (rc) {
 326                         CERROR("Error getattr object %Ld on %d\n",
 327                                tmp.o_id, i);
 328                 }
 329                 /* XXX can do something more sophisticated here... */
 330                 /* This some completely wrong. We only need the size from
 331                    the individual slices. */
 332                 if (i == 0 ) {
 333                         obd_id id = oa->o_id;
 334                         memcpy(oa, &tmp, sizeof(tmp));
 335                         oa->o_id = id;
 336                 } else {
 337                         oa->o_size += tmp.o_size;
 338                 }
 339         }
 340         RETURN(rc);
 341 }
 342
 343 static int lov_setattr(struct lustre_handle *conn, struct obdo *oa,
 344                        struct lov_stripe_md *md)
 345 {
 346         int rc = 0, i;
 347         struct obdo tmp;
 348         struct obd_export *export = class_conn2export(conn);
 349         struct lov_obd *lov;
 350         ENTRY;
 351
 352         if (!md) {
 353                 CERROR("LOV requires striping ea for desctruction\n");
 354                 RETURN(-EINVAL);
 355         }
 356
 357         if (!export || !export->exp_obd)
 358                 RETURN(-ENODEV);
 359
 360         lov = &export->exp_obd->u.lov;
 361         for (i = 0; i < md->lmd_stripe_count; i++) {
 362                 /* create data objects with "parent" OA */
 363                 memcpy(&tmp, oa, sizeof(tmp));
 364                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 365
 366                 rc = obd_setattr(&lov->tgts[i].conn, &tmp, NULL);
 367                 if (!rc) {
 368                         CERROR("Error setattr object %Ld on %d\n",
 369                                oa->o_id, i);
 370                 }
 371         }
 372         RETURN(rc);
 373 }
 374
 375 static int lov_open(struct lustre_handle *conn, struct obdo *oa,
 376                     struct lov_stripe_md *md)
 377 {
 378         int rc = 0, rc2 = 0, i;
 379         struct obdo tmp;
 380         struct obd_export *export = class_conn2export(conn);
 381         struct lov_obd *lov;
 382         ENTRY;
 383
 384         if (!md) {
 385                 CERROR("LOV requires striping ea for opening\n");
 386                 RETURN(-EINVAL);
 387         }
 388
 389         if (!export || !export->exp_obd)
 390                 RETURN(-ENODEV);
 391
 392         lov = &export->exp_obd->u.lov;
 393         for (i = 0; i < md->lmd_stripe_count; i++) {
 394                 /* create data objects with "parent" OA */
 395                 memcpy(&tmp, oa, sizeof(tmp));
 396                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 397
 398                 rc = obd_open(&lov->tgts[i].conn, &tmp, NULL);
 399                 if (rc) {
 400                         rc2 = rc;
 401                         CERROR("Error open object %Ld on %d\n",
 402                                oa->o_id, i);
 403                 }
 404         }
 405         RETURN(rc2);
 406 }
 407
 408
 409 static int lov_close(struct lustre_handle *conn, struct obdo *oa,
 410                      struct lov_stripe_md *md)
 411 {
 412         int rc = 0, i;
 413         struct obdo tmp;
 414         struct obd_export *export = class_conn2export(conn);
 415         struct lov_obd *lov;
 416         ENTRY;
 417
 418         if (!md) {
 419                 CERROR("LOV requires striping ea for desctruction\n");
 420                 RETURN(-EINVAL);
 421         }
 422
 423         if (!export || !export->exp_obd)
 424                 RETURN(-ENODEV);
 425
 426         lov = &export->exp_obd->u.lov;
 427         for (i = 0; i < md->lmd_stripe_count; i++) {
 428                 /* create data objects with "parent" OA */
 429                 memcpy(&tmp, oa, sizeof(tmp));
 430                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 431
 432                 rc = obd_close(&lov->tgts[i].conn, &tmp, NULL);
 433                 if (rc) {
 434                         CERROR("Error close object %Ld on %d\n",
 435                                oa->o_id, i);
 436                 }
 437         }
 438         RETURN(rc);
 439 }
 440
 441 #ifndef log2
 442 #define log2(n) ffz(~(n))
 443 #endif
 444
 445 /* compute offset in stripe i corresponding to offset "in" */
 446 __u64 lov_offset(struct lov_stripe_md *md, __u64 in, int i)
 447 {
 448         __u32 ssz = md->lmd_stripe_size;
 449         /* full stripes across all * stripe size */
 450         __u32 out = ( ((__u32)in) / (md->lmd_stripe_count * ssz)) * ssz;
 451         __u32 off = (__u32)in % (md->lmd_stripe_count * ssz);
 452
 453         if ( in == 0xffffffffffffffff ) {
 454                 return 0xffffffffffffffff;
 455         }
 456
 457         if ( (i+1) * ssz <= off )
 458                 out += (i+1) * ssz;
 459         else if ( i * ssz > off )
 460                 out += 0;
 461         else
 462                 out += (off - (i * ssz)) % ssz;
 463
 464         return (__u64) out;
 465 }
 466
 467 /* compute offset in stripe i corresponding to offset "in" */
 468 __u64 lov_stripe(struct lov_stripe_md *md, __u64 in, int *j)
 469 {
 470         __u32 ssz = md->lmd_stripe_size;
 471         __u32 off, out;
 472         /* full stripes across all * stripe size */
 473         *j = (((__u32) in)/ssz) % md->lmd_stripe_count;
 474         off =  (__u32)in % (md->lmd_stripe_count * ssz);
 475         out = ( ((__u32)in) / (md->lmd_stripe_count * ssz)) * ssz +
 476                 (off - ((*j) * ssz)) % ssz;;
 477
 478         return (__u64) out;
 479 }
 480
 481 int lov_stripe_which(struct lov_stripe_md *md, __u64 in)
 482 {
 483         __u32 ssz = md->lmd_stripe_size;
 484         int j;
 485         j = (((__u32) in)/ssz) % md->lmd_stripe_count;
 486         return j;
 487 }
 488
 489
 490 /* FIXME: maybe we'll just make one node the authoritative attribute node, then
 491  * we can send this 'punch' to just the authoritative node and the nodes
 492  * that the punch will affect. */
 493 static int lov_punch(struct lustre_handle *conn, struct obdo *oa,
 494                      struct lov_stripe_md *md,
 495                      obd_off start, obd_off end)
 496 {
 497         int rc = 0, i;
 498         struct obdo tmp;
 499         struct obd_export *export = class_conn2export(conn);
 500         struct lov_obd *lov;
 501         ENTRY;
 502
 503         if (!md) {
 504                 CERROR("LOV requires striping ea for desctruction\n");
 505                 RETURN(-EINVAL);
 506         }
 507
 508         if (!export || !export->exp_obd)
 509                 RETURN(-ENODEV);
 510
 511         lov = &export->exp_obd->u.lov;
 512         for (i = 0; i < md->lmd_stripe_count; i++) {
 513                 __u64 starti = lov_offset(md, start, i);
 514                 __u64 endi = lov_offset(md, end, i);
 515
 516                 if (starti == endi)
 517                         continue;
 518                 /* create data objects with "parent" OA */
 519                 memcpy(&tmp, oa, sizeof(tmp));
 520                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 521
 522                 rc = obd_punch(&lov->tgts[i].conn, &tmp, NULL,
 523                                starti, endi);
 524                 if (!rc) {
 525                         CERROR("Error punch object %Ld on %d\n",
 526                                oa->o_id, i);
 527                 }
 528         }
 529         RETURN(rc);
 530 }
 531
 532 int lov_osc_brw_callback(struct io_cb_data *cbd, int err, int phase)
 533 {
 534         int ret = 0;
 535         ENTRY;
 536
 537         if (phase == CB_PHASE_START)
 538                 RETURN(0);
 539
 540         if (phase == CB_PHASE_FINISH) {
 541                 if (err)
 542                         cbd->err = err;
 543                 if (atomic_dec_and_test(&cbd->refcount))
 544                         ret = cbd->cb(cbd, cbd->err, phase);
 545                 RETURN(ret);
 546         }
 547
 548         LBUG();
 549         return 0;
 550 }
 551
 552 static inline int lov_brw(int cmd, struct lustre_handle *conn,
 553                           struct lov_stripe_md *md,
 554                           obd_count oa_bufs,
 555                           struct brw_page *pga,
 556                           brw_callback_t callback, struct io_cb_data *cbd)
 557 {
 558         int stripe_count = md->lmd_stripe_count;
 559         struct obd_export *export = class_conn2export(conn);
 560         struct lov_obd *lov;
 561         struct {
 562                 int bufct;
 563                 int index;
 564                 int subcount;
 565                 struct lov_stripe_md md;
 566         } *stripeinfo;
 567         struct brw_page *ioarr;
 568         int rc, i;
 569         ENTRY;
 570
 571         lov = &export->exp_obd->u.lov;
 572
 573
 574         OBD_ALLOC(stripeinfo,  stripe_count * sizeof(*stripeinfo));
 575         if (!stripeinfo)
 576                 RETURN(-ENOMEM);
 577
 578         OBD_ALLOC(ioarr, sizeof(*ioarr) * oa_bufs);
 579         if (!ioarr) {
 580                 OBD_FREE(stripeinfo, stripe_count * sizeof(*stripeinfo));
 581                 RETURN(-ENOMEM);
 582         }
 583
 584         for (i=0 ; i < oa_bufs ; i++ ) {
 585                 int which;
 586                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 587                 stripeinfo[which].bufct++;
 588         }
 589
 590         for (i=0 ; i < stripe_count ; i++) {
 591                 if (i>0)
 592                         stripeinfo[i].index =
 593                                 stripeinfo[i-1].index + stripeinfo[i-1].bufct;
 594                 stripeinfo[i].md.lmd_object_id =
 595                         md->lmd_oinfo[i].loi_id;
 596         }
 597
 598         for (i=0 ; i < oa_bufs ; i++ ) {
 599                 int which, shift;
 600                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 601
 602                 shift = stripeinfo[which].index;
 603                 ioarr[shift + stripeinfo[which].subcount] = pga[i];
 604                 ioarr[shift + stripeinfo[which].subcount].off =
 605                         lov_offset(md, pga[i].pg->index * PAGE_SIZE, which);
 606                 stripeinfo[which].subcount++;
 607         }
 608
 609         cbd->cb = callback;
 610         atomic_set(&cbd->refcount, oa_bufs);
 611         for (i=0 ; i < stripe_count ; i++) {
 612                 int shift = stripeinfo[i].index;
 613                 if (stripeinfo[i].bufct)
 614                         obd_brw(cmd, &lov->tgts[i].conn, &stripeinfo[i].md,
 615                                 stripeinfo[i].bufct, &ioarr[shift],
 616                                 lov_osc_brw_callback, cbd);
 617         }
 618
 619         rc = callback(cbd, 0, CB_PHASE_START);
 620
 621         RETURN(rc);
 622 }
 623
 624 static int lov_enqueue(struct lustre_handle *conn, struct lov_stripe_md *md,
 625                        struct lustre_handle *parent_lock,
 626                        __u32 type, void *cookie, int cookielen, __u32 mode,
 627                        int *flags, void *cb, void *data, int datalen,
 628                        struct lustre_handle *lockhs)
 629 {
 630         int rc = 0, i;
 631         struct obd_export *export = class_conn2export(conn);
 632         struct lov_obd *lov;
 633         struct lov_stripe_md submd;
 634         ENTRY;
 635
 636         if (!md) {
 637                 CERROR("LOV requires striping ea for desctruction\n");
 638                 RETURN(-EINVAL);
 639         }
 640
 641         if (!export || !export->exp_obd)
 642                 RETURN(-ENODEV);
 643
 644         lov = &export->exp_obd->u.lov;
 645         for (i = 0; i < md->lmd_stripe_count; i++) {
 646                 struct ldlm_extent *extent = (struct ldlm_extent *)cookie;
 647                 struct ldlm_extent sub_ext;
 648
 649                 sub_ext.start = lov_offset(md, extent->start, i);
 650                 sub_ext.end = lov_offset(md, extent->end, i);
 651                 if ( sub_ext.start == sub_ext.end )
 652                         continue;
 653
 654                 submd.lmd_object_id = md->lmd_oinfo[i].loi_id;
 655                 submd.lmd_easize = sizeof(struct lov_mds_md);
 656                 submd.lmd_stripe_count = md->lmd_stripe_count;
 657                 /* XXX submd is not fully initialized here */
 658                 rc = obd_enqueue(&(lov->tgts[i].conn), &submd, parent_lock,
 659                                  type, &sub_ext, sizeof(sub_ext), mode,
 660                                  flags, cb, data, datalen, &(lockhs[i]));
 661                 // XXX add a lock debug statement here
 662                 if (rc) {
 663                         CERROR("Error obd_enqueue object %Ld subobj %Ld\n",
 664                                md->lmd_object_id, md->lmd_oinfo[i].loi_id);
 665                 }
 666         }
 667         RETURN(rc);
 668 }
 669
 670 static int lov_cancel(struct lustre_handle *conn, struct lov_stripe_md *md, __u32 mode,
 671                       struct lustre_handle *lockhs)
 672 {
 673         int rc = 0, i;
 674         struct obd_export *export = class_conn2export(conn);
 675         struct lov_obd *lov;
 676         ENTRY;
 677
 678         if (!md) {
 679                 CERROR("LOV requires striping ea for lock cancellation\n");
 680                 RETURN(-EINVAL);
 681         }
 682
 683         if (!export || !export->exp_obd)
 684                 RETURN(-ENODEV);
 685
 686         lov = &export->exp_obd->u.lov;
 687         for (i = 0; i < md->lmd_stripe_count; i++) {
 688                 struct lov_stripe_md submd;
 689
 690                 if ( lockhs[i].addr == 0 )
 691                         continue;
 692
 693                 submd.lmd_object_id = md->lmd_oinfo[i].loi_id;
 694                 submd.lmd_easize = sizeof(struct lov_mds_md);
 695                 rc = obd_cancel(&lov->tgts[i].conn, &submd, mode, &lockhs[i]);
 696                 if (rc) {
 697                         CERROR("Error cancel object %Ld subobj %Ld\n",
 698                                md->lmd_object_id, md->lmd_oinfo[i].loi_id);
 699                 }
 700         }
 701         RETURN(rc);
 702 }
 703
 704
 705
 706
 707 struct obd_ops lov_obd_ops = {
 708         o_setup:       lov_setup,
 709         o_connect:     lov_connect,
 710         o_disconnect:  lov_disconnect,
 711         o_create:      lov_create,
 712         o_destroy:     lov_destroy,
 713         o_getattr:     lov_getattr,
 714         o_setattr:     lov_setattr,
 715         o_open:        lov_open,
 716         o_close:       lov_close,
 717         o_brw:         lov_brw,
 718         o_punch:       lov_punch,
 719         o_enqueue:     lov_enqueue,
 720         o_cancel:      lov_cancel
 721 };
 722
 723
 724 #define LOV_VERSION "v0.1"
 725
 726 static int __init lov_init(void)
 727 {
 728         printk(KERN_INFO "Lustre Logical Object Volume driver " LOV_VERSION
 729                ", info@clusterfs.com\n");
 730         return class_register_type(&lov_obd_ops, OBD_LOV_DEVICENAME);
 731 }
 732
 733 static void __exit lov_exit(void)
 734 {
 735         class_unregister_type(OBD_LOV_DEVICENAME);
 736 }
 737
 738 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 739 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver v0.1");
 740 MODULE_LICENSE("GPL");
 741
 742 module_init(lov_init);
 743 module_exit(lov_exit);