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