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