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