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