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 #include <asm/div64.h>
  28
  29 /* obd methods */
  30 static int lov_connect(struct lustre_handle *conn, struct obd_device *obd,
  31                        obd_uuid_t cluuid)
  32 {
  33         struct ptlrpc_request *req;
  34         struct lov_obd *lov = &obd->u.lov;
  35         struct client_obd *mdc = &lov->mdcobd->u.cli;
  36         struct lov_desc *desc = &lov->desc;
  37         struct lustre_handle mdc_conn;
  38         obd_uuid_t *uuidarray;
  39         int rc, rc2;
  40         int i;
  41
  42         MOD_INC_USE_COUNT;
  43         rc = class_connect(conn, obd, cluuid);
  44         if (rc) {
  45                 MOD_DEC_USE_COUNT;
  46                 RETURN(rc);
  47         }
  48
  49         /* retrieve LOV metadata from MDS */
  50         rc = obd_connect(&mdc_conn, lov->mdcobd, NULL);
  51         if (rc) {
  52                 CERROR("cannot connect to mdc: rc = %d\n", rc);
  53                 GOTO(out, rc = -EINVAL);
  54         }
  55
  56         rc = mdc_getlovinfo(obd, &mdc_conn, &req);
  57         rc2 = obd_disconnect(&mdc_conn);
  58         if (rc || rc2) {
  59                 CERROR("cannot get lov info or disconnect %d/%d\n", rc, rc2);
  60                 GOTO(out, (rc) ? rc : rc2 );
  61         }
  62
  63         /* sanity... */
  64         if (req->rq_repmsg->bufcount < 2 ||
  65             req->rq_repmsg->buflens[0] < sizeof(*desc)) {
  66                 CERROR("LOV desc: invalid descriptor returned\n");
  67                 GOTO(out, rc = -EINVAL);
  68         }
  69
  70         memcpy(desc, lustre_msg_buf(req->rq_repmsg, 0), sizeof(*desc));
  71         lov_unpackdesc(desc);
  72
  73         if (req->rq_repmsg->buflens[1] < sizeof(*uuidarray)*desc->ld_tgt_count){
  74                 CERROR("LOV desc: invalid uuid array returned\n");
  75                 GOTO(out, rc = -EINVAL);
  76         }
  77
  78         mdc->cl_max_mds_easize = lov_mds_md_size(desc->ld_tgt_count);
  79         mdc->cl_max_ost_easize = lov_stripe_md_size(desc->ld_tgt_count);
  80
  81         if (memcmp(obd->obd_uuid, desc->ld_uuid, sizeof(desc->ld_uuid))) {
  82                 CERROR("LOV desc: uuid %s not on mds device (%s)\n",
  83                        obd->obd_uuid, desc->ld_uuid);
  84                 GOTO(out, rc = -EINVAL);
  85         }
  86
  87         if (desc->ld_tgt_count > 1000) {
  88                 CERROR("LOV desc: target count > 1000 (%d)\n",
  89                        desc->ld_tgt_count);
  90                 GOTO(out, rc = -EINVAL);
  91         }
  92
  93         if (desc->ld_default_stripe_count == 0)
  94                 desc->ld_default_stripe_count = desc->ld_tgt_count;
  95
  96         /* Because of 64-bit divide/mod operations only work with a 32-bit
  97          * divisor in a 32-bit kernel, we cannot support a stripe width
  98          * of 4GB or larger.
  99          */
 100         if (desc->ld_default_stripe_size * desc->ld_tgt_count > ~0UL) {
 101                 CERROR("LOV desc: stripe width > %lu on 32-bit system\n",
 102                        ~0UL);
 103                 GOTO(out, rc = -EINVAL);
 104         }
 105
 106         lov->bufsize = sizeof(struct lov_tgt_desc) * desc->ld_tgt_count;
 107         OBD_ALLOC(lov->tgts, lov->bufsize);
 108         if (!lov->tgts) {
 109                 CERROR("Out of memory\n");
 110                 GOTO(out, rc = -ENOMEM);
 111         }
 112
 113         uuidarray = lustre_msg_buf(req->rq_repmsg, 1);
 114         for (i = 0 ; i < desc->ld_tgt_count; i++)
 115                 memcpy(lov->tgts[i].uuid, uuidarray[i], sizeof(*uuidarray));
 116
 117         for (i = 0 ; i < desc->ld_tgt_count; i++) {
 118                 struct obd_device *tgt = class_uuid2obd(uuidarray[i]);
 119                 if (!tgt) {
 120                         CERROR("Target %s not attached\n", uuidarray[i]);
 121                         GOTO(out_mem, rc = -EINVAL);
 122                 }
 123                 if (!(tgt->obd_flags & OBD_SET_UP)) {
 124                         CERROR("Target %s not set up\n", uuidarray[i]);
 125                         GOTO(out_mem, rc = -EINVAL);
 126                 }
 127                 rc = obd_connect(&lov->tgts[i].conn, tgt, NULL);
 128                 if (rc) {
 129                         CERROR("Target %s connect error %d\n",
 130                                uuidarray[i], rc);
 131                         GOTO(out_mem, rc);
 132                 }
 133         }
 134
 135  out_mem:
 136         if (rc) {
 137                 for (i = 0 ; i < desc->ld_tgt_count; i++) {
 138                         rc2 = obd_disconnect(&lov->tgts[i].conn);
 139                         if (rc2)
 140                                 CERROR("BAD: Target %s disconnect error %d\n",
 141                                        uuidarray[i], rc2);
 142                 }
 143                 OBD_FREE(lov->tgts, lov->bufsize);
 144         }
 145  out:
 146         if (rc)
 147                 class_disconnect(conn);
 148         ptlrpc_free_req(req);
 149         return rc;
 150 }
 151
 152 static int lov_disconnect(struct lustre_handle *conn)
 153 {
 154         struct obd_device *obd = class_conn2obd(conn);
 155         struct lov_obd *lov = &obd->u.lov;
 156         int rc;
 157         int i;
 158
 159         if (!lov->tgts)
 160                 goto out_local;
 161
 162         for (i = 0 ; i < lov->desc.ld_tgt_count; i++) {
 163                 rc = obd_disconnect(&lov->tgts[i].conn);
 164                 if (rc) {
 165                         CERROR("Target %s disconnect error %d\n",
 166                                lov->tgts[i].uuid, rc);
 167                         RETURN(rc);
 168                 }
 169         }
 170         OBD_FREE(lov->tgts, lov->bufsize);
 171         lov->bufsize = 0;
 172         lov->tgts = NULL;
 173
 174  out_local:
 175         rc = class_disconnect(conn);
 176         if (!rc)
 177                 MOD_DEC_USE_COUNT;
 178         return rc;
 179 }
 180
 181 static int lov_setup(struct obd_device *obd, obd_count len, void *buf)
 182 {
 183         struct obd_ioctl_data* data = buf;
 184         struct lov_obd *lov = &obd->u.lov;
 185         int rc = 0;
 186         ENTRY;
 187
 188         if (data->ioc_inllen1 < 1) {
 189                 CERROR("osc setup requires an MDC UUID\n");
 190                 RETURN(-EINVAL);
 191         }
 192
 193         if (data->ioc_inllen1 > 37) {
 194                 CERROR("mdc UUID must be 36 characters or less\n");
 195                 RETURN(-EINVAL);
 196         }
 197
 198         lov->mdcobd = class_uuid2obd(data->ioc_inlbuf1);
 199         if (!lov->mdcobd) {
 200                 CERROR("LOV %s cannot locate MDC %s\n", obd->obd_uuid,
 201                        data->ioc_inlbuf1);
 202                 rc = -EINVAL;
 203         }
 204         RETURN(rc);
 205 }
 206
 207
 208 /* the LOV expects oa->o_id to be set to the LOV object id */
 209 static int lov_create(struct lustre_handle *conn, struct obdo *oa,
 210                       struct lov_stripe_md **ea)
 211 {
 212         struct obd_export *export = class_conn2export(conn);
 213         struct lov_obd *lov;
 214         struct lov_stripe_md *lsm;
 215         struct lov_oinfo *loi;
 216         int sub_offset, stripe_offset;
 217         int ost_count;
 218         int rc = 0, i;
 219         ENTRY;
 220
 221         if (!ea) {
 222                 CERROR("lov_create needs ea\n");
 223                 RETURN(-EINVAL);
 224         }
 225
 226         if (!export)
 227                 RETURN(-EINVAL);
 228
 229         lov = &export->exp_obd->u.lov;
 230         ost_count = lov->desc.ld_tgt_count;
 231         oa->o_easize = lov_stripe_md_size(ost_count);
 232         if (!*ea) {
 233                 OBD_ALLOC(*ea, oa->o_easize);
 234                 if (!*ea)
 235                         RETURN(-ENOMEM);
 236         }
 237
 238         lsm = *ea;
 239         LASSERT(oa->o_valid & OBD_MD_FLID);
 240         lsm->lsm_magic = LOV_MAGIC;
 241         lsm->lsm_mds_easize = lov_mds_md_size(ost_count);
 242         lsm->lsm_object_id = oa->o_id;
 243         if (!lsm->lsm_stripe_count)
 244                 lsm->lsm_stripe_count = lov->desc.ld_default_stripe_count;
 245
 246         if (!lsm->lsm_stripe_size)
 247                 lsm->lsm_stripe_size = lov->desc.ld_default_stripe_size;
 248
 249         lsm->lsm_ost_count = ost_count;
 250         stripe_offset = (((int)lsm->lsm_object_id * lsm->lsm_stripe_count) %
 251                          ost_count);
 252         sub_offset = ((int)lsm->lsm_object_id*lsm->lsm_stripe_count/ost_count)%
 253                         lsm->lsm_stripe_count;
 254         lsm->lsm_stripe_offset = stripe_offset + sub_offset;
 255
 256         CDEBUG(D_INODE, "allocating %d subobjs for objid "LPX64" at idx %d\n",
 257                lsm->lsm_stripe_count,lsm->lsm_object_id,lsm->lsm_stripe_offset);
 258
 259         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 260                 struct lov_stripe_md obj_md;
 261                 struct lov_stripe_md *obj_mdp = &obj_md;
 262                 struct obdo tmp;
 263                 int ost_idx = (((sub_offset + i) % lsm->lsm_stripe_count) +
 264                                stripe_offset) % ost_count;
 265
 266                 /* create data objects with "parent" OA */
 267                 memcpy(&tmp, oa, sizeof(tmp));
 268                 tmp.o_easize = sizeof(struct lov_stripe_md);
 269                 rc = obd_create(&lov->tgts[ost_idx].conn, &tmp, &obj_mdp);
 270                 if (rc) {
 271                         CERROR("error creating objid "LPX64" sub-object on "
 272                                "OST idx %d: rc = %d\n", oa->o_id, ost_idx, rc);
 273                         GOTO(out_cleanup, rc);
 274                 }
 275                 loi->loi_id = tmp.o_id;
 276                 loi->loi_size = tmp.o_size;
 277                 loi->loi_ost_idx = ost_idx;
 278                 CDEBUG(D_INODE, "objid "LPX64" has subobj "LPX64" at idx %d\n",
 279                        lsm->lsm_object_id, loi->loi_id, ost_idx);
 280         }
 281
 282  out_cleanup:
 283         if (rc) {
 284                 while (i-- > 0) {
 285                         struct obdo tmp;
 286                         int err;
 287
 288                         --loi;
 289                         /* destroy already created objects here */
 290                         memcpy(&tmp, oa, sizeof(tmp));
 291                         tmp.o_id = loi->loi_id;
 292                         err = obd_destroy(&lov->tgts[loi->loi_ost_idx].conn,
 293                                           &tmp, NULL);
 294                         if (err)
 295                                 CERROR("Failed to remove objid "LPX64" subobj "
 296                                        LPX64" on OST idx %d: rc = %d\n",
 297                                        oa->o_id, loi->loi_id, loi->loi_ost_idx,
 298                                        err);
 299                 }
 300         }
 301         return rc;
 302 }
 303
 304 static int lov_destroy(struct lustre_handle *conn, struct obdo *oa,
 305                        struct lov_stripe_md *lsm)
 306 {
 307         struct obdo tmp;
 308         struct obd_export *export = class_conn2export(conn);
 309         struct lov_obd *lov;
 310         struct lov_oinfo *loi;
 311         int rc = 0, i;
 312         ENTRY;
 313
 314         if (!lsm) {
 315                 CERROR("LOV requires striping ea for destruction\n");
 316                 RETURN(-EINVAL);
 317         }
 318
 319         if (lsm->lsm_magic != LOV_MAGIC) {
 320                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 321                        lsm->lsm_magic, LOV_MAGIC);
 322                 RETURN(-EINVAL);
 323         }
 324
 325         if (!export || !export->exp_obd)
 326                 RETURN(-ENODEV);
 327
 328         lov = &export->exp_obd->u.lov;
 329         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 330                 /* create data objects with "parent" OA */
 331                 memcpy(&tmp, oa, sizeof(tmp));
 332                 tmp.o_id = loi->loi_id;
 333                 rc = obd_destroy(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL);
 334                 if (rc)
 335                         CERROR("Error destroying objid "LPX64" subobj "LPX64
 336                                " on OST idx %d\n: rc = %d",
 337                                oa->o_id, loi->loi_id, loi->loi_ost_idx, rc);
 338         }
 339         RETURN(rc);
 340 }
 341
 342 static int lov_getattr(struct lustre_handle *conn, struct obdo *oa,
 343                        struct lov_stripe_md *lsm)
 344 {
 345         struct obdo tmp;
 346         struct obd_export *export = class_conn2export(conn);
 347         struct lov_obd *lov;
 348         struct lov_oinfo *loi;
 349         int rc = 0, i;
 350         int set = 0;
 351         ENTRY;
 352
 353         if (!lsm) {
 354                 CERROR("LOV requires striping ea\n");
 355                 RETURN(-EINVAL);
 356         }
 357
 358         if (lsm->lsm_magic != LOV_MAGIC) {
 359                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 360                        lsm->lsm_magic, LOV_MAGIC);
 361                 RETURN(-EINVAL);
 362         }
 363
 364         if (!export || !export->exp_obd)
 365                 RETURN(-ENODEV);
 366
 367         lov = &export->exp_obd->u.lov;
 368         oa->o_size = 0;
 369         oa->o_blocks = 0;
 370         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 371                 int err;
 372
 373                 if (loi->loi_id == 0)
 374                         continue;
 375
 376                 /* create data objects with "parent" OA */
 377                 memcpy(&tmp, oa, sizeof(tmp));
 378                 tmp.o_id = loi->loi_id;
 379
 380                 err = obd_getattr(&lov->tgts[loi->loi_ost_idx].conn, &tmp,NULL);
 381                 if (err) {
 382                         CERROR("Error getattr objid "LPX64" subobj "LPX64
 383                                " on OST idx %d: rc = %d\n",
 384                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 385                         if (!rc)
 386                                 rc = err;
 387                         continue; /* XXX or break? */
 388                 }
 389                 if (!set) {
 390                         obdo_cpy_md(oa, &tmp, tmp.o_valid);
 391                         set = 1;
 392                 } else {
 393 #warning FIXME: the size needs to be fixed for sparse files
 394                         if (tmp.o_valid & OBD_MD_FLSIZE)
 395                                 oa->o_size += tmp.o_size;
 396                         if (tmp.o_valid & OBD_MD_FLBLOCKS)
 397                                 oa->o_blocks += tmp.o_blocks;
 398                         if (tmp.o_valid & OBD_MD_FLCTIME &&
 399                             oa->o_ctime < tmp.o_ctime)
 400                                 oa->o_ctime = tmp.o_ctime;
 401                         if (tmp.o_valid & OBD_MD_FLMTIME &&
 402                             oa->o_mtime < tmp.o_mtime)
 403                                 oa->o_mtime = tmp.o_mtime;
 404                 }
 405         }
 406         RETURN(rc);
 407 }
 408
 409 static int lov_setattr(struct lustre_handle *conn, struct obdo *oa,
 410                        struct lov_stripe_md *lsm)
 411 {
 412         int rc = 0, i;
 413         struct obdo tmp;
 414         struct obd_export *export = class_conn2export(conn);
 415         struct lov_obd *lov;
 416         struct lov_oinfo *loi;
 417         ENTRY;
 418
 419         if (!lsm) {
 420                 CERROR("LOV requires striping ea\n");
 421                 RETURN(-EINVAL);
 422         }
 423
 424         if (lsm->lsm_magic != LOV_MAGIC) {
 425                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 426                        lsm->lsm_magic, LOV_MAGIC);
 427                 RETURN(-EINVAL);
 428         }
 429
 430         if (!export || !export->exp_obd)
 431                 RETURN(-ENODEV);
 432
 433         if (oa->o_valid & OBD_MD_FLSIZE)
 434                 CERROR("setting size on an LOV object is totally broken\n");
 435
 436         lov = &export->exp_obd->u.lov;
 437         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 438                 int err;
 439
 440                 /* create data objects with "parent" OA */
 441                 memcpy(&tmp, oa, sizeof(tmp));
 442                 tmp.o_id = loi->loi_id;
 443
 444                 err = obd_setattr(&lov->tgts[loi->loi_ost_idx].conn, &tmp,NULL);
 445                 if (err) {
 446                         CERROR("Error setattr objid "LPX64" subobj "LPX64
 447                                " on OST idx %d: rc = %d\n",
 448                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 449                         if (!rc)
 450                                 rc = err;
 451                 }
 452         }
 453         RETURN(rc);
 454 }
 455
 456 static int lov_open(struct lustre_handle *conn, struct obdo *oa,
 457                     struct lov_stripe_md *lsm)
 458 {
 459         struct obdo tmp;
 460         struct obd_export *export = class_conn2export(conn);
 461         struct lov_obd *lov;
 462         struct lov_oinfo *loi;
 463         int rc = 0, i;
 464         ENTRY;
 465
 466         if (!lsm) {
 467                 CERROR("LOV requires striping ea for opening\n");
 468                 RETURN(-EINVAL);
 469         }
 470
 471         if (lsm->lsm_magic != LOV_MAGIC) {
 472                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 473                        lsm->lsm_magic, LOV_MAGIC);
 474                 RETURN(-EINVAL);
 475         }
 476
 477         if (!export || !export->exp_obd)
 478                 RETURN(-ENODEV);
 479
 480         lov = &export->exp_obd->u.lov;
 481         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 482                 int err;
 483
 484                 /* create data objects with "parent" OA */
 485                 memcpy(&tmp, oa, sizeof(tmp));
 486                 tmp.o_id = loi->loi_id;
 487
 488                 err = obd_open(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL);
 489                 if (err) {
 490                         CERROR("Error open objid "LPX64" subobj "LPX64
 491                                " on OST idx %d: rc = %d\n",
 492                                oa->o_id, lsm->lsm_oinfo[i].loi_id,
 493                                loi->loi_ost_idx, rc);
 494                         if (!rc)
 495                                 rc = err;
 496                 }
 497         }
 498         /* FIXME: returning an error, but having opened some objects is a bad
 499          *        idea, since they will likely never be closed.  We either
 500          *        need to not return an error if _some_ objects could be
 501          *        opened, and leave it to read/write to return -EIO (with
 502          *        hopefully partial error status) or close all opened objects
 503          *        and return an error.
 504          */
 505         RETURN(rc);
 506 }
 507
 508 static int lov_close(struct lustre_handle *conn, struct obdo *oa,
 509                      struct lov_stripe_md *lsm)
 510 {
 511         struct obdo tmp;
 512         struct obd_export *export = class_conn2export(conn);
 513         struct lov_obd *lov;
 514         struct lov_oinfo *loi;
 515         int rc = 0, i;
 516         ENTRY;
 517
 518         if (!lsm) {
 519                 CERROR("LOV requires striping ea\n");
 520                 RETURN(-EINVAL);
 521         }
 522
 523         if (lsm->lsm_magic != LOV_MAGIC) {
 524                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 525                        lsm->lsm_magic, LOV_MAGIC);
 526                 RETURN(-EINVAL);
 527         }
 528
 529         if (!export || !export->exp_obd)
 530                 RETURN(-ENODEV);
 531
 532         lov = &export->exp_obd->u.lov;
 533         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 534                 int err;
 535
 536                 /* create data objects with "parent" OA */
 537                 memcpy(&tmp, oa, sizeof(tmp));
 538                 tmp.o_id = loi->loi_id;
 539
 540                 err = obd_close(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL);
 541                 if (err) {
 542                         CERROR("Error close objid "LPX64" subobj "LPX64
 543                                " on OST idx %d: rc = %d\n",
 544                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 545                         if (!rc)
 546                                 rc = err;
 547                 }
 548         }
 549         RETURN(rc);
 550 }
 551
 552 #ifndef log2
 553 #define log2(n) ffz(~(n))
 554 #endif
 555
 556 #warning FIXME: merge these two functions now that they are nearly the same
 557
 558 /* compute ost offset in stripe "stripeno" corresponding to offset "lov_off" */
 559 static __u64 lov_offset(struct lov_stripe_md *lsm, __u64 lov_off, int stripeno)
 560 {
 561         unsigned long ssize  = lsm->lsm_stripe_size;
 562         unsigned long swidth = ssize * lsm->lsm_stripe_count;
 563         unsigned long stripe_off;
 564
 565         if (lov_off == OBD_PUNCH_EOF)
 566                 return OBD_PUNCH_EOF;
 567
 568         /* do_div(a, b) returns a % b, and a = a / b */
 569         stripe_off = do_div(lov_off, swidth);
 570
 571         if (stripe_off < stripeno * ssize)
 572                 stripe_off = 0;
 573         else
 574                 stripe_off -= stripeno * ssize;
 575
 576         return lov_off * ssize + stripe_off;
 577 }
 578
 579 /* compute which stripe offset "lov_off" will be written into */
 580 static int lov_stripe_which(struct lov_stripe_md *lsm, __u64 lov_off)
 581 {
 582         unsigned long ssize  = lsm->lsm_stripe_size;
 583         unsigned long swidth = ssize * lsm->lsm_stripe_count;
 584         unsigned long stripe_off;
 585
 586         stripe_off = do_div(lov_off, swidth);
 587
 588         return stripe_off / ssize;
 589 }
 590
 591
 592 /* FIXME: maybe we'll just make one node the authoritative attribute node, then
 593  * we can send this 'punch' to just the authoritative node and the nodes
 594  * that the punch will affect. */
 595 static int lov_punch(struct lustre_handle *conn, struct obdo *oa,
 596                      struct lov_stripe_md *lsm,
 597                      obd_off start, obd_off end)
 598 {
 599         struct obdo tmp;
 600         struct obd_export *export = class_conn2export(conn);
 601         struct lov_obd *lov;
 602         struct lov_oinfo *loi;
 603         int rc = 0, i;
 604         ENTRY;
 605
 606         if (!lsm) {
 607                 CERROR("LOV requires striping ea\n");
 608                 RETURN(-EINVAL);
 609         }
 610
 611         if (lsm->lsm_magic != LOV_MAGIC) {
 612                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 613                        lsm->lsm_magic, LOV_MAGIC);
 614                 RETURN(-EINVAL);
 615         }
 616
 617         if (!export || !export->exp_obd)
 618                 RETURN(-ENODEV);
 619
 620         lov = &export->exp_obd->u.lov;
 621         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 622                 __u64 starti = lov_offset(lsm, start, i);
 623                 __u64 endi = lov_offset(lsm, end, i);
 624                 int err;
 625
 626                 if (starti == endi)
 627                         continue;
 628                 /* create data objects with "parent" OA */
 629                 memcpy(&tmp, oa, sizeof(tmp));
 630                 tmp.o_id = loi->loi_id;
 631
 632                 err = obd_punch(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL,
 633                                starti, endi);
 634                 if (err) {
 635                         CERROR("Error punch objid "LPX64" subobj "LPX64
 636                                " on OST idx %d: rc = %d\n",
 637                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 638                         if (!rc)
 639                                 rc = err;
 640                 }
 641         }
 642         RETURN(rc);
 643 }
 644
 645 static int lov_osc_brw_callback(struct io_cb_data *cbd, int err, int phase)
 646 {
 647         int ret = 0;
 648         ENTRY;
 649
 650         if (phase == CB_PHASE_START)
 651                 RETURN(0);
 652
 653         if (phase == CB_PHASE_FINISH) {
 654                 if (err)
 655                         cbd->err = err;
 656                 if (atomic_dec_and_test(&cbd->refcount))
 657                         ret = cbd->cb(cbd->data, cbd->err, phase);
 658                 RETURN(ret);
 659         }
 660
 661         LBUG();
 662         return 0;
 663 }
 664
 665 static inline int lov_brw(int cmd, struct lustre_handle *conn,
 666                           struct lov_stripe_md *lsm, obd_count oa_bufs,
 667                           struct brw_page *pga,
 668                           brw_callback_t callback, struct io_cb_data *cbd)
 669 {
 670         int stripe_count = lsm->lsm_stripe_count;
 671         struct obd_export *export = class_conn2export(conn);
 672         struct lov_obd *lov;
 673         struct {
 674                 int bufct;
 675                 int index;
 676                 int subcount;
 677                 struct lov_stripe_md lsm;
 678                 int ost_idx;
 679         } *stripeinfo, *si, *si_last;
 680         struct brw_page *ioarr;
 681         int rc, i;
 682         struct io_cb_data *our_cb;
 683         struct lov_oinfo *loi;
 684         int *where;
 685         ENTRY;
 686
 687         if (!lsm) {
 688                 CERROR("LOV requires striping ea\n");
 689                 RETURN(-EINVAL);
 690         }
 691
 692         if (lsm->lsm_magic != LOV_MAGIC) {
 693                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 694                        lsm->lsm_magic, LOV_MAGIC);
 695                 RETURN(-EINVAL);
 696         }
 697
 698         lov = &export->exp_obd->u.lov;
 699
 700         our_cb = ll_init_cb();
 701         if (!our_cb)
 702                 RETURN(-ENOMEM);
 703
 704         OBD_ALLOC(stripeinfo, stripe_count * sizeof(*stripeinfo));
 705         if (!stripeinfo)
 706                 GOTO(out_cbdata, rc = -ENOMEM);
 707
 708         OBD_ALLOC(where, sizeof(*where) * oa_bufs);
 709         if (!where)
 710                 GOTO(out_sinfo, rc = -ENOMEM);
 711
 712         OBD_ALLOC(ioarr, sizeof(*ioarr) * oa_bufs);
 713         if (!ioarr)
 714                 GOTO(out_where, rc = -ENOMEM);
 715
 716         /* This is the only race-free way I can think of to get the refcount
 717          * correct. -phil */
 718         atomic_set(&our_cb->refcount, 0);
 719         our_cb->cb = callback;
 720         our_cb->data = cbd;
 721
 722         for (i = 0; i < oa_bufs; i++) {
 723                 where[i] = lov_stripe_which(lsm, pga[i].off);
 724                 if (stripeinfo[where[i]].bufct++ == 0)
 725                         atomic_inc(&our_cb->refcount);
 726         }
 727
 728         for (i = 0, loi = lsm->lsm_oinfo, si_last = si = stripeinfo;
 729              i < stripe_count; i++, loi++, si_last = si, si++) {
 730                 if (i > 0)
 731                         si->index = si_last->index + si_last->bufct;
 732                 si->lsm.lsm_object_id = loi->loi_id;
 733                 si->ost_idx = loi->loi_ost_idx;
 734         }
 735
 736         for (i = 0; i < oa_bufs; i++) {
 737                 int which = where[i];
 738                 int shift;
 739
 740                 shift = stripeinfo[which].index + stripeinfo[which].subcount;
 741                 LASSERT(shift < oa_bufs);
 742                 ioarr[shift] = pga[i];
 743                 ioarr[shift].off = lov_offset(lsm, pga[i].off, which);
 744                 stripeinfo[which].subcount++;
 745         }
 746
 747         for (i = 0, si = stripeinfo; i < stripe_count; i++, si++) {
 748                 int shift = si->index;
 749
 750                 if (si->bufct) {
 751                         LASSERT(shift < oa_bufs);
 752                         obd_brw(cmd, &lov->tgts[si->ost_idx].conn,
 753                                 &si->lsm, si->bufct, &ioarr[shift],
 754                                 lov_osc_brw_callback, our_cb);
 755                 }
 756         }
 757
 758         rc = callback(cbd, 0, CB_PHASE_START);
 759
 760         OBD_FREE(ioarr, sizeof(*ioarr) * oa_bufs);
 761  out_where:
 762         OBD_FREE(where, sizeof(*where) * oa_bufs);
 763  out_sinfo:
 764         OBD_FREE(stripeinfo, stripe_count * sizeof(*stripeinfo));
 765  out_cbdata:
 766         OBD_FREE(our_cb, sizeof(*our_cb));
 767         RETURN(rc);
 768 }
 769
 770 static int lov_enqueue(struct lustre_handle *conn, struct lov_stripe_md *lsm,
 771                        struct lustre_handle *parent_lock,
 772                        __u32 type, void *cookie, int cookielen, __u32 mode,
 773                        int *flags, void *cb, void *data, int datalen,
 774                        struct lustre_handle *lockhs)
 775 {
 776         struct obd_export *export = class_conn2export(conn);
 777         struct lov_obd *lov;
 778         struct lov_oinfo *loi;
 779         int rc = 0, i;
 780         ENTRY;
 781
 782         if (!lsm) {
 783                 CERROR("LOV requires striping ea\n");
 784                 RETURN(-EINVAL);
 785         }
 786
 787         if (lsm->lsm_magic != LOV_MAGIC) {
 788                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 789                        lsm->lsm_magic, LOV_MAGIC);
 790                 RETURN(-EINVAL);
 791         }
 792
 793         if (!export || !export->exp_obd)
 794                 RETURN(-ENODEV);
 795
 796         lov = &export->exp_obd->u.lov;
 797         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 798                 struct ldlm_extent *extent = (struct ldlm_extent *)cookie;
 799                 struct ldlm_extent sub_ext;
 800                 struct lov_stripe_md submd;
 801
 802                 sub_ext.start = lov_offset(lsm, extent->start, i);
 803                 sub_ext.end = lov_offset(lsm, extent->end, i);
 804                 if (sub_ext.start == sub_ext.end)
 805                         continue;
 806
 807                 submd.lsm_object_id = loi->loi_id;
 808                 /* XXX submd lsm_mds_easize should be that from the subobj,
 809                  *     and the subobj should get it opaquely from the LOV.
 810                  */
 811                 submd.lsm_mds_easize = lov_mds_md_size(lsm->lsm_ost_count);
 812                 submd.lsm_stripe_count = 0;
 813                 /* XXX submd is not fully initialized here */
 814                 rc = obd_enqueue(&(lov->tgts[loi->loi_ost_idx].conn), &submd,
 815                                  parent_lock, type, &sub_ext, sizeof(sub_ext),
 816                                  mode, flags, cb, data, datalen, &(lockhs[i]));
 817                 // XXX add a lock debug statement here
 818                 if (rc)
 819                         CERROR("Error enqueue objid "LPX64" subobj "LPX64
 820                                " on OST idx %d: rc = %d\n", lsm->lsm_object_id,
 821                                loi->loi_id, loi->loi_ost_idx, rc);
 822         }
 823         RETURN(rc);
 824 }
 825
 826 static int lov_cancel(struct lustre_handle *conn, struct lov_stripe_md *lsm,
 827                       __u32 mode, struct lustre_handle *lockhs)
 828 {
 829         struct obd_export *export = class_conn2export(conn);
 830         struct lov_obd *lov;
 831         struct lov_oinfo *loi;
 832         int rc = 0, i;
 833         ENTRY;
 834
 835         if (!lsm) {
 836                 CERROR("LOV requires striping ea\n");
 837                 RETURN(-EINVAL);
 838         }
 839
 840         if (lsm->lsm_magic != LOV_MAGIC) {
 841                 CERROR("LOV striping magic bad %#lx != %#lx\n",
 842                        lsm->lsm_magic, LOV_MAGIC);
 843                 RETURN(-EINVAL);
 844         }
 845
 846         if (!export || !export->exp_obd)
 847                 RETURN(-ENODEV);
 848
 849         lov = &export->exp_obd->u.lov;
 850         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 851                 struct lov_stripe_md submd;
 852
 853                 if (lockhs[i].addr == 0)
 854                         continue;
 855
 856                 submd.lsm_object_id = loi->loi_id;
 857                 submd.lsm_mds_easize = lov_mds_md_size(lsm->lsm_ost_count);
 858                 submd.lsm_stripe_count = 0;
 859                 rc = obd_cancel(&lov->tgts[loi->loi_ost_idx].conn, &submd,
 860                                 mode, &lockhs[i]);
 861                 if (rc)
 862                         CERROR("Error cancel objid "LPX64" subobj "LPX64
 863                                " on OST idx %d: rc = %d\n", lsm->lsm_object_id,
 864                                loi->loi_id, loi->loi_ost_idx, rc);
 865         }
 866         RETURN(rc);
 867 }
 868
 869 static int lov_cancel_unused(struct lustre_handle *conn,
 870                              struct lov_stripe_md *lsm)
 871 {
 872         struct obd_export *export = class_conn2export(conn);
 873         struct lov_obd *lov;
 874         struct lov_oinfo *loi;
 875         int rc = 0, i;
 876         ENTRY;
 877
 878         if (!lsm) {
 879                 CERROR("LOV requires striping ea for lock cancellation\n");
 880                 RETURN(-EINVAL);
 881         }
 882
 883         if (!export || !export->exp_obd)
 884                 RETURN(-ENODEV);
 885
 886         lov = &export->exp_obd->u.lov;
 887         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 888                 struct lov_stripe_md submd;
 889
 890                 submd.lsm_object_id = loi->loi_id;
 891                 submd.lsm_mds_easize = lov_mds_md_size(lsm->lsm_ost_count);
 892                 submd.lsm_stripe_count = 0;
 893                 rc = obd_cancel_unused(&lov->tgts[loi->loi_ost_idx].conn,
 894                                        &submd);
 895                 if (rc)
 896                         CERROR("Error cancel unused objid "LPX64" subobj "LPX64
 897                                " on OST idx %d: rc = %d\n", lsm->lsm_object_id,
 898                                loi->loi_id, loi->loi_ost_idx, rc);
 899         }
 900         RETURN(rc);
 901 }
 902
 903 static int lov_statfs(struct lustre_handle *conn, struct obd_statfs *osfs)
 904 {
 905         struct obd_export *export = class_conn2export(conn);
 906         struct lov_obd *lov;
 907         struct obd_statfs lov_sfs;
 908         int set = 0;
 909         int rc = 0;
 910         int i;
 911         ENTRY;
 912
 913         if (!export || !export->exp_obd)
 914                 RETURN(-ENODEV);
 915
 916         lov = &export->exp_obd->u.lov;
 917
 918         /* We only get block data from the OBD */
 919         for (i = 0 ; i < lov->desc.ld_tgt_count; i++) {
 920                 int err;
 921
 922                 err = obd_statfs(&lov->tgts[i].conn, &lov_sfs);
 923                 if (err) {
 924                         CERROR("Error statfs OSC %s idx %d: err = %d\n",
 925                                lov->tgts[i].uuid, i, err);
 926                         if (!rc)
 927                                 rc = err;
 928                         continue; /* XXX or break? - probably OK to continue */
 929                 }
 930                 if (!set) {
 931                         memcpy(osfs, &lov_sfs, sizeof(lov_sfs));
 932                         set = 1;
 933                 } else {
 934                         osfs->os_bfree += lov_sfs.os_bfree;
 935                         osfs->os_bavail += lov_sfs.os_bavail;
 936                         osfs->os_blocks += lov_sfs.os_blocks;
 937                         /* XXX not sure about this one - depends on policy.
 938                          *   - could be minimum if we always stripe on all OBDs
 939                          *     (but that would be wrong for any other policy,
 940                          *     if one of the OBDs has no more objects left)
 941                          *   - could be sum if we stripe whole objects
 942                          *   - could be average, just to give a nice number
 943                          *   - we just pick first OST and hope it is enough
 944                         sfs->f_ffree += lov_sfs.f_ffree;
 945                          */
 946                 }
 947         }
 948         RETURN(rc);
 949 }
 950
 951
 952 struct obd_ops lov_obd_ops = {
 953         o_setup:       lov_setup,
 954         o_connect:     lov_connect,
 955         o_disconnect:  lov_disconnect,
 956         o_create:      lov_create,
 957         o_destroy:     lov_destroy,
 958         o_getattr:     lov_getattr,
 959         o_setattr:     lov_setattr,
 960         o_statfs:      lov_statfs,
 961         o_open:        lov_open,
 962         o_close:       lov_close,
 963         o_brw:         lov_brw,
 964         o_punch:       lov_punch,
 965         o_enqueue:     lov_enqueue,
 966         o_cancel:      lov_cancel,
 967         o_cancel_unused: lov_cancel_unused
 968 };
 969
 970
 971 #define LOV_VERSION "v0.1"
 972
 973 static int __init lov_init(void)
 974 {
 975         printk(KERN_INFO "Lustre Logical Object Volume driver " LOV_VERSION
 976                ", info@clusterfs.com\n");
 977         return class_register_type(&lov_obd_ops, OBD_LOV_DEVICENAME);
 978 }
 979
 980 static void __exit lov_exit(void)
 981 {
 982         class_unregister_type(OBD_LOV_DEVICENAME);
 983 }
 984
 985 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 986 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver v0.1");
 987 MODULE_LICENSE("GPL");
 988
 989 module_init(lov_init);
 990 module_exit(lov_exit);