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