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 = lustre_msg_buf((*reqp)->rq_repmsg,
77 DLM_REPLY_REC_OFF, sizeof(*body));
78 LASSERT(body != NULL);
79 LASSERT(lustre_rep_swabbed(*reqp, DLM_REPLY_REC_OFF));
81 if (!(body->valid & OBD_MD_MDS))
85 * oh, MDS reports that this is remote inode case i.e. we have to ask
86 * for real attrs on another MDS.
88 if (it->it_op & IT_LOOKUP) {
90 * unfortunately, we have to lie to MDC/MDS to retrieve
91 * attributes llite needs.
93 it->it_op = IT_GETATTR;
96 /* we got LOOKUP lock, but we really need attrs */
97 pmode = it->d.lustre.it_lock_mode;
99 plock.cookie = it->d.lustre.it_lock_handle;
100 it->d.lustre.it_lock_mode = 0;
101 it->d.lustre.it_data = 0;
104 LASSERT(fid_is_sane(&body->fid1));
106 it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
108 tgt_exp = lmv_find_export(lmv, &body->fid1);
110 GOTO(out, rc = PTR_ERR(tgt_exp));
112 OBD_ALLOC_PTR(op_data);
114 GOTO(out, rc = -ENOMEM);
116 op_data->op_fid1 = body->fid1;
117 op_data->op_bias = MDS_CROSS_REF;
119 rc = md_intent_lock(tgt_exp, op_data, lmm, lmmsize, it, flags,
120 &req, cb_blocking, extra_lock_flags);
123 * llite needs LOOKUP lock to track dentry revocation in order to
124 * maintain dcache consistency. Thus drop UPDATE lock here and put
128 lmv_drop_intent_lock(it);
129 it->d.lustre.it_lock_handle = plock.cookie;
130 it->d.lustre.it_lock_mode = pmode;
133 OBD_FREE_PTR(op_data);
137 ldlm_lock_decref(&plock, pmode);
139 ptlrpc_req_finished(*reqp);
144 int lmv_alloc_slave_fids(struct obd_device *obd, struct lu_fid *pid,
145 struct md_op_data *op, struct lu_fid *fid)
147 struct lmv_obd *lmv = &obd->u.lmv;
154 obj = lmv_obj_grab(obd, pid);
156 CERROR("Object "DFID" should be split\n",
161 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
162 (char *)op->op_name, op->op_namelen);
163 mds = obj->lo_inodes[mea_idx].li_mds;
166 rc = __lmv_fid_alloc(lmv, fid, mds);
168 CERROR("Can't allocate new fid, rc %d\n",
173 CDEBUG(D_INFO, "Allocate new fid "DFID" for split "
180 * IT_OPEN is intended to open (and create, possible) an object. Parent (pid)
183 int lmv_intent_open(struct obd_export *exp, struct md_op_data *op_data,
184 void *lmm, int lmmsize, struct lookup_intent *it,
185 int flags, struct ptlrpc_request **reqp,
186 ldlm_blocking_callback cb_blocking,
187 int extra_lock_flags)
189 struct obd_device *obd = exp->exp_obd;
190 struct lu_fid rpid = op_data->op_fid1;
191 struct lmv_obd *lmv = &obd->u.lmv;
192 struct md_op_data *sop_data;
193 struct obd_export *tgt_exp;
194 struct lmv_stripe_md *mea;
195 struct mdt_body *body;
200 OBD_ALLOC_PTR(sop_data);
201 if (sop_data == NULL)
204 /* save op_data fro repeat case */
205 *sop_data = *op_data;
211 obj = lmv_obj_grab(obd, &rpid);
216 * Directory is already split, so we have to forward request to
219 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
220 (char *)op_data->op_name,
221 op_data->op_namelen);
223 rpid = obj->lo_inodes[mea_idx].li_fid;
225 sop_data->op_mds = obj->lo_inodes[mea_idx].li_mds;
226 tgt_exp = lmv_get_export(lmv, sop_data->op_mds);
227 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
229 CDEBUG(D_OTHER, "Choose slave dir ("DFID")\n", PFID(&rpid));
231 struct lmv_tgt_desc *tgt;
233 sop_data->op_bias |= MDS_CHECK_SPLIT;
234 tgt = lmv_find_target(lmv, &rpid);
235 sop_data->op_mds = tgt->ltd_idx;
236 tgt_exp = tgt->ltd_exp;
239 GOTO(out_free_sop_data, rc = PTR_ERR(tgt_exp));
241 sop_data->op_fid1 = rpid;
243 if (it->it_op & IT_CREAT) {
245 * For open with IT_CREATE and for IT_CREATE cases allocate new
246 * fid and setup FLD for it.
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 = lustre_msg_buf((*reqp)->rq_repmsg,
319 DLM_REPLY_REC_OFF, sizeof(*body));
320 LASSERT(body != NULL);
321 LASSERT(lustre_rep_swabbed(*reqp, DLM_REPLY_REC_OFF));
323 /* could not find object, FID is not present in response. */
324 if (!(body->valid & OBD_MD_FLID))
325 GOTO(out_free_sop_data, rc = 0);
327 obj = lmv_obj_grab(obd, &body->fid1);
328 if (!obj && (mea = lmv_get_mea(*reqp, DLM_REPLY_REC_OFF))) {
329 /* FIXME: capability for remote! */
330 /* wow! this is split dir, we'd like to handle it */
331 obj = lmv_obj_create(exp, &body->fid1, mea);
333 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
337 /* This is split dir and we'd want to get attrs. */
338 CDEBUG(D_OTHER, "attrs from slaves for "DFID"\n",
341 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
342 cb_blocking, extra_lock_flags);
343 } else if (S_ISDIR(body->mode)) {
344 CDEBUG(D_OTHER, "object "DFID" has not lmv obj?\n",
353 OBD_FREE_PTR(sop_data);
357 int lmv_intent_getattr(struct obd_export *exp, struct md_op_data *op_data,
358 void *lmm, int lmmsize, struct lookup_intent *it,
359 int flags, struct ptlrpc_request **reqp,
360 ldlm_blocking_callback cb_blocking,
361 int extra_lock_flags)
363 struct lmv_obj *obj = NULL, *obj2 = NULL;
364 struct obd_device *obd = exp->exp_obd;
365 struct lu_fid rpid = op_data->op_fid1;
366 struct lmv_obd *lmv = &obd->u.lmv;
367 struct md_op_data *sop_data;
368 struct lmv_stripe_md *mea;
369 struct mdt_body *body;
374 OBD_ALLOC_PTR(sop_data);
375 if (sop_data == NULL)
378 /* save op_data fro repeat case */
379 *sop_data = *op_data;
381 if (fid_is_sane(&op_data->op_fid2)) {
383 * Caller wants to revalidate attrs of obj we have to revalidate
384 * slaves if requested object is split directory.
386 CDEBUG(D_OTHER, "revalidate attrs for "DFID"\n",
387 PFID(&op_data->op_fid2));
389 rc = lmv_fld_lookup(lmv, &op_data->op_fid2, &mds);
391 GOTO(out_free_sop_data, rc);
394 * In fact, we do not need this with current intent_lock(), but
395 * it may change some day.
397 obj = lmv_obj_grab(obd, &op_data->op_fid2);
399 if (!lu_fid_eq(&op_data->op_fid1, &op_data->op_fid2)){
400 rpid = obj->lo_inodes[mds].li_fid;
401 mds = obj->lo_inodes[mds].li_mds;
407 CDEBUG(D_OTHER, "INTENT getattr for %*s on "DFID"\n",
408 op_data->op_namelen, op_data->op_name,
409 PFID(&op_data->op_fid1));
411 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
413 GOTO(out_free_sop_data, rc);
414 obj = lmv_obj_grab(obd, &op_data->op_fid1);
415 if (obj && op_data->op_namelen) {
418 /* directory is already split. calculate mds */
419 mea_idx = raw_name2idx(obj->lo_hashtype, obj->lo_objcount,
420 (char *)op_data->op_name,
421 op_data->op_namelen);
422 rpid = obj->lo_inodes[mea_idx].li_fid;
423 mds = obj->lo_inodes[mea_idx].li_mds;
424 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
427 CDEBUG(D_OTHER, "forward to MDS #"LPU64" (slave "DFID")\n",
430 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
432 GOTO(out_free_sop_data, rc);
433 sop_data->op_bias |= MDS_CHECK_SPLIT;
437 sop_data->op_fid1 = rpid;
439 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm,
440 lmmsize, it, flags, reqp, cb_blocking,
443 LASSERTF(rc != -ERESTART, "GETATTR: Got unhandled -ERESTART!\n");
445 GOTO(out_free_sop_data, rc);
449 * This is split dir. In order to optimize things a bit, we
450 * consider obj valid updating missing parts.
452 * FIXME: do we need to return any lock here? It would be fine
453 * if we don't. This means that nobody should use UPDATE lock to
454 * notify about object * removal.
457 "revalidate slaves for "DFID", rc %d\n",
458 PFID(&op_data->op_fid2), rc);
460 LASSERT(fid_is_sane(&op_data->op_fid2));
461 rc = lmv_revalidate_slaves(exp, reqp, &op_data->op_fid2, it, rc,
462 cb_blocking, extra_lock_flags);
463 GOTO(out_free_sop_data, rc);
467 GOTO(out_free_sop_data, rc);
470 * okay, MDS has returned success. Probably name has been resolved in
473 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags,
474 reqp, cb_blocking, extra_lock_flags);
476 GOTO(out_free_sop_data, rc);
479 * Nothing is found, do not access body->fid1 as it is zero and thus
482 if (it->d.lustre.it_disposition & DISP_LOOKUP_NEG)
483 GOTO(out_free_sop_data, rc = 0);
486 LASSERT((*reqp)->rq_repmsg);
487 body = lustre_msg_buf((*reqp)->rq_repmsg,
488 DLM_REPLY_REC_OFF, sizeof(*body));
489 LASSERT(body != NULL);
490 LASSERT(lustre_rep_swabbed(*reqp, DLM_REPLY_REC_OFF));
492 /* could not find object, FID is not present in response. */
493 if (!(body->valid & OBD_MD_FLID))
494 GOTO(out_free_sop_data, rc = 0);
496 obj2 = lmv_obj_grab(obd, &body->fid1);
498 if (!obj2 && (mea = lmv_get_mea(*reqp, DLM_REPLY_REC_OFF))) {
500 /* FIXME remote capability! */
501 /* wow! this is split dir, we'd like to handle it. */
502 obj2 = lmv_obj_create(exp, &body->fid1, mea);
504 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj2));
508 /* this is split dir and we'd want to get attrs */
509 CDEBUG(D_OTHER, "attrs from slaves for "DFID", rc %d\n",
510 PFID(&body->fid1), rc);
512 rc = lmv_revalidate_slaves(exp, reqp, &body->fid1, it, 1,
513 cb_blocking, extra_lock_flags);
519 OBD_FREE_PTR(sop_data);
523 /* this is not used currently */
524 int lmv_lookup_slaves(struct obd_export *exp, struct ptlrpc_request **reqp)
526 struct obd_device *obd = exp->exp_obd;
527 struct lmv_obd *lmv = &obd->u.lmv;
528 struct lustre_handle *lockh;
529 struct md_op_data *op_data;
530 struct ldlm_lock *lock;
531 struct mdt_body *body2;
532 struct mdt_body *body;
541 * Master is locked. we'd like to take locks on slaves and update
542 * attributes to be returned from the slaves it's important that lookup
543 * is called in two cases:
545 * - for first time (dcache has no such a resolving yet). -
546 * ->d_revalidate() returned false.
548 * Last case possible only if all the objs (master and all slaves aren't
552 OBD_ALLOC_PTR(op_data);
556 body = lustre_msg_buf((*reqp)->rq_repmsg,
557 DLM_REPLY_REC_OFF, sizeof(*body));
558 LASSERT(body != NULL);
559 LASSERT(lustre_rep_swabbed(*reqp, DLM_REPLY_REC_OFF));
561 LASSERT((body->valid & OBD_MD_FLID) != 0);
562 obj = lmv_obj_grab(obd, &body->fid1);
563 LASSERT(obj != NULL);
565 CDEBUG(D_OTHER, "lookup slaves for "DFID"\n",
570 for (i = 0; i < obj->lo_objcount; i++) {
571 struct lu_fid fid = obj->lo_inodes[i].li_fid;
572 struct ptlrpc_request *req = NULL;
573 struct obd_export *tgt_exp;
574 struct lookup_intent it;
576 if (lu_fid_eq(&fid, &obj->lo_fid))
577 /* skip master obj */
580 CDEBUG(D_OTHER, "lookup slave "DFID"\n", PFID(&fid));
583 memset(&it, 0, sizeof(it));
584 it.it_op = IT_GETATTR;
586 memset(op_data, 0, sizeof(*op_data));
587 op_data->op_fid1 = fid;
588 op_data->op_fid2 = fid;
589 op_data->op_bias = MDS_CROSS_REF;
591 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
593 GOTO(cleanup, rc = PTR_ERR(tgt_exp));
595 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0,
596 &req, lmv_blocking_ast, 0);
598 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
599 if (rc > 0 && req == NULL) {
600 /* nice, this slave is valid */
601 LASSERT(req == NULL);
602 CDEBUG(D_OTHER, "cached\n");
607 /* error during lookup */
610 lock = ldlm_handle2lock(lockh);
613 lock->l_ast_data = lmv_obj_get(obj);
615 body2 = lustre_msg_buf(req->rq_repmsg,
616 DLM_REPLY_REC_OFF, sizeof(*body2));
617 LASSERT(body2 != NULL);
618 LASSERT(lustre_rep_swabbed(req, DLM_REPLY_REC_OFF));
620 obj->lo_inodes[i].li_size = body2->size;
622 CDEBUG(D_OTHER, "fresh: %lu\n",
623 (unsigned long)obj->lo_inodes[i].li_size);
628 ptlrpc_req_finished(req);
630 lmv_update_body(body, obj->lo_inodes + i);
632 if (it.d.lustre.it_lock_mode) {
633 ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
634 it.d.lustre.it_lock_mode = 0;
642 OBD_FREE_PTR(op_data);
646 int lmv_intent_lookup(struct obd_export *exp, struct md_op_data *op_data,
647 void *lmm, int lmmsize, struct lookup_intent *it,
648 int flags, struct ptlrpc_request **reqp,
649 ldlm_blocking_callback cb_blocking,
650 int extra_lock_flags)
652 struct obd_device *obd = exp->exp_obd;
653 struct lu_fid rpid = op_data->op_fid1;
654 struct lmv_obd *lmv = &obd->u.lmv;
655 struct md_op_data *sop_data;
656 struct lmv_stripe_md *mea;
657 struct mdt_body *body;
664 OBD_ALLOC_PTR(sop_data);
665 if (sop_data == NULL)
668 /* save op_data fro repeat case */
669 *sop_data = *op_data;
672 * IT_LOOKUP is intended to produce name -> fid resolving (let's call
673 * this lookup below) or to confirm requested resolving is still valid
674 * (let's call this revalidation) fid_is_sane(&sop_data->op_fid2) specifies
677 if (fid_is_sane(&op_data->op_fid2)) {
679 * This is revalidate: we have to check is LOOKUP lock still
680 * valid for given fid. Very important part is that we have to
681 * choose right mds because namespace is per mds.
683 rpid = op_data->op_fid1;
684 obj = lmv_obj_grab(obd, &rpid);
686 mea_idx = raw_name2idx(obj->lo_hashtype,
688 (char *)op_data->op_name,
689 op_data->op_namelen);
690 rpid = obj->lo_inodes[mea_idx].li_fid;
691 mds = obj->lo_inodes[mea_idx].li_mds;
692 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
695 rc = lmv_fld_lookup(lmv, &rpid, &mds);
697 GOTO(out_free_sop_data, rc);
698 sop_data->op_bias |= MDS_CHECK_SPLIT;
701 CDEBUG(D_OTHER, "revalidate lookup for "DFID" to #"LPU64" MDS\n",
702 PFID(&op_data->op_fid2), mds);
709 * This is lookup. During lookup we have to update all the
710 * attributes, because returned values will be put in struct
713 obj = lmv_obj_grab(obd, &op_data->op_fid1);
715 if (op_data->op_namelen) {
716 /* directory is already split. calculate mds */
717 mea_idx = raw_name2idx(obj->lo_hashtype,
719 (char *)op_data->op_name,
720 op_data->op_namelen);
721 rpid = obj->lo_inodes[mea_idx].li_fid;
722 mds = obj->lo_inodes[mea_idx].li_mds;
724 sop_data->op_bias &= ~MDS_CHECK_SPLIT;
727 rc = lmv_fld_lookup(lmv, &op_data->op_fid1, &mds);
729 GOTO(out_free_sop_data, rc);
730 sop_data->op_bias |= MDS_CHECK_SPLIT;
732 fid_zero(&sop_data->op_fid2);
735 sop_data->op_bias &= ~MDS_CROSS_REF;
736 sop_data->op_fid1 = rpid;
738 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, sop_data, lmm, lmmsize,
739 it, flags, reqp, cb_blocking, extra_lock_flags);
741 LASSERT(fid_is_sane(&op_data->op_fid2));
743 * Very interesting. it seems object is still valid but for some
744 * reason llite calls lookup, not revalidate.
746 CDEBUG(D_OTHER, "lookup for "DFID" and data should be uptodate\n",
748 LASSERT(*reqp == NULL);
749 GOTO(out_free_sop_data, rc);
752 if (rc == 0 && *reqp == NULL) {
753 /* once again, we're asked for lookup, not revalidate */
754 CDEBUG(D_OTHER, "lookup for "DFID" and data should be uptodate\n",
756 GOTO(out_free_sop_data, rc);
759 if (rc == -ERESTART) {
760 LASSERT(*reqp != NULL);
761 DEBUG_REQ(D_WARNING|D_RPCTRACE, *reqp,
762 "Got -ERESTART during lookup!\n");
763 ptlrpc_req_finished(*reqp);
765 it->d.lustre.it_data = 0;
767 * Directory got split since last update. This shouldn't be
768 * because splitting causes lock revocation, so revalidate had
769 * to fail and lookup on dir had to return mea.
771 CWARN("we haven't knew about directory splitting!\n");
772 LASSERT(obj == NULL);
774 obj = lmv_obj_create(exp, &rpid, NULL);
776 GOTO(out_free_sop_data, rc = PTR_ERR(obj));
782 GOTO(out_free_sop_data, rc);
785 * Okay, MDS has returned success. Probably name has been resolved in
788 rc = lmv_intent_remote(exp, lmm, lmmsize, it, flags, reqp,
789 cb_blocking, extra_lock_flags);
791 if (rc == 0 && (mea = lmv_get_mea(*reqp, DLM_REPLY_REC_OFF))) {
792 /* Wow! This is split dir, we'd like to handle it. */
793 body = lustre_msg_buf((*reqp)->rq_repmsg,
794 DLM_REPLY_REC_OFF, sizeof(*body));
795 LASSERT(body != NULL);
796 LASSERT(lustre_rep_swabbed(*reqp, DLM_REPLY_REC_OFF));
797 LASSERT((body->valid & OBD_MD_FLID) != 0);
799 obj = lmv_obj_grab(obd, &body->fid1);
801 obj = lmv_obj_create(exp, &body->fid1, mea);
803 GOTO(out_free_sop_data, rc = (int)PTR_ERR(obj));
810 OBD_FREE_PTR(sop_data);
814 int lmv_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
815 void *lmm, int lmmsize, struct lookup_intent *it,
816 int flags, struct ptlrpc_request **reqp,
817 ldlm_blocking_callback cb_blocking,
818 int extra_lock_flags)
820 struct obd_device *obd = exp->exp_obd;
825 LASSERT(fid_is_sane(&op_data->op_fid1));
827 CDEBUG(D_OTHER, "INTENT LOCK '%s' for '%*s' on "DFID"\n",
828 LL_IT2STR(it), op_data->op_namelen, op_data->op_name,
829 PFID(&op_data->op_fid1));
831 rc = lmv_check_connect(obd);
835 if (it->it_op & IT_LOOKUP)
836 rc = lmv_intent_lookup(exp, op_data, lmm, lmmsize, it,
837 flags, reqp, cb_blocking,
839 else if (it->it_op & IT_OPEN)
840 rc = lmv_intent_open(exp, op_data, lmm, lmmsize, it,
841 flags, reqp, cb_blocking,
843 else if (it->it_op & IT_GETATTR)
844 rc = lmv_intent_getattr(exp, op_data,lmm, lmmsize, it,
845 flags, reqp, cb_blocking,
852 int lmv_revalidate_slaves(struct obd_export *exp, struct ptlrpc_request **reqp,
853 const struct lu_fid *mid, struct lookup_intent *oit,
854 int master_valid, ldlm_blocking_callback cb_blocking,
855 int extra_lock_flags)
857 struct obd_device *obd = exp->exp_obd;
858 struct ptlrpc_request *mreq = *reqp;
859 struct lmv_obd *lmv = &obd->u.lmv;
860 struct lustre_handle master_lockh;
861 struct obd_export *tgt_exp;
862 struct md_op_data *op_data;
863 struct ldlm_lock *lock;
864 unsigned long size = 0;
865 struct mdt_body *body;
867 int master_lock_mode;
871 OBD_ALLOC_PTR(op_data);
876 * We have to loop over the subobjects, check validity and update them
877 * from MDSs if needed. it's very useful that we need not to update all
878 * the fields. say, common fields (that are equal on all the subojects
879 * need not to be update, another fields (i_size, for example) are
880 * cached all the time.
882 obj = lmv_obj_grab(obd, mid);
883 LASSERT(obj != NULL);
885 master_lock_mode = 0;
889 for (i = 0; i < obj->lo_objcount; i++) {
890 struct lu_fid fid = obj->lo_inodes[i].li_fid;
891 struct lustre_handle *lockh = NULL;
892 struct ptlrpc_request *req = NULL;
893 ldlm_blocking_callback cb;
894 struct lookup_intent it;
897 CDEBUG(D_OTHER, "revalidate subobj "DFID"\n",
900 memset(op_data, 0, sizeof(*op_data));
901 memset(&it, 0, sizeof(it));
902 it.it_op = IT_GETATTR;
904 cb = lmv_blocking_ast;
906 if (lu_fid_eq(&fid, &obj->lo_fid)) {
909 * lmv_intent_getattr() already checked
910 * validness and took the lock.
914 * It even got the reply refresh attrs
917 body = lustre_msg_buf(mreq->rq_repmsg,
920 LASSERT(body != NULL);
921 LASSERT(lustre_rep_swabbed(
922 mreq, DLM_REPLY_REC_OFF));
925 /* take already cached attrs into account */
927 "master is locked and cached\n");
934 op_data->op_fid1 = fid;
935 op_data->op_fid2 = fid;
936 op_data->op_bias = MDS_CROSS_REF;
939 tgt_exp = lmv_get_export(lmv, obj->lo_inodes[i].li_mds);
941 GOTO(cleanup, rc = PTR_ERR(tgt_exp));
943 rc = md_intent_lock(tgt_exp, op_data, NULL, 0, &it, 0, &req, cb,
946 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
947 if (rc > 0 && req == NULL) {
948 /* Nice, this slave is valid */
949 LASSERT(req == NULL);
950 CDEBUG(D_OTHER, "cached\n");
958 LASSERT(master_valid == 0);
959 /* Save lock on master to be returned to the caller. */
960 CDEBUG(D_OTHER, "no lock on master yet\n");
961 memcpy(&master_lockh, lockh, sizeof(master_lockh));
962 master_lock_mode = it.d.lustre.it_lock_mode;
963 it.d.lustre.it_lock_mode = 0;
965 /* This is slave. We want to control it. */
966 lock = ldlm_handle2lock(lockh);
967 LASSERT(lock != NULL);
968 lock->l_ast_data = lmv_obj_get(obj);
974 * This is first reply, we'll use it to return updated
975 * data back to the caller.
978 ptlrpc_request_addref(req);
982 body = lustre_msg_buf(req->rq_repmsg,
983 DLM_REPLY_REC_OFF, sizeof(*body));
984 LASSERT(body != NULL);
985 LASSERT(lustre_rep_swabbed(req, DLM_REPLY_REC_OFF));
988 obj->lo_inodes[i].li_size = body->size;
990 CDEBUG(D_OTHER, "fresh: %lu\n",
991 (unsigned long)obj->lo_inodes[i].li_size);
994 ptlrpc_req_finished(req);
996 size += obj->lo_inodes[i].li_size;
998 if (it.d.lustre.it_lock_mode) {
999 ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
1000 it.d.lustre.it_lock_mode = 0;
1006 * Some attrs got refreshed, we have reply and it's time to put
1007 * fresh attrs to it.
1009 CDEBUG(D_OTHER, "return refreshed attrs: size = %lu\n",
1010 (unsigned long)size);
1012 body = lustre_msg_buf((*reqp)->rq_repmsg,
1013 DLM_REPLY_REC_OFF, sizeof(*body));
1014 LASSERT(body != NULL);
1015 LASSERT(lustre_rep_swabbed(*reqp, DLM_REPLY_REC_OFF));
1021 * Very important to maintain mds num the same because
1022 * of revalidation. mreq == NULL means that caller has
1023 * no reply and the only attr we can return is size.
1025 body->valid = OBD_MD_FLSIZE;
1027 if (master_valid == 0) {
1028 memcpy(&oit->d.lustre.it_lock_handle,
1029 &master_lockh, sizeof(master_lockh));
1030 oit->d.lustre.it_lock_mode = master_lock_mode;
1034 /* It seems all the attrs are fresh and we did no request */
1035 CDEBUG(D_OTHER, "all the attrs were fresh\n");
1036 if (master_valid == 0)
1037 oit->d.lustre.it_lock_mode = master_lock_mode;
1043 OBD_FREE_PTR(op_data);
1044 lmv_obj_unlock(obj);