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