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         for (i = 0; i < md->lmd_stripe_count; i++) {
 317                 /* create data objects with "parent" OA */
 318                 memcpy(&tmp, oa, sizeof(tmp));
 319                 oa->o_id = md->lmd_oinfo[i].loi_id;
 320
 321                 rc = obd_getattr(&lov->tgts[i].conn, &tmp, NULL);
 322                 if (!rc) {
 323                         CERROR("Error getattr object %Ld on %d\n",
 324                                oa->o_id, i);
 325                 }
 326                 /* XXX can do something more sophisticated here... */
 327                 if (i == 0 ) {
 328                         obd_id id = oa->o_id;
 329                         memcpy(oa, &tmp, sizeof(tmp));
 330                         oa->o_id = id;
 331                 } else {
 332                         oa->o_size += tmp.o_size;
 333                 }
 334
 335         }
 336         RETURN(rc);
 337 }
 338
 339 static int lov_setattr(struct lustre_handle *conn, struct obdo *oa,
 340                        struct lov_stripe_md *md)
 341 {
 342         int rc = 0, i;
 343         struct obdo tmp;
 344         struct obd_export *export = class_conn2export(conn);
 345         struct lov_obd *lov;
 346         ENTRY;
 347
 348         if (!md) {
 349                 CERROR("LOV requires striping ea for desctruction\n");
 350                 RETURN(-EINVAL);
 351         }
 352
 353         if (!export || !export->exp_obd)
 354                 RETURN(-ENODEV);
 355
 356         lov = &export->exp_obd->u.lov;
 357         for (i = 0; i < md->lmd_stripe_count; i++) {
 358                 /* create data objects with "parent" OA */
 359                 memcpy(&tmp, oa, sizeof(tmp));
 360                 oa->o_id = md->lmd_oinfo[i].loi_id;
 361
 362                 rc = obd_setattr(&lov->tgts[i].conn, &tmp, NULL);
 363                 if (!rc) {
 364                         CERROR("Error setattr object %Ld on %d\n",
 365                                oa->o_id, i);
 366                 }
 367         }
 368         RETURN(rc);
 369 }
 370
 371 static int lov_open(struct lustre_handle *conn, struct obdo *oa,
 372                     struct lov_stripe_md *md)
 373 {
 374         int rc = 0, rc2 = 0, i;
 375         struct obdo tmp;
 376         struct obd_export *export = class_conn2export(conn);
 377         struct lov_obd *lov;
 378         ENTRY;
 379
 380         if (!md) {
 381                 CERROR("LOV requires striping ea for opening\n");
 382                 RETURN(-EINVAL);
 383         }
 384
 385         if (!export || !export->exp_obd)
 386                 RETURN(-ENODEV);
 387
 388         lov = &export->exp_obd->u.lov;
 389         for (i = 0; i < md->lmd_stripe_count; i++) {
 390                 /* create data objects with "parent" OA */
 391                 memcpy(&tmp, oa, sizeof(tmp));
 392                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 393
 394                 rc = obd_open(&lov->tgts[i].conn, &tmp, NULL);
 395                 if (rc) {
 396                         rc2 = rc;
 397                         CERROR("Error getattr object %Ld on %d\n",
 398                                oa->o_id, i);
 399                 }
 400         }
 401         RETURN(rc2);
 402 }
 403
 404
 405 static int lov_close(struct lustre_handle *conn, struct obdo *oa,
 406                      struct lov_stripe_md *md)
 407 {
 408         int rc = 0, i;
 409         struct obdo tmp;
 410         struct obd_export *export = class_conn2export(conn);
 411         struct lov_obd *lov;
 412         ENTRY;
 413
 414         if (!md) {
 415                 CERROR("LOV requires striping ea for desctruction\n");
 416                 RETURN(-EINVAL);
 417         }
 418
 419         if (!export || !export->exp_obd)
 420                 RETURN(-ENODEV);
 421
 422         lov = &export->exp_obd->u.lov;
 423         for (i = 0; i < md->lmd_stripe_count; i++) {
 424                 /* create data objects with "parent" OA */
 425                 memcpy(&tmp, oa, sizeof(tmp));
 426                 tmp.o_id = md->lmd_oinfo[i].loi_id;
 427
 428                 rc = obd_close(&lov->tgts[i].conn, &tmp, NULL);
 429                 if (!rc) {
 430                         CERROR("Error getattr object %Ld on %d\n",
 431                                oa->o_id, i);
 432                 }
 433         }
 434         RETURN(rc);
 435 }
 436
 437 #ifndef log2
 438 #define log2(n) ffz(~(n))
 439 #endif
 440
 441 /* compute offset in stripe i corresponds to offset "in" */
 442 __u64 lov_offset(struct lov_stripe_md *md, __u64 in, int i)
 443 {
 444         __u32 ssz = md->lmd_stripe_size;
 445         /* full stripes across all * stripe size */
 446         __u32 out = ( ((__u32)in) / (md->lmd_stripe_count * ssz)) * ssz;
 447         __u32 off = (__u32)in % (md->lmd_stripe_count * ssz);
 448
 449         if ( in == 0xffffffffffffffff ) {
 450                 return 0xffffffffffffffff;
 451         }
 452
 453         if ( (i+1) * ssz <= off )
 454                 out += (i+1) * ssz;
 455         else if ( i * ssz > off )
 456                 out += 0;
 457         else
 458                 out += (off - (i * ssz)) % ssz;
 459
 460         return (__u64) out;
 461 }
 462
 463 /* compute offset in stripe i corresponds to offset "in" */
 464 __u64 lov_stripe(struct lov_stripe_md *md, __u64 in, int *j)
 465 {
 466         __u32 ssz = md->lmd_stripe_size;
 467         __u32 off, out;
 468         /* full stripes across all * stripe size */
 469         *j = (((__u32) in)/ssz) % md->lmd_stripe_count;
 470         off =  (__u32)in % (md->lmd_stripe_count * ssz);
 471         out = ( ((__u32)in) / (md->lmd_stripe_count * ssz)) * ssz +
 472                 (off - ((*j) * ssz)) % ssz;;
 473
 474         return (__u64) out;
 475 }
 476
 477 int lov_stripe_which(struct lov_stripe_md *md, __u64 in)
 478 {
 479         __u32 ssz = md->lmd_stripe_size;
 480         int j;
 481         j = (((__u32) in)/ssz) % md->lmd_stripe_count;
 482         return j;
 483 }
 484
 485
 486 /* FIXME: maybe we'll just make one node the authoritative attribute node, then
 487  * we can send this 'punch' to just the authoritative node and the nodes
 488  * that the punch will affect. */
 489 static int lov_punch(struct lustre_handle *conn, struct obdo *oa,
 490                      struct lov_stripe_md *md,
 491                      obd_off start, obd_off end)
 492 {
 493         int rc = 0, i;
 494         struct obdo tmp;
 495         struct obd_export *export = class_conn2export(conn);
 496         struct lov_obd *lov;
 497         ENTRY;
 498
 499         if (!md) {
 500                 CERROR("LOV requires striping ea for desctruction\n");
 501                 RETURN(-EINVAL);
 502         }
 503
 504         if (!export || !export->exp_obd)
 505                 RETURN(-ENODEV);
 506
 507         lov = &export->exp_obd->u.lov;
 508         for (i = 0; i < md->lmd_stripe_count; i++) {
 509                 __u64 starti = lov_offset(md, start, i);
 510                 __u64 endi = lov_offset(md, end, i);
 511
 512                 if (starti == endi)
 513                         continue;
 514                 /* create data objects with "parent" OA */
 515                 memcpy(&tmp, oa, sizeof(tmp));
 516                 oa->o_id = md->lmd_oinfo[i].loi_id;
 517
 518                 rc = obd_punch(&lov->tgts[i].conn, &tmp, NULL,
 519                                starti, endi);
 520                 if (!rc) {
 521                         CERROR("Error punch object %Ld on %d\n",
 522                                oa->o_id, i);
 523                 }
 524         }
 525         RETURN(rc);
 526 }
 527
 528 int lov_osc_brw_callback(struct io_cb_data *cbd, int err, int phase)
 529 {
 530         int ret = 0;
 531         ENTRY;
 532
 533         if (phase == CB_PHASE_START)
 534                 RETURN(0);
 535
 536         if (phase == CB_PHASE_FINISH) {
 537                 if (err)
 538                         cbd->err = err;
 539                 if (atomic_dec_and_test(&cbd->refcount))
 540                         ret = cbd->cb(cbd, cbd->err, phase);
 541                 RETURN(ret);
 542         }
 543
 544         LBUG();
 545         return 0;
 546 }
 547
 548 static inline int lov_brw(int cmd, struct lustre_handle *conn,
 549                           struct lov_stripe_md *md,
 550                           obd_count oa_bufs,
 551                           struct brw_page *pga,
 552                           brw_callback_t callback, struct io_cb_data *cbd)
 553 {
 554         int stripe_count = md->lmd_stripe_count;
 555         struct obd_export *export = class_conn2export(conn);
 556         struct lov_obd *lov;
 557         struct {
 558                 int bufct;
 559                 int index;
 560                 int subcount;
 561                 struct lov_stripe_md md;
 562         } *stripeinfo;
 563         struct brw_page *ioarr;
 564         int rc, i;
 565         ENTRY;
 566
 567         lov = &export->exp_obd->u.lov;
 568
 569
 570         OBD_ALLOC(stripeinfo,  stripe_count * sizeof(*stripeinfo));
 571         if (!stripeinfo)
 572                 RETURN(-ENOMEM);
 573
 574         OBD_ALLOC(ioarr, sizeof(*ioarr) * oa_bufs);
 575         if (!ioarr) {
 576                 OBD_FREE(stripeinfo, stripe_count * sizeof(*stripeinfo));
 577                 RETURN(-ENOMEM);
 578         }
 579
 580         for (i=0 ; i < oa_bufs ; i++ ) {
 581                 int which;
 582                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 583                 stripeinfo[which].bufct++;
 584         }
 585
 586         for (i=0 ; i < stripe_count ; i++) {
 587                 if (i>0)
 588                         stripeinfo[i].index =
 589                                 stripeinfo[i-1].index + stripeinfo[i-1].bufct;
 590                 stripeinfo[i].md.lmd_object_id =
 591                         md->lmd_oinfo[i].loi_id;
 592         }
 593
 594         for (i=0 ; i < oa_bufs ; i++ ) {
 595                 int which, shift;
 596                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 597
 598                 shift = stripeinfo[which].index;
 599                 ioarr[shift + stripeinfo[which].subcount] = pga[i];
 600                 ioarr[shift + stripeinfo[which].subcount].off = lov_offset(md, pga[i].pg->index * PAGE_SIZE, which);
 601                 stripeinfo[which].subcount++;
 602         }
 603
 604         cbd->cb = callback;
 605         atomic_set(&cbd->refcount, oa_bufs);
 606         for (i=0 ; i < stripe_count ; i++) {
 607                 int shift = stripeinfo[i].index;
 608                 if (stripeinfo[i].bufct)
 609                         obd_brw(cmd, &lov->tgts[i].conn, &stripeinfo[i].md,
 610                                 stripeinfo[i].bufct, &ioarr[shift],
 611                                 lov_osc_brw_callback, cbd);
 612         }
 613
 614         rc = callback(cbd, 0, CB_PHASE_START);
 615
 616         RETURN(rc);
 617 }
 618
 619 static int lov_enqueue(struct lustre_handle *conn, struct lov_stripe_md *md,
 620                        struct lustre_handle *parent_lock,
 621                        __u32 type, void *cookie, int cookielen, __u32 mode,
 622                        int *flags, void *cb, void *data, int datalen,
 623                        struct lustre_handle *lockhs)
 624 {
 625         int rc = 0, i;
 626         struct obd_export *export = class_conn2export(conn);
 627         struct lov_obd *lov;
 628         struct lov_stripe_md submd;
 629         ENTRY;
 630
 631         if (!md) {
 632                 CERROR("LOV requires striping ea for desctruction\n");
 633                 RETURN(-EINVAL);
 634         }
 635
 636         if (!export || !export->exp_obd)
 637                 RETURN(-ENODEV);
 638
 639         lov = &export->exp_obd->u.lov;
 640         for (i = 0; i < md->lmd_stripe_count; i++) {
 641                 struct ldlm_extent *extent = (struct ldlm_extent *)cookie;
 642                 struct ldlm_extent sub_ext;
 643
 644                 sub_ext.start = lov_offset(md, extent->start, i);
 645                 sub_ext.end = lov_offset(md, extent->end, i);
 646                 if ( sub_ext.start == sub_ext.end )
 647                         continue;
 648
 649                 submd.lmd_object_id = md->lmd_oinfo[i].loi_id;
 650                 submd.lmd_easize = sizeof(submd);
 651                 submd.lmd_stripe_count = md->lmd_stripe_count;
 652                 /* XXX submd is not fully initialized here */
 653                 rc = obd_enqueue(&(lov->tgts[i].conn), &submd, parent_lock,  type,
 654                                  &sub_ext, sizeof(sub_ext), mode, flags, cb, data, datalen, &(lockhs[i]));
 655                 // XXX add a lock debug statement here
 656                 if (rc) {
 657                         CERROR("Error obd_enqueu object %Ld subobj %Ld\n", md->lmd_object_id,
 658                                md->lmd_oinfo[i].loi_id);
 659                 }
 660         }
 661         RETURN(rc);
 662 }
 663
 664 static int lov_cancel(struct lustre_handle *conn, struct lov_stripe_md *md, __u32 mode,
 665                       struct lustre_handle *lockhs)
 666 {
 667         int rc = 0, i;
 668         struct obd_export *export = class_conn2export(conn);
 669         struct lov_obd *lov;
 670         ENTRY;
 671
 672         if (!md) {
 673                 CERROR("LOV requires striping ea for lock cancellation\n");
 674                 RETURN(-EINVAL);
 675         }
 676
 677         if (!export || !export->exp_obd)
 678                 RETURN(-ENODEV);
 679
 680         lov = &export->exp_obd->u.lov;
 681         for (i = 0; i < md->lmd_stripe_count; i++) {
 682                 struct lov_stripe_md submd;
 683
 684                 if ( lockhs[i].addr == 0 )
 685                         continue;
 686
 687                 submd.lmd_object_id = md->lmd_oinfo[i].loi_id;
 688                 submd.lmd_easize = sizeof(submd);
 689                 rc = obd_cancel(&lov->tgts[i].conn, &submd, mode, &lockhs[i]);
 690                 if (!rc) {
 691                         CERROR("Error punch object %Ld subobj %Ld\n", md->lmd_object_id,
 692                                md->lmd_oinfo[i].loi_id);
 693                 }
 694         }
 695         RETURN(rc);
 696 }
 697
 698
 699
 700
 701 struct obd_ops lov_obd_ops = {
 702         o_setup:       lov_setup,
 703         o_connect:     lov_connect,
 704         o_disconnect:  lov_disconnect,
 705         o_create:      lov_create,
 706         o_destroy:     lov_destroy,
 707         o_getattr:     lov_getattr,
 708         o_setattr:     lov_setattr,
 709         o_open:        lov_open,
 710         o_close:       lov_close,
 711         o_brw:         lov_brw,
 712         o_punch:       lov_punch,
 713         o_enqueue:     lov_enqueue,
 714         o_cancel:      lov_cancel
 715 };
 716
 717
 718 #define LOV_VERSION "v0.1"
 719
 720 static int __init lov_init(void)
 721 {
 722         printk(KERN_INFO "Lustre Logical Object Volume driver " LOV_VERSION
 723                ", info@clusterfs.com\n");
 724         return class_register_type(&lov_obd_ops, OBD_LOV_DEVICENAME);
 725 }
 726
 727 static void __exit lov_exit(void)
 728 {
 729         class_unregister_type(OBD_LOV_DEVICENAME);
 730 }
 731
 732 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 733 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver v0.1");
 734 MODULE_LICENSE("GPL");
 735
 736 module_init(lov_init);
 737 module_exit(lov_exit);