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