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