lustre/lmv/lmv_intent.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #ifndef EXPORT_SYMTAB
  38 # define EXPORT_SYMTAB
  39 #endif
  40 #define DEBUG_SUBSYSTEM S_LMV
  41 #ifdef __KERNEL__
  42 #include <linux/slab.h>
  43 #include <linux/module.h>
  44 #include <linux/init.h>
  45 #include <linux/slab.h>
  46 #include <linux/pagemap.h>
  47 #include <asm/div64.h>
  48 #include <linux/seq_file.h>
  49 #include <linux/namei.h>
  50 #include <linux/lustre_intent.h>
  51 #else
  52 #include <liblustre.h>
  53 #endif
  54
  55 #include <obd_support.h>
  56 #include <lustre/lustre_idl.h>
  57 #include <lustre_lib.h>
  58 #include <lustre_net.h>
  59 #include <lustre_dlm.h>
  60 #include <obd_class.h>
  61 #include <lprocfs_status.h>
  62 #include "lmv_internal.h"
  63
  64 int lmv_intent_remote(struct obd_export *exp, void *lmm,
  65                       int lmmsize, struct lookup_intent *it,
  66                       int flags, struct ptlrpc_request **reqp,
  67                       ldlm_blocking_callback cb_blocking,
  68                       int extra_lock_flags)
  69 {
  70         struct obd_device      *obd = exp->exp_obd;
  71         struct lmv_obd         *lmv = &obd->u.lmv;
  72         struct ptlrpc_request  *req = NULL;
  73         struct lustre_handle    plock;
  74         struct md_op_data      *op_data;
  75         struct lmv_tgt_desc    *tgt;
  76         struct mdt_body        *body;
  77         int                     pmode;
  78         int                     rc = 0;
  79         ENTRY;
  80
  81         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
  82         if (body == NULL)
  83                 RETURN(-EPROTO);
  84
  85         /*
  86          * Not cross-ref case, just get out of here.
  87          */
  88         if (!(body->valid & OBD_MD_MDS))
  89                 RETURN(0);
  90
  91         /*
  92          * Unfortunately, we have to lie to MDC/MDS to retrieve
  93          * attributes llite needs and provideproper locking.
  94          */
  95         if (it->it_op & IT_LOOKUP)
  96                 it->it_op = IT_GETATTR;
  97
  98         /*
  99          * We got LOOKUP lock, but we really need attrs.
 100          */
 101         pmode = it->d.lustre.it_lock_mode;
 102         if (pmode) {
 103                 plock.cookie = it->d.lustre.it_lock_handle;
 104                 it->d.lustre.it_lock_mode = 0;
 105                 it->d.lustre.it_data = NULL;
 106         }
 107
 108         LASSERT(fid_is_sane(&body->fid1));
 109
 110         tgt = lmv_find_target(lmv, &body->fid1);
 111         if (IS_ERR(tgt))
 112                 GOTO(out, rc = PTR_ERR(tgt));
 113
 114         OBD_ALLOC_PTR(op_data);
 115         if (op_data == NULL)
 116                 GOTO(out, rc = -ENOMEM);
 117
 118         op_data->op_fid1 = body->fid1;
 119         op_data->op_bias = MDS_CROSS_REF;
 120
 121         CDEBUG(D_INODE,
 122                "REMOTE_INTENT with fid="DFID" -> mds #%d\n",
 123                PFID(&body->fid1), tgt->ltd_idx);
 124
 125         it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
 126         rc = md_intent_lock(tgt->ltd_exp, op_data, lmm, lmmsize, it,
 127                             flags, &req, cb_blocking, extra_lock_flags);
 128         if (rc)
 129                 GOTO(out_free_op_data, rc);
 130
 131         /*
 132          * LLite needs LOOKUP lock to track dentry revocation in order to
 133          * maintain dcache consistency. Thus drop UPDATE lock here and put
 134          * LOOKUP in request.
 135          */
 136         if (it->d.lustre.it_lock_mode != 0) {
 137                 ldlm_lock_decref((void *)&it->d.lustre.it_lock_handle,
 138                                  it->d.lustre.it_lock_mode);
 139                 it->d.lustre.it_lock_mode = 0;
 140         }
 141         it->d.lustre.it_lock_handle = plock.cookie;
 142         it->d.lustre.it_lock_mode = pmode;
 143
 144         EXIT;
 145 out_free_op_data:
 146         OBD_FREE_PTR(op_data);
 147 out:
 148         if (rc && pmode)
 149                 ldlm_lock_decref(&plock, pmode);
 150
 151         ptlrpc_req_finished(*reqp);
 152         *reqp = req;
 153         return rc;
 154 }
 155
 156 /*
 157  * IT_OPEN is intended to open (and create, possible) an object. Parent (pid)
 158  * may be split dir.
 159  */
 160 int lmv_intent_open(struct obd_export *exp, struct md_op_data *op_data,
 161                     void *lmm, int lmmsize, struct lookup_intent *it,
 162                     int flags, struct ptlrpc_request **reqp,
 163                     ldlm_blocking_callback cb_blocking,
 164                     int extra_lock_flags)
 165 {
 166         struct obd_device     *obd = exp->exp_obd;
 167         struct lu_fid          rpid = op_data->op_fid1;
 168         struct lmv_obd        *lmv = &obd->u.lmv;
 169         struct md_op_data     *sop_data;
 170         struct lmv_stripe_md  *mea;
 171         struct lmv_tgt_desc   *tgt;
 172         struct mdt_body       *body;
 173         struct lmv_object     *obj;
 174         int                    rc;
 175         int                    loop = 0;
 176         int                    sidx;
 177         ENTRY;
 178
 179         OBD_ALLOC_PTR(sop_data);
 180         if (sop_data == NULL)
 181                 RETURN(-ENOMEM);
 182
 183         /* save op_data fro repeat case */
 184         *sop_data = *op_data;
 185
 186 repeat:
 187
 188         ++loop;
 189         LASSERT(loop <= 2);
 190         obj = lmv_object_find(obd, &rpid);
 191         if (obj) {
 192                 /*
 193                  * Directory is already split, so we have to forward request to
 194                  * the right MDS.
 195                  */
 196                 sidx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
 197                                        (char *)op_data->op_name,
 198                                        op_data->op_namelen);
 199
 200                 rpid = obj->lo_stripes[sidx].ls_fid;
 201
 202                 sop_data->op_mds = obj->lo_stripes[sidx].ls_mds;
 203                 tgt = lmv_get_target(lmv, sop_data->op_mds);
 204                 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
 205                 lmv_object_put(obj);
 206
 207                 CDEBUG(D_INODE,
 208                        "Choose slave dir ("DFID") -> mds #%d\n",
 209                        PFID(&rpid), tgt->ltd_idx);
 210         } else {
 211                 sop_data->op_bias |= MDS_CHECK_SPLIT;
 212                 tgt = lmv_find_target(lmv, &rpid);
 213                 sop_data->op_mds = tgt->ltd_idx;
 214         }
 215         if (IS_ERR(tgt))
 216                 GOTO(out_free_sop_data, rc = PTR_ERR(tgt));
 217
 218         sop_data->op_fid1 = rpid;
 219
 220         if (it->it_op & IT_CREAT) {
 221                 /*
 222                  * For open with IT_CREATE and for IT_CREATE cases allocate new
 223                  * fid and setup FLD for it.
 224                  */
 225                 sop_data->op_fid3 = sop_data->op_fid2;
 226                 rc = lmv_fid_alloc(exp, &sop_data->op_fid2, sop_data);
 227                 if (rc)
 228                         GOTO(out_free_sop_data, rc);
 229
 230                 if (rc == -ERESTART)
 231                         goto repeat;
 232                 else if (rc)
 233                         GOTO(out_free_sop_data, rc);
 234         }
 235
 236         CDEBUG(D_INODE,
 237                "OPEN_INTENT with fid1="DFID", fid2="DFID", name='%s' -> mds #%d\n",
 238                PFID(&sop_data->op_fid1), PFID(&sop_data->op_fid2),
 239                sop_data->op_name, tgt->ltd_idx);
 240
 241         rc = md_intent_lock(tgt->ltd_exp, sop_data, lmm, lmmsize, it, flags,
 242                             reqp, cb_blocking, extra_lock_flags);
 243
 244         if (rc == -ERESTART) {
 245                 LASSERT(*reqp != NULL);
 246                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *reqp,
 247                           "Got -ERESTART during open!\n");
 248                 ptlrpc_req_finished(*reqp);
 249                 *reqp = NULL;
 250                 it->d.lustre.it_data = NULL;
 251
 252                 /*
 253                  * Directory got split. Time to update local object and repeat
 254                  * the request with proper MDS.
 255                  */
 256                 LASSERT(lu_fid_eq(&op_data->op_fid1, &rpid));
 257                 rc = lmv_handle_split(exp, &rpid);
 258                 if (rc == 0) {
 259                         /* We should reallocate child FID. */
 260                         rc = lmv_allocate_slaves(obd, &rpid, op_data,
 261                                                  &sop_data->op_fid2);
 262                         if (rc == 0)
 263                                 goto repeat;
 264                 }
 265         }
 266
 267         if (rc != 0)
 268                 GOTO(out_free_sop_data, rc);
 269
 270         /*
 271          * Nothing is found, do not access body->fid1 as it is zero and thus
 272          * pointless.
 273          */
 274         if ((it->d.lustre.it_disposition & DISP_LOOKUP_NEG) &&
 275             !(it->d.lustre.it_disposition & DISP_OPEN_CREATE) &&
 276             !(it->d.lustre.it_disposition & DISP_OPEN_OPEN))
 277                 GOTO(out_free_sop_data, rc = 0);
 278
 279         /*
 280          * Okay, MDS has returned success. Probably name has been resolved in
 281          * remote inode.
 282          */
 283         rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
 284                                cb_blocking, extra_lock_flags);
 285         if (rc != 0) {
 286                 LASSERT(rc < 0);
 287                 /*
 288                  * This is possible, that some userspace application will try to
 289                  * open file as directory and we will have -ENOTDIR here. As
 290                  * this is normal situation, we should not print error here,
 291                  * only debug info.
 292                  */
 293                 CDEBUG(D_INODE, "Can't handle remote %s: dir "DFID"("DFID"):"
 294                        "%*s: %d\n", LL_IT2STR(it), PFID(&op_data->op_fid2),
 295                        PFID(&rpid), op_data->op_namelen, op_data->op_name, rc);
 296                 GOTO(out_free_sop_data, rc);
 297         }
 298
 299         /*
 300          * Caller may use attrs MDS returns on IT_OPEN lock request so, we have
 301          * to update them for split dir.
 302          */
 303         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
 304         LASSERT(body != NULL);
 305
 306         /*
 307          * Could not find object, FID is not present in response.
 308          */
 309         if (!(body->valid & OBD_MD_FLID))
 310                 GOTO(out_free_sop_data, rc = 0);
 311
 312         obj = lmv_object_find(obd, &body->fid1);
 313         if (obj == NULL) {
 314                 /*
 315                  * XXX: Capability for remote call!
 316                  */
 317                 mea = lmv_get_mea(*reqp);
 318                 if (mea != NULL) {
 319                         obj = lmv_object_create(exp, &body->fid1, mea);
 320                         if (IS_ERR(obj))
 321                                 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
 322                 }
 323         }
 324
 325         if (obj) {
 326                 /*
 327                  * This is split dir and we'd want to get attrs.
 328                  */
 329                 CDEBUG(D_INODE, "Slave attributes for "DFID"\n",
 330                        PFID(&body->fid1));
 331
 332                 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
 333                                            cb_blocking, extra_lock_flags);
 334                 lmv_object_put(obj);
 335         }
 336         EXIT;
 337 out_free_sop_data:
 338         OBD_FREE_PTR(sop_data);
 339         return rc;
 340 }
 341
 342 /*
 343  * Handler for: getattr, lookup and revalidate cases.
 344  */
 345 int lmv_intent_lookup(struct obd_export *exp, struct md_op_data *op_data,
 346                       void *lmm, int lmmsize, struct lookup_intent *it,
 347                       int flags, struct ptlrpc_request **reqp,
 348                       ldlm_blocking_callback cb_blocking,
 349                       int extra_lock_flags)
 350 {
 351         struct obd_device      *obd = exp->exp_obd;
 352         struct lu_fid           rpid = op_data->op_fid1;
 353         struct lmv_obd         *lmv = &obd->u.lmv;
 354         struct lmv_object      *obj = NULL;
 355         struct md_op_data      *sop_data;
 356         struct lmv_stripe_md   *mea;
 357         struct lmv_tgt_desc    *tgt = NULL;
 358         struct mdt_body        *body;
 359         int                     sidx;
 360         int                     loop = 0;
 361         int                     rc = 0;
 362         ENTRY;
 363
 364         OBD_ALLOC_PTR(sop_data);
 365         if (sop_data == NULL)
 366                 RETURN(-ENOMEM);
 367
 368         *sop_data = *op_data;
 369
 370 repeat:
 371         ++loop;
 372         LASSERT(loop <= 2);
 373
 374         obj = lmv_object_find(obd, &op_data->op_fid1);
 375         if (obj && op_data->op_namelen) {
 376                 sidx = raw_name2idx(obj->lo_hashtype,
 377                                        obj->lo_objcount,
 378                                        (char *)op_data->op_name,
 379                                        op_data->op_namelen);
 380                 rpid = obj->lo_stripes[sidx].ls_fid;
 381                 tgt = lmv_get_target(lmv,
 382                                      obj->lo_stripes[sidx].ls_mds);
 383                 CDEBUG(D_INODE,
 384                        "Choose slave dir ("DFID") -> mds #%d\n",
 385                        PFID(&rpid), tgt->ltd_idx);
 386                 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
 387         } else {
 388                 tgt = lmv_find_target(lmv, &op_data->op_fid1);
 389                 sop_data->op_bias |= MDS_CHECK_SPLIT;
 390         }
 391         if (obj)
 392                 lmv_object_put(obj);
 393
 394         if (IS_ERR(tgt))
 395                 GOTO(out_free_sop_data, rc = PTR_ERR(tgt));
 396
 397         if (!fid_is_sane(&sop_data->op_fid2))
 398                 fid_zero(&sop_data->op_fid2);
 399
 400         CDEBUG(D_INODE,
 401                "LOOKUP_INTENT with fid1="DFID", fid2="DFID
 402                ", name='%s' -> mds #%d\n",
 403                PFID(&sop_data->op_fid1), PFID(&sop_data->op_fid2),
 404                sop_data->op_name ? sop_data->op_name : "<NULL>",
 405                tgt->ltd_idx);
 406
 407         sop_data->op_bias &= ~MDS_CROSS_REF;
 408         sop_data->op_fid1 = rpid;
 409
 410         rc = md_intent_lock(tgt->ltd_exp, sop_data, lmm, lmmsize, it,
 411                             flags, reqp, cb_blocking, extra_lock_flags);
 412
 413         if (rc == -ERESTART) {
 414                 LASSERT(*reqp != NULL);
 415                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *reqp,
 416                           "Got -ERESTART during lookup!\n");
 417                 ptlrpc_req_finished(*reqp);
 418                 *reqp = NULL;
 419                 it->d.lustre.it_data = 0;
 420
 421                 /*
 422                  * Directory got split since last update. This shouldn't be
 423                  * because splitting causes lock revocation, so revalidate had
 424                  * to fail and lookup on dir had to return mea.
 425                  */
 426                 LASSERT(obj == NULL);
 427
 428                 obj = lmv_object_create(exp, &rpid, NULL);
 429                 if (IS_ERR(obj))
 430                         GOTO(out_free_sop_data, rc = PTR_ERR(obj));
 431                 lmv_object_put(obj);
 432                 goto repeat;
 433         }
 434
 435         if (rc < 0)
 436                 GOTO(out_free_sop_data, rc);
 437
 438         if (obj && rc > 0) {
 439                 /*
 440                  * This is split dir. In order to optimize things a bit, we
 441                  * consider obj valid updating missing parts.
 442                  */
 443                 CDEBUG(D_INODE,
 444                        "Revalidate slaves for "DFID", rc %d\n",
 445                        PFID(&op_data->op_fid1), rc);
 446
 447                 LASSERT(fid_is_sane(&op_data->op_fid2));
 448                 rc = lmv_revalidate_slaves(exp, reqp, &op_data->op_fid1, it, rc,
 449                                            cb_blocking, extra_lock_flags);
 450                 GOTO(out_free_sop_data, rc);
 451         }
 452
 453         if (*reqp == NULL)
 454                 GOTO(out_free_sop_data, rc);
 455
 456         /*
 457          * MDS has returned success. Probably name has been resolved in
 458          * remote inode. Let's check this.
 459          */
 460         rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags,
 461                                reqp, cb_blocking, extra_lock_flags);
 462         if (rc < 0)
 463                 GOTO(out_free_sop_data, rc);
 464
 465         /*
 466          * Nothing is found, do not access body->fid1 as it is zero and thus
 467          * pointless.
 468          */
 469         if (it->d.lustre.it_disposition & DISP_LOOKUP_NEG)
 470                 GOTO(out_free_sop_data, rc = 0);
 471
 472         LASSERT(*reqp != NULL);
 473         LASSERT((*reqp)->rq_repmsg != NULL);
 474         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
 475         LASSERT(body != NULL);
 476
 477         /*
 478          * Could not find object, FID is not present in response.
 479          */
 480         if (!(body->valid & OBD_MD_FLID))
 481                 GOTO(out_free_sop_data, rc = 0);
 482
 483         obj = lmv_object_find(obd, &body->fid1);
 484         if (obj == NULL) {
 485                 /*
 486                  * XXX: Remote capability is not handled.
 487                  */
 488                 mea = lmv_get_mea(*reqp);
 489                 if (mea != NULL) {
 490                         obj = lmv_object_create(exp, &body->fid1, mea);
 491                         if (IS_ERR(obj))
 492                                 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
 493                 }
 494         } else {
 495                 CDEBUG(D_INODE, "Slave attributes for "DFID", rc %d\n",
 496                        PFID(&body->fid1), rc);
 497
 498                 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
 499                                            cb_blocking, extra_lock_flags);
 500                 lmv_object_put(obj);
 501         }
 502
 503         EXIT;
 504 out_free_sop_data:
 505         OBD_FREE_PTR(sop_data);
 506         return rc;
 507 }
 508
 509 int lmv_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
 510                     void *lmm, int lmmsize, struct lookup_intent *it,
 511                     int flags, struct ptlrpc_request **reqp,
 512                     ldlm_blocking_callback cb_blocking,
 513                     int extra_lock_flags)
 514 {
 515         struct obd_device *obd = exp->exp_obd;
 516         int                rc;
 517         ENTRY;
 518
 519         LASSERT(it != NULL);
 520         LASSERT(fid_is_sane(&op_data->op_fid1));
 521
 522         CDEBUG(D_INODE, "INTENT LOCK '%s' for '%*s' on "DFID"\n",
 523                LL_IT2STR(it), op_data->op_namelen, op_data->op_name,
 524                PFID(&op_data->op_fid1));
 525
 526         rc = lmv_check_connect(obd);
 527         if (rc)
 528                 RETURN(rc);
 529
 530         if (it->it_op & (IT_LOOKUP | IT_GETATTR | IT_LAYOUT))
 531                 rc = lmv_intent_lookup(exp, op_data, lmm, lmmsize, it,
 532                                        flags, reqp, cb_blocking,
 533                                        extra_lock_flags);
 534         else if (it->it_op & IT_OPEN)
 535                 rc = lmv_intent_open(exp, op_data, lmm, lmmsize, it,
 536                                      flags, reqp, cb_blocking,
 537                                      extra_lock_flags);
 538         else
 539                 LBUG();
 540         RETURN(rc);
 541 }
 542
 543 int lmv_revalidate_slaves(struct obd_export *exp, struct ptlrpc_request **reqp,
 544                           const struct lu_fid *mid, struct lookup_intent *oit,
 545                           int master_valid, ldlm_blocking_callback cb_blocking,
 546                           int extra_lock_flags)
 547 {
 548         struct obd_device      *obd = exp->exp_obd;
 549         struct lmv_obd         *lmv = &obd->u.lmv;
 550         int                     master_lockm = 0;
 551         struct lustre_handle   *lockh = NULL;
 552         struct ptlrpc_request  *mreq = *reqp;
 553         struct lustre_handle    master_lockh = { 0 };
 554         struct md_op_data      *op_data;
 555         struct ldlm_lock       *lock;
 556         unsigned long           size = 0;
 557         struct mdt_body        *body;
 558         struct lmv_object      *obj;
 559         int                     i;
 560         int                     rc = 0;
 561         struct lu_fid           fid;
 562         struct ptlrpc_request  *req;
 563         ldlm_blocking_callback  cb;
 564         struct lookup_intent    it;
 565         struct lmv_tgt_desc    *tgt;
 566         int                     master;
 567         ENTRY;
 568
 569         CDEBUG(D_INODE, "Revalidate master obj "DFID"\n", PFID(mid));
 570
 571         OBD_ALLOC_PTR(op_data);
 572         if (op_data == NULL)
 573                 RETURN(-ENOMEM);
 574
 575         /*
 576          * We have to loop over the subobjects, check validity and update them
 577          * from MDS if needed. It's very useful that we need not to update all
 578          * the fields. Say, common fields (that are equal on all the subojects
 579          * need not to be update, another fields (i_size, for example) are
 580          * cached all the time.
 581          */
 582         obj = lmv_object_find_lock(obd, mid);
 583         if (obj == NULL)
 584                 RETURN(-EALREADY);
 585
 586         for (i = 0; i < obj->lo_objcount; i++) {
 587                 fid = obj->lo_stripes[i].ls_fid;
 588                 master = lu_fid_eq(&fid, &obj->lo_fid);
 589                 cb = master ? cb_blocking : lmv_blocking_ast;
 590
 591                 /*
 592                  * We need i_size and we would like to check possible cached locks,
 593                  * so this is is IT_GETATTR intent.
 594                  */
 595                 memset(&it, 0, sizeof(it));
 596                 it.it_op = IT_GETATTR;
 597
 598                 if (master && master_valid) {
 599                         /*
 600                          * lmv_intent_lookup() already checked
 601                          * validness and took the lock.
 602                          */
 603                         if (mreq != NULL) {
 604                                 body = req_capsule_server_get(&mreq->rq_pill,
 605                                                               &RMF_MDT_BODY);
 606                                 LASSERT(body != NULL);
 607                                 goto update;
 608                         }
 609                         /*
 610                          * Take already cached attrs into account.
 611                          */
 612                         CDEBUG(D_INODE,
 613                                "Master "DFID"is locked and cached\n",
 614                                PFID(mid));
 615                         goto release_lock;
 616                 }
 617
 618                 /*
 619                  * Prepare op_data for revalidating. Note that @fid2 shuld be
 620                  * defined otherwise it will go to server and take new lock
 621                  * which is what we reall not need here.
 622                  */
 623                 memset(op_data, 0, sizeof(*op_data));
 624                 op_data->op_bias = MDS_CROSS_REF;
 625                 op_data->op_fid1 = fid;
 626                 op_data->op_fid2 = fid;
 627                 req = NULL;
 628
 629                 tgt = lmv_get_target(lmv, obj->lo_stripes[i].ls_mds);
 630                 if (IS_ERR(tgt))
 631                         GOTO(cleanup, rc = PTR_ERR(tgt));
 632
 633                 CDEBUG(D_INODE, "Revalidate slave obj "DFID" -> mds #%d\n",
 634                        PFID(&fid), tgt->ltd_idx);
 635
 636                 rc = md_intent_lock(tgt->ltd_exp, op_data, NULL, 0, &it, 0,
 637                                     &req, cb, extra_lock_flags);
 638
 639                 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
 640                 if (rc > 0 && req == NULL) {
 641                         /*
 642                          * Nice, this slave is valid.
 643                          */
 644                         CDEBUG(D_INODE, "Cached slave "DFID"\n", PFID(&fid));
 645                         goto release_lock;
 646                 }
 647
 648                 if (rc < 0)
 649                         GOTO(cleanup, rc);
 650
 651                 if (master) {
 652                         /*
 653                          * Save lock on master to be returned to the caller.
 654                          */
 655                         CDEBUG(D_INODE, "No lock on master "DFID" yet\n",
 656                                PFID(mid));
 657                         memcpy(&master_lockh, lockh, sizeof(master_lockh));
 658                         master_lockm = it.d.lustre.it_lock_mode;
 659                         it.d.lustre.it_lock_mode = 0;
 660                 } else {
 661                         /*
 662                          * This is slave. We want to control it.
 663                          */
 664                         lock = ldlm_handle2lock(lockh);
 665                         LASSERT(lock != NULL);
 666                         lock->l_ast_data = lmv_object_get(obj);
 667                         LDLM_LOCK_PUT(lock);
 668                 }
 669
 670                 if (*reqp == NULL) {
 671                         /*
 672                          * This is first reply, we'll use it to return updated
 673                          * data back to the caller.
 674                          */
 675                         LASSERT(req != NULL);
 676                         ptlrpc_request_addref(req);
 677                         *reqp = req;
 678                 }
 679
 680                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 681                 LASSERT(body != NULL);
 682
 683 update:
 684                 obj->lo_stripes[i].ls_size = body->size;
 685
 686                 CDEBUG(D_INODE, "Fresh size %lu from "DFID"\n",
 687                        (unsigned long)obj->lo_stripes[i].ls_size, PFID(&fid));
 688
 689                 if (req)
 690                         ptlrpc_req_finished(req);
 691 release_lock:
 692                 size += obj->lo_stripes[i].ls_size;
 693
 694                 if (it.d.lustre.it_lock_mode && lockh) {
 695                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
 696                         it.d.lustre.it_lock_mode = 0;
 697                 }
 698         }
 699
 700         if (*reqp) {
 701                 /*
 702                  * Some attrs got refreshed, we have reply and it's time to put
 703                  * fresh attrs to it.
 704                  */
 705                 CDEBUG(D_INODE, "Return refreshed attrs: size = %lu for "DFID"\n",
 706                        (unsigned long)size, PFID(mid));
 707
 708                 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
 709                 LASSERT(body != NULL);
 710                 body->size = size;
 711
 712                 if (mreq == NULL) {
 713                         /*
 714                          * Very important to maintain mds num the same because
 715                          * of revalidation. mreq == NULL means that caller has
 716                          * no reply and the only attr we can return is size.
 717                          */
 718                         body->valid = OBD_MD_FLSIZE;
 719                 }
 720                 if (master_valid == 0) {
 721                         oit->d.lustre.it_lock_handle = master_lockh.cookie;
 722                         oit->d.lustre.it_lock_mode = master_lockm;
 723                 }
 724                 rc = 0;
 725         } else {
 726                 /*
 727                  * It seems all the attrs are fresh and we did no request.
 728                  */
 729                 CDEBUG(D_INODE, "All the attrs were fresh on "DFID"\n",
 730                        PFID(mid));
 731                 if (master_valid == 0)
 732                         oit->d.lustre.it_lock_mode = master_lockm;
 733                 rc = 1;
 734         }
 735
 736         EXIT;
 737 cleanup:
 738         OBD_FREE_PTR(op_data);
 739         lmv_object_put_unlock(obj);
 740         return rc;
 741 }
 742
 743 int lmv_allocate_slaves(struct obd_device *obd, struct lu_fid *pid,
 744                         struct md_op_data *op, struct lu_fid *fid)
 745 {
 746         struct lmv_obd          *lmv = &obd->u.lmv;
 747         struct lmv_object       *obj;
 748         mdsno_t                  mds;
 749         int                      sidx;
 750         int                      rc;
 751         ENTRY;
 752
 753         obj = lmv_object_find(obd, pid);
 754         if (obj == NULL)
 755                 RETURN(-EALREADY);
 756
 757         sidx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
 758                             (char *)op->op_name, op->op_namelen);
 759         mds = obj->lo_stripes[sidx].ls_mds;
 760         lmv_object_put(obj);
 761
 762         rc = __lmv_fid_alloc(lmv, fid, mds);
 763         if (rc) {
 764                 CERROR("Can't allocate fid, rc %d\n", rc);
 765                 RETURN(rc);
 766         }
 767
 768         CDEBUG(D_INODE, "Allocate new fid "DFID" for slave "
 769                "obj -> mds #%x\n", PFID(fid), mds);
 770
 771         RETURN(rc);
 772 }