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