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