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