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