1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
6 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
8 * This program is free software; you can redistribute it and/or modify
9 * it under the terms of the GNU General Public License version 2 only,
10 * as published by the Free Software Foundation.
12 * This program is distributed in the hope that it will be useful, but
13 * WITHOUT ANY WARRANTY; without even the implied warranty of
14 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
15 * General Public License version 2 for more details (a copy is included
16 * in the LICENSE file that accompanied this code).
18 * You should have received a copy of the GNU General Public License
19 * version 2 along with this program; If not, see
20 * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
22 * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
23 * CA 95054 USA or visit www.sun.com if you need additional information or
29 * Copyright 2008 Sun Microsystems, Inc. All rights reserved
30 * Use is subject to license terms.
33 * This file is part of Lustre, http://www.lustre.org/
34 * Lustre is a trademark of Sun Microsystems, Inc.
38 # define EXPORT_SYMTAB
40 #define DEBUG_SUBSYSTEM S_LMV
42 #include <linux/slab.h>
43 #include <linux/module.h>
44 #include <linux/init.h>
45 #include <linux/slab.h>
46 #include <linux/pagemap.h>
47 #include <asm/div64.h>
48 #include <linux/seq_file.h>
49 #include <linux/namei.h>
50 # ifndef HAVE_VFS_INTENT_PATCHES
51 # include <linux/lustre_intent.h>
54 #include <liblustre.h>
57 #include <lustre/lustre_idl.h>
58 #include <obd_support.h>
59 #include <lustre_lib.h>
60 #include <lustre_net.h>
61 #include <lustre_dlm.h>
62 #include <obd_class.h>
63 #include <lprocfs_status.h>
64 #include "lmv_internal.h"
66 static inline void lmv_drop_intent_lock(struct lookup_intent *it)
68 if (it->d.lustre.it_lock_mode != 0) {
69 ldlm_lock_decref((void *)&it->d.lustre.it_lock_handle,
70 it->d.lustre.it_lock_mode);
71 it->d.lustre.it_lock_mode = 0;
75 int lmv_intent_remote(struct obd_export *exp, void *lmm,
76 int lmmsize, struct lookup_intent *it,
77 int flags, struct ptlrpc_request **reqp,
78 ldlm_blocking_callback cb_blocking,
81 struct obd_device *obd = exp->exp_obd;
82 struct lmv_obd *lmv = &obd->u.lmv;
83 struct ptlrpc_request *req = NULL;
84 struct lustre_handle plock;
85 struct md_op_data *op_data;
86 struct obd_export *tgt_exp;
87 struct mdt_body *body;
91 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_DLM_REP);
92 LASSERT(body != NULL);
94 if (!(body->valid & OBD_MD_MDS))
98 * oh, MDS reports that this is remote inode case i.e. we have to ask
99 * for real attrs on another MDS.
101 if (it->it_op & IT_LOOKUP) {
103 * unfortunately, we have to lie to MDC/MDS to retrieve
104 * attributes llite needs.
106 it->it_op = IT_GETATTR;
109 /* we got LOOKUP lock, but we really need attrs */
110 pmode = it->d.lustre.it_lock_mode;
112 plock.cookie = it->d.lustre.it_lock_handle;
113 it->d.lustre.it_lock_mode = 0;
114 it->d.lustre.it_data = 0;
117 LASSERT(fid_is_sane(&body->fid1));
119 it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
121 tgt_exp = lmv_find_export(lmv, &body->fid1);
123 GOTO(out, rc = PTR_ERR(tgt_exp));
125 OBD_ALLOC_PTR(op_data);
127 GOTO(out, rc = -ENOMEM);
129 op_data->op_fid1 = body->fid1;
130 op_data->op_bias = MDS_CROSS_REF;
132 rc = md_intent_lock(tgt_exp, op_data, lmm, lmmsize, it, flags,
133 &req, cb_blocking, extra_lock_flags);
136 * llite needs LOOKUP lock to track dentry revocation in order to
137 * maintain dcache consistency. Thus drop UPDATE lock here and put
141 lmv_drop_intent_lock(it);
142 it->d.lustre.it_lock_handle = plock.cookie;
143 it->d.lustre.it_lock_mode = pmode;
146 OBD_FREE_PTR(op_data);
150 ldlm_lock_decref(&plock, pmode);
152 ptlrpc_req_finished(*reqp);
157 int lmv_alloc_slave_fids(struct obd_device *obd, struct lu_fid *pid,
158 struct md_op_data *op, struct lu_fid *fid)
160 struct lmv_obd *lmv = &obd->u.lmv;
167 obj = lmv_obj_grab(obd, pid);
169 CERROR("Object "DFID" should be split\n",
174 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
175 (char *)op->op_name, op->op_namelen);
176 mds = obj->lo_inodes[mea_idx].li_mds;
179 rc = __lmv_fid_alloc(lmv, fid, mds);
181 CERROR("Can't allocate new fid, rc %d\n",
186 CDEBUG(D_INFO, "Allocate new fid "DFID" for split "
193 * IT_OPEN is intended to open (and create, possible) an object. Parent (pid)
196 int lmv_intent_open(struct obd_export *exp, struct md_op_data *op_data,
197 void *lmm, int lmmsize, struct lookup_intent *it,
198 int flags, struct ptlrpc_request **reqp,
199 ldlm_blocking_callback cb_blocking,
200 int extra_lock_flags)
202 struct obd_device *obd = exp->exp_obd;
203 struct lu_fid rpid = op_data->op_fid1;
204 struct lmv_obd *lmv = &obd->u.lmv;
205 struct md_op_data *sop_data;
206 struct obd_export *tgt_exp;
207 struct lmv_stripe_md *mea;
208 struct mdt_body *body;
213 OBD_ALLOC_PTR(sop_data);
214 if (sop_data == NULL)
217 /* save op_data fro repeat case */
218 *sop_data = *op_data;
224 obj = lmv_obj_grab(obd, &rpid);
229 * Directory is already split, so we have to forward request to
232 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
233 (char *)op_data->op_name,
234 op_data->op_namelen);
236 rpid = obj->lo_inodes[mea_idx].li_fid;
238 sop_data->op_mds = obj->lo_inodes[mea_idx].li_mds;
239 tgt_exp = lmv_get_export(lmv, sop_data->op_mds);
240 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
242 CDEBUG(D_OTHER, "Choose slave dir ("DFID")\n", PFID(&rpid));
244 struct lmv_tgt_desc *tgt;
246 sop_data->op_bias |= MDS_CHECK_SPLIT;
247 tgt = lmv_find_target(lmv, &rpid);
248 sop_data->op_mds = tgt->ltd_idx;
249 tgt_exp = tgt->ltd_exp;
252 GOTO(out_free_sop_data, rc = PTR_ERR(tgt_exp));
254 sop_data->op_fid1 = rpid;
256 if (it->it_op & IT_CREAT) {
258 * For open with IT_CREATE and for IT_CREATE cases allocate new
259 * fid and setup FLD for it.
261 /* save old child fid for correctly check stale data*/
262 sop_data->op_fid3 = sop_data->op_fid2;
263 rc = lmv_fid_alloc(exp, &sop_data->op_fid2, sop_data);
265 GOTO(out_free_sop_data, rc);
270 GOTO(out_free_sop_data, rc);
273 rc = md_intent_lock(tgt_exp, sop_data, lmm, lmmsize, it, flags,
274 reqp, cb_blocking, extra_lock_flags);
276 if (rc == -ERESTART) {
277 LASSERT(*reqp != NULL);
278 DEBUG_REQ(D_WARNING|D_RPCTRACE, *reqp,
279 "Got -ERESTART during open!\n");
280 ptlrpc_req_finished(*reqp);
282 it->d.lustre.it_data = 0;
285 * Directory got split. Time to update local object and repeat
286 * the request with proper MDS.
288 LASSERT(lu_fid_eq(&op_data->op_fid1, &rpid));
289 rc = lmv_handle_split(exp, &rpid);
291 /* We should reallocate child FID. */
292 rc = lmv_alloc_slave_fids(obd, &rpid, op_data,
300 GOTO(out_free_sop_data, rc);
303 * Okay, MDS has returned success. Probably name has been resolved in
306 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
307 cb_blocking, extra_lock_flags);
311 * This is possible, that some userspace application will try to
312 * open file as directory and we will have -ENOTDIR here. As
313 * this is normal situation, we should not print error here,
316 CDEBUG(D_OTHER, "can't handle remote %s: dir "DFID"("DFID"):"
317 "%*s: %d\n", LL_IT2STR(it), PFID(&op_data->op_fid2),
318 PFID(&rpid), op_data->op_namelen, op_data->op_name, rc);
319 GOTO(out_free_sop_data, rc);
323 * Nothing is found, do not access body->fid1 as it is zero and thus
326 if ((it->d.lustre.it_disposition & DISP_LOOKUP_NEG) &&
327 !(it->d.lustre.it_disposition & DISP_OPEN_CREATE) &&
328 !(it->d.lustre.it_disposition & DISP_OPEN_OPEN))
329 GOTO(out_free_sop_data, rc = 0);
331 /* caller may use attrs MDS returns on IT_OPEN lock request so, we have
332 * to update them for split dir */
333 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_DLM_REP);
334 LASSERT(body != NULL);
336 /* could not find object, FID is not present in response. */
337 if (!(body->valid & OBD_MD_FLID))
338 GOTO(out_free_sop_data, rc = 0);
340 obj = lmv_obj_grab(obd, &body->fid1);
341 if (!obj && (mea = lmv_get_mea(*reqp))) {
342 /* FIXME: capability for remote! */
343 /* wow! this is split dir, we'd like to handle it */
344 obj = lmv_obj_create(exp, &body->fid1, mea);
346 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
350 /* This is split dir and we'd want to get attrs. */
351 CDEBUG(D_OTHER, "attrs from slaves for "DFID"\n",
354 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
355 cb_blocking, extra_lock_flags);
356 } else if (S_ISDIR(body->mode)) {
357 CDEBUG(D_OTHER, "object "DFID" has not lmv obj?\n",
366 OBD_FREE_PTR(sop_data);
370 int lmv_intent_getattr(struct obd_export *exp, struct md_op_data *op_data,
371 void *lmm, int lmmsize, struct lookup_intent *it,
372 int flags, struct ptlrpc_request **reqp,
373 ldlm_blocking_callback cb_blocking,
374 int extra_lock_flags)
376 struct lmv_obj *obj = NULL, *obj2 = NULL;
377 struct obd_device *obd = exp->exp_obd;
378 struct lu_fid rpid = op_data->op_fid1;
379 struct lmv_obd *lmv = &obd->u.lmv;
380 struct md_op_data *sop_data;
381 struct lmv_stripe_md *mea;
382 struct mdt_body *body;
387 OBD_ALLOC_PTR(sop_data);
388 if (sop_data == NULL)
391 /* save op_data fro repeat case */
392 *sop_data = *op_data;
394 if (fid_is_sane(&op_data->op_fid2)) {
396 * Caller wants to revalidate attrs of obj we have to revalidate
397 * slaves if requested object is split directory.
399 CDEBUG(D_OTHER, "revalidate attrs for "DFID"\n",
400 PFID(&op_data->op_fid2));
402 rc = lmv_fld_lookup(lmv, &op_data->op_fid2, &mds);
404 GOTO(out_free_sop_data, rc);
407 * In fact, we do not need this with current intent_lock(), but
408 * it may change some day.
410 obj = lmv_obj_grab(obd, &op_data->op_fid2);
412 if (!lu_fid_eq(&op_data->op_fid1, &op_data->op_fid2)){
413 rpid = obj->lo_inodes[mds].li_fid;
414 mds = obj->lo_inodes[mds].li_mds;
420 CDEBUG(D_OTHER, "INTENT getattr for %*s on "DFID"\n",
421 op_data->op_namelen, op_data->op_name,
422 PFID(&op_data->op_fid1));
424 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
426 GOTO(out_free_sop_data, rc);
427 obj = lmv_obj_grab(obd, &op_data->op_fid1);
428 if (obj && op_data->op_namelen) {
431 /* directory is already split. calculate mds */
432 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
433 (char *)op_data->op_name,
434 op_data->op_namelen);
435 rpid = obj->lo_inodes[mea_idx].li_fid;
436 mds = obj->lo_inodes[mea_idx].li_mds;
437 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
440 CDEBUG(D_OTHER, "forward to MDS #"LPU64" (slave "DFID")\n",
443 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
445 GOTO(out_free_sop_data, rc);
446 sop_data->op_bias |= MDS_CHECK_SPLIT;
450 sop_data->op_fid1 = rpid;
452 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm,
453 lmmsize, it, flags, reqp, cb_blocking,
456 LASSERTF(rc != -ERESTART, "GETATTR: Got unhandled -ERESTART!\n");
458 GOTO(out_free_sop_data, rc);
462 * This is split dir. In order to optimize things a bit, we
463 * consider obj valid updating missing parts.
465 * FIXME: do we need to return any lock here? It would be fine
466 * if we don't. This means that nobody should use UPDATE lock to
467 * notify about object * removal.
470 "revalidate slaves for "DFID", rc %d\n",
471 PFID(&op_data->op_fid2), rc);
473 LASSERT(fid_is_sane(&op_data->op_fid2));
474 rc = lmv_revalidate_slaves(exp, reqp, &op_data->op_fid2, it, rc,
475 cb_blocking, extra_lock_flags);
476 GOTO(out_free_sop_data, rc);
480 GOTO(out_free_sop_data, rc);
483 * okay, MDS has returned success. Probably name has been resolved in
486 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags,
487 reqp, cb_blocking, extra_lock_flags);
489 GOTO(out_free_sop_data, rc);
492 * Nothing is found, do not access body->fid1 as it is zero and thus
495 if (it->d.lustre.it_disposition & DISP_LOOKUP_NEG)
496 GOTO(out_free_sop_data, rc = 0);
499 LASSERT((*reqp)->rq_repmsg);
500 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
501 LASSERT(body != NULL);
503 /* could not find object, FID is not present in response. */
504 if (!(body->valid & OBD_MD_FLID))
505 GOTO(out_free_sop_data, rc = 0);
507 obj2 = lmv_obj_grab(obd, &body->fid1);
509 if (!obj2 && (mea = lmv_get_mea(*reqp))) {
511 /* FIXME remote capability! */
512 /* wow! this is split dir, we'd like to handle it. */
513 obj2 = lmv_obj_create(exp, &body->fid1, mea);
515 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj2));
519 /* this is split dir and we'd want to get attrs */
520 CDEBUG(D_OTHER, "attrs from slaves for "DFID", rc %d\n",
521 PFID(&body->fid1), rc);
523 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
524 cb_blocking, extra_lock_flags);
530 OBD_FREE_PTR(sop_data);
534 /* this is not used currently */
535 int lmv_lookup_slaves(struct obd_export *exp, struct ptlrpc_request **reqp)
537 struct obd_device *obd = exp->exp_obd;
538 struct lmv_obd *lmv = &obd->u.lmv;
539 struct lustre_handle *lockh;
540 struct md_op_data *op_data;
541 struct ldlm_lock *lock;
542 struct mdt_body *body2;
543 struct mdt_body *body;
552 * Master is locked. we'd like to take locks on slaves and update
553 * attributes to be returned from the slaves it's important that lookup
554 * is called in two cases:
556 * - for first time (dcache has no such a resolving yet). -
557 * ->d_revalidate() returned false.
559 * Last case possible only if all the objs (master and all slaves aren't
563 OBD_ALLOC_PTR(op_data);
567 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
568 LASSERT(body != NULL);
570 LASSERT((body->valid & OBD_MD_FLID) != 0);
571 obj = lmv_obj_grab(obd, &body->fid1);
572 LASSERT(obj != NULL);
574 CDEBUG(D_OTHER, "lookup slaves for "DFID"\n",
579 for (i = 0; i < obj->lo_objcount; i++) {
580 struct lu_fid fid = obj->lo_inodes[i].li_fid;
581 struct ptlrpc_request *req = NULL;
582 struct obd_export *tgt_exp;
583 struct lookup_intent it;
585 if (lu_fid_eq(&fid, &obj->lo_fid))
586 /* skip master obj */
589 CDEBUG(D_OTHER, "lookup slave "DFID"\n", PFID(&fid));
592 memset(&it, 0, sizeof(it));
593 it.it_op = IT_GETATTR;
595 memset(op_data, 0, sizeof(*op_data));
596 op_data->op_fid1 = fid;
597 op_data->op_fid2 = fid;
598 op_data->op_bias = MDS_CROSS_REF;
600 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
602 GOTO(cleanup, rc = PTR_ERR(tgt_exp));
604 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0,
605 &req, lmv_blocking_ast, 0);
607 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
608 if (rc > 0 && req == NULL) {
609 /* nice, this slave is valid */
610 LASSERT(req == NULL);
611 CDEBUG(D_OTHER, "cached\n");
616 /* error during lookup */
619 lock = ldlm_handle2lock(lockh);
622 lock->l_ast_data = lmv_obj_get(obj);
624 body2 = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
625 LASSERT(body2 != NULL);
627 obj->lo_inodes[i].li_size = body2->size;
629 CDEBUG(D_OTHER, "fresh: %lu\n",
630 (unsigned long)obj->lo_inodes[i].li_size);
635 ptlrpc_req_finished(req);
637 lmv_update_body(body, obj->lo_inodes + i);
639 if (it.d.lustre.it_lock_mode) {
640 ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
641 it.d.lustre.it_lock_mode = 0;
649 OBD_FREE_PTR(op_data);
653 int lmv_intent_lookup(struct obd_export *exp, struct md_op_data *op_data,
654 void *lmm, int lmmsize, struct lookup_intent *it,
655 int flags, struct ptlrpc_request **reqp,
656 ldlm_blocking_callback cb_blocking,
657 int extra_lock_flags)
659 struct obd_device *obd = exp->exp_obd;
660 struct lu_fid rpid = op_data->op_fid1;
661 struct lmv_obd *lmv = &obd->u.lmv;
662 struct md_op_data *sop_data;
663 struct lmv_stripe_md *mea;
664 struct mdt_body *body;
671 OBD_ALLOC_PTR(sop_data);
672 if (sop_data == NULL)
675 /* save op_data fro repeat case */
676 *sop_data = *op_data;
679 * IT_LOOKUP is intended to produce name -> fid resolving (let's call
680 * this lookup below) or to confirm requested resolving is still valid
681 * (let's call this revalidation) fid_is_sane(&sop_data->op_fid2) specifies
684 if (fid_is_sane(&op_data->op_fid2)) {
686 * This is revalidate: we have to check is LOOKUP lock still
687 * valid for given fid. Very important part is that we have to
688 * choose right mds because namespace is per mds.
690 rpid = op_data->op_fid1;
691 obj = lmv_obj_grab(obd, &rpid);
693 mea_idx = raw_name2idx(obj->lo_hashtype,
695 (char *)op_data->op_name,
696 op_data->op_namelen);
697 rpid = obj->lo_inodes[mea_idx].li_fid;
698 mds = obj->lo_inodes[mea_idx].li_mds;
699 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
702 rc = lmv_fld_lookup(lmv, &rpid, &mds);
704 GOTO(out_free_sop_data, rc);
705 sop_data->op_bias |= MDS_CHECK_SPLIT;
708 CDEBUG(D_OTHER, "revalidate lookup for "DFID" to #"LPU64" MDS\n",
709 PFID(&op_data->op_fid2), mds);
716 * This is lookup. During lookup we have to update all the
717 * attributes, because returned values will be put in struct
720 obj = lmv_obj_grab(obd, &op_data->op_fid1);
722 if (op_data->op_namelen) {
723 /* directory is already split. calculate mds */
724 mea_idx = raw_name2idx(obj->lo_hashtype,
726 (char *)op_data->op_name,
727 op_data->op_namelen);
728 rpid = obj->lo_inodes[mea_idx].li_fid;
729 mds = obj->lo_inodes[mea_idx].li_mds;
731 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
734 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
736 GOTO(out_free_sop_data, rc);
737 sop_data->op_bias |= MDS_CHECK_SPLIT;
739 fid_zero(&sop_data->op_fid2);
742 sop_data->op_bias &= ~MDS_CROSS_REF;
743 sop_data->op_fid1 = rpid;
745 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm, lmmsize,
746 it, flags, reqp, cb_blocking, extra_lock_flags);
748 LASSERT(fid_is_sane(&op_data->op_fid2));
750 * Very interesting. it seems object is still valid but for some
751 * reason llite calls lookup, not revalidate.
753 CDEBUG(D_OTHER, "lookup for "DFID" and data should be uptodate\n",
755 LASSERT(*reqp == NULL);
756 GOTO(out_free_sop_data, rc);
759 if (rc == 0 && *reqp == NULL) {
760 /* once again, we're asked for lookup, not revalidate */
761 CDEBUG(D_OTHER, "lookup for "DFID" and data should be uptodate\n",
763 GOTO(out_free_sop_data, rc);
766 if (rc == -ERESTART) {
767 LASSERT(*reqp != NULL);
768 DEBUG_REQ(D_WARNING|D_RPCTRACE, *reqp,
769 "Got -ERESTART during lookup!\n");
770 ptlrpc_req_finished(*reqp);
772 it->d.lustre.it_data = 0;
774 * Directory got split since last update. This shouldn't be
775 * because splitting causes lock revocation, so revalidate had
776 * to fail and lookup on dir had to return mea.
778 CWARN("we haven't knew about directory splitting!\n");
779 LASSERT(obj == NULL);
781 obj = lmv_obj_create(exp, &rpid, NULL);
783 GOTO(out_free_sop_data, rc = PTR_ERR(obj));
789 GOTO(out_free_sop_data, rc);
792 * Okay, MDS has returned success. Probably name has been resolved in
795 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
796 cb_blocking, extra_lock_flags);
798 if (rc == 0 && (mea = lmv_get_mea(*reqp))) {
799 /* Wow! This is split dir, we'd like to handle it. */
800 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
801 LASSERT(body != NULL);
802 LASSERT((body->valid & OBD_MD_FLID) != 0);
804 obj = lmv_obj_grab(obd, &body->fid1);
806 obj = lmv_obj_create(exp, &body->fid1, mea);
808 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
815 OBD_FREE_PTR(sop_data);
819 int lmv_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
820 void *lmm, int lmmsize, struct lookup_intent *it,
821 int flags, struct ptlrpc_request **reqp,
822 ldlm_blocking_callback cb_blocking,
823 int extra_lock_flags)
825 struct obd_device *obd = exp->exp_obd;
830 LASSERT(fid_is_sane(&op_data->op_fid1));
832 CDEBUG(D_OTHER, "INTENT LOCK '%s' for '%*s' on "DFID"\n",
833 LL_IT2STR(it), op_data->op_namelen, op_data->op_name,
834 PFID(&op_data->op_fid1));
836 rc = lmv_check_connect(obd);
840 if (it->it_op & IT_LOOKUP)
841 rc = lmv_intent_lookup(exp, op_data, lmm, lmmsize, it,
842 flags, reqp, cb_blocking,
844 else if (it->it_op & IT_OPEN)
845 rc = lmv_intent_open(exp, op_data, lmm, lmmsize, it,
846 flags, reqp, cb_blocking,
848 else if (it->it_op & IT_GETATTR)
849 rc = lmv_intent_getattr(exp, op_data,lmm, lmmsize, it,
850 flags, reqp, cb_blocking,
857 int lmv_revalidate_slaves(struct obd_export *exp, struct ptlrpc_request **reqp,
858 const struct lu_fid *mid, struct lookup_intent *oit,
859 int master_valid, ldlm_blocking_callback cb_blocking,
860 int extra_lock_flags)
862 struct obd_device *obd = exp->exp_obd;
863 struct ptlrpc_request *mreq = *reqp;
864 struct lmv_obd *lmv = &obd->u.lmv;
865 struct lustre_handle master_lockh;
866 struct obd_export *tgt_exp;
867 struct md_op_data *op_data;
868 struct ldlm_lock *lock;
869 unsigned long size = 0;
870 struct mdt_body *body;
872 int master_lock_mode;
876 OBD_ALLOC_PTR(op_data);
881 * We have to loop over the subobjects, check validity and update them
882 * from MDSs if needed. it's very useful that we need not to update all
883 * the fields. say, common fields (that are equal on all the subojects
884 * need not to be update, another fields (i_size, for example) are
885 * cached all the time.
887 obj = lmv_obj_grab(obd, mid);
888 LASSERT(obj != NULL);
890 master_lock_mode = 0;
894 for (i = 0; i < obj->lo_objcount; i++) {
895 struct lu_fid fid = obj->lo_inodes[i].li_fid;
896 struct lustre_handle *lockh = NULL;
897 struct ptlrpc_request *req = NULL;
898 ldlm_blocking_callback cb;
899 struct lookup_intent it;
902 CDEBUG(D_OTHER, "revalidate subobj "DFID"\n",
905 memset(op_data, 0, sizeof(*op_data));
906 memset(&it, 0, sizeof(it));
907 it.it_op = IT_GETATTR;
909 cb = lmv_blocking_ast;
911 if (lu_fid_eq(&fid, &obj->lo_fid)) {
914 * lmv_intent_getattr() already checked
915 * validness and took the lock.
919 * It even got the reply refresh attrs
922 body = req_capsule_server_get(
925 LASSERT(body != NULL);
928 /* take already cached attrs into account */
930 "master is locked and cached\n");
937 op_data->op_fid1 = fid;
938 op_data->op_fid2 = fid;
939 op_data->op_bias = MDS_CROSS_REF;
942 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
944 GOTO(cleanup, rc = PTR_ERR(tgt_exp));
946 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0, &req, cb,
949 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
950 if (rc > 0 && req == NULL) {
951 /* Nice, this slave is valid */
952 LASSERT(req == NULL);
953 CDEBUG(D_OTHER, "cached\n");
961 LASSERT(master_valid == 0);
962 /* Save lock on master to be returned to the caller. */
963 CDEBUG(D_OTHER, "no lock on master yet\n");
964 memcpy(&master_lockh, lockh, sizeof(master_lockh));
965 master_lock_mode = it.d.lustre.it_lock_mode;
966 it.d.lustre.it_lock_mode = 0;
968 /* This is slave. We want to control it. */
969 lock = ldlm_handle2lock(lockh);
970 LASSERT(lock != NULL);
971 lock->l_ast_data = lmv_obj_get(obj);
977 * This is first reply, we'll use it to return updated
978 * data back to the caller.
981 ptlrpc_request_addref(req);
985 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
986 LASSERT(body != NULL);
989 obj->lo_inodes[i].li_size = body->size;
991 CDEBUG(D_OTHER, "fresh: %lu\n",
992 (unsigned long)obj->lo_inodes[i].li_size);
995 ptlrpc_req_finished(req);
997 size += obj->lo_inodes[i].li_size;
999 if (it.d.lustre.it_lock_mode) {
1000 ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
1001 it.d.lustre.it_lock_mode = 0;
1007 * Some attrs got refreshed, we have reply and it's time to put
1008 * fresh attrs to it.
1010 CDEBUG(D_OTHER, "return refreshed attrs: size = %lu\n",
1011 (unsigned long)size);
1013 body = req_capsule_server_get(&(*reqp)->rq_pill, &RMF_MDT_BODY);
1014 LASSERT(body != NULL);
1020 * Very important to maintain mds num the same because
1021 * of revalidation. mreq == NULL means that caller has
1022 * no reply and the only attr we can return is size.
1024 body->valid = OBD_MD_FLSIZE;
1026 if (master_valid == 0) {
1027 oit->d.lustre.it_lock_handle = master_lockh.cookie;
1028 oit->d.lustre.it_lock_mode = master_lock_mode;
1032 /* It seems all the attrs are fresh and we did no request */
1033 CDEBUG(D_OTHER, "all the attrs were fresh\n");
1034 if (master_valid == 0)
1035 oit->d.lustre.it_lock_mode = master_lock_mode;
1041 OBD_FREE_PTR(op_data);
1042 lmv_obj_unlock(obj);