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