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