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 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 <asm/div64.h>
  33 #include <linux/seq_file.h>
  34 #include <linux/namei.h>
  35 #else
  36 #include <liblustre.h>
  37 #endif
  38 #include <linux/ext2_fs.h>
  39
  40 #include <linux/obd_support.h>
  41 #include <linux/lustre_lib.h>
  42 #include <linux/lustre_net.h>
  43 #include <linux/lustre_idl.h>
  44 #include <linux/lustre_dlm.h>
  45 #include <linux/lustre_mds.h>
  46 #include <linux/obd_class.h>
  47 #include <linux/obd_ost.h>
  48 #include <linux/lprocfs_status.h>
  49 #include <linux/lustre_fsfilt.h>
  50 #include <linux/obd_lmv.h>
  51 #include <linux/lustre_lite.h>
  52 #include "lmv_internal.h"
  53
  54 /* not defined for liblustre building */
  55 #if !defined(ATOMIC_INIT)
  56 #define ATOMIC_INIT(val) { (val) }
  57 #endif
  58
  59 /* object cache. */
  60 kmem_cache_t *obj_cache;
  61 atomic_t obj_cache_count = ATOMIC_INIT(0);
  62
  63 static void lmv_activate_target(struct lmv_obd *lmv,
  64                                 struct lmv_tgt_desc *tgt,
  65                                 int activate)
  66 {
  67         if (tgt->active == activate)
  68                 return;
  69
  70         tgt->active = activate;
  71         lmv->desc.ld_active_tgt_count += (activate ? 1 : -1);
  72 }
  73
  74 /* Error codes:
  75  *
  76  *  -EINVAL  : UUID can't be found in the LMV's target list
  77  *  -ENOTCONN: The UUID is found, but the target connection is bad (!)
  78  *  -EBADF   : The UUID is found, but the OBD of the wrong type (!)
  79  */
  80 static int lmv_set_mdc_active(struct lmv_obd *lmv, struct obd_uuid *uuid,
  81                               int activate)
  82 {
  83         struct lmv_tgt_desc *tgt;
  84         struct obd_device *obd;
  85         int i, rc = 0;
  86         ENTRY;
  87
  88         CDEBUG(D_INFO, "Searching in lmv %p for uuid %s (activate=%d)\n",
  89                lmv, uuid->uuid, activate);
  90
  91         spin_lock(&lmv->lmv_lock);
  92         for (i = 0, tgt = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgt++) {
  93                 if (tgt->ltd_exp == NULL)
  94                         continue;
  95
  96                 CDEBUG(D_INFO, "lmv idx %d is %s conn "LPX64"\n",
  97                        i, tgt->uuid.uuid, tgt->ltd_exp->exp_handle.h_cookie);
  98
  99                 if (strncmp(uuid->uuid, tgt->uuid.uuid, sizeof uuid->uuid) == 0)
 100                         break;
 101         }
 102
 103         if (i == lmv->desc.ld_tgt_count)
 104                 GOTO(out_lmv_lock, rc = -EINVAL);
 105
 106         obd = class_exp2obd(tgt->ltd_exp);
 107         if (obd == NULL)
 108                 GOTO(out_lmv_lock, rc = -ENOTCONN);
 109
 110         CDEBUG(D_INFO, "Found OBD %s=%s device %d (%p) type %s at LMV idx %d\n",
 111                obd->obd_name, obd->obd_uuid.uuid, obd->obd_minor, obd,
 112                obd->obd_type->typ_name, i);
 113         LASSERT(strcmp(obd->obd_type->typ_name, LUSTRE_MDC_NAME) == 0);
 114
 115         if (tgt->active == activate) {
 116                 CDEBUG(D_INFO, "OBD %p already %sactive!\n", obd,
 117                        activate ? "" : "in");
 118                 GOTO(out_lmv_lock, rc);
 119         }
 120
 121         CDEBUG(D_INFO, "Marking OBD %p %sactive\n",
 122                obd, activate ? "" : "in");
 123
 124         lmv_activate_target(lmv, tgt, activate);
 125
 126         EXIT;
 127
 128  out_lmv_lock:
 129         spin_unlock(&lmv->lmv_lock);
 130         return rc;
 131 }
 132
 133 static int lmv_notify(struct obd_device *obd, struct obd_device *watched,
 134                       int active, void *data)
 135 {
 136         struct obd_uuid *uuid;
 137         int rc;
 138         ENTRY;
 139
 140         if (strcmp(watched->obd_type->typ_name, LUSTRE_MDC_NAME)) {
 141                 CERROR("unexpected notification of %s %s!\n",
 142                        watched->obd_type->typ_name,
 143                        watched->obd_name);
 144                 RETURN(-EINVAL);
 145         }
 146         uuid = &watched->u.cli.cl_import->imp_target_uuid;
 147
 148         /* Set MDC as active before notifying the observer, so the observer can
 149          * use the MDC normally.
 150          */
 151         rc = lmv_set_mdc_active(&obd->u.lmv, uuid, active);
 152         if (rc) {
 153                 CERROR("%sactivation of %s failed: %d\n",
 154                        active ? "" : "de", uuid->uuid, rc);
 155                 RETURN(rc);
 156         }
 157
 158         if (obd->obd_observer)
 159                 /* Pass the notification up the chain. */
 160                 rc = obd_notify(obd->obd_observer, watched, active, data);
 161
 162         RETURN(rc);
 163 }
 164
 165 int lmv_attach(struct obd_device *dev, obd_count len, void *data)
 166 {
 167         struct lprocfs_static_vars lvars;
 168         int rc;
 169         ENTRY;
 170
 171         lprocfs_init_vars(lmv, &lvars);
 172         rc = lprocfs_obd_attach(dev, lvars.obd_vars);
 173 #ifdef __KERNEL__
 174         if (rc == 0) {
 175                 struct proc_dir_entry *entry;
 176
 177                 entry = create_proc_entry("target_obd_status", 0444,
 178                                            dev->obd_proc_entry);
 179                 if (entry == NULL)
 180                         RETURN(-ENOMEM);
 181                 entry->proc_fops = &lmv_proc_target_fops;
 182                 entry->data = dev;
 183        }
 184 #endif
 185         RETURN (rc);
 186 }
 187
 188 int lmv_detach(struct obd_device *dev)
 189 {
 190         return lprocfs_obd_detach(dev);
 191 }
 192
 193 /* this is fake connect function. Its purpose is to initialize lmv and say
 194  * caller that everything is okay. Real connection will be performed later. */
 195 static int lmv_connect(struct lustre_handle *conn, struct obd_device *obd,
 196                        struct obd_uuid *cluuid, unsigned long flags)
 197 {
 198 #ifdef __KERNEL__
 199         struct proc_dir_entry *lmv_proc_dir;
 200 #endif
 201         struct lmv_obd *lmv = &obd->u.lmv;
 202         struct obd_export *exp;
 203         int rc = 0;
 204         ENTRY;
 205
 206         rc = class_connect(conn, obd, cluuid);
 207         if (rc) {
 208                 CERROR("class_connection() returned %d\n", rc);
 209                 RETURN(rc);
 210         }
 211
 212         exp = class_conn2export(conn);
 213
 214         /* we don't want to actually do the underlying connections more than
 215          * once, so keep track. */
 216         lmv->refcount++;
 217         if (lmv->refcount > 1) {
 218                 class_export_put(exp);
 219                 RETURN(0);
 220         }
 221
 222         lmv->exp = exp;
 223         lmv->connected = 0;
 224         lmv->cluuid = *cluuid;
 225         lmv->connect_flags = flags;
 226         sema_init(&lmv->init_sem, 1);
 227
 228 #ifdef __KERNEL__
 229         lmv_proc_dir = lprocfs_register("target_obds", obd->obd_proc_entry,
 230                                         NULL, NULL);
 231         if (IS_ERR(lmv_proc_dir)) {
 232                 CERROR("could not register /proc/fs/lustre/%s/%s/target_obds.",
 233                        obd->obd_type->typ_name, obd->obd_name);
 234                 lmv_proc_dir = NULL;
 235         }
 236 #endif
 237
 238         /*
 239          * all real clients shouls perform actual connection rightaway, because
 240          * it is possible, that LMV will not have opportunity to connect
 241          * targets, as MDC stuff will bit called directly, for instance while
 242          * reading ../mdc/../kbytesfree procfs file, etc.
 243          */
 244         if (flags & OBD_OPT_REAL_CLIENT)
 245                 rc = lmv_check_connect(obd);
 246
 247 #ifdef __KERNEL__
 248         if (rc) {
 249                 if (lmv_proc_dir)
 250                         lprocfs_remove(lmv_proc_dir);
 251         }
 252 #endif
 253
 254         RETURN(rc);
 255 }
 256
 257 void lmv_set_timeouts(struct obd_device *obd)
 258 {
 259         struct lmv_tgt_desc *tgts;
 260         struct lmv_obd *lmv;
 261         int i;
 262
 263         lmv = &obd->u.lmv;
 264         if (lmv->server_timeout == 0)
 265                 return;
 266
 267         if (lmv->connected == 0)
 268                 return;
 269
 270         for (i = 0, tgts = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgts++) {
 271                 if (tgts->ltd_exp == NULL)
 272                         continue;
 273
 274                 obd_set_info(tgts->ltd_exp, strlen("inter_mds"),
 275                              "inter_mds", 0, NULL);
 276         }
 277 }
 278
 279 #define MAX_STRING_SIZE 128
 280
 281 /* performs a check if passed obd is connected. If no - connect it. */
 282 int lmv_check_connect(struct obd_device *obd)
 283 {
 284 #ifdef __KERNEL__
 285         struct proc_dir_entry *lmv_proc_dir;
 286 #endif
 287         struct lmv_obd *lmv = &obd->u.lmv;
 288         struct lmv_tgt_desc *tgts;
 289         struct obd_uuid *cluuid;
 290         struct obd_export *exp;
 291         int rc, rc2, i;
 292         ENTRY;
 293
 294         if (lmv->connected)
 295                 RETURN(0);
 296
 297         down(&lmv->init_sem);
 298         if (lmv->connected) {
 299                 up(&lmv->init_sem);
 300                 RETURN(0);
 301         }
 302
 303         cluuid = &lmv->cluuid;
 304         exp = lmv->exp;
 305
 306         CDEBUG(D_OTHER, "time to connect %s to %s\n",
 307                cluuid->uuid, obd->obd_name);
 308
 309         for (i = 0, tgts = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgts++) {
 310                 struct obd_uuid lmv_mdc_uuid = { "LMV_MDC_UUID" };
 311                 struct lustre_handle conn = {0, };
 312                 struct obd_device *tgt_obd;
 313
 314                 LASSERT(tgts != NULL);
 315
 316                 tgt_obd = class_find_client_obd(&tgts->uuid, LUSTRE_MDC_NAME,
 317                                                 &obd->obd_uuid);
 318                 if (!tgt_obd) {
 319                         CERROR("target %s not attached\n", tgts->uuid.uuid);
 320                         GOTO(out_disc, rc = -EINVAL);
 321                 }
 322
 323                 /* for MDS: don't connect to yourself */
 324                 if (obd_uuid_equals(&tgts->uuid, cluuid)) {
 325                         CDEBUG(D_OTHER, "don't connect back to %s\n",
 326                                cluuid->uuid);
 327                         tgts->ltd_exp = NULL;
 328                         continue;
 329                 }
 330
 331                 CDEBUG(D_OTHER, "connect to %s(%s) - %s, %s FOR %s\n",
 332                         tgt_obd->obd_name, tgt_obd->obd_uuid.uuid,
 333                         tgts->uuid.uuid, obd->obd_uuid.uuid,
 334                         cluuid->uuid);
 335
 336                 if (!tgt_obd->obd_set_up) {
 337                         CERROR("target %s not set up\n", tgts->uuid.uuid);
 338                         GOTO(out_disc, rc = -EINVAL);
 339                 }
 340
 341                 rc = obd_connect(&conn, tgt_obd, &lmv_mdc_uuid,
 342                                  lmv->connect_flags);
 343                 if (rc) {
 344                         CERROR("target %s connect error %d\n",
 345                                 tgts->uuid.uuid, rc);
 346                         GOTO(out_disc, rc);
 347                 }
 348                 tgts->ltd_exp = class_conn2export(&conn);
 349
 350                 obd_init_ea_size(tgts->ltd_exp, lmv->max_easize,
 351                                  lmv->max_cookiesize);
 352
 353                 rc = obd_register_observer(tgt_obd, obd);
 354                 if (rc) {
 355                         CERROR("target %s register_observer error %d\n",
 356                                tgts->uuid.uuid, rc);
 357                         obd_disconnect(tgts->ltd_exp, 0);
 358                         GOTO(out_disc, rc);
 359                 }
 360
 361                 lmv->desc.ld_active_tgt_count++;
 362                 tgts->active = 1;
 363
 364                 CDEBUG(D_OTHER, "connected to %s(%s) successfully (%d)\n",
 365                         tgt_obd->obd_name, tgt_obd->obd_uuid.uuid,
 366                         atomic_read(&obd->obd_refcount));
 367
 368 #ifdef __KERNEL__
 369                 lmv_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 370                 if (lmv_proc_dir) {
 371                         struct obd_device *mdc_obd = class_conn2obd(&conn);
 372                         struct proc_dir_entry *mdc_symlink;
 373                         char name[MAX_STRING_SIZE + 1];
 374
 375                         LASSERT(mdc_obd != NULL);
 376                         LASSERT(mdc_obd->obd_type != NULL);
 377                         LASSERT(mdc_obd->obd_type->typ_name != NULL);
 378                         name[MAX_STRING_SIZE] = '\0';
 379                         snprintf(name, MAX_STRING_SIZE, "../../../%s/%s",
 380                                  mdc_obd->obd_type->typ_name,
 381                                  mdc_obd->obd_name);
 382                         mdc_symlink = proc_symlink(mdc_obd->obd_name,
 383                                                    lmv_proc_dir, name);
 384                         if (mdc_symlink == NULL) {
 385                                 CERROR("could not register LMV target "
 386                                        "/proc/fs/lustre/%s/%s/target_obds/%s.",
 387                                        obd->obd_type->typ_name, obd->obd_name,
 388                                        mdc_obd->obd_name);
 389                                 lprocfs_remove(lmv_proc_dir);
 390                                 lmv_proc_dir = NULL;
 391                         }
 392                 }
 393 #endif
 394         }
 395
 396         lmv_set_timeouts(obd);
 397         class_export_put(exp);
 398         lmv->connected = 1;
 399         up(&lmv->init_sem);
 400         RETURN(0);
 401
 402  out_disc:
 403         while (i-- > 0) {
 404                 struct obd_uuid uuid;
 405                 --tgts;
 406                 --lmv->desc.ld_active_tgt_count;
 407                 tgts->active = 0;
 408                 /* save for CERROR below; (we know it's terminated) */
 409                 uuid = tgts->uuid;
 410                 rc2 = obd_disconnect(tgts->ltd_exp, 0);
 411                 if (rc2)
 412                         CERROR("error: LMV target %s disconnect on MDC idx %d: "
 413                                "error %d\n", uuid.uuid, i, rc2);
 414         }
 415         class_disconnect(exp, 0);
 416         up(&lmv->init_sem);
 417         return rc;
 418 }
 419
 420 static int lmv_disconnect(struct obd_export *exp, unsigned long flags)
 421 {
 422         struct obd_device *obd = class_exp2obd(exp);
 423         struct lmv_obd *lmv = &obd->u.lmv;
 424
 425 #ifdef __KERNEL__
 426         struct proc_dir_entry *lmv_proc_dir;
 427 #endif
 428         int rc, i;
 429         ENTRY;
 430
 431         if (!lmv->tgts)
 432                 goto out_local;
 433
 434         /* Only disconnect the underlying layers on the final disconnect. */
 435         lmv->refcount--;
 436         if (lmv->refcount != 0)
 437                 goto out_local;
 438
 439 #ifdef __KERNEL__
 440         lmv_proc_dir = lprocfs_srch(obd->obd_proc_entry, "target_obds");
 441 #endif
 442
 443         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 444                 struct obd_device *mdc_obd;
 445
 446                 if (lmv->tgts[i].ltd_exp == NULL)
 447                         continue;
 448
 449                 mdc_obd = class_exp2obd(lmv->tgts[i].ltd_exp);
 450
 451                 if (mdc_obd)
 452                         mdc_obd->obd_no_recov = obd->obd_no_recov;
 453
 454 #ifdef __KERNEL__
 455                 if (lmv_proc_dir) {
 456                         struct proc_dir_entry *mdc_symlink;
 457
 458                         mdc_symlink = lprocfs_srch(lmv_proc_dir, mdc_obd->obd_name);
 459                         if (mdc_symlink) {
 460                                 lprocfs_remove(mdc_symlink);
 461                         } else {
 462                                 CERROR("/proc/fs/lustre/%s/%s/target_obds/%s missing\n",
 463                                        obd->obd_type->typ_name, obd->obd_name,
 464                                        mdc_obd->obd_name);
 465                         }
 466                 }
 467 #endif
 468                 CDEBUG(D_OTHER, "disconnected from %s(%s) successfully\n",
 469                         lmv->tgts[i].ltd_exp->exp_obd->obd_name,
 470                         lmv->tgts[i].ltd_exp->exp_obd->obd_uuid.uuid);
 471
 472                 obd_register_observer(lmv->tgts[i].ltd_exp->exp_obd, NULL);
 473                 rc = obd_disconnect(lmv->tgts[i].ltd_exp, flags);
 474                 if (rc) {
 475                         if (lmv->tgts[i].active) {
 476                                 CERROR("Target %s disconnect error %d\n",
 477                                        lmv->tgts[i].uuid.uuid, rc);
 478                         }
 479                         rc = 0;
 480                 }
 481
 482                 lmv_activate_target(lmv, &lmv->tgts[i], 0);
 483                 lmv->tgts[i].ltd_exp = NULL;
 484         }
 485
 486 #ifdef __KERNEL__
 487         if (lmv_proc_dir) {
 488                 lprocfs_remove(lmv_proc_dir);
 489         } else {
 490                 CERROR("/proc/fs/lustre/%s/%s/target_obds missing\n",
 491                        obd->obd_type->typ_name, obd->obd_name);
 492         }
 493 #endif
 494
 495 out_local:
 496         /* this is the case when no real connection is established by
 497          * lmv_check_connect(). */
 498         if (!lmv->connected)
 499                 class_export_put(exp);
 500         rc = class_disconnect(exp, 0);
 501         if (lmv->refcount == 0)
 502                 lmv->connected = 0;
 503         RETURN(rc);
 504 }
 505
 506 static int lmv_iocontrol(unsigned int cmd, struct obd_export *exp,
 507                          int len, void *karg, void *uarg)
 508 {
 509         struct obd_device *obddev = class_exp2obd(exp);
 510         struct lmv_obd *lmv = &obddev->u.lmv;
 511         int i, rc = 0, set = 0;
 512         ENTRY;
 513
 514         if (lmv->desc.ld_tgt_count == 0)
 515                 RETURN(-ENOTTY);
 516
 517         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 518                 int err;
 519
 520                 if (lmv->tgts[i].ltd_exp == NULL)
 521                         continue;
 522
 523                 err = obd_iocontrol(cmd, lmv->tgts[i].ltd_exp, len, karg, uarg);
 524                 if (err) {
 525                         if (lmv->tgts[i].active) {
 526                                 CERROR("error: iocontrol MDC %s on MDT"
 527                                        "idx %d: err = %d\n",
 528                                        lmv->tgts[i].uuid.uuid, i, err);
 529                                 if (!rc)
 530                                         rc = err;
 531                         }
 532                 } else
 533                         set = 1;
 534         }
 535         if (!set && !rc)
 536                 rc = -EIO;
 537
 538         RETURN(rc);
 539 }
 540
 541 static int lmv_setup(struct obd_device *obd, obd_count len, void *buf)
 542 {
 543         int i, rc = 0;
 544         struct lmv_desc *desc;
 545         struct obd_uuid *uuids;
 546         struct lmv_tgt_desc *tgts;
 547         struct obd_device *tgt_obd;
 548         struct lustre_cfg *lcfg = buf;
 549         struct lmv_obd *lmv = &obd->u.lmv;
 550         ENTRY;
 551
 552         if (lcfg->lcfg_inllen1 < 1) {
 553                 CERROR("LMV setup requires a descriptor\n");
 554                 RETURN(-EINVAL);
 555         }
 556
 557         if (lcfg->lcfg_inllen2 < 1) {
 558                 CERROR("LMV setup requires an MDT UUID list\n");
 559                 RETURN(-EINVAL);
 560         }
 561
 562         desc = (struct lmv_desc *)lcfg->lcfg_inlbuf1;
 563         if (sizeof(*desc) > lcfg->lcfg_inllen1) {
 564                 CERROR("descriptor size wrong: %d > %d\n",
 565                        (int)sizeof(*desc), lcfg->lcfg_inllen1);
 566                 RETURN(-EINVAL);
 567         }
 568
 569         uuids = (struct obd_uuid *)lcfg->lcfg_inlbuf2;
 570         if (sizeof(*uuids) * desc->ld_tgt_count != lcfg->lcfg_inllen2) {
 571                 CERROR("UUID array size wrong: %u * %u != %u\n",
 572                        sizeof(*uuids), desc->ld_tgt_count, lcfg->lcfg_inllen2);
 573                 RETURN(-EINVAL);
 574         }
 575
 576         lmv->tgts_size = sizeof(struct lmv_tgt_desc) * desc->ld_tgt_count;
 577         OBD_ALLOC(lmv->tgts, lmv->tgts_size);
 578         if (lmv->tgts == NULL) {
 579                 CERROR("Out of memory\n");
 580                 RETURN(-ENOMEM);
 581         }
 582
 583         lmv->desc = *desc;
 584         spin_lock_init(&lmv->lmv_lock);
 585
 586         for (i = 0, tgts = lmv->tgts; i < desc->ld_tgt_count; i++, tgts++)
 587                 tgts->uuid = uuids[i];
 588
 589         lmv->max_cookiesize = 0;
 590
 591         lmv->max_easize = sizeof(struct lustre_id) *
 592                 desc->ld_tgt_count + sizeof(struct mea);
 593
 594         rc = lmv_setup_mgr(obd);
 595         if (rc) {
 596                 CERROR("Can't setup LMV object manager, "
 597                        "error %d.\n", rc);
 598                 OBD_FREE(lmv->tgts, lmv->tgts_size);
 599         }
 600
 601         tgt_obd = class_find_client_obd(&lmv->tgts->uuid, LUSTRE_MDC_NAME,
 602                                         &obd->obd_uuid);
 603         if (!tgt_obd) {
 604                 CERROR("Target %s not attached\n", lmv->tgts->uuid.uuid);
 605                 RETURN(-EINVAL);
 606         }
 607
 608         RETURN(rc);
 609 }
 610
 611 static int lmv_cleanup(struct obd_device *obd, int flags)
 612 {
 613         struct lmv_obd *lmv = &obd->u.lmv;
 614         ENTRY;
 615
 616         lmv_cleanup_mgr(obd);
 617         OBD_FREE(lmv->tgts, lmv->tgts_size);
 618
 619         RETURN(0);
 620 }
 621
 622 static int lmv_statfs(struct obd_device *obd, struct obd_statfs *osfs,
 623                       unsigned long max_age)
 624 {
 625         struct lmv_obd *lmv = &obd->u.lmv;
 626         struct obd_statfs *temp;
 627         int rc = 0, i;
 628         ENTRY;
 629
 630         rc = lmv_check_connect(obd);
 631         if (rc)
 632                 RETURN(rc);
 633
 634         OBD_ALLOC(temp, sizeof(*temp));
 635         if (temp == NULL)
 636                 RETURN(-ENOMEM);
 637
 638         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
 639                 if (lmv->tgts[i].ltd_exp == NULL) {
 640                         CWARN("%s: NULL export for %d\n", obd->obd_name, i);
 641                         continue;
 642                 }
 643
 644                 rc = obd_statfs(lmv->tgts[i].ltd_exp->exp_obd, temp, max_age);
 645                 if (rc) {
 646                         CERROR("can't stat MDS #%d (%s), error %d\n", i,
 647                                lmv->tgts[i].ltd_exp->exp_obd->obd_name,
 648                                rc);
 649                         GOTO(out_free_temp, rc);
 650                 }
 651                 if (i == 0) {
 652                         memcpy(osfs, temp, sizeof(*temp));
 653                 } else {
 654                         osfs->os_bavail += temp->os_bavail;
 655                         osfs->os_blocks += temp->os_blocks;
 656                         osfs->os_ffree += temp->os_ffree;
 657                         osfs->os_files += temp->os_files;
 658                 }
 659         }
 660
 661         EXIT;
 662 out_free_temp:
 663         OBD_FREE(temp, sizeof(*temp));
 664         return rc;
 665 }
 666
 667 static int lmv_getstatus(struct obd_export *exp, struct lustre_id *id)
 668 {
 669         struct obd_device *obd = exp->exp_obd;
 670         struct lmv_obd *lmv = &obd->u.lmv;
 671         int rc;
 672         ENTRY;
 673
 674         rc = lmv_check_connect(obd);
 675         if (rc)
 676                 RETURN(rc);
 677
 678         rc = md_getstatus(lmv->tgts[0].ltd_exp, id);
 679         id_group(id) = 0;
 680
 681         RETURN(rc);
 682 }
 683
 684 static int lmv_getattr(struct obd_export *exp, struct lustre_id *id,
 685                        __u64 valid, const char *ea_name, int ea_namelen,
 686                        unsigned int ea_size, struct ptlrpc_request **request)
 687 {
 688         struct obd_device *obd = exp->exp_obd;
 689         struct lmv_obd *lmv = &obd->u.lmv;
 690         int rc, i = id_group(id);
 691         struct lmv_obj *obj;
 692         ENTRY;
 693
 694         rc = lmv_check_connect(obd);
 695         if (rc)
 696                 RETURN(rc);
 697
 698         LASSERT(i < lmv->desc.ld_tgt_count);
 699
 700
 701         rc = md_getattr(lmv->tgts[i].ltd_exp, id, valid,
 702                         ea_name, ea_namelen, ea_size, request);
 703         if (rc)
 704                 RETURN(rc);
 705
 706         obj = lmv_grab_obj(obd, id);
 707
 708         CDEBUG(D_OTHER, "GETATTR for "DLID4" %s\n",
 709                OLID4(id), obj ? "(splitted)" : "");
 710
 711         /*
 712          * if object is splitted, then we loop over all the slaves and gather
 713          * size attribute. In ideal world we would have to gather also mds field
 714          * from all slaves, as object is spread over the cluster and this is
 715          * definitely interesting information and it is not good to loss it,
 716          * but...
 717          */
 718         if (obj) {
 719                 struct mds_body *body;
 720
 721                 if (*request == NULL) {
 722                         lmv_put_obj(obj);
 723                         RETURN(rc);
 724                 }
 725
 726                 body = lustre_msg_buf((*request)->rq_repmsg, 0,
 727                                       sizeof(*body));
 728                 LASSERT(body != NULL);
 729
 730                 lmv_lock_obj(obj);
 731
 732                 for (i = 0; i < obj->objcount; i++) {
 733
 734                         if (lmv->tgts[i].ltd_exp == NULL) {
 735                                 CWARN("%s: NULL export for %d\n",
 736                                       obd->obd_name, i);
 737                                 continue;
 738                         }
 739
 740                         /* skip master obj. */
 741                         if (id_equal_fid(&obj->id, &obj->objs[i].id))
 742                                 continue;
 743
 744                         body->size += obj->objs[i].size;
 745                 }
 746
 747                 lmv_unlock_obj(obj);
 748                 lmv_put_obj(obj);
 749         }
 750
 751         RETURN(rc);
 752 }
 753
 754 static int lmv_change_cbdata(struct obd_export *exp,
 755                              struct lustre_id *id,
 756                              ldlm_iterator_t it,
 757                              void *data)
 758 {
 759         struct obd_device *obd = exp->exp_obd;
 760         struct lmv_obd *lmv = &obd->u.lmv;
 761         int rc = 0;
 762         ENTRY;
 763
 764         rc = lmv_check_connect(obd);
 765         if (rc)
 766                 RETURN(rc);
 767
 768         CDEBUG(D_OTHER, "CBDATA for "DLID4"\n", OLID4(id));
 769         LASSERT(id_group(id) < lmv->desc.ld_tgt_count);
 770
 771         rc = md_change_cbdata(lmv->tgts[id_group(id)].ltd_exp,
 772                               id, it, data);
 773
 774         RETURN(rc);
 775 }
 776
 777 static int lmv_change_cbdata_name(struct obd_export *exp,
 778                                   struct lustre_id *pid,
 779                                   char *name, int len,
 780                                   struct lustre_id *cid,
 781                                   ldlm_iterator_t it,
 782                                   void *data)
 783 {
 784         struct obd_device *obd = exp->exp_obd;
 785         struct lmv_obd *lmv = &obd->u.lmv;
 786         struct lustre_id rcid = *cid;
 787         struct lmv_obj *obj;
 788         int rc = 0, mds;
 789         ENTRY;
 790
 791         rc = lmv_check_connect(obd);
 792         if (rc)
 793                 RETURN(rc);
 794
 795         LASSERT(id_group(pid) < lmv->desc.ld_tgt_count);
 796         LASSERT(id_group(cid) < lmv->desc.ld_tgt_count);
 797
 798         CDEBUG(D_OTHER, "CBDATA for "DLID4":%*s -> "DLID4"\n",
 799                OLID4(pid), len, name, OLID4(cid));
 800
 801         /* this is default mds for directory name belongs to. */
 802         mds = id_group(pid);
 803         obj = lmv_grab_obj(obd, pid);
 804         if (obj) {
 805                 /* directory is splitted. look for right mds for this name. */
 806                 mds = raw_name2idx(obj->hashtype, obj->objcount, name, len);
 807                 rcid = obj->objs[mds].id;
 808                 mds = id_group(&rcid);
 809                 lmv_put_obj(obj);
 810         }
 811         rc = md_change_cbdata(lmv->tgts[mds].ltd_exp, &rcid, it, data);
 812         RETURN(rc);
 813 }
 814
 815 static int lmv_valid_attrs(struct obd_export *exp, struct lustre_id *id)
 816 {
 817         struct obd_device *obd = exp->exp_obd;
 818         struct lmv_obd *lmv = &obd->u.lmv;
 819         int rc = 0;
 820         ENTRY;
 821
 822         rc = lmv_check_connect(obd);
 823         if (rc)
 824                 RETURN(rc);
 825
 826         CDEBUG(D_OTHER, "validate "DLID4"\n", OLID4(id));
 827         LASSERT(id_group(id) < lmv->desc.ld_tgt_count);
 828         rc = md_valid_attrs(lmv->tgts[id_group(id)].ltd_exp, id);
 829         RETURN(rc);
 830 }
 831
 832 int lmv_close(struct obd_export *exp, struct obdo *obdo,
 833               struct obd_client_handle *och,
 834               struct ptlrpc_request **request)
 835 {
 836         struct obd_device *obd = exp->exp_obd;
 837         struct lmv_obd *lmv = &obd->u.lmv;
 838         int rc, i = obdo->o_mds;
 839         ENTRY;
 840
 841         rc = lmv_check_connect(obd);
 842         if (rc)
 843                 RETURN(rc);
 844
 845         LASSERT(i < lmv->desc.ld_tgt_count);
 846         CDEBUG(D_OTHER, "CLOSE %lu/%lu/%lu\n", (unsigned long)obdo->o_mds,
 847                (unsigned long)obdo->o_id, (unsigned long)obdo->o_generation);
 848         rc = md_close(lmv->tgts[i].ltd_exp, obdo, och, request);
 849         RETURN(rc);
 850 }
 851
 852 int lmv_get_mea_and_update_object(struct obd_export *exp,
 853                                   struct lustre_id *id)
 854 {
 855         struct obd_device *obd = exp->exp_obd;
 856         struct lmv_obd *lmv = &obd->u.lmv;
 857         struct ptlrpc_request *req = NULL;
 858         struct lmv_obj *obj;
 859         struct lustre_md md;
 860         int mealen, rc;
 861         __u64 valid;
 862         ENTRY;
 863
 864         md.mea = NULL;
 865         mealen = MEA_SIZE_LMV(lmv);
 866
 867         valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA | OBD_MD_MEA;
 868
 869         /* time to update mea of parent id */
 870         rc = md_getattr(lmv->tgts[id_group(id)].ltd_exp,
 871                         id, valid, NULL, 0, mealen, &req);
 872         if (rc) {
 873                 CERROR("md_getattr() failed, error %d\n", rc);
 874                 GOTO(cleanup, rc);
 875         }
 876
 877         rc = mdc_req2lustre_md(exp, req, 0, NULL, &md);
 878         if (rc) {
 879                 CERROR("mdc_req2lustre_md() failed, error %d\n", rc);
 880                 GOTO(cleanup, rc);
 881         }
 882
 883         if (md.mea == NULL)
 884                 GOTO(cleanup, rc = -ENODATA);
 885
 886         obj = lmv_create_obj(exp, id, md.mea);
 887         if (IS_ERR(obj))
 888                 rc = PTR_ERR(obj);
 889
 890         lmv_put_obj(obj);
 891         obd_free_memmd(exp, (struct lov_stripe_md **)&md.mea);
 892
 893         EXIT;
 894 cleanup:
 895         if (req)
 896                 ptlrpc_req_finished(req);
 897         return rc;
 898 }
 899
 900 int lmv_create(struct obd_export *exp, struct mdc_op_data *op_data,
 901                const void *data, int datalen, int mode, __u32 uid,
 902                __u32 gid, __u64 rdev, struct ptlrpc_request **request)
 903 {
 904         struct obd_device *obd = exp->exp_obd;
 905         struct lmv_obd *lmv = &obd->u.lmv;
 906         struct mds_body *body;
 907         struct lmv_obj *obj;
 908         int rc, mds, loop = 0;
 909         ENTRY;
 910
 911         rc = lmv_check_connect(obd);
 912         if (rc)
 913                 RETURN(rc);
 914
 915         if (!lmv->desc.ld_active_tgt_count)
 916                 RETURN(-EIO);
 917 repeat:
 918         LASSERT(++loop <= 2);
 919         obj = lmv_grab_obj(obd, &op_data->id1);
 920         if (obj) {
 921                 mds = raw_name2idx(obj->hashtype, obj->objcount,
 922                                    op_data->name, op_data->namelen);
 923                 op_data->id1 = obj->objs[mds].id;
 924                 lmv_put_obj(obj);
 925         }
 926
 927         CDEBUG(D_OTHER, "CREATE '%*s' on "DLID4"\n", op_data->namelen,
 928                op_data->name, OLID4(&op_data->id1));
 929
 930         rc = md_create(lmv->tgts[id_group(&op_data->id1)].ltd_exp,
 931                        op_data, data, datalen, mode, uid, gid, rdev,
 932                        request);
 933         if (rc == 0) {
 934                 if (*request == NULL)
 935                         RETURN(rc);
 936
 937                 body = lustre_msg_buf((*request)->rq_repmsg, 0,
 938                                       sizeof(*body));
 939                 if (body == NULL)
 940                         RETURN(-ENOMEM);
 941
 942                 CDEBUG(D_OTHER, "created. "DLID4"\n", OLID4(&op_data->id1));
 943         } else if (rc == -ERESTART) {
 944                 /*
 945                  * directory got splitted. time to update local object and
 946                  * repeat the request with proper MDS.
 947                  */
 948                 rc = lmv_get_mea_and_update_object(exp, &op_data->id1);
 949                 if (rc == 0) {
 950                         ptlrpc_req_finished(*request);
 951                         goto repeat;
 952                 }
 953         }
 954         RETURN(rc);
 955 }
 956
 957 int lmv_done_writing(struct obd_export *exp, struct obdo *obdo)
 958 {
 959         struct obd_device *obd = exp->exp_obd;
 960         struct lmv_obd *lmv = &obd->u.lmv;
 961         int rc;
 962         ENTRY;
 963
 964         rc = lmv_check_connect(obd);
 965         if (rc)
 966                 RETURN(rc);
 967
 968         /* FIXME: choose right MDC here */
 969         CWARN("this method isn't implemented yet\n");
 970         rc = md_done_writing(lmv->tgts[0].ltd_exp, obdo);
 971         RETURN(rc);
 972 }
 973
 974 int lmv_enqueue_slaves(struct obd_export *exp, int locktype,
 975                        struct lookup_intent *it, int lockmode,
 976                        struct mdc_op_data *data, struct lustre_handle *lockh,
 977                        void *lmm, int lmmsize, ldlm_completion_callback cb_compl,
 978                        ldlm_blocking_callback cb_blocking, void *cb_data)
 979 {
 980         struct obd_device *obd = exp->exp_obd;
 981         struct lmv_obd *lmv = &obd->u.lmv;
 982         struct mea *mea = data->mea1;
 983         struct mdc_op_data *data2;
 984         int i, rc, mds;
 985         ENTRY;
 986
 987         OBD_ALLOC(data2, sizeof(*data2));
 988         if (data2 == NULL)
 989                 RETURN(-ENOMEM);
 990
 991         LASSERT(mea != NULL);
 992         for (i = 0; i < mea->mea_count; i++) {
 993                 memset(data2, 0, sizeof(*data2));
 994                 data2->id1 = mea->mea_ids[i];
 995                 mds = id_group(&data2->id1);
 996
 997                 if (lmv->tgts[mds].ltd_exp == NULL)
 998                         continue;
 999
1000                 rc = md_enqueue(lmv->tgts[mds].ltd_exp, locktype, it,
1001                                 lockmode, data2, lockh + i, lmm, lmmsize,
1002                                 cb_compl, cb_blocking, cb_data);
1003
1004                 CDEBUG(D_OTHER, "take lock on slave "DLID4" -> %d/%d\n",
1005                        OLID4(&mea->mea_ids[i]), rc, LUSTRE_IT(it)->it_status);
1006                 if (rc)
1007                         GOTO(cleanup, rc);
1008                 if (LUSTRE_IT(it)->it_data) {
1009                         struct ptlrpc_request *req;
1010                         req = (struct ptlrpc_request *) LUSTRE_IT(it)->it_data;
1011                         ptlrpc_req_finished(req);
1012                 }
1013
1014                 if (LUSTRE_IT(it)->it_status)
1015                         GOTO(cleanup, rc = LUSTRE_IT(it)->it_status);
1016         }
1017
1018         OBD_FREE(data2, sizeof(*data2));
1019         RETURN(0);
1020 cleanup:
1021         OBD_FREE(data2, sizeof(*data2));
1022
1023         /* drop all taken locks */
1024         while (--i >= 0) {
1025                 if (lockh[i].cookie)
1026                         ldlm_lock_decref(lockh + i, lockmode);
1027                 lockh[i].cookie = 0;
1028         }
1029         return rc;
1030 }
1031
1032 int lmv_enqueue(struct obd_export *exp, int lock_type,
1033                 struct lookup_intent *it, int lock_mode,
1034                 struct mdc_op_data *data, struct lustre_handle *lockh,
1035                 void *lmm, int lmmsize, ldlm_completion_callback cb_compl,
1036                 ldlm_blocking_callback cb_blocking, void *cb_data)
1037 {
1038         struct obd_device *obd = exp->exp_obd;
1039         struct lmv_obd *lmv = &obd->u.lmv;
1040         struct lmv_obj *obj;
1041         int rc, mds;
1042         ENTRY;
1043
1044         rc = lmv_check_connect(obd);
1045         if (rc)
1046                 RETURN(rc);
1047
1048         if (data->mea1 && it->it_op == IT_UNLINK) {
1049                 rc = lmv_enqueue_slaves(exp, lock_type, it, lock_mode,
1050                                         data, lockh, lmm, lmmsize,
1051                                         cb_compl, cb_blocking, cb_data);
1052                 RETURN(rc);
1053         }
1054
1055         if (data->namelen) {
1056                 obj = lmv_grab_obj(obd, &data->id1);
1057                 if (obj) {
1058                         /* directory is splitted. look for right mds for this
1059                          * name */
1060                         mds = raw_name2idx(obj->hashtype, obj->objcount,
1061                                            (char *)data->name, data->namelen);
1062                         data->id1 = obj->objs[mds].id;
1063                         lmv_put_obj(obj);
1064                 }
1065         }
1066         CDEBUG(D_OTHER, "ENQUEUE '%s' on "DLID4"\n", LL_IT2STR(it),
1067                OLID4(&data->id1));
1068
1069         rc = md_enqueue(lmv->tgts[id_group(&data->id1)].ltd_exp,
1070                         lock_type, it, lock_mode, data, lockh, lmm,
1071                         lmmsize, cb_compl, cb_blocking, cb_data);
1072         RETURN(rc);
1073 }
1074
1075 int lmv_getattr_lock(struct obd_export *exp, struct lustre_id *id,
1076                      char *filename, int namelen, __u64 valid,
1077                      unsigned int ea_size, struct ptlrpc_request **request)
1078 {
1079         int rc, mds = id_group(id), loop = 0;
1080         struct obd_device *obd = exp->exp_obd;
1081         struct lmv_obd *lmv = &obd->u.lmv;
1082         struct lustre_id rid = *id;
1083         struct mds_body *body;
1084         struct lmv_obj *obj;
1085         ENTRY;
1086
1087         rc = lmv_check_connect(obd);
1088         if (rc)
1089                 RETURN(rc);
1090 repeat:
1091         LASSERT(++loop <= 2);
1092         obj = lmv_grab_obj(obd, id);
1093         if (obj) {
1094                 /* directory is splitted. look for right mds for this name */
1095                 mds = raw_name2idx(obj->hashtype, obj->objcount,
1096                                    filename, namelen - 1);
1097                 rid = obj->objs[mds].id;
1098                 lmv_put_obj(obj);
1099         }
1100
1101         CDEBUG(D_OTHER, "getattr_lock for %*s on "DLID4" -> "DLID4"\n",
1102                namelen, filename, OLID4(id), OLID4(&rid));
1103
1104         rc = md_getattr_lock(lmv->tgts[id_group(&rid)].ltd_exp, &rid,
1105                              filename, namelen, (valid | OBD_MD_FID),
1106                              ea_size, request);
1107         if (rc == 0) {
1108                 /*
1109                  * this could be cross-node reference. in this case all we have
1110                  * right now is lustre_id triple. we'd like to find other
1111                  * attributes.
1112                  */
1113                 body = lustre_msg_buf((*request)->rq_repmsg, 0, sizeof(*body));
1114                 LASSERT(body != NULL);
1115                 LASSERT((body->valid & OBD_MD_FID) != 0);
1116
1117                 if (body->valid & OBD_MD_MDS) {
1118                         struct ptlrpc_request *req = NULL;
1119
1120                         rid = body->id1;
1121                         CDEBUG(D_OTHER, "request attrs for "DLID4"\n", OLID4(&rid));
1122
1123                         rc = md_getattr_lock(lmv->tgts[id_group(&rid)].ltd_exp,
1124                                              &rid, NULL, 1, valid, ea_size, &req);
1125                         ptlrpc_req_finished(*request);
1126                         *request = req;
1127                 }
1128         } else if (rc == -ERESTART) {
1129                 /* directory got splitted. time to update local object and
1130                  * repeat the request with proper MDS */
1131                 rc = lmv_get_mea_and_update_object(exp, &rid);
1132                 if (rc == 0) {
1133                         ptlrpc_req_finished(*request);
1134                         goto repeat;
1135                 }
1136         }
1137         RETURN(rc);
1138 }
1139
1140 /*
1141  * llite passes id of an target inode in data->id1 and id of directory in
1142  * data->id2
1143  */
1144 int lmv_link(struct obd_export *exp, struct mdc_op_data *data,
1145              struct ptlrpc_request **request)
1146 {
1147         struct obd_device *obd = exp->exp_obd;
1148         struct lmv_obd *lmv = &obd->u.lmv;
1149         struct lmv_obj *obj;
1150         int rc;
1151         ENTRY;
1152
1153         rc = lmv_check_connect(obd);
1154         if (rc)
1155                 RETURN(rc);
1156
1157         if (data->namelen != 0) {
1158                 /* usual link request */
1159                 obj = lmv_grab_obj(obd, &data->id1);
1160                 if (obj) {
1161                         rc = raw_name2idx(obj->hashtype, obj->objcount,
1162                                           data->name, data->namelen);
1163                         data->id1 = obj->objs[rc].id;
1164                         lmv_put_obj(obj);
1165                 }
1166
1167                 CDEBUG(D_OTHER,"link "DLID4":%*s to "DLID4"\n",
1168                        OLID4(&data->id2), data->namelen, data->name,
1169                        OLID4(&data->id1));
1170         } else {
1171                 /* request from MDS to acquire i_links for inode by id1 */
1172                 CDEBUG(D_OTHER, "inc i_nlinks for "DLID4"\n",
1173                        OLID4(&data->id1));
1174         }
1175
1176         rc = md_link(lmv->tgts[id_group(&data->id1)].ltd_exp,
1177                      data, request);
1178         RETURN(rc);
1179 }
1180
1181 int lmv_rename(struct obd_export *exp, struct mdc_op_data *data,
1182                const char *old, int oldlen, const char *new, int newlen,
1183                struct ptlrpc_request **request)
1184 {
1185         struct obd_device *obd = exp->exp_obd;
1186         struct lmv_obd *lmv = &obd->u.lmv;
1187         struct lmv_obj *obj;
1188         int rc, mds;
1189         ENTRY;
1190
1191         CDEBUG(D_OTHER, "rename %*s in "DLID4" to %*s in "DLID4"\n",
1192                oldlen, old, OLID4(&data->id1), newlen, new,
1193                OLID4(&data->id2));
1194
1195         rc = lmv_check_connect(obd);
1196         if (rc)
1197                 RETURN(rc);
1198
1199         if (oldlen == 0) {
1200                 /*
1201                  * MDS with old dir entry is asking another MDS to create name
1202                  * there.
1203                  */
1204                 CDEBUG(D_OTHER,
1205                        "create %*s(%d/%d) in "DLID4" pointing "
1206                        "to "DLID4"\n", newlen, new, oldlen, newlen,
1207                        OLID4(&data->id2), OLID4(&data->id1));
1208
1209                 mds = id_group(&data->id2);
1210
1211                 /*
1212                  * target directory can be splitted, sowe should forward request
1213                  * to the right MDS.
1214                  */
1215                 obj = lmv_grab_obj(obd, &data->id2);
1216                 if (obj) {
1217                         mds = raw_name2idx(obj->hashtype, obj->objcount,
1218                                            (char *)new, newlen);
1219                         data->id2 = obj->objs[mds].id;
1220                         CDEBUG(D_OTHER, "forward to MDS #%u ("DLID4")\n", mds,
1221                                OLID4(&data->id2));
1222                         lmv_put_obj(obj);
1223                 }
1224                 goto request;
1225         }
1226
1227         obj = lmv_grab_obj(obd, &data->id1);
1228         if (obj) {
1229                 /*
1230                  * directory is already splitted, so we have to forward request
1231                  * to the right MDS.
1232                  */
1233                 mds = raw_name2idx(obj->hashtype, obj->objcount,
1234                                    (char *)old, oldlen);
1235                 data->id1 = obj->objs[mds].id;
1236                 CDEBUG(D_OTHER, "forward to MDS #%u ("DLID4")\n", mds,
1237                        OLID4(&data->id1));
1238                 lmv_put_obj(obj);
1239         }
1240
1241         obj = lmv_grab_obj(obd, &data->id2);
1242         if (obj) {
1243                 /*
1244                  * directory is already splitted, so we have to forward request
1245                  * to the right MDS.
1246                  */
1247                 mds = raw_name2idx(obj->hashtype, obj->objcount,
1248                                    (char *)new, newlen);
1249
1250                 data->id2 = obj->objs[mds].id;
1251                 CDEBUG(D_OTHER, "forward to MDS #%u ("DLID4")\n", mds,
1252                        OLID4(&data->id2));
1253                 lmv_put_obj(obj);
1254         }
1255
1256         mds = id_group(&data->id1);
1257
1258 request:
1259         if (id_group(&data->id1) != id_group(&data->id2)) {
1260                 CDEBUG(D_OTHER,"cross-node rename "DLID4"/%*s to "DLID4"/%*s\n",
1261                        OLID4(&data->id1), oldlen, old, OLID4(&data->id2),
1262                        newlen, new);
1263         }
1264
1265         rc = md_rename(lmv->tgts[mds].ltd_exp, data, old, oldlen,
1266                        new, newlen, request);
1267         RETURN(rc);
1268 }
1269
1270 int lmv_setattr(struct obd_export *exp, struct mdc_op_data *data,
1271                 struct iattr *iattr, void *ea, int ealen, void *ea2,
1272                 int ea2len, struct ptlrpc_request **request)
1273 {
1274         struct obd_device *obd = exp->exp_obd;
1275         struct lmv_obd *lmv = &obd->u.lmv;
1276         struct ptlrpc_request *req;
1277         struct mds_body *body;
1278         struct lmv_obj *obj;
1279         int rc = 0, i;
1280         ENTRY;
1281
1282         rc = lmv_check_connect(obd);
1283         if (rc)
1284                 RETURN(rc);
1285
1286         obj = lmv_grab_obj(obd, &data->id1);
1287
1288         CDEBUG(D_OTHER, "SETATTR for "DLID4", valid 0x%x%s\n",
1289                OLID4(&data->id1), iattr->ia_valid, obj ? ", splitted" : "");
1290
1291         if (obj) {
1292                 for (i = 0; i < obj->objcount; i++) {
1293                         data->id1 = obj->objs[i].id;
1294
1295                         rc = md_setattr(lmv->tgts[id_group(&data->id1)].ltd_exp,
1296                                         data, iattr, ea, ealen, ea2, ea2len, &req);
1297
1298                         if (id_equal_fid(&obj->id, &obj->objs[i].id)) {
1299                                 /*
1300                                  * this is master object and this request should
1301                                  * be returned back to llite.
1302                                  */
1303                                 *request = req;
1304                         } else {
1305                                 ptlrpc_req_finished(req);
1306                         }
1307
1308                         if (rc)
1309                                 break;
1310                 }
1311                 lmv_put_obj(obj);
1312         } else {
1313                 LASSERT(id_group(&data->id1) < lmv->desc.ld_tgt_count);
1314                 rc = md_setattr(lmv->tgts[id_group(&data->id1)].ltd_exp,
1315                                 data, iattr, ea, ealen, ea2, ea2len, request);
1316                 if (rc == 0) {
1317                         body = lustre_msg_buf((*request)->rq_repmsg, 0,
1318                                               sizeof(*body));
1319                         LASSERT(body != NULL);
1320                         LASSERT((body->valid & OBD_MD_FID) != 0);
1321                         LASSERT(id_group(&body->id1) == id_group(&data->id1));
1322                 }
1323         }
1324         RETURN(rc);
1325 }
1326
1327 int lmv_sync(struct obd_export *exp, struct lustre_id *id,
1328              struct ptlrpc_request **request)
1329 {
1330         struct obd_device *obd = exp->exp_obd;
1331         struct lmv_obd *lmv = &obd->u.lmv;
1332         int rc;
1333         ENTRY;
1334
1335         rc = lmv_check_connect(obd);
1336         if (rc)
1337                 RETURN(rc);
1338
1339         rc = md_sync(lmv->tgts[id_group(id)].ltd_exp,
1340                      id, request);
1341         RETURN(rc);
1342 }
1343
1344 int lmv_dirobj_blocking_ast(struct ldlm_lock *lock,
1345                             struct ldlm_lock_desc *desc,
1346                             void *data, int flag)
1347 {
1348         struct lustre_handle lockh;
1349         struct lmv_obj *obj;
1350         int rc;
1351         ENTRY;
1352
1353         switch (flag) {
1354         case LDLM_CB_BLOCKING:
1355                 ldlm_lock2handle(lock, &lockh);
1356                 rc = ldlm_cli_cancel(&lockh);
1357                 if (rc < 0) {
1358                         CDEBUG(D_INODE, "ldlm_cli_cancel: %d\n", rc);
1359                         RETURN(rc);
1360                 }
1361                 break;
1362         case LDLM_CB_CANCELING:
1363                 /* time to drop cached attrs for dirobj */
1364                 obj = lock->l_ast_data;
1365                 if (obj) {
1366                         CDEBUG(D_OTHER, "cancel %s on "LPU64"/"LPU64
1367                                ", master "DLID4"\n",
1368                                lock->l_resource->lr_name.name[3] == 1 ?
1369                                "LOOKUP" : "UPDATE",
1370                                lock->l_resource->lr_name.name[0],
1371                                lock->l_resource->lr_name.name[1],
1372                                OLID4(&obj->id));
1373                         lmv_put_obj(obj);
1374                 }
1375                 break;
1376         default:
1377                 LBUG();
1378         }
1379         RETURN(0);
1380 }
1381
1382 void lmv_remove_dots(struct page *page)
1383 {
1384         unsigned limit = PAGE_CACHE_SIZE;
1385         char *kaddr = page_address(page);
1386         struct ext2_dir_entry_2 *p;
1387         unsigned offs, rec_len;
1388
1389         for (offs = 0; offs <= limit - EXT2_DIR_REC_LEN(1); offs += rec_len) {
1390                 p = (struct ext2_dir_entry_2 *)(kaddr + offs);
1391                 rec_len = le16_to_cpu(p->rec_len);
1392
1393                 if ((p->name_len == 1 && p->name[0] == '.') ||
1394                     (p->name_len == 2 && p->name[0] == '.' && p->name[1] == '.'))
1395                         p->inode = 0;
1396         }
1397 }
1398
1399 int lmv_readpage(struct obd_export *exp, struct lustre_id *id,
1400                  __u64 offset, struct page *page,
1401                  struct ptlrpc_request **request)
1402 {
1403         struct obd_device *obd = exp->exp_obd;
1404         struct lmv_obd *lmv = &obd->u.lmv;
1405         struct lustre_id rid = *id;
1406         struct lmv_obj *obj;
1407         int rc, i;
1408         ENTRY;
1409
1410 #warning "we need well-desgined readdir() implementation"
1411         rc = lmv_check_connect(obd);
1412         if (rc)
1413                 RETURN(rc);
1414
1415         LASSERT(id_group(id) < lmv->desc.ld_tgt_count);
1416         CDEBUG(D_OTHER, "READPAGE at %llu from "DLID4"\n",
1417                offset, OLID4(&rid));
1418
1419         obj = lmv_grab_obj(obd, id);
1420         if (obj) {
1421                 lmv_lock_obj(obj);
1422
1423                 /* find dirobj containing page with requested offset. */
1424                 for (i = 0; i < obj->objcount; i++) {
1425                         if (offset < obj->objs[i].size)
1426                                 break;
1427                         offset -= obj->objs[i].size;
1428                 }
1429                 rid = obj->objs[i].id;
1430
1431                 lmv_unlock_obj(obj);
1432                 lmv_put_obj(obj);
1433
1434                 CDEBUG(D_OTHER, "forward to "DLID4" with offset %lu\n",
1435                        OLID4(&rid), (unsigned long)offset);
1436         }
1437         rc = md_readpage(lmv->tgts[id_group(&rid)].ltd_exp, &rid,
1438                          offset, page, request);
1439
1440         if (rc == 0 && !id_equal_fid(&rid, id))
1441                 /* this page isn't from master object. To avoid "." and ".."
1442                  * duplication in directory, we have to remove them from all
1443                  * slave objects */
1444                 lmv_remove_dots(page);
1445
1446         RETURN(rc);
1447 }
1448
1449 int lmv_unlink_slaves(struct obd_export *exp, struct mdc_op_data *data,
1450                       struct ptlrpc_request **req)
1451 {
1452         struct obd_device *obd = exp->exp_obd;
1453         struct lmv_obd *lmv = &obd->u.lmv;
1454         struct mea *mea = data->mea1;
1455         struct mdc_op_data *data2;
1456         int i, rc = 0;
1457         ENTRY;
1458
1459         OBD_ALLOC(data2, sizeof(*data2));
1460         if (data2 == NULL)
1461                 RETURN(-ENOMEM);
1462
1463         LASSERT(mea != NULL);
1464         for (i = 0; i < mea->mea_count; i++) {
1465                 memset(data2, 0, sizeof(*data2));
1466                 data2->id1 = mea->mea_ids[i];
1467                 data2->create_mode = MDS_MODE_DONT_LOCK | S_IFDIR;
1468
1469                 if (lmv->tgts[id_group(&data2->id1)].ltd_exp == NULL)
1470                         continue;
1471
1472                 rc = md_unlink(lmv->tgts[id_group(&data2->id1)].ltd_exp,
1473                                data2, req);
1474
1475                 CDEBUG(D_OTHER, "unlink slave "DLID4" -> %d\n",
1476                        OLID4(&mea->mea_ids[i]), rc);
1477
1478                 if (*req) {
1479                         ptlrpc_req_finished(*req);
1480                         *req = NULL;
1481                 }
1482                 if (rc)
1483                         RETURN(rc);
1484         }
1485         OBD_FREE(data2, sizeof(*data2));
1486         RETURN(rc);
1487 }
1488
1489 int lmv_delete_inode(struct obd_export *exp, struct lustre_id *id)
1490 {
1491         ENTRY;
1492
1493         LASSERT(exp && id);
1494         if (lmv_delete_obj(exp, id)) {
1495                 CDEBUG(D_OTHER, "lmv object "DLID4" is destroyed.\n",
1496                        OLID4(id));
1497         }
1498         RETURN(0);
1499 }
1500
1501 int lmv_unlink(struct obd_export *exp, struct mdc_op_data *data,
1502                struct ptlrpc_request **request)
1503 {
1504         struct obd_device *obd = exp->exp_obd;
1505         struct lmv_obd *lmv = &obd->u.lmv;
1506         int rc, i = 0;
1507         ENTRY;
1508
1509         rc = lmv_check_connect(obd);
1510         if (rc)
1511                 RETURN(rc);
1512
1513         if (data->namelen == 0 && data->mea1 != NULL) {
1514                 /* mds asks to remove slave objects */
1515                 rc = lmv_unlink_slaves(exp, data, request);
1516                 RETURN(rc);
1517         }
1518
1519         if (data->namelen != 0) {
1520                 struct lmv_obj *obj;
1521
1522                 obj = lmv_grab_obj(obd, &data->id1);
1523                 if (obj) {
1524                         i = raw_name2idx(obj->hashtype, obj->objcount,
1525                                          data->name, data->namelen);
1526                         data->id1 = obj->objs[i].id;
1527                         lmv_put_obj(obj);
1528                 }
1529                 CDEBUG(D_OTHER, "unlink '%*s' in "DLID4" -> %u\n",
1530                        data->namelen, data->name, OLID4(&data->id1),
1531                        i);
1532         } else {
1533                 CDEBUG(D_OTHER, "drop i_nlink on "DLID4"\n",
1534                        OLID4(&data->id1));
1535         }
1536         rc = md_unlink(lmv->tgts[id_group(&data->id1)].ltd_exp,
1537                        data, request);
1538         RETURN(rc);
1539 }
1540
1541 struct obd_device *lmv_get_real_obd(struct obd_export *exp,
1542                                     struct lustre_id *id)
1543 {
1544         struct obd_device *obd = exp->exp_obd;
1545         struct lmv_obd *lmv = &obd->u.lmv;
1546         int rc;
1547         ENTRY;
1548
1549         rc = lmv_check_connect(obd);
1550         if (rc)
1551                 RETURN(ERR_PTR(rc));
1552         obd = lmv->tgts[id_group(id)].ltd_exp->exp_obd;
1553         EXIT;
1554
1555         return obd;
1556 }
1557
1558 int lmv_init_ea_size(struct obd_export *exp, int easize,
1559                      int cookiesize)
1560 {
1561         struct obd_device *obd = exp->exp_obd;
1562         struct lmv_obd *lmv = &obd->u.lmv;
1563         int i, rc = 0, change = 0;
1564         ENTRY;
1565
1566         if (lmv->max_easize < easize) {
1567                 lmv->max_easize = easize;
1568                 change = 1;
1569         }
1570         if (lmv->max_cookiesize < cookiesize) {
1571                 lmv->max_cookiesize = cookiesize;
1572                 change = 1;
1573         }
1574         if (change == 0)
1575                 RETURN(0);
1576
1577         if (lmv->connected == 0)
1578                 RETURN(0);
1579
1580         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
1581                 if (lmv->tgts[i].ltd_exp == NULL) {
1582                         CWARN("%s: NULL export for %d\n", obd->obd_name, i);
1583                         continue;
1584                 }
1585
1586                 rc = obd_init_ea_size(lmv->tgts[i].ltd_exp, easize, cookiesize);
1587                 if (rc) {
1588                         CERROR("obd_init_ea_size() failed on MDT target %d, "
1589                                "error %d.\n", i, rc);
1590                         break;
1591                 }
1592         }
1593         RETURN(rc);
1594 }
1595
1596 int lmv_obd_create_single(struct obd_export *exp, struct obdo *oa,
1597                           struct lov_stripe_md **ea, struct obd_trans_info *oti)
1598 {
1599         struct obd_device *obd = exp->exp_obd;
1600         struct lmv_obd *lmv = &obd->u.lmv;
1601         struct lov_stripe_md obj_md;
1602         struct lov_stripe_md *obj_mdp = &obj_md;
1603         int rc = 0;
1604         ENTRY;
1605
1606         LASSERT(ea == NULL);
1607         LASSERT(oa->o_mds < lmv->desc.ld_tgt_count);
1608
1609         rc = obd_create(lmv->tgts[oa->o_mds].ltd_exp,
1610                         oa, &obj_mdp, oti);
1611
1612         RETURN(rc);
1613 }
1614
1615 int lmv_getready(struct obd_export *exp)
1616 {
1617         struct obd_device *obd = exp->exp_obd;
1618         int rc = 0;
1619
1620         ENTRY;
1621         rc = lmv_check_connect(obd);
1622         RETURN(rc);
1623 }
1624
1625 /*
1626  * to be called from MDS only. @oa should have correct store cookie and o_fid
1627  * values for "master" object, as it will be used.
1628  */
1629 int lmv_obd_create(struct obd_export *exp, struct obdo *oa,
1630                    struct lov_stripe_md **ea, struct obd_trans_info *oti)
1631 {
1632         struct obd_device *obd = exp->exp_obd;
1633         struct lmv_obd *lmv = &obd->u.lmv;
1634         struct lustre_id mid;
1635         int i, c, rc = 0;
1636         struct mea *mea;
1637         ENTRY;
1638
1639         rc = lmv_check_connect(obd);
1640         if (rc)
1641                 RETURN(rc);
1642
1643         LASSERT(oa != NULL);
1644
1645         if (ea == NULL) {
1646                 rc = lmv_obd_create_single(exp, oa, NULL, oti);
1647                 if (rc)
1648                         CERROR("Can't create object, rc = %d\n", rc);
1649                 RETURN(rc);
1650         }
1651
1652         if (*ea == NULL) {
1653                 rc = obd_alloc_diskmd(exp, (struct lov_mds_md **)ea);
1654                 if (rc < 0) {
1655                         CERROR("obd_alloc_diskmd() failed, error %d\n",
1656                                rc);
1657                         RETURN(rc);
1658                 } else
1659                         rc = 0;
1660
1661                 if (*ea == NULL)
1662                         RETURN(-ENOMEM);
1663         }
1664
1665         /*
1666          * here we should take care about splitted dir, so store cookie and fid
1667          * for "master" object should already be allocated and passed in @oa.
1668          */
1669         LASSERT(oa->o_id != 0);
1670         LASSERT(oa->o_fid != 0);
1671
1672         /* save "master" object id */
1673         obdo2id(&mid, oa);
1674
1675         mea = (struct mea *)*ea;
1676         mea->mea_master = -1;
1677         mea->mea_magic = MEA_MAGIC_ALL_CHARS;
1678
1679         if (!mea->mea_count || mea->mea_count > lmv->desc.ld_tgt_count)
1680                 mea->mea_count = lmv->desc.ld_tgt_count;
1681
1682         for (i = 0, c = 0; c < mea->mea_count && i < lmv->desc.ld_tgt_count; i++) {
1683                 struct lov_stripe_md obj_md;
1684                 struct lov_stripe_md *obj_mdp = &obj_md;
1685
1686                 if (lmv->tgts[i].ltd_exp == NULL) {
1687                         /* this is "master" MDS */
1688                         mea->mea_master = i;
1689                         mea->mea_ids[c] = mid;
1690                         c++;
1691                         continue;
1692                 }
1693
1694                 /*
1695                  * "master" MDS should always be part of stripped dir,
1696                  * so scan for it.
1697                  */
1698                 if (mea->mea_master == -1 && c == mea->mea_count - 1)
1699                         continue;
1700
1701                 oa->o_valid = OBD_MD_FLGENER | OBD_MD_FLTYPE | OBD_MD_FLMODE |
1702                         OBD_MD_FLUID | OBD_MD_FLGID | OBD_MD_FLID;
1703
1704                 rc = obd_create(lmv->tgts[c].ltd_exp, oa, &obj_mdp, oti);
1705                 if (rc) {
1706                         CERROR("obd_create() failed on MDT target %d, "
1707                                "error %d\n", c, rc);
1708                         RETURN(rc);
1709                 }
1710
1711                 CDEBUG(D_OTHER, "dirobj at mds %d: "LPU64"/%u\n",
1712                        i, oa->o_id, oa->o_generation);
1713
1714
1715                 /*
1716                  * here, when object is created (or it is master and was passed
1717                  * from caller) on desired MDS we save its fid to local mea_ids.
1718                  */
1719                 LASSERT(oa->o_fid);
1720
1721                 /*
1722                  * store cookie should be defined here for both cases (master
1723                  * object and not master), because master is already created.
1724                  */
1725                 LASSERT(oa->o_id);
1726
1727                 /* fill mea by store cookie and fid */
1728                 obdo2id(&mea->mea_ids[c], oa);
1729                 c++;
1730         }
1731         LASSERT(c == mea->mea_count);
1732
1733         CDEBUG(D_OTHER, "%d dirobjects created\n",
1734                (int)mea->mea_count);
1735
1736         RETURN(rc);
1737 }
1738
1739 static int lmv_llog_init(struct obd_device *obd, struct obd_llogs *llogs,
1740                          struct obd_device *tgt, int count,
1741                          struct llog_catid *logid)
1742 {
1743         struct llog_ctxt *ctxt;
1744         int rc;
1745         ENTRY;
1746
1747         rc = obd_llog_setup(obd, llogs, LLOG_CONFIG_REPL_CTXT, tgt, 0, NULL,
1748                             &llog_client_ops);
1749         if (rc == 0) {
1750                 ctxt = llog_get_context(llogs, LLOG_CONFIG_REPL_CTXT);
1751                 ctxt->loc_imp = tgt->u.cli.cl_import;
1752         }
1753
1754         RETURN(rc);
1755 }
1756
1757 static int lmv_llog_finish(struct obd_device *obd,
1758                            struct obd_llogs *llogs, int count)
1759 {
1760         int rc;
1761         ENTRY;
1762
1763         rc = obd_llog_cleanup(llog_get_context(llogs, LLOG_CONFIG_REPL_CTXT));
1764         RETURN(rc);
1765 }
1766
1767 static int lmv_get_info(struct obd_export *exp, __u32 keylen,
1768                         void *key, __u32 *vallen, void *val)
1769 {
1770         struct obd_device *obd;
1771         struct lmv_obd *lmv;
1772         int rc = 0;
1773         ENTRY;
1774
1775         obd = class_exp2obd(exp);
1776         if (obd == NULL) {
1777                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
1778                        exp->exp_handle.h_cookie);
1779                 RETURN(-EINVAL);
1780         }
1781
1782         lmv = &obd->u.lmv;
1783         if (keylen == 6 && memcmp(key, "mdsize", 6) == 0) {
1784                 __u32 *mdsize = val;
1785                 *vallen = sizeof(__u32);
1786                 *mdsize = sizeof(struct lustre_id) * lmv->desc.ld_tgt_count
1787                         + sizeof(struct mea);
1788                 RETURN(0);
1789         } else if (keylen == 6 && memcmp(key, "mdsnum", 6) == 0) {
1790                 struct obd_uuid *cluuid = &lmv->cluuid;
1791                 struct lmv_tgt_desc *tgts;
1792                 __u32 *mdsnum = val;
1793                 int i;
1794
1795                 for (i = 0, tgts = lmv->tgts; i < lmv->desc.ld_tgt_count; i++, tgts++) {
1796                         if (obd_uuid_equals(&tgts->uuid, cluuid)) {
1797                                 *vallen = sizeof(__u32);
1798                                 *mdsnum = i;
1799                                 RETURN(0);
1800                         }
1801                 }
1802                 LASSERT(0);
1803         } else if (keylen == 6 && memcmp(key, "rootid", 6) == 0) {
1804                 /* getting rootid from first MDS. */
1805                 rc = obd_get_info(lmv->tgts[0].ltd_exp, keylen, key,
1806                                   vallen, val);
1807                 RETURN(rc);
1808         } else if (keylen >= strlen("lmvdesc") && strcmp(key, "lmvdesc") == 0) {
1809                 struct lmv_desc *desc_ret = val;
1810                 *desc_ret = lmv->desc;
1811                 RETURN(0);
1812         }
1813
1814         CDEBUG(D_IOCTL, "invalid key\n");
1815         RETURN(-EINVAL);
1816 }
1817
1818 int lmv_set_info(struct obd_export *exp, obd_count keylen,
1819                  void *key, obd_count vallen, void *val)
1820 {
1821         struct obd_device *obd;
1822         struct lmv_obd *lmv;
1823         ENTRY;
1824
1825         obd = class_exp2obd(exp);
1826         if (obd == NULL) {
1827                 CDEBUG(D_IOCTL, "invalid client cookie "LPX64"\n",
1828                        exp->exp_handle.h_cookie);
1829                 RETURN(-EINVAL);
1830         }
1831         lmv = &obd->u.lmv;
1832
1833         if (keylen >= strlen("inter_mds") && strcmp(key, "inter_mds") == 0) {
1834                 lmv->server_timeout = 1;
1835                 lmv_set_timeouts(obd);
1836                 RETURN(0);
1837         }
1838
1839         /* maybe this could be default */
1840         if ((keylen == strlen("sec") && strcmp(key, "sec") == 0) ||
1841             (keylen == strlen("nllu") && strcmp(key, "nllu") == 0)) {
1842                 struct lmv_tgt_desc *tgt;
1843                 struct obd_export *exp;
1844                 int rc = 0, err, i;
1845
1846                 spin_lock(&lmv->lmv_lock);
1847                 for (i = 0, tgt = lmv->tgts; i < lmv->desc.ld_tgt_count;
1848                      i++, tgt++) {
1849                         exp = tgt->ltd_exp;
1850                         /* during setup time the connections to mdc might
1851                          * haven't been established.
1852                          */
1853                         if (exp == NULL) {
1854                                 struct obd_device *tgt_obd;
1855
1856                                 tgt_obd = class_find_client_obd(&tgt->uuid,
1857                                                                 LUSTRE_MDC_NAME,
1858                                                                 &obd->obd_uuid);
1859                                 if (!tgt_obd) {
1860                                         CERROR("can't set info %s, "
1861                                                "device %s not attached?\n",
1862                                                 (char *) key, tgt->uuid.uuid);
1863                                         rc = -EINVAL;
1864                                         continue;
1865                                 }
1866                                 exp = tgt_obd->obd_self_export;
1867                         }
1868
1869                         err = obd_set_info(exp, keylen, key, vallen, val);
1870                         if (!rc)
1871                                 rc = err;
1872                 }
1873                 spin_unlock(&lmv->lmv_lock);
1874
1875                 RETURN(rc);
1876         }
1877
1878         RETURN(-EINVAL);
1879 }
1880
1881 int lmv_packmd(struct obd_export *exp, struct lov_mds_md **lmmp,
1882                struct lov_stripe_md *lsm)
1883 {
1884         struct obd_device *obd = class_exp2obd(exp);
1885         struct lmv_obd *lmv = &obd->u.lmv;
1886         struct mea *meap, *lsmp;
1887         int mea_size, i;
1888         ENTRY;
1889
1890         mea_size = (sizeof(struct lustre_id) *
1891                     lmv->desc.ld_tgt_count) + sizeof(struct mea);
1892         if (!lmmp)
1893                 RETURN(mea_size);
1894
1895         if (*lmmp && !lsm) {
1896                 OBD_FREE(*lmmp, mea_size);
1897                 *lmmp = NULL;
1898                 RETURN(0);
1899         }
1900
1901         if (*lmmp == NULL) {
1902                 OBD_ALLOC(*lmmp, mea_size);
1903                 if (*lmmp == NULL)
1904                         RETURN(-ENOMEM);
1905         }
1906
1907         if (!lsm)
1908                 RETURN(mea_size);
1909
1910         lsmp = (struct mea *)lsm;
1911         meap = (struct mea *)*lmmp;
1912
1913         if (lsmp->mea_magic != MEA_MAGIC_LAST_CHAR &&
1914             lsmp->mea_magic != MEA_MAGIC_ALL_CHARS)
1915                 RETURN(-EINVAL);
1916
1917         meap->mea_magic = cpu_to_le32(lsmp->mea_magic);
1918         meap->mea_count = cpu_to_le32(lsmp->mea_count);
1919         meap->mea_master = cpu_to_le32(lsmp->mea_master);
1920
1921         for (i = 0; i < lmv->desc.ld_tgt_count; i++) {
1922                 meap->mea_ids[i] = meap->mea_ids[i];
1923                 id_cpu_to_le(&meap->mea_ids[i]);
1924         }
1925
1926         RETURN(mea_size);
1927 }
1928
1929 int lmv_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
1930                  struct lov_mds_md *lmm, int lmm_size)
1931 {
1932         struct obd_device *obd = class_exp2obd(exp);
1933         struct mea **tmea = (struct mea **)lsmp;
1934         struct mea *mea = (struct mea *)lmm;
1935         struct lmv_obd *lmv = &obd->u.lmv;
1936         int mea_size, i, rc = 0;
1937         __u32 magic;
1938         ENTRY;
1939
1940         mea_size = sizeof(struct lustre_id) *
1941                 lmv->desc.ld_tgt_count + sizeof(struct mea);
1942
1943         if (lsmp == NULL)
1944                 return mea_size;
1945
1946         if (*lsmp != NULL && lmm == NULL) {
1947                 OBD_FREE(*tmea, mea_size);
1948                 RETURN(0);
1949         }
1950
1951         LASSERT(mea_size == lmm_size);
1952
1953         OBD_ALLOC(*tmea, mea_size);
1954         if (*tmea == NULL)
1955                 RETURN(-ENOMEM);
1956
1957         if (!lmm)
1958                 RETURN(mea_size);
1959
1960         if (mea->mea_magic == MEA_MAGIC_LAST_CHAR ||
1961             mea->mea_magic == MEA_MAGIC_ALL_CHARS)
1962         {
1963                 magic = le32_to_cpu(mea->mea_magic);
1964         } else {
1965                 struct mea_old *old = (struct mea_old *)lmm;
1966
1967                 mea_size = sizeof(struct lustre_id) * old->mea_count +
1968                         sizeof(struct mea_old);
1969
1970                 if (old->mea_count > 256 || old->mea_master > 256 ||
1971                     lmm_size < mea_size || old->mea_master > old->mea_count) {
1972                         CWARN("bad MEA: count %u, master %u, size %u\n",
1973                               old->mea_count, old->mea_master, mea_size);
1974                         GOTO(out_free_mea, rc = -EINVAL);
1975                 }
1976                 magic = MEA_MAGIC_LAST_CHAR;
1977         }
1978
1979         (*tmea)->mea_magic = magic;
1980         (*tmea)->mea_count = le32_to_cpu(mea->mea_count);
1981         (*tmea)->mea_master = le32_to_cpu(mea->mea_master);
1982
1983         for (i = 0; i < (*tmea)->mea_count; i++) {
1984                 (*tmea)->mea_ids[i] = mea->mea_ids[i];
1985                 id_le_to_cpu(&(*tmea)->mea_ids[i]);
1986         }
1987         RETURN(mea_size);
1988
1989 out_free_mea:
1990         OBD_FREE(*tmea, mea_size);
1991         return rc;
1992 }
1993
1994 int lmv_brw(int rw, struct obd_export *exp, struct obdo *oa,
1995             struct lov_stripe_md *ea, obd_count oa_bufs,
1996             struct brw_page *pgarr, struct obd_trans_info *oti)
1997 {
1998         struct obd_device *obd = exp->exp_obd;
1999         struct lmv_obd *lmv = &obd->u.lmv;
2000         struct mea *mea = (struct mea *) ea;
2001         int err;
2002
2003         LASSERT(oa != NULL);
2004         LASSERT(ea != NULL);
2005         LASSERT(pgarr != NULL);
2006         LASSERT(oa->o_mds < lmv->desc.ld_tgt_count);
2007
2008         oa->o_gr = id_gen(&mea->mea_ids[oa->o_mds]);
2009         oa->o_id = id_ino(&mea->mea_ids[oa->o_mds]);
2010         oa->o_valid = OBD_MD_FLID | OBD_MD_FLGROUP;
2011
2012         err = obd_brw(rw, lmv->tgts[oa->o_mds].ltd_exp,
2013                       oa, NULL, oa_bufs, pgarr, oti);
2014         RETURN(err);
2015 }
2016
2017 struct obd_ops lmv_obd_ops = {
2018         .o_owner                = THIS_MODULE,
2019         .o_attach               = lmv_attach,
2020         .o_detach               = lmv_detach,
2021         .o_setup                = lmv_setup,
2022         .o_cleanup              = lmv_cleanup,
2023         .o_connect              = lmv_connect,
2024         .o_disconnect           = lmv_disconnect,
2025         .o_statfs               = lmv_statfs,
2026         .o_llog_init            = lmv_llog_init,
2027         .o_llog_finish          = lmv_llog_finish,
2028         .o_get_info             = lmv_get_info,
2029         .o_set_info             = lmv_set_info,
2030         .o_create               = lmv_obd_create,
2031         .o_packmd               = lmv_packmd,
2032         .o_unpackmd             = lmv_unpackmd,
2033         .o_brw                  = lmv_brw,
2034         .o_init_ea_size         = lmv_init_ea_size,
2035         .o_notify               = lmv_notify,
2036         .o_iocontrol            = lmv_iocontrol,
2037         .o_getready             = lmv_getready,
2038 };
2039
2040 struct md_ops lmv_md_ops = {
2041         .m_getstatus           = lmv_getstatus,
2042         .m_getattr             = lmv_getattr,
2043         .m_change_cbdata       = lmv_change_cbdata,
2044         .m_change_cbdata_name  = lmv_change_cbdata_name,
2045         .m_close               = lmv_close,
2046         .m_create              = lmv_create,
2047         .m_done_writing        = lmv_done_writing,
2048         .m_enqueue             = lmv_enqueue,
2049         .m_getattr_lock        = lmv_getattr_lock,
2050         .m_intent_lock         = lmv_intent_lock,
2051         .m_link                = lmv_link,
2052         .m_rename              = lmv_rename,
2053         .m_setattr             = lmv_setattr,
2054         .m_sync                = lmv_sync,
2055         .m_readpage            = lmv_readpage,
2056         .m_unlink              = lmv_unlink,
2057         .m_get_real_obd        = lmv_get_real_obd,
2058         .m_valid_attrs         = lmv_valid_attrs,
2059         .m_delete_inode        = lmv_delete_inode,
2060 };
2061
2062 int __init lmv_init(void)
2063 {
2064         struct lprocfs_static_vars lvars;
2065         int rc;
2066
2067         obj_cache = kmem_cache_create("lmv_objects",
2068                                       sizeof(struct lmv_obj),
2069                                       0, 0, NULL, NULL);
2070         if (!obj_cache) {
2071                 CERROR("error allocating lmv objects cache\n");
2072                 return -ENOMEM;
2073         }
2074
2075         lprocfs_init_vars(lmv, &lvars);
2076         rc = class_register_type(&lmv_obd_ops, &lmv_md_ops,
2077                                  lvars.module_vars,
2078                                  OBD_LMV_DEVICENAME);
2079         if (rc)
2080                 kmem_cache_destroy(obj_cache);
2081
2082         return rc;
2083 }
2084
2085 #ifdef __KERNEL__
2086 static void lmv_exit(void)
2087 {
2088         class_unregister_type(OBD_LMV_DEVICENAME);
2089
2090         LASSERTF(kmem_cache_destroy(obj_cache) == 0,
2091                  "can't free lmv objects cache, %d object(s)"
2092                  "still in use\n", atomic_read(&obj_cache_count));
2093 }
2094
2095 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
2096 MODULE_DESCRIPTION("Lustre Logical Metadata Volume OBD driver");
2097 MODULE_LICENSE("GPL");
2098
2099 module_init(lmv_init);
2100 module_exit(lmv_exit);
2101 #endif