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