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