1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2 * vim:expandtab:shiftwidth=8:tabstop=8:
4 * Copyright (C) 2002, 2003 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>
35 #include <liblustre.h>
38 #include <linux/obd_support.h>
39 #include <linux/lustre_lib.h>
40 #include <linux/lustre_net.h>
41 #include <linux/lustre_idl.h>
42 #include <linux/lustre_dlm.h>
43 #include <linux/lustre_mds.h>
44 #include <linux/obd_class.h>
45 #include <linux/obd_ost.h>
46 #include <linux/lprocfs_status.h>
47 #include <linux/lustre_fsfilt.h>
48 #include <linux/obd_lmv.h>
49 #include "lmv_internal.h"
52 static inline void lmv_drop_intent_lock(struct lookup_intent *it)
54 if (it->d.lustre.it_lock_mode != 0)
55 ldlm_lock_decref((void *)&it->d.lustre.it_lock_handle,
56 it->d.lustre.it_lock_mode);
59 int lmv_handle_remote_inode(struct obd_export *exp, struct ll_uctxt *uctxt,
60 void *lmm, int lmmsize,
61 struct lookup_intent *it, int flags,
62 struct ptlrpc_request **reqp,
63 ldlm_blocking_callback cb_blocking)
65 struct obd_device *obd = exp->exp_obd;
66 struct lmv_obd *lmv = &obd->u.lmv;
67 struct mds_body *body = NULL;
71 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
72 LASSERT(body != NULL);
74 if (body->valid & OBD_MD_MDS) {
75 /* oh, MDS reports that this is remote inode case
76 * i.e. we have to ask for real attrs on another MDS */
77 struct ptlrpc_request *req;
79 struct lustre_handle plock;
82 if (it->it_op == IT_LOOKUP) {
83 /* unfortunately, we have to lie to MDC/MDS to
84 * retrieve attributes llite needs */
85 it->it_op = IT_GETATTR;
88 /* we got LOOKUP lock, but we really need attrs */
89 pmode = it->d.lustre.it_lock_mode;
91 memcpy(&plock, &it->d.lustre.it_lock_handle,
93 it->d.lustre.it_lock_mode = 0;
97 it->d.lustre.it_disposition &= ~DISP_ENQ_COMPLETE;
98 rc = md_intent_lock(lmv->tgts[nfid.mds].ltd_exp, uctxt, &nfid,
99 NULL, 0, lmm, lmmsize, NULL, it, flags,
102 /* llite needs LOOKUP lock to track dentry revocation in
103 * order to maintain dcache consistency. thus drop UPDATE
104 * lock here and put LOOKUP in request */
106 lmv_drop_intent_lock(it);
107 memcpy(&it->d.lustre.it_lock_handle, &plock,
109 it->d.lustre.it_lock_mode = pmode;
112 ldlm_lock_decref(&plock, pmode);
114 ptlrpc_req_finished(*reqp);
120 int lmv_intent_open(struct obd_export *exp, struct ll_uctxt *uctxt,
121 struct ll_fid *pfid, const char *name, int len,
122 void *lmm, int lmmsize, struct ll_fid *cfid,
123 struct lookup_intent *it, int flags,
124 struct ptlrpc_request **reqp,
125 ldlm_blocking_callback cb_blocking)
127 struct obd_device *obd = exp->exp_obd;
128 struct lmv_obd *lmv = &obd->u.lmv;
129 struct mds_body *body = NULL;
130 struct ll_fid rpfid = *pfid;
133 int rc, mds, loop = 0;
136 /* IT_OPEN is intended to open (and create, possible) an object. Parent
137 * (pfid) may be splitted dir */
140 LASSERT(++loop <= 2);
142 obj = lmv_grab_obj(obd, &rpfid);
144 /* directory is already splitted, so we have to forward
145 * request to the right MDS */
146 mds = raw_name2idx(obj->objcount, (char *)name, len);
147 CDEBUG(D_OTHER, "forward to MDS #%u (%lu/%lu/%lu)\n", mds,
148 (unsigned long) rpfid.mds, (unsigned long) rpfid.id,
149 (unsigned long) rpfid.generation);
150 rpfid = obj->objs[mds].fid;
154 rc = md_intent_lock(lmv->tgts[rpfid.mds].ltd_exp, uctxt, &rpfid, name,
155 len, lmm, lmmsize, cfid, it, flags, reqp,
157 if (rc == -ERESTART) {
158 /* directory got splitted. time to update local object
159 * and repeat the request with proper MDS */
160 LASSERT(fid_equal(pfid, &rpfid));
161 rc = lmv_get_mea_and_update_object(exp, &rpfid);
163 ptlrpc_req_finished(*reqp);
170 /* okay, MDS has returned success. Probably name has been resolved in
172 rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it,
173 flags, reqp, cb_blocking);
176 CERROR("can't handle remote %s: dir %lu/%lu/%lu(%lu/%lu/%lu):"
177 "%*s: %d\n", LL_IT2STR(it),
178 (unsigned long) pfid->mds,
179 (unsigned long) pfid->id,
180 (unsigned long) pfid->generation,
181 (unsigned long) rpfid.mds,
182 (unsigned long) rpfid.id,
183 (unsigned long) rpfid.generation,
188 /* caller may use attrs MDS returns on IT_OPEN lock request so, we have
189 * to update them for splitted dir */
190 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
191 LASSERT(body != NULL);
194 obj = lmv_grab_obj(obd, cfid);
195 if (!obj && (mea = body_of_splitted_dir(*reqp, 1))) {
196 /* wow! this is splitted dir, we'd like to handle it */
197 obj = lmv_create_obj(exp, &body->fid1, mea);
199 RETURN(PTR_ERR(obj));
203 /* this is splitted dir and we'd want to get attrs */
204 CDEBUG(D_OTHER, "attrs from slaves for %lu/%lu/%lu\n",
205 (unsigned long)cfid->mds, (unsigned long)cfid->id,
206 (unsigned long)cfid->generation);
207 rc = lmv_revalidate_slaves(exp, reqp, cfid,
209 } else if (S_ISDIR(body->mode)) {
210 /*CWARN("hmmm, %lu/%lu/%lu has not lmv obj?!\n",
211 (unsigned long) cfid->mds,
212 (unsigned long) cfid->id,
213 (unsigned long) cfid->generation);*/
222 int lmv_intent_getattr(struct obd_export *exp, struct ll_uctxt *uctxt,
223 struct ll_fid *pfid, const char *name, int len,
224 void *lmm, int lmmsize, struct ll_fid *cfid,
225 struct lookup_intent *it, int flags,
226 struct ptlrpc_request **reqp,
227 ldlm_blocking_callback cb_blocking)
229 struct obd_device *obd = exp->exp_obd;
230 struct lmv_obd *lmv = &obd->u.lmv;
231 struct mds_body *body = NULL;
232 struct ll_fid rpfid = *pfid;
233 struct lmv_obj *obj, *obj2;
239 /* caller wants to revalidate attrs of obj we have to revalidate
240 * slaves if requested object is splitted directory */
241 CDEBUG(D_OTHER, "revalidate attrs for %lu/%lu/%lu\n",
242 (unsigned long)cfid->mds, (unsigned long)cfid->id,
243 (unsigned long)cfid->generation);
245 obj = lmv_grab_obj(obd, cfid);
247 /* in fact, we need not this with current intent_lock(),
248 * but it may change some day */
249 rpfid = obj->objs[mds].fid;
253 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
254 len, lmm, lmmsize, cfid, it, flags, reqp,
256 if (obj && rc >= 0) {
257 /* this is splitted dir. In order to optimize things a
258 * bit, we consider obj valid updating missing parts.
260 * FIXME: do we need to return any lock here? It would
261 * be fine if we don't. this means that nobody should
262 * use UPDATE lock to notify about object * removal */
264 "revalidate slaves for %lu/%lu/%lu, rc %d\n",
265 (unsigned long)cfid->mds, (unsigned long)cfid->id,
266 (unsigned long)cfid->generation, rc);
268 rc = lmv_revalidate_slaves(exp, reqp, cfid, it, rc,
275 CDEBUG(D_OTHER, "INTENT getattr for %*s on %lu/%lu/%lu\n",
276 len, name, (unsigned long)pfid->mds, (unsigned long)pfid->id,
277 (unsigned long)pfid->generation);
280 obj = lmv_grab_obj(obd, pfid);
282 /* directory is already splitted. calculate mds */
283 mds = raw_name2idx(obj->objcount, (char *) name, len);
284 rpfid = obj->objs[mds].fid;
288 CDEBUG(D_OTHER, "forward to MDS #%u (slave %lu/%lu/%lu)\n",
289 mds, (unsigned long)rpfid.mds, (unsigned long)rpfid.id,
290 (unsigned long)rpfid.generation);
293 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
294 len, lmm, lmmsize, NULL, it, flags, reqp,
302 /* okay, MDS has returned success. probably name has been
303 * resolved in remote inode */
304 rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it,
305 flags, reqp, cb_blocking);
309 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
310 LASSERT(body != NULL);
313 obj2 = lmv_grab_obj(obd, cfid);
315 if (!obj2 && (mea = body_of_splitted_dir(*reqp, 1))) {
316 /* wow! this is splitted dir, we'd like to handle it. */
317 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
318 LASSERT(body != NULL);
320 obj2 = lmv_create_obj(exp, &body->fid1, mea);
322 RETURN(PTR_ERR(obj2));
326 /* this is splitted dir and we'd want to get attrs */
327 CDEBUG(D_OTHER, "attrs from slaves for %lu/%lu/%lu, rc %d\n",
328 (unsigned long)cfid->mds, (unsigned long)cfid->id,
329 (unsigned long)cfid->generation, rc);
331 rc = lmv_revalidate_slaves(exp, reqp, cfid, it, 1, cb_blocking);
337 void lmv_update_body_from_obj(struct mds_body *body, struct lmv_inode *obj)
340 body->size += obj->size;
341 /* body->atime = obj->atime;
342 body->ctime = obj->ctime;
343 body->mtime = obj->mtime;
344 body->nlink = obj->nlink;*/
347 int lmv_lookup_slaves(struct obd_export *exp, struct ptlrpc_request **reqp)
349 struct obd_device *obd = exp->exp_obd;
350 struct lmv_obd *lmv = &obd->u.lmv;
351 struct mds_body *body = NULL;
352 struct lustre_handle *lockh;
353 struct ldlm_lock *lock;
354 struct mds_body *body2;
355 struct ll_uctxt uctxt;
363 /* master is locked. we'd like to take locks on slaves and update
364 * attributes to be returned from the slaves it's important that lookup
365 * is called in two cases:
367 * - for first time (dcache has no such a resolving yet).
368 * - ->d_revalidate() returned false.
370 * last case possible only if all the objs (master and all slaves aren't
373 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
374 LASSERT(body != NULL);
376 obj = lmv_grab_obj(obd, &body->fid1);
377 LASSERT(obj != NULL);
379 CDEBUG(D_OTHER, "lookup slaves for %lu/%lu/%lu\n",
380 (unsigned long)body->fid1.mds,
381 (unsigned long)body->fid1.id,
382 (unsigned long)body->fid1.generation);
389 for (i = 0; i < obj->objcount; i++) {
390 struct ll_fid fid = obj->objs[i].fid;
391 struct ptlrpc_request *req = NULL;
392 struct lookup_intent it;
394 if (fid_equal(&fid, &obj->fid))
395 /* skip master obj */
398 CDEBUG(D_OTHER, "lookup slave %lu/%lu/%lu\n",
399 (unsigned long)fid.mds, (unsigned long)fid.id,
400 (unsigned long)fid.generation);
403 memset(&it, 0, sizeof(it));
404 it.it_op = IT_GETATTR;
405 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
406 NULL, 0, NULL, 0, &fid, &it, 0, &req,
407 lmv_dirobj_blocking_ast);
409 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
411 /* nice, this slave is valid */
412 LASSERT(req == NULL);
413 CDEBUG(D_OTHER, "cached\n");
418 /* error during revalidation */
421 /* rc == 0, this means we have no such a lock and can't think
422 * obj is still valid. lookup it again */
423 LASSERT(req == NULL);
426 memset(&it, 0, sizeof(it));
427 it.it_op = IT_GETATTR;
428 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
429 NULL, 0, NULL, 0, NULL, &it, 0, &req,
430 lmv_dirobj_blocking_ast);
432 lockh = (struct lustre_handle *) &it.d.lustre.it_lock_handle;
436 /* error during lookup */
439 lock = ldlm_handle2lock(lockh);
442 lock->l_ast_data = lmv_get_obj(obj);
444 body2 = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body2));
447 obj->objs[i].size = body2->size;
449 CDEBUG(D_OTHER, "fresh: %lu\n",
450 (unsigned long)obj->objs[i].size);
455 ptlrpc_req_finished(req);
457 lmv_update_body_from_obj(body, obj->objs + i);
459 if (it.d.lustre.it_lock_mode)
460 ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
468 int lmv_intent_lookup(struct obd_export *exp, struct ll_uctxt *uctxt,
469 struct ll_fid *pfid, const char *name, int len,
470 void *lmm, int lmmsize, struct ll_fid *cfid,
471 struct lookup_intent *it, int flags,
472 struct ptlrpc_request **reqp,
473 ldlm_blocking_callback cb_blocking)
475 struct obd_device *obd = exp->exp_obd;
476 struct lmv_obd *lmv = &obd->u.lmv;
477 struct mds_body *body = NULL;
478 struct ll_fid rpfid = *pfid;
481 int rc, mds, loop = 0;
484 /* IT_LOOKUP is intended to produce name -> fid resolving (let's call
485 * this lookup below) or to confirm requested resolving is still valid
486 * (let's call this revalidation) cfid != NULL specifies revalidation */
489 /* this is revalidation: we have to check is LOOKUP lock still
490 * valid for given fid. very important part is that we have to
491 * choose right mds because namespace is per mds */
493 obj = lmv_grab_obj(obd, pfid);
495 mds = raw_name2idx(obj->objcount, (char *) name, len);
496 rpfid = obj->objs[mds].fid;
501 CDEBUG(D_OTHER, "revalidate lookup for %lu/%lu/%lu to %d MDS\n",
502 (unsigned long)cfid->mds, (unsigned long)cfid->id,
503 (unsigned long)cfid->generation, mds);
505 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, pfid, name,
506 len, lmm, lmmsize, cfid, it, flags,
513 LASSERT(++loop <= 2);
514 /* this is lookup. during lookup we have to update all the attributes,
515 * because returned values will be put in struct inode */
517 obj = lmv_grab_obj(obd, pfid);
520 /* directory is already splitted. calculate mds */
521 mds = raw_name2idx(obj->objcount, (char *)name, len);
522 rpfid = obj->objs[mds].fid;
528 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
529 len, lmm, lmmsize, NULL, it, flags, reqp,
532 /* very interesting. it seems object is still valid but for some
533 * reason llite calls lookup, not revalidate */
534 CWARN("lookup for %lu/%lu/%lu and data should be uptodate\n",
535 (unsigned long)rpfid.mds, (unsigned long)rpfid.id,
536 (unsigned long)rpfid.generation);
537 LASSERT(*reqp == NULL);
541 if (rc == 0 && *reqp == NULL) {
542 /* once again, we're asked for lookup, not revalidate */
543 CWARN("lookup for %lu/%lu/%lu and data should be uptodate\n",
544 (unsigned long)rpfid.mds, (unsigned long)rpfid.id,
545 (unsigned long)rpfid.generation);
549 if (rc == -ERESTART) {
550 /* directory got splitted since last update. this shouldn't be
551 * becasue splitting causes lock revocation, so revalidate had
552 * to fail and lookup on dir had to return mea */
553 CWARN("we haven't knew about directory splitting!\n");
554 LASSERT(obj == NULL);
556 obj = lmv_create_obj(exp, &rpfid, NULL);
558 RETURN(PTR_ERR(obj));
566 /* okay, MDS has returned success. probably name has been resolved in
568 rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it, flags,
571 if (rc == 0 && (mea = body_of_splitted_dir(*reqp, 1))) {
572 /* wow! this is splitted dir, we'd like to handle it */
573 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
574 LASSERT(body != NULL);
576 obj = lmv_grab_obj(obd, &body->fid1);
578 obj = lmv_create_obj(exp, &body->fid1, mea);
580 RETURN(PTR_ERR(obj));
588 int lmv_intent_lock(struct obd_export *exp, struct ll_uctxt *uctxt,
589 struct ll_fid *pfid, const char *name, int len,
590 void *lmm, int lmmsize, struct ll_fid *cfid,
591 struct lookup_intent *it, int flags,
592 struct ptlrpc_request **reqp,
593 ldlm_blocking_callback cb_blocking)
595 struct obd_device *obd = exp->exp_obd;
602 CDEBUG(D_OTHER, "INTENT LOCK '%s' for '%*s' on %lu/%lu -> %u\n",
603 LL_IT2STR(it), len, name, (unsigned long) pfid->id,
604 (unsigned long) pfid->generation, pfid->mds);
606 rc = lmv_check_connect(obd);
610 if (it->it_op == IT_LOOKUP)
611 rc = lmv_intent_lookup(exp, uctxt, pfid, name, len, lmm,
612 lmmsize, cfid, it, flags, reqp,
614 else if (it->it_op & IT_OPEN)
615 rc = lmv_intent_open(exp, uctxt, pfid, name, len, lmm,
616 lmmsize, cfid, it, flags, reqp,
618 else if (it->it_op == IT_GETATTR || it->it_op == IT_CHDIR)
619 rc = lmv_intent_getattr(exp, uctxt, pfid, name, len, lmm,
620 lmmsize, cfid, it, flags, reqp,
627 int lmv_revalidate_slaves(struct obd_export *exp, struct ptlrpc_request **reqp,
628 struct ll_fid *mfid, struct lookup_intent *oit,
629 int master_valid, ldlm_blocking_callback cb_blocking)
631 struct obd_device *obd = exp->exp_obd;
632 struct ptlrpc_request *mreq = *reqp;
633 struct lmv_obd *lmv = &obd->u.lmv;
634 struct lustre_handle master_lockh;
635 struct ldlm_lock *lock;
636 unsigned long size = 0;
637 struct mds_body *body;
638 struct ll_uctxt uctxt;
640 int master_lock_mode;
644 /* we have to loop over the subobjects, check validity and update them
645 * from MDSs if needed. it's very useful that we need not to update all
646 * the fields. say, common fields (that are equal on all the subojects
647 * need not to be update, another fields (i_size, for example) are
648 * cached all the time */
649 obj = lmv_grab_obj(obd, mfid);
650 LASSERT(obj != NULL);
654 master_lock_mode = 0;
658 for (i = 0; i < obj->objcount; i++) {
659 struct ll_fid fid = obj->objs[i].fid;
660 struct lustre_handle *lockh = NULL;
661 struct ptlrpc_request *req = NULL;
662 ldlm_blocking_callback cb;
663 struct lookup_intent it;
666 CDEBUG(D_OTHER, "revalidate subobj %lu/%lu/%lu\n",
667 (unsigned long)fid.mds, (unsigned long)fid.id,
668 (unsigned long) fid.generation);
670 memset(&it, 0, sizeof(it));
671 it.it_op = IT_GETATTR;
672 cb = lmv_dirobj_blocking_ast;
674 if (fid_equal(&fid, &obj->fid)) {
676 /* lmv_intent_getattr() already checked
677 * validness and took the lock */
679 /* it even got the reply refresh attrs
681 body = lustre_msg_buf(mreq->rq_repmsg,
683 LASSERT(body != NULL);
686 /* take already cached attrs into account */
688 "master is locked and cached\n");
696 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
697 NULL, 0, NULL, 0, &fid, &it, 0, &req, cb);
698 lockh = (struct lustre_handle *) &it.d.lustre.it_lock_handle;
700 /* nice, this slave is valid */
701 LASSERT(req == NULL);
702 CDEBUG(D_OTHER, "cached\n");
707 /* error during revalidation */
710 /* rc == 0, this means we have no such a lock and can't think
711 * obj is still valid. lookup it again */
712 LASSERT(req == NULL);
715 memset(&it, 0, sizeof(it));
716 it.it_op = IT_GETATTR;
717 rc = md_intent_lock(lmv->tgts[fid.mds].ltd_exp, &uctxt, &fid,
718 NULL, 0, NULL, 0, NULL, &it, 0, &req, cb);
719 lockh = (struct lustre_handle *) &it.d.lustre.it_lock_handle;
723 /* error during lookup */
727 LASSERT(master_valid == 0);
728 /* save lock on master to be returned to the caller */
729 CDEBUG(D_OTHER, "no lock on master yet\n");
730 memcpy(&master_lockh, lockh, sizeof(master_lockh));
731 master_lock_mode = it.d.lustre.it_lock_mode;
732 it.d.lustre.it_lock_mode = 0;
734 /* this is slave. we want to control it */
735 lock = ldlm_handle2lock(lockh);
737 lock->l_ast_data = lmv_get_obj(obj);
742 /* this is first reply, we'll use it to return
743 * updated data back to the caller */
745 ptlrpc_request_addref(req);
750 body = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body));
754 obj->objs[i].size = body->size;
756 CDEBUG(D_OTHER, "fresh: %lu\n",
757 (unsigned long)obj->objs[i].size);
760 ptlrpc_req_finished(req);
762 size += obj->objs[i].size;
764 if (it.d.lustre.it_lock_mode)
765 ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
769 /* some attrs got refreshed, we have reply and it's time to put
770 * fresh attrs to it */
771 CDEBUG(D_OTHER, "return refreshed attrs: size = %lu\n",
772 (unsigned long)size);
774 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
777 /* FIXME: what about other attributes? */
781 /* very important to maintain lli->mds the same because
782 * of revalidation. mreq == NULL means that caller has
783 * no reply and the only attr we can return is size */
784 body->valid = OBD_MD_FLSIZE;
785 body->mds = obj->fid.mds;
787 if (master_valid == 0) {
788 memcpy(&oit->d.lustre.it_lock_handle,
789 &master_lockh, sizeof(master_lockh));
790 oit->d.lustre.it_lock_mode = master_lock_mode;
794 /* it seems all the attrs are fresh and we did no request */
795 CDEBUG(D_OTHER, "all the attrs were fresh\n");
796 if (master_valid == 0)
797 oit->d.lustre.it_lock_mode = master_lock_mode;