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