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                        char *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         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 for striping information\n");
 223                 RETURN(-EINVAL);
 224         }
 225         if (!export)
 226                 RETURN(-EINVAL);
 227
 228         lov = &export->exp_obd->u.lov;
 229         ost_count = lov->desc.ld_tgt_count;
 230         oa->o_easize = lov_stripe_md_size(ost_count);
 231         if (!*ea) {
 232                 OBD_ALLOC(*ea, oa->o_easize);
 233                 if (!*ea)
 234                         RETURN(-ENOMEM);
 235         }
 236
 237         lsm = *ea;
 238         LASSERT(oa->o_valid & OBD_MD_FLID);
 239         lsm->lsm_mds_easize = lov_mds_md_size(ost_count);
 240         lsm->lsm_object_id = oa->o_id;
 241         if (!lsm->lsm_stripe_count)
 242                 lsm->lsm_stripe_count = lov->desc.ld_default_stripe_count;
 243
 244         if (!lsm->lsm_stripe_size)
 245                 lsm->lsm_stripe_size = lov->desc.ld_default_stripe_size;
 246
 247         lsm->lsm_ost_count = ost_count;
 248         sub_offset = ((int)lsm->lsm_object_id / lsm->lsm_stripe_count) %
 249                 lsm->lsm_stripe_count;
 250         stripe_offset = (((int)lsm->lsm_object_id * lsm->lsm_stripe_count) %
 251                          ost_count);
 252         lsm->lsm_stripe_offset = stripe_offset + sub_offset;
 253
 254         CDEBUG(D_INODE, "allocating %d subobjs for objid "LPX64" at idx %d\n",
 255                lsm->lsm_stripe_count,lsm->lsm_object_id,lsm->lsm_stripe_offset);
 256
 257         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 258                 struct lov_stripe_md obj_md;
 259                 struct lov_stripe_md *obj_mdp = &obj_md;
 260                 struct obdo tmp;
 261                 int ost_idx = (((sub_offset + i) % lsm->lsm_stripe_count) +
 262                                stripe_offset) % ost_count;
 263
 264                 /* create data objects with "parent" OA */
 265                 memcpy(&tmp, oa, sizeof(tmp));
 266                 tmp.o_easize = sizeof(struct lov_stripe_md);
 267                 rc = obd_create(&lov->tgts[ost_idx].conn, &tmp, &obj_mdp);
 268                 if (rc) {
 269                         CERROR("error creating objid "LPX64" sub-object on "
 270                                "OST idx %d: rc = %d\n", oa->o_id, ost_idx, rc);
 271                         GOTO(out_cleanup, rc);
 272                 }
 273                 loi->loi_id = tmp.o_id;
 274                 loi->loi_size = tmp.o_size;
 275                 loi->loi_ost_idx = ost_idx;
 276                 CDEBUG(D_INODE, "objid "LPX64" has subobj "LPX64" at idx %d\n",
 277                        lsm->lsm_object_id, loi->loi_id, ost_idx);
 278         }
 279
 280  out_cleanup:
 281         if (rc) {
 282                 while (i-- > 0) {
 283                         struct obdo tmp;
 284                         int err;
 285
 286                         --loi;
 287                         /* destroy already created objects here */
 288                         memcpy(&tmp, oa, sizeof(tmp));
 289                         tmp.o_id = loi->loi_id;
 290                         err = obd_destroy(&lov->tgts[loi->loi_ost_idx].conn,
 291                                           &tmp, NULL);
 292                         if (err)
 293                                 CERROR("Failed to remove objid "LPX64" subobj "
 294                                        LPX64" on OST idx %d: rc = %d\n",
 295                                        oa->o_id, loi->loi_id, loi->loi_ost_idx,
 296                                        err);
 297                 }
 298         }
 299         return rc;
 300 }
 301
 302 static int lov_destroy(struct lustre_handle *conn, struct obdo *oa,
 303                        struct lov_stripe_md *lsm)
 304 {
 305         struct obdo tmp;
 306         struct obd_export *export = class_conn2export(conn);
 307         struct lov_obd *lov;
 308         struct lov_oinfo *loi;
 309         int rc = 0, i;
 310         ENTRY;
 311
 312         if (!lsm) {
 313                 CERROR("LOV requires striping ea for destruction\n");
 314                 RETURN(-EINVAL);
 315         }
 316
 317         if (!export || !export->exp_obd)
 318                 RETURN(-ENODEV);
 319
 320         lov = &export->exp_obd->u.lov;
 321         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 322                 /* create data objects with "parent" OA */
 323                 memcpy(&tmp, oa, sizeof(tmp));
 324                 tmp.o_id = loi->loi_id;
 325                 rc = obd_destroy(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL);
 326                 if (rc)
 327                         CERROR("Error destroying objid "LPX64" subobj "LPX64
 328                                " on OST idx %d\n: rc = %d",
 329                                oa->o_id, loi->loi_id, loi->loi_ost_idx, rc);
 330         }
 331         RETURN(rc);
 332 }
 333
 334 static int lov_getattr(struct lustre_handle *conn, struct obdo *oa,
 335                        struct lov_stripe_md *lsm)
 336 {
 337         struct obdo tmp;
 338         struct obd_export *export = class_conn2export(conn);
 339         struct lov_obd *lov;
 340         struct lov_oinfo *loi;
 341         int rc = 0, i;
 342         int set = 0;
 343         ENTRY;
 344
 345         if (!lsm) {
 346                 CERROR("LOV requires striping ea\n");
 347                 RETURN(-EINVAL);
 348         }
 349
 350         if (!export || !export->exp_obd)
 351                 RETURN(-ENODEV);
 352
 353         lov = &export->exp_obd->u.lov;
 354         oa->o_size = 0;
 355         oa->o_blocks = 0;
 356         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 357                 int err;
 358
 359                 if (loi->loi_id == 0)
 360                         continue;
 361
 362                 /* create data objects with "parent" OA */
 363                 memcpy(&tmp, oa, sizeof(tmp));
 364                 tmp.o_id = loi->loi_id;
 365
 366                 err = obd_getattr(&lov->tgts[loi->loi_ost_idx].conn, &tmp,NULL);
 367                 if (err) {
 368                         CERROR("Error getattr objid "LPX64" subobj "LPX64
 369                                " on OST idx %d: rc = %d\n",
 370                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 371                         if (!rc)
 372                                 rc = err;
 373                         continue; /* XXX or break? */
 374                 }
 375                 if (!set) {
 376                         obdo_cpy_md(oa, &tmp, tmp.o_valid);
 377                         set = 1;
 378                 } else {
 379 #warning FIXME: the size needs to be fixed for sparse files
 380                         if (tmp.o_valid & OBD_MD_FLSIZE)
 381                                 oa->o_size += tmp.o_size;
 382                         if (tmp.o_valid & OBD_MD_FLBLOCKS)
 383                                 oa->o_blocks += tmp.o_blocks;
 384                         if (tmp.o_valid & OBD_MD_FLCTIME &&
 385                             oa->o_ctime < tmp.o_ctime)
 386                                 oa->o_ctime = tmp.o_ctime;
 387                         if (tmp.o_valid & OBD_MD_FLMTIME &&
 388                             oa->o_mtime < tmp.o_mtime)
 389                                 oa->o_mtime = tmp.o_mtime;
 390                 }
 391         }
 392         RETURN(rc);
 393 }
 394
 395 static int lov_setattr(struct lustre_handle *conn, struct obdo *oa,
 396                        struct lov_stripe_md *lsm)
 397 {
 398         int rc = 0, i;
 399         struct obdo tmp;
 400         struct obd_export *export = class_conn2export(conn);
 401         struct lov_obd *lov;
 402         struct lov_oinfo *loi;
 403         ENTRY;
 404
 405         if (!lsm) {
 406                 CERROR("LOV requires striping ea\n");
 407                 RETURN(-EINVAL);
 408         }
 409
 410         if (!export || !export->exp_obd)
 411                 RETURN(-ENODEV);
 412
 413         if (oa->o_valid && OBD_MD_FLSIZE)
 414                 CERROR("setting size on an LOV object is totally broken\n");
 415
 416         lov = &export->exp_obd->u.lov;
 417         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 418                 int err;
 419
 420                 /* create data objects with "parent" OA */
 421                 memcpy(&tmp, oa, sizeof(tmp));
 422                 tmp.o_id = loi->loi_id;
 423
 424                 err = obd_setattr(&lov->tgts[loi->loi_ost_idx].conn, &tmp,NULL);
 425                 if (err) {
 426                         CERROR("Error setattr objid "LPX64" subobj "LPX64
 427                                " on OST idx %d: rc = %d\n",
 428                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 429                         if (!rc)
 430                                 rc = err;
 431                 }
 432         }
 433         RETURN(rc);
 434 }
 435
 436 static int lov_open(struct lustre_handle *conn, struct obdo *oa,
 437                     struct lov_stripe_md *lsm)
 438 {
 439         struct obdo tmp;
 440         struct obd_export *export = class_conn2export(conn);
 441         struct lov_obd *lov;
 442         struct lov_oinfo *loi;
 443         int rc = 0, i;
 444         ENTRY;
 445
 446         if (!lsm) {
 447                 CERROR("LOV requires striping ea for opening\n");
 448                 RETURN(-EINVAL);
 449         }
 450
 451         if (!export || !export->exp_obd)
 452                 RETURN(-ENODEV);
 453
 454         lov = &export->exp_obd->u.lov;
 455         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 456                 int err;
 457
 458                 /* create data objects with "parent" OA */
 459                 memcpy(&tmp, oa, sizeof(tmp));
 460                 tmp.o_id = loi->loi_id;
 461
 462                 err = obd_open(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL);
 463                 if (err) {
 464                         CERROR("Error open objid "LPX64" subobj "LPX64
 465                                " on OST idx %d: rc = %d\n",
 466                                oa->o_id, lsm->lsm_oinfo[i].loi_id,
 467                                loi->loi_ost_idx, rc);
 468                         if (!rc)
 469                                 rc = err;
 470                 }
 471         }
 472         /* FIXME: returning an error, but having opened some objects is a bad
 473          *        idea, since they will likely never be closed.  We either
 474          *        need to not return an error if _some_ objects could be
 475          *        opened, and leave it to read/write to return -EIO (with
 476          *        hopefully partial error status) or close all opened objects
 477          *        and return an error.
 478          */
 479         RETURN(rc);
 480 }
 481
 482 static int lov_close(struct lustre_handle *conn, struct obdo *oa,
 483                      struct lov_stripe_md *lsm)
 484 {
 485         struct obdo tmp;
 486         struct obd_export *export = class_conn2export(conn);
 487         struct lov_obd *lov;
 488         struct lov_oinfo *loi;
 489         int rc = 0, i;
 490         ENTRY;
 491
 492         if (!lsm) {
 493                 CERROR("LOV requires striping ea\n");
 494                 RETURN(-EINVAL);
 495         }
 496
 497         if (!export || !export->exp_obd)
 498                 RETURN(-ENODEV);
 499
 500         lov = &export->exp_obd->u.lov;
 501         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 502                 int err;
 503
 504                 /* create data objects with "parent" OA */
 505                 memcpy(&tmp, oa, sizeof(tmp));
 506                 tmp.o_id = loi->loi_id;
 507
 508                 err = obd_close(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL);
 509                 if (err) {
 510                         CERROR("Error close objid "LPX64" subobj "LPX64
 511                                " on OST idx %d: rc = %d\n",
 512                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 513                         if (!rc)
 514                                 rc = err;
 515                 }
 516         }
 517         RETURN(rc);
 518 }
 519
 520 #ifndef log2
 521 #define log2(n) ffz(~(n))
 522 #endif
 523
 524 #warning FIXME: merge these two functions now that they are nearly the same
 525
 526 /* compute ost offset in stripe "stripeno" corresponding to offset "lov_off" */
 527 static __u64 lov_offset(struct lov_stripe_md *lsm, __u64 lov_off, int stripeno)
 528 {
 529         unsigned long ssize  = lsm->lsm_stripe_size;
 530         unsigned long swidth = ssize * lsm->lsm_stripe_count;
 531         unsigned long stripe_off;
 532
 533         if (lov_off == OBD_PUNCH_EOF)
 534                 return OBD_PUNCH_EOF;
 535
 536         /* do_div(a, b) returns a % b, and a = a / b */
 537         stripe_off = do_div(lov_off, swidth);
 538
 539         if (stripe_off < stripeno * ssize)
 540                 stripe_off = 0;
 541         else
 542                 stripe_off -= stripeno * ssize;
 543
 544         return lov_off + stripe_off;
 545 }
 546
 547 /* compute which stripe offset "lov_off" will be written into */
 548 static int lov_stripe_which(struct lov_stripe_md *lsm, __u64 lov_off)
 549 {
 550         unsigned long ssize  = lsm->lsm_stripe_size;
 551         unsigned long swidth = ssize * lsm->lsm_stripe_count;
 552         unsigned long stripe_off;
 553
 554         stripe_off = do_div(lov_off, swidth);
 555
 556         return stripe_off / ssize;
 557 }
 558
 559
 560 /* FIXME: maybe we'll just make one node the authoritative attribute node, then
 561  * we can send this 'punch' to just the authoritative node and the nodes
 562  * that the punch will affect. */
 563 static int lov_punch(struct lustre_handle *conn, struct obdo *oa,
 564                      struct lov_stripe_md *lsm,
 565                      obd_off start, obd_off end)
 566 {
 567         struct obdo tmp;
 568         struct obd_export *export = class_conn2export(conn);
 569         struct lov_obd *lov;
 570         struct lov_oinfo *loi;
 571         int rc = 0, i;
 572         ENTRY;
 573
 574         if (!lsm) {
 575                 CERROR("LOV requires striping ea for desctruction\n");
 576                 RETURN(-EINVAL);
 577         }
 578
 579         if (!export || !export->exp_obd)
 580                 RETURN(-ENODEV);
 581
 582         lov = &export->exp_obd->u.lov;
 583         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 584                 __u64 starti = lov_offset(lsm, start, i);
 585                 __u64 endi = lov_offset(lsm, end, i);
 586                 int err;
 587
 588                 if (starti == endi)
 589                         continue;
 590                 /* create data objects with "parent" OA */
 591                 memcpy(&tmp, oa, sizeof(tmp));
 592                 tmp.o_id = loi->loi_id;
 593
 594                 err = obd_punch(&lov->tgts[loi->loi_ost_idx].conn, &tmp, NULL,
 595                                starti, endi);
 596                 if (err) {
 597                         CERROR("Error punch objid "LPX64" subobj "LPX64
 598                                " on OST idx %d: rc = %d\n",
 599                                oa->o_id, loi->loi_id, loi->loi_ost_idx, err);
 600                         if (!rc)
 601                                 rc = err;
 602                 }
 603         }
 604         RETURN(rc);
 605 }
 606
 607 static int lov_osc_brw_callback(struct io_cb_data *cbd, int err, int phase)
 608 {
 609         int ret = 0;
 610         ENTRY;
 611
 612         if (phase == CB_PHASE_START)
 613                 RETURN(0);
 614
 615         if (phase == CB_PHASE_FINISH) {
 616                 if (err)
 617                         cbd->err = err;
 618                 if (atomic_dec_and_test(&cbd->refcount))
 619                         ret = cbd->cb(cbd->data, cbd->err, phase);
 620                 RETURN(ret);
 621         }
 622
 623         LBUG();
 624         return 0;
 625 }
 626
 627 static inline int lov_brw(int cmd, struct lustre_handle *conn,
 628                           struct lov_stripe_md *lsm, obd_count oa_bufs,
 629                           struct brw_page *pga,
 630                           brw_callback_t callback, struct io_cb_data *cbd)
 631 {
 632         int stripe_count = lsm->lsm_stripe_count;
 633         struct obd_export *export = class_conn2export(conn);
 634         struct lov_obd *lov;
 635         struct {
 636                 int bufct;
 637                 int index;
 638                 int subcount;
 639                 struct lov_stripe_md lsm;
 640                 int ost_idx;
 641         } *stripeinfo, *si, *si_last;
 642         struct brw_page *ioarr;
 643         int rc, i;
 644         struct io_cb_data *our_cb;
 645         struct lov_oinfo *loi;
 646         int *where;
 647         ENTRY;
 648
 649         lov = &export->exp_obd->u.lov;
 650
 651         our_cb = ll_init_cb();
 652         if (!our_cb)
 653                 RETURN(-ENOMEM);
 654
 655         OBD_ALLOC(stripeinfo, stripe_count * sizeof(*stripeinfo));
 656         if (!stripeinfo)
 657                 GOTO(out_cbdata, rc = -ENOMEM);
 658
 659         OBD_ALLOC(where, sizeof(*where) * oa_bufs);
 660         if (!where)
 661                 GOTO(out_sinfo, rc = -ENOMEM);
 662
 663         OBD_ALLOC(ioarr, sizeof(*ioarr) * oa_bufs);
 664         if (!ioarr)
 665                 GOTO(out_where, rc = -ENOMEM);
 666
 667         /* This is the only race-free way I can think of to get the refcount
 668          * correct. -phil */
 669         atomic_set(&our_cb->refcount, 0);
 670         our_cb->cb = callback;
 671         our_cb->data = cbd;
 672
 673         for (i = 0; i < oa_bufs; i++) {
 674                 where[i] = lov_stripe_which(lsm, pga[i].off);
 675                 if (stripeinfo[where[i]].bufct++ == 0)
 676                         atomic_inc(&our_cb->refcount);
 677         }
 678
 679         for (i = 0, loi = lsm->lsm_oinfo, si = si_last = stripeinfo;
 680              i < stripe_count;
 681              i++, loi++, si_last = si, si++) {
 682                 if (i > 0)
 683                         si->index = si_last->index + si_last->bufct;
 684                 si->lsm.lsm_object_id = loi->loi_id;
 685                 si->ost_idx = loi->loi_ost_idx;
 686         }
 687
 688         for (i = 0; i < oa_bufs; i++) {
 689                 int which = where[i];
 690                 int shift;
 691
 692                 shift = stripeinfo[which].index + stripeinfo[which].subcount;
 693                 LASSERT(shift < oa_bufs);
 694                 ioarr[shift] = pga[i];
 695                 ioarr[shift].off = lov_offset(lsm, pga[i].off, which);
 696                 stripeinfo[which].subcount++;
 697         }
 698
 699         for (i = 0, si = stripeinfo; i < stripe_count; i++) {
 700                 int shift = si->index;
 701
 702                 if (si->bufct) {
 703                         LASSERT(shift < oa_bufs);
 704                         obd_brw(cmd, &lov->tgts[si->ost_idx].conn,
 705                                 &si->lsm, si->bufct, &ioarr[shift],
 706                                 lov_osc_brw_callback, our_cb);
 707                 }
 708         }
 709
 710         rc = callback(cbd, 0, CB_PHASE_START);
 711
 712         OBD_FREE(ioarr, sizeof(*ioarr) * oa_bufs);
 713  out_where:
 714         OBD_FREE(where, sizeof(*where) * oa_bufs);
 715  out_sinfo:
 716         OBD_FREE(stripeinfo, stripe_count * sizeof(*stripeinfo));
 717  out_cbdata:
 718         OBD_FREE(our_cb, sizeof(*our_cb));
 719         RETURN(rc);
 720 }
 721
 722 static int lov_enqueue(struct lustre_handle *conn, struct lov_stripe_md *lsm,
 723                        struct lustre_handle *parent_lock,
 724                        __u32 type, void *cookie, int cookielen, __u32 mode,
 725                        int *flags, void *cb, void *data, int datalen,
 726                        struct lustre_handle *lockhs)
 727 {
 728         struct obd_export *export = class_conn2export(conn);
 729         struct lov_obd *lov;
 730         struct lov_oinfo *loi;
 731         int rc = 0, i;
 732         ENTRY;
 733
 734         if (!lsm) {
 735                 CERROR("LOV requires striping ea for desctruction\n");
 736                 RETURN(-EINVAL);
 737         }
 738
 739         if (!export || !export->exp_obd)
 740                 RETURN(-ENODEV);
 741
 742         lov = &export->exp_obd->u.lov;
 743         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 744                 struct ldlm_extent *extent = (struct ldlm_extent *)cookie;
 745                 struct ldlm_extent sub_ext;
 746                 struct lov_stripe_md submd;
 747
 748                 sub_ext.start = lov_offset(lsm, extent->start, i);
 749                 sub_ext.end = lov_offset(lsm, extent->end, i);
 750                 if (sub_ext.start == sub_ext.end)
 751                         continue;
 752
 753                 submd.lsm_object_id = loi->loi_id;
 754                 /* XXX submd lsm_mds_easize should be that from the subobj,
 755                  *     and the subobj should get it opaquely from the LOV.
 756                  */
 757                 submd.lsm_mds_easize = lov_mds_md_size(lsm->lsm_ost_count);
 758                 submd.lsm_stripe_count = 0;
 759                 /* XXX submd is not fully initialized here */
 760                 rc = obd_enqueue(&(lov->tgts[loi->loi_ost_idx].conn), &submd,
 761                                  parent_lock, type, &sub_ext, sizeof(sub_ext),
 762                                  mode, flags, cb, data, datalen, &(lockhs[i]));
 763                 // XXX add a lock debug statement here
 764                 if (rc)
 765                         CERROR("Error enqueue objid "LPX64" subobj "LPX64
 766                                " on OST idx %d: rc = %d\n", lsm->lsm_object_id,
 767                                loi->loi_id, loi->loi_ost_idx, rc);
 768         }
 769         RETURN(rc);
 770 }
 771
 772 static int lov_cancel(struct lustre_handle *conn, struct lov_stripe_md *lsm,
 773                       __u32 mode, struct lustre_handle *lockhs)
 774 {
 775         struct obd_export *export = class_conn2export(conn);
 776         struct lov_obd *lov;
 777         struct lov_oinfo *loi;
 778         int rc = 0, i;
 779         ENTRY;
 780
 781         if (!lsm) {
 782                 CERROR("LOV requires striping ea for lock cancellation\n");
 783                 RETURN(-EINVAL);
 784         }
 785
 786         if (!export || !export->exp_obd)
 787                 RETURN(-ENODEV);
 788
 789         lov = &export->exp_obd->u.lov;
 790         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
 791                 struct lov_stripe_md submd;
 792
 793                 if (lockhs[i].addr == 0)
 794                         continue;
 795
 796                 submd.lsm_object_id = loi->loi_id;
 797                 submd.lsm_mds_easize = lov_mds_md_size(lsm->lsm_ost_count);
 798                 submd.lsm_stripe_count = 0;
 799                 rc = obd_cancel(&lov->tgts[loi->loi_ost_idx].conn, &submd,
 800                                 mode, &lockhs[i]);
 801                 if (rc)
 802                         CERROR("Error cancel objid "LPX64" subobj "LPX64
 803                                " on OST idx %d: rc = %d\n", lsm->lsm_object_id,
 804                                loi->loi_id, loi->loi_ost_idx, rc);
 805         }
 806         RETURN(rc);
 807 }
 808
 809 static int lov_statfs(struct lustre_handle *conn, struct obd_statfs *osfs)
 810 {
 811         struct obd_export *export = class_conn2export(conn);
 812         struct lov_obd *lov;
 813         struct obd_statfs lov_sfs;
 814         int set = 0;
 815         int rc = 0;
 816         int i;
 817         ENTRY;
 818
 819         if (!export || !export->exp_obd)
 820                 RETURN(-ENODEV);
 821
 822         lov = &export->exp_obd->u.lov;
 823
 824         /* We only get block data from the OBD */
 825         for (i = 0 ; i < lov->desc.ld_tgt_count; i++) {
 826                 int err;
 827
 828                 err = obd_statfs(&lov->tgts[i].conn, &lov_sfs);
 829                 if (err) {
 830                         CERROR("Error statfs OSC %s idx %d: err = %d\n",
 831                                lov->tgts[i].uuid, i, err);
 832                         if (!rc)
 833                                 rc = err;
 834                         continue; /* XXX or break? - probably OK to continue */
 835                 }
 836                 if (!set) {
 837                         memcpy(osfs, &lov_sfs, sizeof(lov_sfs));
 838                         set = 1;
 839                 } else {
 840                         osfs->os_bfree += lov_sfs.os_bfree;
 841                         osfs->os_bavail += lov_sfs.os_bavail;
 842                         osfs->os_blocks += lov_sfs.os_blocks;
 843                         /* XXX not sure about this one - depends on policy.
 844                          *   - could be minimum if we always stripe on all OBDs
 845                          *     (but that would be wrong for any other policy,
 846                          *     if one of the OBDs has no more objects left)
 847                          *   - could be sum if we stripe whole objects
 848                          *   - could be average, just to give a nice number
 849                          *   - we just pick first OST and hope it is enough
 850                         sfs->f_ffree += lov_sfs.f_ffree;
 851                          */
 852                 }
 853         }
 854         RETURN(rc);
 855 }
 856
 857
 858 struct obd_ops lov_obd_ops = {
 859         o_setup:       lov_setup,
 860         o_connect:     lov_connect,
 861         o_disconnect:  lov_disconnect,
 862         o_create:      lov_create,
 863         o_destroy:     lov_destroy,
 864         o_getattr:     lov_getattr,
 865         o_setattr:     lov_setattr,
 866         o_statfs:      lov_statfs,
 867         o_open:        lov_open,
 868         o_close:       lov_close,
 869         o_brw:         lov_brw,
 870         o_punch:       lov_punch,
 871         o_enqueue:     lov_enqueue,
 872         o_cancel:      lov_cancel
 873 };
 874
 875
 876 #define LOV_VERSION "v0.1"
 877
 878 static int __init lov_init(void)
 879 {
 880         printk(KERN_INFO "Lustre Logical Object Volume driver " LOV_VERSION
 881                ", info@clusterfs.com\n");
 882         return class_register_type(&lov_obd_ops, OBD_LOV_DEVICENAME);
 883 }
 884
 885 static void __exit lov_exit(void)
 886 {
 887         class_unregister_type(OBD_LOV_DEVICENAME);
 888 }
 889
 890 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
 891 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver v0.1");
 892 MODULE_LICENSE("GPL");
 893
 894 module_init(lov_init);
 895 module_exit(lov_exit);