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  2008 Sun Microsystems, Inc. All rights reserved
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  */
  36
  37 #ifndef EXPORT_SYMTAB
  38 # define EXPORT_SYMTAB
  39 #endif
  40 #define DEBUG_SUBSYSTEM S_LMV
  41 #ifdef __KERNEL__
  42 #include <linux/slab.h>
  43 #include <linux/module.h>
  44 #include <linux/init.h>
  45 #include <linux/slab.h>
  46 #include <linux/pagemap.h>
  47 #include <asm/div64.h>
  48 #include <linux/seq_file.h>
  49 #include <linux/namei.h>
  50 # ifndef HAVE_VFS_INTENT_PATCHES
  51 # include <linux/lustre_intent.h>
  52 # endif
  53 #else
  54 #include <liblustre.h>
  55 #endif
  56
  57 #include <lustre/lustre_idl.h>
  58 #include <obd_support.h>
  59 #include <lustre_lib.h>
  60 #include <lustre_net.h>
  61 #include <lustre_dlm.h>
  62 #include <obd_class.h>
  63 #include <lprocfs_status.h>
  64 #include "lmv_internal.h"
  65
  66 static inline void lmv_drop_intent_lock(struct lookup_intent *it)
  67 {
  68         if (it->d.lustre.it_lock_mode != 0) {
  69                 ldlm_lock_decref((void *)&it->d.lustre.it_lock_handle,
  70                                  it->d.lustre.it_lock_mode);
  71                 it->d.lustre.it_lock_mode = 0;
  72         }
  73 }
  74
  75 int lmv_intent_remote(struct obd_export *exp, void *lmm,
  76                       int lmmsize, struct lookup_intent *it,
  77                       int flags, struct ptlrpc_request **reqp,
  78                       ldlm_blocking_callback cb_blocking,
  79                       int extra_lock_flags)
  80 {
  81         struct obd_device *obd = exp->exp_obd;
  82         struct lmv_obd *lmv = &obd->u.lmv;
  83         struct ptlrpc_request *req = NULL;
  84         struct lustre_handle plock;
  85         struct md_op_data *op_data;
  86         struct obd_export *tgt_exp;
  87         struct mdt_body *body;
  88         int pmode, rc = 0;
  89         ENTRY;
  90
  91         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_DLM_REP);
  92         LASSERT(body != NULL);
  93
  94         if (!(body->valid & OBD_MD_MDS))
  95                 RETURN(0);
  96
  97         /*
  98          * oh, MDS reports that this is remote inode case i.e. we have to ask
  99          * for real attrs on another MDS.
 100          */
 101         if (it->it_op & IT_LOOKUP) {
 102                 /*
 103                  * unfortunately, we have to lie to MDC/MDS to retrieve
 104                  * attributes llite needs.
 105                  */
 106                 it->it_op = IT_GETATTR;
 107         }
 108
 109         /* we got LOOKUP lock, but we really need attrs */
 110         pmode = it->d.lustre.it_lock_mode;
 111         if (pmode) {
 112                 plock.cookie = it->d.lustre.it_lock_handle;
 113                 it->d.lustre.it_lock_mode = 0;
 114                 it->d.lustre.it_data = 0;
 115         }
 116
 117         LASSERT(fid_is_sane(&body->fid1));
 118
 119         it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
 120
 121         tgt_exp = lmv_find_export(lmv, &body->fid1);
 122         if (IS_ERR(tgt_exp))
 123                 GOTO(out, rc = PTR_ERR(tgt_exp));
 124
 125         OBD_ALLOC_PTR(op_data);
 126         if (op_data == NULL)
 127                 GOTO(out, rc = -ENOMEM);
 128
 129         op_data->op_fid1 = body->fid1;
 130         op_data->op_bias = MDS_CROSS_REF;
 131
 132         rc = md_intent_lock(tgt_exp, op_data, lmm, lmmsize, it, flags,
 133                             &req, cb_blocking, extra_lock_flags);
 134
 135         /*
 136          * llite needs LOOKUP lock to track dentry revocation in order to
 137          * maintain dcache consistency. Thus drop UPDATE lock here and put
 138          * LOOKUP in request.
 139          */
 140         if (rc == 0) {
 141                 lmv_drop_intent_lock(it);
 142                 it->d.lustre.it_lock_handle = plock.cookie;
 143                 it->d.lustre.it_lock_mode = pmode;
 144         }
 145
 146         OBD_FREE_PTR(op_data);
 147         EXIT;
 148 out:
 149         if (rc && pmode)
 150                 ldlm_lock_decref(&plock, pmode);
 151
 152         ptlrpc_req_finished(*reqp);
 153         *reqp = req;
 154         return rc;
 155 }
 156
 157 int lmv_alloc_slave_fids(struct obd_device *obd, struct lu_fid *pid,
 158                          struct md_op_data *op, struct lu_fid *fid)
 159 {
 160         struct lmv_obd *lmv = &obd->u.lmv;
 161         struct lmv_obj *obj;
 162         mdsno_t mds;
 163         int mea_idx;
 164         int rc;
 165         ENTRY;
 166
 167         obj = lmv_obj_grab(obd, pid);
 168         if (!obj) {
 169                 CERROR("Object "DFID" should be split\n",
 170                        PFID(pid));
 171                 RETURN(0);
 172         }
 173
 174         mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
 175                                (char *)op->op_name, op->op_namelen);
 176         mds = obj->lo_inodes[mea_idx].li_mds;
 177         lmv_obj_put(obj);
 178
 179         rc = __lmv_fid_alloc(lmv, fid, mds);
 180         if (rc) {
 181                 CERROR("Can't allocate new fid, rc %d\n",
 182                        rc);
 183                 RETURN(rc);
 184         }
 185
 186         CDEBUG(D_INFO, "Allocate new fid "DFID" for split "
 187                "obj\n", PFID(fid));
 188
 189         RETURN(rc);
 190 }
 191
 192 /*
 193  * IT_OPEN is intended to open (and create, possible) an object. Parent (pid)
 194  * may be split dir.
 195  */
 196 int lmv_intent_open(struct obd_export *exp, struct md_op_data *op_data,
 197                     void *lmm, int lmmsize, struct lookup_intent *it,
 198                     int flags, struct ptlrpc_request **reqp,
 199                     ldlm_blocking_callback cb_blocking,
 200                     int extra_lock_flags)
 201 {
 202         struct obd_device *obd = exp->exp_obd;
 203         struct lu_fid rpid = op_data->op_fid1;
 204         struct lmv_obd *lmv = &obd->u.lmv;
 205         struct md_op_data *sop_data;
 206         struct obd_export *tgt_exp;
 207         struct lmv_stripe_md *mea;
 208         struct mdt_body *body;
 209         struct lmv_obj *obj;
 210         int rc, loop = 0;
 211         ENTRY;
 212
 213         OBD_ALLOC_PTR(sop_data);
 214         if (sop_data == NULL)
 215                 RETURN(-ENOMEM);
 216
 217         /* save op_data fro repeat case */
 218         *sop_data = *op_data;
 219
 220 repeat:
 221
 222         ++loop;
 223         LASSERT(loop <= 2);
 224         obj = lmv_obj_grab(obd, &rpid);
 225         if (obj) {
 226                 int mea_idx;
 227
 228                 /*
 229                  * Directory is already split, so we have to forward request to
 230                  * the right MDS.
 231                  */
 232                 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
 233                                        (char *)op_data->op_name,
 234                                        op_data->op_namelen);
 235
 236                 rpid = obj->lo_inodes[mea_idx].li_fid;
 237
 238                 sop_data->op_mds = obj->lo_inodes[mea_idx].li_mds;
 239                 tgt_exp = lmv_get_export(lmv, sop_data->op_mds);
 240                 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
 241                 lmv_obj_put(obj);
 242                 CDEBUG(D_OTHER, "Choose slave dir ("DFID")\n", PFID(&rpid));
 243         } else {
 244                 struct lmv_tgt_desc *tgt;
 245
 246                 sop_data->op_bias |= MDS_CHECK_SPLIT;
 247                 tgt = lmv_find_target(lmv, &rpid);
 248                 sop_data->op_mds = tgt->ltd_idx;
 249                 tgt_exp = tgt->ltd_exp;
 250         }
 251         if (IS_ERR(tgt_exp))
 252                 GOTO(out_free_sop_data, rc = PTR_ERR(tgt_exp));
 253
 254         sop_data->op_fid1 = rpid;
 255
 256         if (it->it_op & IT_CREAT) {
 257                 /*
 258                  * For open with IT_CREATE and for IT_CREATE cases allocate new
 259                  * fid and setup FLD for it.
 260                  */
 261                 /* save old child fid for correctly check stale data*/
 262                 sop_data->op_fid3 = sop_data->op_fid2;
 263                 rc = lmv_fid_alloc(exp, &sop_data->op_fid2, sop_data);
 264                 if (rc)
 265                         GOTO(out_free_sop_data, rc);
 266
 267                 if (rc == -ERESTART)
 268                         goto repeat;
 269                 else if (rc)
 270                         GOTO(out_free_sop_data, rc);
 271         }
 272
 273         rc = md_intent_lock(tgt_exp, sop_data, lmm, lmmsize, it, flags,
 274                             reqp, cb_blocking, extra_lock_flags);
 275
 276         if (rc == -ERESTART) {
 277                 LASSERT(*reqp != NULL);
 278                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *reqp,
 279                           "Got -ERESTART during open!\n");
 280                 ptlrpc_req_finished(*reqp);
 281                 *reqp = NULL;
 282                 it->d.lustre.it_data = 0;
 283
 284                 /*
 285                  * Directory got split. Time to update local object and repeat
 286                  * the request with proper MDS.
 287                  */
 288                 LASSERT(lu_fid_eq(&op_data->op_fid1, &rpid));
 289                 rc = lmv_handle_split(exp, &rpid);
 290                 if (rc == 0) {
 291                         /* We should reallocate child FID. */
 292                         rc = lmv_alloc_slave_fids(obd, &rpid, op_data,
 293                                                   &sop_data->op_fid2);
 294                         if (rc == 0)
 295                                 goto repeat;
 296                 }
 297         }
 298
 299         if (rc != 0)
 300                 GOTO(out_free_sop_data, rc);
 301
 302         /*
 303          * Okay, MDS has returned success. Probably name has been resolved in
 304          * remote inode.
 305          */
 306         rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
 307                                cb_blocking, extra_lock_flags);
 308         if (rc != 0) {
 309                 LASSERT(rc < 0);
 310                 /*
 311                  * This is possible, that some userspace application will try to
 312                  * open file as directory and we will have -ENOTDIR here. As
 313                  * this is normal situation, we should not print error here,
 314                  * only debug info.
 315                  */
 316                 CDEBUG(D_OTHER, "can't handle remote %s: dir "DFID"("DFID"):"
 317                        "%*s: %d\n", LL_IT2STR(it), PFID(&op_data->op_fid2),
 318                        PFID(&rpid), op_data->op_namelen, op_data->op_name, rc);
 319                 GOTO(out_free_sop_data, rc);
 320         }
 321
 322         /*
 323          * Nothing is found, do not access body->fid1 as it is zero and thus
 324          * pointless.
 325          */
 326         if ((it->d.lustre.it_disposition & DISP_LOOKUP_NEG) &&
 327             !(it->d.lustre.it_disposition & DISP_OPEN_CREATE) &&
 328             !(it->d.lustre.it_disposition & DISP_OPEN_OPEN))
 329                 GOTO(out_free_sop_data, rc = 0);
 330
 331         /* caller may use attrs MDS returns on IT_OPEN lock request so, we have
 332          * to update them for split dir */
 333         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_DLM_REP);
 334         LASSERT(body != NULL);
 335
 336         /* could not find object, FID is not present in response. */
 337         if (!(body->valid & OBD_MD_FLID))
 338                 GOTO(out_free_sop_data, rc = 0);
 339
 340         obj = lmv_obj_grab(obd, &body->fid1);
 341         if (!obj && (mea = lmv_get_mea(*reqp))) {
 342                 /* FIXME: capability for remote! */
 343                 /* wow! this is split dir, we'd like to handle it */
 344                 obj = lmv_obj_create(exp, &body->fid1, mea);
 345                 if (IS_ERR(obj))
 346                         GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
 347         }
 348
 349         if (obj) {
 350                 /* This is split dir and we'd want to get attrs. */
 351                 CDEBUG(D_OTHER, "attrs from slaves for "DFID"\n",
 352                        PFID(&body->fid1));
 353
 354                 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
 355                                            cb_blocking, extra_lock_flags);
 356         } else if (S_ISDIR(body->mode)) {
 357                 CDEBUG(D_OTHER, "object "DFID" has not lmv obj?\n",
 358                        PFID(&body->fid1));
 359         }
 360
 361         if (obj)
 362                 lmv_obj_put(obj);
 363
 364         EXIT;
 365 out_free_sop_data:
 366         OBD_FREE_PTR(sop_data);
 367         return rc;
 368 }
 369
 370 int lmv_intent_getattr(struct obd_export *exp, struct md_op_data *op_data,
 371                        void *lmm, int lmmsize, struct lookup_intent *it,
 372                        int flags, struct ptlrpc_request **reqp,
 373                        ldlm_blocking_callback cb_blocking,
 374                        int extra_lock_flags)
 375 {
 376         struct lmv_obj *obj = NULL, *obj2 = NULL;
 377         struct obd_device *obd = exp->exp_obd;
 378         struct lu_fid rpid = op_data->op_fid1;
 379         struct lmv_obd *lmv = &obd->u.lmv;
 380         struct md_op_data *sop_data;
 381         struct lmv_stripe_md *mea;
 382         struct mdt_body *body;
 383         mdsno_t mds;
 384         int rc = 0;
 385         ENTRY;
 386
 387         OBD_ALLOC_PTR(sop_data);
 388         if (sop_data == NULL)
 389                 RETURN(-ENOMEM);
 390
 391         /* save op_data fro repeat case */
 392         *sop_data = *op_data;
 393
 394         if (fid_is_sane(&op_data->op_fid2)) {
 395                 /*
 396                  * Caller wants to revalidate attrs of obj we have to revalidate
 397                  * slaves if requested object is split directory.
 398                  */
 399                 CDEBUG(D_OTHER, "revalidate attrs for "DFID"\n",
 400                        PFID(&op_data->op_fid2));
 401
 402                 rc = lmv_fld_lookup(lmv, &op_data->op_fid2, &mds);
 403                 if (rc)
 404                         GOTO(out_free_sop_data, rc);
 405 #if 0
 406                 /*
 407                  * In fact, we do not need this with current intent_lock(), but
 408                  * it may change some day.
 409                  */
 410                 obj = lmv_obj_grab(obd, &op_data->op_fid2);
 411                 if (obj) {
 412                         if (!lu_fid_eq(&op_data->op_fid1, &op_data->op_fid2)){
 413                                 rpid = obj->lo_inodes[mds].li_fid;
 414                                 mds = obj->lo_inodes[mds].li_mds;
 415                         }
 416                         lmv_obj_put(obj);
 417                 }
 418 #endif
 419         } else {
 420                 CDEBUG(D_OTHER, "INTENT getattr for %*s on "DFID"\n",
 421                        op_data->op_namelen, op_data->op_name,
 422                        PFID(&op_data->op_fid1));
 423
 424                 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
 425                 if (rc)
 426                         GOTO(out_free_sop_data, rc);
 427                 obj = lmv_obj_grab(obd, &op_data->op_fid1);
 428                 if (obj && op_data->op_namelen) {
 429                         int mea_idx;
 430
 431                         /* directory is already split. calculate mds */
 432                         mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
 433                                                (char *)op_data->op_name,
 434                                                op_data->op_namelen);
 435                         rpid = obj->lo_inodes[mea_idx].li_fid;
 436                         mds = obj->lo_inodes[mea_idx].li_mds;
 437                         sop_data->op_bias &= ~MDS_CHECK_SPLIT;
 438                         lmv_obj_put(obj);
 439
 440                         CDEBUG(D_OTHER, "forward to MDS #"LPU64" (slave "DFID")\n",
 441                                mds, PFID(&rpid));
 442                 } else {
 443                         rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
 444                         if (rc)
 445                                 GOTO(out_free_sop_data, rc);
 446                         sop_data->op_bias |= MDS_CHECK_SPLIT;
 447                 }
 448         }
 449
 450         sop_data->op_fid1 = rpid;
 451
 452         rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm,
 453                             lmmsize, it, flags, reqp, cb_blocking,
 454                             extra_lock_flags);
 455
 456         LASSERTF(rc != -ERESTART, "GETATTR: Got unhandled -ERESTART!\n");
 457         if (rc < 0)
 458                 GOTO(out_free_sop_data, rc);
 459
 460         if (obj && rc > 0) {
 461                 /*
 462                  * This is split dir. In order to optimize things a bit, we
 463                  * consider obj valid updating missing parts.
 464
 465                  * FIXME: do we need to return any lock here? It would be fine
 466                  * if we don't. This means that nobody should use UPDATE lock to
 467                  * notify about object * removal.
 468                  */
 469                 CDEBUG(D_OTHER,
 470                        "revalidate slaves for "DFID", rc %d\n",
 471                        PFID(&op_data->op_fid2), rc);
 472
 473                 LASSERT(fid_is_sane(&op_data->op_fid2));
 474                 rc = lmv_revalidate_slaves(exp, reqp, &op_data->op_fid2, it, rc,
 475                                            cb_blocking, extra_lock_flags);
 476                 GOTO(out_free_sop_data, rc);
 477         }
 478
 479         if (*reqp == NULL)
 480                 GOTO(out_free_sop_data, rc);
 481
 482         /*
 483          * okay, MDS has returned success. Probably name has been resolved in
 484          * remote inode.
 485          */
 486         rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags,
 487                                reqp, cb_blocking, extra_lock_flags);
 488         if (rc < 0)
 489                 GOTO(out_free_sop_data, rc);
 490
 491         /*
 492          * Nothing is found, do not access body->fid1 as it is zero and thus
 493          * pointless.
 494          */
 495         if (it->d.lustre.it_disposition & DISP_LOOKUP_NEG)
 496                 GOTO(out_free_sop_data, rc = 0);
 497
 498         LASSERT(*reqp);
 499         LASSERT((*reqp)->rq_repmsg);
 500         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
 501         LASSERT(body != NULL);
 502
 503         /* could not find object, FID is not present in response. */
 504         if (!(body->valid & OBD_MD_FLID))
 505                 GOTO(out_free_sop_data, rc = 0);
 506
 507         obj2 = lmv_obj_grab(obd, &body->fid1);
 508
 509         if (!obj2 && (mea = lmv_get_mea(*reqp))) {
 510
 511                 /* FIXME remote capability! */
 512                 /* wow! this is split dir, we'd like to handle it. */
 513                 obj2 = lmv_obj_create(exp, &body->fid1, mea);
 514                 if (IS_ERR(obj2))
 515                         GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj2));
 516         }
 517
 518         if (obj2) {
 519                 /* this is split dir and we'd want to get attrs */
 520                 CDEBUG(D_OTHER, "attrs from slaves for "DFID", rc %d\n",
 521                        PFID(&body->fid1), rc);
 522
 523                 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
 524                                            cb_blocking, extra_lock_flags);
 525                 lmv_obj_put(obj2);
 526         }
 527
 528         EXIT;
 529 out_free_sop_data:
 530         OBD_FREE_PTR(sop_data);
 531         return rc;
 532 }
 533
 534 /* this is not used currently */
 535 int lmv_lookup_slaves(struct obd_export *exp, struct ptlrpc_request **reqp)
 536 {
 537         struct obd_device *obd = exp->exp_obd;
 538         struct lmv_obd *lmv = &obd->u.lmv;
 539         struct lustre_handle *lockh;
 540         struct md_op_data *op_data;
 541         struct ldlm_lock *lock;
 542         struct mdt_body *body2;
 543         struct mdt_body *body;
 544         struct lmv_obj *obj;
 545         int i, rc = 0;
 546         ENTRY;
 547
 548         LASSERT(reqp);
 549         LASSERT(*reqp);
 550
 551         /*
 552          * Master is locked. we'd like to take locks on slaves and update
 553          * attributes to be returned from the slaves it's important that lookup
 554          * is called in two cases:
 555
 556          *  - for first time (dcache has no such a resolving yet).  -
 557          *  ->d_revalidate() returned false.
 558
 559          * Last case possible only if all the objs (master and all slaves aren't
 560          * valid.
 561          */
 562
 563         OBD_ALLOC_PTR(op_data);
 564         if (op_data == NULL)
 565                 RETURN(-ENOMEM);
 566
 567         body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
 568         LASSERT(body != NULL);
 569
 570         LASSERT((body->valid & OBD_MD_FLID) != 0);
 571         obj = lmv_obj_grab(obd, &body->fid1);
 572         LASSERT(obj != NULL);
 573
 574         CDEBUG(D_OTHER, "lookup slaves for "DFID"\n",
 575                PFID(&body->fid1));
 576
 577         lmv_obj_lock(obj);
 578
 579         for (i = 0; i < obj->lo_objcount; i++) {
 580                 struct lu_fid fid = obj->lo_inodes[i].li_fid;
 581                 struct ptlrpc_request *req = NULL;
 582                 struct obd_export *tgt_exp;
 583                 struct lookup_intent it;
 584
 585                 if (lu_fid_eq(&fid, &obj->lo_fid))
 586                         /* skip master obj */
 587                         continue;
 588
 589                 CDEBUG(D_OTHER, "lookup slave "DFID"\n", PFID(&fid));
 590
 591                 /* is obj valid? */
 592                 memset(&it, 0, sizeof(it));
 593                 it.it_op = IT_GETATTR;
 594
 595                 memset(op_data, 0, sizeof(*op_data));
 596                 op_data->op_fid1 = fid;
 597                 op_data->op_fid2 = fid;
 598                 op_data->op_bias = MDS_CROSS_REF;
 599
 600                 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
 601                 if (IS_ERR(tgt_exp))
 602                         GOTO(cleanup, rc = PTR_ERR(tgt_exp));
 603
 604                 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0,
 605                                     &req, lmv_blocking_ast, 0);
 606
 607                 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
 608                 if (rc > 0 && req == NULL) {
 609                         /* nice, this slave is valid */
 610                         LASSERT(req == NULL);
 611                         CDEBUG(D_OTHER, "cached\n");
 612                         goto release_lock;
 613                 }
 614
 615                 if (rc < 0) {
 616                         /* error during lookup */
 617                         GOTO(cleanup, rc);
 618                 }
 619                 lock = ldlm_handle2lock(lockh);
 620                 LASSERT(lock);
 621
 622                 lock->l_ast_data = lmv_obj_get(obj);
 623
 624                 body2 = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 625                 LASSERT(body2 != NULL);
 626
 627                 obj->lo_inodes[i].li_size = body2->size;
 628
 629                 CDEBUG(D_OTHER, "fresh: %lu\n",
 630                        (unsigned long)obj->lo_inodes[i].li_size);
 631
 632                 LDLM_LOCK_PUT(lock);
 633
 634                 if (req)
 635                         ptlrpc_req_finished(req);
 636 release_lock:
 637                 lmv_update_body(body, obj->lo_inodes + i);
 638
 639                 if (it.d.lustre.it_lock_mode) {
 640                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
 641                         it.d.lustre.it_lock_mode = 0;
 642                 }
 643         }
 644
 645         EXIT;
 646 cleanup:
 647         lmv_obj_unlock(obj);
 648         lmv_obj_put(obj);
 649         OBD_FREE_PTR(op_data);
 650         return rc;
 651 }
 652
 653 int lmv_intent_lookup(struct obd_export *exp, struct md_op_data *op_data,
 654                       void *lmm, int lmmsize, struct lookup_intent *it,
 655                       int flags, struct ptlrpc_request **reqp,
 656                       ldlm_blocking_callback cb_blocking,
 657                       int extra_lock_flags)
 658 {
 659         struct obd_device *obd = exp->exp_obd;
 660         struct lu_fid rpid = op_data->op_fid1;
 661         struct lmv_obd *lmv = &obd->u.lmv;
 662         struct md_op_data *sop_data;
 663         struct lmv_stripe_md *mea;
 664         struct mdt_body *body;
 665         struct lmv_obj *obj;
 666         int rc, loop = 0;
 667         int mea_idx;
 668         mdsno_t mds;
 669         ENTRY;
 670
 671         OBD_ALLOC_PTR(sop_data);
 672         if (sop_data == NULL)
 673                 RETURN(-ENOMEM);
 674
 675         /* save op_data fro repeat case */
 676         *sop_data = *op_data;
 677
 678         /*
 679          * IT_LOOKUP is intended to produce name -> fid resolving (let's call
 680          * this lookup below) or to confirm requested resolving is still valid
 681          * (let's call this revalidation) fid_is_sane(&sop_data->op_fid2) specifies
 682          * revalidation.
 683          */
 684         if (fid_is_sane(&op_data->op_fid2)) {
 685                 /*
 686                  * This is revalidate: we have to check is LOOKUP lock still
 687                  * valid for given fid. Very important part is that we have to
 688                  * choose right mds because namespace is per mds.
 689                  */
 690                 rpid = op_data->op_fid1;
 691                 obj = lmv_obj_grab(obd, &rpid);
 692                 if (obj) {
 693                         mea_idx = raw_name2idx(obj->lo_hashtype,
 694                                                obj->lo_objcount,
 695                                                (char *)op_data->op_name,
 696                                                op_data->op_namelen);
 697                         rpid = obj->lo_inodes[mea_idx].li_fid;
 698                         mds = obj->lo_inodes[mea_idx].li_mds;
 699                         sop_data->op_bias &= ~MDS_CHECK_SPLIT;
 700                         lmv_obj_put(obj);
 701                 } else {
 702                         rc = lmv_fld_lookup(lmv, &rpid, &mds);
 703                         if (rc)
 704                                 GOTO(out_free_sop_data, rc);
 705                         sop_data->op_bias |= MDS_CHECK_SPLIT;
 706                 }
 707
 708                 CDEBUG(D_OTHER, "revalidate lookup for "DFID" to #"LPU64" MDS\n",
 709                        PFID(&op_data->op_fid2), mds);
 710         } else {
 711 repeat:
 712                 ++loop;
 713                 LASSERT(loop <= 2);
 714
 715                 /*
 716                  * This is lookup. During lookup we have to update all the
 717                  * attributes, because returned values will be put in struct
 718                  * inode.
 719                  */
 720                 obj = lmv_obj_grab(obd, &op_data->op_fid1);
 721                 if (obj) {
 722                         if (op_data->op_namelen) {
 723                                 /* directory is already split. calculate mds */
 724                                 mea_idx = raw_name2idx(obj->lo_hashtype,
 725                                                        obj->lo_objcount,
 726                                                        (char *)op_data->op_name,
 727                                                        op_data->op_namelen);
 728                                 rpid = obj->lo_inodes[mea_idx].li_fid;
 729                                 mds = obj->lo_inodes[mea_idx].li_mds;
 730                         }
 731                         sop_data->op_bias &= ~MDS_CHECK_SPLIT;
 732                         lmv_obj_put(obj);
 733                 } else {
 734                         rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
 735                         if (rc)
 736                                 GOTO(out_free_sop_data, rc);
 737                         sop_data->op_bias |= MDS_CHECK_SPLIT;
 738                 }
 739                 fid_zero(&sop_data->op_fid2);
 740         }
 741
 742         sop_data->op_bias &= ~MDS_CROSS_REF;
 743         sop_data->op_fid1 = rpid;
 744
 745         rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm, lmmsize,
 746                             it, flags, reqp, cb_blocking, extra_lock_flags);
 747         if (rc > 0) {
 748                 LASSERT(fid_is_sane(&op_data->op_fid2));
 749                 /*
 750                  * Very interesting. it seems object is still valid but for some
 751                  * reason llite calls lookup, not revalidate.
 752                  */
 753                 CDEBUG(D_OTHER, "lookup for "DFID" and data should be uptodate\n",
 754                        PFID(&rpid));
 755                 LASSERT(*reqp == NULL);
 756                 GOTO(out_free_sop_data, rc);
 757         }
 758
 759         if (rc == 0 && *reqp == NULL) {
 760                 /* once again, we're asked for lookup, not revalidate */
 761                 CDEBUG(D_OTHER, "lookup for "DFID" and data should be uptodate\n",
 762                        PFID(&rpid));
 763                 GOTO(out_free_sop_data, rc);
 764         }
 765
 766         if (rc == -ERESTART) {
 767                 LASSERT(*reqp != NULL);
 768                 DEBUG_REQ(D_WARNING|D_RPCTRACE, *reqp,
 769                           "Got -ERESTART during lookup!\n");
 770                 ptlrpc_req_finished(*reqp);
 771                 *reqp = NULL;
 772                 it->d.lustre.it_data = 0;
 773                 /*
 774                  * Directory got split since last update. This shouldn't be
 775                  * because splitting causes lock revocation, so revalidate had
 776                  * to fail and lookup on dir had to return mea.
 777                  */
 778                 CWARN("we haven't knew about directory splitting!\n");
 779                 LASSERT(obj == NULL);
 780
 781                 obj = lmv_obj_create(exp, &rpid, NULL);
 782                 if (IS_ERR(obj))
 783                         GOTO(out_free_sop_data, rc = PTR_ERR(obj));
 784                 lmv_obj_put(obj);
 785                 goto repeat;
 786         }
 787
 788         if (rc < 0)
 789                 GOTO(out_free_sop_data, rc);
 790
 791         /*
 792          * Okay, MDS has returned success. Probably name has been resolved in
 793          * remote inode.
 794          */
 795         rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
 796                                cb_blocking, extra_lock_flags);
 797
 798         if (rc == 0 && (mea = lmv_get_mea(*reqp))) {
 799                 /* Wow! This is split dir, we'd like to handle it. */
 800                 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
 801                 LASSERT(body != NULL);
 802                 LASSERT((body->valid & OBD_MD_FLID) != 0);
 803
 804                 obj = lmv_obj_grab(obd, &body->fid1);
 805                 if (!obj) {
 806                         obj = lmv_obj_create(exp, &body->fid1, mea);
 807                         if (IS_ERR(obj))
 808                                 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
 809                 }
 810                 lmv_obj_put(obj);
 811         }
 812
 813         EXIT;
 814 out_free_sop_data:
 815         OBD_FREE_PTR(sop_data);
 816         return rc;
 817 }
 818
 819 int lmv_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
 820                     void *lmm, int lmmsize, struct lookup_intent *it,
 821                     int flags, struct ptlrpc_request **reqp,
 822                     ldlm_blocking_callback cb_blocking,
 823                     int extra_lock_flags)
 824 {
 825         struct obd_device *obd = exp->exp_obd;
 826         int rc;
 827         ENTRY;
 828
 829         LASSERT(it != NULL);
 830         LASSERT(fid_is_sane(&op_data->op_fid1));
 831
 832         CDEBUG(D_OTHER, "INTENT LOCK '%s' for '%*s' on "DFID"\n",
 833                LL_IT2STR(it), op_data->op_namelen, op_data->op_name,
 834                PFID(&op_data->op_fid1));
 835
 836         rc = lmv_check_connect(obd);
 837         if (rc)
 838                 RETURN(rc);
 839
 840         if (it->it_op & IT_LOOKUP)
 841                 rc = lmv_intent_lookup(exp, op_data, lmm, lmmsize, it,
 842                                        flags, reqp, cb_blocking,
 843                                        extra_lock_flags);
 844         else if (it->it_op & IT_OPEN)
 845                 rc = lmv_intent_open(exp, op_data, lmm, lmmsize, it,
 846                                      flags, reqp, cb_blocking,
 847                                      extra_lock_flags);
 848         else if (it->it_op & IT_GETATTR)
 849                 rc = lmv_intent_getattr(exp, op_data,lmm, lmmsize, it,
 850                                         flags, reqp, cb_blocking,
 851                                         extra_lock_flags);
 852         else
 853                 LBUG();
 854         RETURN(rc);
 855 }
 856
 857 int lmv_revalidate_slaves(struct obd_export *exp, struct ptlrpc_request **reqp,
 858                           const struct lu_fid *mid, struct lookup_intent *oit,
 859                           int master_valid, ldlm_blocking_callback cb_blocking,
 860                           int extra_lock_flags)
 861 {
 862         struct obd_device *obd = exp->exp_obd;
 863         struct ptlrpc_request *mreq = *reqp;
 864         struct lmv_obd *lmv = &obd->u.lmv;
 865         struct lustre_handle master_lockh;
 866         struct obd_export *tgt_exp;
 867         struct md_op_data *op_data;
 868         struct ldlm_lock *lock;
 869         unsigned long size = 0;
 870         struct mdt_body *body;
 871         struct lmv_obj *obj;
 872         int master_lock_mode;
 873         int i, rc = 0;
 874         ENTRY;
 875
 876         OBD_ALLOC_PTR(op_data);
 877         if (op_data == NULL)
 878                 RETURN(-ENOMEM);
 879
 880         /*
 881          * We have to loop over the subobjects, check validity and update them
 882          * from MDSs if needed. it's very useful that we need not to update all
 883          * the fields. say, common fields (that are equal on all the subojects
 884          * need not to be update, another fields (i_size, for example) are
 885          * cached all the time.
 886          */
 887         obj = lmv_obj_grab(obd, mid);
 888         LASSERT(obj != NULL);
 889
 890         master_lock_mode = 0;
 891
 892         lmv_obj_lock(obj);
 893
 894         for (i = 0; i < obj->lo_objcount; i++) {
 895                 struct lu_fid fid = obj->lo_inodes[i].li_fid;
 896                 struct lustre_handle *lockh = NULL;
 897                 struct ptlrpc_request *req = NULL;
 898                 ldlm_blocking_callback cb;
 899                 struct lookup_intent it;
 900                 int master = 0;
 901
 902                 CDEBUG(D_OTHER, "revalidate subobj "DFID"\n",
 903                        PFID(&fid));
 904
 905                 memset(op_data, 0, sizeof(*op_data));
 906                 memset(&it, 0, sizeof(it));
 907                 it.it_op = IT_GETATTR;
 908
 909                 cb = lmv_blocking_ast;
 910
 911                 if (lu_fid_eq(&fid, &obj->lo_fid)) {
 912                         if (master_valid) {
 913                                 /*
 914                                  * lmv_intent_getattr() already checked
 915                                  * validness and took the lock.
 916                                  */
 917                                 if (mreq) {
 918                                         /*
 919                                          * It even got the reply refresh attrs
 920                                          * from that reply.
 921                                          */
 922                                         body = req_capsule_server_get(
 923                                                                 &mreq->rq_pill,
 924                                                                 &RMF_MDT_BODY);
 925                                         LASSERT(body != NULL);
 926                                         goto update;
 927                                 }
 928                                 /* take already cached attrs into account */
 929                                 CDEBUG(D_OTHER,
 930                                        "master is locked and cached\n");
 931                                 goto release_lock;
 932                         }
 933                         master = 1;
 934                         cb = cb_blocking;
 935                 }
 936
 937                 op_data->op_fid1 = fid;
 938                 op_data->op_fid2 = fid;
 939                 op_data->op_bias = MDS_CROSS_REF;
 940
 941                 /* Is obj valid? */
 942                 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
 943                 if (IS_ERR(tgt_exp))
 944                         GOTO(cleanup, rc = PTR_ERR(tgt_exp));
 945
 946                 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0, &req, cb,
 947                                     extra_lock_flags);
 948
 949                 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
 950                 if (rc > 0 && req == NULL) {
 951                         /* Nice, this slave is valid */
 952                         LASSERT(req == NULL);
 953                         CDEBUG(D_OTHER, "cached\n");
 954                         goto release_lock;
 955                 }
 956
 957                 if (rc < 0)
 958                         GOTO(cleanup, rc);
 959
 960                 if (master) {
 961                         LASSERT(master_valid == 0);
 962                         /* Save lock on master to be returned to the caller. */
 963                         CDEBUG(D_OTHER, "no lock on master yet\n");
 964                         memcpy(&master_lockh, lockh, sizeof(master_lockh));
 965                         master_lock_mode = it.d.lustre.it_lock_mode;
 966                         it.d.lustre.it_lock_mode = 0;
 967                 } else {
 968                         /* This is slave. We want to control it. */
 969                         lock = ldlm_handle2lock(lockh);
 970                         LASSERT(lock != NULL);
 971                         lock->l_ast_data = lmv_obj_get(obj);
 972                         LDLM_LOCK_PUT(lock);
 973                 }
 974
 975                 if (*reqp == NULL) {
 976                         /*
 977                          * This is first reply, we'll use it to return updated
 978                          * data back to the caller.
 979                          */
 980                         LASSERT(req);
 981                         ptlrpc_request_addref(req);
 982                         *reqp = req;
 983                 }
 984
 985                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
 986                 LASSERT(body != NULL);
 987
 988 update:
 989                 obj->lo_inodes[i].li_size = body->size;
 990
 991                 CDEBUG(D_OTHER, "fresh: %lu\n",
 992                        (unsigned long)obj->lo_inodes[i].li_size);
 993
 994                 if (req)
 995                         ptlrpc_req_finished(req);
 996 release_lock:
 997                 size += obj->lo_inodes[i].li_size;
 998
 999                 if (it.d.lustre.it_lock_mode) {
1000                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
1001                         it.d.lustre.it_lock_mode = 0;
1002                 }
1003         }
1004
1005         if (*reqp) {
1006                 /*
1007                  * Some attrs got refreshed, we have reply and it's time to put
1008                  * fresh attrs to it.
1009                  */
1010                 CDEBUG(D_OTHER, "return refreshed attrs: size = %lu\n",
1011                        (unsigned long)size);
1012
1013                 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
1014                 LASSERT(body != NULL);
1015
1016                 body->size = size;
1017
1018                 if (mreq == NULL) {
1019                         /*
1020                          * Very important to maintain mds num the same because
1021                          * of revalidation. mreq == NULL means that caller has
1022                          * no reply and the only attr we can return is size.
1023                          */
1024                         body->valid = OBD_MD_FLSIZE;
1025                 }
1026                 if (master_valid == 0) {
1027                         oit->d.lustre.it_lock_handle = master_lockh.cookie;
1028                         oit->d.lustre.it_lock_mode = master_lock_mode;
1029                 }
1030                 rc = 0;
1031         } else {
1032                 /* It seems all the attrs are fresh and we did no request */
1033                 CDEBUG(D_OTHER, "all the attrs were fresh\n");
1034                 if (master_valid == 0)
1035                         oit->d.lustre.it_lock_mode = master_lock_mode;
1036                 rc = 1;
1037         }
1038
1039         EXIT;
1040 cleanup:
1041         OBD_FREE_PTR(op_data);
1042         lmv_obj_unlock(obj);
1043         lmv_obj_put(obj);
1044         return rc;
1045 }