lustre/lmv/lmv_obd.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #ifndef EXPORT_SYMTAB
  38 # define EXPORT_SYMTAB
  39 #endif
  40 #define DEBUG_SUBSYSTEM S_LMV
  41 #ifdef __KERNEL__
  42 #include <linux/slab.h>
  43 #include <linux/module.h>
  44 #include <linux/init.h>
  45 #include <linux/slab.h>
  46 #include <linux/pagemap.h>
  47 #include <linux/mm.h>
  48 #include <asm/div64.h>
  49 #include <linux/seq_file.h>
  50 #include <linux/namei.h>
  51 #else
  52 #include <liblustre.h>
  53 #endif
  54
  55 #include <lustre/lustre_idl.h>
  56 #include <lustre_log.h>
  57 #include <obd_support.h>
  58 #include <lustre_lib.h>
  59 #include <lustre_net.h>
  60 #include <obd_class.h>
  61 #include <lprocfs_status.h>
  62 #include <lustre_lite.h>
  63 #include <lustre_fid.h>
  64 #include "lmv_internal.h"
  65
  66 /* not defined for liblustre building */
  67 #if !defined(ATOMIC_INIT)
  68 #define ATOMIC_INIT(val) { (val) }
  69 #endif
  70
  71 /* object cache. */
  72 cfs_mem_cache_t *obj_cache;
  73 atomic_t obj_cache_count = ATOMIC_INIT(0);
  74
  75 static void lmv_activate_target(struct lmv_obd *lmv,
  76                                 struct lmv_tgt_desc *tgt,
  77                                 int activate)
  78 {
  79         if (tgt->ltd_active == activate)
  80                 return;
  81
  82         tgt->ltd_active = activate;
  83         lmv->desc.ld_active_tgt_count += (activate ? 1 : -1);
  84 }
  85
  86 /* Error codes:
  87  *
  88  *  -EINVAL  : UUID can't be found in the LMV's target list
  89  *  -ENOTCONN: The UUID is found, but the target connection is bad (!)
  90  *  -EBADF   : The UUID is found, but the OBD of the wrong type (!)
  91  */
  92 static int lmv_set_mdc_active(struct lmv_obd *lmv, struct obd_uuid *uuid,
  93                               int activate)
  94 {
  95         struct lmv_tgt_desc *tgt;
  96         struct obd_device *obd;
  97         int i, rc = 0;
  98         ENTRY;
  99
 100         CDEBUG(D_INFO, "Searching in lmv %p for uuid %s (activate=%d)\n",
 101                lmv, uuid->uuid, activate);
 102
 103         spin_lock(&lmv->lmv_lock);
 104         for (i = 0, tgt = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgt++) {
 105                 if (tgt->ltd_exp == NULL)
 106                         continue;
 107
 108                 CDEBUG(D_INFO, "lmv idx %d is %s conn "LPX64"\n",
 109                        i, tgt->ltd_uuid.uuid, tgt->ltd_exp->exp_handle.h_cookie);
 110
 111                 if (obd_uuid_equals(uuid, &tgt->ltd_uuid))
 112                         break;
 113         }
 114
 115         if (i == lmv->desc.ld_tgt_count)
 116                 GOTO(out_lmv_lock, rc = -EINVAL);
 117
 118         obd = class_exp2obd(tgt->ltd_exp);
 119         if (obd == NULL)
 120                 GOTO(out_lmv_lock, rc = -ENOTCONN);
 121
 122         CDEBUG(D_INFO, "Found OBD %s=%s device %d (%p) type %s at LMV idx %d\n",
 123                obd->obd_name, obd->obd_uuid.uuid, obd->obd_minor, obd,
 124                obd->obd_type->typ_name, i);
 125         LASSERT(strcmp(obd->obd_type->typ_name, LUSTRE_MDC_NAME) == 0);
 126
 127         if (tgt->ltd_active == activate) {
 128                 CDEBUG(D_INFO, "OBD %p already %sactive!\n", obd,
 129                        activate ? "" : "in");
 130                 GOTO(out_lmv_lock, rc);
 131         }
 132
 133         CDEBUG(D_INFO, "Marking OBD %p %sactive\n",
 134                obd, activate ? "" : "in");
 135
 136         lmv_activate_target(lmv, tgt, activate);
 137
 138         EXIT;
 139
 140  out_lmv_lock:
 141         spin_unlock(&lmv->lmv_lock);
 142         return rc;
 143 }
 144
 145 static int lmv_set_mdc_data(struct lmv_obd *lmv, struct obd_uuid *uuid,
 146                             struct obd_connect_data *data)
 147 {
 148         struct lmv_tgt_desc *tgt;
 149         int i;
 150         ENTRY;
 151
 152         LASSERT(data != NULL);
 153
 154         spin_lock(&lmv->lmv_lock);
 155         for (i = 0, tgt = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgt++) {
 156                 if (tgt->ltd_exp == NULL)
 157                         continue;
 158
 159                 if (obd_uuid_equals(uuid, &tgt->ltd_uuid)) {
 160                         lmv->datas[tgt->ltd_idx] = *data;
 161                         break;
 162                 }
 163         }
 164         spin_unlock(&lmv->lmv_lock);
 165         RETURN(0);
 166 }
 167
 168 static int lmv_notify(struct obd_device *obd, struct obd_device *watched,
 169                       enum obd_notify_event ev, void *data)
 170 {
 171         struct lmv_obd *lmv = &obd->u.lmv;
 172         struct obd_uuid *uuid;
 173         int rc = 0;
 174         ENTRY;
 175
 176         if (strcmp(watched->obd_type->typ_name, LUSTRE_MDC_NAME)) {
 177                 CERROR("unexpected notification of %s %s!\n",
 178                        watched->obd_type->typ_name,
 179                        watched->obd_name);
 180                 RETURN(-EINVAL);
 181         }
 182
 183         uuid = &watched->u.cli.cl_target_uuid;
 184         if (ev == OBD_NOTIFY_ACTIVE || ev == OBD_NOTIFY_INACTIVE) {
 185                 /*
 186                  * Set MDC as active before notifying the observer, so the
 187                  * observer can use the MDC normally.
 188                  */
 189                 rc = lmv_set_mdc_active(lmv, uuid,
 190                                         ev == OBD_NOTIFY_ACTIVE);
 191                 if (rc) {
 192                         CERROR("%sactivation of %s failed: %d\n",
 193                                ev == OBD_NOTIFY_ACTIVE ? "" : "de",
 194                                uuid->uuid, rc);
 195                         RETURN(rc);
 196                 }
 197         } else if (ev == OBD_NOTIFY_OCD) {
 198                 struct obd_connect_data *conn_data =
 199                         &watched->u.cli.cl_import->imp_connect_data;
 200
 201                 /* Set connect data to desired target, update
 202                  * exp_connect_flags. */
 203                 rc = lmv_set_mdc_data(lmv, uuid, conn_data);
 204                 if (rc) {
 205                         CERROR("can't set connect data to target %s, rc %d\n",
 206                                uuid->uuid, rc);
 207                         RETURN(rc);
 208                 }
 209
 210                 /*
 211                  * XXX: Make sure that ocd_connect_flags from all targets are
 212                  * the same. Otherwise one of MDTs runs wrong version or
 213                  * something like this.  --umka
 214                  */
 215                 obd->obd_self_export->exp_connect_flags =
 216                         conn_data->ocd_connect_flags;
 217         }
 218 #if 0
 219         else if (ev == OBD_NOTIFY_DISCON) {
 220                 /* For disconnect event, flush fld cache for failout MDS case. */
 221                 fld_client_flush(&lmv->lmv_fld);
 222         }
 223 #endif
 224         /* Pass the notification up the chain. */
 225         if (obd->obd_observer)
 226                 rc = obd_notify(obd->obd_observer, watched, ev, data);
 227
 228         RETURN(rc);
 229 }
 230
 231 /* this is fake connect function. Its purpose is to initialize lmv and say
 232  * caller that everything is okay. Real connection will be performed later. */
 233 static int lmv_connect(const struct lu_env *env,
 234                        struct lustre_handle *conn, struct obd_device *obd,
 235                        struct obd_uuid *cluuid, struct obd_connect_data *data,
 236                        void *localdata)
 237 {
 238 #ifdef __KERNEL__
 239         struct proc_dir_entry *lmv_proc_dir;
 240 #endif
 241         struct lmv_obd *lmv = &obd->u.lmv;
 242         struct obd_export *exp;
 243         int rc = 0;
 244         ENTRY;
 245
 246         rc = class_connect(conn, obd, cluuid);
 247         if (rc) {
 248                 CERROR("class_connection() returned %d\n", rc);
 249                 RETURN(rc);
 250         }
 251
 252         exp = class_conn2export(conn);
 253
 254         /* we don't want to actually do the underlying connections more than
 255          * once, so keep track. */
 256         lmv->refcount++;
 257         if (lmv->refcount > 1) {
 258                 class_export_put(exp);
 259                 RETURN(0);
 260         }
 261
 262         lmv->exp = exp;
 263         lmv->connected = 0;
 264         lmv->cluuid = *cluuid;
 265
 266         if (data)
 267                 lmv->conn_data = *data;
 268
 269 #ifdef __KERNEL__
 270         lmv_proc_dir = lprocfs_register("target_obds", obd->obd_proc_entry,
 271                                         NULL, NULL);
 272         if (IS_ERR(lmv_proc_dir)) {
 273                 CERROR("could not register /proc/fs/lustre/%s/%s/target_obds.",
 274                        obd->obd_type->typ_name, obd->obd_name);
 275                 lmv_proc_dir = NULL;
 276         }
 277 #endif
 278
 279         /* all real clients should perform actual connection right away, because
 280          * it is possible, that LMV will not have opportunity to connect targets
 281          * and MDC stuff will be called directly, for instance while reading
 282          * ../mdc/../kbytesfree procfs file, etc. */
 283         if (data->ocd_connect_flags & OBD_CONNECT_REAL)
 284                 rc = lmv_check_connect(obd);
 285
 286 #ifdef __KERNEL__
 287         if (rc) {
 288                 if (lmv_proc_dir)
 289                         lprocfs_remove(&lmv_proc_dir);
 290         }
 291 #endif
 292
 293         RETURN(rc);
 294 }
 295
 296 static void lmv_set_timeouts(struct obd_device *obd)
 297 {
 298         struct lmv_tgt_desc *tgts;
 299         struct lmv_obd *lmv;
 300         int i;
 301
 302         lmv = &obd->u.lmv;
 303         if (lmv->server_timeout == 0)
 304                 return;
 305
 306         if (lmv->connected == 0)
 307                 return;
 308
 309         for (i = 0, tgts = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgts++) {
 310                 if (tgts->ltd_exp == NULL)
 311                         continue;
 312
 313                 obd_set_info_async(tgts->ltd_exp, sizeof(KEY_INTERMDS),
 314                                    KEY_INTERMDS, 0, NULL, NULL);
 315         }
 316 }
 317
 318 static int lmv_init_ea_size(struct obd_export *exp, int easize,
 319                             int def_easize, int cookiesize)
 320 {
 321         struct obd_device *obd = exp->exp_obd;
 322         struct lmv_obd *lmv = &obd->u.lmv;
 323         int i, rc = 0, change = 0;
 324         ENTRY;
 325
 326         if (lmv->max_easize < easize) {
 327                 lmv->max_easize = easize;
 328                 change = 1;
 329         }
 330         if (lmv->max_def_easize < def_easize) {
 331                 lmv->max_def_easize = def_easize;
 332                 change = 1;
 333         }
 334         if (lmv->max_cookiesize < cookiesize) {
 335                 lmv->max_cookiesize = cookiesize;
 336                 change = 1;
 337         }
 338         if (change == 0)
 339                 RETURN(0);
 340
 341         if (lmv->connected == 0)
 342                 RETURN(0);
 343
 344         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 345                 if (lmv->tgts[i].ltd_exp == NULL) {
 346                         CWARN("%s: NULL export for %d\n", obd->obd_name, i);
 347                         continue;
 348                 }
 349
 350                 rc = md_init_ea_size(lmv->tgts[i].ltd_exp, easize, def_easize,
 351                                      cookiesize);
 352                 if (rc) {
 353                         CERROR("obd_init_ea_size() failed on MDT target %d, "
 354                                "error %d.\n", i, rc);
 355                         break;
 356                 }
 357         }
 358         RETURN(rc);
 359 }
 360
 361 #define MAX_STRING_SIZE 128
 362
 363 int lmv_connect_mdc(struct obd_device *obd, struct lmv_tgt_desc *tgt)
 364 {
 365         struct lmv_obd *lmv = &obd->u.lmv;
 366         struct obd_uuid *cluuid = &lmv->cluuid;
 367         struct obd_connect_data *mdc_data = NULL;
 368         struct obd_uuid lmv_mdc_uuid = { "LMV_MDC_UUID" };
 369         struct lustre_handle conn = {0, };
 370         struct obd_device *mdc_obd;
 371         struct obd_export *mdc_exp;
 372         struct lu_fld_target target;
 373         int rc;
 374 #ifdef __KERNEL__
 375         struct proc_dir_entry *lmv_proc_dir;
 376 #endif
 377         ENTRY;
 378
 379         /* for MDS: don't connect to yourself */
 380         if (obd_uuid_equals(&tgt->ltd_uuid, cluuid)) {
 381                 CDEBUG(D_CONFIG, "don't connect back to %s\n", cluuid->uuid);
 382                 /* XXX - the old code didn't increment active tgt count.
 383                  *       should we ? */
 384                 RETURN(0);
 385         }
 386
 387         mdc_obd = class_find_client_obd(&tgt->ltd_uuid, LUSTRE_MDC_NAME,
 388                                         &obd->obd_uuid);
 389         if (!mdc_obd) {
 390                 CERROR("target %s not attached\n", tgt->ltd_uuid.uuid);
 391                 RETURN(-EINVAL);
 392         }
 393
 394         CDEBUG(D_CONFIG, "connect to %s(%s) - %s, %s FOR %s\n",
 395                 mdc_obd->obd_name, mdc_obd->obd_uuid.uuid,
 396                 tgt->ltd_uuid.uuid, obd->obd_uuid.uuid,
 397                 cluuid->uuid);
 398
 399         if (!mdc_obd->obd_set_up) {
 400                 CERROR("target %s is not set up\n", tgt->ltd_uuid.uuid);
 401                 RETURN(-EINVAL);
 402         }
 403
 404         rc = obd_connect(NULL, &conn, mdc_obd, &lmv_mdc_uuid,
 405                          &lmv->conn_data, NULL);
 406         if (rc) {
 407                 CERROR("target %s connect error %d\n", tgt->ltd_uuid.uuid, rc);
 408                 RETURN(rc);
 409         }
 410
 411         mdc_exp = class_conn2export(&conn);
 412
 413         /* Init fid sequence client for this mdc. */
 414         rc = obd_fid_init(mdc_exp);
 415         if (rc)
 416                 RETURN(rc);
 417
 418         /* Add new FLD target. */
 419         target.ft_srv = NULL;
 420         target.ft_exp = mdc_exp;
 421         target.ft_idx = tgt->ltd_idx;
 422
 423         fld_client_add_target(&lmv->lmv_fld, &target);
 424
 425         mdc_data = &class_exp2cliimp(mdc_exp)->imp_connect_data;
 426
 427         rc = obd_register_observer(mdc_obd, obd);
 428         if (rc) {
 429                 obd_disconnect(mdc_exp);
 430                 CERROR("target %s register_observer error %d\n",
 431                        tgt->ltd_uuid.uuid, rc);
 432                 RETURN(rc);
 433         }
 434
 435         if (obd->obd_observer) {
 436                 /* tell the mds_lmv about the new target */
 437                 rc = obd_notify(obd->obd_observer, mdc_exp->exp_obd,
 438                                 OBD_NOTIFY_ACTIVE, (void *)(tgt - lmv->tgts));
 439                 if (rc) {
 440                         obd_disconnect(mdc_exp);
 441                         RETURN(rc);
 442                 }
 443         }
 444
 445         tgt->ltd_active = 1;
 446         tgt->ltd_exp = mdc_exp;
 447         lmv->desc.ld_active_tgt_count++;
 448
 449         /* copy connect data, it may be used later */
 450         lmv->datas[tgt->ltd_idx] = *mdc_data;
 451
 452         md_init_ea_size(tgt->ltd_exp, lmv->max_easize,
 453                         lmv->max_def_easize, lmv->max_cookiesize);
 454
 455         CDEBUG(D_CONFIG, "connected to %s(%s) successfully (%d)\n",
 456                 mdc_obd->obd_name, mdc_obd->obd_uuid.uuid,
 457                 atomic_read(&obd->obd_refcount));
 458
 459 #ifdef __KERNEL__
 460         lmv_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 461         if (lmv_proc_dir) {
 462                 struct proc_dir_entry *mdc_symlink;
 463                 char name[MAX_STRING_SIZE + 1];
 464
 465                 LASSERT(mdc_obd->obd_type != NULL);
 466                 LASSERT(mdc_obd->obd_type->typ_name != NULL);
 467                 name[MAX_STRING_SIZE] = '\0';
 468                 snprintf(name, MAX_STRING_SIZE, "../../../%s/%s",
 469                          mdc_obd->obd_type->typ_name,
 470                          mdc_obd->obd_name);
 471                 mdc_symlink = proc_symlink(mdc_obd->obd_name,
 472                                            lmv_proc_dir, name);
 473                 if (mdc_symlink == NULL) {
 474                         CERROR("could not register LMV target "
 475                                "/proc/fs/lustre/%s/%s/target_obds/%s.",
 476                                obd->obd_type->typ_name, obd->obd_name,
 477                                mdc_obd->obd_name);
 478                         lprocfs_remove(&lmv_proc_dir);
 479                         lmv_proc_dir = NULL;
 480                 }
 481         }
 482 #endif
 483         RETURN(0);
 484 }
 485
 486 int lmv_add_target(struct obd_device *obd, struct obd_uuid *tgt_uuid)
 487 {
 488         struct lmv_obd *lmv = &obd->u.lmv;
 489         struct lmv_tgt_desc *tgt;
 490         int rc = 0;
 491         ENTRY;
 492
 493         CDEBUG(D_CONFIG, "tgt_uuid: %s.\n", tgt_uuid->uuid);
 494
 495         lmv_init_lock(lmv);
 496
 497         if (lmv->desc.ld_active_tgt_count >= LMV_MAX_TGT_COUNT) {
 498                 lmv_init_unlock(lmv);
 499                 CERROR("can't add %s, LMV module compiled for %d MDCs. "
 500                        "That many MDCs already configured.\n",
 501                        tgt_uuid->uuid, LMV_MAX_TGT_COUNT);
 502                 RETURN(-EINVAL);
 503         }
 504         if (lmv->desc.ld_tgt_count == 0) {
 505                 struct obd_device *mdc_obd;
 506
 507                 mdc_obd = class_find_client_obd(tgt_uuid, LUSTRE_MDC_NAME,
 508                                                 &obd->obd_uuid);
 509                 if (!mdc_obd) {
 510                         lmv_init_unlock(lmv);
 511                         CERROR("Target %s not attached\n", tgt_uuid->uuid);
 512                         RETURN(-EINVAL);
 513                 }
 514
 515                 rc = obd_llog_init(obd, &obd->obd_olg, mdc_obd, 0, NULL, tgt_uuid);
 516                 if (rc) {
 517                         lmv_init_unlock(lmv);
 518                         CERROR("lmv failed to setup llogging subsystems\n");
 519                 }
 520         }
 521         spin_lock(&lmv->lmv_lock);
 522         tgt = lmv->tgts + lmv->desc.ld_tgt_count++;
 523         tgt->ltd_uuid = *tgt_uuid;
 524         spin_unlock(&lmv->lmv_lock);
 525
 526         if (lmv->connected) {
 527                 rc = lmv_connect_mdc(obd, tgt);
 528                 if (rc) {
 529                         spin_lock(&lmv->lmv_lock);
 530                         lmv->desc.ld_tgt_count--;
 531                         memset(tgt, 0, sizeof(*tgt));
 532                         spin_unlock(&lmv->lmv_lock);
 533                 } else {
 534                         int easize = sizeof(struct lmv_stripe_md) +
 535                                      lmv->desc.ld_tgt_count *
 536                                      sizeof(struct lu_fid);
 537                         lmv_init_ea_size(obd->obd_self_export, easize, 0, 0);
 538                 }
 539         }
 540
 541         lmv_init_unlock(lmv);
 542         RETURN(rc);
 543 }
 544
 545 /* performs a check if passed obd is connected. If no - connect it. */
 546 int lmv_check_connect(struct obd_device *obd)
 547 {
 548         struct lmv_obd *lmv = &obd->u.lmv;
 549         struct lmv_tgt_desc *tgt;
 550         int i, rc, easize;
 551         ENTRY;
 552
 553         if (lmv->connected)
 554                 RETURN(0);
 555
 556         lmv_init_lock(lmv);
 557         if (lmv->connected) {
 558                 lmv_init_unlock(lmv);
 559                 RETURN(0);
 560         }
 561
 562         if (lmv->desc.ld_tgt_count == 0) {
 563                 CERROR("%s: no targets configured.\n", obd->obd_name);
 564                 RETURN(-EINVAL);
 565         }
 566
 567         CDEBUG(D_CONFIG, "time to connect %s to %s\n",
 568                lmv->cluuid.uuid, obd->obd_name);
 569
 570         LASSERT(lmv->tgts != NULL);
 571
 572         for (i = 0, tgt = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgt++) {
 573                 rc = lmv_connect_mdc(obd, tgt);
 574                 if (rc)
 575                         GOTO(out_disc, rc);
 576         }
 577
 578         lmv_set_timeouts(obd);
 579         class_export_put(lmv->exp);
 580         lmv->connected = 1;
 581         easize = lmv_get_easize(lmv);
 582         lmv_init_ea_size(obd->obd_self_export, easize, 0, 0);
 583         lmv_init_unlock(lmv);
 584         RETURN(0);
 585
 586  out_disc:
 587         while (i-- > 0) {
 588                 int rc2;
 589                 --tgt;
 590                 tgt->ltd_active = 0;
 591                 if (tgt->ltd_exp) {
 592                         --lmv->desc.ld_active_tgt_count;
 593                         rc2 = obd_disconnect(tgt->ltd_exp);
 594                         if (rc2) {
 595                                 CERROR("error: LMV target %s disconnect on "
 596                                        "MDC idx %d: error %d\n",
 597                                        tgt->ltd_uuid.uuid, i, rc2);
 598                         }
 599                 }
 600         }
 601         class_disconnect(lmv->exp);
 602         lmv_init_unlock(lmv);
 603         RETURN(rc);
 604 }
 605
 606 static int lmv_disconnect_mdc(struct obd_device *obd, struct lmv_tgt_desc *tgt)
 607 {
 608 #ifdef __KERNEL__
 609         struct proc_dir_entry *lmv_proc_dir;
 610 #endif
 611         struct lmv_obd *lmv = &obd->u.lmv;
 612         struct obd_device *mdc_obd;
 613         int rc;
 614         ENTRY;
 615
 616         LASSERT(tgt != NULL);
 617         LASSERT(obd != NULL);
 618
 619         mdc_obd = class_exp2obd(tgt->ltd_exp);
 620
 621         if (mdc_obd)
 622                 mdc_obd->obd_no_recov = obd->obd_no_recov;
 623
 624 #ifdef __KERNEL__
 625         lmv_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 626         if (lmv_proc_dir) {
 627                 struct proc_dir_entry *mdc_symlink;
 628
 629                 mdc_symlink = lprocfs_srch(lmv_proc_dir, mdc_obd->obd_name);
 630                 if (mdc_symlink) {
 631                         lprocfs_remove(&mdc_symlink);
 632                 } else {
 633                         CERROR("/proc/fs/lustre/%s/%s/target_obds/%s missing\n",
 634                                obd->obd_type->typ_name, obd->obd_name,
 635                                mdc_obd->obd_name);
 636                 }
 637         }
 638 #endif
 639         rc = obd_fid_fini(tgt->ltd_exp);
 640         if (rc)
 641                 CERROR("Can't finanize fids factory\n");
 642
 643         CDEBUG(D_OTHER, "Disconnected from %s(%s) successfully\n",
 644                tgt->ltd_exp->exp_obd->obd_name,
 645                tgt->ltd_exp->exp_obd->obd_uuid.uuid);
 646
 647         obd_register_observer(tgt->ltd_exp->exp_obd, NULL);
 648         rc = obd_disconnect(tgt->ltd_exp);
 649         if (rc) {
 650                 if (tgt->ltd_active) {
 651                         CERROR("Target %s disconnect error %d\n",
 652                                tgt->ltd_uuid.uuid, rc);
 653                 }
 654         }
 655
 656         lmv_activate_target(lmv, tgt, 0);
 657         tgt->ltd_exp = NULL;
 658         RETURN(0);
 659 }
 660
 661 static int lmv_disconnect(struct obd_export *exp)
 662 {
 663         struct obd_device *obd = class_exp2obd(exp);
 664 #ifdef __KERNEL__
 665         struct proc_dir_entry *lmv_proc_dir;
 666 #endif
 667         struct lmv_obd *lmv = &obd->u.lmv;
 668         int rc, i;
 669         ENTRY;
 670
 671         if (!lmv->tgts)
 672                 goto out_local;
 673
 674         /* Only disconnect the underlying layers on the final disconnect. */
 675         lmv->refcount--;
 676         if (lmv->refcount != 0)
 677                 goto out_local;
 678
 679         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 680                 if (lmv->tgts[i].ltd_exp == NULL)
 681                         continue;
 682                 lmv_disconnect_mdc(obd, &lmv->tgts[i]);
 683         }
 684
 685 #ifdef __KERNEL__
 686         lmv_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 687         if (lmv_proc_dir) {
 688                 lprocfs_remove(&lmv_proc_dir);
 689         } else {
 690                 CERROR("/proc/fs/lustre/%s/%s/target_obds missing\n",
 691                        obd->obd_type->typ_name, obd->obd_name);
 692         }
 693 #endif
 694
 695 out_local:
 696         /*
 697          * This is the case when no real connection is established by
 698          * lmv_check_connect().
 699          */
 700         if (!lmv->connected)
 701                 class_export_put(exp);
 702         rc = class_disconnect(exp);
 703         if (lmv->refcount == 0)
 704                 lmv->connected = 0;
 705         RETURN(rc);
 706 }
 707
 708 static int lmv_iocontrol(unsigned int cmd, struct obd_export *exp,
 709                          int len, void *karg, void *uarg)
 710 {
 711         struct obd_device *obddev = class_exp2obd(exp);
 712         struct lmv_obd *lmv = &obddev->u.lmv;
 713         int i, rc = 0, set = 0;
 714         ENTRY;
 715
 716         if (lmv->desc.ld_tgt_count == 0)
 717                 RETURN(-ENOTTY);
 718
 719         switch (cmd) {
 720         case IOC_OBD_STATFS: {
 721                 struct obd_ioctl_data *data = karg;
 722                 struct obd_device *mdc_obd;
 723                 struct obd_statfs stat_buf = {0};
 724                 __u32 index;
 725
 726                 memcpy(&index, data->ioc_inlbuf2, sizeof(__u32));
 727                 LASSERT(data->ioc_plen1 == sizeof(struct obd_statfs));
 728
 729                 if ((index >= lmv->desc.ld_tgt_count))
 730                         RETURN(-ENODEV);
 731
 732                 if (!lmv->tgts[index].ltd_active)
 733                         RETURN(-ENODATA);
 734
 735                 mdc_obd = class_exp2obd(lmv->tgts[index].ltd_exp);
 736                 if (!mdc_obd)
 737                         RETURN(-EINVAL);
 738
 739                 /* got statfs data */
 740                 rc = obd_statfs(mdc_obd, &stat_buf,
 741                                 cfs_time_current_64() - HZ, 0);
 742                 if (rc)
 743                         RETURN(rc);
 744                 if (copy_to_user(data->ioc_pbuf1, &stat_buf, data->ioc_plen1))
 745                         RETURN(rc);
 746                 /* copy UUID */
 747                 rc = copy_to_user(data->ioc_pbuf2, obd2cli_tgt(mdc_obd),
 748                                   data->ioc_plen2);
 749                 break;
 750         }
 751         default : {
 752                 for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 753                         int err;
 754
 755                         if (lmv->tgts[i].ltd_exp == NULL)
 756                                 continue;
 757
 758                         err = obd_iocontrol(cmd, lmv->tgts[i].ltd_exp, len,
 759                                             karg, uarg);
 760                         if (err) {
 761                                 if (lmv->tgts[i].ltd_active) {
 762                                         CERROR("error: iocontrol MDC %s on MDT"
 763                                                "idx %d cmd %x: err = %d\n",
 764                                                 lmv->tgts[i].ltd_uuid.uuid,
 765                                                 i, cmd, err);
 766                                         if (!rc)
 767                                                 rc = err;
 768                                 }
 769                         } else
 770                                 set = 1;
 771                 }
 772                 if (!set && !rc)
 773                         rc = -EIO;
 774         }
 775         }
 776         RETURN(rc);
 777 }
 778
 779 enum MDS_POLICY {
 780      CHAR_TYPE,
 781      NID_TYPE
 782 };
 783
 784 static int lmv_all_chars_policy(int count, const char *name,
 785                                 int len)
 786 {
 787         unsigned int c = 0;
 788
 789         while (len > 0)
 790                 c += name[--len];
 791         c = c % count;
 792         return c;
 793 }
 794
 795 static int lmv_nid_policy(struct lmv_obd *lmv)
 796 {
 797         struct obd_import *imp = class_exp2cliimp(lmv->tgts[0].ltd_exp);
 798         __u32 id;
 799         /*
 800          * XXX Hack: to get nid we assume that underlying obd device is mdc.
 801          */
 802         id = imp->imp_connection->c_self ^ (imp->imp_connection->c_self >> 32);
 803         return id % lmv->desc.ld_tgt_count;
 804 }
 805
 806 static int lmv_choose_mds(struct lmv_obd *lmv, struct md_op_data *op_data,
 807                           int type)
 808 {
 809         switch (type) {
 810         case CHAR_TYPE:
 811                 return lmv_all_chars_policy(lmv->desc.ld_tgt_count,
 812                                             op_data->op_name,
 813                                             op_data->op_namelen);
 814         case NID_TYPE:
 815                 return lmv_nid_policy(lmv);
 816
 817         default:
 818                 break;
 819         }
 820
 821         CERROR("unsupport type %d \n", type);
 822         return -EINVAL;
 823 }
 824
 825 /* This is _inode_ placement policy function (not name). */
 826 static int lmv_placement_policy(struct obd_device *obd,
 827                                 struct md_op_data *op_data,
 828                                 mdsno_t *mds)
 829 {
 830         struct lmv_obd *lmv = &obd->u.lmv;
 831         struct lmv_obj *obj;
 832         int rc;
 833         ENTRY;
 834
 835         LASSERT(mds != NULL);
 836
 837         /*
 838          * Allocate new fid on target according to operation type and parent
 839          * home mds.
 840          */
 841         obj = lmv_obj_grab(obd, &op_data->op_fid1);
 842         if (obj != NULL || op_data->op_name == NULL ||
 843             op_data->op_opc != LUSTRE_OPC_MKDIR) {
 844                 /*
 845                  * Allocate fid for non-dir or for null name or for case parent
 846                  * dir is split.
 847                  */
 848                 if (obj) {
 849                         lmv_obj_put(obj);
 850
 851                         /*
 852                          * If we have this flag turned on, and we see that
 853                          * parent dir is split, this means, that caller did not
 854                          * notice split yet. This is race and we would like to
 855                          * let caller know that.
 856                          */
 857                         if (op_data->op_bias & MDS_CHECK_SPLIT)
 858                                 RETURN(-ERESTART);
 859                 }
 860
 861                 /*
 862                  * Allocate new fid on same mds where parent fid is located and
 863                  * where operation will be sent. In case of split dir, ->op_fid1
 864                  * and ->op_mds here will contain fid and mds of slave directory
 865                  * object (assigned by caller).
 866                  */
 867                 *mds = op_data->op_mds;
 868                 rc = 0;
 869
 870 #if 0
 871                 /* XXX: This should be removed later wehn we sure it is not
 872                  * needed. */
 873                 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, mds);
 874                 if (rc)
 875                         GOTO(out, rc);
 876 #endif
 877         } else {
 878                 /*
 879                  * Parent directory is not split and we want to create a
 880                  * directory in it. Let's calculate where to place it according
 881                  * to name.
 882                  */
 883                 *mds = lmv_choose_mds(lmv, op_data, NID_TYPE);
 884                 rc = 0;
 885         }
 886         EXIT;
 887 #if 0
 888 out:
 889 #endif
 890         if (rc) {
 891                 CERROR("Can't choose MDS, err = %d\n", rc);
 892         } else {
 893                 LASSERT(*mds < lmv->desc.ld_tgt_count);
 894         }
 895
 896         return rc;
 897 }
 898
 899 int __lmv_fid_alloc(struct lmv_obd *lmv, struct lu_fid *fid,
 900                     mdsno_t mds)
 901 {
 902         struct lmv_tgt_desc *tgt = &lmv->tgts[mds];
 903         int rc;
 904         ENTRY;
 905
 906         /* New seq alloc and FLD setup should be atomic. */
 907         down(&tgt->ltd_fid_sem);
 908
 909         /* Asking underlaying tgt layer to allocate new fid. */
 910         rc = obd_fid_alloc(tgt->ltd_exp, fid, NULL);
 911         if (rc > 0) {
 912                 LASSERT(fid_is_sane(fid));
 913
 914                 /* Client switches to new sequence, setup FLD. */
 915                 rc = fld_client_create(&lmv->lmv_fld, fid_seq(fid),
 916                                        mds, NULL);
 917                 if (rc) {
 918                         CERROR("Can't create fld entry, rc %d\n", rc);
 919                         /* Delete just allocated fid sequence */
 920                         obd_fid_delete(tgt->ltd_exp, NULL);
 921                 }
 922         }
 923         up(&tgt->ltd_fid_sem);
 924         RETURN(rc);
 925 }
 926
 927 int lmv_fid_alloc(struct obd_export *exp, struct lu_fid *fid,
 928                   struct md_op_data *op_data)
 929 {
 930         struct obd_device *obd = class_exp2obd(exp);
 931         struct lmv_obd *lmv = &obd->u.lmv;
 932         mdsno_t mds;
 933         int rc;
 934         ENTRY;
 935
 936         LASSERT(op_data != NULL);
 937         LASSERT(fid != NULL);
 938
 939         rc = lmv_placement_policy(obd, op_data, &mds);
 940         if (rc) {
 941                 CERROR("Can't get target for allocating fid, "
 942                        "rc %d\n", rc);
 943                 RETURN(rc);
 944         }
 945
 946         rc = __lmv_fid_alloc(lmv, fid, mds);
 947         if (rc) {
 948                 CERROR("Can't alloc new fid, rc %d\n", rc);
 949                 RETURN(rc);
 950         }
 951
 952         RETURN(rc);
 953 }
 954
 955 static int lmv_fid_delete(struct obd_export *exp, const struct lu_fid *fid)
 956 {
 957         ENTRY;
 958
 959         LASSERT(exp && fid);
 960         if (lmv_obj_delete(exp, fid)) {
 961                 CDEBUG(D_OTHER, "lmv object "DFID" is destroyed.\n",
 962                        PFID(fid));
 963         }
 964         RETURN(0);
 965 }
 966
 967 static int lmv_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
 968 {
 969         struct lmv_obd *lmv = &obd->u.lmv;
 970         struct lprocfs_static_vars lvars;
 971         struct lmv_desc *desc;
 972         int rc, i = 0;
 973         ENTRY;
 974
 975         if (LUSTRE_CFG_BUFLEN(lcfg, 1) < 1) {
 976                 CERROR("LMV setup requires a descriptor\n");
 977                 RETURN(-EINVAL);
 978         }
 979
 980         desc = (struct lmv_desc *)lustre_cfg_buf(lcfg, 1);
 981         if (sizeof(*desc) > LUSTRE_CFG_BUFLEN(lcfg, 1)) {
 982                 CERROR("descriptor size wrong: %d > %d\n",
 983                        (int)sizeof(*desc), LUSTRE_CFG_BUFLEN(lcfg, 1));
 984                 RETURN(-EINVAL);
 985         }
 986
 987         lmv->tgts_size = LMV_MAX_TGT_COUNT * sizeof(struct lmv_tgt_desc);
 988
 989         OBD_ALLOC(lmv->tgts, lmv->tgts_size);
 990         if (lmv->tgts == NULL)
 991                 RETURN(-ENOMEM);
 992
 993         for (i = 0; i < LMV_MAX_TGT_COUNT; i++) {
 994                 sema_init(&lmv->tgts[i].ltd_fid_sem, 1);
 995                 lmv->tgts[i].ltd_idx = i;
 996         }
 997
 998         lmv->datas_size = LMV_MAX_TGT_COUNT * sizeof(struct obd_connect_data);
 999
1000         OBD_ALLOC(lmv->datas, lmv->datas_size);
1001         if (lmv->datas == NULL)
1002                 GOTO(out_free_tgts, rc = -ENOMEM);
1003
1004         obd_str2uuid(&lmv->desc.ld_uuid, desc->ld_uuid.uuid);
1005         lmv->desc.ld_tgt_count = 0;
1006         lmv->desc.ld_active_tgt_count = 0;
1007         lmv->max_cookiesize = 0;
1008         lmv->max_def_easize = 0;
1009         lmv->max_easize = 0;
1010
1011         spin_lock_init(&lmv->lmv_lock);
1012         sema_init(&lmv->init_sem, 1);
1013
1014         rc = lmv_obj_setup(obd);
1015         if (rc) {
1016                 CERROR("Can't setup LMV object manager, "
1017                        "error %d.\n", rc);
1018                 GOTO(out_free_datas, rc);
1019         }
1020
1021         lprocfs_lmv_init_vars(&lvars);
1022         lprocfs_obd_setup(obd, lvars.obd_vars);
1023 #ifdef LPROCFS
1024         {
1025                 rc = lprocfs_seq_create(obd->obd_proc_entry, "target_obd_status",
1026                                         0444, &lmv_proc_target_fops, obd);
1027                 if (rc)
1028                         CWARN("Error adding the target_obd_status file\n");
1029        }
1030 #endif
1031         rc = fld_client_init(&lmv->lmv_fld, obd->obd_name,
1032                              LUSTRE_CLI_FLD_HASH_DHT);
1033         if (rc) {
1034                 CERROR("can't init FLD, err %d\n",
1035                        rc);
1036                 GOTO(out_free_datas, rc);
1037         }
1038
1039         RETURN(0);
1040
1041 out_free_datas:
1042         OBD_FREE(lmv->datas, lmv->datas_size);
1043         lmv->datas = NULL;
1044 out_free_tgts:
1045         OBD_FREE(lmv->tgts, lmv->tgts_size);
1046         lmv->tgts = NULL;
1047         return rc;
1048 }
1049
1050 static int lmv_cleanup(struct obd_device *obd)
1051 {
1052         struct lmv_obd *lmv = &obd->u.lmv;
1053         ENTRY;
1054
1055         fld_client_fini(&lmv->lmv_fld);
1056         lprocfs_obd_cleanup(obd);
1057         lmv_obj_cleanup(obd);
1058         OBD_FREE(lmv->datas, lmv->datas_size);
1059         OBD_FREE(lmv->tgts, lmv->tgts_size);
1060
1061         RETURN(0);
1062 }
1063
1064 static int lmv_process_config(struct obd_device *obd, obd_count len, void *buf)
1065 {
1066         struct lustre_cfg *lcfg = buf;
1067         struct obd_uuid tgt_uuid;
1068         int rc;
1069         ENTRY;
1070
1071         switch(lcfg->lcfg_command) {
1072         case LCFG_ADD_MDC:
1073                 if (LUSTRE_CFG_BUFLEN(lcfg, 1) > sizeof(tgt_uuid.uuid))
1074                         GOTO(out, rc = -EINVAL);
1075
1076                 obd_str2uuid(&tgt_uuid, lustre_cfg_string(lcfg, 1));
1077                 rc = lmv_add_target(obd, &tgt_uuid);
1078                 GOTO(out, rc);
1079         default: {
1080                 CERROR("Unknown command: %d\n", lcfg->lcfg_command);
1081                 GOTO(out, rc = -EINVAL);
1082         }
1083         }
1084 out:
1085         RETURN(rc);
1086 }
1087
1088 static int lmv_statfs(struct obd_device *obd, struct obd_statfs *osfs,
1089                       __u64 max_age, __u32 flags)
1090 {
1091         struct lmv_obd *lmv = &obd->u.lmv;
1092         struct obd_statfs *temp;
1093         int rc = 0, i;
1094         ENTRY;
1095
1096         rc = lmv_check_connect(obd);
1097         if (rc)
1098                 RETURN(rc);
1099
1100         OBD_ALLOC(temp, sizeof(*temp));
1101         if (temp == NULL)
1102                 RETURN(-ENOMEM);
1103
1104         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
1105                 if (lmv->tgts[i].ltd_exp == NULL)
1106                         continue;
1107
1108                 rc = obd_statfs(lmv->tgts[i].ltd_exp->exp_obd, temp,
1109                                 max_age, flags);
1110                 if (rc) {
1111                         CERROR("can't stat MDS #%d (%s), error %d\n", i,
1112                                lmv->tgts[i].ltd_exp->exp_obd->obd_name,
1113                                rc);
1114                         GOTO(out_free_temp, rc);
1115                 }
1116                 if (i == 0) {
1117                         *osfs = *temp;
1118                 } else {
1119                         osfs->os_bavail += temp->os_bavail;
1120                         osfs->os_blocks += temp->os_blocks;
1121                         osfs->os_ffree += temp->os_ffree;
1122                         osfs->os_files += temp->os_files;
1123                 }
1124         }
1125
1126         EXIT;
1127 out_free_temp:
1128         OBD_FREE(temp, sizeof(*temp));
1129         return rc;
1130 }
1131
1132 static int lmv_getstatus(struct obd_export *exp,
1133                          struct lu_fid *fid,
1134                          struct obd_capa **pc)
1135 {
1136         struct obd_device *obd = exp->exp_obd;
1137         struct lmv_obd *lmv = &obd->u.lmv;
1138         int rc;
1139         ENTRY;
1140
1141         rc = lmv_check_connect(obd);
1142         if (rc)
1143                 RETURN(rc);
1144
1145         rc = md_getstatus(lmv->tgts[0].ltd_exp, fid, pc);
1146
1147         RETURN(rc);
1148 }
1149
1150 static int lmv_getxattr(struct obd_export *exp, const struct lu_fid *fid,
1151                         struct obd_capa *oc, obd_valid valid, const char *name,
1152                         const char *input, int input_size, int output_size,
1153                         int flags, struct ptlrpc_request **request)
1154 {
1155         struct obd_device *obd = exp->exp_obd;
1156         struct lmv_obd *lmv = &obd->u.lmv;
1157         struct obd_export *tgt_exp;
1158         int rc;
1159         ENTRY;
1160
1161         rc = lmv_check_connect(obd);
1162         if (rc)
1163                 RETURN(rc);
1164
1165         tgt_exp = lmv_find_export(lmv, fid);
1166         if (IS_ERR(tgt_exp))
1167                 RETURN(PTR_ERR(tgt_exp));
1168
1169         rc = md_getxattr(tgt_exp, fid, oc, valid, name, input, input_size,
1170                          output_size, flags, request);
1171
1172         RETURN(rc);
1173 }
1174
1175 static int lmv_setxattr(struct obd_export *exp, const struct lu_fid *fid,
1176                         struct obd_capa *oc, obd_valid valid, const char *name,
1177                         const char *input, int input_size, int output_size,
1178                         int flags, __u32 suppgid,
1179                         struct ptlrpc_request **request)
1180 {
1181         struct obd_device *obd = exp->exp_obd;
1182         struct lmv_obd *lmv = &obd->u.lmv;
1183         struct obd_export *tgt_exp;
1184         int rc;
1185         ENTRY;
1186
1187         rc = lmv_check_connect(obd);
1188         if (rc)
1189                 RETURN(rc);
1190
1191         tgt_exp = lmv_find_export(lmv, fid);
1192         if (IS_ERR(tgt_exp))
1193                 RETURN(PTR_ERR(tgt_exp));
1194
1195         rc = md_setxattr(tgt_exp, fid, oc, valid, name,
1196                          input, input_size, output_size, flags, suppgid,
1197                          request);
1198
1199         RETURN(rc);
1200 }
1201
1202 static int lmv_getattr(struct obd_export *exp, const struct lu_fid *fid,
1203                        struct obd_capa *oc, obd_valid valid, int ea_size,
1204                        struct ptlrpc_request **request)
1205 {
1206         struct obd_device *obd = exp->exp_obd;
1207         struct lmv_obd *lmv = &obd->u.lmv;
1208         struct obd_export *tgt_exp;
1209         struct lmv_obj *obj;
1210         int rc, i;
1211         ENTRY;
1212
1213         rc = lmv_check_connect(obd);
1214         if (rc)
1215                 RETURN(rc);
1216
1217         tgt_exp = lmv_find_export(lmv, fid);
1218         if (IS_ERR(tgt_exp))
1219                 RETURN(PTR_ERR(tgt_exp));
1220
1221         rc = md_getattr(tgt_exp, fid, oc, valid, ea_size, request);
1222         if (rc)
1223                 RETURN(rc);
1224
1225         obj = lmv_obj_grab(obd, fid);
1226
1227         CDEBUG(D_OTHER, "GETATTR for "DFID" %s\n", PFID(fid),
1228                obj ? "(split)" : "");
1229
1230         /*
1231          * If object is split, then we loop over all the slaves and gather size
1232          * attribute. In ideal world we would have to gather also mds field from
1233          * all slaves, as object is spread over the cluster and this is
1234          * definitely interesting information and it is not good to loss it,
1235          * but...
1236          */
1237         if (obj) {
1238                 struct mdt_body *body;
1239
1240                 if (*request == NULL) {
1241                         lmv_obj_put(obj);
1242                         RETURN(rc);
1243                 }
1244
1245                 body = req_capsule_server_get(&(*request)->rq_pill,
1246                                               &RMF_MDT_BODY);
1247                 LASSERT(body != NULL);
1248
1249                 lmv_obj_lock(obj);
1250
1251                 for (i = 0; i < obj->lo_objcount; i++) {
1252                         if (lmv->tgts[i].ltd_exp == NULL) {
1253                                 CWARN("%s: NULL export for %d\n",
1254                                       obd->obd_name, i);
1255                                 continue;
1256                         }
1257
1258                         /* skip master obj. */
1259                         if (lu_fid_eq(&obj->lo_fid, &obj->lo_inodes[i].li_fid))
1260                                 continue;
1261
1262                         lmv_update_body(body, &obj->lo_inodes[i]);
1263                 }
1264
1265                 lmv_obj_unlock(obj);
1266                 lmv_obj_put(obj);
1267         }
1268
1269         RETURN(rc);
1270 }
1271
1272 static int lmv_change_cbdata(struct obd_export *exp, const struct lu_fid *fid,
1273                              ldlm_iterator_t it, void *data)
1274 {
1275         struct obd_device *obd = exp->exp_obd;
1276         struct lmv_obd *lmv = &obd->u.lmv;
1277         int i, rc;
1278         ENTRY;
1279
1280         rc = lmv_check_connect(obd);
1281         if (rc)
1282                 RETURN(rc);
1283
1284         CDEBUG(D_OTHER, "CBDATA for "DFID"\n", PFID(fid));
1285
1286         /*
1287          * With CMD every object can have two locks in different namespaces:
1288          * lookup lock in space of mds storing direntry and update/open lock in
1289          * space of mds storing inode.
1290          */
1291         for (i = 0; i < lmv->desc.ld_tgt_count; i++)
1292                 md_change_cbdata(lmv->tgts[i].ltd_exp, fid, it, data);
1293
1294         RETURN(0);
1295 }
1296
1297 static int lmv_close(struct obd_export *exp,
1298                      struct md_op_data *op_data,
1299                      struct md_open_data *mod,
1300                      struct ptlrpc_request **request)
1301 {
1302         struct obd_device *obd = exp->exp_obd;
1303         struct lmv_obd *lmv = &obd->u.lmv;
1304         struct obd_export *tgt_exp;
1305         int rc;
1306         ENTRY;
1307
1308         rc = lmv_check_connect(obd);
1309         if (rc)
1310                 RETURN(rc);
1311
1312         tgt_exp = lmv_find_export(lmv, &op_data->op_fid1);
1313         if (IS_ERR(tgt_exp))
1314                 RETURN(PTR_ERR(tgt_exp));
1315
1316         CDEBUG(D_OTHER, "CLOSE "DFID"\n", PFID(&op_data->op_fid1));
1317         rc = md_close(tgt_exp, op_data, mod, request);
1318         RETURN(rc);
1319 }
1320
1321 /*
1322  * Called in the case MDS returns -ERESTART on create on open, what means that
1323  * directory is split and its LMV presentation object has to be updated.
1324  */
1325 int lmv_handle_split(struct obd_export *exp, const struct lu_fid *fid)
1326 {
1327         struct obd_device *obd = exp->exp_obd;
1328         struct lmv_obd *lmv = &obd->u.lmv;
1329         struct ptlrpc_request *req = NULL;
1330         struct obd_export *tgt_exp;
1331         struct lmv_obj *obj;
1332         struct lustre_md md;
1333         int mealen, rc;
1334         __u64 valid;
1335         ENTRY;
1336
1337         md.mea = NULL;
1338         mealen = lmv_get_easize(lmv);
1339
1340         valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA | OBD_MD_MEA;
1341
1342         tgt_exp = lmv_find_export(lmv, fid);
1343         if (IS_ERR(tgt_exp))
1344                 RETURN(PTR_ERR(tgt_exp));
1345
1346         /* time to update mea of parent fid */
1347         rc = md_getattr(tgt_exp, fid, NULL, valid, mealen, &req);
1348         if (rc) {
1349                 CERROR("md_getattr() failed, error %d\n", rc);
1350                 GOTO(cleanup, rc);
1351         }
1352
1353         rc = md_get_lustre_md(tgt_exp, req, NULL, exp, &md);
1354         if (rc) {
1355                 CERROR("mdc_get_lustre_md() failed, error %d\n", rc);
1356                 GOTO(cleanup, rc);
1357         }
1358
1359         if (md.mea == NULL)
1360                 GOTO(cleanup, rc = -ENODATA);
1361
1362         obj = lmv_obj_create(exp, fid, md.mea);
1363         if (IS_ERR(obj))
1364                 rc = PTR_ERR(obj);
1365         else
1366                 lmv_obj_put(obj);
1367
1368         /* XXX LOV STACKING */
1369         obd_free_memmd(exp, (void *)&md.mea);
1370
1371         EXIT;
1372 cleanup:
1373         if (req)
1374                 ptlrpc_req_finished(req);
1375         return rc;
1376 }
1377
1378 int lmv_create(struct obd_export *exp, struct md_op_data *op_data,
1379                const void *data, int datalen, int mode, __u32 uid,
1380                __u32 gid, __u32 cap_effective,  __u64 rdev,
1381                struct ptlrpc_request **request)
1382 {
1383         struct obd_device *obd = exp->exp_obd;
1384         struct lmv_obd *lmv = &obd->u.lmv;
1385         struct obd_export *tgt_exp;
1386         struct lmv_obj *obj;
1387         int rc, loop = 0;
1388         ENTRY;
1389
1390         rc = lmv_check_connect(obd);
1391         if (rc)
1392                 RETURN(rc);
1393
1394         if (!lmv->desc.ld_active_tgt_count)
1395                 RETURN(-EIO);
1396 repeat:
1397         ++loop;
1398         LASSERT(loop <= 2);
1399         obj = lmv_obj_grab(obd, &op_data->op_fid1);
1400         if (obj) {
1401                 int mea_idx;
1402
1403                 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
1404                                        op_data->op_name, op_data->op_namelen);
1405                 op_data->op_fid1 = obj->lo_inodes[mea_idx].li_fid;
1406                 op_data->op_bias &= ~MDS_CHECK_SPLIT;
1407                 op_data->op_mds = obj->lo_inodes[mea_idx].li_mds;
1408                 tgt_exp = lmv_get_export(lmv, op_data->op_mds);
1409                 lmv_obj_put(obj);
1410         } else {
1411                 struct lmv_tgt_desc *tgt;
1412
1413                 tgt = lmv_find_target(lmv, &op_data->op_fid1);
1414                 op_data->op_bias |= MDS_CHECK_SPLIT;
1415                 op_data->op_mds = tgt->ltd_idx;
1416                 tgt_exp = tgt->ltd_exp;
1417         }
1418
1419         if (IS_ERR(tgt_exp))
1420                 RETURN(PTR_ERR(tgt_exp));
1421
1422         rc = lmv_fid_alloc(exp, &op_data->op_fid2, op_data);
1423         if (rc == -ERESTART)
1424                 goto repeat;
1425         else if (rc)
1426                 RETURN(rc);
1427
1428         CDEBUG(D_OTHER, "CREATE '%*s' on "DFID"\n", op_data->op_namelen,
1429                op_data->op_name, PFID(&op_data->op_fid1));
1430
1431         op_data->op_flags |= MF_MDC_CANCEL_FID1;
1432         rc = md_create(tgt_exp, op_data, data, datalen, mode, uid, gid,
1433                        cap_effective, rdev, request);
1434         if (rc == 0) {
1435                 if (*request == NULL)
1436                         RETURN(rc);
1437                 CDEBUG(D_OTHER, "created - "DFID"\n", PFID(&op_data->op_fid1));
1438         } else if (rc == -ERESTART) {
1439                 LASSERT(*request != NULL);
1440                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *request,
1441                           "Got -ERESTART during create!\n");
1442                 ptlrpc_req_finished(*request);
1443                 *request = NULL;
1444
1445                 /*
1446                  * Directory got split. Time to update local object and repeat
1447                  * the request with proper MDS.
1448                  */
1449                 rc = lmv_handle_split(exp, &op_data->op_fid1);
1450                 if (rc == 0) {
1451                         rc = lmv_alloc_slave_fids(obd, &op_data->op_fid1,
1452                                                   op_data, &op_data->op_fid2);
1453                         if (rc)
1454                                 RETURN(rc);
1455                         goto repeat;
1456                 }
1457         }
1458         RETURN(rc);
1459 }
1460
1461 static int lmv_done_writing(struct obd_export *exp,
1462                             struct md_op_data *op_data,
1463                             struct md_open_data *mod)
1464 {
1465         struct obd_device *obd = exp->exp_obd;
1466         struct lmv_obd *lmv = &obd->u.lmv;
1467         struct obd_export *tgt_exp;
1468         int rc;
1469         ENTRY;
1470
1471         rc = lmv_check_connect(obd);
1472         if (rc)
1473                 RETURN(rc);
1474
1475         tgt_exp = lmv_find_export(lmv, &op_data->op_fid1);
1476         if (IS_ERR(tgt_exp))
1477                 RETURN(PTR_ERR(tgt_exp));
1478
1479         rc = md_done_writing(tgt_exp, op_data, mod);
1480         RETURN(rc);
1481 }
1482
1483 static int
1484 lmv_enqueue_slaves(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
1485                    struct lookup_intent *it, struct md_op_data *op_data,
1486                    struct lustre_handle *lockh, void *lmm, int lmmsize)
1487 {
1488         struct obd_device *obd = exp->exp_obd;
1489         struct lmv_obd *lmv = &obd->u.lmv;
1490         struct lmv_stripe_md *mea = op_data->op_mea1;
1491         struct md_op_data *op_data2;
1492         struct obd_export *tgt_exp;
1493         int i, rc = 0;
1494         ENTRY;
1495
1496         OBD_ALLOC_PTR(op_data2);
1497         if (op_data2 == NULL)
1498                 RETURN(-ENOMEM);
1499
1500         LASSERT(mea != NULL);
1501         for (i = 0; i < mea->mea_count; i++) {
1502                 memset(op_data2, 0, sizeof(*op_data2));
1503                 op_data2->op_fid1 = mea->mea_ids[i];
1504                 op_data2->op_bias = 0;
1505
1506                 tgt_exp = lmv_find_export(lmv, &op_data2->op_fid1);
1507                 if (IS_ERR(tgt_exp))
1508                         GOTO(cleanup, rc = PTR_ERR(tgt_exp));
1509
1510                 if (tgt_exp == NULL)
1511                         continue;
1512
1513                 rc = md_enqueue(tgt_exp, einfo, it, op_data2,
1514                                 lockh + i, lmm, lmmsize, 0);
1515
1516                 CDEBUG(D_OTHER, "take lock on slave "DFID" -> %d/%d\n",
1517                        PFID(&mea->mea_ids[i]), rc, it->d.lustre.it_status);
1518
1519                 if (rc)
1520                         GOTO(cleanup, rc);
1521
1522                 if (it->d.lustre.it_data) {
1523                         struct ptlrpc_request *req;
1524                         req = (struct ptlrpc_request *)it->d.lustre.it_data;
1525                         ptlrpc_req_finished(req);
1526                 }
1527
1528                 if (it->d.lustre.it_status)
1529                         GOTO(cleanup, rc = it->d.lustre.it_status);
1530         }
1531
1532         EXIT;
1533 cleanup:
1534         OBD_FREE_PTR(op_data2);
1535
1536         if (rc != 0) {
1537                 /* drop all taken locks */
1538                 while (--i >= 0) {
1539                         if (lockh[i].cookie)
1540                                 ldlm_lock_decref(lockh + i, einfo->ei_mode);
1541                         lockh[i].cookie = 0;
1542                 }
1543         }
1544         return rc;
1545 }
1546
1547 static int
1548 lmv_enqueue_remote(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
1549                    struct lookup_intent *it, struct md_op_data *op_data,
1550                    struct lustre_handle *lockh, void *lmm, int lmmsize,
1551                    int extra_lock_flags)
1552 {
1553         struct ptlrpc_request *req = it->d.lustre.it_data;
1554         struct obd_device *obd = exp->exp_obd;
1555         struct lmv_obd *lmv = &obd->u.lmv;
1556         struct lustre_handle plock;
1557         struct obd_export *tgt_exp;
1558         struct md_op_data *rdata;
1559         struct lu_fid fid_copy;
1560         struct mdt_body *body;
1561         int rc = 0, pmode;
1562         ENTRY;
1563
1564         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
1565         LASSERT(body != NULL);
1566
1567         if (!(body->valid & OBD_MD_MDS))
1568                 RETURN(0);
1569
1570         CDEBUG(D_OTHER, "ENQUEUE '%s' on "DFID" -> "DFID"\n",
1571                LL_IT2STR(it), PFID(&op_data->op_fid1), PFID(&body->fid1));
1572
1573         /* We got LOOKUP lock, but we really need attrs */
1574         pmode = it->d.lustre.it_lock_mode;
1575         LASSERT(pmode != 0);
1576         memcpy(&plock, lockh, sizeof(plock));
1577         it->d.lustre.it_lock_mode = 0;
1578         it->d.lustre.it_data = NULL;
1579         fid_copy = body->fid1;
1580
1581         it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
1582         ptlrpc_req_finished(req);
1583
1584         tgt_exp = lmv_find_export(lmv, &fid_copy);
1585         if (IS_ERR(tgt_exp))
1586                 GOTO(out, rc = PTR_ERR(tgt_exp));
1587
1588         OBD_ALLOC_PTR(rdata);
1589         if (rdata == NULL)
1590                 GOTO(out, rc = -ENOMEM);
1591
1592         rdata->op_fid1 = fid_copy;
1593         rdata->op_bias = MDS_CROSS_REF;
1594
1595         rc = md_enqueue(tgt_exp, einfo, it, rdata, lockh,
1596                         lmm, lmmsize, extra_lock_flags);
1597         OBD_FREE_PTR(rdata);
1598         EXIT;
1599 out:
1600         ldlm_lock_decref(&plock, pmode);
1601         return rc;
1602 }
1603
1604 static int
1605 lmv_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
1606             struct lookup_intent *it, struct md_op_data *op_data,
1607             struct lustre_handle *lockh, void *lmm, int lmmsize,
1608             int extra_lock_flags)
1609 {
1610         struct obd_device *obd = exp->exp_obd;
1611         struct lmv_obd *lmv = &obd->u.lmv;
1612         struct obd_export *tgt_exp = NULL;
1613         struct lmv_obj *obj;
1614         int rc;
1615         ENTRY;
1616
1617         rc = lmv_check_connect(obd);
1618         if (rc)
1619                 RETURN(rc);
1620
1621         if (op_data->op_mea1 && it->it_op == IT_UNLINK) {
1622                 rc = lmv_enqueue_slaves(exp, einfo, it, op_data,
1623                                         lockh, lmm, lmmsize);
1624                 RETURN(rc);
1625         }
1626
1627         if (op_data->op_namelen) {
1628                 obj = lmv_obj_grab(obd, &op_data->op_fid1);
1629                 if (obj) {
1630                         int mea_idx;
1631
1632                         /* directory is split. look for right mds for this
1633                          * name */
1634                         mea_idx = raw_name2idx(obj->lo_hashtype,
1635                                                obj->lo_objcount,
1636                                                (char *)op_data->op_name,
1637                                                op_data->op_namelen);
1638                         op_data->op_fid1 = obj->lo_inodes[mea_idx].li_fid;
1639                         tgt_exp = lmv_get_export(lmv, obj->lo_inodes[mea_idx].li_mds);
1640                         lmv_obj_put(obj);
1641                 }
1642         }
1643
1644         if (tgt_exp == NULL)
1645                 tgt_exp = lmv_find_export(lmv, &op_data->op_fid1);
1646         if (IS_ERR(tgt_exp))
1647                 RETURN(PTR_ERR(tgt_exp));
1648
1649         CDEBUG(D_OTHER, "ENQUEUE '%s' on "DFID"\n", LL_IT2STR(it),
1650                PFID(&op_data->op_fid1));
1651
1652         rc = md_enqueue(tgt_exp, einfo, it, op_data, lockh,
1653                         lmm, lmmsize, extra_lock_flags);
1654
1655         if (rc == 0 && it->it_op == IT_OPEN)
1656                 rc = lmv_enqueue_remote(exp, einfo, it, op_data, lockh,
1657                                         lmm, lmmsize, extra_lock_flags);
1658         RETURN(rc);
1659 }
1660
1661 static int
1662 lmv_getattr_name(struct obd_export *exp, const struct lu_fid *fid,
1663                  struct obd_capa *oc, const char *filename, int namelen,
1664                  obd_valid valid, int ea_size, __u32 suppgid,
1665                  struct ptlrpc_request **request)
1666 {
1667         struct obd_device *obd = exp->exp_obd;
1668         struct lmv_obd *lmv = &obd->u.lmv;
1669         struct lu_fid rid = *fid;
1670         struct obd_export *tgt_exp;
1671         struct mdt_body *body;
1672         struct lmv_obj *obj;
1673         int rc, loop = 0;
1674         ENTRY;
1675
1676         rc = lmv_check_connect(obd);
1677         if (rc)
1678                 RETURN(rc);
1679
1680 repeat:
1681         ++loop;
1682         LASSERT(loop <= 2);
1683         obj = lmv_obj_grab(obd, &rid);
1684         if (obj) {
1685                 int mea_idx;
1686
1687                 /* Directory is split. Look for right mds for this name */
1688                 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
1689                                        filename, namelen - 1);
1690                 rid = obj->lo_inodes[mea_idx].li_fid;
1691                 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[mea_idx].li_mds);
1692                 lmv_obj_put(obj);
1693                 valid &= ~OBD_MD_FLCKSPLIT;
1694         } else {
1695                 tgt_exp = lmv_find_export(lmv, &rid);
1696                 valid |= OBD_MD_FLCKSPLIT;
1697         }
1698         if (IS_ERR(tgt_exp))
1699                 RETURN(PTR_ERR(tgt_exp));
1700
1701         CDEBUG(D_OTHER, "getattr_name for %*s on "DFID" -> "DFID"\n",
1702                namelen, filename, PFID(fid), PFID(&rid));
1703
1704         rc = md_getattr_name(tgt_exp, &rid, oc, filename, namelen, valid,
1705                              ea_size, suppgid, request);
1706         if (rc == 0) {
1707                 body = req_capsule_server_get(&(*request)->rq_pill,
1708                                               &RMF_MDT_BODY);
1709                 LASSERT(body != NULL);
1710
1711                 if (body->valid & OBD_MD_MDS) {
1712                         struct ptlrpc_request *req = NULL;
1713
1714                         rid = body->fid1;
1715                         CDEBUG(D_OTHER, "request attrs for "DFID"\n",
1716                                PFID(&rid));
1717
1718                         tgt_exp = lmv_find_export(lmv, &rid);
1719                         if (IS_ERR(tgt_exp)) {
1720                                 ptlrpc_req_finished(*request);
1721                                 RETURN(PTR_ERR(tgt_exp));
1722                         }
1723
1724                         rc = md_getattr_name(tgt_exp, &rid, NULL, NULL, 1,
1725                                              valid | OBD_MD_FLCROSSREF,
1726                                              ea_size, suppgid, &req);
1727                         ptlrpc_req_finished(*request);
1728                         *request = req;
1729                 }
1730         } else if (rc == -ERESTART) {
1731                 LASSERT(*request != NULL);
1732                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *request,
1733                           "Got -ERESTART during getattr!\n");
1734                 ptlrpc_req_finished(*request);
1735                 *request = NULL;
1736
1737                 /*
1738                  * Directory got split. Time to update local object and repeat
1739                  * the request with proper MDS.
1740                  */
1741                 rc = lmv_handle_split(exp, &rid);
1742                 if (rc == 0)
1743                         goto repeat;
1744         }
1745         RETURN(rc);
1746 }
1747
1748 #define md_op_data_fid(op_data, fl)                     \
1749         (fl == MF_MDC_CANCEL_FID1 ? &op_data->op_fid1 : \
1750          fl == MF_MDC_CANCEL_FID2 ? &op_data->op_fid2 : \
1751          fl == MF_MDC_CANCEL_FID3 ? &op_data->op_fid3 : \
1752          fl == MF_MDC_CANCEL_FID4 ? &op_data->op_fid4 : \
1753          NULL)
1754
1755 /* @tgt_exp is the export the metadata request is sent.
1756  * @fid_exp is the export the cancel should be sent for the current fid.
1757  * if @fid_exp is NULL, the export is found for the current fid.
1758  * @op_data keeps the current fid, which is pointed through @flag.
1759  * @mode, @bits -- lock match parameters. */
1760 static int lmv_early_cancel(struct lmv_obd *lmv, struct obd_export *tgt_exp,
1761                             struct obd_export *fid_exp,
1762                             struct md_op_data *op_data,
1763                             ldlm_mode_t mode, int bits, int flag)
1764 {
1765         struct lu_fid *fid = md_op_data_fid(op_data, flag);
1766         ldlm_policy_data_t policy = {{0}};
1767         int rc = 0;
1768         ENTRY;
1769
1770         if (!fid_is_sane(fid))
1771                 RETURN(0);
1772
1773         if (fid_exp == NULL)
1774                 fid_exp = lmv_find_export(lmv, fid);
1775
1776         if (tgt_exp == fid_exp) {
1777                 /* The export is the same as on the target server, cancel
1778                  * will be sent along with the main metadata operation. */
1779                 op_data->op_flags |= flag;
1780                 RETURN(0);
1781         }
1782
1783         policy.l_inodebits.bits = bits;
1784         rc = md_cancel_unused(fid_exp, fid, &policy, mode, LDLM_FL_ASYNC, NULL);
1785         RETURN(rc);
1786 }
1787
1788 #ifdef EARLY_CANCEL_FOR_STRIPED_DIR_IS_READY
1789 /* Check if the fid in @op_data pointed to by flag is of the same export(s)
1790  * as @tgt_exp. Early cancels will be sent later by mdc code, otherwise, call
1791  * md_cancel_unused for child export(s). */
1792 static int lmv_early_cancel_stripes(struct obd_export *exp,
1793                                     struct obd_export *tgt_exp,
1794                                     struct md_op_data *op_data,
1795                                     ldlm_mode_t mode, int bits, int flag)
1796 {
1797         struct lu_fid *fid = md_op_data_fid(op_data, flag);
1798         struct obd_device *obd = exp->exp_obd;
1799         struct lmv_obd *lmv = &obd->u.lmv;
1800         struct obd_export *st_exp;
1801         struct lmv_obj *obj;
1802         int rc = 0;
1803         ENTRY;
1804
1805         if (!fid_is_sane(fid))
1806                 RETURN(0);
1807
1808         obj = lmv_obj_grab(obd, fid);
1809         if (obj) {
1810                 ldlm_policy_data_t policy = {{0}};
1811                 struct lu_fid *st_fid;
1812                 int i;
1813
1814                 policy.l_inodebits.bits = bits;
1815                 for (i = 0; i < obj->lo_objcount; i++) {
1816                         st_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
1817                         st_fid = &obj->lo_inodes[i].li_fid;
1818                         if (tgt_exp != st_exp) {
1819                                 rc = md_cancel_unused(st_exp, st_fid, &policy,
1820                                                       mode, LDLM_FL_ASYNC,
1821                                                       NULL);
1822                                 if (rc)
1823                                         break;
1824                         } else {
1825                                 /* Some export matches to @tgt_exp, do cancel
1826                                  * for its fid in mdc */
1827                                 *fid = *st_fid;
1828                                 op_data->op_flags |= flag;
1829                         }
1830                 }
1831                 lmv_obj_put(obj);
1832         } else {
1833                 rc = lmv_early_cancel(lmv, tgt_exp, NULL, op_data,
1834                                       mode, bits, flag);
1835         }
1836         RETURN(rc);
1837 }
1838 #endif
1839
1840 /*
1841  * llite passes fid of an target inode in op_data->op_fid1 and id of directory in
1842  * op_data->op_fid2
1843  */
1844 static int lmv_link(struct obd_export *exp, struct md_op_data *op_data,
1845                     struct ptlrpc_request **request)
1846 {
1847         struct obd_device *obd = exp->exp_obd;
1848         struct lmv_obd *lmv = &obd->u.lmv;
1849         struct obd_export *tgt_exp;
1850         struct lmv_obj *obj;
1851         int rc, loop = 0;
1852         mdsno_t mds;
1853         ENTRY;
1854
1855         rc = lmv_check_connect(obd);
1856         if (rc)
1857                 RETURN(rc);
1858
1859 repeat:
1860         ++loop;
1861         LASSERT(loop <= 2);
1862         if (op_data->op_namelen != 0) {
1863                 int mea_idx;
1864
1865                 /* Usual link request */
1866                 obj = lmv_obj_grab(obd, &op_data->op_fid2);
1867                 if (obj) {
1868                         mea_idx = raw_name2idx(obj->lo_hashtype,
1869                                                obj->lo_objcount,
1870                                                op_data->op_name,
1871                                                op_data->op_namelen);
1872                         op_data->op_fid2 = obj->lo_inodes[mea_idx].li_fid;
1873                         mds = obj->lo_inodes[mea_idx].li_mds;
1874                         lmv_obj_put(obj);
1875                 } else {
1876                         rc = lmv_fld_lookup(lmv, &op_data->op_fid2, &mds);
1877                         if (rc)
1878                                 RETURN(rc);
1879                 }
1880
1881                 CDEBUG(D_OTHER,"link "DFID":%*s to "DFID"\n",
1882                        PFID(&op_data->op_fid2), op_data->op_namelen,
1883                        op_data->op_name, PFID(&op_data->op_fid1));
1884         } else {
1885                 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
1886                 if (rc)
1887                         RETURN(rc);
1888
1889                 /* request from MDS to acquire i_links for inode by fid1 */
1890                 CDEBUG(D_OTHER, "inc i_nlinks for "DFID"\n",
1891                        PFID(&op_data->op_fid1));
1892         }
1893
1894         CDEBUG(D_OTHER, "forward to MDS #"LPU64" ("DFID")\n",
1895                mds, PFID(&op_data->op_fid1));
1896
1897         op_data->op_fsuid = current->fsuid;
1898         op_data->op_fsgid = current->fsgid;
1899         op_data->op_cap   = current->cap_effective;
1900
1901         tgt_exp = lmv->tgts[mds].ltd_exp;
1902         if (op_data->op_namelen) {
1903                 op_data->op_flags |= MF_MDC_CANCEL_FID2;
1904                 /* Cancel UPDATE lock on child (fid1). */
1905                 rc = lmv_early_cancel(lmv, tgt_exp, NULL, op_data, LCK_EX,
1906                                       MDS_INODELOCK_UPDATE, MF_MDC_CANCEL_FID1);
1907         }
1908         if (rc == 0)
1909                 rc = md_link(tgt_exp, op_data, request);
1910         if (rc == -ERESTART) {
1911                 LASSERT(*request != NULL);
1912                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *request,
1913                           "Got -ERESTART during link!\n");
1914                 ptlrpc_req_finished(*request);
1915                 *request = NULL;
1916
1917                 /*
1918                  * Directory got split. Time to update local object and repeat
1919                  * the request with proper MDS.
1920                  */
1921                 rc = lmv_handle_split(exp, &op_data->op_fid2);
1922                 if (rc == 0)
1923                         goto repeat;
1924         }
1925
1926         RETURN(rc);
1927 }
1928
1929 static int lmv_rename(struct obd_export *exp, struct md_op_data *op_data,
1930                       const char *old, int oldlen, const char *new, int newlen,
1931                       struct ptlrpc_request **request)
1932 {
1933         struct obd_export *tgt_exp = NULL, *src_exp;
1934         struct obd_device *obd = exp->exp_obd;
1935         struct lmv_obd *lmv = &obd->u.lmv;
1936         int rc, mea_idx, loop = 0;
1937         struct lmv_obj *obj;
1938         mdsno_t mds1, mds2;
1939         ENTRY;
1940
1941         CDEBUG(D_OTHER, "rename %*s in "DFID" to %*s in "DFID"\n",
1942                oldlen, old, PFID(&op_data->op_fid1),
1943                newlen, new, PFID(&op_data->op_fid2));
1944
1945         rc = lmv_check_connect(obd);
1946         if (rc)
1947                 RETURN(rc);
1948
1949         if (oldlen == 0) {
1950                 /*
1951                  * MDS with old dir entry is asking another MDS to create name
1952                  * there.
1953                  */
1954                 CDEBUG(D_OTHER,
1955                        "create %*s(%d/%d) in "DFID" pointing "
1956                        "to "DFID"\n", newlen, new, oldlen, newlen,
1957                        PFID(&op_data->op_fid2), PFID(&op_data->op_fid1));
1958
1959                 rc = lmv_fld_lookup(lmv, &op_data->op_fid2, &mds1);
1960                 if (rc)
1961                         RETURN(rc);
1962
1963                 /*
1964                  * Target directory can be split, sowe should forward request to
1965                  * the right MDS.
1966                  */
1967                 obj = lmv_obj_grab(obd, &op_data->op_fid2);
1968                 if (obj) {
1969                         mea_idx = raw_name2idx(obj->lo_hashtype,
1970                                                obj->lo_objcount,
1971                                                (char *)new, newlen);
1972                         op_data->op_fid2 = obj->lo_inodes[mea_idx].li_fid;
1973                         CDEBUG(D_OTHER, "Parent obj "DFID"\n",
1974                                PFID(&op_data->op_fid2));
1975                         lmv_obj_put(obj);
1976                 }
1977                 goto request;
1978         }
1979
1980 repeat:
1981         ++loop;
1982         LASSERT(loop <= 2);
1983         obj = lmv_obj_grab(obd, &op_data->op_fid1);
1984         if (obj) {
1985                 /*
1986                  * directory is already split, so we have to forward request to
1987                  * the right MDS.
1988                  */
1989                 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
1990                                        (char *)old, oldlen);
1991                 op_data->op_fid1 = obj->lo_inodes[mea_idx].li_fid;
1992                 mds1 = obj->lo_inodes[mea_idx].li_mds;
1993                 CDEBUG(D_OTHER, "Parent obj "DFID"\n", PFID(&op_data->op_fid1));
1994                 lmv_obj_put(obj);
1995         } else {
1996                 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds1);
1997                 if (rc)
1998                         RETURN(rc);
1999         }
2000
2001         obj = lmv_obj_grab(obd, &op_data->op_fid2);
2002         if (obj) {
2003                 /*
2004                  * Directory is already split, so we have to forward request to
2005                  * the right MDS.
2006                  */
2007                 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
2008                                        (char *)new, newlen);
2009
2010                 mds2 = obj->lo_inodes[mea_idx].li_mds;
2011                 op_data->op_fid2 = obj->lo_inodes[mea_idx].li_fid;
2012                 CDEBUG(D_OTHER, "Parent obj "DFID"\n", PFID(&op_data->op_fid2));
2013                 lmv_obj_put(obj);
2014         } else {
2015                 rc = lmv_fld_lookup(lmv, &op_data->op_fid2, &mds2);
2016                 if (rc)
2017                         RETURN(rc);
2018         }
2019
2020 request:
2021         op_data->op_fsuid = current->fsuid;
2022         op_data->op_fsgid = current->fsgid;
2023         op_data->op_cap   = current->cap_effective;
2024
2025         src_exp = lmv_get_export(lmv, mds1);
2026         tgt_exp = lmv_get_export(lmv, mds2);
2027         if (oldlen) {
2028                 /* LOOKUP lock on src child (fid3) should also be cancelled for
2029                  * src_exp in mdc_rename. */
2030                 op_data->op_flags |= MF_MDC_CANCEL_FID1 | MF_MDC_CANCEL_FID3;
2031
2032                 /* Cancel UPDATE locks on tgt parent (fid2), tgt_exp is its
2033                  * own export. */
2034                 rc = lmv_early_cancel(lmv, src_exp, tgt_exp, op_data, LCK_EX,
2035                                       MDS_INODELOCK_UPDATE, MF_MDC_CANCEL_FID2);
2036
2037                 /* Cancel LOOKUP locks on tgt child (fid4) for parent tgt_exp.*/
2038                 if (rc == 0)
2039                         rc = lmv_early_cancel(lmv, src_exp, tgt_exp, op_data,
2040                                               LCK_EX, MDS_INODELOCK_LOOKUP,
2041                                               MF_MDC_CANCEL_FID4);
2042
2043                 /* XXX: the case when child is a striped dir is not supported.
2044                  * Only the master stripe has all locks cancelled early. */
2045                 /* Cancel all the locks on tgt child (fid4). */
2046                 if (rc == 0)
2047                         rc = lmv_early_cancel(lmv, src_exp, NULL, op_data,
2048                                               LCK_EX, MDS_INODELOCK_FULL,
2049                                               MF_MDC_CANCEL_FID4);
2050         }
2051
2052         if (rc == 0)
2053                 rc = md_rename(src_exp, op_data, old, oldlen,
2054                                new, newlen, request);
2055         if (rc == -ERESTART) {
2056                 LASSERT(*request != NULL);
2057                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *request,
2058                           "Got -ERESTART during rename!\n");
2059                 ptlrpc_req_finished(*request);
2060                 *request = NULL;
2061
2062                 /*
2063                  * Directory got split. Time to update local object and repeat
2064                  * the request with proper MDS.
2065                  */
2066                 rc = lmv_handle_split(exp, &op_data->op_fid1);
2067                 if (rc == 0)
2068                         goto repeat;
2069         }
2070         RETURN(rc);
2071 }
2072
2073 static int lmv_setattr(struct obd_export *exp, struct md_op_data *op_data,
2074                        void *ea, int ealen, void *ea2, int ea2len,
2075                        struct ptlrpc_request **request,
2076                        struct md_open_data **mod)
2077 {
2078         struct obd_device *obd = exp->exp_obd;
2079         struct lmv_obd *lmv = &obd->u.lmv;
2080         struct ptlrpc_request *req;
2081         struct obd_export *tgt_exp;
2082         struct lmv_obj *obj;
2083         int rc = 0, i;
2084         ENTRY;
2085
2086         rc = lmv_check_connect(obd);
2087         if (rc)
2088                 RETURN(rc);
2089
2090         obj = lmv_obj_grab(obd, &op_data->op_fid1);
2091
2092         CDEBUG(D_OTHER, "SETATTR for "DFID", valid 0x%x%s\n",
2093                PFID(&op_data->op_fid1), op_data->op_attr.ia_valid,
2094                obj ? ", split" : "");
2095
2096         op_data->op_flags |= MF_MDC_CANCEL_FID1;
2097         if (obj) {
2098                 for (i = 0; i < obj->lo_objcount; i++) {
2099                         op_data->op_fid1 = obj->lo_inodes[i].li_fid;
2100
2101                         tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
2102                         if (IS_ERR(tgt_exp)) {
2103                                 rc = PTR_ERR(tgt_exp);
2104                                 break;
2105                         }
2106
2107                         rc = md_setattr(tgt_exp, op_data, ea, ealen,
2108                                         ea2, ea2len, &req, mod);
2109
2110                         if (lu_fid_eq(&obj->lo_fid, &obj->lo_inodes[i].li_fid)) {
2111                                 /*
2112                                  * this is master object and this request should
2113                                  * be returned back to llite.
2114                                  */
2115                                 *request = req;
2116                         } else {
2117                                 ptlrpc_req_finished(req);
2118                         }
2119
2120                         if (rc)
2121                                 break;
2122                 }
2123                 lmv_obj_put(obj);
2124         } else {
2125                 tgt_exp = lmv_find_export(lmv, &op_data->op_fid1);
2126                 if (IS_ERR(tgt_exp))
2127                         RETURN(PTR_ERR(tgt_exp));
2128
2129                 rc = md_setattr(tgt_exp, op_data, ea, ealen, ea2,
2130                                 ea2len, request, mod);
2131         }
2132         RETURN(rc);
2133 }
2134
2135 static int lmv_sync(struct obd_export *exp, const struct lu_fid *fid,
2136                     struct obd_capa *oc, struct ptlrpc_request **request)
2137 {
2138         struct obd_device *obd = exp->exp_obd;
2139         struct lmv_obd *lmv = &obd->u.lmv;
2140         struct obd_export *tgt_exp;
2141         int rc;
2142         ENTRY;
2143
2144         rc = lmv_check_connect(obd);
2145         if (rc)
2146                 RETURN(rc);
2147
2148         tgt_exp = lmv_find_export(lmv, fid);
2149         if (IS_ERR(tgt_exp))
2150                 RETURN(PTR_ERR(tgt_exp));
2151
2152         rc = md_sync(tgt_exp, fid, oc, request);
2153         RETURN(rc);
2154 }
2155
2156 /* main purpose of LMV blocking ast is to remove split directory LMV
2157  * presentation object (struct lmv_obj) attached to the lock being revoked. */
2158 int lmv_blocking_ast(struct ldlm_lock *lock,
2159                      struct ldlm_lock_desc *desc,
2160                      void *data, int flag)
2161 {
2162         struct lustre_handle lockh;
2163         struct lmv_obj *obj;
2164         int rc;
2165         ENTRY;
2166
2167         switch (flag) {
2168         case LDLM_CB_BLOCKING:
2169                 ldlm_lock2handle(lock, &lockh);
2170                 rc = ldlm_cli_cancel(&lockh);
2171                 if (rc < 0) {
2172                         CDEBUG(D_INODE, "ldlm_cli_cancel: %d\n", rc);
2173                         RETURN(rc);
2174                 }
2175                 break;
2176         case LDLM_CB_CANCELING:
2177                 /* time to drop cached attrs for dirobj */
2178                 obj = lock->l_ast_data;
2179                 if (obj) {
2180                         CDEBUG(D_OTHER, "cancel %s on "LPU64"/"LPU64
2181                                ", master "DFID"\n",
2182                                lock->l_resource->lr_name.name[3] == 1 ?
2183                                "LOOKUP" : "UPDATE",
2184                                lock->l_resource->lr_name.name[0],
2185                                lock->l_resource->lr_name.name[1],
2186                                PFID(&obj->lo_fid));
2187                         lmv_obj_put(obj);
2188                 }
2189                 break;
2190         default:
2191                 LBUG();
2192         }
2193         RETURN(0);
2194 }
2195
2196 static void lmv_hash_adjust(__u64 *hash, __u64 hash_adj)
2197 {
2198         __u64 val;
2199
2200         val = le64_to_cpu(*hash);
2201         if (val < hash_adj)
2202                 val += MAX_HASH_SIZE;
2203         if (val != DIR_END_OFF)
2204                 *hash = cpu_to_le64(val - hash_adj);
2205 }
2206
2207 static __u32 lmv_node_rank(struct obd_export *exp, const struct lu_fid *fid)
2208 {
2209         __u64 id;
2210         struct obd_import *imp;
2211
2212         /*
2213          * XXX Hack: to get nid we assume that underlying obd device is mdc.
2214          */
2215         imp  = class_exp2cliimp(exp);
2216         id   = imp->imp_connection->c_self + fid_flatten(fid);
2217
2218         CDEBUG(D_INFO, "node rank: "LPX64" "DFID" "LPX64" "LPX64"\n",
2219                imp->imp_connection->c_self, PFID(fid), id, id ^ (id >> 32));
2220
2221         return id ^ (id >> 32);
2222 }
2223
2224 static int lmv_readpage(struct obd_export *exp, const struct lu_fid *fid,
2225                         struct obd_capa *oc, __u64 offset64, struct page *page,
2226                         struct ptlrpc_request **request)
2227 {
2228         struct obd_device *obd = exp->exp_obd;
2229         struct lmv_obd *lmv = &obd->u.lmv;
2230         struct obd_export *tgt_exp;
2231         struct lu_fid rid = *fid;
2232         struct lmv_obj *obj;
2233         __u64 offset;
2234         __u64 hash_adj = 0;
2235         __u32 rank = 0;
2236         __u64 seg_size = 0;
2237         __u64 tgt_tmp = 0;
2238         int tgt = 0;
2239         int tgt0 = 0;
2240         int rc;
2241         int nr = 0;
2242         ENTRY;
2243
2244         offset = offset64;
2245
2246         rc = lmv_check_connect(obd);
2247         if (rc)
2248                 RETURN(rc);
2249
2250         CDEBUG(D_INFO, "READPAGE at "LPX64" from "DFID"\n", offset, PFID(&rid));
2251
2252         obj = lmv_obj_grab(obd, fid);
2253         if (obj) {
2254
2255         /*
2256          * This case handle directory lookup in clustered metadata case (i.e.
2257          * split directory is located on multiple md servers.)
2258          * each server keeps directory entries for certain range of hashes.
2259          * E.g. we have N server and suppose hash range is 0 to MAX_HASH.
2260          * first server will keep records with hashes [ 0 ... MAX_HASH / N  - 1],
2261          * second one with hashes [MAX_HASH / N ... 2 * MAX_HASH / N] and
2262          * so on....
2263          *      readdir can simply start reading entries from 0 - N server in
2264          * order but that will not scale well as all client will request dir in
2265          * to server in same order.
2266          * Following algorithm does optimization:
2267          * Instead of doing readdir in 1, 2, ...., N order, client with a
2268          * rank R does readdir in R, R + 1, ..., N, 1, ... R - 1 order.
2269          * (every client has rank R)
2270          *      But ll_readdir() expect offset range [0 to MAX_HASH/N) but
2271          * since client ask dir from MDS{R} client has pages with offsets
2272          * [R*MAX_HASH/N ... (R + 1)*MAX_HASH/N] there for we do hash_adj
2273          * on hash  values that we get.
2274          */
2275
2276                 struct lmv_inode *loi;
2277
2278                 lmv_obj_lock(obj);
2279
2280                 nr       = obj->lo_objcount;
2281                 LASSERT(nr > 0);
2282                 seg_size = MAX_HASH_SIZE;
2283                 do_div(seg_size, nr);
2284                 loi      = obj->lo_inodes;
2285                 rank     = lmv_node_rank(lmv_get_export(lmv, loi[0].li_mds),
2286                                          fid) % nr;
2287                 tgt_tmp = offset;
2288                 do_div(tgt_tmp, seg_size);
2289                 tgt0     = do_div(tgt_tmp,  nr);
2290                 tgt      = (tgt0 + rank) % nr;
2291
2292                 if (tgt < tgt0)
2293                         /*
2294                          * Wrap around.
2295                          *
2296                          * Last segment has unusual length due to division
2297                          * rounding.
2298                          */
2299                         hash_adj = MAX_HASH_SIZE - seg_size * nr;
2300                 else
2301                         hash_adj = 0;
2302
2303                 hash_adj += rank * seg_size;
2304
2305                 CDEBUG(D_INFO, "hash_adj: %x "LPX64" "LPX64"/%x -> "LPX64"/%x\n",
2306                        rank, hash_adj, offset, tgt0, offset + hash_adj, tgt);
2307
2308                 offset = (offset + hash_adj) & MAX_HASH_SIZE;
2309                 rid = obj->lo_inodes[tgt].li_fid;
2310                 tgt_exp = lmv_get_export(lmv, loi[tgt].li_mds);
2311
2312                 CDEBUG(D_INFO, "forward to "DFID" with offset %lu i %d\n",
2313                        PFID(&rid), (unsigned long)offset, tgt);
2314         } else
2315                 tgt_exp = lmv_find_export(lmv, &rid);
2316
2317         if (IS_ERR(tgt_exp))
2318                 GOTO(cleanup, rc = PTR_ERR(tgt_exp));
2319
2320         rc = md_readpage(tgt_exp, &rid, oc, offset, page, request);
2321         if (rc)
2322                 GOTO(cleanup, rc);
2323         if (obj) {
2324                 struct lu_dirpage *dp;
2325                 struct lu_dirent  *ent;
2326
2327                 dp = cfs_kmap(page);
2328
2329                 lmv_hash_adjust(&dp->ldp_hash_start, hash_adj);
2330                 lmv_hash_adjust(&dp->ldp_hash_end,   hash_adj);
2331                 LASSERT(le64_to_cpu(dp->ldp_hash_start) <= offset64);
2332
2333                 for (ent = lu_dirent_start(dp); ent != NULL;
2334                      ent = lu_dirent_next(ent))
2335                         lmv_hash_adjust(&ent->lde_hash, hash_adj);
2336
2337                 if (tgt0 != nr - 1) {
2338                         __u64 end;
2339
2340                         end = le64_to_cpu(dp->ldp_hash_end);
2341                         if (end == DIR_END_OFF) {
2342                                 dp->ldp_hash_end = cpu_to_le32(seg_size *
2343                                                                (tgt0 + 1));
2344                                 CDEBUG(D_INFO,
2345                                        ""DFID" reset end "LPX64" tgt %d\n",
2346                                        PFID(&rid),
2347                                        le64_to_cpu(dp->ldp_hash_end), tgt);
2348                         }
2349                 }
2350                 cfs_kunmap(page);
2351         }
2352         /*
2353          * Here we could remove "." and ".." from all pages which at not from
2354          * master. But MDS has only "." and ".." for master dir.
2355          */
2356         EXIT;
2357 cleanup:
2358         if (obj) {
2359                 lmv_obj_unlock(obj);
2360                 lmv_obj_put(obj);
2361         }
2362         return rc;
2363 }
2364
2365 static int lmv_unlink_slaves(struct obd_export *exp,
2366                              struct md_op_data *op_data,
2367                              struct ptlrpc_request **req)
2368 {
2369         struct obd_device *obd = exp->exp_obd;
2370         struct lmv_obd *lmv = &obd->u.lmv;
2371         struct lmv_stripe_md *mea = op_data->op_mea1;
2372         struct md_op_data *op_data2;
2373         struct obd_export *tgt_exp;
2374         int i, rc = 0;
2375         ENTRY;
2376
2377         OBD_ALLOC_PTR(op_data2);
2378         if (op_data2 == NULL)
2379                 RETURN(-ENOMEM);
2380
2381         op_data2->op_mode = S_IFDIR;
2382         op_data2->op_fsuid = current->fsuid;
2383         op_data2->op_fsgid = current->fsgid;
2384         op_data2->op_bias = 0;
2385
2386         LASSERT(mea != NULL);
2387         for (i = 0; i < mea->mea_count; i++) {
2388                 memset(op_data2, 0, sizeof(*op_data2));
2389                 op_data2->op_fid1 = mea->mea_ids[i];
2390                 tgt_exp = lmv_find_export(lmv, &op_data2->op_fid1);
2391                 if (IS_ERR(tgt_exp))
2392                         GOTO(out_free_op_data2, rc = PTR_ERR(tgt_exp));
2393
2394                 if (tgt_exp == NULL)
2395                         continue;
2396
2397                 rc = md_unlink(tgt_exp, op_data2, req);
2398
2399                 CDEBUG(D_OTHER, "unlink slave "DFID" -> %d\n",
2400                        PFID(&mea->mea_ids[i]), rc);
2401
2402                 if (*req) {
2403                         ptlrpc_req_finished(*req);
2404                         *req = NULL;
2405                 }
2406                 if (rc)
2407                         GOTO(out_free_op_data2, rc);
2408         }
2409
2410         EXIT;
2411 out_free_op_data2:
2412         OBD_FREE_PTR(op_data2);
2413         return rc;
2414 }
2415
2416 static int lmv_unlink(struct obd_export *exp, struct md_op_data *op_data,
2417                       struct ptlrpc_request **request)
2418 {
2419         struct obd_device *obd = exp->exp_obd;
2420         struct lmv_obd *lmv = &obd->u.lmv;
2421         struct obd_export *tgt_exp = NULL;
2422         struct lmv_obj *obj;
2423         int rc, loop = 0;
2424         ENTRY;
2425
2426         rc = lmv_check_connect(obd);
2427         if (rc)
2428                 RETURN(rc);
2429
2430         if (op_data->op_namelen == 0 && op_data->op_mea1 != NULL) {
2431                 /* mds asks to remove slave objects */
2432                 rc = lmv_unlink_slaves(exp, op_data, request);
2433                 RETURN(rc);
2434         }
2435
2436 repeat:
2437         ++loop;
2438         LASSERT(loop <= 2);
2439         if (op_data->op_namelen != 0) {
2440                 int mea_idx;
2441
2442                 obj = lmv_obj_grab(obd, &op_data->op_fid1);
2443                 if (obj) {
2444                         mea_idx = raw_name2idx(obj->lo_hashtype,
2445                                                obj->lo_objcount,
2446                                                op_data->op_name,
2447                                                op_data->op_namelen);
2448                         op_data->op_bias &= ~MDS_CHECK_SPLIT;
2449                         op_data->op_fid1 = obj->lo_inodes[mea_idx].li_fid;
2450                         tgt_exp = lmv_get_export(lmv,
2451                                                  obj->lo_inodes[mea_idx].li_mds);
2452                         lmv_obj_put(obj);
2453                         CDEBUG(D_OTHER, "unlink '%*s' in "DFID" -> %u\n",
2454                                op_data->op_namelen, op_data->op_name,
2455                                PFID(&op_data->op_fid1), mea_idx);
2456                 }
2457         } else {
2458                 CDEBUG(D_OTHER, "drop i_nlink on "DFID"\n",
2459                        PFID(&op_data->op_fid1));
2460         }
2461         if (tgt_exp == NULL) {
2462                 tgt_exp = lmv_find_export(lmv, &op_data->op_fid1);
2463                 if (IS_ERR(tgt_exp))
2464                         RETURN(PTR_ERR(tgt_exp));
2465                 op_data->op_bias |= MDS_CHECK_SPLIT;
2466         }
2467
2468         op_data->op_fsuid = current->fsuid;
2469         op_data->op_fsgid = current->fsgid;
2470         op_data->op_cap   = current->cap_effective;
2471
2472         /* If child's fid is given, cancel unused locks for it if it is from
2473          * another export than parent. */
2474         if (op_data->op_namelen) {
2475                 /* LOOKUP lock for child (fid3) should also be cancelled on
2476                  * parent tgt_exp in mdc_unlink(). */
2477                 op_data->op_flags |= MF_MDC_CANCEL_FID1 | MF_MDC_CANCEL_FID3;
2478
2479                 /* XXX: the case when child is a striped dir is not supported.
2480                  * Only the master stripe has all locks cancelled early. */
2481                 /* Cancel FULL locks on child (fid3). */
2482                 rc = lmv_early_cancel(lmv, tgt_exp, NULL, op_data, LCK_EX,
2483                                       MDS_INODELOCK_FULL, MF_MDC_CANCEL_FID3);
2484         }
2485         if (rc == 0)
2486                 rc = md_unlink(tgt_exp, op_data, request);
2487         if (rc == -ERESTART) {
2488                 LASSERT(*request != NULL);
2489                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *request,
2490                           "Got -ERESTART during unlink!\n");
2491                 ptlrpc_req_finished(*request);
2492                 *request = NULL;
2493
2494                 /*
2495                  * Directory got split. Time to update local object and repeat
2496                  * the request with proper MDS.
2497                  */
2498                 rc = lmv_handle_split(exp, &op_data->op_fid1);
2499                 if (rc == 0)
2500                         goto repeat;
2501         }
2502         RETURN(rc);
2503 }
2504
2505 static int lmv_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
2506                          struct obd_device *tgt, int count,
2507                          struct llog_catid *logid, struct obd_uuid *uuid)
2508 {
2509 #if 0
2510         struct llog_ctxt *ctxt;
2511         int rc;
2512         ENTRY;
2513
2514         LASSERT(group == OBD_LLOG_GROUP);
2515         rc = llog_setup(obd, &obd->obd_olg, LLOG_CONFIG_REPL_CTXT, tgt, 0, NULL,
2516                         &llog_client_ops);
2517         if (rc == 0) {
2518                 ctxt = llog_group_get_ctxt(&obd->obd_olg, LLOG_CONFIG_REPL_CTXT);
2519                 llog_initiator_connect(ctxt, tgt);
2520                 llog_ctxt_put(ctxt);
2521         }
2522         RETURN(rc);
2523 #else
2524         return 0;
2525 #endif
2526 }
2527
2528 static int lmv_llog_finish(struct obd_device *obd, int count)
2529 {
2530         struct llog_ctxt *ctxt;
2531         int rc = 0;
2532         ENTRY;
2533
2534         ctxt = llog_get_context(obd, LLOG_CONFIG_REPL_CTXT);
2535         if (ctxt)
2536                 rc = llog_cleanup(ctxt);
2537
2538         RETURN(rc);
2539 }
2540
2541 static int lmv_precleanup(struct obd_device *obd, enum obd_cleanup_stage stage)
2542 {
2543         int rc = 0;
2544
2545         switch (stage) {
2546         case OBD_CLEANUP_EARLY:
2547                 /* XXX: here should be calling obd_precleanup() down to
2548                  * stack. */
2549                 break;
2550         case OBD_CLEANUP_EXPORTS:
2551                 rc = obd_llog_finish(obd, 0);
2552                 if (rc != 0)
2553                         CERROR("failed to cleanup llogging subsystems\n");
2554                 break;
2555         default:
2556                 break;
2557         }
2558         RETURN(rc);
2559 }
2560
2561 static int lmv_get_info(struct obd_export *exp, __u32 keylen,
2562                         void *key, __u32 *vallen, void *val, struct lov_stripe_md *lsm)
2563 {
2564         struct obd_device *obd;
2565         struct lmv_obd *lmv;
2566         int rc = 0;
2567         ENTRY;
2568
2569         obd = class_exp2obd(exp);
2570         if (obd == NULL) {
2571                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
2572                        exp->exp_handle.h_cookie);
2573                 RETURN(-EINVAL);
2574         }
2575
2576         lmv = &obd->u.lmv;
2577         if (keylen >= strlen("remote_flag") && !strcmp(key, "remote_flag")) {
2578                 struct lmv_tgt_desc *tgts;
2579                 int i;
2580
2581                 rc = lmv_check_connect(obd);
2582                 if (rc)
2583                         RETURN(rc);
2584
2585                 LASSERT(*vallen == sizeof(__u32));
2586                 for (i = 0, tgts = lmv->tgts; i < lmv->desc.ld_tgt_count;
2587                      i++, tgts++) {
2588
2589                         /* all tgts should be connected when this get called. */
2590                         if (!tgts || !tgts->ltd_exp) {
2591                                 CERROR("target not setup?\n");
2592                                 continue;
2593                         }
2594
2595                         if (!obd_get_info(tgts->ltd_exp, keylen, key,
2596                                           vallen, val, NULL))
2597                                 RETURN(0);
2598                 }
2599                 RETURN(-EINVAL);
2600         } else if (KEY_IS(KEY_MAX_EASIZE) || KEY_IS(KEY_CONN_DATA)) {
2601                 rc = lmv_check_connect(obd);
2602                 if (rc)
2603                         RETURN(rc);
2604
2605                 /* forwarding this request to first MDS, it should know LOV
2606                  * desc. */
2607                 rc = obd_get_info(lmv->tgts[0].ltd_exp, keylen, key,
2608                                   vallen, val, NULL);
2609                 if (!rc && KEY_IS(KEY_CONN_DATA)) {
2610                         exp->exp_connect_flags =
2611                         ((struct obd_connect_data *)val)->ocd_connect_flags;
2612                 }
2613                 RETURN(rc);
2614         }
2615
2616         CDEBUG(D_IOCTL, "invalid key\n");
2617         RETURN(-EINVAL);
2618 }
2619
2620 int lmv_set_info_async(struct obd_export *exp, obd_count keylen,
2621                        void *key, obd_count vallen, void *val,
2622                        struct ptlrpc_request_set *set)
2623 {
2624         struct lmv_tgt_desc    *tgt;
2625         struct obd_device      *obd;
2626         struct lmv_obd         *lmv;
2627         int rc = 0;
2628         ENTRY;
2629
2630         obd = class_exp2obd(exp);
2631         if (obd == NULL) {
2632                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
2633                        exp->exp_handle.h_cookie);
2634                 RETURN(-EINVAL);
2635         }
2636         lmv = &obd->u.lmv;
2637
2638         if (KEY_IS(KEY_READ_ONLY) || KEY_IS(KEY_FLUSH_CTX) ||
2639             KEY_IS(KEY_INIT_RECOV_BACKUP)) {
2640                 int i, err = 0;
2641
2642                 for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
2643                         tgt = &lmv->tgts[i];
2644
2645                         if (!tgt->ltd_exp)
2646                                 continue;
2647
2648                         err = obd_set_info_async(tgt->ltd_exp,
2649                                                  keylen, key, vallen, val, set);
2650                         if (err && rc == 0)
2651                                 rc = err;
2652                 }
2653
2654                 RETURN(rc);
2655         }
2656
2657         RETURN(-EINVAL);
2658 }
2659
2660 int lmv_packmd(struct obd_export *exp, struct lov_mds_md **lmmp,
2661                struct lov_stripe_md *lsm)
2662 {
2663         struct obd_device *obd = class_exp2obd(exp);
2664         struct lmv_obd *lmv = &obd->u.lmv;
2665         struct lmv_stripe_md *meap, *lsmp;
2666         int mea_size, i;
2667         ENTRY;
2668
2669         mea_size = lmv_get_easize(lmv);
2670         if (!lmmp)
2671                 RETURN(mea_size);
2672
2673         if (*lmmp && !lsm) {
2674                 OBD_FREE(*lmmp, mea_size);
2675                 *lmmp = NULL;
2676                 RETURN(0);
2677         }
2678
2679         if (*lmmp == NULL) {
2680                 OBD_ALLOC(*lmmp, mea_size);
2681                 if (*lmmp == NULL)
2682                         RETURN(-ENOMEM);
2683         }
2684
2685         if (!lsm)
2686                 RETURN(mea_size);
2687
2688         lsmp = (struct lmv_stripe_md *)lsm;
2689         meap = (struct lmv_stripe_md *)*lmmp;
2690
2691         if (lsmp->mea_magic != MEA_MAGIC_LAST_CHAR &&
2692             lsmp->mea_magic != MEA_MAGIC_ALL_CHARS)
2693                 RETURN(-EINVAL);
2694
2695         meap->mea_magic = cpu_to_le32(lsmp->mea_magic);
2696         meap->mea_count = cpu_to_le32(lsmp->mea_count);
2697         meap->mea_master = cpu_to_le32(lsmp->mea_master);
2698
2699         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
2700                 meap->mea_ids[i] = meap->mea_ids[i];
2701                 fid_cpu_to_le(&meap->mea_ids[i], &meap->mea_ids[i]);
2702         }
2703
2704         RETURN(mea_size);
2705 }
2706
2707 int lmv_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
2708                  struct lov_mds_md *lmm, int lmm_size)
2709 {
2710         struct obd_device *obd = class_exp2obd(exp);
2711         struct lmv_stripe_md **tmea = (struct lmv_stripe_md **)lsmp;
2712         struct lmv_stripe_md *mea = (struct lmv_stripe_md *)lmm;
2713         struct lmv_obd *lmv = &obd->u.lmv;
2714         int mea_size, i;
2715         __u32 magic;
2716         ENTRY;
2717
2718         mea_size = lmv_get_easize(lmv);
2719         if (lsmp == NULL)
2720                 return mea_size;
2721
2722         if (*lsmp != NULL && lmm == NULL) {
2723                 OBD_FREE(*tmea, mea_size);
2724                 *lsmp = NULL;
2725                 RETURN(0);
2726         }
2727
2728         LASSERT(mea_size == lmm_size);
2729
2730         OBD_ALLOC(*tmea, mea_size);
2731         if (*tmea == NULL)
2732                 RETURN(-ENOMEM);
2733
2734         if (!lmm)
2735                 RETURN(mea_size);
2736
2737         if (mea->mea_magic == MEA_MAGIC_LAST_CHAR ||
2738             mea->mea_magic == MEA_MAGIC_ALL_CHARS ||
2739             mea->mea_magic == MEA_MAGIC_HASH_SEGMENT)
2740         {
2741                 magic = le32_to_cpu(mea->mea_magic);
2742         } else {
2743                 /* old mea is not handled here */
2744                 LBUG();
2745         }
2746
2747         (*tmea)->mea_magic = magic;
2748         (*tmea)->mea_count = le32_to_cpu(mea->mea_count);
2749         (*tmea)->mea_master = le32_to_cpu(mea->mea_master);
2750
2751         for (i = 0; i < (*tmea)->mea_count; i++) {
2752                 (*tmea)->mea_ids[i] = mea->mea_ids[i];
2753                 fid_le_to_cpu(&(*tmea)->mea_ids[i], &(*tmea)->mea_ids[i]);
2754         }
2755         RETURN(mea_size);
2756 }
2757
2758 static int lmv_cancel_unused(struct obd_export *exp,
2759                              const struct lu_fid *fid,
2760                              ldlm_policy_data_t *policy,
2761                              ldlm_mode_t mode, int flags, void *opaque)
2762 {
2763         struct obd_device *obd = exp->exp_obd;
2764         struct lmv_obd *lmv = &obd->u.lmv;
2765         int rc = 0, err, i;
2766         ENTRY;
2767
2768         LASSERT(fid != NULL);
2769
2770         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
2771                 if (!lmv->tgts[i].ltd_exp || !lmv->tgts[i].ltd_active)
2772                         continue;
2773
2774                 err = md_cancel_unused(lmv->tgts[i].ltd_exp, fid,
2775                                        policy, mode, flags, opaque);
2776                 if (!rc)
2777                         rc = err;
2778         }
2779         RETURN(rc);
2780 }
2781
2782 int lmv_set_lock_data(struct obd_export *exp, __u64 *lockh, void *data)
2783 {
2784         struct obd_device *obd = exp->exp_obd;
2785         struct lmv_obd *lmv = &obd->u.lmv;
2786
2787         ENTRY;
2788         RETURN(md_set_lock_data(lmv->tgts[0].ltd_exp, lockh, data));
2789 }
2790
2791 ldlm_mode_t lmv_lock_match(struct obd_export *exp, int flags,
2792                            const struct lu_fid *fid, ldlm_type_t type,
2793                            ldlm_policy_data_t *policy, ldlm_mode_t mode,
2794                            struct lustre_handle *lockh)
2795 {
2796         struct obd_device *obd = exp->exp_obd;
2797         struct lmv_obd *lmv = &obd->u.lmv;
2798         ldlm_mode_t rc;
2799         int i;
2800         ENTRY;
2801
2802         CDEBUG(D_OTHER, "lock match for "DFID"\n", PFID(fid));
2803
2804         /* with CMD every object can have two locks in different namespaces:
2805          * lookup lock in space of mds storing direntry and update/open lock in
2806          * space of mds storing inode. Thus we check all targets, not only that
2807          * one fid was created in. */
2808         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
2809                 rc = md_lock_match(lmv->tgts[i].ltd_exp, flags, fid,
2810                                    type, policy, mode, lockh);
2811                 if (rc)
2812                         RETURN(rc);
2813         }
2814
2815         RETURN(0);
2816 }
2817
2818 int lmv_get_lustre_md(struct obd_export *exp, struct ptlrpc_request *req,
2819                       struct obd_export *dt_exp, struct obd_export *md_exp,
2820                       struct lustre_md *md)
2821 {
2822         struct obd_device *obd = exp->exp_obd;
2823         struct lmv_obd *lmv = &obd->u.lmv;
2824         int rc;
2825
2826         ENTRY;
2827         rc = md_get_lustre_md(lmv->tgts[0].ltd_exp, req, dt_exp, md_exp, md);
2828         RETURN(rc);
2829 }
2830
2831 int lmv_free_lustre_md(struct obd_export *exp, struct lustre_md *md)
2832 {
2833         struct obd_device *obd = exp->exp_obd;
2834         struct lmv_obd *lmv = &obd->u.lmv;
2835
2836         ENTRY;
2837
2838         /* XXX LOV STACKING */
2839         if (md->mea)
2840                 obd_free_memmd(exp, (void *)&md->mea);
2841         RETURN(md_free_lustre_md(lmv->tgts[0].ltd_exp, md));
2842 }
2843
2844 int lmv_set_open_replay_data(struct obd_export *exp,
2845                              struct obd_client_handle *och,
2846                              struct ptlrpc_request *open_req)
2847 {
2848         struct obd_device *obd = exp->exp_obd;
2849         struct lmv_obd *lmv = &obd->u.lmv;
2850         struct obd_export *tgt_exp;
2851
2852         ENTRY;
2853
2854         tgt_exp = lmv_find_export(lmv, &och->och_fid);
2855         if (IS_ERR(tgt_exp))
2856                 RETURN(PTR_ERR(tgt_exp));
2857
2858         RETURN(md_set_open_replay_data(tgt_exp, och, open_req));
2859 }
2860
2861 int lmv_clear_open_replay_data(struct obd_export *exp,
2862                                struct obd_client_handle *och)
2863 {
2864         struct obd_device *obd = exp->exp_obd;
2865         struct lmv_obd *lmv = &obd->u.lmv;
2866         struct obd_export *tgt_exp;
2867         ENTRY;
2868
2869         tgt_exp = lmv_find_export(lmv, &och->och_fid);
2870         if (IS_ERR(tgt_exp))
2871                 RETURN(PTR_ERR(tgt_exp));
2872
2873         RETURN(md_clear_open_replay_data(tgt_exp, och));
2874 }
2875
2876 static int lmv_get_remote_perm(struct obd_export *exp,
2877                                const struct lu_fid *fid,
2878                                struct obd_capa *oc, __u32 suppgid,
2879                                struct ptlrpc_request **request)
2880 {
2881         struct obd_device *obd = exp->exp_obd;
2882         struct lmv_obd *lmv = &obd->u.lmv;
2883         struct obd_export *tgt_exp;
2884         int rc;
2885
2886         ENTRY;
2887
2888         rc = lmv_check_connect(obd);
2889         if (rc)
2890                 RETURN(rc);
2891
2892         tgt_exp = lmv_find_export(lmv, fid);
2893         if (IS_ERR(tgt_exp))
2894                 RETURN(PTR_ERR(tgt_exp));
2895
2896         rc = md_get_remote_perm(tgt_exp, fid, oc, suppgid, request);
2897
2898         RETURN(rc);
2899 }
2900
2901 static int lmv_renew_capa(struct obd_export *exp, struct obd_capa *oc,
2902                           renew_capa_cb_t cb)
2903 {
2904         struct obd_device *obd = exp->exp_obd;
2905         struct lmv_obd *lmv = &obd->u.lmv;
2906         struct obd_export *tgt_exp;
2907         int rc;
2908         ENTRY;
2909
2910         rc = lmv_check_connect(obd);
2911         if (rc)
2912                 RETURN(rc);
2913
2914         tgt_exp = lmv_find_export(lmv, &oc->c_capa.lc_fid);
2915         if (IS_ERR(tgt_exp))
2916                 RETURN(PTR_ERR(tgt_exp));
2917
2918         rc = md_renew_capa(tgt_exp, oc, cb);
2919         RETURN(rc);
2920 }
2921
2922 int lmv_intent_getattr_async(struct obd_export *exp,
2923                              struct md_enqueue_info *minfo,
2924                              struct ldlm_enqueue_info *einfo)
2925 {
2926         struct obd_device *obd = exp->exp_obd;
2927         struct lmv_obd    *lmv = &obd->u.lmv;
2928         struct obd_export *tgt_exp;
2929         int rc;
2930         ENTRY;
2931
2932         rc = lmv_check_connect(obd);
2933         if (rc)
2934                 RETURN(rc);
2935
2936         if (fid_is_zero(&minfo->mi_data.op_fid2))
2937                 tgt_exp = lmv_find_export(lmv, &minfo->mi_data.op_fid1);
2938         else
2939                 tgt_exp = lmv_find_export(lmv, &minfo->mi_data.op_fid2);
2940         if (IS_ERR(tgt_exp))
2941                 RETURN(PTR_ERR(tgt_exp));
2942
2943         rc = md_intent_getattr_async(tgt_exp, minfo, einfo);
2944         RETURN(rc);
2945 }
2946
2947 int lmv_revalidate_lock(struct obd_export *exp,
2948                         struct lookup_intent *it,
2949                         struct lu_fid *fid)
2950 {
2951         struct obd_device *obd = exp->exp_obd;
2952         struct lmv_obd    *lmv = &obd->u.lmv;
2953         struct obd_export *tgt_exp;
2954         int rc;
2955         ENTRY;
2956
2957         rc = lmv_check_connect(obd);
2958         if (rc)
2959                 RETURN(rc);
2960
2961         tgt_exp = lmv_find_export(lmv, fid);
2962         if (IS_ERR(tgt_exp))
2963                 RETURN(PTR_ERR(tgt_exp));
2964
2965         rc = md_revalidate_lock(tgt_exp, it, fid);
2966         RETURN(rc);
2967 }
2968
2969
2970 struct obd_ops lmv_obd_ops = {
2971         .o_owner                = THIS_MODULE,
2972         .o_setup                = lmv_setup,
2973         .o_cleanup              = lmv_cleanup,
2974         .o_precleanup           = lmv_precleanup,
2975         .o_process_config       = lmv_process_config,
2976         .o_connect              = lmv_connect,
2977         .o_disconnect           = lmv_disconnect,
2978         .o_statfs               = lmv_statfs,
2979         .o_llog_init            = lmv_llog_init,
2980         .o_llog_finish          = lmv_llog_finish,
2981         .o_get_info             = lmv_get_info,
2982         .o_set_info_async       = lmv_set_info_async,
2983         .o_packmd               = lmv_packmd,
2984         .o_unpackmd             = lmv_unpackmd,
2985         .o_notify               = lmv_notify,
2986         .o_iocontrol            = lmv_iocontrol,
2987         .o_fid_delete           = lmv_fid_delete
2988 };
2989
2990 struct md_ops lmv_md_ops = {
2991         .m_getstatus            = lmv_getstatus,
2992         .m_change_cbdata        = lmv_change_cbdata,
2993         .m_close                = lmv_close,
2994         .m_create               = lmv_create,
2995         .m_done_writing         = lmv_done_writing,
2996         .m_enqueue              = lmv_enqueue,
2997         .m_getattr              = lmv_getattr,
2998         .m_getxattr             = lmv_getxattr,
2999         .m_getattr_name         = lmv_getattr_name,
3000         .m_intent_lock          = lmv_intent_lock,
3001         .m_link                 = lmv_link,
3002         .m_rename               = lmv_rename,
3003         .m_setattr              = lmv_setattr,
3004         .m_setxattr             = lmv_setxattr,
3005         .m_sync                 = lmv_sync,
3006         .m_readpage             = lmv_readpage,
3007         .m_unlink               = lmv_unlink,
3008         .m_init_ea_size         = lmv_init_ea_size,
3009         .m_cancel_unused        = lmv_cancel_unused,
3010         .m_set_lock_data        = lmv_set_lock_data,
3011         .m_lock_match           = lmv_lock_match,
3012         .m_get_lustre_md        = lmv_get_lustre_md,
3013         .m_free_lustre_md       = lmv_free_lustre_md,
3014         .m_set_open_replay_data = lmv_set_open_replay_data,
3015         .m_clear_open_replay_data = lmv_clear_open_replay_data,
3016         .m_renew_capa           = lmv_renew_capa,
3017         .m_get_remote_perm      = lmv_get_remote_perm,
3018         .m_intent_getattr_async = lmv_intent_getattr_async,
3019         .m_revalidate_lock      = lmv_revalidate_lock
3020 };
3021
3022 int __init lmv_init(void)
3023 {
3024         struct lprocfs_static_vars lvars;
3025         int rc;
3026
3027         obj_cache = cfs_mem_cache_create("lmv_objects",
3028                                       sizeof(struct lmv_obj),
3029                                       0, 0);
3030         if (!obj_cache) {
3031                 CERROR("error allocating lmv objects cache\n");
3032                 return -ENOMEM;
3033         }
3034
3035         lprocfs_lmv_init_vars(&lvars);
3036         rc = class_register_type(&lmv_obd_ops, &lmv_md_ops,
3037                                  lvars.module_vars, LUSTRE_LMV_NAME, NULL);
3038         if (rc)
3039                 cfs_mem_cache_destroy(obj_cache);
3040
3041         return rc;
3042 }
3043
3044 #ifdef __KERNEL__
3045 static void lmv_exit(void)
3046 {
3047         int rc;
3048
3049         class_unregister_type(LUSTRE_LMV_NAME);
3050
3051         rc = cfs_mem_cache_destroy(obj_cache);
3052         LASSERTF(rc == 0,
3053                  "can't free lmv objects cache, %d object(s)"
3054                  "still in use\n", atomic_read(&obj_cache_count));
3055 }
3056
3057 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
3058 MODULE_DESCRIPTION("Lustre Logical Metadata Volume OBD driver");
3059 MODULE_LICENSE("GPL");
3060
3061 module_init(lmv_init);
3062 module_exit(lmv_exit);
3063 #endif