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  * Copyright (C) 2002, 2003 Cluster File Systems, Inc.
   5  * Author: Phil Schwan <phil@clusterfs.com>
   6  *         Peter Braam <braam@clusterfs.com>
   7  *         Mike Shaver <shaver@clusterfs.com>
   8  *
   9  *   This file is part of Lustre, http://www.lustre.org.
  10  *
  11  *   Lustre is free software; you can redistribute it and/or
  12  *   modify it under the terms of version 2 of the GNU General Public
  13  *   License as published by the Free Software Foundation.
  14  *
  15  *   Lustre is distributed in the hope that it will be useful,
  16  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
  17  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  18  *   GNU General Public License for more details.
  19  *
  20  *   You should have received a copy of the GNU General Public License
  21  *   along with Lustre; if not, write to the Free Software
  22  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  23  */
  24
  25 #ifndef EXPORT_SYMTAB
  26 # define EXPORT_SYMTAB
  27 #endif
  28 #define DEBUG_SUBSYSTEM S_LOV
  29 #ifdef __KERNEL__
  30 #include <linux/slab.h>
  31 #include <linux/module.h>
  32 #include <linux/init.h>
  33 #include <linux/slab.h>
  34 #include <linux/pagemap.h>
  35 #include <linux/seq_file.h>
  36 #include <asm/div64.h>
  37 #else
  38 #include <liblustre.h>
  39 #endif
  40
  41 #include <linux/obd_support.h>
  42 #include <linux/lustre_lib.h>
  43 #include <linux/lustre_net.h>
  44 #include <linux/lustre_idl.h>
  45 #include <linux/lustre_dlm.h>
  46 #include <linux/lustre_mds.h>
  47 #include <linux/obd_class.h>
  48 #include <linux/obd_lov.h>
  49 #include <linux/obd_ost.h>
  50 #include <linux/lprocfs_status.h>
  51 #include <linux/lustre_audit.h>
  52 #include "lov_internal.h"
  53
  54 /* obd methods */
  55 #define MAX_STRING_SIZE 128
  56 static int lov_connect_obd(struct obd_device *obd, struct lov_tgt_desc *tgt,
  57                            int activate, struct obd_connect_data *conn_data,
  58                            unsigned long connect_flags)
  59 {
  60         struct obd_uuid lov_osc_uuid = { "LOV_OSC_UUID" };
  61         struct obd_uuid *tgt_uuid = &tgt->uuid;
  62
  63 #ifdef __KERNEL__
  64         struct proc_dir_entry *lov_proc_dir;
  65 #endif
  66         struct lov_obd *lov = &obd->u.lov;
  67         struct lustre_handle conn = {0, };
  68         struct obd_device *tgt_obd;
  69         int rc;
  70         ENTRY;
  71
  72         tgt_obd = class_find_client_obd(tgt_uuid, OBD_OSC_DEVICENAME,
  73                                         &obd->obd_uuid);
  74
  75         if (!tgt_obd) {
  76                 CERROR("Target %s not attached\n", tgt_uuid->uuid);
  77                 RETURN(-EINVAL);
  78         }
  79
  80         if (!tgt_obd->obd_set_up) {
  81                 CERROR("Target %s not set up\n", tgt_uuid->uuid);
  82                 RETURN(-EINVAL);
  83         }
  84
  85         if (activate) {
  86                 tgt_obd->obd_no_recov = 0;
  87                 ptlrpc_activate_import(tgt_obd->u.cli.cl_import);
  88         }
  89
  90         if (tgt_obd->u.cli.cl_import->imp_invalid) {
  91                 CERROR("not connecting OSC %s; administratively "
  92                        "disabled\n", tgt_uuid->uuid);
  93                 rc = obd_register_observer(tgt_obd, obd);
  94                 if (rc) {
  95                         CERROR("Target %s register_observer error %d; "
  96                                "will not be able to reactivate\n",
  97                                tgt_uuid->uuid, rc);
  98                 }
  99                 RETURN(0);
 100         }
 101
 102         rc = obd_connect(&conn, tgt_obd, &lov_osc_uuid, conn_data,
 103                          connect_flags);
 104         if (rc) {
 105                 CERROR("Target %s connect error %d\n", tgt_uuid->uuid, rc);
 106                 RETURN(rc);
 107         }
 108         tgt->ltd_exp = class_conn2export(&conn);
 109
 110         rc = obd_register_observer(tgt_obd, obd);
 111         if (rc) {
 112                 CERROR("Target %s register_observer error %d\n",
 113                        tgt_uuid->uuid, rc);
 114                 obd_disconnect(tgt->ltd_exp, 0);
 115                 tgt->ltd_exp = NULL;
 116                 RETURN(rc);
 117         }
 118
 119         lov_tgt_set_flags(lov, tgt, LTD_ACTIVE);
 120
 121 #ifdef __KERNEL__
 122         lov_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 123         if (lov_proc_dir) {
 124                 struct obd_device *osc_obd = class_conn2obd(&conn);
 125                 struct proc_dir_entry *osc_symlink;
 126                 char name[MAX_STRING_SIZE + 1];
 127
 128                 LASSERT(osc_obd != NULL);
 129                 LASSERT(osc_obd->obd_type != NULL);
 130                 LASSERT(osc_obd->obd_type->typ_name != NULL);
 131                 name[MAX_STRING_SIZE] = '\0';
 132                 snprintf(name, MAX_STRING_SIZE, "../../../%s/%s",
 133                          osc_obd->obd_type->typ_name,
 134                          osc_obd->obd_name);
 135                 osc_symlink = proc_symlink(osc_obd->obd_name, lov_proc_dir,
 136                                            name);
 137                 if (osc_symlink == NULL) {
 138                         CERROR("could not register LOV target "
 139                                "/proc/fs/lustre/%s/%s/target_obds/%s\n",
 140                                obd->obd_type->typ_name, obd->obd_name,
 141                                osc_obd->obd_name);
 142                         lprocfs_remove(lov_proc_dir);
 143                         lov_proc_dir = NULL;
 144                 }
 145         }
 146 #endif
 147
 148         RETURN(0);
 149 }
 150
 151 static int lov_connect(struct lustre_handle *conn, struct obd_device *obd,
 152                        struct obd_uuid *cluuid, struct obd_connect_data *data,
 153                        unsigned long flags)
 154 {
 155 #ifdef __KERNEL__
 156         struct proc_dir_entry *lov_proc_dir;
 157 #endif
 158         struct lov_obd *lov = &obd->u.lov;
 159         struct lov_tgt_desc *tgt;
 160         struct obd_export *exp;
 161         int rc, rc2, i;
 162         ENTRY;
 163
 164         rc = class_connect(conn, obd, cluuid);
 165         if (rc)
 166                 RETURN(rc);
 167
 168         exp = class_conn2export(conn);
 169
 170         /* We don't want to actually do the underlying connections more than
 171          * once, so keep track. */
 172         lov->refcount++;
 173         if (lov->refcount > 1) {
 174                 class_export_put(exp);
 175                 RETURN(0);
 176         }
 177
 178 #ifdef __KERNEL__
 179         lov_proc_dir = lprocfs_register("target_obds", obd->obd_proc_entry,
 180                                         NULL, NULL);
 181         if (IS_ERR(lov_proc_dir)) {
 182                 CERROR("could not register /proc/fs/lustre/%s/%s/target_obds.",
 183                        obd->obd_type->typ_name, obd->obd_name);
 184                 lov_proc_dir = NULL;
 185         }
 186 #endif
 187
 188         /* connect_flags is the MDS number, save for use in lov_add_obd */
 189         lov->lov_connect_flags = flags;
 190         for (i = 0, tgt = lov->tgts; i < lov->desc.ld_tgt_count; i++, tgt++) {
 191                 if (obd_uuid_empty(&tgt->uuid))
 192                         continue;
 193                 rc = lov_connect_obd(obd, tgt, 0, data, flags);
 194                 if (rc)
 195                         GOTO(out_disc, rc);
 196         }
 197
 198         class_export_put(exp);
 199         RETURN (0);
 200
 201  out_disc:
 202 #ifdef __KERNEL__
 203         if (lov_proc_dir)
 204                 lprocfs_remove(lov_proc_dir);
 205 #endif
 206
 207         while (i-- > 0) {
 208                 struct obd_uuid uuid;
 209                 --tgt;
 210                 lov_tgt_clear_flags(lov, tgt, LTD_ACTIVE);
 211
 212                 /* save for CERROR below; (we know it's terminated) */
 213                 uuid = tgt->uuid;
 214                 rc2 = obd_disconnect(tgt->ltd_exp, 0);
 215                 if (rc2)
 216                         CERROR("error: LOV target %s disconnect on OST idx %d: "
 217                                "rc = %d\n", uuid.uuid, i, rc2);
 218         }
 219         class_disconnect(exp, 0);
 220         RETURN (rc);
 221 }
 222
 223 static int lov_disconnect_obd(struct obd_device *obd,
 224                               struct lov_tgt_desc *tgt,
 225                               unsigned long flags)
 226 {
 227 #ifdef __KERNEL__
 228         struct proc_dir_entry *lov_proc_dir;
 229 #endif
 230         struct obd_device *osc_obd = class_exp2obd(tgt->ltd_exp);
 231         struct lov_obd *lov = &obd->u.lov;
 232         int rc;
 233         ENTRY;
 234
 235 #ifdef __KERNEL__
 236         lov_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 237         if (lov_proc_dir) {
 238                 struct proc_dir_entry *osc_symlink;
 239
 240                 osc_symlink = lprocfs_srch(lov_proc_dir, osc_obd->obd_name);
 241                 if (osc_symlink) {
 242                         lprocfs_remove(osc_symlink);
 243                 } else {
 244                         CERROR("/proc/fs/lustre/%s/%s/target_obds/%s missing\n",
 245                                obd->obd_type->typ_name, obd->obd_name,
 246                                osc_obd->obd_name);
 247                 }
 248         }
 249 #endif
 250         if (obd->obd_no_recov) {
 251                 /* Pass it on to our clients.
 252                  * XXX This should be an argument to disconnect,
 253                  * XXX not a back-door flag on the OBD.  Ah well.
 254                  */
 255                 if (osc_obd)
 256                         osc_obd->obd_no_recov = 1;
 257         }
 258
 259         /* XXX - shouldn't this be after the obd_disconnect() ? */
 260         obd_register_observer(tgt->ltd_exp->exp_obd, NULL);
 261
 262         rc = obd_disconnect(tgt->ltd_exp, flags);
 263         if (rc) {
 264                 if (lov_tgt_active(lov, tgt, 0)) {
 265                         lov_tgt_decref(lov, tgt);
 266                         CERROR("Target %s disconnect error %d\n",
 267                                tgt->uuid.uuid, rc);
 268                 }
 269         }
 270
 271         CDEBUG(D_CONFIG, "idx: %d flags: 0x%x active: %d\n",
 272                tgt - lov->tgts, tgt->ltd_flags,
 273                lov->desc.ld_active_tgt_count);
 274
 275         lov_tgt_clear_flags(lov, tgt, LTD_ACTIVE);
 276         tgt->ltd_exp = NULL;
 277         RETURN(0);
 278 }
 279
 280 static int lov_disconnect(struct obd_export *exp, unsigned long flags)
 281 {
 282         struct obd_device *obd = class_exp2obd(exp);
 283 #ifdef __KERNEL__
 284         struct proc_dir_entry *lov_proc_dir;
 285 #endif
 286         struct lov_obd *lov = &obd->u.lov;
 287         struct lov_tgt_desc *tgt;
 288         int rc, i;
 289         ENTRY;
 290
 291         if (!lov->tgts)
 292                 goto out_local;
 293
 294         /* Only disconnect the underlying layers on the final disconnect. */
 295         lov->refcount--;
 296         if (lov->refcount != 0)
 297                 goto out_local;
 298
 299         for (i = 0, tgt = lov->tgts; i < lov->desc.ld_tgt_count; i++, tgt++) {
 300                 if (tgt->ltd_exp)
 301                         lov_disconnect_obd(obd, tgt, flags);
 302         }
 303
 304 #ifdef __KERNEL__
 305         lov_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 306         if (lov_proc_dir) {
 307                 lprocfs_remove(lov_proc_dir);
 308         } else {
 309                 CERROR("/proc/fs/lustre/%s/%s/target_obds missing.",
 310                        obd->obd_type->typ_name, obd->obd_name);
 311         }
 312 #endif
 313
 314  out_local:
 315         rc = class_disconnect(exp, 0);
 316         RETURN(rc);
 317 }
 318
 319 /* Error codes:
 320  *
 321  *  -EINVAL  : UUID can't be found in the LOV's target list
 322  *  -ENOTCONN: The UUID is found, but the target connection is bad (!)
 323  *  -EBADF   : The UUID is found, but the OBD is the wrong type (!)
 324  */
 325 static int lov_set_osc_active(struct lov_obd *lov, struct obd_uuid *uuid,
 326                               int activate)
 327 {
 328         struct lov_tgt_desc *tgt;
 329         int i, rc = 0;
 330         ENTRY;
 331
 332         CDEBUG(D_INFO, "Searching in lov %p for uuid %s (activate=%d)\n",
 333                lov, uuid->uuid, activate);
 334
 335         lov_tgts_lock(lov);
 336         for (i = 0, tgt = lov->tgts; i < lov->desc.ld_tgt_count; i++, tgt++) {
 337                 if (tgt->ltd_exp == NULL)
 338                         continue;
 339
 340                 CDEBUG(D_INFO, "lov idx %d is %s conn "LPX64"\n",
 341                        i, tgt->uuid.uuid, tgt->ltd_exp->exp_handle.h_cookie);
 342
 343                 if (obd_uuid_equals(uuid, &tgt->uuid))
 344                         break;
 345         }
 346
 347         if (i == lov->desc.ld_tgt_count)
 348                 GOTO(out, rc = -EINVAL);
 349
 350         if (activate) {
 351                 if (tgt->ltd_flags & LTD_ACTIVE) {
 352                         CDEBUG(D_CONFIG|D_INFO, "OSC %s already active!\n",
 353                                uuid->uuid);
 354                         GOTO(out, rc);
 355                 }
 356                 CDEBUG(D_INFO, "Marking OSC %s active\n", uuid->uuid);
 357                 tgt->ltd_flags |= LTD_ACTIVE;
 358                 lov->desc.ld_active_tgt_count++;
 359         } else {
 360                 if ((tgt->ltd_flags & LTD_ACTIVE) == 0) {
 361                         CDEBUG(D_CONFIG|D_INFO, "OSC %s already inactive!\n",
 362                                uuid->uuid);
 363                         GOTO(out, rc);
 364                 }
 365                 CDEBUG(D_INFO, "Marking OSC %s inactive\n", uuid->uuid);
 366                 tgt->ltd_flags &= ~LTD_ACTIVE;
 367                 lov->desc.ld_active_tgt_count--;
 368         }
 369
 370         EXIT;
 371  out:
 372         lov_tgts_unlock(lov);
 373         return rc;
 374 }
 375
 376 static int lov_notify(struct obd_device *obd, struct obd_device *watched,
 377                       int active, void *data)
 378 {
 379         struct obd_uuid *uuid;
 380         int rc;
 381         ENTRY;
 382
 383         if (strcmp(watched->obd_type->typ_name, OBD_OSC_DEVICENAME)) {
 384                 CERROR("unexpected notification of %s %s!\n",
 385                        watched->obd_type->typ_name,
 386                        watched->obd_name);
 387                 return -EINVAL;
 388         }
 389
 390         uuid = &watched->u.cli.cl_import->imp_target_uuid;
 391
 392         /* Set OSC as active before notifying the observer, so the
 393          * observer can use the OSC normally.
 394          */
 395         rc = lov_set_osc_active(&obd->u.lov, uuid, active);
 396         if (rc) {
 397                 CERROR("%sactivation of %s failed: %d\n",
 398                        active ? "" : "de", uuid->uuid, rc);
 399                 RETURN(rc);
 400         }
 401
 402         if (obd->obd_observer)
 403                 /* Pass the notification up the chain. */
 404                 rc = obd_notify(obd->obd_observer, watched, active, data);
 405
 406         RETURN(rc);
 407 }
 408
 409 int lov_attach(struct obd_device *dev, obd_count len, void *data)
 410 {
 411         struct lprocfs_static_vars lvars;
 412         int rc;
 413
 414         lprocfs_init_vars(lov, &lvars);
 415         rc = lprocfs_obd_attach(dev, lvars.obd_vars);
 416         if (rc == 0) {
 417 #ifdef __KERNEL__
 418                 struct proc_dir_entry *entry;
 419
 420                 entry = create_proc_entry("target_obd_status", 0444,
 421                                           dev->obd_proc_entry);
 422                 if (entry == NULL) {
 423                         rc = -ENOMEM;
 424                 } else {
 425                         entry->proc_fops = &lov_proc_target_fops;
 426                         entry->data = dev;
 427                 }
 428 #endif
 429         }
 430         return rc;
 431 }
 432
 433 int lov_detach(struct obd_device *dev)
 434 {
 435         return lprocfs_obd_detach(dev);
 436 }
 437
 438 static int lov_setup(struct obd_device *obd, obd_count len, void *buf)
 439 {
 440         struct lov_obd *lov = &obd->u.lov;
 441         struct lustre_cfg *lcfg = buf;
 442         struct lov_desc *desc;
 443         int count;
 444         ENTRY;
 445
 446         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
 447                 CERROR("LOV setup requires a descriptor\n");
 448                 RETURN(-EINVAL);
 449         }
 450
 451         desc = (struct lov_desc *)lustre_cfg_string(lcfg, 1);
 452         if (sizeof(*desc) > LUSTRE_CFG_BUFLEN(lcfg, 1)) {
 453                 CERROR("descriptor size wrong: %d > %d\n",
 454                        (int)sizeof(*desc), LUSTRE_CFG_BUFLEN(lcfg, 1));
 455                 RETURN(-EINVAL);
 456         }
 457
 458         /* Because of 64-bit divide/mod operations only work with a 32-bit
 459          * divisor in a 32-bit kernel, we cannot support a stripe width
 460          * of 4GB or larger on 32-bit CPUs.
 461          */
 462
 463         count = desc->ld_default_stripe_count;
 464         if (count && (count * desc->ld_default_stripe_size) > ~0UL) {
 465                 CERROR("LOV: stripe width "LPU64"x%u > %lu on 32-bit system\n",
 466                        desc->ld_default_stripe_size, count, ~0UL);
 467                 RETURN(-EINVAL);
 468         }
 469
 470         lov->bufsize = sizeof(struct lov_tgt_desc) * LOV_MAX_TGT_COUNT;
 471         OBD_ALLOC(lov->tgts, lov->bufsize);
 472         if (lov->tgts == NULL) {
 473                 lov->bufsize = 0;
 474                 CERROR("couldn't allocate %d bytes for target table.\n",
 475                        lov->bufsize);
 476                 RETURN(-EINVAL);
 477         }
 478
 479         desc->ld_tgt_count = 0;
 480         desc->ld_active_tgt_count = 0;
 481         lov->desc = *desc;
 482         spin_lock_init(&lov->lov_lock);
 483         sema_init(&lov->lov_llog_sem, 1);
 484         init_waitqueue_head(&lov->lov_tgt_waitq);
 485
 486         RETURN(0);
 487 }
 488
 489 static int lov_cleanup(struct obd_device *obd, int flags)
 490 {
 491         struct lov_obd *lov = &obd->u.lov;
 492
 493         OBD_FREE(lov->tgts, lov->bufsize);
 494         RETURN(0);
 495 }
 496
 497 static int
 498 lov_add_obd(struct obd_device *obd, struct obd_uuid *uuidp, int index, int gen)
 499 {
 500         struct lov_obd *lov = &obd->u.lov;
 501         struct lov_tgt_desc *tgt;
 502         int rc;
 503         ENTRY;
 504
 505         CDEBUG(D_CONFIG, "uuid: %s idx: %d gen: %d\n",
 506                uuidp->uuid, index, gen);
 507
 508         if ((index < 0) || (index >= LOV_MAX_TGT_COUNT)) {
 509                 CERROR("request to add OBD %s at invalid index: %d\n",
 510                        uuidp->uuid, index);
 511                 RETURN(-EINVAL);
 512         }
 513
 514         if (gen <= 0) {
 515                 CERROR("request to add OBD %s with invalid generation: %d\n",
 516                        uuidp->uuid, gen);
 517                 RETURN(-EINVAL);
 518         }
 519
 520         tgt = lov->tgts + index;
 521
 522         lov_tgts_lock(lov);
 523
 524         if (!obd_uuid_empty(&tgt->uuid)) {
 525                 lov_tgts_unlock(lov);
 526                 CERROR("OBD already assigned at LOV target index %d\n",
 527                        index);
 528                 RETURN(-EEXIST);
 529         }
 530
 531         tgt->uuid = *uuidp;
 532         /* XXX - add a sanity check on the generation number. */
 533         tgt->ltd_gen = gen;
 534         tgt->ltd_flags = 0;
 535
 536         if (index >= lov->desc.ld_tgt_count)
 537                 lov->desc.ld_tgt_count = index + 1;
 538
 539         lov_tgts_unlock(lov);
 540
 541         CDEBUG(D_CONFIG, "idx: %d ltd_gen: %d ld_tgt_count: %d\n",
 542                 index, tgt->ltd_gen, lov->desc.ld_tgt_count);
 543
 544         if (lov->refcount == 0)
 545                 RETURN(0);
 546
 547         /* We don't need to lock the tgt entry here because the
 548          * config events are single threaded. Therefore only the
 549          * current thread can be changing the tgts tbl. */
 550         if (tgt->ltd_exp) {
 551                 struct obd_device *osc_obd;
 552
 553                 osc_obd = class_exp2obd(tgt->ltd_exp);
 554                 if (osc_obd)
 555                         osc_obd->obd_no_recov = 0;
 556         }
 557
 558         rc = lov_connect_obd(obd, tgt, 0, NULL, lov->lov_connect_flags);
 559         if (rc)
 560                 GOTO(out, rc);
 561
 562         if (obd->obd_observer) {
 563                 /* tell the mds_lov about the new target */
 564                 rc = obd_notify(obd->obd_observer, tgt->ltd_exp->exp_obd, 1,
 565                                 (void *)index);
 566         }
 567
 568         GOTO(out, rc);
 569  out:
 570         if (rc && tgt->ltd_exp != NULL)
 571                 lov_disconnect_obd(obd, tgt, 0);
 572         return rc;
 573 }
 574
 575 static int
 576 lov_del_obd(struct obd_device *obd, struct obd_uuid *uuidp, int index, int gen,
 577             struct obd_export *md_exp)
 578 {
 579         struct lov_obd *lov = &obd->u.lov;
 580         struct lov_tgt_desc *tgt;
 581         struct l_wait_info lwi = { 0 };
 582         int count = lov->desc.ld_tgt_count;
 583         int rc = 0;
 584         ENTRY;
 585
 586         CDEBUG(D_CONFIG, "uuid: %s idx: %d gen: %d\n",
 587                uuidp->uuid, index, gen);
 588
 589         if (index >= count) {
 590                 CERROR("LOV target index %d >= number of LOV OBDs %d.\n",
 591                        index, count);
 592                 RETURN(-EINVAL);
 593         }
 594
 595         tgt = lov->tgts + index;
 596
 597         if (obd_uuid_empty(&tgt->uuid)) {
 598                 CERROR("LOV target at index %d is not setup.\n", index);
 599                 RETURN(-EINVAL);
 600         }
 601
 602         if (!obd_uuid_equals(uuidp, &tgt->uuid)) {
 603                 CERROR("LOV target UUID %s at index %d doesn't match %s.\n",
 604                        tgt->uuid.uuid, index, uuidp->uuid);
 605                 RETURN(-EINVAL);
 606         }
 607
 608         lov_tgt_set_flags(lov, tgt, LTD_DEL_PENDING);
 609
 610         if (tgt->ltd_exp) {
 611                 struct obd_device *osc_obd;
 612
 613                 rc = obd_cancel_unused(tgt->ltd_exp, NULL,
 614                                        LDLM_FL_CONFIG_CHANGE, NULL);
 615                 if (rc != 0)
 616                         CWARN("obd_cancel_unused(osc): %d\n", rc);
 617
 618                 if (md_exp) {
 619                         rc = obd_cancel_unused(md_exp, NULL,
 620                                        LDLM_FL_CONFIG_CHANGE, NULL);
 621                         if (rc != 0)
 622                                 CWARN("obd_cancel_unused(md): %d\n", rc);
 623                 }
 624
 625                 osc_obd = class_exp2obd(tgt->ltd_exp);
 626                 if (osc_obd) {
 627                         osc_obd->obd_no_recov = 1;
 628                         rc = obd_llog_finish(osc_obd, &osc_obd->obd_llogs, 1);
 629                         if (rc)
 630                                 CERROR("osc_llog_finish error: %d\n", rc);
 631                 }
 632
 633                 lov_disconnect_obd(obd, tgt, 0);
 634         }
 635
 636         CDEBUG(D_CONFIG, "Sleeping until LOV target index %d UUID %s "
 637                "is quiesced; ltd_refcount: %d ld_active_tgt_count: %d.\n",
 638                index, uuidp->uuid, tgt->ltd_refcount,
 639                lov->desc.ld_active_tgt_count);
 640
 641         /* XXX - should we set a timeout ? */
 642         lwi = LWI_TIMEOUT_INTR(0, NULL, NULL, NULL);
 643         rc = l_wait_event(lov->lov_tgt_waitq, (tgt->ltd_refcount == 0), &lwi);
 644         if (rc) {
 645                 lov_tgt_clear_flags(lov, tgt, LTD_DEL_PENDING);
 646                 CERROR("LOV target delete UUID %s index %d aborted: %d.\n",
 647                        uuidp->uuid, index, rc);
 648                 RETURN(rc);
 649         }
 650
 651         CDEBUG(D_CONFIG, "LOV target index %d UUID %s is quiesced; "
 652                "ltd_refcount: %d ld_active_tgt_count: %d.\n",
 653                index, uuidp->uuid, tgt->ltd_refcount,
 654                lov->desc.ld_active_tgt_count);
 655
 656         /* XXX - right now there is a dependency on ld_tgt_count being the
 657          * maximum tgt index for computing the mds_max_easize. So we can't
 658          * shrink it. */
 659
 660         /* lt_gen = 0 will mean it will not match the gen of any valid loi */
 661         memset(tgt, 0, sizeof(*tgt));
 662         RETURN(rc);
 663 }
 664
 665 static int lov_process_config(struct obd_device *obd, obd_count len, void *buf)
 666 {
 667         struct lustre_cfg *lcfg = buf;
 668         struct obd_uuid obd_uuid;
 669         int cmd;
 670         int index;
 671         int gen;
 672         int rc = 0;
 673         ENTRY;
 674
 675         switch(cmd = lcfg->lcfg_command) {
 676         case LCFG_LOV_ADD_OBD:
 677         case LCFG_LOV_DEL_OBD: {
 678                 if (LUSTRE_CFG_BUFLEN(lcfg, 1) > sizeof(obd_uuid.uuid))
 679                         GOTO(out, rc = -EINVAL);
 680
 681                 obd_str2uuid(&obd_uuid, lustre_cfg_string(lcfg, 1));
 682
 683                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
 684                         GOTO(out, rc = -EINVAL);
 685                 if (sscanf(lustre_cfg_buf(lcfg, 3), "%d", &gen) != 1)
 686                         GOTO(out, rc = -EINVAL);
 687                 if (cmd == LCFG_LOV_ADD_OBD)
 688                         rc = lov_add_obd(obd, &obd_uuid, index, gen);
 689                 else {
 690                         struct obd_export *md_exp = (struct obd_export *)lcfg->lcfg_nal;
 691                         rc = lov_del_obd(obd, &obd_uuid, index, gen, md_exp);
 692                 }
 693                 GOTO(out, rc);
 694         }
 695         default: {
 696                 CERROR("Unknown command: %d\n", lcfg->lcfg_command);
 697                 GOTO(out, rc = -EINVAL);
 698
 699         }
 700         }
 701 out:
 702         RETURN(rc);
 703 }
 704
 705 #ifndef log2
 706 #define log2(n) ffz(~(n))
 707 #endif
 708
 709 static int lov_clear_orphans(struct obd_export *export,
 710                              struct obdo *src_oa,
 711                              struct lov_stripe_md **ea,
 712                              struct obd_trans_info *oti)
 713 {
 714         struct lov_obd *lov;
 715         struct obdo *tmp_oa;
 716         struct obd_uuid *ost_uuid = NULL;
 717         struct lov_tgt_desc *tgt;
 718         int rc = 0, i;
 719         ENTRY;
 720
 721         LASSERT(src_oa->o_valid & OBD_MD_FLFLAGS &&
 722                 src_oa->o_flags == OBD_FL_DELORPHAN);
 723
 724         lov = &export->exp_obd->u.lov;
 725
 726         tmp_oa = obdo_alloc();
 727         if (tmp_oa == NULL)
 728                 RETURN(-ENOMEM);
 729
 730         if (src_oa->o_valid & OBD_MD_FLINLINE) {
 731                 ost_uuid = (struct obd_uuid *)src_oa->o_inline;
 732                 CDEBUG(D_HA, "clearing orphans only for %s\n",
 733                        ost_uuid->uuid);
 734         }
 735
 736         for (i = 0, tgt = lov->tgts; i < lov->desc.ld_tgt_count; i++, tgt++) {
 737                 struct lov_stripe_md obj_md;
 738                 struct lov_stripe_md *obj_mdp = &obj_md;
 739                 int active = 0;
 740                 int err;
 741
 742                 /* if called for a specific target, we don't
 743                    care if it is not active. */
 744                 if (lov_tgt_active(lov, tgt, 0)) {
 745                         active = 1;
 746                 } else if (ost_uuid == NULL) {
 747                         CDEBUG(D_HA, "lov idx %d inactive\n", i);
 748                         continue;
 749                 }
 750
 751                 if (ost_uuid && !obd_uuid_equals(ost_uuid, &tgt->uuid)) {
 752                         if (active)
 753                                 lov_tgt_decref(lov, tgt);
 754                         continue;
 755                 }
 756
 757                 /*
 758                  * setting up objid OSS objects should be destroyed starting
 759                  * from it.
 760                  */
 761                 memcpy(tmp_oa, src_oa, sizeof(*tmp_oa));
 762                 tmp_oa->o_valid |= OBD_MD_FLID;
 763                 tmp_oa->o_id = oti->oti_objid[i];
 764
 765                 /* XXX: LOV STACKING: use real "obj_mdp" sub-data */
 766                 err = obd_create(tgt->ltd_exp, tmp_oa, NULL, 0, &obj_mdp, oti);
 767                 if (err)
 768                         /* This export will be disabled until it is recovered,
 769                            and then orphan recovery will be completed. */
 770                         CERROR("error in orphan recovery on OST idx %d/%d: "
 771                                "rc = %d\n", i, lov->desc.ld_tgt_count, err);
 772
 773                 if (active)
 774                         lov_tgt_decref(lov, tgt);
 775
 776                 if (ost_uuid)
 777                         break;
 778         }
 779
 780         obdo_free(tmp_oa);
 781         RETURN(rc);
 782 }
 783
 784 /* the LOV expects oa->o_id to be set to the LOV object id */
 785 static int
 786 lov_create(struct obd_export *exp, struct obdo *src_oa,
 787            void *acl, int acl_size, struct lov_stripe_md **ea,
 788            struct obd_trans_info *oti)
 789 {
 790         struct lov_request_set *set = NULL;
 791         struct list_head *pos;
 792         struct lov_obd *lov;
 793         int rc = 0;
 794         ENTRY;
 795
 796         LASSERT(ea != NULL);
 797         if (exp == NULL)
 798                 RETURN(-EINVAL);
 799
 800         if ((src_oa->o_valid & OBD_MD_FLFLAGS) &&
 801             src_oa->o_flags == OBD_FL_DELORPHAN) {
 802                 rc = lov_clear_orphans(exp, src_oa, ea, oti);
 803                 RETURN(rc);
 804         }
 805
 806         LASSERT(oti->oti_flags & OBD_MODE_CROW);
 807
 808         /* main creation loop */
 809         rc = lov_prep_create_set(exp, ea, src_oa, oti, &set);
 810         if (rc)
 811                 RETURN(rc);
 812
 813         lov = &exp->exp_obd->u.lov;
 814
 815         list_for_each (pos, &set->set_list) {
 816                 struct lov_request *req =
 817                         list_entry(pos, struct lov_request, rq_link);
 818
 819                 /* XXX: LOV STACKING: use real "obj_mdp" sub-data */
 820                 rc = obd_create(lov->tgts[req->rq_idx].ltd_exp,
 821                                 req->rq_oa, NULL, 0, &req->rq_md, oti);
 822                 lov_update_create_set(set, req, rc);
 823         }
 824         rc = lov_fini_create_set(set, ea);
 825         RETURN(rc);
 826 }
 827
 828 #define lsm_bad_magic(LSMP)                                     \
 829 ({                                                              \
 830         struct lov_stripe_md *_lsm__ = (LSMP);                  \
 831         int _ret__ = 0;                                         \
 832         if (!_lsm__) {                                          \
 833                 CERROR("LOV requires striping ea\n");           \
 834                 _ret__ = 1;                                     \
 835         } else if (_lsm__->lsm_magic != LOV_MAGIC) {            \
 836                 CERROR("LOV striping magic bad %#x != %#x\n",   \
 837                        _lsm__->lsm_magic, LOV_MAGIC);           \
 838                 _ret__ = 1;                                     \
 839         }                                                       \
 840         _ret__;                                                 \
 841 })
 842
 843 static int lov_destroy(struct obd_export *exp, struct obdo *oa,
 844                        struct lov_stripe_md *lsm, struct obd_trans_info *oti)
 845 {
 846         struct lov_request_set *set;
 847         struct lov_request *req;
 848         struct list_head *pos;
 849         struct lov_obd *lov;
 850         int rc = 0;
 851         ENTRY;
 852
 853         if (lsm_bad_magic(lsm))
 854                 RETURN(-EINVAL);
 855
 856         if (!exp || !exp->exp_obd)
 857                 RETURN(-ENODEV);
 858
 859         lov = &exp->exp_obd->u.lov;
 860         rc = lov_prep_destroy_set(exp, oa, lsm, oti, &set);
 861         if (rc)
 862                 RETURN(rc);
 863
 864         list_for_each (pos, &set->set_list) {
 865                 int err;
 866                 req = list_entry(pos, struct lov_request, rq_link);
 867
 868                 /* XXX update the cookie position */
 869                 oti->oti_logcookies = set->set_cookies + req->rq_stripe;
 870                 rc = obd_destroy(lov->tgts[req->rq_idx].ltd_exp, req->rq_oa,
 871                                  NULL, oti);
 872                 err = lov_update_common_set(set, req, rc);
 873                 if (rc) {
 874                         CERROR("error: destroying objid "LPX64" subobj "
 875                                LPX64" on OST idx %d: rc = %d\n",
 876                                set->set_oa->o_id, req->rq_oa->o_id,
 877                                req->rq_idx, rc);
 878                         if (!rc)
 879                                 rc = err;
 880                 }
 881         }
 882         lov_fini_destroy_set(set);
 883         RETURN(rc);
 884 }
 885
 886 static int lov_getattr(struct obd_export *exp, struct obdo *oa,
 887                        struct lov_stripe_md *lsm)
 888 {
 889         struct lov_request_set *set;
 890         struct lov_request *req;
 891         struct list_head *pos;
 892         struct lov_obd *lov;
 893         int err = 0, rc = 0;
 894         ENTRY;
 895
 896         if (lsm_bad_magic(lsm))
 897                 RETURN(-EINVAL);
 898
 899         if (!exp || !exp->exp_obd)
 900                 RETURN(-ENODEV);
 901
 902         lov = &exp->exp_obd->u.lov;
 903
 904         rc = lov_prep_getattr_set(exp, oa, lsm, &set);
 905         if (rc)
 906                 RETURN(rc);
 907
 908         list_for_each (pos, &set->set_list) {
 909                 req = list_entry(pos, struct lov_request, rq_link);
 910
 911                 CDEBUG(D_INFO, "objid "LPX64"[%d] has subobj "LPX64" at idx "
 912                        "%u\n", oa->o_id, req->rq_stripe, req->rq_oa->o_id,
 913                        req->rq_idx);
 914
 915                 rc = obd_getattr(lov->tgts[req->rq_idx].ltd_exp,
 916                                  req->rq_oa, NULL);
 917                 err = lov_update_common_set(set, req, rc);
 918                 if (err) {
 919                         CERROR("error: getattr objid "LPX64" subobj "
 920                                LPX64" on OST idx %d: rc = %d\n",
 921                                set->set_oa->o_id, req->rq_oa->o_id,
 922                                req->rq_idx, err);
 923                         break;
 924                 }
 925         }
 926
 927         rc = lov_fini_getattr_set(set);
 928         if (err)
 929                 rc = err;
 930         RETURN(rc);
 931 }
 932
 933 static int lov_getattr_interpret(struct ptlrpc_request_set *rqset, void *data,
 934                                  int rc)
 935 {
 936         struct lov_request_set *lovset = (struct lov_request_set *)data;
 937         ENTRY;
 938
 939         /* don't do attribute merge if this aysnc op failed */
 940         if (rc) {
 941                 lovset->set_completes = 0;
 942                 lov_fini_getattr_set(lovset);
 943         } else {
 944                 rc = lov_fini_getattr_set(lovset);
 945         }
 946         RETURN (rc);
 947 }
 948
 949 static int lov_getattr_async(struct obd_export *exp, struct obdo *oa,
 950                               struct lov_stripe_md *lsm,
 951                               struct ptlrpc_request_set *rqset)
 952 {
 953         struct lov_request_set *lovset;
 954         struct lov_obd *lov;
 955         struct list_head *pos;
 956         struct lov_request *req;
 957         int rc = 0;
 958         ENTRY;
 959
 960         if (lsm_bad_magic(lsm))
 961                 RETURN(-EINVAL);
 962
 963         if (!exp || !exp->exp_obd)
 964                 RETURN(-ENODEV);
 965
 966         lov = &exp->exp_obd->u.lov;
 967
 968         rc = lov_prep_getattr_set(exp, oa, lsm, &lovset);
 969         if (rc)
 970                 RETURN(rc);
 971
 972         CDEBUG(D_INFO, "objid "LPX64": %ux%u byte stripes\n",
 973                lsm->lsm_object_id, lsm->lsm_stripe_count, lsm->lsm_stripe_size);
 974
 975         list_for_each (pos, &lovset->set_list) {
 976                 req = list_entry(pos, struct lov_request, rq_link);
 977
 978                 CDEBUG(D_INFO, "objid "LPX64"[%d] has subobj "LPX64" at idx "
 979                        "%u\n", oa->o_id, req->rq_stripe, req->rq_oa->o_id,
 980                        req->rq_idx);
 981                 rc = obd_getattr_async(lov->tgts[req->rq_idx].ltd_exp,
 982                                        req->rq_oa, NULL, rqset);
 983                 if (rc) {
 984                         CERROR("error: getattr objid "LPX64" subobj "
 985                                LPX64" on OST idx %d: rc = %d\n",
 986                                lovset->set_oa->o_id, req->rq_oa->o_id,
 987                                req->rq_idx, rc);
 988                         GOTO(out, rc);
 989                 }
 990                 lov_update_common_set(lovset, req, rc);
 991         }
 992
 993         LASSERT(rc == 0);
 994         LASSERT (rqset->set_interpret == NULL);
 995         rqset->set_interpret = lov_getattr_interpret;
 996         rqset->set_arg = (void *)lovset;
 997         RETURN(rc);
 998 out:
 999         LASSERT(rc);
1000         lov_fini_getattr_set(lovset);
1001         RETURN(rc);
1002 }
1003
1004 static int lov_setattr(struct obd_export *exp, struct obdo *src_oa,
1005                        struct lov_stripe_md *lsm, struct obd_trans_info *oti,
1006                        struct lustre_capa *capa)
1007 {
1008         struct lov_request_set *set;
1009         struct lov_obd *lov;
1010         struct list_head *pos;
1011         struct lov_request *req;
1012         int err = 0, rc = 0;
1013         ENTRY;
1014
1015         if (lsm_bad_magic(lsm))
1016                 RETURN(-EINVAL);
1017
1018         if (!exp || !exp->exp_obd)
1019                 RETURN(-ENODEV);
1020
1021         LASSERT(!(src_oa->o_valid & ~(OBD_MD_FLID|OBD_MD_FLTYPE | OBD_MD_FLMODE|
1022                                       OBD_MD_FLATIME | OBD_MD_FLMTIME |
1023                                       OBD_MD_FLCTIME | OBD_MD_FLFLAGS |
1024                                       OBD_MD_FLSIZE | OBD_MD_FLGROUP |
1025                                       OBD_MD_FLUID | OBD_MD_FLGID |
1026                                       OBD_MD_FLINLINE | OBD_MD_FLIFID)));
1027
1028         LASSERT(!(src_oa->o_valid & OBD_MD_FLGROUP) || src_oa->o_gr > 0);
1029
1030         lov = &exp->exp_obd->u.lov;
1031         rc = lov_prep_setattr_set(exp, src_oa, lsm, NULL, &set);
1032         if (rc)
1033                 RETURN(rc);
1034
1035         list_for_each (pos, &set->set_list) {
1036                 req = list_entry(pos, struct lov_request, rq_link);
1037
1038                 rc = obd_setattr(lov->tgts[req->rq_idx].ltd_exp, req->rq_oa,
1039                                  NULL, NULL, NULL);
1040                 err = lov_update_common_set(set, req, rc);
1041                 if (err) {
1042                         CERROR("error: setattr objid "LPX64" subobj "
1043                                LPX64" on OST idx %d: rc = %d\n",
1044                                set->set_oa->o_id, req->rq_oa->o_id,
1045                                req->rq_idx, err);
1046                         if (!rc)
1047                                 rc = err;
1048                 }
1049         }
1050         err = lov_fini_setattr_set(set);
1051         if (!rc)
1052                 rc = err;
1053         RETURN(rc);
1054 }
1055
1056 static int lov_revalidate_policy(struct lov_obd *lov, struct lov_stripe_md *lsm)
1057 {
1058         static int next_idx = 0;
1059         struct lov_tgt_desc *tgt;
1060         int i, count;
1061         ENTRY;
1062
1063         /* XXX - we should do something clever and take lsm
1064          * into account but just do round robin for now. */
1065
1066         /* last_idx must always be less that count because
1067          * ld_tgt_count currently cannot shrink. */
1068         count = lov->desc.ld_tgt_count;
1069
1070         for (i = next_idx, tgt = lov->tgts + i; i < count; i++, tgt++) {
1071                 if (lov_tgt_active(lov, tgt, 0)) {
1072                         next_idx = (i + 1) % count;
1073                         RETURN(i);
1074                 }
1075         }
1076
1077         for (i = 0, tgt = lov->tgts; i < next_idx; i++, tgt++) {
1078                 if (lov_tgt_active(lov, tgt, 0)) {
1079                         next_idx = (i + 1) % count;
1080                         RETURN(i);
1081                 }
1082         }
1083
1084         RETURN(-EIO);
1085 }
1086
1087 static int lov_revalidate_md(struct obd_export *exp, struct obdo *src_oa,
1088                              struct lov_stripe_md *ea,
1089                              struct obd_trans_info *oti)
1090 {
1091         struct obd_export *osc_exp;
1092         struct lov_obd *lov = &exp->exp_obd->u.lov;
1093         struct lov_stripe_md *lsm = ea;
1094         struct lov_stripe_md obj_md;
1095         struct lov_stripe_md *obj_mdp = &obj_md;
1096         struct lov_oinfo *loi;
1097         struct obdo *tmp_oa;
1098         int ost_idx, updates = 0, i;
1099         ENTRY;
1100
1101         tmp_oa = obdo_alloc();
1102         if (tmp_oa == NULL)
1103                 RETURN(-ENOMEM);
1104
1105         loi = lsm->lsm_oinfo;
1106         for (i = 0; i < lsm->lsm_stripe_count; i++, loi++) {
1107                 struct lov_tgt_desc *tgt;
1108                 int rc;
1109
1110                 if (!lov_tgt_changed(lov, loi))
1111                         continue;
1112
1113                 /* returns an active tgt. tgt->ltd_refcount is incremented. */
1114                 ost_idx = lov_revalidate_policy(lov, lsm);
1115                 if (ost_idx < 0) {
1116                         /* FIXME: punt for now. */
1117                         CERROR("lov_revalidate_policy failed; no active "
1118                                "OSCs?\n");
1119                         continue;
1120                 }
1121
1122                 /* create a new object */
1123                 memcpy(tmp_oa, src_oa, sizeof(*tmp_oa));
1124                 /* XXX: LOV STACKING: use real "obj_mdp" sub-data */
1125                 tgt = lov->tgts + ost_idx;
1126                 osc_exp = tgt->ltd_exp;
1127                 rc = obd_create(osc_exp, tmp_oa, NULL, 0, &obj_mdp, oti);
1128                 if (rc) {
1129                         lov_tgt_decref(lov, tgt);
1130                         CERROR("error creating new subobj at idx %d; "
1131                                "rc = %d\n", ost_idx, rc);
1132                         continue;
1133                 }
1134
1135                 CDEBUG(D_INODE,
1136                        "replacing idx %d gen %d objid "LPX64" subobj "LPX64" "
1137                        "with idx %d gen %d objid "LPX64" subobj "LPX64".\n",
1138                        loi->loi_ost_idx, loi->loi_ost_gen,
1139                        loi->loi_id, loi->loi_gr,
1140                        ost_idx, tgt->ltd_gen, tmp_oa->o_id, tmp_oa->o_gr);
1141
1142                 if (oti->oti_objid)
1143                         oti->oti_objid[ost_idx] = tmp_oa->o_id;
1144                 loi->loi_id = tmp_oa->o_id;
1145                 loi->loi_gr = tmp_oa->o_gr;
1146                 loi->loi_ost_idx = ost_idx;
1147                 loi->loi_ost_gen = tgt->ltd_gen;
1148                 lov_tgt_decref(lov, tgt);
1149                 updates = 1;
1150         }
1151
1152         /* If we got an error revalidating an entry there's no need to
1153          * cleanup up objects we allocated here because the bad entry
1154          * still points to a deleted OST. */
1155
1156         obdo_free(tmp_oa);
1157         RETURN(updates);
1158 }
1159
1160 /* FIXME: maybe we'll just make one node the authoritative attribute node, then
1161  * we can send this 'punch' to just the authoritative node and the nodes
1162  * that the punch will affect. */
1163 static int lov_punch(struct obd_export *exp, struct obdo *oa,
1164                      struct lov_stripe_md *lsm, obd_off start,
1165                      obd_off end, struct obd_trans_info *oti,
1166                      struct lustre_capa *capa)
1167 {
1168         struct lov_request_set *set;
1169         struct lov_obd *lov;
1170         struct list_head *pos;
1171         struct lov_request *req;
1172         int err = 0, rc = 0;
1173         ENTRY;
1174
1175         if (lsm_bad_magic(lsm))
1176                 RETURN(-EINVAL);
1177
1178         if (!exp || !exp->exp_obd)
1179                 RETURN(-ENODEV);
1180
1181         lov = &exp->exp_obd->u.lov;
1182         rc = lov_prep_punch_set(exp, oa, lsm, start, end, oti, &set);
1183         if (rc)
1184                 RETURN(rc);
1185
1186         list_for_each (pos, &set->set_list) {
1187                 req = list_entry(pos, struct lov_request, rq_link);
1188
1189                 rc = obd_punch(lov->tgts[req->rq_idx].ltd_exp, req->rq_oa,
1190                                NULL, req->rq_extent.start,
1191                                req->rq_extent.end, NULL, capa);
1192                 err = lov_update_punch_set(set, req, rc);
1193                 if (err) {
1194                         CERROR("error: punch objid "LPX64" subobj "LPX64
1195                                " on OST idx %d: rc = %d\n", set->set_oa->o_id,
1196                                req->rq_oa->o_id, req->rq_idx, rc);
1197                         if (!rc)
1198                                 rc = err;
1199                 }
1200         }
1201         err = lov_fini_punch_set(set);
1202         if (!rc)
1203                 rc = err;
1204         RETURN(rc);
1205 }
1206
1207 static int lov_sync(struct obd_export *exp, struct obdo *oa,
1208                     struct lov_stripe_md *lsm, obd_off start, obd_off end)
1209 {
1210         struct lov_request_set *set;
1211         struct lov_obd *lov;
1212         struct list_head *pos;
1213         struct lov_request *req;
1214         int err = 0, rc = 0;
1215         ENTRY;
1216
1217         if (lsm_bad_magic(lsm))
1218                 RETURN(-EINVAL);
1219
1220         if (!exp->exp_obd)
1221                 RETURN(-ENODEV);
1222
1223         lov = &exp->exp_obd->u.lov;
1224         rc = lov_prep_sync_set(exp, oa, lsm, start, end, &set);
1225         if (rc)
1226                 RETURN(rc);
1227
1228         list_for_each (pos, &set->set_list) {
1229                 req = list_entry(pos, struct lov_request, rq_link);
1230
1231                 rc = obd_sync(lov->tgts[req->rq_idx].ltd_exp, req->rq_oa,
1232                               NULL, req->rq_extent.start, req->rq_extent.end);
1233                 err = lov_update_common_set(set, req, rc);
1234                 if (err) {
1235                         CERROR("error: fsync objid "LPX64" subobj "LPX64
1236                                " on OST idx %d: rc = %d\n", set->set_oa->o_id,
1237                                req->rq_oa->o_id, req->rq_idx, rc);
1238                         if (!rc)
1239                                 rc = err;
1240                 }
1241         }
1242         err = lov_fini_sync_set(set);
1243         if (!rc)
1244                 rc = err;
1245         RETURN(rc);
1246 }
1247
1248 static int lov_brw_check(struct lov_obd *lov, struct obdo *oa,
1249                          struct lov_stripe_md *lsm,
1250                          obd_count oa_bufs, struct brw_page *pga)
1251 {
1252         int i, rc = 0;
1253         ENTRY;
1254
1255         /* The caller just wants to know if there's a chance that this
1256          * I/O can succeed */
1257         for (i = 0; i < oa_bufs; i++) {
1258                 struct lov_oinfo *loi;
1259                 struct lov_tgt_desc *tgt;
1260                 int stripe;
1261                 obd_off start, end;
1262
1263                 if (!lov_stripe_intersects(lsm, i, pga[i].disk_offset,
1264                                            pga[i].disk_offset + pga[i].count,
1265                                            &start, &end))
1266                         continue;
1267
1268                 stripe = lov_stripe_number(lsm, pga[i].disk_offset);
1269                 loi = lsm->lsm_oinfo + stripe;
1270                 tgt = lov->tgts + loi->loi_ost_idx;
1271
1272                 if (!lov_tgt_ready(lov, tgt, loi->loi_ost_gen)) {
1273                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
1274                         RETURN(-EIO);
1275                 }
1276                 rc = obd_brw(OBD_BRW_CHECK, tgt->ltd_exp, oa,
1277                              NULL, 1, &pga[i], NULL);
1278                 lov_tgt_decref(lov, tgt);
1279                 if (rc)
1280                         break;
1281         }
1282         RETURN(rc);
1283 }
1284
1285 static int lov_brw(int cmd, struct obd_export *exp, struct obdo *src_oa,
1286                    struct lov_stripe_md *lsm, obd_count oa_bufs,
1287                    struct brw_page *pga, struct obd_trans_info *oti)
1288 {
1289         struct lov_request_set *set;
1290         struct lov_request *req;
1291         struct list_head *pos;
1292         struct lov_obd *lov = &exp->exp_obd->u.lov;
1293         int err, rc = 0;
1294         ENTRY;
1295
1296         if (lsm_bad_magic(lsm))
1297                 RETURN(-EINVAL);
1298
1299         if (cmd == OBD_BRW_CHECK) {
1300                 rc = lov_brw_check(lov, src_oa, lsm, oa_bufs, pga);
1301                 RETURN(rc);
1302         }
1303
1304         rc = lov_prep_brw_set(exp, src_oa, lsm, oa_bufs, pga, oti, &set);
1305         if (rc)
1306                 RETURN(rc);
1307
1308         list_for_each (pos, &set->set_list) {
1309                 struct obd_export *sub_exp;
1310                 struct brw_page *sub_pga;
1311                 req = list_entry(pos, struct lov_request, rq_link);
1312
1313                 sub_exp = lov->tgts[req->rq_idx].ltd_exp;
1314                 sub_pga = set->set_pga + req->rq_pgaidx;
1315                 rc = obd_brw(cmd, sub_exp, req->rq_oa, req->rq_md,
1316                              req->rq_oabufs, sub_pga, oti);
1317                 if (rc)
1318                         break;
1319                 lov_update_common_set(set, req, rc);
1320         }
1321
1322         err = lov_fini_brw_set(set);
1323         if (!rc)
1324                 rc = err;
1325         RETURN(rc);
1326 }
1327
1328 static int lov_brw_interpret(struct ptlrpc_request_set *reqset, void *data,
1329                              int rc)
1330 {
1331         struct lov_request_set *lovset = (struct lov_request_set *)data;
1332         ENTRY;
1333
1334         if (rc) {
1335                 lovset->set_completes = 0;
1336                 lov_fini_brw_set(lovset);
1337         } else {
1338                 rc = lov_fini_brw_set(lovset);
1339         }
1340
1341         RETURN(rc);
1342 }
1343
1344 static int lov_brw_async(int cmd, struct obd_export *exp, struct obdo *oa,
1345                          struct lov_stripe_md *lsm, obd_count oa_bufs,
1346                          struct brw_page *pga, struct ptlrpc_request_set *set,
1347                          struct obd_trans_info *oti)
1348 {
1349         struct lov_request_set *lovset;
1350         struct lov_request *req;
1351         struct list_head *pos;
1352         struct lov_obd *lov = &exp->exp_obd->u.lov;
1353         int rc = 0;
1354         ENTRY;
1355
1356         if (lsm_bad_magic(lsm))
1357                 RETURN(-EINVAL);
1358
1359         if (cmd == OBD_BRW_CHECK) {
1360                 rc = lov_brw_check(lov, oa, lsm, oa_bufs, pga);
1361                 RETURN(rc);
1362         }
1363
1364         rc = lov_prep_brw_set(exp, oa, lsm, oa_bufs, pga, oti, &lovset);
1365         if (rc)
1366                 RETURN(rc);
1367
1368         list_for_each (pos, &lovset->set_list) {
1369                 struct obd_export *sub_exp;
1370                 struct brw_page *sub_pga;
1371                 req = list_entry(pos, struct lov_request, rq_link);
1372
1373                 sub_exp = lov->tgts[req->rq_idx].ltd_exp;
1374                 sub_pga = lovset->set_pga + req->rq_pgaidx;
1375                 rc = obd_brw_async(cmd, sub_exp, req->rq_oa, req->rq_md,
1376                                    req->rq_oabufs, sub_pga, set, oti);
1377                 if (rc)
1378                         GOTO(out, rc);
1379                 lov_update_common_set(lovset, req, rc);
1380         }
1381         LASSERT(rc == 0);
1382         LASSERT(set->set_interpret == NULL);
1383         set->set_interpret = (set_interpreter_func)lov_brw_interpret;
1384         set->set_arg = (void *)lovset;
1385
1386         RETURN(rc);
1387 out:
1388         lov_fini_brw_set(lovset);
1389         RETURN(rc);
1390 }
1391
1392 static int lov_ap_make_ready(void *data, int cmd)
1393 {
1394         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1395
1396         return lap->lap_caller_ops->ap_make_ready(lap->lap_caller_data, cmd);
1397 }
1398 static int lov_ap_refresh_count(void *data, int cmd)
1399 {
1400         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1401
1402         return lap->lap_caller_ops->ap_refresh_count(lap->lap_caller_data,
1403                                                      cmd);
1404 }
1405 static void lov_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
1406 {
1407         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1408
1409         lap->lap_caller_ops->ap_fill_obdo(lap->lap_caller_data, cmd, oa);
1410         /* XXX woah, shouldn't we be altering more here?  size? */
1411         oa->o_id = lap->lap_loi_id;
1412 }
1413
1414 static void lov_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
1415 {
1416         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1417
1418         /* in a raid1 regime this would down a count of many ios
1419          * in flight, onl calling the caller_ops completion when all
1420          * the raid1 ios are complete */
1421         lap->lap_caller_ops->ap_completion(lap->lap_caller_data, cmd, oa, rc);
1422 }
1423
1424 static struct obd_async_page_ops lov_async_page_ops = {
1425         .ap_make_ready =        lov_ap_make_ready,
1426         .ap_refresh_count =     lov_ap_refresh_count,
1427         .ap_fill_obdo =         lov_ap_fill_obdo,
1428         .ap_completion =        lov_ap_completion,
1429 };
1430
1431 static int lov_prep_async_page(struct obd_export *exp,
1432                                struct lov_stripe_md *lsm,
1433                                struct lov_oinfo *loi, struct page *page,
1434                                obd_off offset, struct obd_async_page_ops *ops,
1435                                void *data, void **res)
1436 {
1437         struct lov_obd *lov = &exp->exp_obd->u.lov;
1438         struct lov_tgt_desc *tgt;
1439         struct lov_async_page *lap;
1440         int rc, stripe;
1441         ENTRY;
1442
1443         if (lsm_bad_magic(lsm))
1444                 RETURN(-EINVAL);
1445         LASSERT(loi == NULL);
1446
1447         stripe = lov_stripe_number(lsm, offset);
1448         loi = &lsm->lsm_oinfo[stripe];
1449
1450         OBD_ALLOC(lap, sizeof(*lap));
1451         if (lap == NULL)
1452                 RETURN(-ENOMEM);
1453
1454         lap->lap_magic = LAP_MAGIC;
1455         lap->lap_caller_ops = ops;
1456         lap->lap_caller_data = data;
1457
1458         /* FIXME handle multiple oscs after landing b_raid1 */
1459         lap->lap_stripe = stripe;
1460         switch (lsm->lsm_pattern) {
1461                 case LOV_PATTERN_RAID0:
1462                         lov_stripe_offset(lsm, offset, lap->lap_stripe,
1463                                           &lap->lap_sub_offset);
1464                         break;
1465                 case LOV_PATTERN_CMOBD:
1466                         lap->lap_sub_offset = offset;
1467                         break;
1468                 default:
1469                         LBUG();
1470         }
1471
1472         /* so the callback doesn't need the lsm */
1473         lap->lap_loi_id = loi->loi_id;
1474
1475         tgt = lov->tgts + loi->loi_ost_idx;
1476
1477         if (lov_tgt_ready(lov, tgt, loi->loi_ost_gen)) {
1478                 rc = obd_prep_async_page(tgt->ltd_exp, lsm, loi, page,
1479                                          lap->lap_sub_offset,
1480                                          &lov_async_page_ops, lap,
1481                                          &lap->lap_sub_cookie);
1482                 lov_tgt_decref(lov, tgt);
1483         } else {
1484                 rc = -EIO;
1485         }
1486
1487         if (rc) {
1488                 OBD_FREE(lap, sizeof(*lap));
1489                 RETURN(rc);
1490         }
1491         CDEBUG(D_CACHE, "lap %p page %p cookie %p off "LPU64"\n", lap, page,
1492                lap->lap_sub_cookie, offset);
1493         *res = lap;
1494         RETURN(0);
1495 }
1496
1497 static int lov_queue_async_io(struct obd_export *exp,
1498                               struct lov_stripe_md *lsm,
1499                               struct lov_oinfo *loi, void *cookie,
1500                               int cmd, obd_off off, int count,
1501                               obd_flags brw_flags, obd_flags async_flags)
1502 {
1503         struct lov_obd *lov = &exp->exp_obd->u.lov;
1504         struct lov_tgt_desc *tgt;
1505         struct lov_async_page *lap;
1506         int rc;
1507
1508         LASSERT(loi == NULL);
1509
1510         if (lsm_bad_magic(lsm))
1511                 RETURN(-EINVAL);
1512
1513         lap = LAP_FROM_COOKIE(cookie);
1514         loi = &lsm->lsm_oinfo[lap->lap_stripe];
1515         tgt = lov->tgts + loi->loi_ost_idx;
1516
1517         if (!lov_tgt_valid(lov, tgt, loi->loi_ost_gen))
1518                 RETURN(-EIO);
1519
1520         rc = obd_queue_async_io(tgt->ltd_exp, lsm, loi, lap->lap_sub_cookie,
1521                                 cmd, off, count, brw_flags, async_flags);
1522
1523         lov_tgt_decref(lov, tgt);
1524         RETURN(rc);
1525 }
1526
1527 static int lov_set_async_flags(struct obd_export *exp,
1528                                struct lov_stripe_md *lsm,
1529                                struct lov_oinfo *loi, void *cookie,
1530                                obd_flags async_flags)
1531 {
1532         struct lov_obd *lov = &exp->exp_obd->u.lov;
1533         struct lov_tgt_desc *tgt;
1534         struct lov_async_page *lap;
1535         int rc;
1536
1537         LASSERT(loi == NULL);
1538
1539         if (lsm_bad_magic(lsm))
1540                 RETURN(-EINVAL);
1541
1542         lap = LAP_FROM_COOKIE(cookie);
1543         loi = &lsm->lsm_oinfo[lap->lap_stripe];
1544         tgt = lov->tgts + loi->loi_ost_idx;
1545
1546         if (!lov_tgt_ready(lov, tgt, loi->loi_ost_gen))
1547                  RETURN(-EIO);
1548
1549         rc = obd_set_async_flags(tgt->ltd_exp, lsm, loi, lap->lap_sub_cookie,
1550                                  async_flags);
1551
1552         lov_tgt_decref(lov, tgt);
1553         RETURN(rc);
1554 }
1555
1556 static int lov_queue_group_io(struct obd_export *exp,
1557                               struct lov_stripe_md *lsm,
1558                               struct lov_oinfo *loi,
1559                               struct obd_io_group *oig, void *cookie,
1560                               int cmd, obd_off off, int count,
1561                               obd_flags brw_flags, obd_flags async_flags)
1562 {
1563         struct lov_obd *lov = &exp->exp_obd->u.lov;
1564         struct lov_tgt_desc *tgt;
1565         struct lov_async_page *lap;
1566         int rc;
1567
1568         LASSERT(loi == NULL);
1569
1570         if (lsm_bad_magic(lsm))
1571                 RETURN(-EINVAL);
1572
1573         lap = LAP_FROM_COOKIE(cookie);
1574         loi = &lsm->lsm_oinfo[lap->lap_stripe];
1575         tgt = lov->tgts + loi->loi_ost_idx;
1576
1577         if (!lov_tgt_valid(lov, tgt, loi->loi_ost_gen))
1578                 RETURN(-EIO);
1579
1580         rc = obd_queue_group_io(tgt->ltd_exp, lsm, loi, oig,
1581                                 lap->lap_sub_cookie, cmd, off, count,
1582                                 brw_flags, async_flags);
1583
1584         lov_tgt_decref(lov, tgt);
1585         RETURN(rc);
1586 }
1587
1588 /* this isn't exactly optimal.  we may have queued sync io in oscs on
1589  * all stripes, but we don't record that fact at queue time.  so we
1590  * trigger sync io on all stripes. */
1591 static int lov_trigger_group_io(struct obd_export *exp,
1592                                 struct lov_stripe_md *lsm,
1593                                 struct lov_oinfo *loi,
1594                                 struct obd_io_group *oig)
1595 {
1596         struct lov_obd *lov = &exp->exp_obd->u.lov;
1597         int rc = 0, i, err;
1598
1599         LASSERT(loi == NULL);
1600
1601         if (lsm_bad_magic(lsm))
1602                 RETURN(-EINVAL);
1603
1604         loi = lsm->lsm_oinfo;
1605         for (i = 0; i < lsm->lsm_stripe_count; i++, loi++) {
1606                 struct lov_tgt_desc *tgt = lov->tgts + loi->loi_ost_idx;
1607
1608                 if (!lov_tgt_ready(lov, tgt, loi->loi_ost_gen)) {
1609                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
1610                         continue;
1611                 }
1612
1613                 err = obd_trigger_group_io(tgt->ltd_exp, lsm, loi, oig);
1614
1615                 lov_tgt_decref(lov, tgt);
1616
1617                 if (rc == 0 && err != 0)
1618                         rc = err;
1619         };
1620         RETURN(rc);
1621 }
1622
1623 static int lov_teardown_async_page(struct obd_export *exp,
1624                                    struct lov_stripe_md *lsm,
1625                                    struct lov_oinfo *loi, void *cookie)
1626 {
1627         struct lov_obd *lov = &exp->exp_obd->u.lov;
1628         struct lov_tgt_desc *tgt;
1629         struct lov_async_page *lap;
1630         int rc;
1631         ENTRY;
1632
1633         LASSERT(loi == NULL);
1634
1635         if (lsm_bad_magic(lsm))
1636                 RETURN(-EINVAL);
1637
1638         lap = LAP_FROM_COOKIE(cookie);
1639
1640         loi = &lsm->lsm_oinfo[lap->lap_stripe];
1641         tgt = lov->tgts + loi->loi_ost_idx;
1642
1643         /* FIXME: this leaks the page, but it should never really happen.
1644          *        Should we make this an LBUG() ? */
1645         if (!lov_tgt_valid(lov, tgt, loi->loi_ost_gen)) {
1646                 CERROR("page found with invalid OSC !");
1647                 RETURN(-EIO);
1648         }
1649
1650         rc = obd_teardown_async_page(tgt->ltd_exp, lsm, loi,
1651                                      lap->lap_sub_cookie);
1652         lov_tgt_decref(lov, tgt);
1653
1654         if (rc) {
1655                 CERROR("unable to teardown sub cookie %p: %d\n",
1656                        lap->lap_sub_cookie, rc);
1657                 RETURN(rc);
1658         }
1659         OBD_FREE(lap, sizeof(*lap));
1660         RETURN(rc);
1661 }
1662
1663 static int lov_enqueue(struct obd_export *exp, struct lov_stripe_md *lsm,
1664                        __u32 type, ldlm_policy_data_t *policy, __u32 mode,
1665                        int *flags, void *bl_cb, void *cp_cb, void *gl_cb,
1666                        void *data,__u32 lvb_len, void *lvb_swabber,
1667                        struct lustre_handle *lockh)
1668 {
1669         struct lov_request_set *set;
1670         struct lov_request *req;
1671         struct list_head *pos;
1672         struct lustre_handle *lov_lockhp;
1673         struct lov_obd *lov;
1674         ldlm_error_t rc;
1675         int save_flags = *flags;
1676         ENTRY;
1677
1678         if (lsm_bad_magic(lsm))
1679                 RETURN(-EINVAL);
1680
1681         /* we should never be asked to replay a lock this way. */
1682         LASSERT((*flags & LDLM_FL_REPLAY) == 0);
1683
1684         if (!exp || !exp->exp_obd)
1685                 RETURN(-ENODEV);
1686
1687         lov = &exp->exp_obd->u.lov;
1688         rc = lov_prep_enqueue_set(exp, lsm, policy, mode, lockh, &set);
1689         if (rc)
1690                 RETURN(rc);
1691
1692         list_for_each (pos, &set->set_list) {
1693                 ldlm_policy_data_t sub_policy;
1694                 req = list_entry(pos, struct lov_request, rq_link);
1695                 lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
1696                 LASSERT(lov_lockhp);
1697
1698                 *flags = save_flags;
1699                 sub_policy.l_extent = req->rq_extent;
1700
1701                 rc = obd_enqueue(lov->tgts[req->rq_idx].ltd_exp, req->rq_md,
1702                                  type, &sub_policy, mode, flags, bl_cb,
1703                                  cp_cb, gl_cb, data, lvb_len, lvb_swabber,
1704                                  lov_lockhp);
1705                 rc = lov_update_enqueue_set(set, req, rc, save_flags);
1706                 if (rc != ELDLM_OK)
1707                         break;
1708         }
1709
1710         lov_fini_enqueue_set(set, mode);
1711         RETURN(rc);
1712 }
1713
1714 static int lov_match(struct obd_export *exp, struct lov_stripe_md *lsm,
1715                      __u32 type, ldlm_policy_data_t *policy, __u32 mode,
1716                      int *flags, void *data, struct lustre_handle *lockh)
1717 {
1718         struct lov_request_set *set;
1719         struct lov_request *req;
1720         struct list_head *pos;
1721         struct lov_obd *lov = &exp->exp_obd->u.lov;
1722         struct lustre_handle *lov_lockhp;
1723         int lov_flags, rc = 0;
1724         ENTRY;
1725
1726         if (lsm_bad_magic(lsm))
1727                 RETURN(-EINVAL);
1728
1729         if (!exp || !exp->exp_obd)
1730                 RETURN(-ENODEV);
1731
1732         lov = &exp->exp_obd->u.lov;
1733         rc = lov_prep_match_set(exp, lsm, policy, mode, lockh, &set);
1734         if (rc)
1735                 RETURN(rc);
1736
1737         list_for_each (pos, &set->set_list) {
1738                 ldlm_policy_data_t sub_policy;
1739                 req = list_entry(pos, struct lov_request, rq_link);
1740                 lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
1741                 LASSERT(lov_lockhp);
1742
1743                 sub_policy.l_extent = req->rq_extent;
1744                 lov_flags = *flags;
1745
1746                 rc = obd_match(lov->tgts[req->rq_idx].ltd_exp, req->rq_md,
1747                                type, &sub_policy, mode, &lov_flags, data,
1748                                lov_lockhp);
1749                 rc = lov_update_match_set(set, req, rc);
1750                 if (rc != 1)
1751                         break;
1752         }
1753         lov_fini_match_set(set, mode, *flags);
1754         RETURN(rc);
1755 }
1756
1757 static int lov_change_cbdata(struct obd_export *exp,
1758                              struct lov_stripe_md *lsm, ldlm_iterator_t it,
1759                              void *data)
1760 {
1761         struct lov_obd *lov;
1762         struct lov_oinfo *loi;
1763         int rc = 0, i;
1764         ENTRY;
1765
1766         if (lsm_bad_magic(lsm))
1767                 RETURN(-EINVAL);
1768
1769         if (!exp || !exp->exp_obd)
1770                 RETURN(-ENODEV);
1771
1772         LASSERT(lsm->lsm_object_gr > 0);
1773
1774         lov = &exp->exp_obd->u.lov;
1775         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
1776                 struct lov_tgt_desc *tgt = lov->tgts + loi->loi_ost_idx;
1777                 struct lov_stripe_md submd;
1778
1779                 submd.lsm_object_id = loi->loi_id;
1780                 submd.lsm_object_gr = lsm->lsm_object_gr;
1781                 submd.lsm_stripe_count = 0;
1782
1783                 if (!lov_tgt_valid(lov, tgt, loi->loi_ost_gen)) {
1784                         CDEBUG(D_HA, "lov idx %d invalid.\n", loi->loi_ost_idx);
1785                         continue;
1786                 }
1787                 rc = obd_change_cbdata(tgt->ltd_exp, &submd, it, data);
1788                 lov_tgt_decref(lov, tgt);
1789         }
1790         RETURN(rc);
1791 }
1792
1793 static int lov_cancel(struct obd_export *exp, struct lov_stripe_md *lsm,
1794                       __u32 mode, struct lustre_handle *lockh)
1795 {
1796         struct lov_request_set *set;
1797         struct lov_request *req;
1798         struct list_head *pos;
1799         struct lov_obd *lov = &exp->exp_obd->u.lov;
1800         struct lustre_handle *lov_lockhp;
1801         int err = 0, rc = 0;
1802         ENTRY;
1803
1804         if (lsm_bad_magic(lsm))
1805                 RETURN(-EINVAL);
1806
1807         if (!exp || !exp->exp_obd)
1808                 RETURN(-ENODEV);
1809
1810         LASSERT(lsm->lsm_object_gr > 0);
1811
1812         LASSERT(lockh);
1813         lov = &exp->exp_obd->u.lov;
1814         rc = lov_prep_cancel_set(exp, lsm, mode, lockh, &set);
1815         if (rc)
1816                 RETURN(rc);
1817
1818         list_for_each (pos, &set->set_list) {
1819                 req = list_entry(pos, struct lov_request, rq_link);
1820                 lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
1821
1822                 rc = obd_cancel(lov->tgts[req->rq_idx].ltd_exp, req->rq_md,
1823                                 mode, lov_lockhp);
1824                 rc = lov_update_common_set(set, req, rc);
1825                 if (rc) {
1826                         CERROR("error: cancel objid "LPX64" subobj "
1827                                LPX64" on OST idx %d: rc = %d\n",
1828                                lsm->lsm_object_id,
1829                                req->rq_md->lsm_object_id, req->rq_idx, rc);
1830                         err = rc;
1831                 }
1832
1833         }
1834         lov_fini_cancel_set(set);
1835         RETURN(err);
1836 }
1837
1838 static int lov_cancel_unused(struct obd_export *exp,
1839                              struct lov_stripe_md *lsm,
1840                              int flags, void *opaque)
1841 {
1842         struct lov_obd *lov;
1843         struct lov_oinfo *loi;
1844         struct lov_tgt_desc *tgt;
1845         int rc = 0, i;
1846         ENTRY;
1847
1848         lov = &exp->exp_obd->u.lov;
1849         if (lsm == NULL) {
1850                 tgt = lov->tgts;
1851                 for (i = 0; i < lov->desc.ld_tgt_count; i++, tgt++) {
1852                         int err;
1853
1854                         if (lov_tgt_ready(lov, tgt, 0)) {
1855                                 err = obd_cancel_unused(tgt->ltd_exp, NULL,
1856                                                         flags, opaque);
1857                                 lov_tgt_decref(lov, tgt);
1858                         } else {
1859                                 if (obd_uuid_empty(&tgt->uuid))
1860                                         continue;
1861                                 CERROR("error: cancel unused "
1862                                        "OST idx %d: OST inactive.\n", i);
1863                                 err = -EIO;
1864                         }
1865                         if (!rc)
1866                                 rc = err;
1867                 }
1868                 RETURN(rc);
1869         }
1870
1871         if (lsm_bad_magic(lsm))
1872                 RETURN(-EINVAL);
1873
1874         if (!exp || !exp->exp_obd)
1875                 RETURN(-ENODEV);
1876
1877         LASSERT(lsm->lsm_object_gr > 0);
1878
1879         for (i = 0,loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count; i++,loi++) {
1880                 struct lov_stripe_md submd;
1881                 int err;
1882
1883                 submd.lsm_object_id = loi->loi_id;
1884                 submd.lsm_object_gr = lsm->lsm_object_gr;
1885                 submd.lsm_stripe_count = 0;
1886
1887                 tgt = lov->tgts + loi->loi_ost_idx;
1888                 if (lov_tgt_ready(lov, tgt, loi->loi_ost_gen)) {
1889                         err = obd_cancel_unused(tgt->ltd_exp, &submd, flags,
1890                                                 opaque);
1891                         lov_tgt_decref(lov, tgt);
1892                         if (err && lov_tgt_ready(lov, tgt, loi->loi_ost_gen)) {
1893                                 lov_tgt_decref(lov, tgt);
1894                                 CERROR("error: cancel unused objid "LPX64" "
1895                                        "subobj "LPX64" OST idx %d: rc = %d\n",
1896                                        lsm->lsm_object_id, loi->loi_id,
1897                                        loi->loi_ost_idx, err);
1898                         }
1899                 } else {
1900                         err = -EIO;
1901                         CERROR("error: cancel unused objid "LPX64" "
1902                                "subobj "LPX64" on OST idx %d: OST inactive.\n",
1903                                lsm->lsm_object_id,
1904                                loi->loi_id, loi->loi_ost_idx);
1905                 }
1906
1907                 if (!rc)
1908                         rc = err;
1909         }
1910         RETURN(rc);
1911 }
1912
1913 #define LOV_U64_MAX ((__u64)~0ULL)
1914 #define LOV_SUM_MAX(tot, add)                                           \
1915         do {                                                            \
1916                 if ((tot) + (add) < (tot))                              \
1917                         (tot) = LOV_U64_MAX;                            \
1918                 else                                                    \
1919                         (tot) += (add);                                 \
1920         } while(0)
1921
1922 static int lov_statfs(struct obd_device *obd, struct obd_statfs *osfs,
1923                       unsigned long max_age)
1924 {
1925         struct lov_obd *lov = &obd->u.lov;
1926         struct lov_tgt_desc *tgt = lov->tgts;
1927         struct obd_statfs lov_sfs;
1928         int set = 0;
1929         int rc = 0;
1930         int i;
1931         ENTRY;
1932
1933
1934         /* We only get block data from the OBD */
1935         for (i = 0; i < lov->desc.ld_tgt_count; i++, tgt++) {
1936                 int err;
1937
1938                 if (!lov_tgt_active(lov, tgt, 0)) {
1939                         CDEBUG(D_HA, "lov idx %d inactive\n", i);
1940                         continue;
1941                 }
1942
1943                 err = obd_statfs(class_exp2obd(tgt->ltd_exp), &lov_sfs,max_age);
1944                 lov_tgt_decref(lov, tgt);
1945                 if (err) {
1946                         if (lov_tgt_active(lov, tgt, 0)) {
1947                                 lov_tgt_decref(lov, tgt);
1948                                 if (!rc)
1949                                         rc = err;
1950                         }
1951                         continue;
1952                 }
1953
1954                 if (!set) {
1955                         memcpy(osfs, &lov_sfs, sizeof(lov_sfs));
1956                         set = 1;
1957                 } else {
1958                         osfs->os_bfree += lov_sfs.os_bfree;
1959                         osfs->os_bavail += lov_sfs.os_bavail;
1960                         osfs->os_blocks += lov_sfs.os_blocks;
1961                         /* XXX not sure about this one - depends on policy.
1962                          *   - could be minimum if we always stripe on all OBDs
1963                          *     (but that would be wrong for any other policy,
1964                          *     if one of the OBDs has no more objects left)
1965                          *   - could be sum if we stripe whole objects
1966                          *   - could be average, just to give a nice number
1967                          *
1968                          * To give a "reasonable" (if not wholly accurate)
1969                          * number, we divide the total number of free objects
1970                          * by expected stripe count (watch out for overflow).
1971                          */
1972                         LOV_SUM_MAX(osfs->os_files, lov_sfs.os_files);
1973                         LOV_SUM_MAX(osfs->os_ffree, lov_sfs.os_ffree);
1974                 }
1975         }
1976
1977         if (set) {
1978                 __u32 expected_stripes = lov->desc.ld_default_stripe_count ?
1979                                          lov->desc.ld_default_stripe_count :
1980                                          lov->desc.ld_active_tgt_count;
1981
1982                 if ((lov->desc.ld_default_stripe_count == 0) ||
1983                     (lov->desc.ld_default_stripe_count >
1984                      lov->desc.ld_active_tgt_count)) {
1985                         expected_stripes = lov->desc.ld_active_tgt_count;
1986                 } else {
1987                         expected_stripes = lov->desc.ld_default_stripe_count;
1988                 }
1989
1990                 if (osfs->os_files != LOV_U64_MAX)
1991                         do_div(osfs->os_files, expected_stripes);
1992                 if (osfs->os_ffree != LOV_U64_MAX)
1993                         do_div(osfs->os_ffree, expected_stripes);
1994         } else if (!rc)
1995                 rc = -EIO;
1996
1997         RETURN(rc);
1998 }
1999
2000 static int lov_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2001                          void *karg, void *uarg)
2002 {
2003         struct obd_device *obddev = class_exp2obd(exp);
2004         struct lov_obd *lov = &obddev->u.lov;
2005         struct lov_tgt_desc *tgt;
2006         int i, rc, count = lov->desc.ld_tgt_count;
2007         struct obd_uuid *uuidp;
2008         ENTRY;
2009
2010         switch (cmd) {
2011         case OBD_IOC_LOV_GET_CONFIG: {
2012                 struct obd_ioctl_data *data = karg;
2013                 struct lov_desc *desc;
2014                 char *buf = NULL;
2015                 __u32 *genp;
2016
2017                 buf = NULL;
2018                 len = 0;
2019                 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
2020                         RETURN(-EINVAL);
2021
2022                 data = (struct obd_ioctl_data *)buf;
2023
2024                 if (sizeof(*desc) > data->ioc_inllen1) {
2025                         obd_ioctl_freedata(buf, len);
2026                         RETURN(-EINVAL);
2027                 }
2028
2029                 if (sizeof(uuidp->uuid) * count > data->ioc_inllen2) {
2030                         obd_ioctl_freedata(buf, len);
2031                         RETURN(-EINVAL);
2032                 }
2033
2034                 if (sizeof(__u32) * count > data->ioc_inllen3) {
2035                         obd_ioctl_freedata(buf, len);
2036                         RETURN(-EINVAL);
2037                 }
2038
2039                 desc = (struct lov_desc *)data->ioc_inlbuf1;
2040                 memcpy(desc, &(lov->desc), sizeof(*desc));
2041
2042                 uuidp = (struct obd_uuid *)data->ioc_inlbuf2;
2043                 genp = (__u32 *)data->ioc_inlbuf3;
2044                 /* the uuid will be empty for deleted OSTs */
2045                 lov_tgts_lock(lov);
2046                 tgt = lov->tgts;
2047                 for (i = 0; i < count; i++, uuidp++, genp++, tgt++) {
2048                         obd_str2uuid(uuidp, (char *)tgt->uuid.uuid);
2049                         *genp = tgt->ltd_gen;
2050                 }
2051                 lov_tgts_unlock(lov);
2052
2053                 rc = copy_to_user((void *)uarg, buf, len);
2054                 if (rc)
2055                         rc = -EFAULT;
2056                 obd_ioctl_freedata(buf, len);
2057                 break;
2058         }
2059         case LL_IOC_LOV_SETSTRIPE:
2060                 rc = lov_setstripe(exp, karg, uarg);
2061                 break;
2062         case LL_IOC_LOV_GETSTRIPE:
2063                 rc = lov_getstripe(exp, karg, uarg);
2064                 break;
2065         case LL_IOC_LOV_SETEA:
2066                 rc = lov_setea(exp, karg, uarg);
2067                 break;
2068         default: {
2069                 int set = 0;
2070                 if (count == 0)
2071                         RETURN(-ENOTTY);
2072                 rc = 0;
2073                 for (i = 0, tgt = lov->tgts; i < count; i++, tgt++) {
2074                         int err;
2075
2076                         if (!lov_tgt_active(lov, tgt, 0)) {
2077                                 CERROR("error: iocontrol failed for OSC %s on "
2078                                        "OST idx %d: OST inactive.\n",
2079                                        tgt->uuid.uuid, i);
2080                                 if (!rc)
2081                                         rc = -EIO;
2082                                 continue;
2083                         }
2084
2085                         err = obd_iocontrol(cmd, tgt->ltd_exp, len, karg, uarg);
2086                         lov_tgt_decref(lov, tgt);
2087                         if (err) {
2088                                 if (lov_tgt_ready(lov, tgt, 0)) {
2089                                         lov_tgt_decref(lov, tgt);
2090                                         CERROR("error: iocontrol OSC %s on OST "
2091                                                "idx %d cmd %x: err = %d\n",
2092                                                lov->tgts[i].uuid.uuid, i,
2093                                                cmd, err);
2094                                         if (!rc)
2095                                                 rc = err;
2096                                 }
2097                         } else
2098                                 set = 1;
2099                 }
2100                 if (!set && !rc)
2101                         rc = -EIO;
2102         }
2103         }
2104
2105         RETURN(rc);
2106 }
2107
2108 static int lov_get_info(struct obd_export *exp, __u32 keylen,
2109                         void *key, __u32 *vallen, void *val)
2110 {
2111         struct obd_device *obddev = class_exp2obd(exp);
2112         struct lov_obd *lov = &obddev->u.lov;
2113         struct lov_tgt_desc *tgt;
2114         int i;
2115         ENTRY;
2116
2117         if (!vallen || !val)
2118                 RETURN(-EFAULT);
2119
2120         if (keylen > strlen("lock_to_stripe") &&
2121             strcmp(key, "lock_to_stripe") == 0) {
2122                 struct {
2123                         char name[16];
2124                         struct ldlm_lock *lock;
2125                         struct lov_stripe_md *lsm;
2126                 } *data = key;
2127                 struct lov_oinfo *loi;
2128                 struct ldlm_res_id *res_id = &data->lock->l_resource->lr_name;
2129                 __u32 *stripe = val;
2130
2131                 if (*vallen < sizeof(*stripe))
2132                         RETURN(-EFAULT);
2133                 *vallen = sizeof(*stripe);
2134
2135                 /* XXX This is another one of those bits that will need to
2136                  * change if we ever actually support nested LOVs.  It uses
2137                  * the lock's export to find out which stripe it is. */
2138                 /* XXX - it's assumed all the locks for deleted OSTs have
2139                  * been cancelled. Also, the export for deleted OSTs will
2140                  * be NULL and won't match the lock's export. */
2141                 lov_tgts_lock(lov);
2142                 for (i = 0, loi = data->lsm->lsm_oinfo;
2143                      i < data->lsm->lsm_stripe_count;
2144                      i++, loi++) {
2145                         if (lov->tgts[loi->loi_ost_idx].ltd_exp ==
2146                                         data->lock->l_conn_export &&
2147                             loi->loi_id == res_id->name[0] &&
2148                             loi->loi_gr == res_id->name[2]) {
2149                                 lov_tgts_unlock(lov);
2150                                 *stripe = i;
2151                                 RETURN(0);
2152                         }
2153                 }
2154                 lov_tgts_unlock(lov);
2155
2156                 /* This can happen if a deleted OST has been replaced
2157                  * in the lsm by the MDS.  */
2158 #if 0
2159                 LDLM_ERROR(data->lock, "lock on inode without such object");
2160                 dump_lsm(D_ERROR, data->lsm);
2161                 portals_debug_dumpstack(NULL);
2162 #endif
2163                 RETURN(-ENXIO);
2164         } else if (keylen >= strlen("size_to_stripe") &&
2165                    strcmp(key, "size_to_stripe") == 0) {
2166                 struct {
2167                         int stripe_number;
2168                         __u64 size;
2169                         struct lov_stripe_md *lsm;
2170                 } *data = val;
2171
2172                 if (*vallen < sizeof(*data))
2173                         RETURN(-EFAULT);
2174
2175                 data->size = lov_size_to_stripe(data->lsm, data->size,
2176                                                 data->stripe_number);
2177                 RETURN(0);
2178         } else if (keylen >= strlen("last_id") && strcmp(key, "last_id") == 0) {
2179                 __u32 size = sizeof(obd_id);
2180                 obd_id *ids = val;
2181                 int rc = 0;
2182
2183                 tgt = lov->tgts;
2184                 for (i = 0; i < lov->desc.ld_tgt_count; i++, tgt++) {
2185                         if (!lov_tgt_active(lov, tgt, 0))
2186                                 continue;
2187                         rc = obd_get_info(tgt->ltd_exp, keylen, key, &size,
2188                                           ids + i);
2189                         lov_tgt_decref(lov, tgt);
2190                         if (rc != 0)
2191                                 RETURN(rc);
2192                 }
2193                 RETURN(0);
2194         } else if (keylen >= strlen("lovdesc") && strcmp(key, "lovdesc") == 0) {
2195                 struct lov_desc *desc_ret = val;
2196                 *desc_ret = lov->desc;
2197
2198                 RETURN(0);
2199         }
2200
2201         RETURN(-EINVAL);
2202 }
2203
2204 static int lov_set_info(struct obd_export *exp, obd_count keylen,
2205                         void *key, obd_count vallen, void *val)
2206 {
2207         struct obd_device *obddev = class_exp2obd(exp);
2208         struct lov_obd *lov = &obddev->u.lov;
2209         struct lov_tgt_desc *tgt;
2210         int i, rc = 0, err;
2211         ENTRY;
2212
2213 #define KEY_IS(str) \
2214         (keylen == strlen(str) && memcmp(key, str, keylen) == 0)
2215
2216         if (KEY_IS("async")) {
2217                 struct lov_desc *desc = &lov->desc;
2218
2219                 if (vallen != sizeof(int))
2220                         RETURN(-EINVAL);
2221                 lov->async = *((int*) val);
2222
2223                 lov_tgts_lock(lov);
2224                 tgt = lov->tgts;
2225                 for (i = 0; i < desc->ld_tgt_count; i++, tgt++) {
2226                         struct obd_uuid *tgt_uuid = &tgt->uuid;
2227                         struct obd_device *tgt_obd;
2228
2229                         tgt_obd = class_find_client_obd(tgt_uuid,
2230                                                         OBD_OSC_DEVICENAME,
2231                                                         &obddev->obd_uuid);
2232                         if (!tgt_obd) {
2233                                 CERROR("Target %s not attached\n",
2234                                         tgt_uuid->uuid);
2235                                 if (!rc)
2236                                         rc = -EINVAL;
2237                                 continue;
2238                         }
2239
2240                         err = obd_set_info(tgt_obd->obd_self_export,
2241                                            keylen, key, vallen, val);
2242                         if (err) {
2243                                 CERROR("Failed to set async on target %s\n",
2244                                         tgt_obd->obd_name);
2245                                 if (!rc)
2246                                         rc = err;
2247                         }
2248                 }
2249                 lov_tgts_unlock(lov);
2250                 RETURN(rc);
2251         }
2252
2253         if (KEY_IS("mds_conn")) {
2254                 if (vallen != sizeof(__u32))
2255                         RETURN(-EINVAL);
2256         } else if (KEY_IS("unlinked") || KEY_IS("unrecovery")) {
2257                 if (vallen != 0)
2258                         RETURN(-EINVAL);
2259         } else if (KEY_IS("sec") || KEY_IS("sec_flags")) {
2260                 struct lov_tgt_desc *tgt;
2261                 struct obd_export *exp;
2262                 int rc = 0, err, i;
2263
2264                 lov_tgts_lock(lov);
2265                 tgt = lov->tgts;
2266                 for (i = 0; i < lov->desc.ld_tgt_count; i++, tgt++) {
2267                         exp = tgt->ltd_exp;
2268                         /* during setup time the connections to osc might
2269                          * haven't been established.
2270                          */
2271                         if (exp == NULL) {
2272                                 struct obd_device *tgt_obd;
2273
2274                                 tgt_obd = class_find_client_obd(&tgt->uuid,
2275                                                                 OBD_OSC_DEVICENAME,
2276                                                                 &obddev->obd_uuid);
2277                                 if (!tgt_obd) {
2278                                         CERROR("can't set security flavor, "
2279                                                "device %s not attached?\n",
2280                                                 tgt->uuid.uuid);
2281                                         rc = -EINVAL;
2282                                         continue;
2283                                 }
2284                                 exp = tgt_obd->obd_self_export;
2285                         }
2286
2287                         err = obd_set_info(exp, keylen, key, vallen, val);
2288                         if (!rc)
2289                                 rc = err;
2290                 }
2291                 lov_tgts_unlock(lov);
2292
2293                 RETURN(rc);
2294         } else if (KEY_IS("auditlog")) {
2295                 ptl_nid_t nid = 0;
2296                 struct audit_msg * msg = val;
2297                 int i = 0, len = sizeof(nid);
2298                 struct obd_export * ost_exp;
2299                 //try to find client nid
2300                 for (i = 0; i < lov->desc.ld_tgt_count; i++) {
2301                         if (!lov->tgts[i].ltd_exp)
2302                                 continue;
2303                         rc = obd_get_info(lov->tgts[i].ltd_exp, 10,
2304                                           "client_nid", &len, &nid);
2305                         if (!rc)
2306                                 break;
2307                 }
2308
2309                 i = (__u32)nid % lov->desc.ld_tgt_count;
2310
2311                 CDEBUG(D_INFO, "Select OST %i for nid %.8x\n",i, (__u32)nid);
2312
2313                 ost_exp = lov->tgts[i].ltd_exp;
2314                 if (!ost_exp) {
2315                         CERROR("can't send auditlog data,"
2316                                "device %s is not attached?\n",
2317                                lov->tgts[i].uuid.uuid);
2318                         RETURN(-EFAULT);
2319                 }
2320                 msg->nid = nid;
2321                 rc = obd_set_info(ost_exp, keylen, key, vallen, val);
2322
2323                 RETURN(rc);
2324         } else if (KEY_IS("audit_obj")) {
2325                 struct audit_lov_msg * msg = val;
2326                 int i = 0, index;
2327                 struct obd_export * ost_exp;
2328                 struct lov_oinfo *loi;
2329                 struct obdo *oa = NULL;
2330
2331                 oa = obdo_alloc();
2332                 if (oa == NULL)
2333                         RETURN(-ENOMEM);
2334
2335                 //set audit for all objects of file
2336                 CDEBUG(D_INFO, "Set audit 0x%x for objects\n", (__u32)msg->mask);
2337                 //use o_fid to store audit mask
2338                 oa->o_fid = msg->mask;
2339                 oa->o_uid = msg->uid;
2340                 oa->o_gid = msg->gid;
2341                 //iterate over all objects and set audit for each
2342                 for (i = 0, loi = msg->lsm->lsm_oinfo;
2343                      i < msg->lsm->lsm_stripe_count; i++, loi++) {
2344                         index = loi->loi_ost_idx;
2345                         ost_exp = lov->tgts[index].ltd_exp;
2346                         if (!ost_exp) {
2347                                 CERROR("can't send audit_obj data,"
2348                                        "device %s is not attached?\n",
2349                                        lov->tgts[i].uuid.uuid);
2350                                 continue;
2351                         }
2352                         oa->o_id = loi->loi_id;
2353                         oa->o_gr = loi->loi_gr;
2354
2355                         obd_set_info(ost_exp, keylen, key, sizeof(*oa), oa);
2356                 }
2357
2358                 obdo_free(oa);
2359                 RETURN(0);
2360         } else if (KEY_IS("audit")) {
2361                 int i = 0;
2362                 struct obd_export * ost_exp;
2363
2364                 //set audit for whole fs on OSS
2365                 for (i = 0; i < lov->desc.ld_tgt_count; i++) {
2366                         ost_exp = lov->tgts[i].ltd_exp;
2367                         if (!ost_exp) {
2368                                 CERROR("can't send audit_fs data,"
2369                                        "device %s is not attached?\n",
2370                                        lov->tgts[i].uuid.uuid);
2371                                 continue;
2372                         }
2373                         obd_set_info(ost_exp, keylen, key, vallen, val);
2374                 }
2375
2376                 RETURN(rc);
2377         } else if (KEY_IS("flush_cred") || KEY_IS("crypto_cb") ||
2378                    KEY_IS("capa_key")) {
2379                 struct lov_tgt_desc *tgt;
2380                 int rc = 0, i;
2381
2382                 tgt = lov->tgts;
2383                 for (i = 0; i < lov->desc.ld_tgt_count; i++, tgt++) {
2384                         if (!lov_tgt_active(lov, tgt, 0))
2385                                 continue;
2386                         rc = obd_set_info(tgt->ltd_exp,
2387                                           keylen, key, vallen, val);
2388                         lov_tgt_decref(lov, tgt);
2389                         if (rc)
2390                                 RETURN(rc);
2391                 }
2392
2393                 RETURN(0);
2394         } else if (KEY_IS("setext")) {
2395                 struct lov_tgt_desc *tgt;
2396                 int rc = 0, i;
2397
2398                 for (i = 0, tgt = lov->tgts; i < lov->desc.ld_tgt_count;
2399                      i++, tgt++) {
2400                         if (!tgt->ltd_exp)
2401                                 continue;
2402                         rc = obd_set_info(tgt->ltd_exp,
2403                                           keylen, key, vallen, val);
2404                         if (rc)
2405                                 RETURN(rc);
2406                 }
2407
2408                 RETURN(0);
2409         } else {
2410                 RETURN(-EINVAL);
2411         }
2412
2413         tgt = lov->tgts;
2414         for (i = 0; i < lov->desc.ld_tgt_count; i++, tgt++) {
2415                 int active = 0;
2416
2417                 if (val) {
2418                         if (!obd_uuid_equals(val, &tgt->uuid))
2419                                 continue;
2420                 } else if (!lov_tgt_active(lov, tgt, 0)) {
2421                         continue;
2422                 } else {
2423                         active = 1;
2424                 }
2425
2426                 err = obd_set_info(tgt->ltd_exp, keylen, key, vallen, val);
2427                 if (!rc)
2428                         rc = err;
2429                 if (active)
2430                         lov_tgt_decref(lov, tgt);
2431         }
2432         RETURN(rc);
2433 #undef KEY_IS
2434 }
2435
2436 #if 0
2437 struct lov_multi_wait {
2438         struct ldlm_lock *lock;
2439         wait_queue_t      wait;
2440         int               completed;
2441         int               generation;
2442 };
2443
2444 int lov_complete_many(struct obd_export *exp, struct lov_stripe_md *lsm,
2445                       struct lustre_handle *lockh)
2446 {
2447         struct lov_lock_handles *lov_lockh = NULL;
2448         struct lustre_handle *lov_lockhp;
2449         struct lov_obd *lov;
2450         struct lov_oinfo *loi;
2451         struct lov_multi_wait *queues;
2452         int rc = 0, i;
2453         ENTRY;
2454
2455         if (lsm_bad_magic(lsm))
2456                 RETURN(-EINVAL);
2457
2458         if (!exp || !exp->exp_obd)
2459                 RETURN(-ENODEV);
2460
2461         LASSERT(lockh != NULL);
2462         if (lsm->lsm_stripe_count > 1) {
2463                 lov_lockh = lov_handle2llh(lockh);
2464                 if (lov_lockh == NULL) {
2465                         CERROR("LOV: invalid lov lock handle %p\n", lockh);
2466                         RETURN(-EINVAL);
2467                 }
2468
2469                 lov_lockhp = lov_lockh->llh_handles;
2470         } else {
2471                 lov_lockhp = lockh;
2472         }
2473
2474         OBD_ALLOC(queues, lsm->lsm_stripe_count * sizeof(*queues));
2475         if (queues == NULL)
2476                 GOTO(out, rc = -ENOMEM);
2477
2478         lov = &exp->exp_obd->u.lov;
2479         for (i = 0, loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count;
2480              i++, loi++, lov_lockhp++) {
2481                 struct ldlm_lock *lock;
2482                 struct obd_device *obd;
2483                 unsigned long irqflags;
2484
2485                 lock = ldlm_handle2lock(lov_lockhp);
2486                 if (lock == NULL) {
2487                         CDEBUG(D_HA, "lov idx %d subobj "LPX64" no lock?\n",
2488                                loi->loi_ost_idx, loi->loi_id);
2489                         queues[i].completed = 1;
2490                         continue;
2491                 }
2492
2493                 queues[i].lock = lock;
2494                 init_waitqueue_entry(&(queues[i].wait), current);
2495                 add_wait_queue(lock->l_waitq, &(queues[i].wait));
2496
2497                 obd = class_exp2obd(lock->l_conn_export);
2498                 if (obd != NULL)
2499                         imp = obd->u.cli.cl_import;
2500                 if (imp != NULL) {
2501                         spin_lock_irqsave(&imp->imp_lock, irqflags);
2502                         queues[i].generation = imp->imp_generation;
2503                         spin_unlock_irqrestore(&imp->imp_lock, irqflags);
2504                 }
2505         }
2506
2507         lwi = LWI_TIMEOUT_INTR(obd_timeout * HZ, ldlm_expired_completion_wait,
2508                                interrupted_completion_wait, &lwd);
2509         rc = l_wait_event_added(check_multi_complete(queues, lsm), &lwi);
2510
2511         for (i = 0; i < lsm->lsm_stripe_count; i++)
2512                 remove_wait_queue(lock->l_waitq, &(queues[i].wait));
2513
2514         if (rc == -EINTR || rc == -ETIMEDOUT) {
2515
2516
2517         }
2518
2519  out:
2520         if (lov_lockh != NULL)
2521                 lov_llh_put(lov_lockh);
2522         RETURN(rc);
2523 }
2524 #endif
2525
2526 struct obd_ops lov_obd_ops = {
2527         .o_owner               = THIS_MODULE,
2528         .o_attach              = lov_attach,
2529         .o_detach              = lov_detach,
2530         .o_setup               = lov_setup,
2531         .o_cleanup             = lov_cleanup,
2532         .o_process_config      = lov_process_config,
2533         .o_connect             = lov_connect,
2534         .o_disconnect          = lov_disconnect,
2535         .o_statfs              = lov_statfs,
2536         .o_packmd              = lov_packmd,
2537         .o_unpackmd            = lov_unpackmd,
2538         .o_revalidate_md       = lov_revalidate_md,
2539         .o_create              = lov_create,
2540         .o_destroy             = lov_destroy,
2541         .o_getattr             = lov_getattr,
2542         .o_getattr_async       = lov_getattr_async,
2543         .o_setattr             = lov_setattr,
2544         .o_brw                 = lov_brw,
2545         .o_brw_async           = lov_brw_async,
2546         .o_prep_async_page     = lov_prep_async_page,
2547         .o_queue_async_io      = lov_queue_async_io,
2548         .o_set_async_flags     = lov_set_async_flags,
2549         .o_queue_group_io      = lov_queue_group_io,
2550         .o_trigger_group_io    = lov_trigger_group_io,
2551         .o_teardown_async_page = lov_teardown_async_page,
2552         .o_adjust_kms          = lov_adjust_kms,
2553         .o_punch               = lov_punch,
2554         .o_sync                = lov_sync,
2555         .o_enqueue             = lov_enqueue,
2556         .o_match               = lov_match,
2557         .o_change_cbdata       = lov_change_cbdata,
2558         .o_cancel              = lov_cancel,
2559         .o_cancel_unused       = lov_cancel_unused,
2560         .o_iocontrol           = lov_iocontrol,
2561         .o_get_info            = lov_get_info,
2562         .o_set_info            = lov_set_info,
2563         .o_llog_init           = lov_llog_init,
2564         .o_llog_finish         = lov_llog_finish,
2565         .o_notify              = lov_notify,
2566 };
2567
2568 int __init lov_init(void)
2569 {
2570         struct lprocfs_static_vars lvars;
2571         int rc;
2572         ENTRY;
2573
2574         lprocfs_init_vars(lov, &lvars);
2575         rc = class_register_type(&lov_obd_ops, NULL, lvars.module_vars,
2576                                  OBD_LOV_DEVICENAME);
2577         RETURN(rc);
2578 }
2579
2580 #ifdef __KERNEL__
2581 static void /*__exit*/ lov_exit(void)
2582 {
2583         class_unregister_type(OBD_LOV_DEVICENAME);
2584 }
2585
2586 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2587 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver");
2588 MODULE_LICENSE("GPL");
2589
2590 module_init(lov_init);
2591 module_exit(lov_exit);
2592 #endif