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_object_id);
 188         return size;
 189 }
 190
 191 /* the LOV counts on oa->o_id to be set as the LOV object id */
 192 static int lov_create(struct lustre_handle *conn, struct obdo *oa, struct lov_stripe_md **ea)
 193 {
 194         int rc = 0, i;
 195         struct obdo tmp;
 196         struct obd_export *export = class_conn2export(conn);
 197         struct lov_obd *lov;
 198         struct lov_stripe_md *md;
 199         ENTRY;
 200
 201         if (!ea) {
 202                 CERROR("lov_create needs EA for striping information\n");
 203                 RETURN(-EINVAL);
 204         }
 205         if (!export)
 206                 RETURN(-EINVAL);
 207         lov = &export->exp_obd->u.lov;
 208
 209         oa->o_easize =  lov_stripe_md_size(export->exp_obd);
 210         if (!*ea) {
 211                 OBD_ALLOC(*ea, oa->o_easize);
 212                 if (! *ea)
 213                         RETURN(-ENOMEM);
 214         }
 215
 216         md = *ea;
 217         md->lmd_easize = oa->o_easize;
 218         md->lmd_object_id = oa->o_id;
 219         if (!md->lmd_stripe_count) {
 220                 md->lmd_stripe_count = lov->desc.ld_default_stripe_count;
 221         }
 222
 223         if (!md->lmd_stripe_size)
 224                 md->lmd_stripe_size = lov->desc.ld_default_stripe_size;
 225
 226
 227
 228         for (i = 0; i < md->lmd_stripe_count; i++) {
 229                 struct lov_stripe_md obj_md;
 230                 struct lov_stripe_md *obj_mdp = &obj_md;
 231                 /* create data objects with "parent" OA */
 232                 memcpy(&tmp, oa, sizeof(tmp));
 233                 tmp.o_easize = sizeof(struct lov_stripe_md);
 234                 rc = obd_create(&lov->tgts[i].conn, &tmp, &obj_mdp);
 235                 if (rc)
 236                         GOTO(out_cleanup, rc);
 237                 md->lmd_objects[i].l_object_id = tmp.o_id;
 238         }
 239
 240  out_cleanup:
 241         if (rc) {
 242                 int i2, rc2;
 243                 for (i2 = 0 ; i2 < i ; i2++) {
 244                         /* destroy already created objects here */
 245                         tmp.o_id = md->lmd_objects[i].l_object_id;
 246                         rc2 = obd_destroy(&lov->tgts[i].conn, &tmp, NULL);
 247                         if (rc2) {
 248                                 CERROR("Failed to remove object from target %d\n",
 249                                        i2);
 250                         }
 251                 }
 252         }
 253         return rc;
 254 }
 255
 256 static int lov_destroy(struct lustre_handle *conn, struct obdo *oa,
 257                        struct lov_stripe_md *md)
 258 {
 259         int rc = 0, i;
 260         struct obdo tmp;
 261         struct obd_export *export = class_conn2export(conn);
 262         struct lov_obd *lov;
 263         ENTRY;
 264
 265         if (!md) {
 266                 CERROR("LOV requires striping ea for desctruction\n");
 267                 RETURN(-EINVAL);
 268         }
 269
 270         if (!export || !export->exp_obd)
 271                 RETURN(-ENODEV);
 272
 273         lov = &export->exp_obd->u.lov;
 274         for (i = 0; i < md->lmd_stripe_count; i++) {
 275                 /* create data objects with "parent" OA */
 276                 memcpy(&tmp, oa, sizeof(tmp));
 277                 tmp.o_id = md->lmd_objects[i].l_object_id;
 278                 rc = obd_destroy(&lov->tgts[i].conn, &tmp, NULL);
 279                 if (!rc) {
 280                         CERROR("Error destroying object %Ld on %d\n",
 281                                oa->o_id, i);
 282                 }
 283         }
 284         RETURN(rc);
 285 }
 286
 287 static int lov_getattr(struct lustre_handle *conn, struct obdo *oa,
 288                        struct lov_stripe_md *md)
 289 {
 290         int rc = 0, i;
 291         struct obdo tmp;
 292         struct obd_export *export = class_conn2export(conn);
 293         struct lov_obd *lov;
 294         ENTRY;
 295
 296         if (!md) {
 297                 CERROR("LOV requires striping ea for desctruction\n");
 298                 RETURN(-EINVAL);
 299         }
 300
 301         if (!export || !export->exp_obd)
 302                 RETURN(-ENODEV);
 303
 304         lov = &export->exp_obd->u.lov;
 305         for (i = 0; i < md->lmd_stripe_count; i++) {
 306                 /* create data objects with "parent" OA */
 307                 memcpy(&tmp, oa, sizeof(tmp));
 308                 oa->o_id = md->lmd_objects[i].l_object_id;
 309
 310                 rc = obd_getattr(&lov->tgts[i].conn, &tmp, NULL);
 311                 if (!rc) {
 312                         CERROR("Error getattr object %Ld on %d\n",
 313                                oa->o_id, i);
 314                 }
 315                 /* XXX can do something more sophisticated here... */
 316                 if (i == 0 ) {
 317                         obd_id id = oa->o_id;
 318                         memcpy(oa, &tmp, sizeof(tmp));
 319                         oa->o_id = id;
 320                 }
 321         }
 322         RETURN(rc);
 323 }
 324
 325 static int lov_setattr(struct lustre_handle *conn, struct obdo *oa,
 326                        struct lov_stripe_md *md)
 327 {
 328         int rc = 0, i;
 329         struct obdo tmp;
 330         struct obd_export *export = class_conn2export(conn);
 331         struct lov_obd *lov;
 332         ENTRY;
 333
 334         if (!md) {
 335                 CERROR("LOV requires striping ea for desctruction\n");
 336                 RETURN(-EINVAL);
 337         }
 338
 339         if (!export || !export->exp_obd)
 340                 RETURN(-ENODEV);
 341
 342         lov = &export->exp_obd->u.lov;
 343         for (i = 0; i < md->lmd_stripe_count; i++) {
 344                 /* create data objects with "parent" OA */
 345                 memcpy(&tmp, oa, sizeof(tmp));
 346                 oa->o_id = md->lmd_objects[i].l_object_id;
 347
 348                 rc = obd_setattr(&lov->tgts[i].conn, &tmp, NULL);
 349                 if (!rc) {
 350                         CERROR("Error setattr object %Ld on %d\n",
 351                                oa->o_id, i);
 352                 }
 353         }
 354         RETURN(rc);
 355 }
 356
 357 static int lov_open(struct lustre_handle *conn, struct obdo *oa,
 358                     struct lov_stripe_md *md)
 359 {
 360         int rc = 0, rc2 = 0, i;
 361         struct obdo tmp;
 362         struct obd_export *export = class_conn2export(conn);
 363         struct lov_obd *lov;
 364         ENTRY;
 365
 366         if (!md) {
 367                 CERROR("LOV requires striping ea for opening\n");
 368                 RETURN(-EINVAL);
 369         }
 370
 371         if (!export || !export->exp_obd)
 372                 RETURN(-ENODEV);
 373
 374         lov = &export->exp_obd->u.lov;
 375         for (i = 0; i < md->lmd_stripe_count; i++) {
 376                 /* create data objects with "parent" OA */
 377                 memcpy(&tmp, oa, sizeof(tmp));
 378                 oa->o_id = md->lmd_objects[i].l_object_id;
 379
 380                 rc = obd_open(&lov->tgts[i].conn, &tmp, NULL);
 381                 if (rc) {
 382                         rc2 = rc;
 383                         CERROR("Error getattr object %Ld on %d\n",
 384                                oa->o_id, i);
 385                 }
 386         }
 387         RETURN(rc2);
 388 }
 389
 390
 391 static int lov_close(struct lustre_handle *conn, struct obdo *oa,
 392                      struct lov_stripe_md *md)
 393 {
 394         int rc = 0, i;
 395         struct obdo tmp;
 396         struct obd_export *export = class_conn2export(conn);
 397         struct lov_obd *lov;
 398         ENTRY;
 399
 400         if (!md) {
 401                 CERROR("LOV requires striping ea for desctruction\n");
 402                 RETURN(-EINVAL);
 403         }
 404
 405         if (!export || !export->exp_obd)
 406                 RETURN(-ENODEV);
 407
 408         lov = &export->exp_obd->u.lov;
 409         for (i = 0; i < md->lmd_stripe_count; i++) {
 410                 /* create data objects with "parent" OA */
 411                 memcpy(&tmp, oa, sizeof(tmp));
 412                 oa->o_id = md->lmd_objects[i].l_object_id;
 413
 414                 rc = obd_close(&lov->tgts[i].conn, &tmp, NULL);
 415                 if (!rc) {
 416                         CERROR("Error getattr object %Ld on %d\n",
 417                                oa->o_id, i);
 418                 }
 419         }
 420         RETURN(rc);
 421 }
 422
 423 #ifndef log2
 424 #define log2(n) ffz(~(n))
 425 #endif
 426
 427 /* compute offset in stripe i corresponds to offset "in" */
 428 __u64 lov_offset(struct lov_stripe_md *md, __u64 in, int i)
 429 {
 430         __u32 ssz = md->lmd_stripe_size;
 431         /* full stripes across all * stripe size */
 432         __u32 out = ( ((__u32)in) / (md->lmd_stripe_count * ssz)) * ssz;
 433         __u32 off = (__u32)in % (md->lmd_stripe_count * ssz);
 434
 435         if ( in == 0xffffffffffffffff ) {
 436                 return 0xffffffffffffffff;
 437         }
 438
 439         if ( (i+1) * ssz < off )
 440                 out += ssz;
 441         else if ( i * ssz > off )
 442                 out += 0;
 443         else
 444                 out += (off - (i * ssz)) % ssz;
 445
 446         return (__u64) out;
 447 }
 448
 449 /* compute offset in stripe i corresponds to offset "in" */
 450 __u64 lov_stripe(struct lov_stripe_md *md, __u64 in, int *j)
 451 {
 452         __u32 ssz = md->lmd_stripe_size;
 453         __u32 off, out;
 454         /* full stripes across all * stripe size */
 455         *j = (((__u32) in)/ssz) % md->lmd_stripe_count;
 456         off =  (__u32)in % (md->lmd_stripe_count * ssz);
 457         out = ( ((__u32)in) / (md->lmd_stripe_count * ssz)) * ssz +
 458                 (off - ((*j) * ssz)) % ssz;;
 459
 460         return (__u64) out;
 461 }
 462
 463 int lov_stripe_which(struct lov_stripe_md *md, __u64 in)
 464 {
 465         __u32 ssz = md->lmd_stripe_size;
 466         int j;
 467         j = (((__u32) in)/ssz) % md->lmd_stripe_count;
 468         return j;
 469 }
 470
 471
 472 /* FIXME: maybe we'll just make one node the authoritative attribute node, then
 473  * we can send this 'punch' to just the authoritative node and the nodes
 474  * that the punch will affect. */
 475 static int lov_punch(struct lustre_handle *conn, struct obdo *oa,
 476                      struct lov_stripe_md *md,
 477                      obd_off start, obd_off end)
 478 {
 479         int rc = 0, i;
 480         struct obdo tmp;
 481         struct obd_export *export = class_conn2export(conn);
 482         struct lov_obd *lov;
 483         ENTRY;
 484
 485         if (!md) {
 486                 CERROR("LOV requires striping ea for desctruction\n");
 487                 RETURN(-EINVAL);
 488         }
 489
 490         if (!export || !export->exp_obd)
 491                 RETURN(-ENODEV);
 492
 493         lov = &export->exp_obd->u.lov;
 494         for (i = 0; i < md->lmd_stripe_count; i++) {
 495                 __u64 starti = lov_offset(md, start, i);
 496                 __u64 endi = lov_offset(md, end, i);
 497
 498                 if (starti == endi)
 499                         continue;
 500                 /* create data objects with "parent" OA */
 501                 memcpy(&tmp, oa, sizeof(tmp));
 502                 oa->o_id = md->lmd_objects[i].l_object_id;
 503
 504                 rc = obd_punch(&lov->tgts[i].conn, &tmp, NULL,
 505                                starti, endi);
 506                 if (!rc) {
 507                         CERROR("Error punch object %Ld on %d\n",
 508                                oa->o_id, i);
 509                 }
 510         }
 511         RETURN(rc);
 512 }
 513
 514 int lov_osc_brw_callback(struct io_cb_data *cbd, int err, int phase)
 515 {
 516         int ret = 0;
 517         ENTRY;
 518
 519         if (phase == CB_PHASE_START)
 520                 RETURN(0);
 521
 522         if (phase == CB_PHASE_FINISH) {
 523                 if (err)
 524                         cbd->err = err;
 525                 if (atomic_dec_and_test(&cbd->refcount))
 526                         ret = cbd->cb(cbd, cbd->err, phase);
 527                 RETURN(ret);
 528         }
 529
 530         LBUG();
 531         return 0;
 532 }
 533
 534 static inline int lov_brw(int cmd, struct lustre_handle *conn,
 535                           struct lov_stripe_md *md,
 536                           obd_count oa_bufs,
 537                           struct brw_page *pga,
 538                           brw_callback_t callback, struct io_cb_data *cbd)
 539 {
 540         int stripe_count = md->lmd_stripe_count;
 541         struct obd_export *export = class_conn2export(conn);
 542         struct lov_obd *lov;
 543         struct {
 544                 int bufct;
 545                 int index;
 546                 int subcount;
 547                 struct lov_stripe_md md;
 548         } *stripeinfo;
 549         struct brw_page *ioarr;
 550         int rc, i;
 551         ENTRY;
 552
 553         lov = &export->exp_obd->u.lov;
 554
 555
 556         OBD_ALLOC(stripeinfo,  stripe_count * sizeof(*stripeinfo));
 557         if (!stripeinfo)
 558                 RETURN(-ENOMEM);
 559
 560         OBD_ALLOC(ioarr, sizeof(*ioarr) * oa_bufs);
 561         if (!ioarr) {
 562                 OBD_FREE(stripeinfo, stripe_count * sizeof(*stripeinfo));
 563                 RETURN(-ENOMEM);
 564         }
 565
 566         for (i=0 ; i < oa_bufs ; i++ ) {
 567                 int which;
 568                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 569                 stripeinfo[which].bufct++;
 570         }
 571
 572         for (i=0 ; i < stripe_count ; i++) {
 573                 if (i>0)
 574                         stripeinfo[i].index =
 575                                 stripeinfo[i-1].index + stripeinfo[i-1].bufct;
 576                 stripeinfo[i].md.lmd_object_id =
 577                         md->lmd_objects[i].l_object_id;
 578         }
 579
 580         for (i=0 ; i < oa_bufs ; i++ ) {
 581                 int which, shift;
 582                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 583
 584                 shift = stripeinfo[which].index;
 585                 ioarr[shift + stripeinfo[which].subcount] = pga[i];
 586                 pga[i].off = lov_offset(md, pga[i].pg->index * PAGE_SIZE, which);
 587                 stripeinfo[which].subcount++;
 588         }
 589
 590         cbd->cb = callback;
 591         atomic_set(&cbd->refcount, oa_bufs);
 592         for (i=0 ; i < stripe_count ; i++) {
 593                 int shift = stripeinfo[i].index;
 594                 if (stripeinfo[i].bufct)
 595                         obd_brw(cmd, &lov->tgts[i].conn, &stripeinfo[i].md,
 596                                 stripeinfo[i].bufct, &ioarr[shift],
 597                                 lov_osc_brw_callback, cbd);
 598         }
 599
 600         rc = callback(cbd, 0, CB_PHASE_START);
 601
 602         RETURN(rc);
 603 }
 604
 605 static int lov_enqueue(struct lustre_handle *conn, struct lov_stripe_md *md,
 606                        struct lustre_handle *parent_lock,
 607                        __u32 type, void *cookie, int cookielen, __u32 mode,
 608                        int *flags, void *cb, void *data, int datalen,
 609                        struct lustre_handle *lockhs)
 610 {
 611         int rc = 0, i;
 612         struct obd_export *export = class_conn2export(conn);
 613         struct lov_obd *lov;
 614         ENTRY;
 615
 616         if (!md) {
 617                 CERROR("LOV requires striping ea for desctruction\n");
 618                 RETURN(-EINVAL);
 619         }
 620
 621         if (!export || !export->exp_obd)
 622                 RETURN(-ENODEV);
 623
 624         lov = &export->exp_obd->u.lov;
 625         for (i = 0; i < md->lmd_stripe_count; i++) {
 626                 struct ldlm_extent *extent = (struct ldlm_extent *)cookie;
 627                 struct ldlm_extent sub_ext;
 628                 struct lov_stripe_md submd;
 629
 630                 sub_ext.start = lov_offset(md, extent->start, i);
 631                 sub_ext.end = lov_offset(md, extent->end, i);
 632                 if ( sub_ext.start == sub_ext.end )
 633                         continue;
 634
 635                 submd.lmd_object_id = md->lmd_objects[i].l_object_id;
 636                 submd.lmd_easize = sizeof(submd);
 637                 rc = obd_enqueue(&(lov->tgts[i].conn), &submd, parent_lock,  type,
 638                                  &sub_ext, sizeof(sub_ext), mode, flags, cb, data, datalen, &(lockhs[i]));
 639                 // XXX add a lock debug statement here
 640                 if (rc) {
 641                         CERROR("Error obd_enqueu object %Ld subobj %Ld\n", md->lmd_object_id,
 642                                md->lmd_objects[i].l_object_id);
 643                 }
 644         }
 645         RETURN(rc);
 646 }
 647
 648 static int lov_cancel(struct lustre_handle *conn, struct lov_stripe_md *md, __u32 mode,
 649                       struct lustre_handle *lockhs)
 650 {
 651         int rc = 0, i;
 652         struct obd_export *export = class_conn2export(conn);
 653         struct lov_obd *lov;
 654         ENTRY;
 655
 656         if (!md) {
 657                 CERROR("LOV requires striping ea for lock cancellation\n");
 658                 RETURN(-EINVAL);
 659         }
 660
 661         if (!export || !export->exp_obd)
 662                 RETURN(-ENODEV);
 663
 664         lov = &export->exp_obd->u.lov;
 665         for (i = 0; i < md->lmd_stripe_count; i++) {
 666                 struct lov_stripe_md submd;
 667
 668                 if ( lockhs[i].addr == 0 )
 669                         continue;
 670
 671                 submd.lmd_object_id = md->lmd_objects[i].l_object_id;
 672                 submd.lmd_easize = sizeof(submd);
 673                 rc = obd_cancel(&lov->tgts[i].conn, &submd, mode, &lockhs[i]);
 674                 if (!rc) {
 675                         CERROR("Error punch object %Ld subobj %Ld\n", md->lmd_object_id,
 676                                md->lmd_objects[i].l_object_id);
 677                 }
 678         }
 679         RETURN(rc);
 680 }
 681
 682
 683
 684
 685 struct obd_ops lov_obd_ops = {
 686         o_setup:       lov_setup,
 687         o_connect:     lov_connect,
 688         o_disconnect:  lov_disconnect,
 689         o_create:      lov_create,
 690         o_destroy:     lov_destroy,
 691         o_getattr:     lov_getattr,
 692         o_setattr:     lov_setattr,
 693         o_open:        lov_open,
 694         o_close:       lov_close,
 695         o_brw:         lov_brw,
 696         o_punch:       lov_punch,
 697         o_enqueue:     lov_enqueue,
 698         o_cancel:      lov_cancel
 699 };
 700
 701
 702 #define LOV_VERSION "v0.1"
 703
 704 static int __init lov_init(void)
 705 {
 706         printk(KERN_INFO "Lustre Logical Object Volume driver " LOV_VERSION
 707                ", info@clusterfs.com\n");
 708         return class_register_type(&lov_obd_ops, OBD_LOV_DEVICENAME);
 709 }
 710
 711 static void __exit lov_exit(void)
 712 {
 713         class_unregister_type(OBD_LOV_DEVICENAME);
 714 }
 715
 716 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 717 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver v0.1");
 718 MODULE_LICENSE("GPL");
 719
 720 module_init(lov_init);
 721 module_exit(lov_exit);