1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Copyright (C) 2002, 2003, 2004, 2005, 2006 Cluster File Systems, Inc.
6 * This file is part of Lustre, http://www.lustre.org.
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.
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.
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.
23 # define EXPORT_SYMTAB
25 #define DEBUG_SUBSYSTEM S_LMV
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>
39 #include <liblustre.h>
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"
51 static inline void lmv_drop_intent_lock(struct lookup_intent *it)
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;
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,
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;
76 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_DLM_REP);
77 LASSERT(body != NULL);
79 if (!(body->valid & OBD_MD_MDS))
83 * oh, MDS reports that this is remote inode case i.e. we have to ask
84 * for real attrs on another MDS.
86 if (it->it_op & IT_LOOKUP) {
88 * unfortunately, we have to lie to MDC/MDS to retrieve
89 * attributes llite needs.
91 it->it_op = IT_GETATTR;
94 /* we got LOOKUP lock, but we really need attrs */
95 pmode = it->d.lustre.it_lock_mode;
97 plock.cookie = it->d.lustre.it_lock_handle;
98 it->d.lustre.it_lock_mode = 0;
99 it->d.lustre.it_data = 0;
102 LASSERT(fid_is_sane(&body->fid1));
104 it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
106 tgt_exp = lmv_find_export(lmv, &body->fid1);
108 GOTO(out, rc = PTR_ERR(tgt_exp));
110 OBD_ALLOC_PTR(op_data);
112 GOTO(out, rc = -ENOMEM);
114 op_data->op_fid1 = body->fid1;
115 op_data->op_bias = MDS_CROSS_REF;
117 rc = md_intent_lock(tgt_exp, op_data, lmm, lmmsize, it, flags,
118 &req, cb_blocking, extra_lock_flags);
121 * llite needs LOOKUP lock to track dentry revocation in order to
122 * maintain dcache consistency. Thus drop UPDATE lock here and put
126 lmv_drop_intent_lock(it);
127 it->d.lustre.it_lock_handle = plock.cookie;
128 it->d.lustre.it_lock_mode = pmode;
131 OBD_FREE_PTR(op_data);
135 ldlm_lock_decref(&plock, pmode);
137 ptlrpc_req_finished(*reqp);
142 int lmv_alloc_slave_fids(struct obd_device *obd, struct lu_fid *pid,
143 struct md_op_data *op, struct lu_fid *fid)
145 struct lmv_obd *lmv = &obd->u.lmv;
152 obj = lmv_obj_grab(obd, pid);
154 CERROR("Object "DFID" should be split\n",
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;
164 rc = __lmv_fid_alloc(lmv, fid, mds);
166 CERROR("Can't allocate new fid, rc %d\n",
171 CDEBUG(D_INFO, "Allocate new fid "DFID" for split "
178 * IT_OPEN is intended to open (and create, possible) an object. Parent (pid)
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)
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;
198 OBD_ALLOC_PTR(sop_data);
199 if (sop_data == NULL)
202 /* save op_data fro repeat case */
203 *sop_data = *op_data;
209 obj = lmv_obj_grab(obd, &rpid);
214 * Directory is already split, so we have to forward request to
217 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
218 (char *)op_data->op_name,
219 op_data->op_namelen);
221 rpid = obj->lo_inodes[mea_idx].li_fid;
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;
227 CDEBUG(D_OTHER, "Choose slave dir ("DFID")\n", PFID(&rpid));
229 struct lmv_tgt_desc *tgt;
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;
237 GOTO(out_free_sop_data, rc = PTR_ERR(tgt_exp));
239 sop_data->op_fid1 = rpid;
241 if (it->it_op & IT_CREAT) {
243 * For open with IT_CREATE and for IT_CREATE cases allocate new
244 * fid and setup FLD for it.
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);
250 GOTO(out_free_sop_data, rc);
255 GOTO(out_free_sop_data, rc);
258 rc = md_intent_lock(tgt_exp, sop_data, lmm, lmmsize, it, flags,
259 reqp, cb_blocking, extra_lock_flags);
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);
267 it->d.lustre.it_data = 0;
270 * Directory got split. Time to update local object and repeat
271 * the request with proper MDS.
273 LASSERT(lu_fid_eq(&op_data->op_fid1, &rpid));
274 rc = lmv_handle_split(exp, &rpid);
276 /* We should reallocate child FID. */
277 rc = lmv_alloc_slave_fids(obd, &rpid, op_data,
285 GOTO(out_free_sop_data, rc);
288 * Okay, MDS has returned success. Probably name has been resolved in
291 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
292 cb_blocking, extra_lock_flags);
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,
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);
308 * Nothing is found, do not access body->fid1 as it is zero and thus
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);
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);
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);
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);
331 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
335 /* This is split dir and we'd want to get attrs. */
336 CDEBUG(D_OTHER, "attrs from slaves for "DFID"\n",
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",
351 OBD_FREE_PTR(sop_data);
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)
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;
372 OBD_ALLOC_PTR(sop_data);
373 if (sop_data == NULL)
376 /* save op_data fro repeat case */
377 *sop_data = *op_data;
379 if (fid_is_sane(&op_data->op_fid2)) {
381 * Caller wants to revalidate attrs of obj we have to revalidate
382 * slaves if requested object is split directory.
384 CDEBUG(D_OTHER, "revalidate attrs for "DFID"\n",
385 PFID(&op_data->op_fid2));
387 rc = lmv_fld_lookup(lmv, &op_data->op_fid2, &mds);
389 GOTO(out_free_sop_data, rc);
392 * In fact, we do not need this with current intent_lock(), but
393 * it may change some day.
395 obj = lmv_obj_grab(obd, &op_data->op_fid2);
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;
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));
409 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
411 GOTO(out_free_sop_data, rc);
412 obj = lmv_obj_grab(obd, &op_data->op_fid1);
413 if (obj && op_data->op_namelen) {
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;
425 CDEBUG(D_OTHER, "forward to MDS #"LPU64" (slave "DFID")\n",
428 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
430 GOTO(out_free_sop_data, rc);
431 sop_data->op_bias |= MDS_CHECK_SPLIT;
435 sop_data->op_fid1 = rpid;
437 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm,
438 lmmsize, it, flags, reqp, cb_blocking,
441 LASSERTF(rc != -ERESTART, "GETATTR: Got unhandled -ERESTART!\n");
443 GOTO(out_free_sop_data, rc);
447 * This is split dir. In order to optimize things a bit, we
448 * consider obj valid updating missing parts.
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.
455 "revalidate slaves for "DFID", rc %d\n",
456 PFID(&op_data->op_fid2), rc);
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);
465 GOTO(out_free_sop_data, rc);
468 * okay, MDS has returned success. Probably name has been resolved in
471 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags,
472 reqp, cb_blocking, extra_lock_flags);
474 GOTO(out_free_sop_data, rc);
477 * Nothing is found, do not access body->fid1 as it is zero and thus
480 if (it->d.lustre.it_disposition & DISP_LOOKUP_NEG)
481 GOTO(out_free_sop_data, rc = 0);
484 LASSERT((*reqp)->rq_repmsg);
485 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
486 LASSERT(body != NULL);
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);
492 obj2 = lmv_obj_grab(obd, &body->fid1);
494 if (!obj2 && (mea = lmv_get_mea(*reqp))) {
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);
500 GOTO(out_free_sop_data, rc = (int)PTR_ERR(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);
508 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
509 cb_blocking, extra_lock_flags);
515 OBD_FREE_PTR(sop_data);
519 /* this is not used currently */
520 int lmv_lookup_slaves(struct obd_export *exp, struct ptlrpc_request **reqp)
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;
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:
541 * - for first time (dcache has no such a resolving yet). -
542 * ->d_revalidate() returned false.
544 * Last case possible only if all the objs (master and all slaves aren't
548 OBD_ALLOC_PTR(op_data);
552 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
553 LASSERT(body != NULL);
555 LASSERT((body->valid & OBD_MD_FLID) != 0);
556 obj = lmv_obj_grab(obd, &body->fid1);
557 LASSERT(obj != NULL);
559 CDEBUG(D_OTHER, "lookup slaves for "DFID"\n",
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;
570 if (lu_fid_eq(&fid, &obj->lo_fid))
571 /* skip master obj */
574 CDEBUG(D_OTHER, "lookup slave "DFID"\n", PFID(&fid));
577 memset(&it, 0, sizeof(it));
578 it.it_op = IT_GETATTR;
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;
585 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
587 GOTO(cleanup, rc = PTR_ERR(tgt_exp));
589 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0,
590 &req, lmv_blocking_ast, 0);
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");
601 /* error during lookup */
604 lock = ldlm_handle2lock(lockh);
607 lock->l_ast_data = lmv_obj_get(obj);
609 body2 = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
610 LASSERT(body2 != NULL);
612 obj->lo_inodes[i].li_size = body2->size;
614 CDEBUG(D_OTHER, "fresh: %lu\n",
615 (unsigned long)obj->lo_inodes[i].li_size);
620 ptlrpc_req_finished(req);
622 lmv_update_body(body, obj->lo_inodes + i);
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;
634 OBD_FREE_PTR(op_data);
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)
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;
656 OBD_ALLOC_PTR(sop_data);
657 if (sop_data == NULL)
660 /* save op_data fro repeat case */
661 *sop_data = *op_data;
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
669 if (fid_is_sane(&op_data->op_fid2)) {
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.
675 rpid = op_data->op_fid1;
676 obj = lmv_obj_grab(obd, &rpid);
678 mea_idx = raw_name2idx(obj->lo_hashtype,
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;
687 rc = lmv_fld_lookup(lmv, &rpid, &mds);
689 GOTO(out_free_sop_data, rc);
690 sop_data->op_bias |= MDS_CHECK_SPLIT;
693 CDEBUG(D_OTHER, "revalidate lookup for "DFID" to #"LPU64" MDS\n",
694 PFID(&op_data->op_fid2), mds);
701 * This is lookup. During lookup we have to update all the
702 * attributes, because returned values will be put in struct
705 obj = lmv_obj_grab(obd, &op_data->op_fid1);
707 if (op_data->op_namelen) {
708 /* directory is already split. calculate mds */
709 mea_idx = raw_name2idx(obj->lo_hashtype,
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;
716 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
719 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
721 GOTO(out_free_sop_data, rc);
722 sop_data->op_bias |= MDS_CHECK_SPLIT;
724 fid_zero(&sop_data->op_fid2);
727 sop_data->op_bias &= ~MDS_CROSS_REF;
728 sop_data->op_fid1 = rpid;
730 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm, lmmsize,
731 it, flags, reqp, cb_blocking, extra_lock_flags);
733 LASSERT(fid_is_sane(&op_data->op_fid2));
735 * Very interesting. it seems object is still valid but for some
736 * reason llite calls lookup, not revalidate.
738 CDEBUG(D_OTHER, "lookup for "DFID" and data should be uptodate\n",
740 LASSERT(*reqp == NULL);
741 GOTO(out_free_sop_data, rc);
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",
748 GOTO(out_free_sop_data, rc);
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);
757 it->d.lustre.it_data = 0;
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.
763 CWARN("we haven't knew about directory splitting!\n");
764 LASSERT(obj == NULL);
766 obj = lmv_obj_create(exp, &rpid, NULL);
768 GOTO(out_free_sop_data, rc = PTR_ERR(obj));
774 GOTO(out_free_sop_data, rc);
777 * Okay, MDS has returned success. Probably name has been resolved in
780 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
781 cb_blocking, extra_lock_flags);
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);
789 obj = lmv_obj_grab(obd, &body->fid1);
791 obj = lmv_obj_create(exp, &body->fid1, mea);
793 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
800 OBD_FREE_PTR(sop_data);
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)
810 struct obd_device *obd = exp->exp_obd;
815 LASSERT(fid_is_sane(&op_data->op_fid1));
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));
821 rc = lmv_check_connect(obd);
825 if (it->it_op & IT_LOOKUP)
826 rc = lmv_intent_lookup(exp, op_data, lmm, lmmsize, it,
827 flags, reqp, cb_blocking,
829 else if (it->it_op & IT_OPEN)
830 rc = lmv_intent_open(exp, op_data, lmm, lmmsize, it,
831 flags, reqp, cb_blocking,
833 else if (it->it_op & IT_GETATTR)
834 rc = lmv_intent_getattr(exp, op_data,lmm, lmmsize, it,
835 flags, reqp, cb_blocking,
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)
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;
857 int master_lock_mode;
861 OBD_ALLOC_PTR(op_data);
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.
872 obj = lmv_obj_grab(obd, mid);
873 LASSERT(obj != NULL);
875 master_lock_mode = 0;
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;
887 CDEBUG(D_OTHER, "revalidate subobj "DFID"\n",
890 memset(op_data, 0, sizeof(*op_data));
891 memset(&it, 0, sizeof(it));
892 it.it_op = IT_GETATTR;
894 cb = lmv_blocking_ast;
896 if (lu_fid_eq(&fid, &obj->lo_fid)) {
899 * lmv_intent_getattr() already checked
900 * validness and took the lock.
904 * It even got the reply refresh attrs
907 body = req_capsule_server_get(
910 LASSERT(body != NULL);
913 /* take already cached attrs into account */
915 "master is locked and cached\n");
922 op_data->op_fid1 = fid;
923 op_data->op_fid2 = fid;
924 op_data->op_bias = MDS_CROSS_REF;
927 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
929 GOTO(cleanup, rc = PTR_ERR(tgt_exp));
931 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0, &req, cb,
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");
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;
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);
962 * This is first reply, we'll use it to return updated
963 * data back to the caller.
966 ptlrpc_request_addref(req);
970 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
971 LASSERT(body != NULL);
974 obj->lo_inodes[i].li_size = body->size;
976 CDEBUG(D_OTHER, "fresh: %lu\n",
977 (unsigned long)obj->lo_inodes[i].li_size);
980 ptlrpc_req_finished(req);
982 size += obj->lo_inodes[i].li_size;
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;
992 * Some attrs got refreshed, we have reply and it's time to put
995 CDEBUG(D_OTHER, "return refreshed attrs: size = %lu\n",
996 (unsigned long)size);
998 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
999 LASSERT(body != NULL);
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.
1009 body->valid = OBD_MD_FLSIZE;
1011 if (master_valid == 0) {
1012 memcpy(&oit->d.lustre.it_lock_handle,
1013 &master_lockh, sizeof(master_lockh));
1014 oit->d.lustre.it_lock_mode = master_lock_mode;
1018 /* It seems all the attrs are fresh and we did no request */
1019 CDEBUG(D_OTHER, "all the attrs were fresh\n");
1020 if (master_valid == 0)
1021 oit->d.lustre.it_lock_mode = master_lock_mode;
1027 OBD_FREE_PTR(op_data);
1028 lmv_obj_unlock(obj);