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