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