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(struct lov_desc *desc)
 656 {
 657         if (desc->ld_default_stripe_size < PTLRPC_MAX_BRW_SIZE) {
 658                 LCONSOLE_WARN("Increasing default stripe size to min %u\n",
 659                               PTLRPC_MAX_BRW_SIZE);
 660                 desc->ld_default_stripe_size = PTLRPC_MAX_BRW_SIZE;
 661         } else if (desc->ld_default_stripe_size & (LOV_MIN_STRIPE_SIZE - 1)) {
 662                 desc->ld_default_stripe_size &= ~(LOV_MIN_STRIPE_SIZE - 1);
 663                 LCONSOLE_WARN("Changing default stripe size to "LPU64" (a "
 664                               "multiple of %u)\n",
 665                               desc->ld_default_stripe_size,LOV_MIN_STRIPE_SIZE);
 666         }
 667
 668         if (desc->ld_default_stripe_count == 0)
 669                 desc->ld_default_stripe_count = 1;
 670
 671         /* from lov_setstripe */
 672         if ((desc->ld_pattern != 0) &&
 673             (desc->ld_pattern != LOV_PATTERN_RAID0)) {
 674                 LCONSOLE_WARN("Unknown stripe pattern: %#x\n",desc->ld_pattern);
 675                 desc->ld_pattern = 0;
 676         }
 677
 678         if (desc->ld_qos_maxage == 0)
 679                 desc->ld_qos_maxage = QOS_DEFAULT_MAXAGE;
 680 }
 681
 682 static int lov_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
 683 {
 684         struct lprocfs_static_vars lvars = { 0 };
 685         struct lov_desc *desc;
 686         struct lov_obd *lov = &obd->u.lov;
 687         int count;
 688         ENTRY;
 689
 690         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
 691                 CERROR("LOV setup requires a descriptor\n");
 692                 RETURN(-EINVAL);
 693         }
 694
 695         desc = (struct lov_desc *)lustre_cfg_buf(lcfg, 1);
 696
 697         if (sizeof(*desc) > LUSTRE_CFG_BUFLEN(lcfg, 1)) {
 698                 CERROR("descriptor size wrong: %d > %d\n",
 699                        (int)sizeof(*desc), LUSTRE_CFG_BUFLEN(lcfg, 1));
 700                 RETURN(-EINVAL);
 701         }
 702
 703         if (desc->ld_magic != LOV_DESC_MAGIC) {
 704                 if (desc->ld_magic == __swab32(LOV_DESC_MAGIC)) {
 705                             CDEBUG(D_OTHER, "%s: Swabbing lov desc %p\n",
 706                                    obd->obd_name, desc);
 707                             lustre_swab_lov_desc(desc);
 708                 } else {
 709                         CERROR("%s: Bad lov desc magic: %#x\n",
 710                                obd->obd_name, desc->ld_magic);
 711                         RETURN(-EINVAL);
 712                 }
 713         }
 714
 715         lov_fix_desc(desc);
 716
 717         /* Because of 64-bit divide/mod operations only work with a 32-bit
 718          * divisor in a 32-bit kernel, we cannot support a stripe width
 719          * of 4GB or larger on 32-bit CPUs. */
 720         count = desc->ld_default_stripe_count;
 721         if ((count > 0 ? count : desc->ld_tgt_count) *
 722             desc->ld_default_stripe_size > 0xffffffff) {
 723                 CERROR("LOV: stripe width "LPU64"x%u > 4294967295 bytes\n",
 724                        desc->ld_default_stripe_size, count);
 725                 RETURN(-EINVAL);
 726         }
 727
 728         desc->ld_active_tgt_count = 0;
 729         lov->desc = *desc;
 730         lov->lov_tgt_size = 0;
 731         sema_init(&lov->lov_lock, 1);
 732         atomic_set(&lov->lov_refcount, 0);
 733         INIT_LIST_HEAD(&lov->lov_qos.lq_oss_list);
 734         init_rwsem(&lov->lov_qos.lq_rw_sem);
 735         lov->lov_qos.lq_dirty = 1;
 736         lov->lov_qos.lq_dirty_rr = 1;
 737         lov->lov_qos.lq_reset = 1;
 738         /* Default priority is toward free space balance */
 739         lov->lov_qos.lq_prio_free = 232;
 740
 741         lprocfs_lov_init_vars(&lvars);
 742         lprocfs_obd_setup(obd, lvars.obd_vars);
 743 #ifdef LPROCFS
 744         {
 745                 cfs_proc_dir_entry_t *entry;
 746
 747                 entry = create_proc_entry("target_obd", 0444,
 748                                           obd->obd_proc_entry);
 749                 if (entry != NULL) {
 750                         entry->proc_fops = &lov_proc_target_fops;
 751                         entry->data = obd;
 752                 }
 753         }
 754 #endif
 755
 756         RETURN(0);
 757 }
 758
 759 static int lov_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
 760 {
 761         int rc = 0;
 762         ENTRY;
 763
 764         switch (stage) {
 765         case OBD_CLEANUP_EARLY: {
 766                 struct lov_obd *lov = &obd->u.lov;
 767                 int i;
 768                 for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 769                         if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active)
 770                                 continue;
 771                         obd_precleanup(class_exp2obd(lov->lov_tgts[i]->ltd_exp),
 772                                        OBD_CLEANUP_EARLY);
 773                 }
 774                 break;
 775         }
 776         case OBD_CLEANUP_EXPORTS:
 777                 break;
 778         case OBD_CLEANUP_SELF_EXP:
 779                 rc = obd_llog_finish(obd, 0);
 780                 if (rc != 0)
 781                         CERROR("failed to cleanup llogging subsystems\n");
 782                 break;
 783         case OBD_CLEANUP_OBD:
 784                 break;
 785         }
 786         RETURN(rc);
 787 }
 788
 789 static int lov_cleanup(struct obd_device *obd)
 790 {
 791         struct lov_obd *lov = &obd->u.lov;
 792
 793         lprocfs_obd_cleanup(obd);
 794         if (lov->lov_tgts) {
 795                 int i;
 796                 for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 797                         if (!lov->lov_tgts[i])
 798                                 continue;
 799
 800                         /* Inactive targets may never have connected */
 801                         if (lov->lov_tgts[i]->ltd_active ||
 802                             atomic_read(&lov->lov_refcount))
 803                             /* We should never get here - these
 804                                should have been removed in the
 805                              disconnect. */
 806                                 CERROR("lov tgt %d not cleaned!"
 807                                        " deathrow=%d, lovrc=%d\n",
 808                                        i, lov->lov_death_row,
 809                                        atomic_read(&lov->lov_refcount));
 810                         lov_del_target(obd, i, 0, 0);
 811                 }
 812                 OBD_FREE(lov->lov_tgts, sizeof(*lov->lov_tgts) *
 813                          lov->lov_tgt_size);
 814                 lov->lov_tgt_size = 0;
 815         }
 816
 817         if (lov->lov_qos.lq_rr_size)
 818                 OBD_FREE(lov->lov_qos.lq_rr_array, lov->lov_qos.lq_rr_size);
 819
 820         RETURN(0);
 821 }
 822
 823 static int lov_process_config(struct obd_device *obd, obd_count len, void *buf)
 824 {
 825         struct lustre_cfg *lcfg = buf;
 826         struct obd_uuid obd_uuid;
 827         int cmd;
 828         int rc = 0;
 829         ENTRY;
 830
 831         switch(cmd = lcfg->lcfg_command) {
 832         case LCFG_LOV_ADD_OBD:
 833         case LCFG_LOV_ADD_INA:
 834         case LCFG_LOV_DEL_OBD: {
 835                 __u32 index;
 836                 int gen;
 837                 /* lov_modify_tgts add  0:lov_mdsA  1:ost1_UUID  2:0  3:1 */
 838                 if (LUSTRE_CFG_BUFLEN(lcfg, 1) > sizeof(obd_uuid.uuid))
 839                         GOTO(out, rc = -EINVAL);
 840
 841                 obd_str2uuid(&obd_uuid,  lustre_cfg_buf(lcfg, 1));
 842
 843                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%d", &index) != 1)
 844                         GOTO(out, rc = -EINVAL);
 845                 if (sscanf(lustre_cfg_buf(lcfg, 3), "%d", &gen) != 1)
 846                         GOTO(out, rc = -EINVAL);
 847                 if (cmd == LCFG_LOV_ADD_OBD)
 848                         rc = lov_add_target(obd, &obd_uuid, index, gen, 1);
 849                 else if (cmd == LCFG_LOV_ADD_INA)
 850                         rc = lov_add_target(obd, &obd_uuid, index, gen, 0);
 851                 else
 852                         rc = lov_del_target(obd, index, &obd_uuid, gen);
 853                 GOTO(out, rc);
 854         }
 855         case LCFG_PARAM: {
 856                 struct lprocfs_static_vars lvars = { 0 };
 857                 struct lov_desc *desc = &(obd->u.lov.desc);
 858
 859                 if (!desc)
 860                         GOTO(out, rc = -EINVAL);
 861
 862                 lprocfs_lov_init_vars(&lvars);
 863
 864                 rc = class_process_proc_param(PARAM_LOV, lvars.obd_vars,
 865                                               lcfg, obd);
 866                 GOTO(out, rc);
 867         }
 868         default: {
 869                 CERROR("Unknown command: %d\n", lcfg->lcfg_command);
 870                 GOTO(out, rc = -EINVAL);
 871
 872         }
 873         }
 874 out:
 875         RETURN(rc);
 876 }
 877
 878 #ifndef log2
 879 #define log2(n) ffz(~(n))
 880 #endif
 881
 882 static int lov_clear_orphans(struct obd_export *export, struct obdo *src_oa,
 883                              struct lov_stripe_md **ea,
 884                              struct obd_trans_info *oti)
 885 {
 886         struct lov_obd *lov;
 887         struct obdo *tmp_oa;
 888         struct obd_uuid *ost_uuid = NULL;
 889         int rc = 0, i;
 890         ENTRY;
 891
 892         LASSERT(src_oa->o_valid & OBD_MD_FLFLAGS &&
 893                 src_oa->o_flags == OBD_FL_DELORPHAN);
 894
 895         lov = &export->exp_obd->u.lov;
 896
 897         OBDO_ALLOC(tmp_oa);
 898         if (tmp_oa == NULL)
 899                 RETURN(-ENOMEM);
 900
 901         if (src_oa->o_valid & OBD_MD_FLINLINE) {
 902                 ost_uuid = (struct obd_uuid *)src_oa->o_inline;
 903                 CDEBUG(D_HA, "clearing orphans only for %s\n",
 904                        ost_uuid->uuid);
 905         }
 906
 907         lov_getref(export->exp_obd);
 908         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
 909                 struct lov_stripe_md obj_md;
 910                 struct lov_stripe_md *obj_mdp = &obj_md;
 911                 struct lov_tgt_desc *tgt;
 912                 int err;
 913
 914                 tgt = lov->lov_tgts[i];
 915                 if (!tgt)
 916                         continue;
 917
 918                 /* if called for a specific target, we don't
 919                    care if it is not active. */
 920                 if (!lov->lov_tgts[i]->ltd_active && ost_uuid == NULL) {
 921                         CDEBUG(D_HA, "lov idx %d inactive\n", i);
 922                         continue;
 923                 }
 924
 925                 if (ost_uuid && !obd_uuid_equals(ost_uuid, &tgt->ltd_uuid))
 926                         continue;
 927
 928                 CDEBUG(D_CONFIG,"Clear orphans for %d:%s\n", i,
 929                        obd_uuid2str(ost_uuid));
 930
 931                 memcpy(tmp_oa, src_oa, sizeof(*tmp_oa));
 932
 933                 LASSERT(lov->lov_tgts[i]->ltd_exp);
 934                 /* XXX: LOV STACKING: use real "obj_mdp" sub-data */
 935                 err = obd_create(lov->lov_tgts[i]->ltd_exp,
 936                                  tmp_oa, &obj_mdp, oti);
 937                 if (err)
 938                         /* This export will be disabled until it is recovered,
 939                            and then orphan recovery will be completed. */
 940                         CERROR("error in orphan recovery on OST idx %d/%d: "
 941                                "rc = %d\n", i, lov->desc.ld_tgt_count, err);
 942
 943                 if (ost_uuid)
 944                         break;
 945         }
 946         lov_putref(export->exp_obd);
 947
 948         OBDO_FREE(tmp_oa);
 949         RETURN(rc);
 950 }
 951
 952 static int lov_recreate(struct obd_export *exp, struct obdo *src_oa,
 953                         struct lov_stripe_md **ea, struct obd_trans_info *oti)
 954 {
 955         struct lov_stripe_md *obj_mdp, *lsm;
 956         struct lov_obd *lov = &exp->exp_obd->u.lov;
 957         unsigned ost_idx;
 958         int rc, i;
 959         ENTRY;
 960
 961         LASSERT(src_oa->o_valid & OBD_MD_FLFLAGS &&
 962                 src_oa->o_flags & OBD_FL_RECREATE_OBJS);
 963
 964         OBD_ALLOC(obj_mdp, sizeof(*obj_mdp));
 965         if (obj_mdp == NULL)
 966                 RETURN(-ENOMEM);
 967
 968         ost_idx = src_oa->o_nlink;
 969         lsm = *ea;
 970         if (lsm == NULL)
 971                 GOTO(out, rc = -EINVAL);
 972         if (ost_idx >= lov->desc.ld_tgt_count ||
 973             !lov->lov_tgts[ost_idx])
 974                 GOTO(out, rc = -EINVAL);
 975
 976         for (i = 0; i < lsm->lsm_stripe_count; i++) {
 977                 if (lsm->lsm_oinfo[i]->loi_ost_idx == ost_idx) {
 978                         if (lsm->lsm_oinfo[i]->loi_id != src_oa->o_id)
 979                                 GOTO(out, rc = -EINVAL);
 980                         break;
 981                 }
 982         }
 983         if (i == lsm->lsm_stripe_count)
 984                 GOTO(out, rc = -EINVAL);
 985
 986         rc = obd_create(lov->lov_tgts[ost_idx]->ltd_exp, src_oa, &obj_mdp, oti);
 987 out:
 988         OBD_FREE(obj_mdp, sizeof(*obj_mdp));
 989         RETURN(rc);
 990 }
 991
 992 /* the LOV expects oa->o_id to be set to the LOV object id */
 993 static int lov_create(struct obd_export *exp, struct obdo *src_oa,
 994                       struct lov_stripe_md **ea, struct obd_trans_info *oti)
 995 {
 996         struct lov_obd *lov;
 997         struct obd_info oinfo;
 998         struct lov_request_set *set = NULL;
 999         struct lov_request *req;
1000         struct obd_statfs osfs;
1001         __u64 maxage;
1002         int rc = 0;
1003         ENTRY;
1004
1005         LASSERT(ea != NULL);
1006         if (exp == NULL)
1007                 RETURN(-EINVAL);
1008
1009         if ((src_oa->o_valid & OBD_MD_FLFLAGS) &&
1010             src_oa->o_flags == OBD_FL_DELORPHAN) {
1011                 rc = lov_clear_orphans(exp, src_oa, ea, oti);
1012                 RETURN(rc);
1013         }
1014
1015         lov = &exp->exp_obd->u.lov;
1016         if (!lov->desc.ld_active_tgt_count)
1017                 RETURN(-EIO);
1018
1019         /* Recreate a specific object id at the given OST index */
1020         if ((src_oa->o_valid & OBD_MD_FLFLAGS) &&
1021             (src_oa->o_flags & OBD_FL_RECREATE_OBJS)) {
1022                  rc = lov_recreate(exp, src_oa, ea, oti);
1023                  RETURN(rc);
1024         }
1025
1026         maxage = cfs_time_shift_64(-lov->desc.ld_qos_maxage);
1027         obd_statfs_rqset(exp->exp_obd, &osfs, maxage, 1);
1028
1029         rc = lov_prep_create_set(exp, &oinfo, ea, src_oa, oti, &set);
1030         if (rc)
1031                 RETURN(rc);
1032
1033         list_for_each_entry(req, &set->set_list, rq_link) {
1034                 /* XXX: LOV STACKING: use real "obj_mdp" sub-data */
1035                 rc = obd_create(lov->lov_tgts[req->rq_idx]->ltd_exp,
1036                                 req->rq_oi.oi_oa, &req->rq_oi.oi_md, oti);
1037                 lov_update_create_set(set, req, rc);
1038         }
1039         rc = lov_fini_create_set(set, ea);
1040         RETURN(rc);
1041 }
1042
1043 #define ASSERT_LSM_MAGIC(lsmp)                                                  \
1044 do {                                                                            \
1045         LASSERT((lsmp) != NULL);                                                \
1046         LASSERTF(((lsmp)->lsm_magic == LOV_MAGIC ||                             \
1047                  (lsmp)->lsm_magic == LOV_MAGIC_JOIN), "%p->lsm_magic=%x\n",    \
1048                  (lsmp), (lsmp)->lsm_magic);                                    \
1049 } while (0)
1050
1051 static int lov_destroy(struct obd_export *exp, struct obdo *oa,
1052                        struct lov_stripe_md *lsm, struct obd_trans_info *oti,
1053                        struct obd_export *md_exp)
1054 {
1055         struct lov_request_set *set;
1056         struct obd_info oinfo;
1057         struct lov_request *req;
1058         struct list_head *pos;
1059         struct lov_obd *lov;
1060         int rc = 0, err;
1061         ENTRY;
1062
1063         ASSERT_LSM_MAGIC(lsm);
1064
1065         if (!exp || !exp->exp_obd)
1066                 RETURN(-ENODEV);
1067
1068         if (oa->o_valid & OBD_MD_FLCOOKIE) {
1069                 LASSERT(oti);
1070                 LASSERT(oti->oti_logcookies);
1071         }
1072
1073         lov = &exp->exp_obd->u.lov;
1074         rc = lov_prep_destroy_set(exp, &oinfo, oa, lsm, oti, &set);
1075         if (rc)
1076                 RETURN(rc);
1077
1078         list_for_each (pos, &set->set_list) {
1079                 int err;
1080                 req = list_entry(pos, struct lov_request, rq_link);
1081
1082                 if (oa->o_valid & OBD_MD_FLCOOKIE)
1083                         oti->oti_logcookies = set->set_cookies + req->rq_stripe;
1084
1085                 err = obd_destroy(lov->lov_tgts[req->rq_idx]->ltd_exp,
1086                                   req->rq_oi.oi_oa, NULL, oti, NULL);
1087                 err = lov_update_common_set(set, req, err);
1088                 if (err) {
1089                         CERROR("error: destroying objid "LPX64" subobj "
1090                                LPX64" on OST idx %d: rc = %d\n",
1091                                oa->o_id, req->rq_oi.oi_oa->o_id,
1092                                req->rq_idx, err);
1093                         if (!rc)
1094                                 rc = err;
1095                 }
1096         }
1097
1098         if (rc == 0) {
1099                 LASSERT(lsm_op_find(lsm->lsm_magic) != NULL);
1100                 rc = lsm_op_find(lsm->lsm_magic)->lsm_destroy(lsm, oa, md_exp);
1101         }
1102         err = lov_fini_destroy_set(set);
1103         RETURN(rc ? rc : err);
1104 }
1105
1106 static int lov_getattr(struct obd_export *exp, struct obd_info *oinfo)
1107 {
1108         struct lov_request_set *set;
1109         struct lov_request *req;
1110         struct list_head *pos;
1111         struct lov_obd *lov;
1112         int err = 0, rc = 0;
1113         ENTRY;
1114
1115         LASSERT(oinfo);
1116         ASSERT_LSM_MAGIC(oinfo->oi_md);
1117
1118         if (!exp || !exp->exp_obd)
1119                 RETURN(-ENODEV);
1120
1121         lov = &exp->exp_obd->u.lov;
1122
1123         rc = lov_prep_getattr_set(exp, oinfo, &set);
1124         if (rc)
1125                 RETURN(rc);
1126
1127         list_for_each (pos, &set->set_list) {
1128                 req = list_entry(pos, struct lov_request, rq_link);
1129
1130                 CDEBUG(D_INFO, "objid "LPX64"[%d] has subobj "LPX64" at idx "
1131                        "%u\n", oinfo->oi_oa->o_id, req->rq_stripe,
1132                        req->rq_oi.oi_oa->o_id, req->rq_idx);
1133
1134                 rc = obd_getattr(lov->lov_tgts[req->rq_idx]->ltd_exp,
1135                                  &req->rq_oi);
1136                 err = lov_update_common_set(set, req, rc);
1137                 if (err) {
1138                         CERROR("error: getattr objid "LPX64" subobj "
1139                                LPX64" on OST idx %d: rc = %d\n",
1140                                oinfo->oi_oa->o_id, req->rq_oi.oi_oa->o_id,
1141                                req->rq_idx, err);
1142                         break;
1143                 }
1144         }
1145
1146         rc = lov_fini_getattr_set(set);
1147         if (err)
1148                 rc = err;
1149         RETURN(rc);
1150 }
1151
1152 static int lov_getattr_interpret(struct ptlrpc_request_set *rqset,
1153                                  void *data, int rc)
1154 {
1155         struct lov_request_set *lovset = (struct lov_request_set *)data;
1156         int err;
1157         ENTRY;
1158
1159         /* don't do attribute merge if this aysnc op failed */
1160         if (rc)
1161                 lovset->set_completes = 0;
1162         err = lov_fini_getattr_set(lovset);
1163         RETURN(rc ? rc : err);
1164 }
1165
1166 static int lov_getattr_async(struct obd_export *exp, struct obd_info *oinfo,
1167                               struct ptlrpc_request_set *rqset)
1168 {
1169         struct lov_request_set *lovset;
1170         struct lov_obd *lov;
1171         struct list_head *pos;
1172         struct lov_request *req;
1173         int rc = 0, err;
1174         ENTRY;
1175
1176         LASSERT(oinfo);
1177         ASSERT_LSM_MAGIC(oinfo->oi_md);
1178
1179         if (!exp || !exp->exp_obd)
1180                 RETURN(-ENODEV);
1181
1182         lov = &exp->exp_obd->u.lov;
1183
1184         rc = lov_prep_getattr_set(exp, oinfo, &lovset);
1185         if (rc)
1186                 RETURN(rc);
1187
1188         CDEBUG(D_INFO, "objid "LPX64": %ux%u byte stripes\n",
1189                oinfo->oi_md->lsm_object_id, oinfo->oi_md->lsm_stripe_count,
1190                oinfo->oi_md->lsm_stripe_size);
1191
1192         list_for_each (pos, &lovset->set_list) {
1193                 req = list_entry(pos, struct lov_request, rq_link);
1194
1195                 CDEBUG(D_INFO, "objid "LPX64"[%d] has subobj "LPX64" at idx "
1196                        "%u\n", oinfo->oi_oa->o_id, req->rq_stripe,
1197                        req->rq_oi.oi_oa->o_id, req->rq_idx);
1198                 rc = obd_getattr_async(lov->lov_tgts[req->rq_idx]->ltd_exp,
1199                                        &req->rq_oi, rqset);
1200                 if (rc) {
1201                         CERROR("error: getattr objid "LPX64" subobj "
1202                                LPX64" on OST idx %d: rc = %d\n",
1203                                oinfo->oi_oa->o_id, req->rq_oi.oi_oa->o_id,
1204                                req->rq_idx, rc);
1205                         GOTO(out, rc);
1206                 }
1207         }
1208
1209         if (!list_empty(&rqset->set_requests)) {
1210                 LASSERT(rc == 0);
1211                 LASSERT (rqset->set_interpret == NULL);
1212                 rqset->set_interpret = lov_getattr_interpret;
1213                 rqset->set_arg = (void *)lovset;
1214                 RETURN(rc);
1215         }
1216 out:
1217         if (rc)
1218                 lovset->set_completes = 0;
1219         err = lov_fini_getattr_set(lovset);
1220         RETURN(rc ? rc : err);
1221 }
1222
1223 static int lov_setattr(struct obd_export *exp, struct obd_info *oinfo,
1224                        struct obd_trans_info *oti)
1225 {
1226         struct lov_request_set *set;
1227         struct lov_obd *lov;
1228         struct list_head *pos;
1229         struct lov_request *req;
1230         int err = 0, rc = 0;
1231         ENTRY;
1232
1233         LASSERT(oinfo);
1234         ASSERT_LSM_MAGIC(oinfo->oi_md);
1235
1236         if (!exp || !exp->exp_obd)
1237                 RETURN(-ENODEV);
1238
1239         /* for now, we only expect the following updates here */
1240         LASSERT(!(oinfo->oi_oa->o_valid & ~(OBD_MD_FLID | OBD_MD_FLTYPE |
1241                                             OBD_MD_FLMODE | OBD_MD_FLATIME |
1242                                             OBD_MD_FLMTIME | OBD_MD_FLCTIME |
1243                                             OBD_MD_FLFLAGS | OBD_MD_FLSIZE |
1244                                             OBD_MD_FLGROUP | OBD_MD_FLUID |
1245                                             OBD_MD_FLGID | OBD_MD_FLINLINE |
1246                                             OBD_MD_FLFID | OBD_MD_FLGENER)));
1247         lov = &exp->exp_obd->u.lov;
1248         rc = lov_prep_setattr_set(exp, oinfo, oti, &set);
1249         if (rc)
1250                 RETURN(rc);
1251
1252         list_for_each (pos, &set->set_list) {
1253                 req = list_entry(pos, struct lov_request, rq_link);
1254
1255                 rc = obd_setattr(lov->lov_tgts[req->rq_idx]->ltd_exp,
1256                                  &req->rq_oi, NULL);
1257                 err = lov_update_setattr_set(set, req, rc);
1258                 if (err) {
1259                         CERROR("error: setattr objid "LPX64" subobj "
1260                                LPX64" on OST idx %d: rc = %d\n",
1261                                set->set_oi->oi_oa->o_id,
1262                                req->rq_oi.oi_oa->o_id, req->rq_idx, err);
1263                         if (!rc)
1264                                 rc = err;
1265                 }
1266         }
1267         err = lov_fini_setattr_set(set);
1268         if (!rc)
1269                 rc = err;
1270         RETURN(rc);
1271 }
1272
1273 static int lov_setattr_interpret(struct ptlrpc_request_set *rqset,
1274                                  void *data, int rc)
1275 {
1276         struct lov_request_set *lovset = (struct lov_request_set *)data;
1277         int err;
1278         ENTRY;
1279
1280         if (rc)
1281                 lovset->set_completes = 0;
1282         err = lov_fini_setattr_set(lovset);
1283         RETURN(rc ? rc : err);
1284 }
1285
1286 /* If @oti is given, the request goes from MDS and responses from OSTs are not
1287    needed. Otherwise, a client is waiting for responses. */
1288 static int lov_setattr_async(struct obd_export *exp, struct obd_info *oinfo,
1289                              struct obd_trans_info *oti,
1290                              struct ptlrpc_request_set *rqset)
1291 {
1292         struct lov_request_set *set;
1293         struct lov_request *req;
1294         struct list_head *pos;
1295         struct lov_obd *lov;
1296         int rc = 0;
1297         ENTRY;
1298
1299         LASSERT(oinfo);
1300         ASSERT_LSM_MAGIC(oinfo->oi_md);
1301         if (oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE) {
1302                 LASSERT(oti);
1303                 LASSERT(oti->oti_logcookies);
1304         }
1305
1306         if (!exp || !exp->exp_obd)
1307                 RETURN(-ENODEV);
1308
1309         lov = &exp->exp_obd->u.lov;
1310         rc = lov_prep_setattr_set(exp, oinfo, oti, &set);
1311         if (rc)
1312                 RETURN(rc);
1313
1314         CDEBUG(D_INFO, "objid "LPX64": %ux%u byte stripes\n",
1315                oinfo->oi_md->lsm_object_id, oinfo->oi_md->lsm_stripe_count,
1316                oinfo->oi_md->lsm_stripe_size);
1317
1318         list_for_each (pos, &set->set_list) {
1319                 req = list_entry(pos, struct lov_request, rq_link);
1320
1321                 if (oinfo->oi_oa->o_valid & OBD_MD_FLCOOKIE)
1322                         oti->oti_logcookies = set->set_cookies + req->rq_stripe;
1323
1324                 CDEBUG(D_INFO, "objid "LPX64"[%d] has subobj "LPX64" at idx "
1325                        "%u\n", oinfo->oi_oa->o_id, req->rq_stripe,
1326                        req->rq_oi.oi_oa->o_id, req->rq_idx);
1327
1328                 rc = obd_setattr_async(lov->lov_tgts[req->rq_idx]->ltd_exp,
1329                                        &req->rq_oi, oti, rqset);
1330                 if (rc) {
1331                         CERROR("error: setattr objid "LPX64" subobj "
1332                                LPX64" on OST idx %d: rc = %d\n",
1333                                set->set_oi->oi_oa->o_id,
1334                                req->rq_oi.oi_oa->o_id,
1335                                req->rq_idx, rc);
1336                         break;
1337                 }
1338         }
1339
1340         /* If we are not waiting for responses on async requests, return. */
1341         if (rc || !rqset || list_empty(&rqset->set_requests)) {
1342                 int err;
1343                 if (rc)
1344                         set->set_completes = 0;
1345                 err = lov_fini_setattr_set(set);
1346                 RETURN(rc ? rc : err);
1347         }
1348
1349         LASSERT(rqset->set_interpret == NULL);
1350         rqset->set_interpret = lov_setattr_interpret;
1351         rqset->set_arg = (void *)set;
1352
1353         RETURN(0);
1354 }
1355
1356 static int lov_punch_interpret(struct ptlrpc_request_set *rqset,
1357                                void *data, int rc)
1358 {
1359         struct lov_request_set *lovset = (struct lov_request_set *)data;
1360         int err;
1361         ENTRY;
1362
1363         if (rc)
1364                 lovset->set_completes = 0;
1365         err = lov_fini_punch_set(lovset);
1366         RETURN(rc ? rc : err);
1367 }
1368
1369 /* FIXME: maybe we'll just make one node the authoritative attribute node, then
1370  * we can send this 'punch' to just the authoritative node and the nodes
1371  * that the punch will affect. */
1372 static int lov_punch(struct obd_export *exp, struct obd_info *oinfo,
1373                      struct obd_trans_info *oti,
1374                      struct ptlrpc_request_set *rqset)
1375 {
1376         struct lov_request_set *set;
1377         struct lov_obd *lov;
1378         struct list_head *pos;
1379         struct lov_request *req;
1380         int rc = 0;
1381         ENTRY;
1382
1383         LASSERT(oinfo);
1384         ASSERT_LSM_MAGIC(oinfo->oi_md);
1385
1386         if (!exp || !exp->exp_obd)
1387                 RETURN(-ENODEV);
1388
1389         lov = &exp->exp_obd->u.lov;
1390         rc = lov_prep_punch_set(exp, oinfo, oti, &set);
1391         if (rc)
1392                 RETURN(rc);
1393
1394         list_for_each (pos, &set->set_list) {
1395                 req = list_entry(pos, struct lov_request, rq_link);
1396
1397                 rc = obd_punch(lov->lov_tgts[req->rq_idx]->ltd_exp,
1398                                &req->rq_oi, NULL, rqset);
1399                 if (rc) {
1400                         CERROR("error: punch objid "LPX64" subobj "LPX64
1401                                " on OST idx %d: rc = %d\n",
1402                                set->set_oi->oi_oa->o_id,
1403                                req->rq_oi.oi_oa->o_id, req->rq_idx, rc);
1404                         break;
1405                 }
1406         }
1407
1408         if (rc || list_empty(&rqset->set_requests)) {
1409                 int err;
1410                 err = lov_fini_punch_set(set);
1411                 RETURN(rc ? rc : err);
1412         }
1413
1414         LASSERT(rqset->set_interpret == NULL);
1415         rqset->set_interpret = lov_punch_interpret;
1416         rqset->set_arg = (void *)set;
1417
1418         RETURN(0);
1419 }
1420
1421 static int lov_sync(struct obd_export *exp, struct obdo *oa,
1422                     struct lov_stripe_md *lsm, obd_off start, obd_off end,
1423                     void *capa)
1424 {
1425         struct lov_request_set *set;
1426         struct obd_info oinfo;
1427         struct lov_obd *lov;
1428         struct list_head *pos;
1429         struct lov_request *req;
1430         int err = 0, rc = 0;
1431         ENTRY;
1432
1433         ASSERT_LSM_MAGIC(lsm);
1434
1435         if (!exp->exp_obd)
1436                 RETURN(-ENODEV);
1437
1438         lov = &exp->exp_obd->u.lov;
1439         rc = lov_prep_sync_set(exp, &oinfo, oa, lsm, start, end, &set);
1440         if (rc)
1441                 RETURN(rc);
1442
1443         list_for_each (pos, &set->set_list) {
1444                 req = list_entry(pos, struct lov_request, rq_link);
1445
1446                 rc = obd_sync(lov->lov_tgts[req->rq_idx]->ltd_exp,
1447                               req->rq_oi.oi_oa, NULL,
1448                               req->rq_oi.oi_policy.l_extent.start,
1449                               req->rq_oi.oi_policy.l_extent.end, capa);
1450                 err = lov_update_common_set(set, req, rc);
1451                 if (err) {
1452                         CERROR("error: fsync objid "LPX64" subobj "LPX64
1453                                " on OST idx %d: rc = %d\n",
1454                                set->set_oi->oi_oa->o_id,
1455                                req->rq_oi.oi_oa->o_id, req->rq_idx, rc);
1456                         if (!rc)
1457                                 rc = err;
1458                 }
1459         }
1460         err = lov_fini_sync_set(set);
1461         if (!rc)
1462                 rc = err;
1463         RETURN(rc);
1464 }
1465
1466 static int lov_brw_check(struct lov_obd *lov, struct obd_info *lov_oinfo,
1467                          obd_count oa_bufs, struct brw_page *pga)
1468 {
1469         struct obd_info oinfo = { { { 0 } } };
1470         int i, rc = 0;
1471
1472         oinfo.oi_oa = lov_oinfo->oi_oa;
1473
1474         /* The caller just wants to know if there's a chance that this
1475          * I/O can succeed */
1476         for (i = 0; i < oa_bufs; i++) {
1477                 int stripe = lov_stripe_number(lov_oinfo->oi_md, pga[i].off);
1478                 int ost = lov_oinfo->oi_md->lsm_oinfo[stripe]->loi_ost_idx;
1479                 obd_off start, end;
1480
1481                 if (!lov_stripe_intersects(lov_oinfo->oi_md, i, pga[i].off,
1482                                            pga[i].off + pga[i].count,
1483                                            &start, &end))
1484                         continue;
1485
1486                 if (!lov->lov_tgts[ost] || !lov->lov_tgts[ost]->ltd_active) {
1487                         CDEBUG(D_HA, "lov idx %d inactive\n", ost);
1488                         return -EIO;
1489                 }
1490
1491                 rc = obd_brw(OBD_BRW_CHECK, lov->lov_tgts[ost]->ltd_exp, &oinfo,
1492                              1, &pga[i], NULL);
1493                 if (rc)
1494                         break;
1495         }
1496         return rc;
1497 }
1498
1499 static int lov_brw(int cmd, struct obd_export *exp, struct obd_info *oinfo,
1500                    obd_count oa_bufs, struct brw_page *pga,
1501                    struct obd_trans_info *oti)
1502 {
1503         struct lov_request_set *set;
1504         struct lov_request *req;
1505         struct list_head *pos;
1506         struct lov_obd *lov = &exp->exp_obd->u.lov;
1507         int err, rc = 0;
1508         ENTRY;
1509
1510         ASSERT_LSM_MAGIC(oinfo->oi_md);
1511
1512         if (cmd == OBD_BRW_CHECK) {
1513                 rc = lov_brw_check(lov, oinfo, oa_bufs, pga);
1514                 RETURN(rc);
1515         }
1516
1517         rc = lov_prep_brw_set(exp, oinfo, oa_bufs, pga, oti, &set);
1518         if (rc)
1519                 RETURN(rc);
1520
1521         list_for_each (pos, &set->set_list) {
1522                 struct obd_export *sub_exp;
1523                 struct brw_page *sub_pga;
1524                 req = list_entry(pos, struct lov_request, rq_link);
1525
1526                 sub_exp = lov->lov_tgts[req->rq_idx]->ltd_exp;
1527                 sub_pga = set->set_pga + req->rq_pgaidx;
1528                 rc = obd_brw(cmd, sub_exp, &req->rq_oi, req->rq_oabufs,
1529                              sub_pga, oti);
1530                 if (rc)
1531                         break;
1532                 lov_update_common_set(set, req, rc);
1533         }
1534
1535         err = lov_fini_brw_set(set);
1536         if (!rc)
1537                 rc = err;
1538         RETURN(rc);
1539 }
1540
1541 static int lov_brw_interpret(struct ptlrpc_request_set *reqset, void *data,
1542                              int rc)
1543 {
1544         struct lov_request_set *lovset = (struct lov_request_set *)data;
1545         ENTRY;
1546
1547         if (rc) {
1548                 lovset->set_completes = 0;
1549                 lov_fini_brw_set(lovset);
1550         } else {
1551                 rc = lov_fini_brw_set(lovset);
1552         }
1553
1554         RETURN(rc);
1555 }
1556
1557 static int lov_brw_async(int cmd, struct obd_export *exp,
1558                          struct obd_info *oinfo, obd_count oa_bufs,
1559                          struct brw_page *pga, struct obd_trans_info *oti,
1560                          struct ptlrpc_request_set *set)
1561 {
1562         struct lov_request_set *lovset;
1563         struct lov_request *req;
1564         struct list_head *pos;
1565         struct lov_obd *lov = &exp->exp_obd->u.lov;
1566         int rc = 0;
1567         ENTRY;
1568
1569         LASSERT(oinfo);
1570         ASSERT_LSM_MAGIC(oinfo->oi_md);
1571
1572         if (cmd == OBD_BRW_CHECK) {
1573                 rc = lov_brw_check(lov, oinfo, oa_bufs, pga);
1574                 RETURN(rc);
1575         }
1576
1577         rc = lov_prep_brw_set(exp, oinfo, oa_bufs, pga, oti, &lovset);
1578         if (rc)
1579                 RETURN(rc);
1580
1581         list_for_each (pos, &lovset->set_list) {
1582                 struct obd_export *sub_exp;
1583                 struct brw_page *sub_pga;
1584                 req = list_entry(pos, struct lov_request, rq_link);
1585
1586                 sub_exp = lov->lov_tgts[req->rq_idx]->ltd_exp;
1587                 sub_pga = lovset->set_pga + req->rq_pgaidx;
1588                 rc = obd_brw_async(cmd, sub_exp, &req->rq_oi, req->rq_oabufs,
1589                                    sub_pga, oti, set);
1590                 if (rc)
1591                         GOTO(out, rc);
1592                 lov_update_common_set(lovset, req, rc);
1593         }
1594         LASSERT(rc == 0);
1595         LASSERT(set->set_interpret == NULL);
1596         LASSERT(set->set_arg == NULL);
1597         rc = ptlrpc_set_add_cb(set, lov_brw_interpret, lovset);
1598         if (rc)
1599                 GOTO(out, rc);
1600
1601         RETURN(rc);
1602 out:
1603         lov_fini_brw_set(lovset);
1604         RETURN(rc);
1605 }
1606
1607 static int lov_ap_make_ready(void *data, int cmd)
1608 {
1609         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1610
1611         return lap->lap_caller_ops->ap_make_ready(lap->lap_caller_data, cmd);
1612 }
1613
1614 static int lov_ap_refresh_count(void *data, int cmd)
1615 {
1616         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1617
1618         return lap->lap_caller_ops->ap_refresh_count(lap->lap_caller_data,
1619                                                      cmd);
1620 }
1621
1622 static void lov_ap_fill_obdo(void *data, int cmd, struct obdo *oa)
1623 {
1624         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1625
1626         lap->lap_caller_ops->ap_fill_obdo(lap->lap_caller_data, cmd, oa);
1627         /* XXX woah, shouldn't we be altering more here?  size? */
1628         oa->o_id = lap->lap_loi_id;
1629         oa->o_gr = lap->lap_loi_gr;
1630         oa->o_valid |= OBD_MD_FLGROUP;
1631         oa->o_stripe_idx = lap->lap_stripe;
1632 }
1633
1634 static void lov_ap_update_obdo(void *data, int cmd, struct obdo *oa,
1635                                obd_valid valid)
1636 {
1637         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1638
1639         lap->lap_caller_ops->ap_update_obdo(lap->lap_caller_data, cmd,oa,valid);
1640 }
1641
1642 static int lov_ap_completion(void *data, int cmd, struct obdo *oa, int rc)
1643 {
1644         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1645
1646         /* in a raid1 regime this would down a count of many ios
1647          * in flight, onl calling the caller_ops completion when all
1648          * the raid1 ios are complete */
1649         rc = lap->lap_caller_ops->ap_completion(lap->lap_caller_data,cmd,oa,rc);
1650         return rc;
1651 }
1652
1653 static struct obd_capa *lov_ap_lookup_capa(void *data, int cmd)
1654 {
1655         struct lov_async_page *lap = LAP_FROM_COOKIE(data);
1656         return lap->lap_caller_ops->ap_lookup_capa(lap->lap_caller_data, cmd);
1657 }
1658
1659 static struct obd_async_page_ops lov_async_page_ops = {
1660         .ap_make_ready =        lov_ap_make_ready,
1661         .ap_refresh_count =     lov_ap_refresh_count,
1662         .ap_fill_obdo =         lov_ap_fill_obdo,
1663         .ap_update_obdo =       lov_ap_update_obdo,
1664         .ap_completion =        lov_ap_completion,
1665         .ap_lookup_capa =       lov_ap_lookup_capa,
1666 };
1667
1668 int lov_prep_async_page(struct obd_export *exp, struct lov_stripe_md *lsm,
1669                            struct lov_oinfo *loi, cfs_page_t *page,
1670                            obd_off offset, struct obd_async_page_ops *ops,
1671                            void *data, void **res)
1672 {
1673         struct lov_obd *lov = &exp->exp_obd->u.lov;
1674         struct lov_async_page *lap;
1675         int rc = 0;
1676         ENTRY;
1677
1678         if (!page) {
1679                 int i = 0;
1680                 /* Find an existing osc so we can get it's stupid sizeof(*oap).
1681                    Only because of this layering limitation will a client
1682                    mount with no osts fail */
1683                 while (!lov->lov_tgts || !lov->lov_tgts[i] ||
1684                        !lov->lov_tgts[i]->ltd_exp) {
1685                         i++;
1686                         if (i >= lov->desc.ld_tgt_count)
1687                                 RETURN(-ENOMEDIUM);
1688                 }
1689                 rc = size_round(sizeof(*lap)) +
1690                         obd_prep_async_page(lov->lov_tgts[i]->ltd_exp, NULL,
1691                                             NULL, NULL, 0, NULL, NULL, NULL);
1692                 RETURN(rc);
1693         }
1694         ASSERT_LSM_MAGIC(lsm);
1695         LASSERT(loi == NULL);
1696
1697         lap = *res;
1698         lap->lap_magic = LOV_AP_MAGIC;
1699         lap->lap_caller_ops = ops;
1700         lap->lap_caller_data = data;
1701
1702         /* for now only raid 0 which passes through */
1703         lap->lap_stripe = lov_stripe_number(lsm, offset);
1704         lov_stripe_offset(lsm, offset, lap->lap_stripe, &lap->lap_sub_offset);
1705         loi = lsm->lsm_oinfo[lap->lap_stripe];
1706
1707         /* so the callback doesn't need the lsm */
1708         lap->lap_loi_id = loi->loi_id;
1709         lap->lap_loi_gr = lsm->lsm_object_gr;
1710         LASSERT(lsm->lsm_object_gr > 0);
1711
1712         lap->lap_sub_cookie = (void *)lap + size_round(sizeof(*lap));
1713
1714         rc = obd_prep_async_page(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
1715                                  lsm, loi, page, lap->lap_sub_offset,
1716                                  &lov_async_page_ops, lap,
1717                                  &lap->lap_sub_cookie);
1718         if (rc)
1719                 RETURN(rc);
1720         CDEBUG(D_CACHE, "lap %p page %p cookie %p off "LPU64"\n", lap, page,
1721                lap->lap_sub_cookie, offset);
1722         RETURN(0);
1723 }
1724
1725 static int lov_queue_async_io(struct obd_export *exp,
1726                               struct lov_stripe_md *lsm,
1727                               struct lov_oinfo *loi, void *cookie,
1728                               int cmd, obd_off off, int count,
1729                               obd_flag brw_flags, obd_flag async_flags)
1730 {
1731         struct lov_obd *lov = &exp->exp_obd->u.lov;
1732         struct lov_async_page *lap;
1733         int rc;
1734
1735         LASSERT(loi == NULL);
1736
1737         ASSERT_LSM_MAGIC(lsm);
1738
1739         lap = LAP_FROM_COOKIE(cookie);
1740
1741         loi = lsm->lsm_oinfo[lap->lap_stripe];
1742
1743         rc = obd_queue_async_io(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp, lsm,
1744                                 loi, lap->lap_sub_cookie, cmd, off, count,
1745                                 brw_flags, async_flags);
1746         RETURN(rc);
1747 }
1748
1749 static int lov_set_async_flags(struct obd_export *exp,
1750                                struct lov_stripe_md *lsm,
1751                                struct lov_oinfo *loi, void *cookie,
1752                                obd_flag async_flags)
1753 {
1754         struct lov_obd *lov = &exp->exp_obd->u.lov;
1755         struct lov_async_page *lap;
1756         int rc;
1757
1758         LASSERT(loi == NULL);
1759
1760         ASSERT_LSM_MAGIC(lsm);
1761
1762         lap = LAP_FROM_COOKIE(cookie);
1763
1764         loi = lsm->lsm_oinfo[lap->lap_stripe];
1765
1766         rc = obd_set_async_flags(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
1767                                  lsm, loi, lap->lap_sub_cookie, async_flags);
1768         RETURN(rc);
1769 }
1770
1771 static int lov_queue_group_io(struct obd_export *exp,
1772                               struct lov_stripe_md *lsm,
1773                               struct lov_oinfo *loi,
1774                               struct obd_io_group *oig, void *cookie,
1775                               int cmd, obd_off off, int count,
1776                               obd_flag brw_flags, obd_flag async_flags)
1777 {
1778         struct lov_obd *lov = &exp->exp_obd->u.lov;
1779         struct lov_async_page *lap;
1780         int rc;
1781
1782         LASSERT(loi == NULL);
1783
1784         ASSERT_LSM_MAGIC(lsm);
1785
1786         lap = LAP_FROM_COOKIE(cookie);
1787
1788         loi = lsm->lsm_oinfo[lap->lap_stripe];
1789
1790         rc = obd_queue_group_io(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp, lsm,
1791                                 loi, oig, lap->lap_sub_cookie, cmd, off, count,
1792                                 brw_flags, async_flags);
1793         RETURN(rc);
1794 }
1795
1796 /* this isn't exactly optimal.  we may have queued sync io in oscs on
1797  * all stripes, but we don't record that fact at queue time.  so we
1798  * trigger sync io on all stripes. */
1799 static int lov_trigger_group_io(struct obd_export *exp,
1800                                 struct lov_stripe_md *lsm,
1801                                 struct lov_oinfo *loi,
1802                                 struct obd_io_group *oig)
1803 {
1804         struct lov_obd *lov = &exp->exp_obd->u.lov;
1805         int rc = 0, i, err;
1806
1807         LASSERT(loi == NULL);
1808
1809         ASSERT_LSM_MAGIC(lsm);
1810
1811         for (i = 0; i < lsm->lsm_stripe_count; i++) {
1812                 loi = lsm->lsm_oinfo[i];
1813                 if (!lov->lov_tgts[loi->loi_ost_idx] ||
1814                     !lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
1815                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
1816                         continue;
1817                 }
1818
1819                 err = obd_trigger_group_io(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
1820                                            lsm, loi, oig);
1821                 if (rc == 0 && err != 0)
1822                         rc = err;
1823         };
1824         RETURN(rc);
1825 }
1826
1827 static int lov_teardown_async_page(struct obd_export *exp,
1828                                    struct lov_stripe_md *lsm,
1829                                    struct lov_oinfo *loi, void *cookie)
1830 {
1831         struct lov_obd *lov = &exp->exp_obd->u.lov;
1832         struct lov_async_page *lap;
1833         int rc;
1834
1835         LASSERT(loi == NULL);
1836
1837         ASSERT_LSM_MAGIC(lsm);
1838
1839         lap = LAP_FROM_COOKIE(cookie);
1840
1841         loi = lsm->lsm_oinfo[lap->lap_stripe];
1842
1843         rc = obd_teardown_async_page(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
1844                                      lsm, loi, lap->lap_sub_cookie);
1845         if (rc) {
1846                 CERROR("unable to teardown sub cookie %p: %d\n",
1847                        lap->lap_sub_cookie, rc);
1848                 RETURN(rc);
1849         }
1850         RETURN(rc);
1851 }
1852
1853 static int lov_enqueue_interpret(struct ptlrpc_request_set *rqset,
1854                                  void *data, int rc)
1855 {
1856         struct lov_request_set *lovset = (struct lov_request_set *)data;
1857         ENTRY;
1858         rc = lov_fini_enqueue_set(lovset, lovset->set_ei->ei_mode, rc, rqset);
1859         RETURN(rc);
1860 }
1861
1862 static int lov_enqueue(struct obd_export *exp, struct obd_info *oinfo,
1863                        struct ldlm_enqueue_info *einfo,
1864                        struct ptlrpc_request_set *rqset)
1865 {
1866         ldlm_mode_t mode = einfo->ei_mode;
1867         struct lov_request_set *set;
1868         struct lov_request *req;
1869         struct list_head *pos;
1870         struct lov_obd *lov;
1871         ldlm_error_t rc;
1872         ENTRY;
1873
1874         LASSERT(oinfo);
1875         ASSERT_LSM_MAGIC(oinfo->oi_md);
1876         LASSERT(mode == (mode & -mode));
1877
1878         /* we should never be asked to replay a lock this way. */
1879         LASSERT((oinfo->oi_flags & LDLM_FL_REPLAY) == 0);
1880
1881         if (!exp || !exp->exp_obd)
1882                 RETURN(-ENODEV);
1883
1884         lov = &exp->exp_obd->u.lov;
1885         rc = lov_prep_enqueue_set(exp, oinfo, einfo, &set);
1886         if (rc)
1887                 RETURN(rc);
1888
1889         list_for_each (pos, &set->set_list) {
1890                 req = list_entry(pos, struct lov_request, rq_link);
1891
1892                 rc = obd_enqueue(lov->lov_tgts[req->rq_idx]->ltd_exp,
1893                                  &req->rq_oi, einfo, rqset);
1894                 if (rc != ELDLM_OK)
1895                         GOTO(out, rc);
1896         }
1897
1898         if (rqset && !list_empty(&rqset->set_requests)) {
1899                 LASSERT(rc == 0);
1900                 LASSERT(rqset->set_interpret == NULL);
1901                 rqset->set_interpret = lov_enqueue_interpret;
1902                 rqset->set_arg = (void *)set;
1903                 RETURN(rc);
1904         }
1905 out:
1906         rc = lov_fini_enqueue_set(set, mode, rc, rqset);
1907         RETURN(rc);
1908 }
1909
1910 static int lov_match(struct obd_export *exp, struct lov_stripe_md *lsm,
1911                      __u32 type, ldlm_policy_data_t *policy, __u32 mode,
1912                      int *flags, void *data, struct lustre_handle *lockh)
1913 {
1914         struct lov_request_set *set;
1915         struct obd_info oinfo;
1916         struct lov_request *req;
1917         struct list_head *pos;
1918         struct lov_obd *lov = &exp->exp_obd->u.lov;
1919         struct lustre_handle *lov_lockhp;
1920         int lov_flags, rc = 0;
1921         ENTRY;
1922
1923         ASSERT_LSM_MAGIC(lsm);
1924         LASSERT((*flags & LDLM_FL_TEST_LOCK) || mode == (mode & -mode));
1925
1926         if (!exp || !exp->exp_obd)
1927                 RETURN(-ENODEV);
1928
1929         lov = &exp->exp_obd->u.lov;
1930         rc = lov_prep_match_set(exp, &oinfo, lsm, policy, mode, lockh, &set);
1931         if (rc)
1932                 RETURN(rc);
1933
1934         list_for_each (pos, &set->set_list) {
1935                 ldlm_policy_data_t sub_policy;
1936                 req = list_entry(pos, struct lov_request, rq_link);
1937                 lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
1938                 LASSERT(lov_lockhp);
1939
1940                 lov_flags = *flags;
1941                 sub_policy.l_extent = req->rq_oi.oi_policy.l_extent;
1942
1943                 rc = obd_match(lov->lov_tgts[req->rq_idx]->ltd_exp,
1944                                req->rq_oi.oi_md, type, &sub_policy,
1945                                mode, &lov_flags, data, lov_lockhp);
1946                 rc = lov_update_match_set(set, req, rc);
1947                 if (rc <= 0)
1948                         break;
1949         }
1950         lov_fini_match_set(set, mode, *flags);
1951         RETURN(rc);
1952 }
1953
1954 static int lov_change_cbdata(struct obd_export *exp,
1955                              struct lov_stripe_md *lsm, ldlm_iterator_t it,
1956                              void *data)
1957 {
1958         struct lov_obd *lov;
1959         int rc = 0, i;
1960         ENTRY;
1961
1962         ASSERT_LSM_MAGIC(lsm);
1963
1964         if (!exp || !exp->exp_obd)
1965                 RETURN(-ENODEV);
1966
1967         LASSERT(lsm->lsm_object_gr > 0);
1968
1969         lov = &exp->exp_obd->u.lov;
1970         for (i = 0; i < lsm->lsm_stripe_count; i++) {
1971                 struct lov_stripe_md submd;
1972                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
1973
1974                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
1975                         CDEBUG(D_HA, "lov idx %d NULL \n", loi->loi_ost_idx);
1976                         continue;
1977                 }
1978
1979                 submd.lsm_object_id = loi->loi_id;
1980                 submd.lsm_object_gr = lsm->lsm_object_gr;
1981                 submd.lsm_stripe_count = 0;
1982                 rc = obd_change_cbdata(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
1983                                        &submd, it, data);
1984         }
1985         RETURN(rc);
1986 }
1987
1988 static int lov_cancel(struct obd_export *exp, struct lov_stripe_md *lsm,
1989                       __u32 mode, struct lustre_handle *lockh)
1990 {
1991         struct lov_request_set *set;
1992         struct obd_info oinfo;
1993         struct lov_request *req;
1994         struct list_head *pos;
1995         struct lov_obd *lov = &exp->exp_obd->u.lov;
1996         struct lustre_handle *lov_lockhp;
1997         int err = 0, rc = 0;
1998         ENTRY;
1999
2000         ASSERT_LSM_MAGIC(lsm);
2001
2002         if (!exp || !exp->exp_obd)
2003                 RETURN(-ENODEV);
2004
2005         LASSERT(lsm->lsm_object_gr > 0);
2006         LASSERT(lockh);
2007         lov = &exp->exp_obd->u.lov;
2008         rc = lov_prep_cancel_set(exp, &oinfo, lsm, mode, lockh, &set);
2009         if (rc)
2010                 RETURN(rc);
2011
2012         list_for_each (pos, &set->set_list) {
2013                 req = list_entry(pos, struct lov_request, rq_link);
2014                 lov_lockhp = set->set_lockh->llh_handles + req->rq_stripe;
2015
2016                 rc = obd_cancel(lov->lov_tgts[req->rq_idx]->ltd_exp,
2017                                 req->rq_oi.oi_md, mode, lov_lockhp);
2018                 rc = lov_update_common_set(set, req, rc);
2019                 if (rc) {
2020                         CERROR("error: cancel objid "LPX64" subobj "
2021                                LPX64" on OST idx %d: rc = %d\n",
2022                                lsm->lsm_object_id,
2023                                req->rq_oi.oi_md->lsm_object_id,
2024                                req->rq_idx, rc);
2025                         err = rc;
2026                 }
2027
2028         }
2029         lov_fini_cancel_set(set);
2030         RETURN(err);
2031 }
2032
2033 static int lov_cancel_unused(struct obd_export *exp,
2034                              struct lov_stripe_md *lsm,
2035                              int flags, void *opaque)
2036 {
2037         struct lov_obd *lov;
2038         int rc = 0, i;
2039         ENTRY;
2040
2041         if (!exp || !exp->exp_obd)
2042                 RETURN(-ENODEV);
2043
2044         lov = &exp->exp_obd->u.lov;
2045         if (lsm == NULL) {
2046                 for (i = 0; i < lov->desc.ld_tgt_count; i++) {
2047                         int err;
2048                         if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_exp)
2049                                 continue;
2050
2051                         err = obd_cancel_unused(lov->lov_tgts[i]->ltd_exp, NULL,
2052                                                 flags, opaque);
2053                         if (!rc)
2054                                 rc = err;
2055                 }
2056                 RETURN(rc);
2057         }
2058
2059         ASSERT_LSM_MAGIC(lsm);
2060
2061         LASSERT(lsm->lsm_object_gr > 0);
2062         for (i = 0; i < lsm->lsm_stripe_count; i++) {
2063                 struct lov_stripe_md submd;
2064                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
2065                 int err;
2066
2067                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
2068                         CDEBUG(D_HA, "lov idx %d NULL\n", loi->loi_ost_idx);
2069                         continue;
2070                 }
2071
2072                 if (!lov->lov_tgts[loi->loi_ost_idx]->ltd_active)
2073                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
2074
2075                 submd.lsm_object_id = loi->loi_id;
2076                 submd.lsm_object_gr = lsm->lsm_object_gr;
2077                 submd.lsm_stripe_count = 0;
2078                 err = obd_cancel_unused(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
2079                                         &submd, flags, opaque);
2080                 if (err && lov->lov_tgts[loi->loi_ost_idx]->ltd_active) {
2081                         CERROR("error: cancel unused objid "LPX64" subobj "LPX64
2082                                " on OST idx %d: rc = %d\n", lsm->lsm_object_id,
2083                                loi->loi_id, loi->loi_ost_idx, err);
2084                         if (!rc)
2085                                 rc = err;
2086                 }
2087         }
2088         RETURN(rc);
2089 }
2090
2091 static int lov_join_lru(struct obd_export *exp,
2092                         struct lov_stripe_md *lsm, int join)
2093 {
2094         struct lov_obd *lov;
2095         int i, count = 0;
2096         ENTRY;
2097
2098         ASSERT_LSM_MAGIC(lsm);
2099         if (!exp || !exp->exp_obd)
2100                 RETURN(-ENODEV);
2101
2102         lov = &exp->exp_obd->u.lov;
2103         for (i = 0; i < lsm->lsm_stripe_count; i++) {
2104                 struct lov_stripe_md submd;
2105                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
2106                 int rc = 0;
2107
2108                 if (!lov->lov_tgts[loi->loi_ost_idx]) {
2109                         CDEBUG(D_HA, "lov idx %d NULL\n", loi->loi_ost_idx);
2110                         continue;
2111                 }
2112
2113                 if (!lov->lov_tgts[loi->loi_ost_idx]->ltd_active)
2114                         CDEBUG(D_HA, "lov idx %d inactive\n", loi->loi_ost_idx);
2115
2116                 submd.lsm_object_id = loi->loi_id;
2117                 submd.lsm_object_gr = lsm->lsm_object_gr;
2118                 submd.lsm_stripe_count = 0;
2119                 rc = obd_join_lru(lov->lov_tgts[loi->loi_ost_idx]->ltd_exp,
2120                                   &submd, join);
2121                 if (rc < 0) {
2122                         CERROR("join lru failed. objid: "LPX64" subobj: "LPX64
2123                                " ostidx: %d rc: %d\n", lsm->lsm_object_id,
2124                                loi->loi_id, loi->loi_ost_idx, rc);
2125                         return rc;
2126                 } else {
2127                         count += rc;
2128                 }
2129         }
2130         RETURN(count);
2131 }
2132
2133 static int lov_statfs_interpret(struct ptlrpc_request_set *rqset,
2134                                 void *data, int rc)
2135 {
2136         struct lov_request_set *lovset = (struct lov_request_set *)data;
2137         int err;
2138         ENTRY;
2139
2140         if (rc)
2141                 lovset->set_completes = 0;
2142
2143         err = lov_fini_statfs_set(lovset);
2144         RETURN(rc ? rc : err);
2145 }
2146
2147 static int lov_statfs_async(struct obd_device *obd, struct obd_info *oinfo,
2148                             __u64 max_age, struct ptlrpc_request_set *rqset)
2149 {
2150         struct lov_request_set *set;
2151         struct lov_request *req;
2152         struct list_head *pos;
2153         struct lov_obd *lov;
2154         int rc = 0;
2155         ENTRY;
2156
2157         LASSERT(oinfo != NULL);
2158         LASSERT(oinfo->oi_osfs != NULL);
2159
2160         lov = &obd->u.lov;
2161         rc = lov_prep_statfs_set(obd, oinfo, &set);
2162         if (rc)
2163                 RETURN(rc);
2164
2165         list_for_each (pos, &set->set_list) {
2166                 struct obd_device *osc_obd;
2167
2168                 req = list_entry(pos, struct lov_request, rq_link);
2169
2170                 osc_obd = class_exp2obd(lov->lov_tgts[req->rq_idx]->ltd_exp);
2171                 rc = obd_statfs_async(osc_obd, &req->rq_oi, max_age, rqset);
2172                 if (rc)
2173                         break;
2174         }
2175
2176         if (rc || list_empty(&rqset->set_requests)) {
2177                 int err;
2178                 if (rc)
2179                         set->set_completes = 0;
2180                 err = lov_fini_statfs_set(set);
2181                 RETURN(rc ? rc : err);
2182         }
2183
2184         LASSERT(rqset->set_interpret == NULL);
2185         rqset->set_interpret = lov_statfs_interpret;
2186         rqset->set_arg = (void *)set;
2187         RETURN(0);
2188 }
2189
2190 static int lov_statfs(struct obd_device *obd, struct obd_statfs *osfs,
2191                       __u64 max_age)
2192 {
2193         struct lov_obd *lov = &obd->u.lov;
2194         struct obd_statfs lov_sfs;
2195         int set = 0;
2196         int rc = 0, err;
2197         int i;
2198         ENTRY;
2199
2200         /* We only get block data from the OBD */
2201         for (i = 0; i < lov->desc.ld_tgt_count; i++) {
2202                 if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_active) {
2203                         CDEBUG(D_HA, "lov idx %d inactive\n", i);
2204                         continue;
2205                 }
2206
2207                 err = obd_statfs(class_exp2obd(lov->lov_tgts[i]->ltd_exp),
2208                                  &lov_sfs, max_age);
2209                 if (err) {
2210                         if (lov->lov_tgts[i]->ltd_active && !rc)
2211                                 rc = err;
2212                         continue;
2213                 }
2214                 lov_update_statfs(class_exp2obd(lov->lov_tgts[i]->ltd_exp),
2215                                   osfs, &lov_sfs, set);
2216                 set++;
2217         }
2218
2219         err = lov_fini_statfs(obd, osfs, set);
2220         qos_update(lov);
2221
2222         RETURN(rc ? rc : err);
2223 }
2224
2225 static int lov_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
2226                          void *karg, void *uarg)
2227 {
2228         struct obd_device *obddev = class_exp2obd(exp);
2229         struct lov_obd *lov = &obddev->u.lov;
2230         int i, rc, count = lov->desc.ld_tgt_count;
2231         struct obd_uuid *uuidp;
2232         ENTRY;
2233
2234         switch (cmd) {
2235         case OBD_IOC_LOV_GET_CONFIG: {
2236                 struct obd_ioctl_data *data;
2237                 struct lov_desc *desc;
2238                 char *buf = NULL;
2239                 __u32 *genp;
2240
2241                 len = 0;
2242                 if (obd_ioctl_getdata(&buf, &len, (void *)uarg))
2243                         RETURN(-EINVAL);
2244
2245                 data = (struct obd_ioctl_data *)buf;
2246
2247                 if (sizeof(*desc) > data->ioc_inllen1) {
2248                         obd_ioctl_freedata(buf, len);
2249                         RETURN(-EINVAL);
2250                 }
2251
2252                 if (sizeof(uuidp->uuid) * count > data->ioc_inllen2) {
2253                         obd_ioctl_freedata(buf, len);
2254                         RETURN(-EINVAL);
2255                 }
2256
2257                 if (sizeof(__u32) * count > data->ioc_inllen3) {
2258                         obd_ioctl_freedata(buf, len);
2259                         RETURN(-EINVAL);
2260                 }
2261
2262                 desc = (struct lov_desc *)data->ioc_inlbuf1;
2263                 memcpy(desc, &(lov->desc), sizeof(*desc));
2264
2265                 uuidp = (struct obd_uuid *)data->ioc_inlbuf2;
2266                 genp = (__u32 *)data->ioc_inlbuf3;
2267                 /* the uuid will be empty for deleted OSTs */
2268                 for (i = 0; i < count; i++, uuidp++, genp++) {
2269                         if (!lov->lov_tgts[i])
2270                                 continue;
2271                         *uuidp = lov->lov_tgts[i]->ltd_uuid;
2272                         *genp = lov->lov_tgts[i]->ltd_gen;
2273                 }
2274
2275                 rc = copy_to_user((void *)uarg, buf, len);
2276                 if (rc)
2277                         rc = -EFAULT;
2278                 obd_ioctl_freedata(buf, len);
2279                 break;
2280         }
2281         case LL_IOC_LOV_SETSTRIPE:
2282                 rc = lov_setstripe(exp, karg, uarg);
2283                 break;
2284         case LL_IOC_LOV_GETSTRIPE:
2285                 rc = lov_getstripe(exp, karg, uarg);
2286                 break;
2287         case LL_IOC_LOV_SETEA:
2288                 rc = lov_setea(exp, karg, uarg);
2289                 break;
2290         default: {
2291                 int set = 0;
2292
2293                 if (count == 0)
2294                         RETURN(-ENOTTY);
2295
2296                 rc = 0;
2297                 for (i = 0; i < count; i++) {
2298                         int err;
2299
2300                         /* OST was disconnected */
2301                         if (!lov->lov_tgts[i] || !lov->lov_tgts[i]->ltd_exp)
2302                                 continue;
2303
2304                         err = obd_iocontrol(cmd, lov->lov_tgts[i]->ltd_exp,
2305                                             len, karg, uarg);
2306                         if (err == -ENODATA && cmd == OBD_IOC_POLL_QUOTACHECK) {
2307                                 RETURN(err);
2308                         } else if (err) {
2309                                 if (lov->lov_tgts[i]->ltd_active) {
2310                                         CDEBUG(err == -ENOTTY ?
2311                                                D_IOCTL : D_WARNING,
2312                                                "iocontrol OSC %s on OST "
2313                                                "idx %d cmd %x: err = %d\n",
2314                                                lov_uuid2str(lov, i),
2315                                                i, cmd, err);
2316                                         if (!rc)
2317                                                 rc = err;
2318                                 }
2319                         } else {
2320                                 set = 1;
2321                         }
2322                 }
2323                 if (!set && !rc)
2324                         rc = -EIO;
2325         }
2326         }
2327
2328         RETURN(rc);
2329 }
2330
2331 static int lov_get_info(struct obd_export *exp, __u32 keylen,
2332                         void *key, __u32 *vallen, void *val)
2333 {
2334         struct obd_device *obddev = class_exp2obd(exp);
2335         struct lov_obd *lov = &obddev->u.lov;
2336         int i, rc;
2337         ENTRY;
2338
2339         if (!vallen || !val)
2340                 RETURN(-EFAULT);
2341
2342         lov_getref(obddev);
2343
2344         if (KEY_IS("lock_to_stripe")) {
2345                 struct {
2346                         char name[16];
2347                         struct ldlm_lock *lock;
2348                         struct lov_stripe_md *lsm;
2349                 } *data = key;
2350                 struct ldlm_res_id *res_id = &data->lock->l_resource->lr_name;
2351                 struct lov_oinfo *loi;
2352                 __u32 *stripe = val;
2353
2354                 if (*vallen < sizeof(*stripe))
2355                         GOTO(out, rc = -EFAULT);
2356                 *vallen = sizeof(*stripe);
2357
2358                 /* XXX This is another one of those bits that will need to
2359                  * change if we ever actually support nested LOVs.  It uses
2360                  * the lock's export to find out which stripe it is. */
2361                 /* XXX - it's assumed all the locks for deleted OSTs have
2362                  * been cancelled. Also, the export for deleted OSTs will
2363                  * be NULL and won't match the lock's export. */
2364                 for (i = 0; i < data->lsm->lsm_stripe_count; i++) {
2365                         loi = data->lsm->lsm_oinfo[i];
2366                         if (!lov->lov_tgts[loi->loi_ost_idx])
2367                                 continue;
2368                         if (lov->lov_tgts[loi->loi_ost_idx]->ltd_exp ==
2369                             data->lock->l_conn_export &&
2370                             loi->loi_id == res_id->name[0] &&
2371                             loi->loi_gr == res_id->name[2]) {
2372                                 *stripe = i;
2373                                 GOTO(out, rc = 0);
2374                         }
2375                 }
2376                 LDLM_ERROR(data->lock, "lock on inode without such object");
2377                 dump_lsm(D_ERROR, data->lsm);
2378                 GOTO(out, rc = -ENXIO);
2379         } else if (KEY_IS(KEY_LAST_ID)) {
2380                 struct obd_id_info *info = val;
2381                 int size = sizeof(obd_id);
2382                 struct lov_tgt_desc *tgt;
2383
2384                 LASSERT(*vallen == sizeof(struct obd_id_info));
2385                 tgt = lov->lov_tgts[info->idx];
2386
2387                 if (!tgt || !tgt->ltd_active)
2388                         GOTO(out, rc = -ESRCH);
2389
2390                 rc = obd_get_info(tgt->ltd_exp, keylen, key, &size, info->data);
2391                 GOTO(out, rc = 0);
2392         } else if (KEY_IS(KEY_LOVDESC)) {
2393                 struct lov_desc *desc_ret = val;
2394                 *desc_ret = lov->desc;
2395
2396                 GOTO(out, rc = 0);
2397         } else if (KEY_IS(KEY_LOV_IDX)) {
2398                 struct lov_tgt_desc *tgt;
2399
2400                 for(i = 0; i < lov->desc.ld_tgt_count; i++) {
2401                         tgt = lov->lov_tgts[i];
2402                         if (tgt && obd_uuid_equals(val, &tgt->ltd_uuid))
2403                                 GOTO(out, rc = i);
2404                 }
2405         }
2406
2407         rc = -EINVAL;
2408 out:
2409         lov_putref(obddev);
2410         RETURN(rc);
2411 }
2412
2413 static int lov_set_info_async(struct obd_export *exp, obd_count keylen,
2414                               void *key, obd_count vallen, void *val,
2415                               struct ptlrpc_request_set *set)
2416 {
2417         struct obd_device *obddev = class_exp2obd(exp);
2418         struct lov_obd *lov = &obddev->u.lov;
2419         obd_count count;
2420         int i, rc = 0, err;
2421         struct lov_tgt_desc *tgt;
2422         unsigned incr, check_uuid,
2423                  do_inactive, no_set;
2424         unsigned next_id = 0,  mds_con = 0;
2425         ENTRY;
2426
2427         incr = check_uuid = do_inactive = no_set = 0;
2428         if (set == NULL) {
2429                 no_set = 1;
2430                 set = ptlrpc_prep_set();
2431                 if (!set)
2432                         RETURN(-ENOMEM);
2433         }
2434
2435         lov_getref(obddev);
2436         count = lov->desc.ld_tgt_count;
2437
2438         if (KEY_IS(KEY_NEXT_ID)) {
2439                 count = vallen / sizeof(struct obd_id_info);
2440                 vallen = sizeof(obd_id);
2441                 incr = sizeof(struct obd_id_info);
2442                 do_inactive = 1;
2443                 next_id = 1;
2444         } else if (KEY_IS("checksum")) {
2445                 do_inactive = 1;
2446         } else if (KEY_IS("unlinked")) {
2447                 check_uuid = val ? 1 : 0;
2448         } else if (KEY_IS("evict_by_nid")) {
2449                 /* use defaults:  do_inactive = incr = 0; */
2450         } else if (KEY_IS(KEY_MDS_CONN)) {
2451                 mds_con = 1;
2452         }
2453
2454         for (i = 0; i < count; i++, val = (char *)val + incr) {
2455                 if (next_id) {
2456                         tgt = lov->lov_tgts[((struct obd_id_info*)val)->idx];
2457                 } else {
2458                         tgt = lov->lov_tgts[i];
2459                 }
2460                 /* OST was disconnected */
2461                 if (!tgt || !tgt->ltd_exp)
2462                         continue;
2463
2464                 /* OST is inactive and we don't want inactive OSCs */
2465                 if (!tgt->ltd_active && !do_inactive)
2466                         continue;
2467
2468                 if (mds_con) {
2469                         struct mds_group_info *mgi;
2470
2471                         LASSERT(vallen == sizeof(*mgi));
2472                         mgi = (struct mds_group_info *)val;
2473
2474                         /* Only want a specific OSC */
2475                         if (mgi->uuid && !obd_uuid_equals(mgi->uuid,
2476                                                 &tgt->ltd_uuid))
2477                                 continue;
2478
2479                         err = obd_set_info_async(tgt->ltd_exp,
2480                                          keylen, key, sizeof(int),
2481                                          &mgi->group, set);
2482                 } else if (next_id) {
2483                         err = obd_set_info_async(tgt->ltd_exp,
2484                                          keylen, key, vallen,
2485                                          ((struct obd_id_info*)val)->data, set);
2486                 } else  {
2487                         /* Only want a specific OSC */
2488                         if (check_uuid &&
2489                             !obd_uuid_equals(val, &tgt->ltd_uuid))
2490                                 continue;
2491
2492                         err = obd_set_info_async(tgt->ltd_exp,
2493                                          keylen, key, vallen, val, set);
2494                 }
2495
2496                 if (!rc)
2497                         rc = err;
2498         }
2499
2500         lov_putref(obddev);
2501         if (no_set) {
2502                 err = ptlrpc_set_wait(set);
2503                 if (!rc)
2504                         rc = err;
2505                 ptlrpc_set_destroy(set);
2506         }
2507         RETURN(rc);
2508 }
2509
2510 static int lov_checkmd(struct obd_export *exp, struct obd_export *md_exp,
2511                        struct lov_stripe_md *lsm)
2512 {
2513         int rc;
2514         ENTRY;
2515
2516         if (!lsm)
2517                 RETURN(0);
2518         LASSERT(md_exp);
2519         LASSERT(lsm_op_find(lsm->lsm_magic) != NULL);
2520         rc = lsm_op_find(lsm->lsm_magic)->lsm_revalidate(lsm, md_exp->exp_obd);
2521
2522         RETURN(rc);
2523 }
2524
2525 int lov_test_and_clear_async_rc(struct lov_stripe_md *lsm)
2526 {
2527         int i, rc = 0;
2528         ENTRY;
2529
2530         for (i = 0; i < lsm->lsm_stripe_count; i++) {
2531                 struct lov_oinfo *loi = lsm->lsm_oinfo[i];
2532                 if (loi->loi_ar.ar_rc && !rc)
2533                         rc = loi->loi_ar.ar_rc;
2534                 loi->loi_ar.ar_rc = 0;
2535         }
2536         RETURN(rc);
2537 }
2538 EXPORT_SYMBOL(lov_test_and_clear_async_rc);
2539
2540
2541 static int lov_extent_calc(struct obd_export *exp, struct lov_stripe_md *lsm,
2542                            int cmd, __u64 *offset)
2543 {
2544         __u32 ssize = lsm->lsm_stripe_size;
2545         __u64 start;
2546
2547         start = *offset;
2548         do_div(start, ssize);
2549         start = start * ssize;
2550
2551         CDEBUG(D_DLMTRACE, "offset "LPU64", stripe %u, start "LPU64
2552                            ", end "LPU64"\n", *offset, ssize, start,
2553                            start + ssize - 1);
2554         if (cmd == OBD_CALC_STRIPE_END) {
2555                 *offset = start + ssize - 1;
2556         } else if (cmd == OBD_CALC_STRIPE_START) {
2557                 *offset = start;
2558         } else {
2559                 LBUG();
2560         }
2561
2562         RETURN(0);
2563 }
2564
2565
2566 #if 0
2567 struct lov_multi_wait {
2568         struct ldlm_lock *lock;
2569         wait_queue_t      wait;
2570         int               completed;
2571         int               generation;
2572 };
2573
2574 int lov_complete_many(struct obd_export *exp, struct lov_stripe_md *lsm,
2575                       struct lustre_handle *lockh)
2576 {
2577         struct lov_lock_handles *lov_lockh = NULL;
2578         struct lustre_handle *lov_lockhp;
2579         struct lov_obd *lov;
2580         struct lov_oinfo *loi;
2581         struct lov_multi_wait *queues;
2582         int rc = 0, i;
2583         ENTRY;
2584
2585         ASSERT_LSM_MAGIC(lsm);
2586
2587         if (!exp || !exp->exp_obd)
2588                 RETURN(-ENODEV);
2589
2590         LASSERT(lockh != NULL);
2591         if (lsm->lsm_stripe_count > 1) {
2592                 lov_lockh = lov_handle2llh(lockh);
2593                 if (lov_lockh == NULL) {
2594                         CERROR("LOV: invalid lov lock handle %p\n", lockh);
2595                         RETURN(-EINVAL);
2596                 }
2597
2598                 lov_lockhp = lov_lockh->llh_handles;
2599         } else {
2600                 lov_lockhp = lockh;
2601         }
2602
2603         OBD_ALLOC(queues, lsm->lsm_stripe_count * sizeof(*queues));
2604         if (queues == NULL)
2605                 GOTO(out, rc = -ENOMEM);
2606
2607         lov = &exp->exp_obd->u.lov;
2608         for (i = 0, loi = lsm->lsm_oinfo; i < lsm->lsm_stripe_count;
2609              i++, loi++, lov_lockhp++) {
2610                 struct ldlm_lock *lock;
2611                 struct obd_device *obd;
2612
2613                 lock = ldlm_handle2lock(lov_lockhp);
2614                 if (lock == NULL) {
2615                         CDEBUG(D_HA, "lov idx %d subobj "LPX64" no lock?\n",
2616                                loi->loi_ost_idx, loi->loi_id);
2617                         queues[i].completed = 1;
2618                         continue;
2619                 }
2620
2621                 queues[i].lock = lock;
2622                 init_waitqueue_entry(&(queues[i].wait), current);
2623                 add_wait_queue(lock->l_waitq, &(queues[i].wait));
2624
2625                 obd = class_exp2obd(lock->l_conn_export);
2626                 if (obd != NULL)
2627                         imp = obd->u.cli.cl_import;
2628                 if (imp != NULL) {
2629                         spin_lock(&imp->imp_lock);
2630                         queues[i].generation = imp->imp_generation;
2631                         spin_unlock(&imp->imp_lock);
2632                 }
2633         }
2634
2635         lwi = LWI_TIMEOUT_INTR(obd_timeout * HZ, ldlm_expired_completion_wait,
2636                                interrupted_completion_wait, &lwd);
2637         rc = l_wait_event_added(check_multi_complete(queues, lsm), &lwi);
2638
2639         for (i = 0; i < lsm->lsm_stripe_count; i++)
2640                 remove_wait_queue(lock->l_waitq, &(queues[i].wait));
2641
2642         if (rc == -EINTR || rc == -ETIMEDOUT) {
2643
2644
2645         }
2646
2647  out:
2648         if (lov_lockh != NULL)
2649                 lov_llh_put(lov_lockh);
2650         RETURN(rc);
2651 }
2652 #endif
2653
2654 void lov_stripe_lock(struct lov_stripe_md *md)
2655 {
2656         LASSERT(md->lsm_lock_owner != cfs_current());
2657         spin_lock(&md->lsm_lock);
2658         LASSERT(md->lsm_lock_owner == NULL);
2659         md->lsm_lock_owner = cfs_current();
2660 }
2661 EXPORT_SYMBOL(lov_stripe_lock);
2662
2663 void lov_stripe_unlock(struct lov_stripe_md *md)
2664 {
2665         LASSERT(md->lsm_lock_owner == cfs_current());
2666         md->lsm_lock_owner = NULL;
2667         spin_unlock(&md->lsm_lock);
2668 }
2669 EXPORT_SYMBOL(lov_stripe_unlock);
2670
2671 struct obd_ops lov_obd_ops = {
2672         .o_owner               = THIS_MODULE,
2673         .o_setup               = lov_setup,
2674         .o_precleanup          = lov_precleanup,
2675         .o_cleanup             = lov_cleanup,
2676         .o_process_config      = lov_process_config,
2677         .o_connect             = lov_connect,
2678         .o_disconnect          = lov_disconnect,
2679         .o_statfs              = lov_statfs,
2680         .o_statfs_async        = lov_statfs_async,
2681         .o_packmd              = lov_packmd,
2682         .o_unpackmd            = lov_unpackmd,
2683         .o_checkmd             = lov_checkmd,
2684         .o_create              = lov_create,
2685         .o_destroy             = lov_destroy,
2686         .o_getattr             = lov_getattr,
2687         .o_getattr_async       = lov_getattr_async,
2688         .o_setattr             = lov_setattr,
2689         .o_setattr_async       = lov_setattr_async,
2690         .o_brw                 = lov_brw,
2691         .o_brw_async           = lov_brw_async,
2692         .o_prep_async_page     = lov_prep_async_page,
2693         .o_queue_async_io      = lov_queue_async_io,
2694         .o_set_async_flags     = lov_set_async_flags,
2695         .o_queue_group_io      = lov_queue_group_io,
2696         .o_trigger_group_io    = lov_trigger_group_io,
2697         .o_teardown_async_page = lov_teardown_async_page,
2698         .o_merge_lvb           = lov_merge_lvb,
2699         .o_adjust_kms          = lov_adjust_kms,
2700         .o_punch               = lov_punch,
2701         .o_sync                = lov_sync,
2702         .o_enqueue             = lov_enqueue,
2703         .o_match               = lov_match,
2704         .o_change_cbdata       = lov_change_cbdata,
2705         .o_cancel              = lov_cancel,
2706         .o_cancel_unused       = lov_cancel_unused,
2707         .o_join_lru            = lov_join_lru,
2708         .o_iocontrol           = lov_iocontrol,
2709         .o_get_info            = lov_get_info,
2710         .o_set_info_async      = lov_set_info_async,
2711         .o_extent_calc         = lov_extent_calc,
2712         .o_llog_init           = lov_llog_init,
2713         .o_llog_finish         = lov_llog_finish,
2714         .o_notify              = lov_notify,
2715 };
2716
2717 static quota_interface_t *quota_interface;
2718 extern quota_interface_t lov_quota_interface;
2719
2720 cfs_mem_cache_t *lov_oinfo_slab;
2721
2722 int __init lov_init(void)
2723 {
2724         struct lprocfs_static_vars lvars = { 0 };
2725         int rc, rc2;
2726         ENTRY;
2727
2728         lov_oinfo_slab = cfs_mem_cache_create("lov_oinfo",
2729                                               sizeof(struct lov_oinfo),
2730                                               0, SLAB_HWCACHE_ALIGN);
2731         if (lov_oinfo_slab == NULL)
2732                 return -ENOMEM;
2733         lprocfs_lov_init_vars(&lvars);
2734
2735         request_module("lquota");
2736         quota_interface = PORTAL_SYMBOL_GET(lov_quota_interface);
2737         init_obd_quota_ops(quota_interface, &lov_obd_ops);
2738
2739         rc = class_register_type(&lov_obd_ops, NULL, lvars.module_vars,
2740                                  LUSTRE_LOV_NAME, NULL);
2741         if (rc) {
2742                 if (quota_interface)
2743                         PORTAL_SYMBOL_PUT(lov_quota_interface);
2744                 rc2 = cfs_mem_cache_destroy(lov_oinfo_slab);
2745                 LASSERT(rc2 == 0);
2746         }
2747
2748         RETURN(rc);
2749 }
2750
2751 #ifdef __KERNEL__
2752 static void /*__exit*/ lov_exit(void)
2753 {
2754         int rc;
2755
2756         if (quota_interface)
2757                 PORTAL_SYMBOL_PUT(lov_quota_interface);
2758
2759         class_unregister_type(LUSTRE_LOV_NAME);
2760         rc = cfs_mem_cache_destroy(lov_oinfo_slab);
2761         LASSERT(rc == 0);
2762 }
2763
2764 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2765 MODULE_DESCRIPTION("Lustre Logical Object Volume OBD driver");
2766 MODULE_LICENSE("GPL");
2767
2768 cfs_module(lov, LUSTRE_VERSION_STRING, lov_init, lov_exit);
2769 #endif