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 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 #else
  35 #include <liblustre.h>
  36 #endif
  37
  38 #include <linux/obd_support.h>
  39 #include <linux/lustre_lib.h>
  40 #include <linux/lustre_net.h>
  41 #include <linux/lustre_idl.h>
  42 #include <linux/lustre_dlm.h>
  43 #include <linux/lustre_mds.h>
  44 #include <linux/obd_class.h>
  45 #include <linux/obd_ost.h>
  46 #include <linux/lprocfs_status.h>
  47 #include <linux/lustre_fsfilt.h>
  48 #include <linux/obd_lmv.h>
  49 #include "lmv_internal.h"
  50
  51
  52 static inline void lmv_drop_intent_lock(struct lookup_intent *it)
  53 {
  54         if (it->d.lustre.it_lock_mode != 0)
  55                 ldlm_lock_decref((void *)&it->d.lustre.it_lock_handle,
  56                                  it->d.lustre.it_lock_mode);
  57 }
  58
  59 int lmv_handle_remote_inode(struct obd_export *exp, struct ll_uctxt *uctxt,
  60                             void *lmm, int lmmsize,
  61                             struct lookup_intent *it, int flags,
  62                             struct ptlrpc_request **reqp,
  63                             ldlm_blocking_callback cb_blocking)
  64 {
  65         struct obd_device *obd = exp->exp_obd;
  66         struct lmv_obd *lmv = &obd->u.lmv;
  67         struct mds_body *body = NULL;
  68         int rc = 0;
  69         ENTRY;
  70
  71         body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
  72         LASSERT(body != NULL);
  73
  74         if (body->valid & OBD_MD_MDS) {
  75                 /* oh, MDS reports that this is remote inode case
  76                  * i.e. we have to ask for real attrs on another MDS */
  77                 struct ptlrpc_request *req;
  78                 struct ll_fid nfid;
  79                 struct lustre_handle plock;
  80                 int pmode;
  81
  82                 if (it->it_op == IT_LOOKUP) {
  83                         /* unfortunately, we have to lie to MDC/MDS to
  84                          * retrieve attributes llite needs */
  85                         it->it_op = IT_GETATTR;
  86                 }
  87
  88                 /* we got LOOKUP lock, but we really need attrs */
  89                 pmode = it->d.lustre.it_lock_mode;
  90                 if (pmode) {
  91                         memcpy(&plock, &it->d.lustre.it_lock_handle,
  92                                         sizeof(plock));
  93                         it->d.lustre.it_lock_mode = 0;
  94                 }
  95
  96                 nfid = body->fid1;
  97                 it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
  98                 rc = md_intent_lock(lmv->tgts[nfid.mds].ltd_exp, uctxt, &nfid,
  99                                     NULL, 0, lmm, lmmsize, NULL, it, flags,
 100                                     &req, cb_blocking);
 101
 102                 /* llite needs LOOKUP lock to track dentry revocation in
 103                  * order to maintain dcache consistency. thus drop UPDATE
 104                  * lock here and put LOOKUP in request */
 105                 if (rc == 0) {
 106                         lmv_drop_intent_lock(it);
 107                         memcpy(&it->d.lustre.it_lock_handle, &plock,
 108                                         sizeof(plock));
 109                         it->d.lustre.it_lock_mode = pmode;
 110
 111                 } else if (pmode)
 112                         ldlm_lock_decref(&plock, pmode);
 113
 114                 ptlrpc_req_finished(*reqp);
 115                 *reqp = req;
 116         }
 117         RETURN(rc);
 118 }
 119
 120 int lmv_intent_open(struct obd_export *exp, struct ll_uctxt *uctxt,
 121                     struct ll_fid *pfid, const char *name, int len,
 122                     void *lmm, int lmmsize, struct ll_fid *cfid,
 123                     struct lookup_intent *it, int flags,
 124                     struct ptlrpc_request **reqp,
 125                     ldlm_blocking_callback cb_blocking)
 126 {
 127         struct obd_device *obd = exp->exp_obd;
 128         struct lmv_obd *lmv = &obd->u.lmv;
 129         struct mds_body *body = NULL;
 130         struct ll_fid rpfid = *pfid;
 131         struct lmv_obj *obj;
 132         struct mea *mea;
 133         int rc, mds;
 134         ENTRY;
 135
 136         /* IT_OPEN is intended to open (and create, possible) an object. Parent
 137          * (pfid) may be splitted dir */
 138
 139 repeat:
 140         mds = rpfid.mds;
 141         obj = lmv_grab_obj(obd, &rpfid);
 142         if (obj) {
 143                 /* directory is already splitted, so we have to forward
 144                  * request to the right MDS */
 145                 mds = raw_name2idx(obj->objcount, (char *)name, len);
 146                 CDEBUG(D_OTHER, "forward to MDS #%u\n", mds);
 147
 148                 rpfid = obj->objs[mds].fid;
 149                 lmv_put_obj(obj);
 150         }
 151
 152         rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
 153                             len, lmm, lmmsize, cfid, it, flags, reqp,
 154                             cb_blocking);
 155         if (rc == -ERESTART) {
 156                 /* directory got splitted. time to update local object
 157                  * and repeat the request with proper MDS */
 158                 LASSERT(fid_equal(pfid, &rpfid));
 159                 rc = lmv_get_mea_and_update_object(exp, &rpfid);
 160                 if (rc == 0) {
 161                         ptlrpc_req_finished(*reqp);
 162                         goto repeat;
 163                 }
 164         }
 165         if (rc != 0)
 166                 RETURN(rc);
 167
 168         /* okay, MDS has returned success. Probably name has been resolved in
 169          * remote inode */
 170         rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it,
 171                                      flags, reqp, cb_blocking);
 172         if (rc != 0) {
 173                 LASSERT(rc < 0);
 174                 RETURN(rc);
 175         }
 176
 177         /* caller may use attrs MDS returns on IT_OPEN lock request so, we have
 178          * to update them for splitted dir */
 179         body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
 180         LASSERT(body != NULL);
 181
 182         cfid = &body->fid1;
 183         obj = lmv_grab_obj(obd, cfid);
 184         if (!obj && (mea = body_of_splitted_dir(*reqp, 1))) {
 185                 /* wow! this is splitted dir, we'd like to handle it */
 186                 obj = lmv_create_obj(exp, &body->fid1, mea);
 187                 if (IS_ERR(obj))
 188                         RETURN(PTR_ERR(obj));
 189         }
 190
 191         if (obj) {
 192                 /* this is splitted dir and we'd want to get attrs */
 193                 CDEBUG(D_OTHER, "attrs from slaves for %lu/%lu/%lu\n",
 194                        (unsigned long)cfid->mds, (unsigned long)cfid->id,
 195                        (unsigned long)cfid->generation);
 196                 rc = lmv_revalidate_slaves(exp, reqp, cfid,
 197                                            it, 1, cb_blocking);
 198         } else if (S_ISDIR(body->mode)) {
 199                 /*CWARN("hmmm, %lu/%lu/%lu has not lmv obj?!\n",
 200                                 (unsigned long) cfid->mds,
 201                                 (unsigned long) cfid->id,
 202                                 (unsigned long) cfid->generation);*/
 203         }
 204
 205         if (obj)
 206                 lmv_put_obj(obj);
 207
 208         RETURN(rc);
 209 }
 210
 211 int lmv_intent_getattr(struct obd_export *exp, struct ll_uctxt *uctxt,
 212                        struct ll_fid *pfid, const char *name, int len,
 213                        void *lmm, int lmmsize, struct ll_fid *cfid,
 214                        struct lookup_intent *it, int flags,
 215                        struct ptlrpc_request **reqp,
 216                        ldlm_blocking_callback cb_blocking)
 217 {
 218         struct obd_device *obd = exp->exp_obd;
 219         struct lmv_obd *lmv = &obd->u.lmv;
 220         struct mds_body *body = NULL;
 221         struct ll_fid rpfid = *pfid;
 222         struct lmv_obj *obj, *obj2;
 223         struct mea *mea;
 224         int rc = 0, mds;
 225         ENTRY;
 226
 227         if (cfid) {
 228                 /* caller wants to revalidate attrs of obj we have to revalidate
 229                  * slaves if requested object is splitted directory */
 230                 CDEBUG(D_OTHER, "revalidate attrs for %lu/%lu/%lu\n",
 231                        (unsigned long)cfid->mds, (unsigned long)cfid->id,
 232                        (unsigned long)cfid->generation);
 233                 mds = cfid->mds;
 234                 obj = lmv_grab_obj(obd, cfid);
 235                 if (obj) {
 236                         /* in fact, we need not this with current intent_lock(),
 237                          * but it may change some day */
 238                         rpfid = obj->objs[mds].fid;
 239                         lmv_put_obj(obj);
 240                 }
 241                 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
 242                                     len, lmm, lmmsize, cfid, it, flags, reqp,
 243                                     cb_blocking);
 244                 if (obj && rc >= 0) {
 245                         /* this is splitted dir. In order to optimize things a
 246                          * bit, we consider obj valid updating missing parts.
 247
 248                          * FIXME: do we need to return any lock here? It would
 249                          * be fine if we don't. this means that nobody should
 250                          * use UPDATE lock to notify about object * removal */
 251                         CDEBUG(D_OTHER,
 252                                "revalidate slaves for %lu/%lu/%lu, rc %d\n",
 253                                (unsigned long)cfid->mds, (unsigned long)cfid->id,
 254                                (unsigned long)cfid->generation, rc);
 255
 256                         rc = lmv_revalidate_slaves(exp, reqp, cfid, it, rc,
 257                                                    cb_blocking);
 258                 }
 259
 260                 RETURN(rc);
 261         }
 262
 263         CDEBUG(D_OTHER, "INTENT getattr for %*s on %lu/%lu/%lu\n",
 264                len, name, (unsigned long)pfid->mds, (unsigned long)pfid->id,
 265                (unsigned long)pfid->generation);
 266
 267         mds = pfid->mds;
 268         obj = lmv_grab_obj(obd, pfid);
 269         if (obj && len) {
 270                 /* directory is already splitted. calculate mds */
 271                 mds = raw_name2idx(obj->objcount, (char *) name, len);
 272                 rpfid = obj->objs[mds].fid;
 273                 lmv_put_obj(obj);
 274
 275                 CDEBUG(D_OTHER, "forward to MDS #%u (slave %lu/%lu/%lu)\n",
 276                        mds, (unsigned long)rpfid.mds, (unsigned long)rpfid.id,
 277                        (unsigned long)rpfid.generation);
 278         }
 279
 280         rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
 281                             len, lmm, lmmsize, NULL, it, flags, reqp,
 282                             cb_blocking);
 283
 284         if (rc < 0)
 285                 RETURN(rc);
 286
 287         LASSERT(rc == 0);
 288
 289         /* okay, MDS has returned success. probably name has been
 290          * resolved in remote inode */
 291         rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it,
 292                                      flags, reqp, cb_blocking);
 293         if (rc < 0)
 294                 RETURN(rc);
 295
 296         body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
 297         LASSERT(body != NULL);
 298
 299         cfid = &body->fid1;
 300         obj2 = lmv_grab_obj(obd, cfid);
 301
 302         if (!obj2 && (mea = body_of_splitted_dir(*reqp, 1))) {
 303                 /* wow! this is splitted dir, we'd like to handle it. */
 304                 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
 305                 LASSERT(body != NULL);
 306
 307                 obj2 = lmv_create_obj(exp, &body->fid1, mea);
 308                 if (IS_ERR(obj2))
 309                         RETURN(PTR_ERR(obj2));
 310         }
 311
 312         if (obj2) {
 313                 /* this is splitted dir and we'd want to get attrs */
 314                 CDEBUG(D_OTHER, "attrs from slaves for %lu/%lu/%lu, rc %d\n",
 315                        (unsigned long)cfid->mds, (unsigned long)cfid->id,
 316                        (unsigned long)cfid->generation, rc);
 317
 318                 rc = lmv_revalidate_slaves(exp, reqp, cfid, it, 1, cb_blocking);
 319                 lmv_put_obj(obj2);
 320         }
 321         RETURN(rc);
 322 }
 323
 324 void lmv_update_body_from_obj(struct mds_body *body, struct lmv_inode *obj)
 325 {
 326         /* update size */
 327         body->size += obj->size;
 328
 329         /* update atime */
 330         /* update ctime */
 331         /* update mtime */
 332         /* update nlink */
 333 }
 334
 335 int lmv_lookup_slaves(struct obd_export *exp, struct ptlrpc_request **reqp)
 336 {
 337         struct obd_device *obd = exp->exp_obd;
 338         struct lmv_obd *lmv = &obd->u.lmv;
 339         struct mds_body *body = NULL;
 340         struct lustre_handle *lockh;
 341         struct ldlm_lock *lock;
 342         struct mds_body *body2;
 343         struct ll_uctxt uctxt;
 344         struct lmv_obj *obj;
 345         int i, rc = 0;
 346         ENTRY;
 347
 348         LASSERT(reqp);
 349         LASSERT(*reqp);
 350
 351         /* master is locked. we'd like to take locks on slaves and update
 352          * attributes to be returned from the slaves it's important that lookup
 353          * is called in two cases:
 354
 355          *  - for first time (dcache has no such a resolving yet).
 356          *  - ->d_revalidate() returned false.
 357
 358          * last case possible only if all the objs (master and all slaves aren't
 359          * valid */
 360
 361         body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
 362         LASSERT(body != NULL);
 363
 364         obj = lmv_grab_obj(obd, &body->fid1);
 365         LASSERT(obj != NULL);
 366
 367         CDEBUG(D_OTHER, "lookup slaves for %lu/%lu/%lu\n",
 368                (unsigned long)body->fid1.mds,
 369                (unsigned long)body->fid1.id,
 370                (unsigned long)body->fid1.generation);
 371
 372         uctxt.gid1 = 0;
 373         uctxt.gid2 = 0;
 374
 375         lmv_lock_obj(obj);
 376
 377         for (i = 0; i < obj->objcount; i++) {
 378                 struct ll_fid fid = obj->objs[i].fid;
 379                 struct ptlrpc_request *req = NULL;
 380                 struct lookup_intent it;
 381
 382                 if (fid_equal(&fid, &obj->fid))
 383                         /* skip master obj */
 384                         continue;
 385
 386                 CDEBUG(D_OTHER, "lookup slave %lu/%lu/%lu\n",
 387                        (unsigned long)fid.mds, (unsigned long)fid.id,
 388                        (unsigned long)fid.generation);
 389
 390                 /* is obj valid? */
 391                 memset(&it, 0, sizeof(it));
 392                 it.it_op = IT_GETATTR;
 393                 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
 394                                     NULL, 0, NULL, 0, &fid, &it, 0, &req,
 395                                     lmv_dirobj_blocking_ast);
 396
 397                 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
 398                 if (rc > 0) {
 399                         /* nice, this slave is valid */
 400                         LASSERT(req == NULL);
 401                         CDEBUG(D_OTHER, "cached\n");
 402                         goto release_lock;
 403                 }
 404
 405                 if (rc < 0)
 406                         /* error during revalidation */
 407                         GOTO(cleanup, rc);
 408
 409                 /* rc == 0, this means we have no such a lock and can't think
 410                  * obj is still valid. lookup it again */
 411                 LASSERT(req == NULL);
 412                 req = NULL;
 413
 414                 memset(&it, 0, sizeof(it));
 415                 it.it_op = IT_GETATTR;
 416                 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
 417                                     NULL, 0, NULL, 0, NULL, &it, 0, &req,
 418                                     lmv_dirobj_blocking_ast);
 419
 420                 lockh = (struct lustre_handle *) &it.d.lustre.it_lock_handle;
 421                 LASSERT(rc <= 0);
 422
 423                 if (rc < 0)
 424                         /* error during lookup */
 425                         GOTO(cleanup, rc);
 426
 427                 lock = ldlm_handle2lock(lockh);
 428                 LASSERT(lock);
 429
 430                 lock->l_ast_data = lmv_get_obj(obj);
 431
 432                 body2 = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body2));
 433                 LASSERT(body2);
 434
 435                 obj->objs[i].size = body2->size;
 436
 437                 CDEBUG(D_OTHER, "fresh: %lu\n",
 438                        (unsigned long)obj->objs[i].size);
 439
 440                 LDLM_LOCK_PUT(lock);
 441
 442                 if (req)
 443                         ptlrpc_req_finished(req);
 444 release_lock:
 445                 lmv_update_body_from_obj(body, obj->objs + i);
 446
 447                 if (it.d.lustre.it_lock_mode)
 448                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
 449         }
 450 cleanup:
 451         lmv_unlock_obj(obj);
 452         lmv_put_obj(obj);
 453         RETURN(rc);
 454 }
 455
 456 int lmv_intent_lookup(struct obd_export *exp, struct ll_uctxt *uctxt,
 457                       struct ll_fid *pfid, const char *name, int len,
 458                       void *lmm, int lmmsize, struct ll_fid *cfid,
 459                       struct lookup_intent *it, int flags,
 460                       struct ptlrpc_request **reqp,
 461                       ldlm_blocking_callback cb_blocking)
 462 {
 463         struct obd_device *obd = exp->exp_obd;
 464         struct lmv_obd *lmv = &obd->u.lmv;
 465         struct mds_body *body = NULL;
 466         struct ll_fid rpfid = *pfid;
 467         struct lmv_obj *obj;
 468         struct mea *mea;
 469         int rc, mds;
 470         ENTRY;
 471
 472         /* IT_LOOKUP is intended to produce name -> fid resolving (let's call
 473          * this lookup below) or to confirm requested resolving is still valid
 474          * (let's call this revalidation) cfid != NULL specifies revalidation */
 475
 476         if (cfid) {
 477                 /* this is revalidation: we have to check is LOOKUP lock still
 478                  * valid for given fid. very important part is that we have to
 479                  * choose right mds because namespace is per mds */
 480                 rpfid = *pfid;
 481                 obj = lmv_grab_obj(obd, pfid);
 482                 if (obj) {
 483                         mds = raw_name2idx(obj->objcount, (char *) name, len);
 484                         rpfid = obj->objs[mds].fid;
 485                         lmv_put_obj(obj);
 486                 }
 487                 mds = rpfid.mds;
 488
 489                 CDEBUG(D_OTHER, "revalidate lookup for %lu/%lu/%lu to %d MDS\n",
 490                        (unsigned long)cfid->mds, (unsigned long)cfid->id,
 491                        (unsigned long)cfid->generation, mds);
 492
 493                 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, pfid, name,
 494                                     len, lmm, lmmsize, cfid, it, flags,
 495                                     reqp, cb_blocking);
 496                 RETURN(rc);
 497         }
 498
 499         mds = pfid->mds;
 500 repeat:
 501         /* this is lookup. during lookup we have to update all the attributes,
 502          * because returned values will be put in struct inode */
 503
 504         obj = lmv_grab_obj(obd, pfid);
 505         if (obj && len) {
 506                 /* directory is already splitted. calculate mds */
 507                 mds = raw_name2idx(obj->objcount, (char *)name, len);
 508                 rpfid = obj->objs[mds].fid;
 509                 lmv_put_obj(obj);
 510         }
 511
 512         rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
 513                             len, lmm, lmmsize, NULL, it, flags, reqp,
 514                             cb_blocking);
 515         if (rc > 0) {
 516                 /* very interesting. it seems object is still valid but for some
 517                  * reason llite calls lookup, not revalidate */
 518                 CWARN("lookup for %lu/%lu/%lu and data should be uptodate\n",
 519                       (unsigned long)rpfid.mds, (unsigned long)rpfid.id,
 520                       (unsigned long)rpfid.generation);
 521
 522                 LASSERT(*reqp == NULL);
 523                 RETURN(rc);
 524         }
 525
 526         if (rc == 0 && *reqp == NULL) {
 527                 /* once again, we're asked for lookup, not revalidate */
 528                 CWARN("lookup for %lu/%lu/%lu and data should be uptodate\n",
 529                        (unsigned long)rpfid.mds, (unsigned long)rpfid.id,
 530                        (unsigned long)rpfid.generation);
 531                 RETURN(rc);
 532         }
 533
 534         if (rc == -ERESTART) {
 535                 /* directory got splitted since last update. this shouldn't be
 536                  * becasue splitting causes lock revocation, so revalidate had
 537                  * to fail and lookup on dir had to return mea */
 538                 CWARN("we haven't knew about directory splitting!\n");
 539                 LASSERT(obj == NULL);
 540
 541                 obj = lmv_create_obj(exp, &rpfid, NULL);
 542                 if (IS_ERR(obj))
 543                         RETURN(PTR_ERR(obj));
 544
 545                 goto repeat;
 546         }
 547
 548         if (rc < 0)
 549                 RETURN(rc);
 550
 551         /* okay, MDS has returned success. probably name has been resolved in
 552          * remote inode */
 553         rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it, flags,
 554                                      reqp, cb_blocking);
 555
 556         if (rc == 0 && (mea = body_of_splitted_dir(*reqp, 1))) {
 557                 /* wow! this is splitted dir, we'd like to handle it */
 558                 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
 559                 LASSERT(body != NULL);
 560
 561                 obj = lmv_grab_obj(obd, &body->fid1);
 562                 if (!obj) {
 563                         obj = lmv_create_obj(exp, &body->fid1, mea);
 564                         if (IS_ERR(obj))
 565                                 RETURN(PTR_ERR(obj));
 566                 } else {
 567                         lmv_put_obj(obj);
 568                 }
 569         }
 570
 571         RETURN(rc);
 572 }
 573
 574 int lmv_intent_lock(struct obd_export *exp, struct ll_uctxt *uctxt,
 575                     struct ll_fid *pfid, const char *name, int len,
 576                     void *lmm, int lmmsize, struct ll_fid *cfid,
 577                     struct lookup_intent *it, int flags,
 578                     struct ptlrpc_request **reqp,
 579                     ldlm_blocking_callback cb_blocking)
 580 {
 581         struct obd_device *obd = exp->exp_obd;
 582         int rc = 0;
 583         ENTRY;
 584
 585         LASSERT(it);
 586         LASSERT(pfid);
 587
 588         CDEBUG(D_OTHER, "INTENT LOCK '%s' for '%*s' on %lu/%lu -> %u\n",
 589                         LL_IT2STR(it), len, name, (unsigned long) pfid->id,
 590                         (unsigned long) pfid->generation, pfid->mds);
 591
 592         rc = lmv_check_connect(obd);
 593         if (rc)
 594                 RETURN(rc);
 595
 596         if (it->it_op == IT_LOOKUP)
 597                 rc = lmv_intent_lookup(exp, uctxt, pfid, name, len, lmm,
 598                                        lmmsize, cfid, it, flags, reqp,
 599                                        cb_blocking);
 600         else if (it->it_op & IT_OPEN)
 601                 rc = lmv_intent_open(exp, uctxt, pfid, name, len, lmm,
 602                                      lmmsize, cfid, it, flags, reqp,
 603                                      cb_blocking);
 604         else if (it->it_op == IT_GETATTR || it->it_op == IT_CHDIR)
 605                 rc = lmv_intent_getattr(exp, uctxt, pfid, name, len, lmm,
 606                                         lmmsize, cfid, it, flags, reqp,
 607                                         cb_blocking);
 608         else
 609                 LBUG();
 610         RETURN(rc);
 611 }
 612
 613 int lmv_revalidate_slaves(struct obd_export *exp, struct ptlrpc_request **reqp,
 614                           struct ll_fid *mfid, struct lookup_intent *oit,
 615                           int master_valid, ldlm_blocking_callback cb_blocking)
 616 {
 617         struct obd_device *obd = exp->exp_obd;
 618         struct ptlrpc_request *mreq = *reqp;
 619         struct lmv_obd *lmv = &obd->u.lmv;
 620         struct lustre_handle master_lockh;
 621         struct ldlm_lock *lock;
 622         unsigned long size = 0;
 623         struct mds_body *body;
 624         struct ll_uctxt uctxt;
 625         struct lmv_obj *obj;
 626         int master_lock_mode;
 627         int i, rc = 0;
 628         ENTRY;
 629
 630         /* we have to loop over the subobjects, check validity and update them
 631          * from MDSs if needed. it's very useful that we need not to update all
 632          * the fields. say, common fields (that are equal on all the subojects
 633          * need not to be update, another fields (i_size, for example) are
 634          * cached all the time */
 635         obj = lmv_grab_obj(obd, mfid);
 636         LASSERT(obj != NULL);
 637
 638         uctxt.gid1 = 0;
 639         uctxt.gid2 = 0;
 640         master_lock_mode = 0;
 641
 642         lmv_lock_obj(obj);
 643
 644         for (i = 0; i < obj->objcount; i++) {
 645                 struct ll_fid fid = obj->objs[i].fid;
 646                 struct lustre_handle *lockh = NULL;
 647                 struct ptlrpc_request *req = NULL;
 648                 ldlm_blocking_callback cb;
 649                 struct lookup_intent it;
 650                 int master = 0;
 651
 652                 CDEBUG(D_OTHER, "revalidate subobj %lu/%lu/%lu\n",
 653                        (unsigned long)fid.mds, (unsigned long)fid.id,
 654                        (unsigned long) fid.generation);
 655
 656                 memset(&it, 0, sizeof(it));
 657                 it.it_op = IT_GETATTR;
 658                 cb = lmv_dirobj_blocking_ast;
 659
 660                 if (fid_equal(&fid, &obj->fid)) {
 661                         if (master_valid) {
 662                                 /* lmv_intent_getattr() already checked
 663                                  * validness and took the lock */
 664                                 if (mreq) {
 665                                         /* it even got the reply refresh attrs
 666                                          * from that reply */
 667                                         body = lustre_msg_buf(mreq->rq_repmsg,
 668                                                               1, sizeof(*body));
 669                                         LASSERT(body != NULL);
 670                                         goto update;
 671                                 }
 672                                 /* take already cached attrs into account */
 673                                 CDEBUG(D_OTHER,
 674                                        "master is locked and cached\n");
 675                                 goto release_lock;
 676                         }
 677                         master = 1;
 678                         cb = cb_blocking;
 679                 }
 680
 681                 /* is obj valid? */
 682                 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
 683                                     NULL, 0, NULL, 0, &fid, &it, 0, &req, cb);
 684                 lockh = (struct lustre_handle *) &it.d.lustre.it_lock_handle;
 685                 if (rc > 0) {
 686                         /* nice, this slave is valid */
 687                         LASSERT(req == NULL);
 688                         CDEBUG(D_OTHER, "cached\n");
 689                         goto release_lock;
 690                 }
 691
 692                 if (rc < 0)
 693                         /* error during revalidation */
 694                         GOTO(cleanup, rc);
 695
 696                 /* rc == 0, this means we have no such a lock and can't think
 697                  * obj is still valid. lookup it again */
 698                 LASSERT(req == NULL);
 699                 req = NULL;
 700
 701                 memset(&it, 0, sizeof(it));
 702                 it.it_op = IT_GETATTR;
 703                 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
 704                                     NULL, 0, NULL, 0, NULL, &it, 0, &req, cb);
 705                 lockh = (struct lustre_handle *) &it.d.lustre.it_lock_handle;
 706                 LASSERT(rc <= 0);
 707
 708                 if (rc < 0)
 709                         /* error during lookup */
 710                         GOTO(cleanup, rc);
 711
 712                 if (master) {
 713                         LASSERT(master_valid == 0);
 714                         /* save lock on master to be returned to the caller */
 715                         CDEBUG(D_OTHER, "no lock on master yet\n");
 716                         memcpy(&master_lockh, lockh, sizeof(master_lockh));
 717                         master_lock_mode = it.d.lustre.it_lock_mode;
 718                         it.d.lustre.it_lock_mode = 0;
 719                 } else {
 720                         /* this is slave. we want to control it */
 721                         lock = ldlm_handle2lock(lockh);
 722                         LASSERT(lock);
 723                         lock->l_ast_data = lmv_get_obj(obj);
 724                         LDLM_LOCK_PUT(lock);
 725                 }
 726
 727                 if (*reqp == NULL) {
 728                         /* this is first reply, we'll use it to return
 729                          * updated data back to the caller */
 730                         LASSERT(req);
 731                         ptlrpc_request_addref(req);
 732                         *reqp = req;
 733
 734                 }
 735
 736                 body = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body));
 737                 LASSERT(body);
 738
 739 update:
 740                 obj->objs[i].size = body->size;
 741
 742                 CDEBUG(D_OTHER, "fresh: %lu\n",
 743                        (unsigned long)obj->objs[i].size);
 744
 745                 if (req)
 746                         ptlrpc_req_finished(req);
 747 release_lock:
 748                 size += obj->objs[i].size;
 749
 750                 if (it.d.lustre.it_lock_mode)
 751                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
 752         }
 753
 754         if (*reqp) {
 755                 /* some attrs got refreshed, we have reply and it's time to put
 756                  * fresh attrs to it */
 757                 CDEBUG(D_OTHER, "return refreshed attrs: size = %lu\n",
 758                        (unsigned long)size);
 759
 760                 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
 761                 LASSERT(body);
 762
 763                 /* FIXME: what about other attributes? */
 764                 body->size = size;
 765
 766                 if (mreq == NULL) {
 767                         /* very important to maintain lli->mds the same because
 768                          * of revalidation. mreq == NULL means that caller has
 769                          * no reply and the only attr we can return is size */
 770                         body->valid = OBD_MD_FLSIZE;
 771                         body->mds = obj->fid.mds;
 772                 }
 773                 if (master_valid == 0) {
 774                         memcpy(&oit->d.lustre.it_lock_handle,
 775                                &master_lockh, sizeof(master_lockh));
 776                         oit->d.lustre.it_lock_mode = master_lock_mode;
 777                 }
 778                 rc = 0;
 779         } else {
 780                 /* it seems all the attrs are fresh and we did no request */
 781                 CDEBUG(D_OTHER, "all the attrs were fresh\n");
 782                 if (master_valid == 0)
 783                         oit->d.lustre.it_lock_mode = master_lock_mode;
 784                 rc = 1;
 785         }
 786 cleanup:
 787         lmv_unlock_obj(obj);
 788         lmv_put_obj(obj);
 789         RETURN(rc);
 790 }