Whamcloud - gitweb
e012cbb8e294ee09403d1353f97e85a9ef83368a
[fs/lustre-release.git] / lustre / lmv / lmv_intent.c
1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
2  * vim:expandtab:shiftwidth=8:tabstop=8:
3  *
4  * Copyright (C) 2002, 2003 Cluster File Systems, Inc.
5  *
6  *   This file is part of Lustre, http://www.lustre.org.
7  *
8  *   Lustre is free software; you can redistribute it and/or
9  *   modify it under the terms of version 2 of the GNU General Public
10  *   License as published by the Free Software Foundation.
11  *
12  *   Lustre is distributed in the hope that it will be useful,
13  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
14  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
15  *   GNU General Public License for more details.
16  *
17  *   You should have received a copy of the GNU General Public License
18  *   along with Lustre; if not, write to the Free Software
19  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
20  */
21
22 #ifndef EXPORT_SYMTAB
23 # define EXPORT_SYMTAB
24 #endif
25 #define DEBUG_SUBSYSTEM S_LMV
26 #ifdef __KERNEL__
27 #include <linux/slab.h>
28 #include <linux/module.h>
29 #include <linux/init.h>
30 #include <linux/slab.h>
31 #include <linux/pagemap.h>
32 #include <asm/div64.h>
33 #include <linux/seq_file.h>
34 #else
35 #include <liblustre.h>
36 #endif
37
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"
50
51
52 static inline void lmv_drop_intent_lock(struct lookup_intent *it)
53 {
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);
57 }
58
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)
64 {
65         struct obd_device *obd = exp->exp_obd;
66         struct lmv_obd *lmv = &obd->u.lmv;
67         struct mds_body *body = NULL;
68         int rc = 0;
69         ENTRY;
70
71         body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
72         LASSERT(body != NULL);
73
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;
78                 struct ll_fid nfid;
79                 struct lustre_handle plock;
80                 int pmode;
81
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;
86                 }
87
88                 /* we got LOOKUP lock, but we really need attrs */
89                 pmode = it->d.lustre.it_lock_mode;
90                 if (pmode) {
91                         memcpy(&plock, &it->d.lustre.it_lock_handle,
92                                         sizeof(plock));
93                         it->d.lustre.it_lock_mode = 0;
94                 }
95
96                 nfid = body->fid1;
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,
100                                     &req, cb_blocking);
101
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 */
105                 if (rc == 0) {
106                         lmv_drop_intent_lock(it);
107                         memcpy(&it->d.lustre.it_lock_handle, &plock,
108                                         sizeof(plock));
109                         it->d.lustre.it_lock_mode = pmode;
110                         
111                 } else if (pmode)
112                         ldlm_lock_decref(&plock, pmode);
113
114                 ptlrpc_req_finished(*reqp);
115                 *reqp = req;
116         }
117         RETURN(rc);
118 }
119
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)
126 {
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;
131         struct lmv_obj *obj;
132         struct mea *mea;
133         int rc, mds, loop = 0;
134         ENTRY;
135
136         /* IT_OPEN is intended to open (and create, possible) an object. Parent
137          * (pfid) may be splitted dir */
138
139 repeat:
140         LASSERT(++loop <= 2);
141         mds = rpfid.mds;
142         obj = lmv_grab_obj(obd, &rpfid);
143         if (obj) {
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;
151                 lmv_put_obj(obj);
152         }
153
154         rc = md_intent_lock(lmv->tgts[rpfid.mds].ltd_exp, uctxt, &rpfid, name,
155                             len, lmm, lmmsize, cfid, it, flags, reqp,
156                             cb_blocking);
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);
162                 if (rc == 0) {
163                         ptlrpc_req_finished(*reqp);
164                         goto repeat;
165                 }
166         }
167         if (rc != 0)
168                 RETURN(rc);
169
170         /* okay, MDS has returned success. Probably name has been resolved in
171          * remote inode */
172         rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it,
173                                      flags, reqp, cb_blocking);
174         if (rc != 0) {
175                 LASSERT(rc < 0);
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,
184                        len, name, rc);
185                 RETURN(rc);
186         }
187
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);
192
193         cfid = &body->fid1;
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);
198                 if (IS_ERR(obj))
199                         RETURN(PTR_ERR(obj));
200         }
201
202         if (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,
208                                            it, 1, cb_blocking);
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);*/
214         }
215         
216         if (obj)
217                 lmv_put_obj(obj);
218         
219         RETURN(rc);
220 }
221
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)
228 {
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;
234         struct mea *mea;
235         int rc = 0, mds;
236         ENTRY;
237
238         if (cfid) {
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);
244                 mds = cfid->mds;
245                 obj = lmv_grab_obj(obd, cfid);
246                 if (obj) {
247                         /* in fact, we need not this with current intent_lock(),
248                          * but it may change some day */
249                         rpfid = obj->objs[mds].fid;
250                         mds = rpfid.mds;
251                         lmv_put_obj(obj);
252                 }
253                 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
254                                     len, lmm, lmmsize, cfid, it, flags, reqp,
255                                     cb_blocking);
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.
259
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 */
263                         CDEBUG(D_OTHER,
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);
267                         
268                         rc = lmv_revalidate_slaves(exp, reqp, cfid, it, rc,
269                                                    cb_blocking);
270                 }
271
272                 RETURN(rc);                
273         }
274
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);
278
279         mds = pfid->mds;
280         obj = lmv_grab_obj(obd, pfid);
281         if (obj && len) {
282                 /* directory is already splitted. calculate mds */
283                 mds = raw_name2idx(obj->objcount, (char *) name, len);
284                 rpfid = obj->objs[mds].fid;
285                 mds = rpfid.mds;
286                 lmv_put_obj(obj);
287                 
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);
291         }
292         
293         rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
294                             len, lmm, lmmsize, NULL, it, flags, reqp,
295                             cb_blocking);
296         
297         if (rc < 0)
298                 RETURN(rc);
299         
300         LASSERT(rc == 0);
301
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);
306         if (rc < 0)
307                 RETURN(rc);
308
309         body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
310         LASSERT(body != NULL);
311         
312         cfid = &body->fid1;
313         obj2 = lmv_grab_obj(obd, cfid);
314
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);
319
320                 obj2 = lmv_create_obj(exp, &body->fid1, mea);
321                 if (IS_ERR(obj2))
322                         RETURN(PTR_ERR(obj2));
323         }
324
325         if (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);
330                 
331                 rc = lmv_revalidate_slaves(exp, reqp, cfid, it, 1, cb_blocking);
332                 lmv_put_obj(obj2);
333         }
334         RETURN(rc);
335 }
336
337 void lmv_update_body_from_obj(struct mds_body *body, struct lmv_inode *obj)
338 {
339         /* update size */
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;*/
345 }
346
347 int lmv_lookup_slaves(struct obd_export *exp, struct ptlrpc_request **reqp)
348 {
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;
356         struct lmv_obj *obj;
357         int i, rc = 0;
358         ENTRY;
359
360         LASSERT(reqp);
361         LASSERT(*reqp);
362
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:
366          
367          *  - for first time (dcache has no such a resolving yet).
368          *  - ->d_revalidate() returned false.
369          
370          * last case possible only if all the objs (master and all slaves aren't
371          * valid */
372
373         body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
374         LASSERT(body != NULL);
375
376         obj = lmv_grab_obj(obd, &body->fid1);
377         LASSERT(obj != NULL);
378
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);
383
384         uctxt.gid1 = 0;
385         uctxt.gid2 = 0;
386
387         lmv_lock_obj(obj);
388         
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;
393
394                 if (fid_equal(&fid, &obj->fid))
395                         /* skip master obj */
396                         continue;
397
398                 CDEBUG(D_OTHER, "lookup slave %lu/%lu/%lu\n",
399                        (unsigned long)fid.mds, (unsigned long)fid.id,
400                        (unsigned long)fid.generation);
401
402                 /* is obj valid? */
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);
408                 
409                 lockh = (struct lustre_handle *)&it.d.lustre.it_lock_handle;
410                 if (rc > 0) {
411                         /* nice, this slave is valid */
412                         LASSERT(req == NULL);
413                         CDEBUG(D_OTHER, "cached\n");
414                         goto release_lock;
415                 }
416
417                 if (rc < 0)
418                         /* error during revalidation */
419                         GOTO(cleanup, rc);
420
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);
424                 req = NULL;
425
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);
431
432                 lockh = (struct lustre_handle *) &it.d.lustre.it_lock_handle;
433                 LASSERT(rc <= 0);
434
435                 if (rc < 0)
436                         /* error during lookup */
437                         GOTO(cleanup, rc);
438                 
439                 lock = ldlm_handle2lock(lockh);
440                 LASSERT(lock);
441
442                 lock->l_ast_data = lmv_get_obj(obj);
443
444                 body2 = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body2));
445                 LASSERT(body2);
446
447                 obj->objs[i].size = body2->size;
448                 
449                 CDEBUG(D_OTHER, "fresh: %lu\n",
450                        (unsigned long)obj->objs[i].size);
451
452                 LDLM_LOCK_PUT(lock);
453
454                 if (req)
455                         ptlrpc_req_finished(req);
456 release_lock:
457                 lmv_update_body_from_obj(body, obj->objs + i);
458
459                 if (it.d.lustre.it_lock_mode)
460                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
461         }
462 cleanup:
463         lmv_unlock_obj(obj);
464         lmv_put_obj(obj);
465         RETURN(rc);
466 }
467
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)
474 {
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;
479         struct lmv_obj *obj;
480         struct mea *mea;
481         int rc, mds, loop = 0;
482         ENTRY;
483
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 */
487
488         if (cfid) {
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 */
492                 rpfid = *pfid;
493                 obj = lmv_grab_obj(obd, pfid);
494                 if (obj) {
495                         mds = raw_name2idx(obj->objcount, (char *) name, len);
496                         rpfid = obj->objs[mds].fid;
497                         lmv_put_obj(obj);
498                 }
499                 mds = rpfid.mds;
500                 
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);
504                 
505                 rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, pfid, name,
506                                     len, lmm, lmmsize, cfid, it, flags,
507                                     reqp, cb_blocking);
508                 RETURN(rc);
509         }
510
511         mds = pfid->mds;
512 repeat:
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 */
516
517         obj = lmv_grab_obj(obd, pfid);
518         if (obj) {
519                 if (len) {
520                         /* directory is already splitted. calculate mds */
521                         mds = raw_name2idx(obj->objcount, (char *)name, len);
522                         rpfid = obj->objs[mds].fid;
523                         mds = rpfid.mds;
524                 }
525                 lmv_put_obj(obj);
526         }
527
528         rc = md_intent_lock(lmv->tgts[mds].ltd_exp, uctxt, &rpfid, name,
529                             len, lmm, lmmsize, NULL, it, flags, reqp,
530                             cb_blocking);
531         if (rc > 0) {
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);
538                 RETURN(rc);
539         }
540
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);
546                 RETURN(rc);
547         }
548        
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);
555
556                 obj = lmv_create_obj(exp, &rpfid, NULL);
557                 if (IS_ERR(obj))
558                         RETURN(PTR_ERR(obj));
559                 lmv_put_obj(obj);
560                 goto repeat;
561         }
562
563         if (rc < 0)
564                 RETURN(rc);
565
566         /* okay, MDS has returned success. probably name has been resolved in
567          * remote inode */
568         rc = lmv_handle_remote_inode(exp, uctxt, lmm, lmmsize, it, flags,
569                                      reqp, cb_blocking);
570
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);
575                 
576                 obj = lmv_grab_obj(obd, &body->fid1);
577                 if (!obj) {
578                         obj = lmv_create_obj(exp, &body->fid1, mea);
579                         if (IS_ERR(obj))
580                                 RETURN(PTR_ERR(obj));
581                 }
582                 lmv_put_obj(obj);
583         }
584
585         RETURN(rc);
586 }
587
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)
594 {
595         struct obd_device *obd = exp->exp_obd;
596         int rc = 0;
597         ENTRY;
598
599         LASSERT(it);
600         LASSERT(pfid);
601
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);
605
606         rc = lmv_check_connect(obd);
607         if (rc)
608                 RETURN(rc);
609
610         if (it->it_op == IT_LOOKUP)
611                 rc = lmv_intent_lookup(exp, uctxt, pfid, name, len, lmm,
612                                        lmmsize, cfid, it, flags, reqp,
613                                        cb_blocking);
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,
617                                      cb_blocking);
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,
621                                         cb_blocking);
622         else
623                 LBUG();
624         RETURN(rc);
625 }
626
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)
630 {
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;
639         struct lmv_obj *obj;
640         int master_lock_mode;
641         int i, rc = 0;
642         ENTRY;
643
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);
651
652         uctxt.gid1 = 0;
653         uctxt.gid2 = 0;
654         master_lock_mode = 0;
655
656         lmv_lock_obj(obj);
657         
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;
664                 int master = 0;
665
666                 CDEBUG(D_OTHER, "revalidate subobj %lu/%lu/%lu\n",
667                        (unsigned long)fid.mds, (unsigned long)fid.id,
668                        (unsigned long) fid.generation);
669
670                 memset(&it, 0, sizeof(it));
671                 it.it_op = IT_GETATTR;
672                 cb = lmv_dirobj_blocking_ast;
673
674                 if (fid_equal(&fid, &obj->fid)) {
675                         if (master_valid) {
676                                 /* lmv_intent_getattr() already checked
677                                  * validness and took the lock */
678                                 if (mreq) {
679                                         /* it even got the reply refresh attrs
680                                          * from that reply */
681                                         body = lustre_msg_buf(mreq->rq_repmsg,
682                                                               1, sizeof(*body));
683                                         LASSERT(body != NULL);
684                                         goto update; 
685                                 }
686                                 /* take already cached attrs into account */
687                                 CDEBUG(D_OTHER,
688                                        "master is locked and cached\n");
689                                 goto release_lock;
690                         }
691                         master = 1;
692                         cb = cb_blocking;
693                 }
694
695                 /* is obj valid? */
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;
699                 if (rc > 0) {
700                         /* nice, this slave is valid */
701                         LASSERT(req == NULL);
702                         CDEBUG(D_OTHER, "cached\n");
703                         goto release_lock;
704                 }
705
706                 if (rc < 0)
707                         /* error during revalidation */
708                         GOTO(cleanup, rc);
709
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);
713                 req = NULL;
714                 
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;
720                 LASSERT(rc <= 0);
721
722                 if (rc < 0)
723                         /* error during lookup */
724                         GOTO(cleanup, rc);
725               
726                 if (master) {
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;
733                 } else {
734                         /* this is slave. we want to control it */
735                         lock = ldlm_handle2lock(lockh);
736                         LASSERT(lock);
737                         lock->l_ast_data = lmv_get_obj(obj);
738                         LDLM_LOCK_PUT(lock);
739                 }
740
741                 if (*reqp == NULL) {
742                         /* this is first reply, we'll use it to return
743                          * updated data back to the caller */
744                         LASSERT(req);
745                         ptlrpc_request_addref(req);
746                         *reqp = req;
747
748                 }
749
750                 body = lustre_msg_buf(req->rq_repmsg, 1, sizeof(*body));
751                 LASSERT(body);
752                 
753 update:
754                 obj->objs[i].size = body->size;
755                 
756                 CDEBUG(D_OTHER, "fresh: %lu\n",
757                        (unsigned long)obj->objs[i].size);
758
759                 if (req)
760                         ptlrpc_req_finished(req);
761 release_lock:
762                 size += obj->objs[i].size;
763
764                 if (it.d.lustre.it_lock_mode)
765                         ldlm_lock_decref(lockh, it.d.lustre.it_lock_mode);
766         }
767
768         if (*reqp) {
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);
773                 
774                 body = lustre_msg_buf((*reqp)->rq_repmsg, 1, sizeof(*body));
775                 LASSERT(body);
776
777                 /* FIXME: what about other attributes? */
778                 body->size = size;
779                 
780                 if (mreq == NULL) {
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;
786                 }
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;
791                 }
792                 rc = 0;
793         } else {
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;
798                 rc = 1;
799         }
800 cleanup:
801         lmv_unlock_obj(obj);
802         lmv_put_obj(obj);
803         RETURN(rc);
804 }