Whamcloud - gitweb
da658b55715912bb6b9af3c457f54c62e11d9563
[fs/lustre-release.git] / lustre / mdc / mdc_locks.c
1 /*
2  * GPL HEADER START
3  *
4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
5  *
6  * This program is free software; you can redistribute it and/or modify
7  * it under the terms of the GNU General Public License version 2 only,
8  * as published by the Free Software Foundation.
9  *
10  * This program is distributed in the hope that it will be useful, but
11  * WITHOUT ANY WARRANTY; without even the implied warranty of
12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
13  * General Public License version 2 for more details (a copy is included
14  * in the LICENSE file that accompanied this code).
15  *
16  * You should have received a copy of the GNU General Public License
17  * version 2 along with this program; If not, see
18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
19  *
20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
21  * CA 95054 USA or visit www.sun.com if you need additional information or
22  * have any questions.
23  *
24  * GPL HEADER END
25  */
26 /*
27  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2011, 2013, Intel Corporation.
31  */
32 /*
33  * This file is part of Lustre, http://www.lustre.org/
34  * Lustre is a trademark of Sun Microsystems, Inc.
35  */
36
37 #define DEBUG_SUBSYSTEM S_MDC
38
39 #ifdef __KERNEL__
40 # include <linux/module.h>
41 # include <linux/lustre_intent.h>
42 #else
43 # include <liblustre.h>
44 #endif
45
46 #include <obd.h>
47 #include <obd_class.h>
48 #include <lustre_dlm.h>
49 #include <lustre_fid.h> /* fid_res_name_eq() */
50 #include <lustre_mdc.h>
51 #include <lustre_net.h>
52 #include <lustre_req_layout.h>
53 #include "mdc_internal.h"
54
55 struct mdc_getattr_args {
56         struct obd_export           *ga_exp;
57         struct md_enqueue_info      *ga_minfo;
58         struct ldlm_enqueue_info    *ga_einfo;
59 };
60
61 int it_open_error(int phase, struct lookup_intent *it)
62 {
63         if (it_disposition(it, DISP_OPEN_LEASE)) {
64                 if (phase >= DISP_OPEN_LEASE)
65                         return it->d.lustre.it_status;
66                 else
67                         return 0;
68         }
69         if (it_disposition(it, DISP_OPEN_OPEN)) {
70                 if (phase >= DISP_OPEN_OPEN)
71                         return it->d.lustre.it_status;
72                 else
73                         return 0;
74         }
75
76         if (it_disposition(it, DISP_OPEN_CREATE)) {
77                 if (phase >= DISP_OPEN_CREATE)
78                         return it->d.lustre.it_status;
79                 else
80                         return 0;
81         }
82
83         if (it_disposition(it, DISP_LOOKUP_EXECD)) {
84                 if (phase >= DISP_LOOKUP_EXECD)
85                         return it->d.lustre.it_status;
86                 else
87                         return 0;
88         }
89
90         if (it_disposition(it, DISP_IT_EXECD)) {
91                 if (phase >= DISP_IT_EXECD)
92                         return it->d.lustre.it_status;
93                 else
94                         return 0;
95         }
96         CERROR("it disp: %X, status: %d\n", it->d.lustre.it_disposition,
97                it->d.lustre.it_status);
98         LBUG();
99         return 0;
100 }
101 EXPORT_SYMBOL(it_open_error);
102
103 /* this must be called on a lockh that is known to have a referenced lock */
104 int mdc_set_lock_data(struct obd_export *exp, __u64 *lockh, void *data,
105                       __u64 *bits)
106 {
107         struct ldlm_lock *lock;
108         struct inode *new_inode = data;
109         ENTRY;
110
111         if(bits)
112                 *bits = 0;
113
114         if (!*lockh)
115                 RETURN(0);
116
117         lock = ldlm_handle2lock((struct lustre_handle *)lockh);
118
119         LASSERT(lock != NULL);
120         lock_res_and_lock(lock);
121 #ifdef __KERNEL__
122         if (lock->l_resource->lr_lvb_inode &&
123             lock->l_resource->lr_lvb_inode != data) {
124                 struct inode *old_inode = lock->l_resource->lr_lvb_inode;
125                 LASSERTF(old_inode->i_state & I_FREEING,
126                          "Found existing inode %p/%lu/%u state %lu in lock: "
127                          "setting data to %p/%lu/%u\n", old_inode,
128                          old_inode->i_ino, old_inode->i_generation,
129                          old_inode->i_state,
130                          new_inode, new_inode->i_ino, new_inode->i_generation);
131         }
132 #endif
133         lock->l_resource->lr_lvb_inode = new_inode;
134         if (bits)
135                 *bits = lock->l_policy_data.l_inodebits.bits;
136
137         unlock_res_and_lock(lock);
138         LDLM_LOCK_PUT(lock);
139
140         RETURN(0);
141 }
142
143 ldlm_mode_t mdc_lock_match(struct obd_export *exp, __u64 flags,
144                            const struct lu_fid *fid, ldlm_type_t type,
145                            ldlm_policy_data_t *policy, ldlm_mode_t mode,
146                            struct lustre_handle *lockh)
147 {
148         struct ldlm_res_id res_id;
149         ldlm_mode_t rc;
150         ENTRY;
151
152         fid_build_reg_res_name(fid, &res_id);
153         /* LU-4405: Clear bits not supported by server */
154         policy->l_inodebits.bits &= exp_connect_ibits(exp);
155         rc = ldlm_lock_match(class_exp2obd(exp)->obd_namespace, flags,
156                              &res_id, type, policy, mode, lockh, 0);
157         RETURN(rc);
158 }
159
160 int mdc_cancel_unused(struct obd_export *exp,
161                       const struct lu_fid *fid,
162                       ldlm_policy_data_t *policy,
163                       ldlm_mode_t mode,
164                       ldlm_cancel_flags_t flags,
165                       void *opaque)
166 {
167         struct ldlm_res_id res_id;
168         struct obd_device *obd = class_exp2obd(exp);
169         int rc;
170
171         ENTRY;
172
173         fid_build_reg_res_name(fid, &res_id);
174         rc = ldlm_cli_cancel_unused_resource(obd->obd_namespace, &res_id,
175                                              policy, mode, flags, opaque);
176         RETURN(rc);
177 }
178
179 int mdc_null_inode(struct obd_export *exp,
180                    const struct lu_fid *fid)
181 {
182         struct ldlm_res_id res_id;
183         struct ldlm_resource *res;
184         struct ldlm_namespace *ns = class_exp2obd(exp)->obd_namespace;
185         ENTRY;
186
187         LASSERTF(ns != NULL, "no namespace passed\n");
188
189         fid_build_reg_res_name(fid, &res_id);
190
191         res = ldlm_resource_get(ns, NULL, &res_id, 0, 0);
192         if(res == NULL)
193                 RETURN(0);
194
195         lock_res(res);
196         res->lr_lvb_inode = NULL;
197         unlock_res(res);
198
199         ldlm_resource_putref(res);
200         RETURN(0);
201 }
202
203 /* find any ldlm lock of the inode in mdc
204  * return 0    not find
205  *        1    find one
206  *      < 0    error */
207 int mdc_find_cbdata(struct obd_export *exp,
208                     const struct lu_fid *fid,
209                     ldlm_iterator_t it, void *data)
210 {
211         struct ldlm_res_id res_id;
212         int rc = 0;
213         ENTRY;
214
215         fid_build_reg_res_name((struct lu_fid*)fid, &res_id);
216         rc = ldlm_resource_iterate(class_exp2obd(exp)->obd_namespace, &res_id,
217                                    it, data);
218         if (rc == LDLM_ITER_STOP)
219                 RETURN(1);
220         else if (rc == LDLM_ITER_CONTINUE)
221                 RETURN(0);
222         RETURN(rc);
223 }
224
225 static inline void mdc_clear_replay_flag(struct ptlrpc_request *req, int rc)
226 {
227         /* Don't hold error requests for replay. */
228         if (req->rq_replay) {
229                 spin_lock(&req->rq_lock);
230                 req->rq_replay = 0;
231                 spin_unlock(&req->rq_lock);
232         }
233         if (rc && req->rq_transno != 0) {
234                 DEBUG_REQ(D_ERROR, req, "transno returned on error rc %d", rc);
235                 LBUG();
236         }
237 }
238
239 /* Save a large LOV EA into the request buffer so that it is available
240  * for replay.  We don't do this in the initial request because the
241  * original request doesn't need this buffer (at most it sends just the
242  * lov_mds_md) and it is a waste of RAM/bandwidth to send the empty
243  * buffer and may also be difficult to allocate and save a very large
244  * request buffer for each open. (bug 5707)
245  *
246  * OOM here may cause recovery failure if lmm is needed (only for the
247  * original open if the MDS crashed just when this client also OOM'd)
248  * but this is incredibly unlikely, and questionable whether the client
249  * could do MDS recovery under OOM anyways... */
250 static void mdc_realloc_openmsg(struct ptlrpc_request *req,
251                                 struct mdt_body *body)
252 {
253         int     rc;
254
255         /* FIXME: remove this explicit offset. */
256         rc = sptlrpc_cli_enlarge_reqbuf(req, DLM_INTENT_REC_OFF + 4,
257                                         body->mbo_eadatasize);
258         if (rc) {
259                 CERROR("Can't enlarge segment %d size to %d\n",
260                        DLM_INTENT_REC_OFF + 4, body->mbo_eadatasize);
261                 body->mbo_valid &= ~OBD_MD_FLEASIZE;
262                 body->mbo_eadatasize = 0;
263         }
264 }
265
266 static struct ptlrpc_request *
267 mdc_intent_open_pack(struct obd_export *exp, struct lookup_intent *it,
268                      struct md_op_data *op_data)
269 {
270         struct ptlrpc_request   *req;
271         struct obd_device       *obddev = class_exp2obd(exp);
272         struct ldlm_intent      *lit;
273         const void              *lmm = op_data->op_data;
274         int                      lmmsize = op_data->op_data_size;
275         struct list_head         cancels = LIST_HEAD_INIT(cancels);
276         int                      count = 0;
277         int                      mode;
278         int                      rc;
279         ENTRY;
280
281         it->it_create_mode = (it->it_create_mode & ~S_IFMT) | S_IFREG;
282
283         /* XXX: openlock is not cancelled for cross-refs. */
284         /* If inode is known, cancel conflicting OPEN locks. */
285         if (fid_is_sane(&op_data->op_fid2)) {
286                 if (it->it_flags & MDS_OPEN_LEASE) { /* try to get lease */
287                         if (it->it_flags & FMODE_WRITE)
288                                 mode = LCK_EX;
289                         else
290                                 mode = LCK_PR;
291                 } else {
292                         if (it->it_flags & (FMODE_WRITE|MDS_OPEN_TRUNC))
293                                 mode = LCK_CW;
294 #ifdef FMODE_EXEC
295                         else if (it->it_flags & FMODE_EXEC)
296                                 mode = LCK_PR;
297 #endif
298                         else
299                                 mode = LCK_CR;
300                 }
301                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
302                                                 &cancels, mode,
303                                                 MDS_INODELOCK_OPEN);
304         }
305
306         /* If CREATE, cancel parent's UPDATE lock. */
307         if (it->it_op & IT_CREAT)
308                 mode = LCK_EX;
309         else
310                 mode = LCK_CR;
311         count += mdc_resource_get_unused(exp, &op_data->op_fid1,
312                                          &cancels, mode,
313                                          MDS_INODELOCK_UPDATE);
314
315         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
316                                    &RQF_LDLM_INTENT_OPEN);
317         if (req == NULL) {
318                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
319                 RETURN(ERR_PTR(-ENOMEM));
320         }
321
322         /* parent capability */
323         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
324         /* child capability, reserve the size according to parent capa, it will
325          * be filled after we get the reply */
326         mdc_set_capa_size(req, &RMF_CAPA2, op_data->op_capa1);
327
328         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
329                              op_data->op_namelen + 1);
330         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
331                              max(lmmsize, obddev->u.cli.cl_default_mds_easize));
332
333         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
334         if (rc < 0) {
335                 ptlrpc_request_free(req);
336                 RETURN(ERR_PTR(rc));
337         }
338
339         spin_lock(&req->rq_lock);
340         req->rq_replay = req->rq_import->imp_replayable;
341         spin_unlock(&req->rq_lock);
342
343         /* pack the intent */
344         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
345         lit->opc = (__u64)it->it_op;
346
347         /* pack the intended request */
348         mdc_open_pack(req, op_data, it->it_create_mode, 0, it->it_flags, lmm,
349                       lmmsize);
350
351         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
352                              obddev->u.cli.cl_max_mds_easize);
353
354         /* for remote client, fetch remote perm for current user */
355         if (client_is_remote(exp))
356                 req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
357                                      sizeof(struct mdt_remote_perm));
358         ptlrpc_request_set_replen(req);
359         return req;
360 }
361
362 static struct ptlrpc_request *
363 mdc_intent_getxattr_pack(struct obd_export *exp,
364                          struct lookup_intent *it,
365                          struct md_op_data *op_data)
366 {
367         struct ptlrpc_request   *req;
368         struct ldlm_intent      *lit;
369         int                     rc, count = 0, maxdata;
370         struct list_head        cancels = LIST_HEAD_INIT(cancels);
371
372         ENTRY;
373
374         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
375                                         &RQF_LDLM_INTENT_GETXATTR);
376         if (req == NULL)
377                 RETURN(ERR_PTR(-ENOMEM));
378
379         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
380
381         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
382         if (rc) {
383                 ptlrpc_request_free(req);
384                 RETURN(ERR_PTR(rc));
385         }
386
387         /* pack the intent */
388         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
389         lit->opc = IT_GETXATTR;
390
391         maxdata = class_exp2cliimp(exp)->imp_connect_data.ocd_max_easize;
392
393         /* pack the intended request */
394         mdc_pack_body(req, &op_data->op_fid1, op_data->op_capa1,
395                         op_data->op_valid, maxdata, -1, 0);
396
397         req_capsule_set_size(&req->rq_pill, &RMF_EADATA,
398                                 RCL_SERVER, maxdata);
399
400         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS,
401                                 RCL_SERVER, maxdata);
402
403         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS_LENS,
404                                 RCL_SERVER, maxdata);
405
406         ptlrpc_request_set_replen(req);
407
408         RETURN(req);
409 }
410
411 static struct ptlrpc_request *mdc_intent_unlink_pack(struct obd_export *exp,
412                                                      struct lookup_intent *it,
413                                                      struct md_op_data *op_data)
414 {
415         struct ptlrpc_request *req;
416         struct obd_device     *obddev = class_exp2obd(exp);
417         struct ldlm_intent    *lit;
418         int                    rc;
419         ENTRY;
420
421         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
422                                    &RQF_LDLM_INTENT_UNLINK);
423         if (req == NULL)
424                 RETURN(ERR_PTR(-ENOMEM));
425
426         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
427         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
428                              op_data->op_namelen + 1);
429
430         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
431         if (rc) {
432                 ptlrpc_request_free(req);
433                 RETURN(ERR_PTR(rc));
434         }
435
436         /* pack the intent */
437         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
438         lit->opc = (__u64)it->it_op;
439
440         /* pack the intended request */
441         mdc_unlink_pack(req, op_data);
442
443         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
444                              obddev->u.cli.cl_default_mds_easize);
445         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
446                              obddev->u.cli.cl_default_mds_cookiesize);
447         ptlrpc_request_set_replen(req);
448         RETURN(req);
449 }
450
451 static struct ptlrpc_request *mdc_intent_getattr_pack(struct obd_export *exp,
452                                                       struct lookup_intent *it,
453                                                       struct md_op_data *op_data)
454 {
455         struct ptlrpc_request *req;
456         struct obd_device     *obddev = class_exp2obd(exp);
457         obd_valid              valid = OBD_MD_FLGETATTR | OBD_MD_FLEASIZE |
458                                        OBD_MD_FLMODEASIZE | OBD_MD_FLDIREA |
459                                        OBD_MD_FLMDSCAPA | OBD_MD_MEA |
460                                        (client_is_remote(exp) ?
461                                                OBD_MD_FLRMTPERM : OBD_MD_FLACL);
462         struct ldlm_intent    *lit;
463         int                    rc;
464         int                     easize;
465         ENTRY;
466
467         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
468                                    &RQF_LDLM_INTENT_GETATTR);
469         if (req == NULL)
470                 RETURN(ERR_PTR(-ENOMEM));
471
472         mdc_set_capa_size(req, &RMF_CAPA1, op_data->op_capa1);
473         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
474                              op_data->op_namelen + 1);
475
476         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
477         if (rc) {
478                 ptlrpc_request_free(req);
479                 RETURN(ERR_PTR(rc));
480         }
481
482         /* pack the intent */
483         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
484         lit->opc = (__u64)it->it_op;
485
486         if (obddev->u.cli.cl_default_mds_easize > 0)
487                 easize = obddev->u.cli.cl_default_mds_easize;
488         else
489                 easize = obddev->u.cli.cl_max_mds_easize;
490
491         /* pack the intended request */
492         mdc_getattr_pack(req, valid, it->it_flags, op_data, easize);
493
494         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, easize);
495         if (client_is_remote(exp))
496                 req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER,
497                                      sizeof(struct mdt_remote_perm));
498         ptlrpc_request_set_replen(req);
499         RETURN(req);
500 }
501
502 static struct ptlrpc_request *mdc_intent_layout_pack(struct obd_export *exp,
503                                                      struct lookup_intent *it,
504                                                      struct md_op_data *unused)
505 {
506         struct obd_device     *obd = class_exp2obd(exp);
507         struct ptlrpc_request *req;
508         struct ldlm_intent    *lit;
509         struct layout_intent  *layout;
510         int rc;
511         ENTRY;
512
513         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
514                                 &RQF_LDLM_INTENT_LAYOUT);
515         if (req == NULL)
516                 RETURN(ERR_PTR(-ENOMEM));
517
518         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, 0);
519         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
520         if (rc) {
521                 ptlrpc_request_free(req);
522                 RETURN(ERR_PTR(rc));
523         }
524
525         /* pack the intent */
526         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
527         lit->opc = (__u64)it->it_op;
528
529         /* pack the layout intent request */
530         layout = req_capsule_client_get(&req->rq_pill, &RMF_LAYOUT_INTENT);
531         /* LAYOUT_INTENT_ACCESS is generic, specific operation will be
532          * set for replication */
533         layout->li_opc = LAYOUT_INTENT_ACCESS;
534
535         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
536                              obd->u.cli.cl_default_mds_easize);
537         ptlrpc_request_set_replen(req);
538         RETURN(req);
539 }
540
541 static struct ptlrpc_request *
542 mdc_enqueue_pack(struct obd_export *exp, int lvb_len)
543 {
544         struct ptlrpc_request *req;
545         int rc;
546         ENTRY;
547
548         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LDLM_ENQUEUE);
549         if (req == NULL)
550                 RETURN(ERR_PTR(-ENOMEM));
551
552         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
553         if (rc) {
554                 ptlrpc_request_free(req);
555                 RETURN(ERR_PTR(rc));
556         }
557
558         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, lvb_len);
559         ptlrpc_request_set_replen(req);
560         RETURN(req);
561 }
562
563 static int mdc_finish_enqueue(struct obd_export *exp,
564                               struct ptlrpc_request *req,
565                               struct ldlm_enqueue_info *einfo,
566                               struct lookup_intent *it,
567                               struct lustre_handle *lockh,
568                               int rc)
569 {
570         struct req_capsule  *pill = &req->rq_pill;
571         struct ldlm_request *lockreq;
572         struct ldlm_reply   *lockrep;
573         struct lustre_intent_data *intent = &it->d.lustre;
574         struct ldlm_lock    *lock;
575         void                *lvb_data = NULL;
576         int                  lvb_len = 0;
577         ENTRY;
578
579         LASSERT(rc >= 0);
580         /* Similarly, if we're going to replay this request, we don't want to
581          * actually get a lock, just perform the intent. */
582         if (req->rq_transno || req->rq_replay) {
583                 lockreq = req_capsule_client_get(pill, &RMF_DLM_REQ);
584                 lockreq->lock_flags |= ldlm_flags_to_wire(LDLM_FL_INTENT_ONLY);
585         }
586
587         if (rc == ELDLM_LOCK_ABORTED) {
588                 einfo->ei_mode = 0;
589                 memset(lockh, 0, sizeof(*lockh));
590                 rc = 0;
591         } else { /* rc = 0 */
592                 lock = ldlm_handle2lock(lockh);
593                 LASSERT(lock != NULL);
594
595                 /* If the server gave us back a different lock mode, we should
596                  * fix up our variables. */
597                 if (lock->l_req_mode != einfo->ei_mode) {
598                         ldlm_lock_addref(lockh, lock->l_req_mode);
599                         ldlm_lock_decref(lockh, einfo->ei_mode);
600                         einfo->ei_mode = lock->l_req_mode;
601                 }
602                 LDLM_LOCK_PUT(lock);
603         }
604
605         lockrep = req_capsule_server_get(pill, &RMF_DLM_REP);
606         LASSERT(lockrep != NULL); /* checked by ldlm_cli_enqueue() */
607
608         intent->it_disposition = (int)lockrep->lock_policy_res1;
609         intent->it_status = (int)lockrep->lock_policy_res2;
610         intent->it_lock_mode = einfo->ei_mode;
611         intent->it_lock_handle = lockh->cookie;
612         intent->it_data = req;
613
614         /* Technically speaking rq_transno must already be zero if
615          * it_status is in error, so the check is a bit redundant */
616         if ((!req->rq_transno || intent->it_status < 0) && req->rq_replay)
617                 mdc_clear_replay_flag(req, intent->it_status);
618
619         /* If we're doing an IT_OPEN which did not result in an actual
620          * successful open, then we need to remove the bit which saves
621          * this request for unconditional replay.
622          *
623          * It's important that we do this first!  Otherwise we might exit the
624          * function without doing so, and try to replay a failed create
625          * (bug 3440) */
626         if (it->it_op & IT_OPEN && req->rq_replay &&
627             (!it_disposition(it, DISP_OPEN_OPEN) ||intent->it_status != 0))
628                 mdc_clear_replay_flag(req, intent->it_status);
629
630         DEBUG_REQ(D_RPCTRACE, req, "op: %d disposition: %x, status: %d",
631                   it->it_op, intent->it_disposition, intent->it_status);
632
633         /* We know what to expect, so we do any byte flipping required here */
634         if (it->it_op & (IT_OPEN | IT_UNLINK | IT_LOOKUP | IT_GETATTR)) {
635                 struct mdt_body *body;
636
637                 body = req_capsule_server_get(pill, &RMF_MDT_BODY);
638                 if (body == NULL) {
639                         CERROR ("Can't swab mdt_body\n");
640                         RETURN (-EPROTO);
641                 }
642
643                 if (it_disposition(it, DISP_OPEN_OPEN) &&
644                     !it_open_error(DISP_OPEN_OPEN, it)) {
645                         /*
646                          * If this is a successful OPEN request, we need to set
647                          * replay handler and data early, so that if replay
648                          * happens immediately after swabbing below, new reply
649                          * is swabbed by that handler correctly.
650                          */
651                         mdc_set_open_replay_data(NULL, NULL, it);
652                 }
653
654                 if (body->mbo_valid & (OBD_MD_FLDIREA | OBD_MD_FLEASIZE)) {
655                         void *eadata;
656
657                         mdc_update_max_ea_from_body(exp, body);
658
659                         /*
660                          * The eadata is opaque; just check that it is there.
661                          * Eventually, obd_unpackmd() will check the contents.
662                          */
663                         eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
664                                                         body->mbo_eadatasize);
665                         if (eadata == NULL)
666                                 RETURN(-EPROTO);
667
668                         /* save lvb data and length in case this is for layout
669                          * lock */
670                         lvb_data = eadata;
671                         lvb_len = body->mbo_eadatasize;
672
673                         /*
674                          * We save the reply LOV EA in case we have to replay a
675                          * create for recovery.  If we didn't allocate a large
676                          * enough request buffer above we need to reallocate it
677                          * here to hold the actual LOV EA.
678                          *
679                          * To not save LOV EA if request is not going to replay
680                          * (for example error one).
681                          */
682                         if ((it->it_op & IT_OPEN) && req->rq_replay) {
683                                 void *lmm;
684                                 if (req_capsule_get_size(pill, &RMF_EADATA,
685                                                          RCL_CLIENT) <
686                                     body->mbo_eadatasize)
687                                         mdc_realloc_openmsg(req, body);
688                                 else
689                                         req_capsule_shrink(pill, &RMF_EADATA,
690                                                            body->mbo_eadatasize,
691                                                            RCL_CLIENT);
692
693                                 req_capsule_set_size(pill, &RMF_EADATA,
694                                                      RCL_CLIENT,
695                                                      body->mbo_eadatasize);
696
697                                 lmm = req_capsule_client_get(pill, &RMF_EADATA);
698                                 if (lmm)
699                                         memcpy(lmm, eadata,
700                                                body->mbo_eadatasize);
701                         }
702                 }
703
704                 if (body->mbo_valid & OBD_MD_FLRMTPERM) {
705                         struct mdt_remote_perm *perm;
706
707                         LASSERT(client_is_remote(exp));
708                         perm = req_capsule_server_swab_get(pill, &RMF_ACL,
709                                                 lustre_swab_mdt_remote_perm);
710                         if (perm == NULL)
711                                 RETURN(-EPROTO);
712                 }
713                 if (body->mbo_valid & OBD_MD_FLMDSCAPA) {
714                         struct lustre_capa *capa, *p;
715
716                         capa = req_capsule_server_get(pill, &RMF_CAPA1);
717                         if (capa == NULL)
718                                 RETURN(-EPROTO);
719
720                         if (it->it_op & IT_OPEN) {
721                                 /* client fid capa will be checked in replay */
722                                 p = req_capsule_client_get(pill, &RMF_CAPA2);
723                                 LASSERT(p);
724                                 *p = *capa;
725                         }
726                 }
727                 if (body->mbo_valid & OBD_MD_FLOSSCAPA) {
728                         struct lustre_capa *capa;
729
730                         capa = req_capsule_server_get(pill, &RMF_CAPA2);
731                         if (capa == NULL)
732                                 RETURN(-EPROTO);
733                 }
734         } else if (it->it_op & IT_LAYOUT) {
735                 /* maybe the lock was granted right away and layout
736                  * is packed into RMF_DLM_LVB of req */
737                 lvb_len = req_capsule_get_size(pill, &RMF_DLM_LVB, RCL_SERVER);
738                 if (lvb_len > 0) {
739                         lvb_data = req_capsule_server_sized_get(pill,
740                                                         &RMF_DLM_LVB, lvb_len);
741                         if (lvb_data == NULL)
742                                 RETURN(-EPROTO);
743                 }
744         }
745
746         /* fill in stripe data for layout lock */
747         lock = ldlm_handle2lock(lockh);
748         if (lock != NULL && ldlm_has_layout(lock) && lvb_data != NULL) {
749                 void *lmm;
750
751                 LDLM_DEBUG(lock, "layout lock returned by: %s, lvb_len: %d\n",
752                         ldlm_it2str(it->it_op), lvb_len);
753
754                 OBD_ALLOC_LARGE(lmm, lvb_len);
755                 if (lmm == NULL) {
756                         LDLM_LOCK_PUT(lock);
757                         RETURN(-ENOMEM);
758                 }
759                 memcpy(lmm, lvb_data, lvb_len);
760
761                 /* install lvb_data */
762                 lock_res_and_lock(lock);
763                 if (lock->l_lvb_data == NULL) {
764                         lock->l_lvb_type = LVB_T_LAYOUT;
765                         lock->l_lvb_data = lmm;
766                         lock->l_lvb_len = lvb_len;
767                         lmm = NULL;
768                 }
769                 unlock_res_and_lock(lock);
770                 if (lmm != NULL)
771                         OBD_FREE_LARGE(lmm, lvb_len);
772         }
773         if (lock != NULL)
774                 LDLM_LOCK_PUT(lock);
775
776         RETURN(rc);
777 }
778
779 /* We always reserve enough space in the reply packet for a stripe MD, because
780  * we don't know in advance the file type. */
781 int mdc_enqueue(struct obd_export *exp,
782                 struct ldlm_enqueue_info *einfo,
783                 const union ldlm_policy_data *policy,
784                 struct lookup_intent *it, struct md_op_data *op_data,
785                 struct lustre_handle *lockh, __u64 extra_lock_flags)
786 {
787         struct obd_device     *obddev = class_exp2obd(exp);
788         struct ptlrpc_request *req = NULL;
789         __u64                  flags, saved_flags = extra_lock_flags;
790         int                    rc;
791         struct ldlm_res_id res_id;
792         static const ldlm_policy_data_t lookup_policy =
793                             { .l_inodebits = { MDS_INODELOCK_LOOKUP } };
794         static const ldlm_policy_data_t update_policy =
795                             { .l_inodebits = { MDS_INODELOCK_UPDATE } };
796         static const ldlm_policy_data_t layout_policy =
797                             { .l_inodebits = { MDS_INODELOCK_LAYOUT } };
798         static const ldlm_policy_data_t getxattr_policy = {
799                               .l_inodebits = { MDS_INODELOCK_XATTR } };
800         int                    generation, resends = 0;
801         struct ldlm_reply     *lockrep;
802         enum lvb_type          lvb_type = 0;
803         ENTRY;
804
805         LASSERTF(!it || einfo->ei_type == LDLM_IBITS, "lock type %d\n",
806                  einfo->ei_type);
807         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
808
809         if (it != NULL) {
810                 LASSERT(policy == NULL);
811
812                 saved_flags |= LDLM_FL_HAS_INTENT;
813                 if (it->it_op & (IT_OPEN | IT_UNLINK | IT_GETATTR | IT_READDIR))
814                         policy = &update_policy;
815                 else if (it->it_op & IT_LAYOUT)
816                         policy = &layout_policy;
817                 else if (it->it_op & (IT_GETXATTR | IT_SETXATTR))
818                         policy = &getxattr_policy;
819                 else
820                         policy = &lookup_policy;
821         }
822
823         generation = obddev->u.cli.cl_import->imp_generation;
824 resend:
825         flags = saved_flags;
826         if (it == NULL) {
827                 /* The only way right now is FLOCK. */
828                 LASSERTF(einfo->ei_type == LDLM_FLOCK, "lock type %d\n",
829                          einfo->ei_type);
830                 res_id.name[3] = LDLM_FLOCK;
831         } else if (it->it_op & IT_OPEN) {
832                 req = mdc_intent_open_pack(exp, it, op_data);
833         } else if (it->it_op & IT_UNLINK) {
834                 req = mdc_intent_unlink_pack(exp, it, op_data);
835         } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
836                 req = mdc_intent_getattr_pack(exp, it, op_data);
837         } else if (it->it_op & IT_READDIR) {
838                 req = mdc_enqueue_pack(exp, 0);
839         } else if (it->it_op & IT_LAYOUT) {
840                 if (!imp_connect_lvb_type(class_exp2cliimp(exp)))
841                         RETURN(-EOPNOTSUPP);
842                 req = mdc_intent_layout_pack(exp, it, op_data);
843                 lvb_type = LVB_T_LAYOUT;
844         } else if (it->it_op & IT_GETXATTR) {
845                 req = mdc_intent_getxattr_pack(exp, it, op_data);
846         } else {
847                 LBUG();
848                 RETURN(-EINVAL);
849         }
850
851         if (IS_ERR(req))
852                 RETURN(PTR_ERR(req));
853
854         if (req != NULL && it && it->it_op & IT_CREAT)
855                 /* ask ptlrpc not to resend on EINPROGRESS since we have our own
856                  * retry logic */
857                 req->rq_no_retry_einprogress = 1;
858
859         if (resends) {
860                 req->rq_generation_set = 1;
861                 req->rq_import_generation = generation;
862                 req->rq_sent = cfs_time_current_sec() + resends;
863         }
864
865         /* It is important to obtain rpc_lock first (if applicable), so that
866          * threads that are serialised with rpc_lock are not polluting our
867          * rpcs in flight counter. We do not do flock request limiting, though*/
868         if (it) {
869                 mdc_get_rpc_lock(obddev->u.cli.cl_rpc_lock, it);
870                 rc = obd_get_request_slot(&obddev->u.cli);
871                 if (rc != 0) {
872                         mdc_put_rpc_lock(obddev->u.cli.cl_rpc_lock, it);
873                         mdc_clear_replay_flag(req, 0);
874                         ptlrpc_req_finished(req);
875                         RETURN(rc);
876                 }
877         }
878
879         rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, policy, &flags, NULL,
880                               0, lvb_type, lockh, 0);
881         if (!it) {
882                 /* For flock requests we immediatelly return without further
883                    delay and let caller deal with the rest, since rest of
884                    this function metadata processing makes no sense for flock
885                    requests anyway. But in case of problem during comms with
886                    Server (ETIMEDOUT) or any signal/kill attempt (EINTR), we
887                    can not rely on caller and this mainly for F_UNLCKs
888                    (explicits or automatically generated by Kernel to clean
889                    current FLocks upon exit) that can't be trashed */
890                 if (((rc == -EINTR) || (rc == -ETIMEDOUT)) &&
891                     (einfo->ei_type == LDLM_FLOCK) &&
892                     (einfo->ei_mode == LCK_NL))
893                         goto resend;
894                 RETURN(rc);
895         }
896
897         obd_put_request_slot(&obddev->u.cli);
898         mdc_put_rpc_lock(obddev->u.cli.cl_rpc_lock, it);
899
900         if (rc < 0) {
901                 CDEBUG_LIMIT((rc == -EACCES || rc == -EIDRM) ? D_INFO : D_ERROR,
902                              "%s: ldlm_cli_enqueue failed: rc = %d\n",
903                              obddev->obd_name, rc);
904
905                 mdc_clear_replay_flag(req, rc);
906                 ptlrpc_req_finished(req);
907                 RETURN(rc);
908         }
909
910         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
911         LASSERT(lockrep != NULL);
912
913         lockrep->lock_policy_res2 =
914                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
915
916         /* Retry the create infinitely when we get -EINPROGRESS from
917          * server. This is required by the new quota design. */
918         if (it && it->it_op & IT_CREAT &&
919             (int)lockrep->lock_policy_res2 == -EINPROGRESS) {
920                 mdc_clear_replay_flag(req, rc);
921                 ptlrpc_req_finished(req);
922                 resends++;
923
924                 CDEBUG(D_HA, "%s: resend:%d op:%d "DFID"/"DFID"\n",
925                        obddev->obd_name, resends, it->it_op,
926                        PFID(&op_data->op_fid1), PFID(&op_data->op_fid2));
927
928                 if (generation == obddev->u.cli.cl_import->imp_generation) {
929                         goto resend;
930                 } else {
931                         CDEBUG(D_HA, "resend cross eviction\n");
932                         RETURN(-EIO);
933                 }
934         }
935
936         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
937         if (rc < 0) {
938                 if (lustre_handle_is_used(lockh)) {
939                         ldlm_lock_decref(lockh, einfo->ei_mode);
940                         memset(lockh, 0, sizeof(*lockh));
941                 }
942                 ptlrpc_req_finished(req);
943         }
944         RETURN(rc);
945 }
946
947 static int mdc_finish_intent_lock(struct obd_export *exp,
948                                   struct ptlrpc_request *request,
949                                   struct md_op_data *op_data,
950                                   struct lookup_intent *it,
951                                   struct lustre_handle *lockh)
952 {
953         struct lustre_handle old_lock;
954         struct mdt_body *mdt_body;
955         struct ldlm_lock *lock;
956         int rc;
957         ENTRY;
958
959         LASSERT(request != NULL);
960         LASSERT(request != LP_POISON);
961         LASSERT(request->rq_repmsg != LP_POISON);
962
963         if (it->it_op & IT_READDIR)
964                 RETURN(0);
965
966         if (!it_disposition(it, DISP_IT_EXECD)) {
967                 /* The server failed before it even started executing the
968                  * intent, i.e. because it couldn't unpack the request. */
969                 LASSERT(it->d.lustre.it_status != 0);
970                 RETURN(it->d.lustre.it_status);
971         }
972         rc = it_open_error(DISP_IT_EXECD, it);
973         if (rc)
974                 RETURN(rc);
975
976         mdt_body = req_capsule_server_get(&request->rq_pill, &RMF_MDT_BODY);
977         LASSERT(mdt_body != NULL);      /* mdc_enqueue checked */
978
979         /* If we were revalidating a fid/name pair, mark the intent in
980          * case we fail and get called again from lookup */
981         if (fid_is_sane(&op_data->op_fid2) &&
982             it->it_create_mode & M_CHECK_STALE &&
983             it->it_op != IT_GETATTR) {
984                 /* Also: did we find the same inode? */
985                 /* sever can return one of two fids:
986                  * op_fid2 - new allocated fid - if file is created.
987                  * op_fid3 - existent fid - if file only open.
988                  * op_fid3 is saved in lmv_intent_open */
989                 if ((!lu_fid_eq(&op_data->op_fid2, &mdt_body->mbo_fid1)) &&
990                     (!lu_fid_eq(&op_data->op_fid3, &mdt_body->mbo_fid1))) {
991                         CDEBUG(D_DENTRY, "Found stale data "DFID"("DFID")/"DFID
992                                "\n", PFID(&op_data->op_fid2),
993                                PFID(&op_data->op_fid2),
994                                PFID(&mdt_body->mbo_fid1));
995                         RETURN(-ESTALE);
996                 }
997         }
998
999         rc = it_open_error(DISP_LOOKUP_EXECD, it);
1000         if (rc)
1001                 RETURN(rc);
1002
1003         /* keep requests around for the multiple phases of the call
1004          * this shows the DISP_XX must guarantee we make it into the call
1005          */
1006         if (!it_disposition(it, DISP_ENQ_CREATE_REF) &&
1007             it_disposition(it, DISP_OPEN_CREATE) &&
1008             !it_open_error(DISP_OPEN_CREATE, it)) {
1009                 it_set_disposition(it, DISP_ENQ_CREATE_REF);
1010                 ptlrpc_request_addref(request); /* balanced in ll_create_node */
1011         }
1012         if (!it_disposition(it, DISP_ENQ_OPEN_REF) &&
1013             it_disposition(it, DISP_OPEN_OPEN) &&
1014             !it_open_error(DISP_OPEN_OPEN, it)) {
1015                 it_set_disposition(it, DISP_ENQ_OPEN_REF);
1016                 ptlrpc_request_addref(request); /* balanced in ll_file_open */
1017                 /* BUG 11546 - eviction in the middle of open rpc processing */
1018                 OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_ENQUEUE_PAUSE, obd_timeout);
1019         }
1020
1021         if (it->it_op & IT_CREAT) {
1022                 /* XXX this belongs in ll_create_it */
1023         } else if (it->it_op == IT_OPEN) {
1024                 LASSERT(!it_disposition(it, DISP_OPEN_CREATE));
1025         } else {
1026                 LASSERT(it->it_op & (IT_GETATTR | IT_LOOKUP | IT_LAYOUT));
1027         }
1028
1029         /* If we already have a matching lock, then cancel the new
1030          * one.  We have to set the data here instead of in
1031          * mdc_enqueue, because we need to use the child's inode as
1032          * the l_ast_data to match, and that's not available until
1033          * intent_finish has performed the iget().) */
1034         lock = ldlm_handle2lock(lockh);
1035         if (lock) {
1036                 ldlm_policy_data_t policy = lock->l_policy_data;
1037                 LDLM_DEBUG(lock, "matching against this");
1038
1039                 LASSERTF(fid_res_name_eq(&mdt_body->mbo_fid1,
1040                                          &lock->l_resource->lr_name),
1041                          "Lock res_id: "DLDLMRES", fid: "DFID"\n",
1042                          PLDLMRES(lock->l_resource), PFID(&mdt_body->mbo_fid1));
1043                 LDLM_LOCK_PUT(lock);
1044
1045                 memcpy(&old_lock, lockh, sizeof(*lockh));
1046                 if (ldlm_lock_match(NULL, LDLM_FL_BLOCK_GRANTED, NULL,
1047                                     LDLM_IBITS, &policy, LCK_NL, &old_lock, 0)) {
1048                         ldlm_lock_decref_and_cancel(lockh,
1049                                                     it->d.lustre.it_lock_mode);
1050                         memcpy(lockh, &old_lock, sizeof(old_lock));
1051                         it->d.lustre.it_lock_handle = lockh->cookie;
1052                 }
1053         }
1054         CDEBUG(D_DENTRY,"D_IT dentry %.*s intent: %s status %d disp %x rc %d\n",
1055                op_data->op_namelen, op_data->op_name, ldlm_it2str(it->it_op),
1056                it->d.lustre.it_status, it->d.lustre.it_disposition, rc);
1057         RETURN(rc);
1058 }
1059
1060 int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
1061                         struct lu_fid *fid, __u64 *bits)
1062 {
1063         /* We could just return 1 immediately, but since we should only
1064          * be called in revalidate_it if we already have a lock, let's
1065          * verify that. */
1066         struct ldlm_res_id res_id;
1067         struct lustre_handle lockh;
1068         ldlm_policy_data_t policy;
1069         ldlm_mode_t mode;
1070         ENTRY;
1071
1072         if (it->d.lustre.it_lock_handle) {
1073                 lockh.cookie = it->d.lustre.it_lock_handle;
1074                 mode = ldlm_revalidate_lock_handle(&lockh, bits);
1075         } else {
1076                 fid_build_reg_res_name(fid, &res_id);
1077                 switch (it->it_op) {
1078                 case IT_GETATTR:
1079                         /* File attributes are held under multiple bits:
1080                          * nlink is under lookup lock, size and times are
1081                          * under UPDATE lock and recently we've also got
1082                          * a separate permissions lock for owner/group/acl that
1083                          * were protected by lookup lock before.
1084                          * Getattr must provide all of that information,
1085                          * so we need to ensure we have all of those locks.
1086                          * Unfortunately, if the bits are split across multiple
1087                          * locks, there's no easy way to match all of them here,
1088                          * so an extra RPC would be performed to fetch all
1089                          * of those bits at once for now. */
1090                         /* For new MDTs(> 2.4), UPDATE|PERM should be enough,
1091                          * but for old MDTs (< 2.4), permission is covered
1092                          * by LOOKUP lock, so it needs to match all bits here.*/
1093                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE |
1094                                                   MDS_INODELOCK_LOOKUP |
1095                                                   MDS_INODELOCK_PERM;
1096                         break;
1097                 case IT_READDIR:
1098                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
1099                         break;
1100                 case IT_LAYOUT:
1101                         policy.l_inodebits.bits = MDS_INODELOCK_LAYOUT;
1102                         break;
1103                 default:
1104                         policy.l_inodebits.bits = MDS_INODELOCK_LOOKUP;
1105                         break;
1106                 }
1107
1108                 mode = mdc_lock_match(exp, LDLM_FL_BLOCK_GRANTED, fid,
1109                                       LDLM_IBITS, &policy,
1110                                       LCK_CR | LCK_CW | LCK_PR | LCK_PW,
1111                                       &lockh);
1112         }
1113
1114         if (mode) {
1115                 it->d.lustre.it_lock_handle = lockh.cookie;
1116                 it->d.lustre.it_lock_mode = mode;
1117         } else {
1118                 it->d.lustre.it_lock_handle = 0;
1119                 it->d.lustre.it_lock_mode = 0;
1120         }
1121
1122         RETURN(!!mode);
1123 }
1124
1125 /*
1126  * This long block is all about fixing up the lock and request state
1127  * so that it is correct as of the moment _before_ the operation was
1128  * applied; that way, the VFS will think that everything is normal and
1129  * call Lustre's regular VFS methods.
1130  *
1131  * If we're performing a creation, that means that unless the creation
1132  * failed with EEXIST, we should fake up a negative dentry.
1133  *
1134  * For everything else, we want to lookup to succeed.
1135  *
1136  * One additional note: if CREATE or OPEN succeeded, we add an extra
1137  * reference to the request because we need to keep it around until
1138  * ll_create/ll_open gets called.
1139  *
1140  * The server will return to us, in it_disposition, an indication of
1141  * exactly what d.lustre.it_status refers to.
1142  *
1143  * If DISP_OPEN_OPEN is set, then d.lustre.it_status refers to the open() call,
1144  * otherwise if DISP_OPEN_CREATE is set, then it status is the
1145  * creation failure mode.  In either case, one of DISP_LOOKUP_NEG or
1146  * DISP_LOOKUP_POS will be set, indicating whether the child lookup
1147  * was successful.
1148  *
1149  * Else, if DISP_LOOKUP_EXECD then d.lustre.it_status is the rc of the
1150  * child lookup.
1151  */
1152 int mdc_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
1153                     struct lookup_intent *it, struct ptlrpc_request **reqp,
1154                     ldlm_blocking_callback cb_blocking, __u64 extra_lock_flags)
1155 {
1156         struct ldlm_enqueue_info einfo = {
1157                 .ei_type        = LDLM_IBITS,
1158                 .ei_mode        = it_to_lock_mode(it),
1159                 .ei_cb_bl       = cb_blocking,
1160                 .ei_cb_cp       = ldlm_completion_ast,
1161         };
1162         struct lustre_handle lockh;
1163         int rc = 0;
1164         ENTRY;
1165         LASSERT(it);
1166
1167         CDEBUG(D_DLMTRACE, "(name: %.*s,"DFID") in obj "DFID
1168                 ", intent: %s flags %#"LPF64"o\n", op_data->op_namelen,
1169                 op_data->op_name, PFID(&op_data->op_fid2),
1170                 PFID(&op_data->op_fid1), ldlm_it2str(it->it_op),
1171                 it->it_flags);
1172
1173         lockh.cookie = 0;
1174         if (fid_is_sane(&op_data->op_fid2) &&
1175             (it->it_op & (IT_LOOKUP | IT_GETATTR | IT_READDIR))) {
1176                 /* We could just return 1 immediately, but since we should only
1177                  * be called in revalidate_it if we already have a lock, let's
1178                  * verify that. */
1179                 it->d.lustre.it_lock_handle = 0;
1180                 rc = mdc_revalidate_lock(exp, it, &op_data->op_fid2, NULL);
1181                 /* Only return failure if it was not GETATTR by cfid
1182                    (from inode_revalidate) */
1183                 if (rc || op_data->op_namelen != 0)
1184                         RETURN(rc);
1185         }
1186
1187         /* For case if upper layer did not alloc fid, do it now. */
1188         if (!fid_is_sane(&op_data->op_fid2) && it->it_op & IT_CREAT) {
1189                 rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
1190                 if (rc < 0) {
1191                         CERROR("Can't alloc new fid, rc %d\n", rc);
1192                         RETURN(rc);
1193                 }
1194         }
1195
1196         rc = mdc_enqueue(exp, &einfo, NULL, it, op_data, &lockh,
1197                          extra_lock_flags);
1198         if (rc < 0)
1199                 RETURN(rc);
1200
1201         *reqp = it->d.lustre.it_data;
1202         rc = mdc_finish_intent_lock(exp, *reqp, op_data, it, &lockh);
1203         RETURN(rc);
1204 }
1205
1206 static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
1207                                               struct ptlrpc_request *req,
1208                                               void *args, int rc)
1209 {
1210         struct mdc_getattr_args  *ga = args;
1211         struct obd_export        *exp = ga->ga_exp;
1212         struct md_enqueue_info   *minfo = ga->ga_minfo;
1213         struct ldlm_enqueue_info *einfo = ga->ga_einfo;
1214         struct lookup_intent     *it;
1215         struct lustre_handle     *lockh;
1216         struct obd_device        *obddev;
1217         struct ldlm_reply        *lockrep;
1218         __u64                     flags = LDLM_FL_HAS_INTENT;
1219         ENTRY;
1220
1221         it    = &minfo->mi_it;
1222         lockh = &minfo->mi_lockh;
1223
1224         obddev = class_exp2obd(exp);
1225
1226         obd_put_request_slot(&obddev->u.cli);
1227         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
1228                 rc = -ETIMEDOUT;
1229
1230         rc = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, 1, einfo->ei_mode,
1231                                    &flags, NULL, 0, lockh, rc);
1232         if (rc < 0) {
1233                 CERROR("ldlm_cli_enqueue_fini: %d\n", rc);
1234                 mdc_clear_replay_flag(req, rc);
1235                 GOTO(out, rc);
1236         }
1237
1238         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1239         LASSERT(lockrep != NULL);
1240
1241         lockrep->lock_policy_res2 =
1242                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
1243
1244         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
1245         if (rc)
1246                 GOTO(out, rc);
1247
1248         rc = mdc_finish_intent_lock(exp, req, &minfo->mi_data, it, lockh);
1249         EXIT;
1250
1251 out:
1252         OBD_FREE_PTR(einfo);
1253         minfo->mi_cb(req, minfo, rc);
1254         return 0;
1255 }
1256
1257 int mdc_intent_getattr_async(struct obd_export *exp,
1258                              struct md_enqueue_info *minfo,
1259                              struct ldlm_enqueue_info *einfo)
1260 {
1261         struct md_op_data       *op_data = &minfo->mi_data;
1262         struct lookup_intent    *it = &minfo->mi_it;
1263         struct ptlrpc_request   *req;
1264         struct mdc_getattr_args *ga;
1265         struct obd_device       *obddev = class_exp2obd(exp);
1266         struct ldlm_res_id       res_id;
1267         /*XXX: Both MDS_INODELOCK_LOOKUP and MDS_INODELOCK_UPDATE are needed
1268          *     for statahead currently. Consider CMD in future, such two bits
1269          *     maybe managed by different MDS, should be adjusted then. */
1270         ldlm_policy_data_t       policy = {
1271                                         .l_inodebits = { MDS_INODELOCK_LOOKUP | 
1272                                                          MDS_INODELOCK_UPDATE }
1273                                  };
1274         int                      rc = 0;
1275         __u64                    flags = LDLM_FL_HAS_INTENT;
1276         ENTRY;
1277
1278         CDEBUG(D_DLMTRACE, "name: %.*s in inode "DFID", intent: %s flags %#"
1279                 LPF64"o\n",
1280                 op_data->op_namelen, op_data->op_name, PFID(&op_data->op_fid1),
1281                 ldlm_it2str(it->it_op), it->it_flags);
1282
1283         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
1284         req = mdc_intent_getattr_pack(exp, it, op_data);
1285         if (IS_ERR(req))
1286                 RETURN(PTR_ERR(req));
1287
1288         rc = obd_get_request_slot(&obddev->u.cli);
1289         if (rc != 0) {
1290                 ptlrpc_req_finished(req);
1291                 RETURN(rc);
1292         }
1293
1294         rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, &policy, &flags, NULL,
1295                               0, LVB_T_NONE, &minfo->mi_lockh, 1);
1296         if (rc < 0) {
1297                 obd_put_request_slot(&obddev->u.cli);
1298                 ptlrpc_req_finished(req);
1299                 RETURN(rc);
1300         }
1301
1302         CLASSERT(sizeof(*ga) <= sizeof(req->rq_async_args));
1303         ga = ptlrpc_req_async_args(req);
1304         ga->ga_exp = exp;
1305         ga->ga_minfo = minfo;
1306         ga->ga_einfo = einfo;
1307
1308         req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
1309         ptlrpcd_add_req(req, PDL_POLICY_LOCAL, -1);
1310
1311         RETURN(0);
1312 }