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