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 struct lov_callback_data {
 515         struct io_cb_data cbd;
 516         brw_callback_t    cb;
 517 };
 518
 519 int lov_osc_brw_callback(struct io_cb_data *data, int err, int phase)
 520 {
 521         struct lov_callback_data *lovcbd = (struct lov_callback_data *)data;
 522         int ret = 0;
 523         ENTRY;
 524
 525         if (phase == CB_PHASE_START)
 526                 RETURN(0);
 527
 528         if (phase == CB_PHASE_FINISH) {
 529                 if (err)
 530                         lovcbd->cbd.err = err;
 531                 if (atomic_dec_and_test(&lovcbd->cbd.refcount))
 532                         ret = lovcbd->cb(&lovcbd->cbd, 0, lovcbd->cbd.err);
 533                 RETURN(ret);
 534         }
 535
 536         LBUG();
 537         return 0;
 538 }
 539
 540 static inline int lov_brw(int cmd, struct lustre_handle *conn,
 541                           struct lov_stripe_md *md,
 542                           obd_count oa_bufs,
 543                           struct brw_page *pga,
 544                           brw_callback_t callback, struct io_cb_data *data)
 545 {
 546         int stripe_count = md->lmd_stripe_count;
 547         struct obd_export *export = class_conn2export(conn);
 548         struct lov_obd *lov;
 549         struct {
 550                 int bufct;
 551                 int index;
 552                 int subcount;
 553                 struct lov_stripe_md md;
 554         } *stripeinfo;
 555         struct brw_page *ioarr;
 556         int rc, i;
 557         struct lov_callback_data *lov_cb_data;
 558         ENTRY;
 559
 560         lov = &export->exp_obd->u.lov;
 561
 562         OBD_ALLOC(lov_cb_data, sizeof(*lov_cb_data));
 563         if (!lov_cb_data)
 564                 RETURN(-ENOMEM);
 565
 566         OBD_ALLOC(stripeinfo,  stripe_count * sizeof(*stripeinfo));
 567         if (!stripeinfo)
 568                 RETURN(-ENOMEM);
 569
 570         OBD_ALLOC(ioarr, sizeof(*ioarr) * oa_bufs);
 571         if (!ioarr) {
 572                 OBD_FREE(stripeinfo, stripe_count * sizeof(*stripeinfo));
 573                 RETURN(-ENOMEM);
 574         }
 575
 576         for (i=0 ; i < oa_bufs ; i++ ) {
 577                 int which;
 578                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 579                 stripeinfo[which].bufct++;
 580         }
 581
 582         for (i=0 ; i < stripe_count ; i++) {
 583                 if (i>0)
 584                         stripeinfo[i].index =
 585                                 stripeinfo[i-1].index + stripeinfo[i-1].bufct;
 586                 stripeinfo[i].md.lmd_object_id =
 587                         md->lmd_objects[i].l_object_id;
 588         }
 589
 590         for (i=0 ; i < oa_bufs ; i++ ) {
 591                 int which, shift;
 592                 which = lov_stripe_which(md, pga[i].pg->index * PAGE_SIZE);
 593
 594                 shift = stripeinfo[which].index;
 595                 ioarr[shift + stripeinfo[which].subcount] = pga[i];
 596                 pga[i].off = lov_offset(md, pga[i].pg->index * PAGE_SIZE, which);
 597                 stripeinfo[which].subcount++;
 598         }
 599
 600         lov_cb_data->cbd = *data;
 601         lov_cb_data->cb = callback;
 602         atomic_set(&lov_cb_data->cbd.refcount, oa_bufs);
 603         for (i=0 ; i < stripe_count ; i++) {
 604                 int shift = stripeinfo[i].index;
 605
 606                 obd_brw(cmd, &lov->tgts[i].conn, &stripeinfo[i].md,
 607                         stripeinfo[i].bufct, &ioarr[shift],
 608                         lov_osc_brw_callback, (struct io_cb_data *)lov_cb_data);
 609         }
 610
 611         rc = callback((struct io_cb_data *)lov_cb_data, 0, CB_PHASE_START);
 612
 613         RETURN(rc);
 614 }
 615
 616 static int lov_enqueue(struct lustre_handle *conn, struct lov_stripe_md *md,
 617                        struct lustre_handle *parent_lock,
 618                        __u32 type, void *cookie, int cookielen, __u32 mode,
 619                        int *flags, void *cb, void *data, int datalen,
 620                        struct lustre_handle *lockhs)
 621 {
 622         int rc = 0, i;
 623         struct obd_export *export = class_conn2export(conn);
 624         struct lov_obd *lov;
 625         ENTRY;
 626
 627         if (!md) {
 628                 CERROR("LOV requires striping ea for desctruction\n");
 629                 RETURN(-EINVAL);
 630         }
 631
 632         if (!export || !export->exp_obd)
 633                 RETURN(-ENODEV);
 634
 635         lov = &export->exp_obd->u.lov;
 636         for (i = 0; i < md->lmd_stripe_count; i++) {
 637                 struct ldlm_extent *extent = (struct ldlm_extent *)cookie;
 638                 struct ldlm_extent sub_ext;
 639                 struct lov_stripe_md submd;
 640
 641                 sub_ext.start = lov_offset(md, extent->start, i);
 642                 sub_ext.end = lov_offset(md, extent->end, i);
 643                 if ( sub_ext.start == sub_ext.end )
 644                         continue;
 645
 646                 submd.lmd_object_id = md->lmd_objects[i].l_object_id;
 647                 submd.lmd_easize = sizeof(submd);
 648                 rc = obd_enqueue(&(lov->tgts[i].conn), &submd, parent_lock,  type,
 649                                  &sub_ext, sizeof(sub_ext), mode, flags, cb, data, datalen, &(lockhs[i]));
 650                 // XXX add a lock debug statement here
 651                 if (!rc) {
 652                         CERROR("Error punch object %Ld subobj %Ld\n", md->lmd_object_id,
 653                                md->lmd_objects[i].l_object_id);
 654                 }
 655         }
 656         RETURN(rc);
 657 }
 658
 659 static int lov_cancel(struct lustre_handle *conn, struct lov_stripe_md *md, __u32 mode,
 660                       struct lustre_handle *lockhs)
 661 {
 662         int rc = 0, i;
 663         struct obd_export *export = class_conn2export(conn);
 664         struct lov_obd *lov;
 665         ENTRY;
 666
 667         if (!md) {
 668                 CERROR("LOV requires striping ea for lock cancellation\n");
 669                 RETURN(-EINVAL);
 670         }
 671
 672         if (!export || !export->exp_obd)
 673                 RETURN(-ENODEV);
 674
 675         lov = &export->exp_obd->u.lov;
 676         for (i = 0; i < md->lmd_stripe_count; i++) {
 677                 struct lov_stripe_md submd;
 678
 679                 if ( lockhs[i].addr == 0 )
 680                         continue;
 681
 682                 submd.lmd_object_id = md->lmd_objects[i].l_object_id;
 683                 submd.lmd_easize = sizeof(submd);
 684                 rc = obd_cancel(&lov->tgts[i].conn, &submd, mode, &lockhs[i]);
 685                 if (!rc) {
 686                         CERROR("Error punch object %Ld subobj %Ld\n", md->lmd_object_id,
 687                                md->lmd_objects[i].l_object_id);
 688                 }
 689         }
 690         RETURN(rc);
 691 }
 692
 693
 694
 695
 696 struct obd_ops lov_obd_ops = {
 697         o_setup:       lov_setup,
 698         o_connect:     lov_connect,
 699         o_disconnect:  lov_disconnect,
 700         o_create:      lov_create,
 701         o_destroy:     lov_destroy,
 702         o_getattr:     lov_getattr,
 703         o_setattr:     lov_setattr,
 704         o_open:        lov_open,
 705         o_close:       lov_close,
 706         o_brw:         lov_brw,
 707         o_punch:       lov_punch,
 708         o_enqueue:     lov_enqueue,
 709         o_cancel:      lov_cancel
 710 };
 711
 712
 713 #define LOV_VERSION "v0.1"
 714
 715 static int __init lov_init(void)
 716 {
 717         printk(KERN_INFO "Lustre Logical Object Volume driver " LOV_VERSION
 718                ", info@clusterfs.com\n");
 719         return class_register_type(&lov_obd_ops, OBD_LOV_DEVICENAME);
 720 }
 721
 722 static void __exit lov_exit(void)
 723 {
 724         class_unregister_type(OBD_LOV_DEVICENAME);
 725 }
 726
 727 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 728 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver v0.1");
 729 MODULE_LICENSE("GPL");
 730
 731 module_init(lov_init);
 732 module_exit(lov_exit);