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