lustre/lmv/lmv_obd.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2013, Intel Corporation.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #define DEBUG_SUBSYSTEM S_LMV
  38 #include <linux/slab.h>
  39 #include <linux/module.h>
  40 #include <linux/init.h>
  41 #include <linux/slab.h>
  42 #include <linux/pagemap.h>
  43 #include <linux/mm.h>
  44 #include <linux/math64.h>
  45 #include <linux/seq_file.h>
  46 #include <linux/namei.h>
  47
  48 #include <lustre/lustre_idl.h>
  49 #include <obd_support.h>
  50 #include <lustre_lib.h>
  51 #include <lustre_net.h>
  52 #include <obd_class.h>
  53 #include <lustre_lmv.h>
  54 #include <lprocfs_status.h>
  55 #include <cl_object.h>
  56 #include <lclient.h>
  57 #include <lustre_fid.h>
  58 #include <lustre_ioctl.h>
  59 #include "lmv_internal.h"
  60
  61 /* This hash is only for testing purpose */
  62 static inline unsigned int
  63 lmv_hash_all_chars(unsigned int count, const char *name, int namelen)
  64 {
  65         unsigned int c = 0;
  66         const unsigned char *p = (const unsigned char *)name;
  67
  68         while (--namelen >= 0)
  69                 c += p[namelen];
  70
  71         c = c % count;
  72
  73         return c;
  74 }
  75
  76 static inline unsigned int
  77 lmv_hash_fnv1a(unsigned int count, const char *name, int namelen)
  78 {
  79         __u64   hash;
  80
  81         hash = lustre_hash_fnv_1a_64(name, namelen);
  82
  83         hash = hash % count;
  84
  85         return hash;
  86 }
  87
  88 int lmv_name_to_stripe_index(__u32 lmv_hash_type, unsigned int stripe_count,
  89                              const char *name, int namelen)
  90 {
  91         int     idx;
  92         __u32   hash_type = lmv_hash_type & LMV_HASH_TYPE_MASK;
  93
  94         LASSERT(namelen > 0);
  95         if (stripe_count <= 1)
  96                 return 0;
  97
  98         /* for migrating object, always start from 0 stripe */
  99         if (lmv_hash_type & LMV_HASH_FLAG_MIGRATION)
 100                 return 0;
 101
 102         switch (hash_type) {
 103         case LMV_HASH_TYPE_ALL_CHARS:
 104                 idx = lmv_hash_all_chars(stripe_count, name, namelen);
 105                 break;
 106         case LMV_HASH_TYPE_FNV_1A_64:
 107                 idx = lmv_hash_fnv1a(stripe_count, name, namelen);
 108                 break;
 109         default:
 110                 idx = -EBADFD;
 111                 break;
 112         }
 113
 114         CDEBUG(D_INFO, "name %.*s hash_type %d idx %d\n", namelen, name,
 115                hash_type, idx);
 116
 117         return idx;
 118 }
 119
 120 static void lmv_activate_target(struct lmv_obd *lmv,
 121                                 struct lmv_tgt_desc *tgt,
 122                                 int activate)
 123 {
 124         if (tgt->ltd_active == activate)
 125                 return;
 126
 127         tgt->ltd_active = activate;
 128         lmv->desc.ld_active_tgt_count += (activate ? 1 : -1);
 129 }
 130
 131 /**
 132  * Error codes:
 133  *
 134  *  -EINVAL  : UUID can't be found in the LMV's target list
 135  *  -ENOTCONN: The UUID is found, but the target connection is bad (!)
 136  *  -EBADF   : The UUID is found, but the OBD of the wrong type (!)
 137  */
 138 static int lmv_set_mdc_active(struct lmv_obd *lmv,
 139                               const struct obd_uuid *uuid,
 140                               int activate)
 141 {
 142         struct lmv_tgt_desc     *tgt = NULL;
 143         struct obd_device       *obd;
 144         __u32                    i;
 145         int                      rc = 0;
 146         ENTRY;
 147
 148         CDEBUG(D_INFO, "Searching in lmv %p for uuid %s (activate=%d)\n",
 149                         lmv, uuid->uuid, activate);
 150
 151         spin_lock(&lmv->lmv_lock);
 152         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 153                 tgt = lmv->tgts[i];
 154                 if (tgt == NULL || tgt->ltd_exp == NULL)
 155                         continue;
 156
 157                 CDEBUG(D_INFO, "Target idx %d is %s conn "LPX64"\n", i,
 158                        tgt->ltd_uuid.uuid, tgt->ltd_exp->exp_handle.h_cookie);
 159
 160                 if (obd_uuid_equals(uuid, &tgt->ltd_uuid))
 161                         break;
 162         }
 163
 164         if (i == lmv->desc.ld_tgt_count)
 165                 GOTO(out_lmv_lock, rc = -EINVAL);
 166
 167         obd = class_exp2obd(tgt->ltd_exp);
 168         if (obd == NULL)
 169                 GOTO(out_lmv_lock, rc = -ENOTCONN);
 170
 171         CDEBUG(D_INFO, "Found OBD %s=%s device %d (%p) type %s at LMV idx %d\n",
 172                obd->obd_name, obd->obd_uuid.uuid, obd->obd_minor, obd,
 173                obd->obd_type->typ_name, i);
 174         LASSERT(strcmp(obd->obd_type->typ_name, LUSTRE_MDC_NAME) == 0);
 175
 176         if (tgt->ltd_active == activate) {
 177                 CDEBUG(D_INFO, "OBD %p already %sactive!\n", obd,
 178                        activate ? "" : "in");
 179                 GOTO(out_lmv_lock, rc);
 180         }
 181
 182         CDEBUG(D_INFO, "Marking OBD %p %sactive\n", obd,
 183                activate ? "" : "in");
 184         lmv_activate_target(lmv, tgt, activate);
 185         EXIT;
 186
 187  out_lmv_lock:
 188         spin_unlock(&lmv->lmv_lock);
 189         return rc;
 190 }
 191
 192 struct obd_uuid *lmv_get_uuid(struct obd_export *exp)
 193 {
 194         struct lmv_obd          *lmv = &exp->exp_obd->u.lmv;
 195         struct lmv_tgt_desc     *tgt = lmv->tgts[0];
 196
 197         return (tgt == NULL) ? NULL : obd_get_uuid(tgt->ltd_exp);
 198 }
 199
 200 static int lmv_notify(struct obd_device *obd, struct obd_device *watched,
 201                       enum obd_notify_event ev, void *data)
 202 {
 203         struct obd_connect_data *conn_data;
 204         struct lmv_obd          *lmv = &obd->u.lmv;
 205         struct obd_uuid         *uuid;
 206         int                      rc = 0;
 207         ENTRY;
 208
 209         if (strcmp(watched->obd_type->typ_name, LUSTRE_MDC_NAME)) {
 210                 CERROR("unexpected notification of %s %s!\n",
 211                        watched->obd_type->typ_name,
 212                        watched->obd_name);
 213                 RETURN(-EINVAL);
 214         }
 215
 216         uuid = &watched->u.cli.cl_target_uuid;
 217         if (ev == OBD_NOTIFY_ACTIVE || ev == OBD_NOTIFY_INACTIVE) {
 218                 /*
 219                  * Set MDC as active before notifying the observer, so the
 220                  * observer can use the MDC normally.
 221                  */
 222                 rc = lmv_set_mdc_active(lmv, uuid,
 223                                         ev == OBD_NOTIFY_ACTIVE);
 224                 if (rc) {
 225                         CERROR("%sactivation of %s failed: %d\n",
 226                                ev == OBD_NOTIFY_ACTIVE ? "" : "de",
 227                                uuid->uuid, rc);
 228                         RETURN(rc);
 229                 }
 230         } else if (ev == OBD_NOTIFY_OCD) {
 231                 conn_data = &watched->u.cli.cl_import->imp_connect_data;
 232                 /*
 233                  * XXX: Make sure that ocd_connect_flags from all targets are
 234                  * the same. Otherwise one of MDTs runs wrong version or
 235                  * something like this.  --umka
 236                  */
 237                 obd->obd_self_export->exp_connect_data = *conn_data;
 238         }
 239 #if 0
 240         else if (ev == OBD_NOTIFY_DISCON) {
 241                 /*
 242                  * For disconnect event, flush fld cache for failout MDS case.
 243                  */
 244                 fld_client_flush(&lmv->lmv_fld);
 245         }
 246 #endif
 247         /*
 248          * Pass the notification up the chain.
 249          */
 250         if (obd->obd_observer)
 251                 rc = obd_notify(obd->obd_observer, watched, ev, data);
 252
 253         RETURN(rc);
 254 }
 255
 256 /**
 257  * This is fake connect function. Its purpose is to initialize lmv and say
 258  * caller that everything is okay. Real connection will be performed later.
 259  */
 260 static int lmv_connect(const struct lu_env *env,
 261                        struct obd_export **exp, struct obd_device *obd,
 262                        struct obd_uuid *cluuid, struct obd_connect_data *data,
 263                        void *localdata)
 264 {
 265         struct lmv_obd        *lmv = &obd->u.lmv;
 266         struct lustre_handle  conn = { 0 };
 267         int                    rc = 0;
 268         ENTRY;
 269
 270         /*
 271          * We don't want to actually do the underlying connections more than
 272          * once, so keep track.
 273          */
 274         lmv->refcount++;
 275         if (lmv->refcount > 1) {
 276                 *exp = NULL;
 277                 RETURN(0);
 278         }
 279
 280         rc = class_connect(&conn, obd, cluuid);
 281         if (rc) {
 282                 CERROR("class_connection() returned %d\n", rc);
 283                 RETURN(rc);
 284         }
 285
 286         *exp = class_conn2export(&conn);
 287         class_export_get(*exp);
 288
 289         lmv->exp = *exp;
 290         lmv->connected = 0;
 291         lmv->cluuid = *cluuid;
 292
 293         if (data)
 294                 lmv->conn_data = *data;
 295
 296         if (lmv->targets_proc_entry == NULL) {
 297                 lmv->targets_proc_entry = lprocfs_seq_register("target_obds",
 298                                                         obd->obd_proc_entry,
 299                                                         NULL, NULL);
 300                 if (IS_ERR(lmv->targets_proc_entry)) {
 301                         CERROR("%s: cannot register "
 302                                "/proc/fs/lustre/%s/%s/target_obds\n",
 303                                obd->obd_name, obd->obd_type->typ_name,
 304                                obd->obd_name);
 305                         lmv->targets_proc_entry = NULL;
 306                 }
 307         }
 308
 309         /*
 310          * All real clients should perform actual connection right away, because
 311          * it is possible, that LMV will not have opportunity to connect targets
 312          * and MDC stuff will be called directly, for instance while reading
 313          * ../mdc/../kbytesfree procfs file, etc.
 314          */
 315         if (data != NULL && (data->ocd_connect_flags & OBD_CONNECT_REAL))
 316                 rc = lmv_check_connect(obd);
 317
 318         if (rc && lmv->targets_proc_entry != NULL)
 319                 lprocfs_remove(&lmv->targets_proc_entry);
 320         RETURN(rc);
 321 }
 322
 323 static int lmv_init_ea_size(struct obd_export *exp, int easize,
 324                             int def_easize, int cookiesize, int def_cookiesize)
 325 {
 326         struct obd_device       *obd = exp->exp_obd;
 327         struct lmv_obd          *lmv = &obd->u.lmv;
 328         __u32                    i;
 329         int                      rc = 0;
 330         int                      change = 0;
 331         ENTRY;
 332
 333         if (lmv->max_easize < easize) {
 334                 lmv->max_easize = easize;
 335                 change = 1;
 336         }
 337         if (lmv->max_def_easize < def_easize) {
 338                 lmv->max_def_easize = def_easize;
 339                 change = 1;
 340         }
 341         if (lmv->max_cookiesize < cookiesize) {
 342                 lmv->max_cookiesize = cookiesize;
 343                 change = 1;
 344         }
 345         if (lmv->max_def_cookiesize < def_cookiesize) {
 346                 lmv->max_def_cookiesize = def_cookiesize;
 347                 change = 1;
 348         }
 349         if (change == 0)
 350                 RETURN(0);
 351
 352         if (lmv->connected == 0)
 353                 RETURN(0);
 354
 355         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 356                 struct lmv_tgt_desc *tgt = lmv->tgts[i];
 357
 358                 if (tgt == NULL || tgt->ltd_exp == NULL || !tgt->ltd_active) {
 359                         CWARN("%s: NULL export for %d\n", obd->obd_name, i);
 360                         continue;
 361                 }
 362
 363                 rc = md_init_ea_size(tgt->ltd_exp, easize, def_easize,
 364                                      cookiesize, def_cookiesize);
 365                 if (rc) {
 366                         CERROR("%s: obd_init_ea_size() failed on MDT target %d:"
 367                                " rc = %d\n", obd->obd_name, i, rc);
 368                         break;
 369                 }
 370         }
 371         RETURN(rc);
 372 }
 373
 374 #define MAX_STRING_SIZE 128
 375
 376 int lmv_connect_mdc(struct obd_device *obd, struct lmv_tgt_desc *tgt)
 377 {
 378         struct lmv_obd          *lmv = &obd->u.lmv;
 379         struct obd_uuid         *cluuid = &lmv->cluuid;
 380         struct obd_uuid          lmv_mdc_uuid = { "LMV_MDC_UUID" };
 381         struct obd_device       *mdc_obd;
 382         struct obd_export       *mdc_exp;
 383         struct lu_fld_target     target;
 384         int                      rc;
 385         ENTRY;
 386
 387         mdc_obd = class_find_client_obd(&tgt->ltd_uuid, LUSTRE_MDC_NAME,
 388                                         &obd->obd_uuid);
 389         if (!mdc_obd) {
 390                 CERROR("target %s not attached\n", tgt->ltd_uuid.uuid);
 391                 RETURN(-EINVAL);
 392         }
 393
 394         CDEBUG(D_CONFIG, "connect to %s(%s) - %s, %s FOR %s\n",
 395                 mdc_obd->obd_name, mdc_obd->obd_uuid.uuid,
 396                 tgt->ltd_uuid.uuid, obd->obd_uuid.uuid,
 397                 cluuid->uuid);
 398
 399         if (!mdc_obd->obd_set_up) {
 400                 CERROR("target %s is not set up\n", tgt->ltd_uuid.uuid);
 401                 RETURN(-EINVAL);
 402         }
 403
 404         rc = obd_connect(NULL, &mdc_exp, mdc_obd, &lmv_mdc_uuid,
 405                          &lmv->conn_data, NULL);
 406         if (rc) {
 407                 CERROR("target %s connect error %d\n", tgt->ltd_uuid.uuid, rc);
 408                 RETURN(rc);
 409         }
 410
 411         /*
 412          * Init fid sequence client for this mdc and add new fld target.
 413          */
 414         rc = obd_fid_init(mdc_obd, mdc_exp, LUSTRE_SEQ_METADATA);
 415         if (rc)
 416                 RETURN(rc);
 417
 418         target.ft_srv = NULL;
 419         target.ft_exp = mdc_exp;
 420         target.ft_idx = tgt->ltd_idx;
 421
 422         fld_client_add_target(&lmv->lmv_fld, &target);
 423
 424         rc = obd_register_observer(mdc_obd, obd);
 425         if (rc) {
 426                 obd_disconnect(mdc_exp);
 427                 CERROR("target %s register_observer error %d\n",
 428                        tgt->ltd_uuid.uuid, rc);
 429                 RETURN(rc);
 430         }
 431
 432         if (obd->obd_observer) {
 433                 /*
 434                  * Tell the observer about the new target.
 435                  */
 436                 rc = obd_notify(obd->obd_observer, mdc_exp->exp_obd,
 437                                 OBD_NOTIFY_ACTIVE,
 438                                 (void *)(tgt - lmv->tgts[0]));
 439                 if (rc) {
 440                         obd_disconnect(mdc_exp);
 441                         RETURN(rc);
 442                 }
 443         }
 444
 445         tgt->ltd_active = 1;
 446         tgt->ltd_exp = mdc_exp;
 447         lmv->desc.ld_active_tgt_count++;
 448
 449         md_init_ea_size(tgt->ltd_exp, lmv->max_easize, lmv->max_def_easize,
 450                         lmv->max_cookiesize, lmv->max_def_cookiesize);
 451
 452         CDEBUG(D_CONFIG, "Connected to %s(%s) successfully (%d)\n",
 453                 mdc_obd->obd_name, mdc_obd->obd_uuid.uuid,
 454                 atomic_read(&obd->obd_refcount));
 455
 456         if (lmv->targets_proc_entry != NULL) {
 457                 struct proc_dir_entry *mdc_symlink;
 458
 459                 LASSERT(mdc_obd->obd_type != NULL);
 460                 LASSERT(mdc_obd->obd_type->typ_name != NULL);
 461                 mdc_symlink = lprocfs_add_symlink(mdc_obd->obd_name,
 462                                                   lmv->targets_proc_entry,
 463                                                   "../../../%s/%s",
 464                                                   mdc_obd->obd_type->typ_name,
 465                                                   mdc_obd->obd_name);
 466                 if (mdc_symlink == NULL) {
 467                         CERROR("cannot register LMV target "
 468                                "/proc/fs/lustre/%s/%s/target_obds/%s\n",
 469                                obd->obd_type->typ_name, obd->obd_name,
 470                                mdc_obd->obd_name);
 471                 }
 472         }
 473         RETURN(0);
 474 }
 475
 476 static void lmv_del_target(struct lmv_obd *lmv, int index)
 477 {
 478         if (lmv->tgts[index] == NULL)
 479                 return;
 480
 481         OBD_FREE_PTR(lmv->tgts[index]);
 482         lmv->tgts[index] = NULL;
 483         return;
 484 }
 485
 486 static int lmv_add_target(struct obd_device *obd, struct obd_uuid *uuidp,
 487                            __u32 index, int gen)
 488 {
 489         struct lmv_obd      *lmv = &obd->u.lmv;
 490         struct lmv_tgt_desc *tgt;
 491         int                  orig_tgt_count = 0;
 492         int                  rc = 0;
 493         ENTRY;
 494
 495         CDEBUG(D_CONFIG, "Target uuid: %s. index %d\n", uuidp->uuid, index);
 496
 497         lmv_init_lock(lmv);
 498
 499         if (lmv->desc.ld_tgt_count == 0) {
 500                 struct obd_device *mdc_obd;
 501
 502                 mdc_obd = class_find_client_obd(uuidp, LUSTRE_MDC_NAME,
 503                                                 &obd->obd_uuid);
 504                 if (!mdc_obd) {
 505                         lmv_init_unlock(lmv);
 506                         CERROR("%s: Target %s not attached: rc = %d\n",
 507                                obd->obd_name, uuidp->uuid, -EINVAL);
 508                         RETURN(-EINVAL);
 509                 }
 510         }
 511
 512         if ((index < lmv->tgts_size) && (lmv->tgts[index] != NULL)) {
 513                 tgt = lmv->tgts[index];
 514                 CERROR("%s: UUID %s already assigned at LOV target index %d:"
 515                        " rc = %d\n", obd->obd_name,
 516                        obd_uuid2str(&tgt->ltd_uuid), index, -EEXIST);
 517                 lmv_init_unlock(lmv);
 518                 RETURN(-EEXIST);
 519         }
 520
 521         if (index >= lmv->tgts_size) {
 522                 /* We need to reallocate the lmv target array. */
 523                 struct lmv_tgt_desc **newtgts, **old = NULL;
 524                 __u32 newsize = 1;
 525                 __u32 oldsize = 0;
 526
 527                 while (newsize < index + 1)
 528                         newsize = newsize << 1;
 529                 OBD_ALLOC(newtgts, sizeof(*newtgts) * newsize);
 530                 if (newtgts == NULL) {
 531                         lmv_init_unlock(lmv);
 532                         RETURN(-ENOMEM);
 533                 }
 534
 535                 if (lmv->tgts_size) {
 536                         memcpy(newtgts, lmv->tgts,
 537                                sizeof(*newtgts) * lmv->tgts_size);
 538                         old = lmv->tgts;
 539                         oldsize = lmv->tgts_size;
 540                 }
 541
 542                 lmv->tgts = newtgts;
 543                 lmv->tgts_size = newsize;
 544                 smp_rmb();
 545                 if (old)
 546                         OBD_FREE(old, sizeof(*old) * oldsize);
 547
 548                 CDEBUG(D_CONFIG, "tgts: %p size: %d\n", lmv->tgts,
 549                        lmv->tgts_size);
 550         }
 551
 552         OBD_ALLOC_PTR(tgt);
 553         if (!tgt) {
 554                 lmv_init_unlock(lmv);
 555                 RETURN(-ENOMEM);
 556         }
 557
 558         mutex_init(&tgt->ltd_fid_mutex);
 559         tgt->ltd_idx = index;
 560         tgt->ltd_uuid = *uuidp;
 561         tgt->ltd_active = 0;
 562         lmv->tgts[index] = tgt;
 563         if (index >= lmv->desc.ld_tgt_count) {
 564                 orig_tgt_count = lmv->desc.ld_tgt_count;
 565                 lmv->desc.ld_tgt_count = index + 1;
 566         }
 567
 568         if (lmv->connected) {
 569                 rc = lmv_connect_mdc(obd, tgt);
 570                 if (rc != 0) {
 571                         spin_lock(&lmv->lmv_lock);
 572                         if (lmv->desc.ld_tgt_count == index + 1)
 573                                 lmv->desc.ld_tgt_count = orig_tgt_count;
 574                         memset(tgt, 0, sizeof(*tgt));
 575                         spin_unlock(&lmv->lmv_lock);
 576                 } else {
 577                         int easize = sizeof(struct lmv_stripe_md) +
 578                                 lmv->desc.ld_tgt_count * sizeof(struct lu_fid);
 579                         lmv_init_ea_size(obd->obd_self_export, easize, 0, 0, 0);
 580                 }
 581         }
 582
 583         lmv_init_unlock(lmv);
 584         RETURN(rc);
 585 }
 586
 587 int lmv_check_connect(struct obd_device *obd)
 588 {
 589         struct lmv_obd          *lmv = &obd->u.lmv;
 590         struct lmv_tgt_desc     *tgt;
 591         __u32                    i;
 592         int                      rc;
 593         int                      easize;
 594         ENTRY;
 595
 596         if (lmv->connected)
 597                 RETURN(0);
 598
 599         lmv_init_lock(lmv);
 600         if (lmv->connected) {
 601                 lmv_init_unlock(lmv);
 602                 RETURN(0);
 603         }
 604
 605         if (lmv->desc.ld_tgt_count == 0) {
 606                 lmv_init_unlock(lmv);
 607                 CERROR("%s: no targets configured.\n", obd->obd_name);
 608                 RETURN(-EINVAL);
 609         }
 610
 611         LASSERT(lmv->tgts != NULL);
 612
 613         if (lmv->tgts[0] == NULL) {
 614                 lmv_init_unlock(lmv);
 615                 CERROR("%s: no target configured for index 0.\n",
 616                        obd->obd_name);
 617                 RETURN(-EINVAL);
 618         }
 619
 620         CDEBUG(D_CONFIG, "Time to connect %s to %s\n",
 621                lmv->cluuid.uuid, obd->obd_name);
 622
 623         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 624                 tgt = lmv->tgts[i];
 625                 if (tgt == NULL)
 626                         continue;
 627                 rc = lmv_connect_mdc(obd, tgt);
 628                 if (rc)
 629                         GOTO(out_disc, rc);
 630         }
 631
 632         class_export_put(lmv->exp);
 633         lmv->connected = 1;
 634         easize = lmv_mds_md_size(lmv->desc.ld_tgt_count, LMV_MAGIC);
 635         lmv_init_ea_size(obd->obd_self_export, easize, 0, 0, 0);
 636         lmv_init_unlock(lmv);
 637         RETURN(0);
 638
 639  out_disc:
 640         while (i-- > 0) {
 641                 int rc2;
 642                 tgt = lmv->tgts[i];
 643                 if (tgt == NULL)
 644                         continue;
 645                 tgt->ltd_active = 0;
 646                 if (tgt->ltd_exp) {
 647                         --lmv->desc.ld_active_tgt_count;
 648                         rc2 = obd_disconnect(tgt->ltd_exp);
 649                         if (rc2) {
 650                                 CERROR("LMV target %s disconnect on "
 651                                        "MDC idx %d: error %d\n",
 652                                        tgt->ltd_uuid.uuid, i, rc2);
 653                         }
 654                 }
 655         }
 656         class_disconnect(lmv->exp);
 657         lmv_init_unlock(lmv);
 658         RETURN(rc);
 659 }
 660
 661 static int lmv_disconnect_mdc(struct obd_device *obd, struct lmv_tgt_desc *tgt)
 662 {
 663         struct lmv_obd         *lmv = &obd->u.lmv;
 664         struct obd_device      *mdc_obd;
 665         int                     rc;
 666         ENTRY;
 667
 668         LASSERT(tgt != NULL);
 669         LASSERT(obd != NULL);
 670
 671         mdc_obd = class_exp2obd(tgt->ltd_exp);
 672
 673         if (mdc_obd) {
 674                 mdc_obd->obd_force = obd->obd_force;
 675                 mdc_obd->obd_fail = obd->obd_fail;
 676                 mdc_obd->obd_no_recov = obd->obd_no_recov;
 677         }
 678
 679         if (lmv->targets_proc_entry != NULL)
 680                 lprocfs_remove_proc_entry(mdc_obd->obd_name,
 681                                           lmv->targets_proc_entry);
 682
 683         rc = obd_fid_fini(tgt->ltd_exp->exp_obd);
 684         if (rc)
 685                 CERROR("Can't finanize fids factory\n");
 686
 687         CDEBUG(D_INFO, "Disconnected from %s(%s) successfully\n",
 688                tgt->ltd_exp->exp_obd->obd_name,
 689                tgt->ltd_exp->exp_obd->obd_uuid.uuid);
 690
 691         obd_register_observer(tgt->ltd_exp->exp_obd, NULL);
 692         rc = obd_disconnect(tgt->ltd_exp);
 693         if (rc) {
 694                 if (tgt->ltd_active) {
 695                         CERROR("Target %s disconnect error %d\n",
 696                                tgt->ltd_uuid.uuid, rc);
 697                 }
 698         }
 699
 700         lmv_activate_target(lmv, tgt, 0);
 701         tgt->ltd_exp = NULL;
 702         RETURN(0);
 703 }
 704
 705 static int lmv_disconnect(struct obd_export *exp)
 706 {
 707         struct obd_device       *obd = class_exp2obd(exp);
 708         struct lmv_obd          *lmv = &obd->u.lmv;
 709         int                      rc;
 710         __u32                    i;
 711         ENTRY;
 712
 713         if (!lmv->tgts)
 714                 goto out_local;
 715
 716         /*
 717          * Only disconnect the underlying layers on the final disconnect.
 718          */
 719         lmv->refcount--;
 720         if (lmv->refcount != 0)
 721                 goto out_local;
 722
 723         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 724                 if (lmv->tgts[i] == NULL || lmv->tgts[i]->ltd_exp == NULL)
 725                         continue;
 726
 727                 lmv_disconnect_mdc(obd, lmv->tgts[i]);
 728         }
 729
 730         if (lmv->targets_proc_entry != NULL)
 731                 lprocfs_remove(&lmv->targets_proc_entry);
 732         else
 733                 CERROR("/proc/fs/lustre/%s/%s/target_obds missing\n",
 734                        obd->obd_type->typ_name, obd->obd_name);
 735
 736 out_local:
 737         /*
 738          * This is the case when no real connection is established by
 739          * lmv_check_connect().
 740          */
 741         if (!lmv->connected)
 742                 class_export_put(exp);
 743         rc = class_disconnect(exp);
 744         if (lmv->refcount == 0)
 745                 lmv->connected = 0;
 746         RETURN(rc);
 747 }
 748
 749 static int lmv_fid2path(struct obd_export *exp, int len, void *karg, void *uarg)
 750 {
 751         struct obd_device       *obddev = class_exp2obd(exp);
 752         struct lmv_obd          *lmv = &obddev->u.lmv;
 753         struct getinfo_fid2path *gf;
 754         struct lmv_tgt_desc     *tgt;
 755         struct getinfo_fid2path *remote_gf = NULL;
 756         int                     remote_gf_size = 0;
 757         int                     rc;
 758
 759         gf = (struct getinfo_fid2path *)karg;
 760         tgt = lmv_find_target(lmv, &gf->gf_fid);
 761         if (IS_ERR(tgt))
 762                 RETURN(PTR_ERR(tgt));
 763
 764 repeat_fid2path:
 765         rc = obd_iocontrol(OBD_IOC_FID2PATH, tgt->ltd_exp, len, gf, uarg);
 766         if (rc != 0 && rc != -EREMOTE)
 767                 GOTO(out_fid2path, rc);
 768
 769         /* If remote_gf != NULL, it means just building the
 770          * path on the remote MDT, copy this path segement to gf */
 771         if (remote_gf != NULL) {
 772                 struct getinfo_fid2path *ori_gf;
 773                 char *ptr;
 774
 775                 ori_gf = (struct getinfo_fid2path *)karg;
 776                 if (strlen(ori_gf->gf_path) +
 777                     strlen(gf->gf_path) > ori_gf->gf_pathlen)
 778                         GOTO(out_fid2path, rc = -EOVERFLOW);
 779
 780                 ptr = ori_gf->gf_path;
 781
 782                 memmove(ptr + strlen(gf->gf_path) + 1, ptr,
 783                         strlen(ori_gf->gf_path));
 784
 785                 strncpy(ptr, gf->gf_path, strlen(gf->gf_path));
 786                 ptr += strlen(gf->gf_path);
 787                 *ptr = '/';
 788         }
 789
 790         CDEBUG(D_INFO, "%s: get path %s "DFID" rec: "LPU64" ln: %u\n",
 791                tgt->ltd_exp->exp_obd->obd_name,
 792                gf->gf_path, PFID(&gf->gf_fid), gf->gf_recno,
 793                gf->gf_linkno);
 794
 795         if (rc == 0)
 796                 GOTO(out_fid2path, rc);
 797
 798         /* sigh, has to go to another MDT to do path building further */
 799         if (remote_gf == NULL) {
 800                 remote_gf_size = sizeof(*remote_gf) + PATH_MAX;
 801                 OBD_ALLOC(remote_gf, remote_gf_size);
 802                 if (remote_gf == NULL)
 803                         GOTO(out_fid2path, rc = -ENOMEM);
 804                 remote_gf->gf_pathlen = PATH_MAX;
 805         }
 806
 807         if (!fid_is_sane(&gf->gf_fid)) {
 808                 CERROR("%s: invalid FID "DFID": rc = %d\n",
 809                        tgt->ltd_exp->exp_obd->obd_name,
 810                        PFID(&gf->gf_fid), -EINVAL);
 811                 GOTO(out_fid2path, rc = -EINVAL);
 812         }
 813
 814         tgt = lmv_find_target(lmv, &gf->gf_fid);
 815         if (IS_ERR(tgt))
 816                 GOTO(out_fid2path, rc = -EINVAL);
 817
 818         remote_gf->gf_fid = gf->gf_fid;
 819         remote_gf->gf_recno = -1;
 820         remote_gf->gf_linkno = -1;
 821         memset(remote_gf->gf_path, 0, remote_gf->gf_pathlen);
 822         gf = remote_gf;
 823         goto repeat_fid2path;
 824
 825 out_fid2path:
 826         if (remote_gf != NULL)
 827                 OBD_FREE(remote_gf, remote_gf_size);
 828         RETURN(rc);
 829 }
 830
 831 static int lmv_hsm_req_count(struct lmv_obd *lmv,
 832                              const struct hsm_user_request *hur,
 833                              const struct lmv_tgt_desc *tgt_mds)
 834 {
 835         __u32                    i;
 836         int                      nr = 0;
 837         struct lmv_tgt_desc     *curr_tgt;
 838
 839         /* count how many requests must be sent to the given target */
 840         for (i = 0; i < hur->hur_request.hr_itemcount; i++) {
 841                 curr_tgt = lmv_find_target(lmv, &hur->hur_user_item[i].hui_fid);
 842                 if (obd_uuid_equals(&curr_tgt->ltd_uuid, &tgt_mds->ltd_uuid))
 843                         nr++;
 844         }
 845         return nr;
 846 }
 847
 848 static void lmv_hsm_req_build(struct lmv_obd *lmv,
 849                               struct hsm_user_request *hur_in,
 850                               const struct lmv_tgt_desc *tgt_mds,
 851                               struct hsm_user_request *hur_out)
 852 {
 853         __u32                    i, nr_out;
 854         struct lmv_tgt_desc     *curr_tgt;
 855
 856         /* build the hsm_user_request for the given target */
 857         hur_out->hur_request = hur_in->hur_request;
 858         nr_out = 0;
 859         for (i = 0; i < hur_in->hur_request.hr_itemcount; i++) {
 860                 curr_tgt = lmv_find_target(lmv,
 861                                            &hur_in->hur_user_item[i].hui_fid);
 862                 if (obd_uuid_equals(&curr_tgt->ltd_uuid, &tgt_mds->ltd_uuid)) {
 863                         hur_out->hur_user_item[nr_out] =
 864                                                 hur_in->hur_user_item[i];
 865                         nr_out++;
 866                 }
 867         }
 868         hur_out->hur_request.hr_itemcount = nr_out;
 869         memcpy(hur_data(hur_out), hur_data(hur_in),
 870                hur_in->hur_request.hr_data_len);
 871 }
 872
 873 static int lmv_hsm_ct_unregister(struct lmv_obd *lmv, unsigned int cmd, int len,
 874                                  struct lustre_kernelcomm *lk, void *uarg)
 875 {
 876         __u32                    i;
 877         int                      rc;
 878         struct kkuc_ct_data     *kcd = NULL;
 879         ENTRY;
 880
 881         /* unregister request (call from llapi_hsm_copytool_fini) */
 882         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 883                 struct lmv_tgt_desc *tgt = lmv->tgts[i];
 884
 885                 if (tgt == NULL || tgt->ltd_exp == NULL)
 886                         continue;
 887                 /* best effort: try to clean as much as possible
 888                  * (continue on error) */
 889                 obd_iocontrol(cmd, tgt->ltd_exp, len, lk, uarg);
 890         }
 891
 892         /* Whatever the result, remove copytool from kuc groups.
 893          * Unreached coordinators will get EPIPE on next requests
 894          * and will unregister automatically.
 895          */
 896         rc = libcfs_kkuc_group_rem(lk->lk_uid, lk->lk_group, (void **)&kcd);
 897         if (kcd != NULL)
 898                 OBD_FREE_PTR(kcd);
 899
 900         RETURN(rc);
 901 }
 902
 903 static int lmv_hsm_ct_register(struct lmv_obd *lmv, unsigned int cmd, int len,
 904                                struct lustre_kernelcomm *lk, void *uarg)
 905 {
 906         struct file             *filp;
 907         __u32                    i, j;
 908         int                      err, rc;
 909         bool                     any_set = false;
 910         struct kkuc_ct_data     *kcd;
 911         ENTRY;
 912
 913         /* All or nothing: try to register to all MDS.
 914          * In case of failure, unregister from previous MDS,
 915          * except if it because of inactive target. */
 916         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 917                 struct lmv_tgt_desc *tgt = lmv->tgts[i];
 918
 919                 if (tgt == NULL || tgt->ltd_exp == NULL)
 920                         continue;
 921                 err = obd_iocontrol(cmd, tgt->ltd_exp, len, lk, uarg);
 922                 if (err) {
 923                         if (tgt->ltd_active) {
 924                                 /* permanent error */
 925                                 CERROR("%s: iocontrol MDC %s on MDT"
 926                                        " idx %d cmd %x: err = %d\n",
 927                                        class_exp2obd(lmv->exp)->obd_name,
 928                                        tgt->ltd_uuid.uuid, i, cmd, err);
 929                                 rc = err;
 930                                 lk->lk_flags |= LK_FLG_STOP;
 931                                 /* unregister from previous MDS */
 932                                 for (j = 0; j < i; j++) {
 933                                         tgt = lmv->tgts[j];
 934                                         if (tgt == NULL || tgt->ltd_exp == NULL)
 935                                                 continue;
 936                                         obd_iocontrol(cmd, tgt->ltd_exp, len,
 937                                                       lk, uarg);
 938                                 }
 939                                 RETURN(rc);
 940                         }
 941                         /* else: transient error.
 942                          * kuc will register to the missing MDT
 943                          * when it is back */
 944                 } else {
 945                         any_set = true;
 946                 }
 947         }
 948
 949         if (!any_set)
 950                 /* no registration done: return error */
 951                 RETURN(-ENOTCONN);
 952
 953         /* at least one registration done, with no failure */
 954         filp = fget(lk->lk_wfd);
 955         if (filp == NULL)
 956                 RETURN(-EBADF);
 957
 958         OBD_ALLOC_PTR(kcd);
 959         if (kcd == NULL) {
 960                 fput(filp);
 961                 RETURN(-ENOMEM);
 962         }
 963         kcd->kcd_magic = KKUC_CT_DATA_MAGIC;
 964         kcd->kcd_uuid = lmv->cluuid;
 965         kcd->kcd_archive = lk->lk_data;
 966
 967         rc = libcfs_kkuc_group_add(filp, lk->lk_uid, lk->lk_group, kcd);
 968         if (rc != 0) {
 969                 if (filp != NULL)
 970                         fput(filp);
 971                 OBD_FREE_PTR(kcd);
 972         }
 973
 974         RETURN(rc);
 975 }
 976
 977
 978
 979
 980 static int lmv_iocontrol(unsigned int cmd, struct obd_export *exp,
 981                          int len, void *karg, void *uarg)
 982 {
 983         struct obd_device       *obddev = class_exp2obd(exp);
 984         struct lmv_obd          *lmv = &obddev->u.lmv;
 985         struct lmv_tgt_desc     *tgt = NULL;
 986         __u32                    i = 0;
 987         int                      rc = 0;
 988         int                      set = 0;
 989         __u32                    count = lmv->desc.ld_tgt_count;
 990         ENTRY;
 991
 992         if (count == 0)
 993                 RETURN(-ENOTTY);
 994
 995         switch (cmd) {
 996         case IOC_OBD_STATFS: {
 997                 struct obd_ioctl_data *data = karg;
 998                 struct obd_device *mdc_obd;
 999                 struct obd_statfs stat_buf = {0};
1000                 __u32 index;
1001
1002                 memcpy(&index, data->ioc_inlbuf2, sizeof(__u32));
1003                 if ((index >= count))
1004                         RETURN(-ENODEV);
1005
1006                 tgt = lmv->tgts[index];
1007                 if (tgt == NULL || !tgt->ltd_active)
1008                         RETURN(-ENODATA);
1009
1010                 mdc_obd = class_exp2obd(tgt->ltd_exp);
1011                 if (!mdc_obd)
1012                         RETURN(-EINVAL);
1013
1014                 /* copy UUID */
1015                 if (copy_to_user(data->ioc_pbuf2, obd2cli_tgt(mdc_obd),
1016                                  min((int) data->ioc_plen2,
1017                                      (int) sizeof(struct obd_uuid))))
1018                         RETURN(-EFAULT);
1019
1020                 rc = obd_statfs(NULL, tgt->ltd_exp, &stat_buf,
1021                                 cfs_time_shift_64(-OBD_STATFS_CACHE_SECONDS),
1022                                 0);
1023                 if (rc)
1024                         RETURN(rc);
1025                 if (copy_to_user(data->ioc_pbuf1, &stat_buf,
1026                                  min((int) data->ioc_plen1,
1027                                      (int) sizeof(stat_buf))))
1028                         RETURN(-EFAULT);
1029                 break;
1030         }
1031         case OBD_IOC_QUOTACTL: {
1032                 struct if_quotactl *qctl = karg;
1033                 struct obd_quotactl *oqctl;
1034
1035                 if (qctl->qc_valid == QC_MDTIDX) {
1036                         if (count <= qctl->qc_idx)
1037                                 RETURN(-EINVAL);
1038
1039                         tgt = lmv->tgts[qctl->qc_idx];
1040                         if (tgt == NULL || tgt->ltd_exp == NULL)
1041                                 RETURN(-EINVAL);
1042                 } else if (qctl->qc_valid == QC_UUID) {
1043                         for (i = 0; i < count; i++) {
1044                                 tgt = lmv->tgts[i];
1045                                 if (tgt == NULL)
1046                                         continue;
1047                                 if (!obd_uuid_equals(&tgt->ltd_uuid,
1048                                                      &qctl->obd_uuid))
1049                                         continue;
1050
1051                                 if (tgt->ltd_exp == NULL)
1052                                         RETURN(-EINVAL);
1053
1054                                 break;
1055                         }
1056                 } else {
1057                         RETURN(-EINVAL);
1058                 }
1059
1060                 if (i >= count)
1061                         RETURN(-EAGAIN);
1062
1063                 LASSERT(tgt != NULL && tgt->ltd_exp != NULL);
1064                 OBD_ALLOC_PTR(oqctl);
1065                 if (!oqctl)
1066                         RETURN(-ENOMEM);
1067
1068                 QCTL_COPY(oqctl, qctl);
1069                 rc = obd_quotactl(tgt->ltd_exp, oqctl);
1070                 if (rc == 0) {
1071                         QCTL_COPY(qctl, oqctl);
1072                         qctl->qc_valid = QC_MDTIDX;
1073                         qctl->obd_uuid = tgt->ltd_uuid;
1074                 }
1075                 OBD_FREE_PTR(oqctl);
1076                 break;
1077         }
1078         case OBD_IOC_CHANGELOG_SEND:
1079         case OBD_IOC_CHANGELOG_CLEAR: {
1080                 struct ioc_changelog *icc = karg;
1081
1082                 if (icc->icc_mdtindex >= count)
1083                         RETURN(-ENODEV);
1084
1085                 tgt = lmv->tgts[icc->icc_mdtindex];
1086                 if (tgt == NULL || tgt->ltd_exp == NULL || !tgt->ltd_active)
1087                         RETURN(-ENODEV);
1088                 rc = obd_iocontrol(cmd, tgt->ltd_exp, sizeof(*icc), icc, NULL);
1089                 break;
1090         }
1091         case LL_IOC_GET_CONNECT_FLAGS: {
1092                 tgt = lmv->tgts[0];
1093                 if (tgt == NULL || tgt->ltd_exp == NULL)
1094                         RETURN(-ENODATA);
1095                 rc = obd_iocontrol(cmd, tgt->ltd_exp, len, karg, uarg);
1096                 break;
1097         }
1098         case LL_IOC_FID2MDTIDX: {
1099                 struct lu_fid *fid = karg;
1100                 int             mdt_index;
1101
1102                 rc = lmv_fld_lookup(lmv, fid, &mdt_index);
1103                 if (rc != 0)
1104                         RETURN(rc);
1105
1106                 /* Note: this is from llite(see ll_dir_ioctl()), @uarg does not
1107                  * point to user space memory for FID2MDTIDX. */
1108                 *(__u32 *)uarg = mdt_index;
1109                 break;
1110         }
1111         case OBD_IOC_FID2PATH: {
1112                 rc = lmv_fid2path(exp, len, karg, uarg);
1113                 break;
1114         }
1115         case LL_IOC_HSM_STATE_GET:
1116         case LL_IOC_HSM_STATE_SET:
1117         case LL_IOC_HSM_ACTION: {
1118                 struct md_op_data       *op_data = karg;
1119
1120                 tgt = lmv_find_target(lmv, &op_data->op_fid1);
1121                 if (IS_ERR(tgt))
1122                         RETURN(PTR_ERR(tgt));
1123
1124                 if (tgt->ltd_exp == NULL)
1125                         RETURN(-EINVAL);
1126
1127                 rc = obd_iocontrol(cmd, tgt->ltd_exp, len, karg, uarg);
1128                 break;
1129         }
1130         case LL_IOC_HSM_PROGRESS: {
1131                 const struct hsm_progress_kernel *hpk = karg;
1132
1133                 tgt = lmv_find_target(lmv, &hpk->hpk_fid);
1134                 if (IS_ERR(tgt))
1135                         RETURN(PTR_ERR(tgt));
1136                 rc = obd_iocontrol(cmd, tgt->ltd_exp, len, karg, uarg);
1137                 break;
1138         }
1139         case LL_IOC_HSM_REQUEST: {
1140                 struct hsm_user_request *hur = karg;
1141                 unsigned int reqcount = hur->hur_request.hr_itemcount;
1142
1143                 if (reqcount == 0)
1144                         RETURN(0);
1145
1146                 /* if the request is about a single fid
1147                  * or if there is a single MDS, no need to split
1148                  * the request. */
1149                 if (reqcount == 1 || count == 1) {
1150                         tgt = lmv_find_target(lmv,
1151                                               &hur->hur_user_item[0].hui_fid);
1152                         if (IS_ERR(tgt))
1153                                 RETURN(PTR_ERR(tgt));
1154                         rc = obd_iocontrol(cmd, tgt->ltd_exp, len, karg, uarg);
1155                 } else {
1156                         /* split fid list to their respective MDS */
1157                         for (i = 0; i < count; i++) {
1158                                 unsigned int            nr, reqlen;
1159                                 int                     rc1;
1160                                 struct hsm_user_request *req;
1161
1162                                 tgt = lmv->tgts[i];
1163                                 if (tgt == NULL || tgt->ltd_exp == NULL)
1164                                         continue;
1165
1166                                 nr = lmv_hsm_req_count(lmv, hur, tgt);
1167                                 if (nr == 0) /* nothing for this MDS */
1168                                         continue;
1169
1170                                 /* build a request with fids for this MDS */
1171                                 reqlen = offsetof(typeof(*hur),
1172                                                   hur_user_item[nr])
1173                                                 + hur->hur_request.hr_data_len;
1174                                 OBD_ALLOC_LARGE(req, reqlen);
1175                                 if (req == NULL)
1176                                         RETURN(-ENOMEM);
1177
1178                                 lmv_hsm_req_build(lmv, hur, tgt, req);
1179
1180                                 rc1 = obd_iocontrol(cmd, tgt->ltd_exp, reqlen,
1181                                                     req, uarg);
1182                                 if (rc1 != 0 && rc == 0)
1183                                         rc = rc1;
1184                                 OBD_FREE_LARGE(req, reqlen);
1185                         }
1186                 }
1187                 break;
1188         }
1189         case LL_IOC_LOV_SWAP_LAYOUTS: {
1190                 struct md_op_data       *op_data = karg;
1191                 struct lmv_tgt_desc     *tgt1, *tgt2;
1192
1193                 tgt1 = lmv_find_target(lmv, &op_data->op_fid1);
1194                 if (IS_ERR(tgt1))
1195                         RETURN(PTR_ERR(tgt1));
1196
1197                 tgt2 = lmv_find_target(lmv, &op_data->op_fid2);
1198                 if (IS_ERR(tgt2))
1199                         RETURN(PTR_ERR(tgt2));
1200
1201                 if ((tgt1->ltd_exp == NULL) || (tgt2->ltd_exp == NULL))
1202                         RETURN(-EINVAL);
1203
1204                 /* only files on same MDT can have their layouts swapped */
1205                 if (tgt1->ltd_idx != tgt2->ltd_idx)
1206                         RETURN(-EPERM);
1207
1208                 rc = obd_iocontrol(cmd, tgt1->ltd_exp, len, karg, uarg);
1209                 break;
1210         }
1211         case LL_IOC_HSM_CT_START: {
1212                 struct lustre_kernelcomm *lk = karg;
1213                 if (lk->lk_flags & LK_FLG_STOP)
1214                         rc = lmv_hsm_ct_unregister(lmv, cmd, len, lk, uarg);
1215                 else
1216                         rc = lmv_hsm_ct_register(lmv, cmd, len, lk, uarg);
1217                 break;
1218         }
1219         default:
1220                 for (i = 0; i < count; i++) {
1221                         struct obd_device *mdc_obd;
1222                         int err;
1223
1224                         tgt = lmv->tgts[i];
1225                         if (tgt == NULL || tgt->ltd_exp == NULL)
1226                                 continue;
1227                         /* ll_umount_begin() sets force flag but for lmv, not
1228                          * mdc. Let's pass it through */
1229                         mdc_obd = class_exp2obd(tgt->ltd_exp);
1230                         mdc_obd->obd_force = obddev->obd_force;
1231                         err = obd_iocontrol(cmd, tgt->ltd_exp, len, karg, uarg);
1232                         if (err == -ENODATA && cmd == OBD_IOC_POLL_QUOTACHECK) {
1233                                 RETURN(err);
1234                         } else if (err) {
1235                                 if (tgt->ltd_active) {
1236                                         CERROR("error: iocontrol MDC %s on MDT"
1237                                                " idx %d cmd %x: err = %d\n",
1238                                                tgt->ltd_uuid.uuid, i, cmd, err);
1239                                         if (!rc)
1240                                                 rc = err;
1241                                 }
1242                         } else
1243                                 set = 1;
1244                 }
1245                 if (!set && !rc)
1246                         rc = -EIO;
1247         }
1248         RETURN(rc);
1249 }
1250
1251 #if 0
1252 static int lmv_all_chars_policy(int count, const char *name,
1253                                 int len)
1254 {
1255         unsigned int c = 0;
1256
1257         while (len > 0)
1258                 c += name[--len];
1259         c = c % count;
1260         return c;
1261 }
1262
1263 static int lmv_nid_policy(struct lmv_obd *lmv)
1264 {
1265         struct obd_import *imp;
1266         __u32              id;
1267
1268         /*
1269          * XXX: To get nid we assume that underlying obd device is mdc.
1270          */
1271         imp = class_exp2cliimp(lmv->tgts[0].ltd_exp);
1272         id = imp->imp_connection->c_self ^ (imp->imp_connection->c_self >> 32);
1273         return id % lmv->desc.ld_tgt_count;
1274 }
1275
1276 static int lmv_choose_mds(struct lmv_obd *lmv, struct md_op_data *op_data,
1277                           placement_policy_t placement)
1278 {
1279         switch (placement) {
1280         case PLACEMENT_CHAR_POLICY:
1281                 return lmv_all_chars_policy(lmv->desc.ld_tgt_count,
1282                                             op_data->op_name,
1283                                             op_data->op_namelen);
1284         case PLACEMENT_NID_POLICY:
1285                 return lmv_nid_policy(lmv);
1286
1287         default:
1288                 break;
1289         }
1290
1291         CERROR("Unsupported placement policy %x\n", placement);
1292         return -EINVAL;
1293 }
1294 #endif
1295
1296 /**
1297  * This is _inode_ placement policy function (not name).
1298  */
1299 static int lmv_placement_policy(struct obd_device *obd,
1300                                 struct md_op_data *op_data, u32 *mds)
1301 {
1302         struct lmv_obd          *lmv = &obd->u.lmv;
1303         ENTRY;
1304
1305         LASSERT(mds != NULL);
1306
1307         if (lmv->desc.ld_tgt_count == 1) {
1308                 *mds = 0;
1309                 RETURN(0);
1310         }
1311
1312         /**
1313          * If stripe_offset is provided during setdirstripe
1314          * (setdirstripe -i xx), xx MDS will be choosen.
1315          */
1316         if (op_data->op_cli_flags & CLI_SET_MEA && op_data->op_data != NULL) {
1317                 struct lmv_user_md *lum;
1318
1319                 lum = op_data->op_data;
1320
1321                 if (le32_to_cpu(lum->lum_stripe_offset) != (__u32)-1) {
1322                         *mds = le32_to_cpu(lum->lum_stripe_offset);
1323                 } else {
1324                         /* -1 means default, which will be in the same MDT with
1325                          * the stripe */
1326                         *mds = op_data->op_mds;
1327                         lum->lum_stripe_offset = cpu_to_le32(op_data->op_mds);
1328                 }
1329         } else {
1330                 /* Allocate new fid on target according to operation type and
1331                  * parent home mds. */
1332                 *mds = op_data->op_mds;
1333         }
1334
1335         RETURN(0);
1336 }
1337
1338 int __lmv_fid_alloc(struct lmv_obd *lmv, struct lu_fid *fid, u32 mds)
1339 {
1340         struct lmv_tgt_desc     *tgt;
1341         int                      rc;
1342         ENTRY;
1343
1344         tgt = lmv_get_target(lmv, mds, NULL);
1345         if (IS_ERR(tgt))
1346                 RETURN(PTR_ERR(tgt));
1347
1348         /*
1349          * New seq alloc and FLD setup should be atomic. Otherwise we may find
1350          * on server that seq in new allocated fid is not yet known.
1351          */
1352         mutex_lock(&tgt->ltd_fid_mutex);
1353
1354         if (tgt->ltd_active == 0 || tgt->ltd_exp == NULL)
1355                 GOTO(out, rc = -ENODEV);
1356
1357         /*
1358          * Asking underlying tgt layer to allocate new fid.
1359          */
1360         rc = obd_fid_alloc(NULL, tgt->ltd_exp, fid, NULL);
1361         if (rc > 0) {
1362                 LASSERT(fid_is_sane(fid));
1363                 rc = 0;
1364         }
1365
1366         EXIT;
1367 out:
1368         mutex_unlock(&tgt->ltd_fid_mutex);
1369         return rc;
1370 }
1371
1372 int lmv_fid_alloc(const struct lu_env *env, struct obd_export *exp,
1373                   struct lu_fid *fid, struct md_op_data *op_data)
1374 {
1375         struct obd_device     *obd = class_exp2obd(exp);
1376         struct lmv_obd        *lmv = &obd->u.lmv;
1377         u32                    mds = 0;
1378         int                    rc;
1379         ENTRY;
1380
1381         LASSERT(op_data != NULL);
1382         LASSERT(fid != NULL);
1383
1384         rc = lmv_placement_policy(obd, op_data, &mds);
1385         if (rc) {
1386                 CERROR("Can't get target for allocating fid, "
1387                        "rc %d\n", rc);
1388                 RETURN(rc);
1389         }
1390
1391         rc = __lmv_fid_alloc(lmv, fid, mds);
1392         if (rc) {
1393                 CERROR("Can't alloc new fid, rc %d\n", rc);
1394                 RETURN(rc);
1395         }
1396
1397         RETURN(rc);
1398 }
1399
1400 static int lmv_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
1401 {
1402         struct lmv_obd  *lmv = &obd->u.lmv;
1403         struct lmv_desc *desc;
1404         int             rc;
1405         ENTRY;
1406
1407         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
1408                 CERROR("LMV setup requires a descriptor\n");
1409                 RETURN(-EINVAL);
1410         }
1411
1412         desc = (struct lmv_desc *)lustre_cfg_buf(lcfg, 1);
1413         if (sizeof(*desc) > LUSTRE_CFG_BUFLEN(lcfg, 1)) {
1414                 CERROR("Lmv descriptor size wrong: %d > %d\n",
1415                        (int)sizeof(*desc), LUSTRE_CFG_BUFLEN(lcfg, 1));
1416                 RETURN(-EINVAL);
1417         }
1418
1419         lmv->tgts_size = 32U;
1420         OBD_ALLOC(lmv->tgts, sizeof(*lmv->tgts) * lmv->tgts_size);
1421         if (lmv->tgts == NULL)
1422                 RETURN(-ENOMEM);
1423
1424         obd_str2uuid(&lmv->desc.ld_uuid, desc->ld_uuid.uuid);
1425         lmv->desc.ld_tgt_count = 0;
1426         lmv->desc.ld_active_tgt_count = 0;
1427         lmv->max_cookiesize = 0;
1428         lmv->max_def_easize = 0;
1429         lmv->max_easize = 0;
1430         lmv->lmv_placement = PLACEMENT_CHAR_POLICY;
1431
1432         spin_lock_init(&lmv->lmv_lock);
1433         mutex_init(&lmv->init_mutex);
1434
1435 #ifdef LPROCFS
1436         obd->obd_vars = lprocfs_lmv_obd_vars;
1437         lprocfs_obd_setup(obd);
1438         lprocfs_alloc_md_stats(obd, 0);
1439         rc = lprocfs_seq_create(obd->obd_proc_entry, "target_obd",
1440                                 0444, &lmv_proc_target_fops, obd);
1441         if (rc)
1442                 CWARN("%s: error adding LMV target_obd file: rc = %d\n",
1443                       obd->obd_name, rc);
1444 #endif
1445         rc = fld_client_init(&lmv->lmv_fld, obd->obd_name,
1446                              LUSTRE_CLI_FLD_HASH_DHT);
1447         if (rc) {
1448                 CERROR("Can't init FLD, err %d\n", rc);
1449                 GOTO(out, rc);
1450         }
1451
1452         RETURN(0);
1453
1454 out:
1455         return rc;
1456 }
1457
1458 static int lmv_cleanup(struct obd_device *obd)
1459 {
1460         struct lmv_obd   *lmv = &obd->u.lmv;
1461         ENTRY;
1462
1463         fld_client_fini(&lmv->lmv_fld);
1464         if (lmv->tgts != NULL) {
1465                 int i;
1466                 for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
1467                         if (lmv->tgts[i] == NULL)
1468                                 continue;
1469                         lmv_del_target(lmv, i);
1470                 }
1471                 OBD_FREE(lmv->tgts, sizeof(*lmv->tgts) * lmv->tgts_size);
1472                 lmv->tgts_size = 0;
1473         }
1474         RETURN(0);
1475 }
1476
1477 static int lmv_process_config(struct obd_device *obd, obd_count len, void *buf)
1478 {
1479         struct lustre_cfg       *lcfg = buf;
1480         struct obd_uuid         obd_uuid;
1481         int                     gen;
1482         __u32                   index;
1483         int                     rc;
1484         ENTRY;
1485
1486         switch (lcfg->lcfg_command) {
1487         case LCFG_ADD_MDC:
1488                 /* modify_mdc_tgts add 0:lustre-clilmv  1:lustre-MDT0000_UUID
1489                  * 2:0  3:1  4:lustre-MDT0000-mdc_UUID */
1490                 if (LUSTRE_CFG_BUFLEN(lcfg, 1) > sizeof(obd_uuid.uuid))
1491                         GOTO(out, rc = -EINVAL);
1492
1493                 obd_str2uuid(&obd_uuid,  lustre_cfg_buf(lcfg, 1));
1494
1495                 if (sscanf(lustre_cfg_buf(lcfg, 2), "%u", &index) != 1)
1496                         GOTO(out, rc = -EINVAL);
1497                 if (sscanf(lustre_cfg_buf(lcfg, 3), "%d", &gen) != 1)
1498                         GOTO(out, rc = -EINVAL);
1499                 rc = lmv_add_target(obd, &obd_uuid, index, gen);
1500                 GOTO(out, rc);
1501         default:
1502                 CERROR("Unknown command: %d\n", lcfg->lcfg_command);
1503                 GOTO(out, rc = -EINVAL);
1504         }
1505 out:
1506         RETURN(rc);
1507 }
1508
1509 static int lmv_statfs(const struct lu_env *env, struct obd_export *exp,
1510                       struct obd_statfs *osfs, __u64 max_age, __u32 flags)
1511 {
1512         struct obd_device       *obd = class_exp2obd(exp);
1513         struct lmv_obd          *lmv = &obd->u.lmv;
1514         struct obd_statfs       *temp;
1515         int                      rc = 0;
1516         __u32                    i;
1517         ENTRY;
1518
1519         rc = lmv_check_connect(obd);
1520         if (rc)
1521                 RETURN(rc);
1522
1523         OBD_ALLOC(temp, sizeof(*temp));
1524         if (temp == NULL)
1525                 RETURN(-ENOMEM);
1526
1527         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
1528                 if (lmv->tgts[i] == NULL || lmv->tgts[i]->ltd_exp == NULL)
1529                         continue;
1530
1531                 rc = obd_statfs(env, lmv->tgts[i]->ltd_exp, temp,
1532                                 max_age, flags);
1533                 if (rc) {
1534                         CERROR("can't stat MDS #%d (%s), error %d\n", i,
1535                                lmv->tgts[i]->ltd_exp->exp_obd->obd_name,
1536                                rc);
1537                         GOTO(out_free_temp, rc);
1538                 }
1539
1540                 if (i == 0) {
1541                         *osfs = *temp;
1542                         /* If the statfs is from mount, it will needs
1543                          * retrieve necessary information from MDT0.
1544                          * i.e. mount does not need the merged osfs
1545                          * from all of MDT.
1546                          * And also clients can be mounted as long as
1547                          * MDT0 is in service*/
1548                         if (flags & OBD_STATFS_FOR_MDT0)
1549                                 GOTO(out_free_temp, rc);
1550                 } else {
1551                         osfs->os_bavail += temp->os_bavail;
1552                         osfs->os_blocks += temp->os_blocks;
1553                         osfs->os_ffree += temp->os_ffree;
1554                         osfs->os_files += temp->os_files;
1555                 }
1556         }
1557
1558         EXIT;
1559 out_free_temp:
1560         OBD_FREE(temp, sizeof(*temp));
1561         return rc;
1562 }
1563
1564 static int lmv_getstatus(struct obd_export *exp,
1565                          struct lu_fid *fid,
1566                          struct obd_capa **pc)
1567 {
1568         struct obd_device    *obd = exp->exp_obd;
1569         struct lmv_obd       *lmv = &obd->u.lmv;
1570         int                   rc;
1571         ENTRY;
1572
1573         rc = lmv_check_connect(obd);
1574         if (rc)
1575                 RETURN(rc);
1576
1577         rc = md_getstatus(lmv->tgts[0]->ltd_exp, fid, pc);
1578         RETURN(rc);
1579 }
1580
1581 static int lmv_getxattr(struct obd_export *exp, const struct lu_fid *fid,
1582                         struct obd_capa *oc, obd_valid valid, const char *name,
1583                         const char *input, int input_size, int output_size,
1584                         int flags, struct ptlrpc_request **request)
1585 {
1586         struct obd_device      *obd = exp->exp_obd;
1587         struct lmv_obd         *lmv = &obd->u.lmv;
1588         struct lmv_tgt_desc    *tgt;
1589         int                     rc;
1590         ENTRY;
1591
1592         rc = lmv_check_connect(obd);
1593         if (rc)
1594                 RETURN(rc);
1595
1596         tgt = lmv_find_target(lmv, fid);
1597         if (IS_ERR(tgt))
1598                 RETURN(PTR_ERR(tgt));
1599
1600         rc = md_getxattr(tgt->ltd_exp, fid, oc, valid, name, input,
1601                          input_size, output_size, flags, request);
1602
1603         RETURN(rc);
1604 }
1605
1606 static int lmv_setxattr(struct obd_export *exp, const struct lu_fid *fid,
1607                         struct obd_capa *oc, obd_valid valid, const char *name,
1608                         const char *input, int input_size, int output_size,
1609                         int flags, __u32 suppgid,
1610                         struct ptlrpc_request **request)
1611 {
1612         struct obd_device      *obd = exp->exp_obd;
1613         struct lmv_obd         *lmv = &obd->u.lmv;
1614         struct lmv_tgt_desc    *tgt;
1615         int                     rc;
1616         ENTRY;
1617
1618         rc = lmv_check_connect(obd);
1619         if (rc)
1620                 RETURN(rc);
1621
1622         tgt = lmv_find_target(lmv, fid);
1623         if (IS_ERR(tgt))
1624                 RETURN(PTR_ERR(tgt));
1625
1626         rc = md_setxattr(tgt->ltd_exp, fid, oc, valid, name, input,
1627                          input_size, output_size, flags, suppgid,
1628                          request);
1629
1630         RETURN(rc);
1631 }
1632
1633 static int lmv_getattr(struct obd_export *exp, struct md_op_data *op_data,
1634                        struct ptlrpc_request **request)
1635 {
1636         struct obd_device       *obd = exp->exp_obd;
1637         struct lmv_obd          *lmv = &obd->u.lmv;
1638         struct lmv_tgt_desc     *tgt;
1639         int                      rc;
1640         ENTRY;
1641
1642         rc = lmv_check_connect(obd);
1643         if (rc)
1644                 RETURN(rc);
1645
1646         tgt = lmv_find_target(lmv, &op_data->op_fid1);
1647         if (IS_ERR(tgt))
1648                 RETURN(PTR_ERR(tgt));
1649
1650         if (op_data->op_flags & MF_GET_MDT_IDX) {
1651                 op_data->op_mds = tgt->ltd_idx;
1652                 RETURN(0);
1653         }
1654
1655         rc = md_getattr(tgt->ltd_exp, op_data, request);
1656
1657         RETURN(rc);
1658 }
1659
1660 static int lmv_null_inode(struct obd_export *exp, const struct lu_fid *fid)
1661 {
1662         struct obd_device   *obd = exp->exp_obd;
1663         struct lmv_obd      *lmv = &obd->u.lmv;
1664         __u32                i;
1665         int                  rc;
1666         ENTRY;
1667
1668         rc = lmv_check_connect(obd);
1669         if (rc)
1670                 RETURN(rc);
1671
1672         CDEBUG(D_INODE, "CBDATA for "DFID"\n", PFID(fid));
1673
1674         /*
1675          * With DNE every object can have two locks in different namespaces:
1676          * lookup lock in space of MDT storing direntry and update/open lock in
1677          * space of MDT storing inode.
1678          */
1679         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
1680                 if (lmv->tgts[i] == NULL || lmv->tgts[i]->ltd_exp == NULL)
1681                         continue;
1682                 md_null_inode(lmv->tgts[i]->ltd_exp, fid);
1683         }
1684
1685         RETURN(0);
1686 }
1687
1688 static int lmv_find_cbdata(struct obd_export *exp, const struct lu_fid *fid,
1689                            ldlm_iterator_t it, void *data)
1690 {
1691         struct obd_device       *obd = exp->exp_obd;
1692         struct lmv_obd          *lmv = &obd->u.lmv;
1693         int                     i;
1694         int                     tgt;
1695         int                     rc;
1696         ENTRY;
1697
1698         rc = lmv_check_connect(obd);
1699         if (rc)
1700                 RETURN(rc);
1701
1702         CDEBUG(D_INODE, "CBDATA for "DFID"\n", PFID(fid));
1703
1704         /*
1705          * With DNE every object can have two locks in different namespaces:
1706          * lookup lock in space of MDT storing direntry and update/open lock in
1707          * space of MDT storing inode.  Try the MDT that the FID maps to first,
1708          * since this can be easily found, and only try others if that fails.
1709          */
1710         for (i = 0, tgt = lmv_find_target_index(lmv, fid);
1711              i < lmv->desc.ld_tgt_count;
1712              i++, tgt = (tgt + 1) % lmv->desc.ld_tgt_count) {
1713                 if (tgt < 0) {
1714                         CDEBUG(D_HA, "%s: "DFID" is inaccessible: rc = %d\n",
1715                                obd->obd_name, PFID(fid), tgt);
1716                         tgt = 0;
1717                 }
1718
1719                 if (lmv->tgts[tgt] == NULL ||
1720                     lmv->tgts[tgt]->ltd_exp == NULL)
1721                         continue;
1722
1723                 rc = md_find_cbdata(lmv->tgts[tgt]->ltd_exp, fid, it, data);
1724                 if (rc)
1725                         RETURN(rc);
1726         }
1727
1728         RETURN(rc);
1729 }
1730
1731
1732 static int lmv_close(struct obd_export *exp, struct md_op_data *op_data,
1733                      struct md_open_data *mod, struct ptlrpc_request **request)
1734 {
1735         struct obd_device     *obd = exp->exp_obd;
1736         struct lmv_obd        *lmv = &obd->u.lmv;
1737         struct lmv_tgt_desc   *tgt;
1738         int                    rc;
1739         ENTRY;
1740
1741         rc = lmv_check_connect(obd);
1742         if (rc)
1743                 RETURN(rc);
1744
1745         tgt = lmv_find_target(lmv, &op_data->op_fid1);
1746         if (IS_ERR(tgt))
1747                 RETURN(PTR_ERR(tgt));
1748
1749         CDEBUG(D_INODE, "CLOSE "DFID"\n", PFID(&op_data->op_fid1));
1750         rc = md_close(tgt->ltd_exp, op_data, mod, request);
1751         RETURN(rc);
1752 }
1753
1754 /**
1755  * Choosing the MDT by name or FID in @op_data.
1756  * For non-striped directory, it will locate MDT by fid.
1757  * For striped-directory, it will locate MDT by name. And also
1758  * it will reset op_fid1 with the FID of the choosen stripe.
1759  **/
1760 struct lmv_tgt_desc *
1761 lmv_locate_target_for_name(struct lmv_obd *lmv, struct lmv_stripe_md *lsm,
1762                            const char *name, int namelen, struct lu_fid *fid,
1763                            u32 *mds)
1764 {
1765         struct lmv_tgt_desc     *tgt;
1766         const struct lmv_oinfo  *oinfo;
1767
1768         oinfo = lsm_name_to_stripe_info(lsm, name, namelen);
1769         if (IS_ERR(oinfo))
1770                 RETURN(ERR_CAST(oinfo));
1771         *fid = oinfo->lmo_fid;
1772         *mds = oinfo->lmo_mds;
1773         tgt = lmv_get_target(lmv, *mds, NULL);
1774
1775         CDEBUG(D_INFO, "locate on mds %u "DFID"\n", *mds, PFID(fid));
1776         return tgt;
1777 }
1778
1779 /**
1780  * Locate mds by fid or name
1781  *
1782  * For striped directory (lsm != NULL), it will locate the stripe
1783  * by name hash (see lsm_name_to_stripe_info()). Note: if the hash_type
1784  * is unknown, it will return -EBADFD, and lmv_intent_lookup might need
1785  * walk through all of stripes to locate the entry.
1786  *
1787  * For normal direcotry, it will locate MDS by FID directly.
1788  * \param[in] lmv       LMV device
1789  * \param[in] op_data   client MD stack parameters, name, namelen
1790  *                      mds_num etc.
1791  * \param[in] fid       object FID used to locate MDS.
1792  *
1793  * retval               pointer to the lmv_tgt_desc if succeed.
1794  *                      ERR_PTR(errno) if failed.
1795  */
1796 struct lmv_tgt_desc*
1797 lmv_locate_mds(struct lmv_obd *lmv, struct md_op_data *op_data,
1798                struct lu_fid *fid)
1799 {
1800         struct lmv_stripe_md    *lsm = op_data->op_mea1;
1801         struct lmv_tgt_desc     *tgt;
1802
1803         /* During creating VOLATILE file, it should honor the mdt
1804          * index if the file under striped dir is being restored, see
1805          * ct_restore(). */
1806         if (op_data->op_bias & MDS_CREATE_VOLATILE &&
1807             (int)op_data->op_mds != -1 && lsm != NULL) {
1808                 int i;
1809                 tgt = lmv_get_target(lmv, op_data->op_mds, NULL);
1810                 if (IS_ERR(tgt))
1811                         return tgt;
1812
1813                 /* refill the right parent fid */
1814                 for (i = 0; i < lsm->lsm_md_stripe_count; i++) {
1815                         struct lmv_oinfo *oinfo;
1816
1817                         oinfo = &lsm->lsm_md_oinfo[i];
1818                         if (oinfo->lmo_mds == op_data->op_mds) {
1819                                 *fid = oinfo->lmo_fid;
1820                                 break;
1821                         }
1822                 }
1823
1824                 /* Hmm, can not find the stripe by mdt_index(op_mds) */
1825                 if (i == lsm->lsm_md_stripe_count)
1826                         tgt = ERR_PTR(-EINVAL);
1827
1828                 return tgt;
1829         }
1830
1831         if (lsm == NULL || op_data->op_namelen == 0) {
1832                 tgt = lmv_find_target(lmv, fid);
1833                 if (IS_ERR(tgt))
1834                         return tgt;
1835
1836                 op_data->op_mds = tgt->ltd_idx;
1837                 return tgt;
1838         }
1839
1840         return lmv_locate_target_for_name(lmv, lsm, op_data->op_name,
1841                                           op_data->op_namelen, fid,
1842                                           &op_data->op_mds);
1843 }
1844
1845 int lmv_create(struct obd_export *exp, struct md_op_data *op_data,
1846                const void *data, int datalen, int mode, __u32 uid,
1847                __u32 gid, cfs_cap_t cap_effective, __u64 rdev,
1848                struct ptlrpc_request **request)
1849 {
1850         struct obd_device       *obd = exp->exp_obd;
1851         struct lmv_obd          *lmv = &obd->u.lmv;
1852         struct lmv_tgt_desc     *tgt;
1853         int                      rc;
1854         ENTRY;
1855
1856         rc = lmv_check_connect(obd);
1857         if (rc)
1858                 RETURN(rc);
1859
1860         if (!lmv->desc.ld_active_tgt_count)
1861                 RETURN(-EIO);
1862
1863         tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
1864         if (IS_ERR(tgt))
1865                 RETURN(PTR_ERR(tgt));
1866
1867         CDEBUG(D_INODE, "CREATE name '%.*s' on "DFID" -> mds #%x\n",
1868                op_data->op_namelen, op_data->op_name, PFID(&op_data->op_fid1),
1869                op_data->op_mds);
1870
1871         rc = lmv_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
1872         if (rc)
1873                 RETURN(rc);
1874
1875         /* Send the create request to the MDT where the object
1876          * will be located */
1877         tgt = lmv_find_target(lmv, &op_data->op_fid2);
1878         if (IS_ERR(tgt))
1879                 RETURN(PTR_ERR(tgt));
1880
1881         op_data->op_mds = tgt->ltd_idx;
1882
1883         CDEBUG(D_INODE, "CREATE obj "DFID" -> mds #%x\n",
1884                PFID(&op_data->op_fid2), op_data->op_mds);
1885
1886         op_data->op_flags |= MF_MDC_CANCEL_FID1;
1887         rc = md_create(tgt->ltd_exp, op_data, data, datalen, mode, uid, gid,
1888                        cap_effective, rdev, request);
1889         if (rc == 0) {
1890                 if (*request == NULL)
1891                         RETURN(rc);
1892                 CDEBUG(D_INODE, "Created - "DFID"\n", PFID(&op_data->op_fid2));
1893         }
1894         RETURN(rc);
1895 }
1896
1897 static int lmv_done_writing(struct obd_export *exp,
1898                             struct md_op_data *op_data,
1899                             struct md_open_data *mod)
1900 {
1901         struct obd_device     *obd = exp->exp_obd;
1902         struct lmv_obd        *lmv = &obd->u.lmv;
1903         struct lmv_tgt_desc   *tgt;
1904         int                    rc;
1905         ENTRY;
1906
1907         rc = lmv_check_connect(obd);
1908         if (rc)
1909                 RETURN(rc);
1910
1911         tgt = lmv_find_target(lmv, &op_data->op_fid1);
1912         if (IS_ERR(tgt))
1913                 RETURN(PTR_ERR(tgt));
1914
1915         rc = md_done_writing(tgt->ltd_exp, op_data, mod);
1916         RETURN(rc);
1917 }
1918
1919 static int
1920 lmv_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
1921             const union ldlm_policy_data *policy,
1922             struct lookup_intent *it, struct md_op_data *op_data,
1923             struct lustre_handle *lockh, __u64 extra_lock_flags)
1924 {
1925         struct obd_device        *obd = exp->exp_obd;
1926         struct lmv_obd           *lmv = &obd->u.lmv;
1927         struct lmv_tgt_desc      *tgt;
1928         int                       rc;
1929         ENTRY;
1930
1931         rc = lmv_check_connect(obd);
1932         if (rc)
1933                 RETURN(rc);
1934
1935         CDEBUG(D_INODE, "ENQUEUE '%s' on "DFID"\n",
1936                LL_IT2STR(it), PFID(&op_data->op_fid1));
1937
1938         tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
1939         if (IS_ERR(tgt))
1940                 RETURN(PTR_ERR(tgt));
1941
1942         CDEBUG(D_INODE, "ENQUEUE '%s' on "DFID" -> mds #%u\n",
1943                LL_IT2STR(it), PFID(&op_data->op_fid1), tgt->ltd_idx);
1944
1945         rc = md_enqueue(tgt->ltd_exp, einfo, policy, it, op_data, lockh,
1946                         extra_lock_flags);
1947
1948         RETURN(rc);
1949 }
1950
1951 static int
1952 lmv_getattr_name(struct obd_export *exp,struct md_op_data *op_data,
1953                  struct ptlrpc_request **preq)
1954 {
1955         struct ptlrpc_request   *req = NULL;
1956         struct obd_device       *obd = exp->exp_obd;
1957         struct lmv_obd          *lmv = &obd->u.lmv;
1958         struct lmv_tgt_desc     *tgt;
1959         struct mdt_body         *body;
1960         int                      rc;
1961         ENTRY;
1962
1963         rc = lmv_check_connect(obd);
1964         if (rc)
1965                 RETURN(rc);
1966
1967         tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
1968         if (IS_ERR(tgt))
1969                 RETURN(PTR_ERR(tgt));
1970
1971         CDEBUG(D_INODE, "GETATTR_NAME for %*s on "DFID" -> mds #%u\n",
1972                op_data->op_namelen, op_data->op_name, PFID(&op_data->op_fid1),
1973                tgt->ltd_idx);
1974
1975         rc = md_getattr_name(tgt->ltd_exp, op_data, preq);
1976         if (rc != 0)
1977                 RETURN(rc);
1978
1979         body = req_capsule_server_get(&(*preq)->rq_pill, &RMF_MDT_BODY);
1980         LASSERT(body != NULL);
1981
1982         if (body->mbo_valid & OBD_MD_MDS) {
1983                 struct lu_fid rid = body->mbo_fid1;
1984                 CDEBUG(D_INODE, "Request attrs for "DFID"\n",
1985                        PFID(&rid));
1986
1987                 tgt = lmv_find_target(lmv, &rid);
1988                 if (IS_ERR(tgt)) {
1989                         ptlrpc_req_finished(*preq);
1990                         preq = NULL;
1991                         RETURN(PTR_ERR(tgt));
1992                 }
1993
1994                 op_data->op_fid1 = rid;
1995                 op_data->op_valid |= OBD_MD_FLCROSSREF;
1996                 op_data->op_namelen = 0;
1997                 op_data->op_name = NULL;
1998                 rc = md_getattr_name(tgt->ltd_exp, op_data, &req);
1999                 ptlrpc_req_finished(*preq);
2000                 *preq = req;
2001         }
2002
2003         RETURN(rc);
2004 }
2005
2006 #define md_op_data_fid(op_data, fl)                     \
2007         (fl == MF_MDC_CANCEL_FID1 ? &op_data->op_fid1 : \
2008          fl == MF_MDC_CANCEL_FID2 ? &op_data->op_fid2 : \
2009          fl == MF_MDC_CANCEL_FID3 ? &op_data->op_fid3 : \
2010          fl == MF_MDC_CANCEL_FID4 ? &op_data->op_fid4 : \
2011          NULL)
2012
2013 static int lmv_early_cancel(struct obd_export *exp, struct lmv_tgt_desc *tgt,
2014                             struct md_op_data *op_data,
2015                             __u32 op_tgt, ldlm_mode_t mode, int bits, int flag)
2016 {
2017         struct lu_fid          *fid = md_op_data_fid(op_data, flag);
2018         struct obd_device      *obd = exp->exp_obd;
2019         struct lmv_obd         *lmv = &obd->u.lmv;
2020         ldlm_policy_data_t      policy = {{ 0 }};
2021         int                     rc = 0;
2022         ENTRY;
2023
2024         if (!fid_is_sane(fid))
2025                 RETURN(0);
2026
2027         if (tgt == NULL) {
2028                 tgt = lmv_find_target(lmv, fid);
2029                 if (IS_ERR(tgt))
2030                         RETURN(PTR_ERR(tgt));
2031         }
2032
2033         if (tgt->ltd_idx != op_tgt) {
2034                 CDEBUG(D_INODE, "EARLY_CANCEL on "DFID"\n", PFID(fid));
2035                 policy.l_inodebits.bits = bits;
2036                 rc = md_cancel_unused(tgt->ltd_exp, fid, &policy,
2037                                       mode, LCF_ASYNC, NULL);
2038         } else {
2039                 CDEBUG(D_INODE,
2040                        "EARLY_CANCEL skip operation target %d on "DFID"\n",
2041                        op_tgt, PFID(fid));
2042                 op_data->op_flags |= flag;
2043                 rc = 0;
2044         }
2045
2046         RETURN(rc);
2047 }
2048
2049 /*
2050  * llite passes fid of an target inode in op_data->op_fid1 and id of directory in
2051  * op_data->op_fid2
2052  */
2053 static int lmv_link(struct obd_export *exp, struct md_op_data *op_data,
2054                     struct ptlrpc_request **request)
2055 {
2056         struct obd_device       *obd = exp->exp_obd;
2057         struct lmv_obd          *lmv = &obd->u.lmv;
2058         struct lmv_tgt_desc     *tgt;
2059         int                      rc;
2060         ENTRY;
2061
2062         rc = lmv_check_connect(obd);
2063         if (rc)
2064                 RETURN(rc);
2065
2066         LASSERT(op_data->op_namelen != 0);
2067
2068         CDEBUG(D_INODE, "LINK "DFID":%*s to "DFID"\n",
2069                PFID(&op_data->op_fid2), op_data->op_namelen,
2070                op_data->op_name, PFID(&op_data->op_fid1));
2071
2072         op_data->op_fsuid = from_kuid(&init_user_ns, current_fsuid());
2073         op_data->op_fsgid = from_kgid(&init_user_ns, current_fsgid());
2074         op_data->op_cap = cfs_curproc_cap_pack();
2075         if (op_data->op_mea2 != NULL) {
2076                 struct lmv_stripe_md    *lsm = op_data->op_mea2;
2077                 const struct lmv_oinfo  *oinfo;
2078
2079                 oinfo = lsm_name_to_stripe_info(lsm, op_data->op_name,
2080                                                 op_data->op_namelen);
2081                 if (IS_ERR(oinfo))
2082                         RETURN(PTR_ERR(oinfo));
2083
2084                 op_data->op_fid2 = oinfo->lmo_fid;
2085         }
2086
2087         tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid2);
2088         if (IS_ERR(tgt))
2089                 RETURN(PTR_ERR(tgt));
2090
2091         /*
2092          * Cancel UPDATE lock on child (fid1).
2093          */
2094         op_data->op_flags |= MF_MDC_CANCEL_FID2;
2095         rc = lmv_early_cancel(exp, NULL, op_data, tgt->ltd_idx, LCK_EX,
2096                               MDS_INODELOCK_UPDATE, MF_MDC_CANCEL_FID1);
2097         if (rc != 0)
2098                 RETURN(rc);
2099
2100         rc = md_link(tgt->ltd_exp, op_data, request);
2101
2102         RETURN(rc);
2103 }
2104
2105 static int lmv_rename(struct obd_export *exp, struct md_op_data *op_data,
2106                       const char *old, int oldlen, const char *new, int newlen,
2107                       struct ptlrpc_request **request)
2108 {
2109         struct obd_device       *obd = exp->exp_obd;
2110         struct lmv_obd          *lmv = &obd->u.lmv;
2111         struct lmv_tgt_desc     *src_tgt;
2112         int                     rc;
2113         ENTRY;
2114
2115         LASSERT(oldlen != 0);
2116
2117         CDEBUG(D_INODE, "RENAME %.*s in "DFID":%d to %.*s in "DFID":%d\n",
2118                oldlen, old, PFID(&op_data->op_fid1),
2119                op_data->op_mea1 ? op_data->op_mea1->lsm_md_stripe_count : 0,
2120                newlen, new, PFID(&op_data->op_fid2),
2121                op_data->op_mea2 ? op_data->op_mea2->lsm_md_stripe_count : 0);
2122
2123         rc = lmv_check_connect(obd);
2124         if (rc)
2125                 RETURN(rc);
2126
2127         op_data->op_fsuid = from_kuid(&init_user_ns, current_fsuid());
2128         op_data->op_fsgid = from_kgid(&init_user_ns, current_fsgid());
2129         op_data->op_cap = cfs_curproc_cap_pack();
2130         if (op_data->op_cli_flags & CLI_MIGRATE) {
2131                 LASSERTF(fid_is_sane(&op_data->op_fid3), "invalid FID "DFID"\n",
2132                          PFID(&op_data->op_fid3));
2133                 rc = lmv_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
2134                 if (rc)
2135                         RETURN(rc);
2136                 src_tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid3);
2137         } else {
2138                 if (op_data->op_mea1 != NULL) {
2139                         struct lmv_stripe_md    *lsm = op_data->op_mea1;
2140
2141                         src_tgt = lmv_locate_target_for_name(lmv, lsm, old,
2142                                                              oldlen,
2143                                                              &op_data->op_fid1,
2144                                                              &op_data->op_mds);
2145                         if (IS_ERR(src_tgt))
2146                                 RETURN(PTR_ERR(src_tgt));
2147                 } else {
2148                         src_tgt = lmv_find_target(lmv, &op_data->op_fid1);
2149                         if (IS_ERR(src_tgt))
2150                                 RETURN(PTR_ERR(src_tgt));
2151
2152                         op_data->op_mds = src_tgt->ltd_idx;
2153                 }
2154
2155                 if (op_data->op_mea2) {
2156                         struct lmv_stripe_md    *lsm = op_data->op_mea2;
2157                         const struct lmv_oinfo  *oinfo;
2158
2159                         oinfo = lsm_name_to_stripe_info(lsm, new, newlen);
2160                         if (IS_ERR(oinfo))
2161                                 RETURN(PTR_ERR(oinfo));
2162
2163                         op_data->op_fid2 = oinfo->lmo_fid;
2164                 }
2165         }
2166         if (IS_ERR(src_tgt))
2167                 RETURN(PTR_ERR(src_tgt));
2168
2169         /*
2170          * LOOKUP lock on src child (fid3) should also be cancelled for
2171          * src_tgt in mdc_rename.
2172          */
2173         op_data->op_flags |= MF_MDC_CANCEL_FID1 | MF_MDC_CANCEL_FID3;
2174
2175         /*
2176          * Cancel UPDATE locks on tgt parent (fid2), tgt_tgt is its
2177          * own target.
2178          */
2179         rc = lmv_early_cancel(exp, NULL, op_data, src_tgt->ltd_idx,
2180                               LCK_EX, MDS_INODELOCK_UPDATE,
2181                               MF_MDC_CANCEL_FID2);
2182
2183         if (rc != 0)
2184                 RETURN(rc);
2185         /*
2186          * Cancel LOOKUP locks on source child (fid3) for parent tgt_tgt.
2187          */
2188         if (fid_is_sane(&op_data->op_fid3)) {
2189                 struct lmv_tgt_desc *tgt;
2190
2191                 tgt = lmv_find_target(lmv, &op_data->op_fid1);
2192                 if (IS_ERR(tgt))
2193                         RETURN(PTR_ERR(tgt));
2194
2195                 /* Cancel LOOKUP lock on its parent */
2196                 rc = lmv_early_cancel(exp, tgt, op_data, src_tgt->ltd_idx,
2197                                       LCK_EX, MDS_INODELOCK_LOOKUP,
2198                                       MF_MDC_CANCEL_FID3);
2199                 if (rc != 0)
2200                         RETURN(rc);
2201
2202                 rc = lmv_early_cancel(exp, NULL, op_data, src_tgt->ltd_idx,
2203                                       LCK_EX, MDS_INODELOCK_FULL,
2204                                       MF_MDC_CANCEL_FID3);
2205                 if (rc != 0)
2206                         RETURN(rc);
2207         }
2208
2209         /*
2210          * Cancel all the locks on tgt child (fid4).
2211          */
2212         if (fid_is_sane(&op_data->op_fid4))
2213                 rc = lmv_early_cancel(exp, NULL, op_data, src_tgt->ltd_idx,
2214                                       LCK_EX, MDS_INODELOCK_FULL,
2215                                       MF_MDC_CANCEL_FID4);
2216
2217         CDEBUG(D_INODE, DFID":m%d to "DFID"\n", PFID(&op_data->op_fid1),
2218                op_data->op_mds, PFID(&op_data->op_fid2));
2219
2220         rc = md_rename(src_tgt->ltd_exp, op_data, old, oldlen, new, newlen,
2221                        request);
2222
2223         RETURN(rc);
2224 }
2225
2226 static int lmv_setattr(struct obd_export *exp, struct md_op_data *op_data,
2227                        void *ea, int ealen, void *ea2, int ea2len,
2228                        struct ptlrpc_request **request,
2229                        struct md_open_data **mod)
2230 {
2231         struct obd_device       *obd = exp->exp_obd;
2232         struct lmv_obd          *lmv = &obd->u.lmv;
2233         struct lmv_tgt_desc     *tgt;
2234         int                      rc = 0;
2235         ENTRY;
2236
2237         rc = lmv_check_connect(obd);
2238         if (rc)
2239                 RETURN(rc);
2240
2241         CDEBUG(D_INODE, "SETATTR for "DFID", valid 0x%x\n",
2242                PFID(&op_data->op_fid1), op_data->op_attr.ia_valid);
2243
2244         op_data->op_flags |= MF_MDC_CANCEL_FID1;
2245         tgt = lmv_find_target(lmv, &op_data->op_fid1);
2246         if (IS_ERR(tgt))
2247                 RETURN(PTR_ERR(tgt));
2248
2249         rc = md_setattr(tgt->ltd_exp, op_data, ea, ealen, ea2,
2250                         ea2len, request, mod);
2251
2252         RETURN(rc);
2253 }
2254
2255 static int lmv_fsync(struct obd_export *exp, const struct lu_fid *fid,
2256                      struct obd_capa *oc, struct ptlrpc_request **request)
2257 {
2258         struct obd_device       *obd = exp->exp_obd;
2259         struct lmv_obd          *lmv = &obd->u.lmv;
2260         struct lmv_tgt_desc     *tgt;
2261         int                      rc;
2262         ENTRY;
2263
2264         rc = lmv_check_connect(obd);
2265         if (rc != 0)
2266                 RETURN(rc);
2267
2268         tgt = lmv_find_target(lmv, fid);
2269         if (IS_ERR(tgt))
2270                 RETURN(PTR_ERR(tgt));
2271
2272         rc = md_fsync(tgt->ltd_exp, fid, oc, request);
2273         RETURN(rc);
2274 }
2275
2276 /**
2277  * Get current minimum entry from striped directory
2278  *
2279  * This function will search the dir entry, whose hash value is the
2280  * closest(>=) to @hash_offset, from all of sub-stripes, and it is
2281  * only being called for striped directory.
2282  *
2283  * \param[in] exp               export of LMV
2284  * \param[in] op_data           parameters transferred beween client MD stack
2285  *                              stripe_information will be included in this
2286  *                              parameter
2287  * \param[in] cb_op             ldlm callback being used in enqueue in
2288  *                              mdc_read_page
2289  * \param[in] hash_offset       the hash value, which is used to locate
2290  *                              minum(closet) dir entry
2291  * \param[in|out] stripe_offset the caller use this to indicate the stripe
2292  *                              index of last entry, so to avoid hash conflict
2293  *                              between stripes. It will also be used to
2294  *                              return the stripe index of current dir entry.
2295  * \param[in|out] entp          the minum entry and it also is being used
2296  *                              to input the last dir entry to resolve the
2297  *                              hash conflict
2298  *
2299  * \param[out] ppage            the page which holds the minum entry
2300  *
2301  * \retval                      = 0 get the entry successfully
2302  *                              negative errno (< 0) does not get the entry
2303  */
2304 static int lmv_get_min_striped_entry(struct obd_export *exp,
2305                                      struct md_op_data *op_data,
2306                                      struct md_callback *cb_op,
2307                                      __u64 hash_offset, int *stripe_offset,
2308                                      struct lu_dirent **entp,
2309                                      struct page **ppage)
2310 {
2311         struct obd_device       *obd = exp->exp_obd;
2312         struct lmv_obd          *lmv = &obd->u.lmv;
2313         struct lmv_stripe_md    *lsm = op_data->op_mea1;
2314         struct lmv_tgt_desc     *tgt;
2315         int                     stripe_count;
2316         struct lu_dirent        *min_ent = NULL;
2317         struct page             *min_page = NULL;
2318         int                     min_idx = 0;
2319         int                     i;
2320         int                     rc = 0;
2321         ENTRY;
2322
2323         stripe_count = lsm->lsm_md_stripe_count;
2324         for (i = 0; i < stripe_count; i++) {
2325                 struct lu_dirent        *ent = NULL;
2326                 struct page             *page = NULL;
2327                 struct lu_dirpage       *dp;
2328                 __u64                   stripe_hash = hash_offset;
2329
2330                 tgt = lmv_get_target(lmv, lsm->lsm_md_oinfo[i].lmo_mds, NULL);
2331                 if (IS_ERR(tgt))
2332                         GOTO(out, rc = PTR_ERR(tgt));
2333
2334                 /* op_data will be shared by each stripe, so we need
2335                  * reset these value for each stripe */
2336                 op_data->op_stripe_offset = i;
2337                 op_data->op_fid1 = lsm->lsm_md_oinfo[i].lmo_fid;
2338                 op_data->op_fid2 = lsm->lsm_md_oinfo[i].lmo_fid;
2339                 op_data->op_data = lsm->lsm_md_oinfo[i].lmo_root;
2340 next:
2341                 rc = md_read_page(tgt->ltd_exp, op_data, cb_op, stripe_hash,
2342                                   &page);
2343                 if (rc != 0)
2344                         GOTO(out, rc);
2345
2346                 dp = page_address(page);
2347                 for (ent = lu_dirent_start(dp); ent != NULL;
2348                      ent = lu_dirent_next(ent)) {
2349                         /* Skip dummy entry */
2350                         if (le16_to_cpu(ent->lde_namelen) == 0)
2351                                 continue;
2352
2353                         if (le64_to_cpu(ent->lde_hash) < hash_offset)
2354                                 continue;
2355
2356                         if (le64_to_cpu(ent->lde_hash) == hash_offset &&
2357                             (*entp == ent || i < *stripe_offset))
2358                                 continue;
2359
2360                         /* skip . and .. for other stripes */
2361                         if (i != 0 &&
2362                             (strncmp(ent->lde_name, ".",
2363                                      le16_to_cpu(ent->lde_namelen)) == 0 ||
2364                              strncmp(ent->lde_name, "..",
2365                                      le16_to_cpu(ent->lde_namelen)) == 0))
2366                                 continue;
2367                         break;
2368                 }
2369
2370                 if (ent == NULL) {
2371                         stripe_hash = le64_to_cpu(dp->ldp_hash_end);
2372
2373                         kunmap(page);
2374                         page_cache_release(page);
2375                         page = NULL;
2376
2377                         /* reach the end of current stripe, go to next stripe */
2378                         if (stripe_hash == MDS_DIR_END_OFF)
2379                                 continue;
2380                         else
2381                                 goto next;
2382                 }
2383
2384                 if (min_ent != NULL) {
2385                         if (le64_to_cpu(min_ent->lde_hash) >
2386                             le64_to_cpu(ent->lde_hash)) {
2387                                 min_ent = ent;
2388                                 kunmap(min_page);
2389                                 page_cache_release(min_page);
2390                                 min_idx = i;
2391                                 min_page = page;
2392                         } else {
2393                                 kunmap(page);
2394                                 page_cache_release(page);
2395                                 page = NULL;
2396                         }
2397                 } else {
2398                         min_ent = ent;
2399                         min_page = page;
2400                         min_idx = i;
2401                 }
2402         }
2403
2404 out:
2405         if (*ppage != NULL) {
2406                 kunmap(*ppage);
2407                 page_cache_release(*ppage);
2408         }
2409         *stripe_offset = min_idx;
2410         *entp = min_ent;
2411         *ppage = min_page;
2412         RETURN(rc);
2413 }
2414
2415 /**
2416  * Build dir entry page from a striped directory
2417  *
2418  * This function gets one entry by @offset from a striped directory. It will
2419  * read entries from all of stripes, and choose one closest to the required
2420  * offset(&offset). A few notes
2421  * 1. skip . and .. for non-zero stripes, because there can only have one .
2422  * and .. in a directory.
2423  * 2. op_data will be shared by all of stripes, instead of allocating new
2424  * one, so need to restore before reusing.
2425  * 3. release the entry page if that is not being chosen.
2426  *
2427  * \param[in] exp       obd export refer to LMV
2428  * \param[in] op_data   hold those MD parameters of read_entry
2429  * \param[in] cb_op     ldlm callback being used in enqueue in mdc_read_entry
2430  * \param[out] ldp      the entry being read
2431  * \param[out] ppage    the page holding the entry. Note: because the entry
2432  *                      will be accessed in upper layer, so we need hold the
2433  *                      page until the usages of entry is finished, see
2434  *                      ll_dir_entry_next.
2435  *
2436  * retval               =0 if get entry successfully
2437  *                      <0 cannot get entry
2438  */
2439 static int lmv_read_striped_page(struct obd_export *exp,
2440                                  struct md_op_data *op_data,
2441                                  struct md_callback *cb_op,
2442                                  __u64 offset, struct page **ppage)
2443 {
2444         struct obd_device       *obd = exp->exp_obd;
2445         struct lu_fid           master_fid = op_data->op_fid1;
2446         struct inode            *master_inode = op_data->op_data;
2447         __u64                   hash_offset = offset;
2448         struct lu_dirpage       *dp;
2449         struct page             *min_ent_page = NULL;
2450         struct page             *ent_page = NULL;
2451         struct lu_dirent        *ent;
2452         void                    *area;
2453         int                     ent_idx = 0;
2454         struct lu_dirent        *min_ent = NULL;
2455         struct lu_dirent        *last_ent;
2456         int                     left_bytes;
2457         int                     rc;
2458         ENTRY;
2459
2460         rc = lmv_check_connect(obd);
2461         if (rc)
2462                 RETURN(rc);
2463
2464         /* Allocate a page and read entries from all of stripes and fill
2465          * the page by hash order */
2466         ent_page = alloc_page(GFP_KERNEL);
2467         if (ent_page == NULL)
2468                 RETURN(-ENOMEM);
2469
2470         /* Initialize the entry page */
2471         dp = kmap(ent_page);
2472         memset(dp, 0, sizeof(*dp));
2473         dp->ldp_hash_start = cpu_to_le64(offset);
2474         dp->ldp_flags |= LDF_COLLIDE;
2475
2476         area = dp + 1;
2477         left_bytes = PAGE_CACHE_SIZE - sizeof(*dp);
2478         ent = area;
2479         last_ent = ent;
2480         do {
2481                 __u16   ent_size;
2482
2483                 /* Find the minum entry from all sub-stripes */
2484                 rc = lmv_get_min_striped_entry(exp, op_data, cb_op, hash_offset,
2485                                                &ent_idx, &min_ent,
2486                                                &min_ent_page);
2487                 if (rc != 0)
2488                         GOTO(out, rc);
2489
2490                 /* If it can not get minum entry, it means it already reaches
2491                  * the end of this directory */
2492                 if (min_ent == NULL) {
2493                         last_ent->lde_reclen = 0;
2494                         hash_offset = MDS_DIR_END_OFF;
2495                         GOTO(out, rc);
2496                 }
2497
2498                 ent_size = le16_to_cpu(min_ent->lde_reclen);
2499
2500                 /* the last entry lde_reclen is 0, but it might not
2501                  * the end of this entry of this temporay entry */
2502                 if (ent_size == 0)
2503                         ent_size = lu_dirent_calc_size(
2504                                         le16_to_cpu(min_ent->lde_namelen),
2505                                         le32_to_cpu(min_ent->lde_attrs));
2506                 if (ent_size > left_bytes) {
2507                         last_ent->lde_reclen = cpu_to_le16(0);
2508                         hash_offset = le64_to_cpu(min_ent->lde_hash);
2509                         GOTO(out, rc);
2510                 }
2511
2512                 memcpy(ent, min_ent, ent_size);
2513
2514                 /* Replace . with master FID and Replace .. with the parent FID
2515                  * of master object */
2516                 if (strncmp(ent->lde_name, ".",
2517                             le16_to_cpu(ent->lde_namelen)) == 0 &&
2518                     le16_to_cpu(ent->lde_namelen) == 1)
2519                         fid_cpu_to_le(&ent->lde_fid, &master_fid);
2520                 else if (strncmp(ent->lde_name, "..",
2521                                    le16_to_cpu(ent->lde_namelen)) == 0 &&
2522                            le16_to_cpu(ent->lde_namelen) == 2)
2523                         fid_cpu_to_le(&ent->lde_fid, &op_data->op_fid3);
2524
2525                 left_bytes -= ent_size;
2526                 ent->lde_reclen = cpu_to_le16(ent_size);
2527                 last_ent = ent;
2528                 ent = (void *)ent + ent_size;
2529                 hash_offset = le64_to_cpu(min_ent->lde_hash);
2530                 if (hash_offset == MDS_DIR_END_OFF) {
2531                         last_ent->lde_reclen = 0;
2532                         break;
2533                 }
2534         } while (1);
2535 out:
2536         if (min_ent_page != NULL) {
2537                 kunmap(min_ent_page);
2538                 page_cache_release(min_ent_page);
2539         }
2540
2541         if (unlikely(rc != 0)) {
2542                 __free_page(ent_page);
2543                 ent_page = NULL;
2544         } else {
2545                 if (ent == area)
2546                         dp->ldp_flags |= LDF_EMPTY;
2547                 dp->ldp_flags = cpu_to_le32(dp->ldp_flags);
2548                 dp->ldp_hash_end = cpu_to_le64(hash_offset);
2549         }
2550
2551         /* We do not want to allocate md_op_data during each
2552          * dir entry reading, so op_data will be shared by every stripe,
2553          * then we need to restore it back to original value before
2554          * return to the upper layer */
2555         op_data->op_fid1 = master_fid;
2556         op_data->op_fid2 = master_fid;
2557         op_data->op_data = master_inode;
2558
2559         *ppage = ent_page;
2560
2561         RETURN(rc);
2562 }
2563
2564 int lmv_read_page(struct obd_export *exp, struct md_op_data *op_data,
2565                   struct md_callback *cb_op, __u64 offset,
2566                   struct page **ppage)
2567 {
2568         struct obd_device       *obd = exp->exp_obd;
2569         struct lmv_obd          *lmv = &obd->u.lmv;
2570         struct lmv_stripe_md    *lsm = op_data->op_mea1;
2571         struct lmv_tgt_desc     *tgt;
2572         int                     rc;
2573         ENTRY;
2574
2575         rc = lmv_check_connect(obd);
2576         if (rc != 0)
2577                 RETURN(rc);
2578
2579         if (unlikely(lsm != NULL)) {
2580                 rc = lmv_read_striped_page(exp, op_data, cb_op, offset, ppage);
2581                 RETURN(rc);
2582         }
2583
2584         tgt = lmv_find_target(lmv, &op_data->op_fid1);
2585         if (IS_ERR(tgt))
2586                 RETURN(PTR_ERR(tgt));
2587
2588         rc = md_read_page(tgt->ltd_exp, op_data, cb_op, offset, ppage);
2589
2590         RETURN(rc);
2591 }
2592
2593 /**
2594  * Unlink a file/directory
2595  *
2596  * Unlink a file or directory under the parent dir. The unlink request
2597  * usually will be sent to the MDT where the child is located, but if
2598  * the client does not have the child FID then request will be sent to the
2599  * MDT where the parent is located.
2600  *
2601  * If the parent is a striped directory then it also needs to locate which
2602  * stripe the name of the child is located, and replace the parent FID
2603  * (@op->op_fid1) with the stripe FID. Note: if the stripe is unknown,
2604  * it will walk through all of sub-stripes until the child is being
2605  * unlinked finally.
2606  *
2607  * \param[in] exp       export refer to LMV
2608  * \param[in] op_data   different parameters transferred beween client
2609  *                      MD stacks, name, namelen, FIDs etc.
2610  *                      op_fid1 is the parent FID, op_fid2 is the child
2611  *                      FID.
2612  * \param[out] request  point to the request of unlink.
2613  *
2614  * retval               0 if succeed
2615  *                      negative errno if failed.
2616  */
2617 static int lmv_unlink(struct obd_export *exp, struct md_op_data *op_data,
2618                       struct ptlrpc_request **request)
2619 {
2620         struct obd_device       *obd = exp->exp_obd;
2621         struct lmv_obd          *lmv = &obd->u.lmv;
2622         struct lmv_tgt_desc     *tgt = NULL;
2623         struct lmv_tgt_desc     *parent_tgt = NULL;
2624         struct mdt_body         *body;
2625         int                     rc;
2626         int                     stripe_index = 0;
2627         struct lmv_stripe_md    *lsm = op_data->op_mea1;
2628         ENTRY;
2629
2630         rc = lmv_check_connect(obd);
2631         if (rc)
2632                 RETURN(rc);
2633 retry_unlink:
2634         /* For striped dir, we need to locate the parent as well */
2635         if (lsm != NULL) {
2636                 struct lmv_tgt_desc *tmp;
2637
2638                 LASSERT(op_data->op_name != NULL &&
2639                         op_data->op_namelen != 0);
2640
2641                 tmp = lmv_locate_target_for_name(lmv, lsm,
2642                                                  op_data->op_name,
2643                                                  op_data->op_namelen,
2644                                                  &op_data->op_fid1,
2645                                                  &op_data->op_mds);
2646
2647                 /* return -EBADFD means unknown hash type, might
2648                  * need try all sub-stripe here */
2649                 if (IS_ERR(tmp) && PTR_ERR(tmp) != -EBADFD)
2650                         RETURN(PTR_ERR(tmp));
2651
2652                 /* Note: both migrating dir and unknown hash dir need to
2653                  * try all of sub-stripes, so we need start search the
2654                  * name from stripe 0, but migrating dir is already handled
2655                  * inside lmv_locate_target_for_name(), so we only check
2656                  * unknown hash type directory here */
2657                 if (!lmv_is_known_hash_type(lsm)) {
2658                         struct lmv_oinfo *oinfo;
2659
2660                         oinfo = &lsm->lsm_md_oinfo[stripe_index];
2661
2662                         op_data->op_fid1 = oinfo->lmo_fid;
2663                         op_data->op_mds = oinfo->lmo_mds;
2664                 }
2665         }
2666
2667 try_next_stripe:
2668         /* Send unlink requests to the MDT where the child is located */
2669         if (likely(!fid_is_zero(&op_data->op_fid2)))
2670                 tgt = lmv_find_target(lmv, &op_data->op_fid2);
2671         else if (lsm != NULL)
2672                 tgt = lmv_get_target(lmv, op_data->op_mds, NULL);
2673         else
2674                 tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
2675
2676         if (IS_ERR(tgt))
2677                 RETURN(PTR_ERR(tgt));
2678
2679         op_data->op_fsuid = from_kuid(&init_user_ns, current_fsuid());
2680         op_data->op_fsgid = from_kgid(&init_user_ns, current_fsgid());
2681         op_data->op_cap = cfs_curproc_cap_pack();
2682
2683         /*
2684          * If child's fid is given, cancel unused locks for it if it is from
2685          * another export than parent.
2686          *
2687          * LOOKUP lock for child (fid3) should also be cancelled on parent
2688          * tgt_tgt in mdc_unlink().
2689          */
2690         op_data->op_flags |= MF_MDC_CANCEL_FID1 | MF_MDC_CANCEL_FID3;
2691
2692         /*
2693          * Cancel FULL locks on child (fid3).
2694          */
2695         parent_tgt = lmv_find_target(lmv, &op_data->op_fid1);
2696         if (IS_ERR(parent_tgt))
2697                 RETURN(PTR_ERR(parent_tgt));
2698
2699         if (parent_tgt != tgt) {
2700                 rc = lmv_early_cancel(exp, parent_tgt, op_data, tgt->ltd_idx,
2701                                       LCK_EX, MDS_INODELOCK_LOOKUP,
2702                                       MF_MDC_CANCEL_FID3);
2703         }
2704
2705         rc = lmv_early_cancel(exp, NULL, op_data, tgt->ltd_idx, LCK_EX,
2706                               MDS_INODELOCK_FULL, MF_MDC_CANCEL_FID3);
2707         if (rc != 0)
2708                 RETURN(rc);
2709
2710         CDEBUG(D_INODE, "unlink with fid="DFID"/"DFID" -> mds #%u\n",
2711                PFID(&op_data->op_fid1), PFID(&op_data->op_fid2), tgt->ltd_idx);
2712
2713         rc = md_unlink(tgt->ltd_exp, op_data, request);
2714         if (rc != 0 && rc != -EREMOTE && rc != -ENOENT)
2715                 RETURN(rc);
2716
2717         /* Try next stripe if it is needed. */
2718         if (rc == -ENOENT && lsm != NULL && lmv_need_try_all_stripes(lsm)) {
2719                 struct lmv_oinfo *oinfo;
2720
2721                 stripe_index++;
2722                 if (stripe_index >= lsm->lsm_md_stripe_count)
2723                         RETURN(rc);
2724
2725                 oinfo = &lsm->lsm_md_oinfo[stripe_index];
2726
2727                 op_data->op_fid1 = oinfo->lmo_fid;
2728                 op_data->op_mds = oinfo->lmo_mds;
2729
2730                 ptlrpc_req_finished(*request);
2731                 *request = NULL;
2732
2733                 goto try_next_stripe;
2734         }
2735
2736         body = req_capsule_server_get(&(*request)->rq_pill, &RMF_MDT_BODY);
2737         if (body == NULL)
2738                 RETURN(-EPROTO);
2739
2740         /* Not cross-ref case, just get out of here. */
2741         if (likely(!(body->mbo_valid & OBD_MD_MDS)))
2742                 RETURN(rc);
2743
2744         CDEBUG(D_INODE, "%s: try unlink to another MDT for "DFID"\n",
2745                exp->exp_obd->obd_name, PFID(&body->mbo_fid1));
2746
2747         /* This is a remote object, try remote MDT, Note: it may
2748          * try more than 1 time here, Considering following case
2749          * /mnt/lustre is root on MDT0, remote1 is on MDT1
2750          * 1. Initially A does not know where remote1 is, it send
2751          *    unlink RPC to MDT0, MDT0 return -EREMOTE, it will
2752          *    resend unlink RPC to MDT1 (retry 1st time).
2753          *
2754          * 2. During the unlink RPC in flight,
2755          *    client B mv /mnt/lustre/remote1 /mnt/lustre/remote2
2756          *    and create new remote1, but on MDT0
2757          *
2758          * 3. MDT1 get unlink RPC(from A), then do remote lock on
2759          *    /mnt/lustre, then lookup get fid of remote1, and find
2760          *    it is remote dir again, and replay -EREMOTE again.
2761          *
2762          * 4. Then A will resend unlink RPC to MDT0. (retry 2nd times).
2763          *
2764          * In theory, it might try unlimited time here, but it should
2765          * be very rare case.  */
2766         op_data->op_fid2 = body->mbo_fid1;
2767         ptlrpc_req_finished(*request);
2768         *request = NULL;
2769
2770         goto retry_unlink;
2771 }
2772
2773 static int lmv_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
2774 {
2775         struct lmv_obd *lmv = &obd->u.lmv;
2776         int rc = 0;
2777
2778         switch (stage) {
2779         case OBD_CLEANUP_EARLY:
2780                 /* XXX: here should be calling obd_precleanup() down to
2781                  * stack. */
2782                 break;
2783         case OBD_CLEANUP_EXPORTS:
2784                 fld_client_proc_fini(&lmv->lmv_fld);
2785                 lprocfs_obd_cleanup(obd);
2786                 lprocfs_free_md_stats(obd);
2787                 break;
2788         default:
2789                 break;
2790         }
2791         RETURN(rc);
2792 }
2793
2794 static int lmv_get_info(const struct lu_env *env, struct obd_export *exp,
2795                         __u32 keylen, void *key, __u32 *vallen, void *val,
2796                         struct lov_stripe_md *lsm)
2797 {
2798         struct obd_device       *obd;
2799         struct lmv_obd          *lmv;
2800         int                      rc = 0;
2801         ENTRY;
2802
2803         obd = class_exp2obd(exp);
2804         if (obd == NULL) {
2805                 CDEBUG(D_IOCTL, "Invalid client cookie "LPX64"\n",
2806                        exp->exp_handle.h_cookie);
2807                 RETURN(-EINVAL);
2808         }
2809
2810         lmv = &obd->u.lmv;
2811         if (keylen >= strlen("remote_flag") && !strcmp(key, "remote_flag")) {
2812                 int i;
2813
2814                 rc = lmv_check_connect(obd);
2815                 if (rc)
2816                         RETURN(rc);
2817
2818                 LASSERT(*vallen == sizeof(__u32));
2819                 for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
2820                         struct lmv_tgt_desc *tgt = lmv->tgts[i];
2821                         /*
2822                          * All tgts should be connected when this gets called.
2823                          */
2824                         if (tgt == NULL || tgt->ltd_exp == NULL)
2825                                 continue;
2826
2827                         if (!obd_get_info(env, tgt->ltd_exp, keylen, key,
2828                                           vallen, val, NULL))
2829                                 RETURN(0);
2830                 }
2831                 RETURN(-EINVAL);
2832         } else if (KEY_IS(KEY_MAX_EASIZE) ||
2833                    KEY_IS(KEY_DEFAULT_EASIZE) ||
2834                    KEY_IS(KEY_MAX_COOKIESIZE) ||
2835                    KEY_IS(KEY_DEFAULT_COOKIESIZE) ||
2836                    KEY_IS(KEY_CONN_DATA)) {
2837                 rc = lmv_check_connect(obd);
2838                 if (rc)
2839                         RETURN(rc);
2840
2841                 /*
2842                  * Forwarding this request to first MDS, it should know LOV
2843                  * desc.
2844                  */
2845                 rc = obd_get_info(env, lmv->tgts[0]->ltd_exp, keylen, key,
2846                                   vallen, val, NULL);
2847                 if (!rc && KEY_IS(KEY_CONN_DATA))
2848                         exp->exp_connect_data = *(struct obd_connect_data *)val;
2849                 RETURN(rc);
2850         } else if (KEY_IS(KEY_TGT_COUNT)) {
2851                 *((int *)val) = lmv->desc.ld_tgt_count;
2852                 RETURN(0);
2853         }
2854
2855         CDEBUG(D_IOCTL, "Invalid key\n");
2856         RETURN(-EINVAL);
2857 }
2858
2859 int lmv_set_info_async(const struct lu_env *env, struct obd_export *exp,
2860                        obd_count keylen, void *key, obd_count vallen,
2861                        void *val, struct ptlrpc_request_set *set)
2862 {
2863         struct lmv_tgt_desc    *tgt = NULL;
2864         struct obd_device      *obd;
2865         struct lmv_obd         *lmv;
2866         int rc = 0;
2867         ENTRY;
2868
2869         obd = class_exp2obd(exp);
2870         if (obd == NULL) {
2871                 CDEBUG(D_IOCTL, "Invalid client cookie "LPX64"\n",
2872                        exp->exp_handle.h_cookie);
2873                 RETURN(-EINVAL);
2874         }
2875         lmv = &obd->u.lmv;
2876
2877         if (KEY_IS(KEY_READ_ONLY) || KEY_IS(KEY_FLUSH_CTX)) {
2878                 int i, err = 0;
2879
2880                 for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
2881                         tgt = lmv->tgts[i];
2882
2883                         if (tgt == NULL || tgt->ltd_exp == NULL)
2884                                 continue;
2885
2886                         err = obd_set_info_async(env, tgt->ltd_exp,
2887                                                  keylen, key, vallen, val, set);
2888                         if (err && rc == 0)
2889                                 rc = err;
2890                 }
2891
2892                 RETURN(rc);
2893         }
2894
2895         RETURN(-EINVAL);
2896 }
2897
2898 static int lmv_pack_md_v1(const struct lmv_stripe_md *lsm,
2899                           struct lmv_mds_md_v1 *lmm1)
2900 {
2901         int     cplen;
2902         int     i;
2903
2904         lmm1->lmv_magic = cpu_to_le32(lsm->lsm_md_magic);
2905         lmm1->lmv_stripe_count = cpu_to_le32(lsm->lsm_md_stripe_count);
2906         lmm1->lmv_master_mdt_index = cpu_to_le32(lsm->lsm_md_master_mdt_index);
2907         lmm1->lmv_hash_type = cpu_to_le32(lsm->lsm_md_hash_type);
2908         cplen = strlcpy(lmm1->lmv_pool_name, lsm->lsm_md_pool_name,
2909                         sizeof(lmm1->lmv_pool_name));
2910         if (cplen >= sizeof(lmm1->lmv_pool_name))
2911                 return -E2BIG;
2912
2913         for (i = 0; i < lsm->lsm_md_stripe_count; i++)
2914                 fid_cpu_to_le(&lmm1->lmv_stripe_fids[i],
2915                               &lsm->lsm_md_oinfo[i].lmo_fid);
2916         return 0;
2917 }
2918
2919 int lmv_pack_md(union lmv_mds_md **lmmp, const struct lmv_stripe_md *lsm,
2920                 int stripe_count)
2921 {
2922         int     lmm_size = 0;
2923         bool    allocated = false;
2924         int     rc = 0;
2925         ENTRY;
2926
2927         LASSERT(lmmp != NULL);
2928         /* Free lmm */
2929         if (*lmmp != NULL && lsm == NULL) {
2930                 int stripe_count;
2931
2932                 stripe_count = lmv_mds_md_stripe_count_get(*lmmp);
2933                 lmm_size = lmv_mds_md_size(stripe_count,
2934                                            le32_to_cpu((*lmmp)->lmv_magic));
2935                 if (lmm_size == 0)
2936                         RETURN(-EINVAL);
2937                 OBD_FREE(*lmmp, lmm_size);
2938                 *lmmp = NULL;
2939                 RETURN(0);
2940         }
2941
2942         /* Alloc lmm */
2943         if (*lmmp == NULL && lsm == NULL) {
2944                 lmm_size = lmv_mds_md_size(stripe_count, LMV_MAGIC);
2945                 LASSERT(lmm_size > 0);
2946                 OBD_ALLOC(*lmmp, lmm_size);
2947                 if (*lmmp == NULL)
2948                         RETURN(-ENOMEM);
2949                 lmv_mds_md_stripe_count_set(*lmmp, stripe_count);
2950                 (*lmmp)->lmv_magic = cpu_to_le32(LMV_MAGIC);
2951                 RETURN(lmm_size);
2952         }
2953
2954         /* pack lmm */
2955         LASSERT(lsm != NULL);
2956         lmm_size = lmv_mds_md_size(lsm->lsm_md_stripe_count, lsm->lsm_md_magic);
2957         if (*lmmp == NULL) {
2958                 OBD_ALLOC(*lmmp, lmm_size);
2959                 if (*lmmp == NULL)
2960                         RETURN(-ENOMEM);
2961                 allocated = true;
2962         }
2963
2964         switch (lsm->lsm_md_magic) {
2965         case LMV_MAGIC_V1:
2966                 rc = lmv_pack_md_v1(lsm, &(*lmmp)->lmv_md_v1);
2967                 break;
2968         default:
2969                 rc = -EINVAL;
2970                 break;
2971         }
2972
2973         if (rc != 0 && allocated) {
2974                 OBD_FREE(*lmmp, lmm_size);
2975                 *lmmp = NULL;
2976         }
2977
2978         RETURN(lmm_size);
2979 }
2980 EXPORT_SYMBOL(lmv_pack_md);
2981
2982 static int lmv_unpack_md_v1(struct obd_export *exp, struct lmv_stripe_md *lsm,
2983                             const struct lmv_mds_md_v1 *lmm1)
2984 {
2985         struct lmv_obd  *lmv = &exp->exp_obd->u.lmv;
2986         int             stripe_count;
2987         int             cplen;
2988         int             i;
2989         int             rc = 0;
2990         ENTRY;
2991
2992         lsm->lsm_md_magic = le32_to_cpu(lmm1->lmv_magic);
2993         lsm->lsm_md_stripe_count = le32_to_cpu(lmm1->lmv_stripe_count);
2994         lsm->lsm_md_master_mdt_index = le32_to_cpu(lmm1->lmv_master_mdt_index);
2995         if (OBD_FAIL_CHECK(OBD_FAIL_UNKNOWN_LMV_STRIPE))
2996                 lsm->lsm_md_hash_type = LMV_HASH_TYPE_UNKNOWN;
2997         else
2998                 lsm->lsm_md_hash_type = le32_to_cpu(lmm1->lmv_hash_type);
2999         lsm->lsm_md_layout_version = le32_to_cpu(lmm1->lmv_layout_version);
3000         cplen = strlcpy(lsm->lsm_md_pool_name, lmm1->lmv_pool_name,
3001                         sizeof(lsm->lsm_md_pool_name));
3002
3003         if (cplen >= sizeof(lsm->lsm_md_pool_name))
3004                 RETURN(-E2BIG);
3005
3006         CDEBUG(D_INFO, "unpack lsm count %d, master %d hash_type %d"
3007                "layout_version %d\n", lsm->lsm_md_stripe_count,
3008                lsm->lsm_md_master_mdt_index, lsm->lsm_md_hash_type,
3009                lsm->lsm_md_layout_version);
3010
3011         stripe_count = le32_to_cpu(lmm1->lmv_stripe_count);
3012         for (i = 0; i < stripe_count; i++) {
3013                 fid_le_to_cpu(&lsm->lsm_md_oinfo[i].lmo_fid,
3014                               &lmm1->lmv_stripe_fids[i]);
3015                 rc = lmv_fld_lookup(lmv, &lsm->lsm_md_oinfo[i].lmo_fid,
3016                                     &lsm->lsm_md_oinfo[i].lmo_mds);
3017                 if (rc != 0)
3018                         RETURN(rc);
3019                 CDEBUG(D_INFO, "unpack fid #%d "DFID"\n", i,
3020                        PFID(&lsm->lsm_md_oinfo[i].lmo_fid));
3021         }
3022
3023         RETURN(rc);
3024 }
3025
3026 int lmv_unpack_md(struct obd_export *exp, struct lmv_stripe_md **lsmp,
3027                   const union lmv_mds_md *lmm, int stripe_count)
3028 {
3029         struct lmv_stripe_md     *lsm;
3030         int                      lsm_size;
3031         int                      rc;
3032         bool                     allocated = false;
3033         ENTRY;
3034
3035         LASSERT(lsmp != NULL);
3036
3037         lsm = *lsmp;
3038         /* Free memmd */
3039         if (lsm != NULL && lmm == NULL) {
3040                 int i;
3041                 for (i = 0; i < lsm->lsm_md_stripe_count; i++) {
3042                         /* For migrating inode, the master stripe and master
3043                          * object will be the same, so do not need iput, see
3044                          * ll_update_lsm_md */
3045                         if (!(lsm->lsm_md_hash_type & LMV_HASH_FLAG_MIGRATION &&
3046                               i == 0) && lsm->lsm_md_oinfo[i].lmo_root != NULL)
3047                                 iput(lsm->lsm_md_oinfo[i].lmo_root);
3048                 }
3049                 lsm_size = lmv_stripe_md_size(lsm->lsm_md_stripe_count);
3050                 OBD_FREE(lsm, lsm_size);
3051                 *lsmp = NULL;
3052                 RETURN(0);
3053         }
3054
3055         /* Alloc memmd */
3056         if (lsm == NULL && lmm == NULL) {
3057                 lsm_size = lmv_stripe_md_size(stripe_count);
3058                 OBD_ALLOC(lsm, lsm_size);
3059                 if (lsm == NULL)
3060                         RETURN(-ENOMEM);
3061                 lsm->lsm_md_stripe_count = stripe_count;
3062                 *lsmp = lsm;
3063                 RETURN(0);
3064         }
3065
3066         if (le32_to_cpu(lmm->lmv_magic) == LMV_MAGIC_STRIPE)
3067                 RETURN(-EPERM);
3068
3069         /* Unpack memmd */
3070         if (le32_to_cpu(lmm->lmv_magic) != LMV_MAGIC_V1 &&
3071             le32_to_cpu(lmm->lmv_magic) != LMV_USER_MAGIC) {
3072                 CERROR("%s: invalid lmv magic %x: rc = %d\n",
3073                        exp->exp_obd->obd_name, le32_to_cpu(lmm->lmv_magic),
3074                        -EIO);
3075                 RETURN(-EIO);
3076         }
3077
3078         if (le32_to_cpu(lmm->lmv_magic) == LMV_MAGIC_V1)
3079                 lsm_size = lmv_stripe_md_size(lmv_mds_md_stripe_count_get(lmm));
3080         else
3081                 /**
3082                  * Unpack default dirstripe(lmv_user_md) to lmv_stripe_md,
3083                  * stripecount should be 0 then.
3084                  */
3085                 lsm_size = lmv_stripe_md_size(0);
3086
3087         lsm_size = lmv_stripe_md_size(lmv_mds_md_stripe_count_get(lmm));
3088         if (lsm == NULL) {
3089                 OBD_ALLOC(lsm, lsm_size);
3090                 if (lsm == NULL)
3091                         RETURN(-ENOMEM);
3092                 allocated = true;
3093                 *lsmp = lsm;
3094         }
3095
3096         switch (le32_to_cpu(lmm->lmv_magic)) {
3097         case LMV_MAGIC_V1:
3098                 rc = lmv_unpack_md_v1(exp, lsm, &lmm->lmv_md_v1);
3099                 break;
3100         default:
3101                 CERROR("%s: unrecognized magic %x\n", exp->exp_obd->obd_name,
3102                        le32_to_cpu(lmm->lmv_magic));
3103                 rc = -EINVAL;
3104                 break;
3105         }
3106
3107         if (rc != 0 && allocated) {
3108                 OBD_FREE(lsm, lsm_size);
3109                 *lsmp = NULL;
3110                 lsm_size = rc;
3111         }
3112         RETURN(lsm_size);
3113 }
3114
3115 int lmv_alloc_memmd(struct lmv_stripe_md **lsmp, int stripes)
3116 {
3117         return lmv_unpack_md(NULL, lsmp, NULL, stripes);
3118 }
3119 EXPORT_SYMBOL(lmv_alloc_memmd);
3120
3121 void lmv_free_memmd(struct lmv_stripe_md *lsm)
3122 {
3123         lmv_unpack_md(NULL, &lsm, NULL, 0);
3124 }
3125 EXPORT_SYMBOL(lmv_free_memmd);
3126
3127 int lmv_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
3128                  struct lov_mds_md *lmm, int disk_len)
3129 {
3130         return lmv_unpack_md(exp, (struct lmv_stripe_md **)lsmp,
3131                              (union lmv_mds_md *)lmm, disk_len);
3132 }
3133
3134 int lmv_packmd(struct obd_export *exp, struct lov_mds_md **lmmp,
3135                struct lov_stripe_md *lsm)
3136 {
3137         struct obd_device               *obd = exp->exp_obd;
3138         struct lmv_obd                  *lmv_obd = &obd->u.lmv;
3139         const struct lmv_stripe_md      *lmv = (struct lmv_stripe_md *)lsm;
3140         int                             stripe_count;
3141
3142         if (lmmp == NULL) {
3143                 if (lsm != NULL)
3144                         stripe_count = lmv->lsm_md_stripe_count;
3145                 else
3146                         stripe_count = lmv_obd->desc.ld_tgt_count;
3147
3148                 return lmv_mds_md_size(stripe_count, LMV_MAGIC_V1);
3149         }
3150
3151         return lmv_pack_md((union lmv_mds_md **)lmmp, lmv, 0);
3152 }
3153
3154 static int lmv_cancel_unused(struct obd_export *exp, const struct lu_fid *fid,
3155                              ldlm_policy_data_t *policy, ldlm_mode_t mode,
3156                              ldlm_cancel_flags_t flags, void *opaque)
3157 {
3158         struct obd_device       *obd = exp->exp_obd;
3159         struct lmv_obd          *lmv = &obd->u.lmv;
3160         int                      rc = 0;
3161         int                      err;
3162         __u32                    i;
3163         ENTRY;
3164
3165         LASSERT(fid != NULL);
3166
3167         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
3168                 struct lmv_tgt_desc *tgt = lmv->tgts[i];
3169
3170                 if (tgt == NULL || tgt->ltd_exp == NULL || !tgt->ltd_active)
3171                         continue;
3172
3173                 err = md_cancel_unused(tgt->ltd_exp, fid, policy, mode, flags,
3174                                        opaque);
3175                 if (!rc)
3176                         rc = err;
3177         }
3178         RETURN(rc);
3179 }
3180
3181 int lmv_set_lock_data(struct obd_export *exp, __u64 *lockh, void *data,
3182                       __u64 *bits)
3183 {
3184         struct lmv_obd          *lmv = &exp->exp_obd->u.lmv;
3185         struct lmv_tgt_desc     *tgt = lmv->tgts[0];
3186         int                      rc;
3187         ENTRY;
3188
3189         if (tgt == NULL || tgt->ltd_exp == NULL)
3190                 RETURN(-EINVAL);
3191         rc =  md_set_lock_data(tgt->ltd_exp, lockh, data, bits);
3192         RETURN(rc);
3193 }
3194
3195 ldlm_mode_t lmv_lock_match(struct obd_export *exp, __u64 flags,
3196                            const struct lu_fid *fid, ldlm_type_t type,
3197                            ldlm_policy_data_t *policy, ldlm_mode_t mode,
3198                            struct lustre_handle *lockh)
3199 {
3200         struct obd_device       *obd = exp->exp_obd;
3201         struct lmv_obd          *lmv = &obd->u.lmv;
3202         ldlm_mode_t             rc;
3203         int                     tgt;
3204         int                     i;
3205         ENTRY;
3206
3207         CDEBUG(D_INODE, "Lock match for "DFID"\n", PFID(fid));
3208
3209         /*
3210          * With DNE every object can have two locks in different namespaces:
3211          * lookup lock in space of MDT storing direntry and update/open lock in
3212          * space of MDT storing inode.  Try the MDT that the FID maps to first,
3213          * since this can be easily found, and only try others if that fails.
3214          */
3215         for (i = 0, tgt = lmv_find_target_index(lmv, fid);
3216              i < lmv->desc.ld_tgt_count;
3217              i++, tgt = (tgt + 1) % lmv->desc.ld_tgt_count) {
3218                 if (tgt < 0) {
3219                         CDEBUG(D_HA, "%s: "DFID" is inaccessible: rc = %d\n",
3220                                obd->obd_name, PFID(fid), tgt);
3221                         tgt = 0;
3222                 }
3223
3224                 if (lmv->tgts[tgt] == NULL ||
3225                     lmv->tgts[tgt]->ltd_exp == NULL ||
3226                     lmv->tgts[tgt]->ltd_active == 0)
3227                         continue;
3228
3229                 rc = md_lock_match(lmv->tgts[tgt]->ltd_exp, flags, fid,
3230                                    type, policy, mode, lockh);
3231                 if (rc)
3232                         RETURN(rc);
3233         }
3234
3235         RETURN(0);
3236 }
3237
3238 int lmv_get_lustre_md(struct obd_export *exp, struct ptlrpc_request *req,
3239                       struct obd_export *dt_exp, struct obd_export *md_exp,
3240                       struct lustre_md *md)
3241 {
3242         struct lmv_obd          *lmv = &exp->exp_obd->u.lmv;
3243         struct lmv_tgt_desc     *tgt = lmv->tgts[0];
3244
3245         if (tgt == NULL || tgt->ltd_exp == NULL)
3246                 RETURN(-EINVAL);
3247
3248         return md_get_lustre_md(lmv->tgts[0]->ltd_exp, req, dt_exp, md_exp, md);
3249 }
3250
3251 int lmv_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
3252 {
3253         struct obd_device       *obd = exp->exp_obd;
3254         struct lmv_obd          *lmv = &obd->u.lmv;
3255         struct lmv_tgt_desc     *tgt = lmv->tgts[0];
3256         ENTRY;
3257
3258         if (md->lmv != NULL) {
3259                 lmv_free_memmd(md->lmv);
3260                 md->lmv = NULL;
3261         }
3262         if (tgt == NULL || tgt->ltd_exp == NULL)
3263                 RETURN(-EINVAL);
3264         RETURN(md_free_lustre_md(lmv->tgts[0]->ltd_exp, md));
3265 }
3266
3267 int lmv_set_open_replay_data(struct obd_export *exp,
3268                              struct obd_client_handle *och,
3269                              struct lookup_intent *it)
3270 {
3271         struct obd_device       *obd = exp->exp_obd;
3272         struct lmv_obd          *lmv = &obd->u.lmv;
3273         struct lmv_tgt_desc     *tgt;
3274         ENTRY;
3275
3276         tgt = lmv_find_target(lmv, &och->och_fid);
3277         if (IS_ERR(tgt))
3278                 RETURN(PTR_ERR(tgt));
3279
3280         RETURN(md_set_open_replay_data(tgt->ltd_exp, och, it));
3281 }
3282
3283 int lmv_clear_open_replay_data(struct obd_export *exp,
3284                                struct obd_client_handle *och)
3285 {
3286         struct obd_device       *obd = exp->exp_obd;
3287         struct lmv_obd          *lmv = &obd->u.lmv;
3288         struct lmv_tgt_desc     *tgt;
3289         ENTRY;
3290
3291         tgt = lmv_find_target(lmv, &och->och_fid);
3292         if (IS_ERR(tgt))
3293                 RETURN(PTR_ERR(tgt));
3294
3295         RETURN(md_clear_open_replay_data(tgt->ltd_exp, och));
3296 }
3297
3298 static int lmv_get_remote_perm(struct obd_export *exp,
3299                                const struct lu_fid *fid,
3300                                struct obd_capa *oc, __u32 suppgid,
3301                                struct ptlrpc_request **request)
3302 {
3303         struct obd_device       *obd = exp->exp_obd;
3304         struct lmv_obd          *lmv = &obd->u.lmv;
3305         struct lmv_tgt_desc     *tgt;
3306         int                      rc;
3307         ENTRY;
3308
3309         rc = lmv_check_connect(obd);
3310         if (rc)
3311                 RETURN(rc);
3312
3313         tgt = lmv_find_target(lmv, fid);
3314         if (IS_ERR(tgt))
3315                 RETURN(PTR_ERR(tgt));
3316
3317         rc = md_get_remote_perm(tgt->ltd_exp, fid, oc, suppgid, request);
3318         RETURN(rc);
3319 }
3320
3321 static int lmv_renew_capa(struct obd_export *exp, struct obd_capa *oc,
3322                           renew_capa_cb_t cb)
3323 {
3324         struct obd_device       *obd = exp->exp_obd;
3325         struct lmv_obd          *lmv = &obd->u.lmv;
3326         struct lmv_tgt_desc     *tgt;
3327         int                      rc;
3328         ENTRY;
3329
3330         rc = lmv_check_connect(obd);
3331         if (rc)
3332                 RETURN(rc);
3333
3334         tgt = lmv_find_target(lmv, &oc->c_capa.lc_fid);
3335         if (IS_ERR(tgt))
3336                 RETURN(PTR_ERR(tgt));
3337
3338         rc = md_renew_capa(tgt->ltd_exp, oc, cb);
3339         RETURN(rc);
3340 }
3341
3342 int lmv_unpack_capa(struct obd_export *exp, struct ptlrpc_request *req,
3343                     const struct req_msg_field *field, struct obd_capa **oc)
3344 {
3345         struct lmv_obd          *lmv = &exp->exp_obd->u.lmv;
3346         struct lmv_tgt_desc     *tgt = lmv->tgts[0];
3347
3348         if (tgt == NULL || tgt->ltd_exp == NULL)
3349                 RETURN(-EINVAL);
3350         return md_unpack_capa(tgt->ltd_exp, req, field, oc);
3351 }
3352
3353 int lmv_intent_getattr_async(struct obd_export *exp,
3354                              struct md_enqueue_info *minfo,
3355                              struct ldlm_enqueue_info *einfo)
3356 {
3357         struct md_op_data       *op_data = &minfo->mi_data;
3358         struct obd_device       *obd = exp->exp_obd;
3359         struct lmv_obd          *lmv = &obd->u.lmv;
3360         struct lmv_tgt_desc     *tgt = NULL;
3361         int                      rc;
3362         ENTRY;
3363
3364         rc = lmv_check_connect(obd);
3365         if (rc)
3366                 RETURN(rc);
3367
3368         tgt = lmv_locate_mds(lmv, op_data, &op_data->op_fid1);
3369         if (IS_ERR(tgt))
3370                 RETURN(PTR_ERR(tgt));
3371
3372         rc = md_intent_getattr_async(tgt->ltd_exp, minfo, einfo);
3373         RETURN(rc);
3374 }
3375
3376 int lmv_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
3377                         struct lu_fid *fid, __u64 *bits)
3378 {
3379         struct obd_device       *obd = exp->exp_obd;
3380         struct lmv_obd          *lmv = &obd->u.lmv;
3381         struct lmv_tgt_desc     *tgt;
3382         int                      rc;
3383         ENTRY;
3384
3385         rc = lmv_check_connect(obd);
3386         if (rc)
3387                 RETURN(rc);
3388
3389         tgt = lmv_find_target(lmv, fid);
3390         if (IS_ERR(tgt))
3391                 RETURN(PTR_ERR(tgt));
3392
3393         rc = md_revalidate_lock(tgt->ltd_exp, it, fid, bits);
3394         RETURN(rc);
3395 }
3396
3397 int lmv_get_fid_from_lsm(struct obd_export *exp,
3398                          const struct lmv_stripe_md *lsm,
3399                          const char *name, int namelen, struct lu_fid *fid)
3400 {
3401         const struct lmv_oinfo *oinfo;
3402
3403         LASSERT(lsm != NULL);
3404         oinfo = lsm_name_to_stripe_info(lsm, name, namelen);
3405         if (IS_ERR(oinfo))
3406                 return PTR_ERR(oinfo);
3407
3408         *fid = oinfo->lmo_fid;
3409
3410         RETURN(0);
3411 }
3412
3413 /**
3414  * For lmv, only need to send request to master MDT, and the master MDT will
3415  * process with other slave MDTs. The only exception is Q_GETOQUOTA for which
3416  * we directly fetch data from the slave MDTs.
3417  */
3418 int lmv_quotactl(struct obd_device *unused, struct obd_export *exp,
3419                  struct obd_quotactl *oqctl)
3420 {
3421         struct obd_device   *obd = class_exp2obd(exp);
3422         struct lmv_obd      *lmv = &obd->u.lmv;
3423         struct lmv_tgt_desc *tgt = lmv->tgts[0];
3424         int                  rc = 0;
3425         __u32                i;
3426         __u64                curspace, curinodes;
3427         ENTRY;
3428
3429         if (tgt == NULL ||
3430             tgt->ltd_exp == NULL ||
3431             !tgt->ltd_active ||
3432             lmv->desc.ld_tgt_count == 0) {
3433                 CERROR("master lmv inactive\n");
3434                 RETURN(-EIO);
3435         }
3436
3437         if (oqctl->qc_cmd != Q_GETOQUOTA) {
3438                 rc = obd_quotactl(tgt->ltd_exp, oqctl);
3439                 RETURN(rc);
3440         }
3441
3442         curspace = curinodes = 0;
3443         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
3444                 int err;
3445                 tgt = lmv->tgts[i];
3446
3447                 if (tgt == NULL || tgt->ltd_exp == NULL || !tgt->ltd_active)
3448                         continue;
3449
3450                 err = obd_quotactl(tgt->ltd_exp, oqctl);
3451                 if (err) {
3452                         CERROR("getquota on mdt %d failed. %d\n", i, err);
3453                         if (!rc)
3454                                 rc = err;
3455                 } else {
3456                         curspace += oqctl->qc_dqblk.dqb_curspace;
3457                         curinodes += oqctl->qc_dqblk.dqb_curinodes;
3458                 }
3459         }
3460         oqctl->qc_dqblk.dqb_curspace = curspace;
3461         oqctl->qc_dqblk.dqb_curinodes = curinodes;
3462
3463         RETURN(rc);
3464 }
3465
3466 int lmv_quotacheck(struct obd_device *unused, struct obd_export *exp,
3467                    struct obd_quotactl *oqctl)
3468 {
3469         struct obd_device       *obd = class_exp2obd(exp);
3470         struct lmv_obd          *lmv = &obd->u.lmv;
3471         struct lmv_tgt_desc     *tgt;
3472         __u32                    i;
3473         int                      rc = 0;
3474         ENTRY;
3475
3476         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
3477                 int err;
3478                 tgt = lmv->tgts[i];
3479                 if (tgt == NULL || tgt->ltd_exp == NULL || !tgt->ltd_active) {
3480                         CERROR("lmv idx %d inactive\n", i);
3481                         RETURN(-EIO);
3482                 }
3483
3484                 err = obd_quotacheck(tgt->ltd_exp, oqctl);
3485                 if (err && !rc)
3486                         rc = err;
3487         }
3488
3489         RETURN(rc);
3490 }
3491
3492 int lmv_update_lsm_md(struct obd_export *exp, struct lmv_stripe_md *lsm,
3493                       struct mdt_body *body, ldlm_blocking_callback cb_blocking)
3494 {
3495         return lmv_revalidate_slaves(exp, body, lsm, cb_blocking, 0);
3496 }
3497
3498 int lmv_merge_attr(struct obd_export *exp, const struct lmv_stripe_md *lsm,
3499                    struct cl_attr *attr)
3500 {
3501         int i;
3502
3503         for (i = 0; i < lsm->lsm_md_stripe_count; i++) {
3504                 struct inode *inode = lsm->lsm_md_oinfo[i].lmo_root;
3505
3506                 CDEBUG(D_INFO, ""DFID" size %llu, nlink %u, atime %lu ctime"
3507                        "%lu, mtime %lu.\n", PFID(&lsm->lsm_md_oinfo[i].lmo_fid),
3508                        i_size_read(inode), inode->i_nlink,
3509                        LTIME_S(inode->i_atime), LTIME_S(inode->i_ctime),
3510                        LTIME_S(inode->i_mtime));
3511
3512                 /* for slave stripe, it needs to subtract nlink for . and .. */
3513                 if (i != 0)
3514                         attr->cat_nlink += inode->i_nlink - 2;
3515                 else
3516                         attr->cat_nlink = inode->i_nlink;
3517
3518                 attr->cat_size += i_size_read(inode);
3519
3520                 if (attr->cat_atime < LTIME_S(inode->i_atime))
3521                         attr->cat_atime = LTIME_S(inode->i_atime);
3522
3523                 if (attr->cat_ctime < LTIME_S(inode->i_ctime))
3524                         attr->cat_ctime = LTIME_S(inode->i_ctime);
3525
3526                 if (attr->cat_mtime < LTIME_S(inode->i_mtime))
3527                         attr->cat_mtime = LTIME_S(inode->i_mtime);
3528         }
3529         return 0;
3530 }
3531
3532 struct obd_ops lmv_obd_ops = {
3533         .o_owner                = THIS_MODULE,
3534         .o_setup                = lmv_setup,
3535         .o_cleanup              = lmv_cleanup,
3536         .o_precleanup           = lmv_precleanup,
3537         .o_process_config       = lmv_process_config,
3538         .o_connect              = lmv_connect,
3539         .o_disconnect           = lmv_disconnect,
3540         .o_statfs               = lmv_statfs,
3541         .o_get_info             = lmv_get_info,
3542         .o_set_info_async       = lmv_set_info_async,
3543         .o_packmd               = lmv_packmd,
3544         .o_unpackmd             = lmv_unpackmd,
3545         .o_notify               = lmv_notify,
3546         .o_get_uuid             = lmv_get_uuid,
3547         .o_iocontrol            = lmv_iocontrol,
3548         .o_quotacheck           = lmv_quotacheck,
3549         .o_quotactl             = lmv_quotactl
3550 };
3551
3552 struct md_ops lmv_md_ops = {
3553         .m_getstatus            = lmv_getstatus,
3554         .m_null_inode           = lmv_null_inode,
3555         .m_find_cbdata          = lmv_find_cbdata,
3556         .m_close                = lmv_close,
3557         .m_create               = lmv_create,
3558         .m_done_writing         = lmv_done_writing,
3559         .m_enqueue              = lmv_enqueue,
3560         .m_getattr              = lmv_getattr,
3561         .m_getxattr             = lmv_getxattr,
3562         .m_getattr_name         = lmv_getattr_name,
3563         .m_intent_lock          = lmv_intent_lock,
3564         .m_link                 = lmv_link,
3565         .m_rename               = lmv_rename,
3566         .m_setattr              = lmv_setattr,
3567         .m_setxattr             = lmv_setxattr,
3568         .m_fsync                = lmv_fsync,
3569         .m_read_page            = lmv_read_page,
3570         .m_unlink               = lmv_unlink,
3571         .m_init_ea_size         = lmv_init_ea_size,
3572         .m_cancel_unused        = lmv_cancel_unused,
3573         .m_set_lock_data        = lmv_set_lock_data,
3574         .m_lock_match           = lmv_lock_match,
3575         .m_get_lustre_md        = lmv_get_lustre_md,
3576         .m_free_lustre_md       = lmv_free_lustre_md,
3577         .m_update_lsm_md        = lmv_update_lsm_md,
3578         .m_merge_attr           = lmv_merge_attr,
3579         .m_set_open_replay_data = lmv_set_open_replay_data,
3580         .m_clear_open_replay_data = lmv_clear_open_replay_data,
3581         .m_renew_capa           = lmv_renew_capa,
3582         .m_unpack_capa          = lmv_unpack_capa,
3583         .m_get_remote_perm      = lmv_get_remote_perm,
3584         .m_intent_getattr_async = lmv_intent_getattr_async,
3585         .m_revalidate_lock      = lmv_revalidate_lock,
3586         .m_get_fid_from_lsm     = lmv_get_fid_from_lsm,
3587 };
3588
3589 int __init lmv_init(void)
3590 {
3591         return class_register_type(&lmv_obd_ops, &lmv_md_ops, true, NULL,
3592                                    LUSTRE_LMV_NAME, NULL);
3593 }
3594
3595 static void lmv_exit(void)
3596 {
3597         class_unregister_type(LUSTRE_LMV_NAME);
3598 }
3599
3600 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
3601 MODULE_DESCRIPTION("Lustre Logical Metadata Volume OBD driver");
3602 MODULE_LICENSE("GPL");
3603
3604 module_init(lmv_init);
3605 module_exit(lmv_exit);