Whamcloud - gitweb
e9c794e9115db43aaeed36db3d043d620f48011e
[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, 2017, 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 #include <lustre_acl.h>
47
48 #include "mdc_internal.h"
49
50 struct mdc_getattr_args {
51         struct obd_export               *ga_exp;
52         struct md_enqueue_info          *ga_minfo;
53 };
54
55 int it_open_error(int phase, struct lookup_intent *it)
56 {
57         if (it_disposition(it, DISP_OPEN_LEASE)) {
58                 if (phase >= DISP_OPEN_LEASE)
59                         return it->it_status;
60                 else
61                         return 0;
62         }
63         if (it_disposition(it, DISP_OPEN_OPEN)) {
64                 if (phase >= DISP_OPEN_OPEN)
65                         return it->it_status;
66                 else
67                         return 0;
68         }
69
70         if (it_disposition(it, DISP_OPEN_CREATE)) {
71                 if (phase >= DISP_OPEN_CREATE)
72                         return it->it_status;
73                 else
74                         return 0;
75         }
76
77         if (it_disposition(it, DISP_LOOKUP_EXECD)) {
78                 if (phase >= DISP_LOOKUP_EXECD)
79                         return it->it_status;
80                 else
81                         return 0;
82         }
83
84         if (it_disposition(it, DISP_IT_EXECD)) {
85                 if (phase >= DISP_IT_EXECD)
86                         return it->it_status;
87                 else
88                         return 0;
89         }
90
91         CERROR("it disp: %X, status: %d\n", it->it_disposition, it->it_status);
92         LBUG();
93
94         return 0;
95 }
96 EXPORT_SYMBOL(it_open_error);
97
98 /* this must be called on a lockh that is known to have a referenced lock */
99 int mdc_set_lock_data(struct obd_export *exp, const struct lustre_handle *lockh,
100                       void *data, __u64 *bits)
101 {
102         struct ldlm_lock *lock;
103         struct inode *new_inode = data;
104
105         ENTRY;
106         if (bits)
107                 *bits = 0;
108
109         if (!lustre_handle_is_used(lockh))
110                 RETURN(0);
111
112         lock = ldlm_handle2lock(lockh);
113
114         LASSERT(lock != NULL);
115         lock_res_and_lock(lock);
116         if (lock->l_resource->lr_lvb_inode &&
117             lock->l_resource->lr_lvb_inode != data) {
118                 struct inode *old_inode = lock->l_resource->lr_lvb_inode;
119
120                 LASSERTF(old_inode->i_state & I_FREEING,
121                          "Found existing inode %p/%lu/%u state %lu in lock: setting data to %p/%lu/%u\n",
122                          old_inode, old_inode->i_ino, old_inode->i_generation,
123                          old_inode->i_state,
124                          new_inode, new_inode->i_ino, new_inode->i_generation);
125         }
126         lock->l_resource->lr_lvb_inode = new_inode;
127         if (bits)
128                 *bits = lock->l_policy_data.l_inodebits.bits;
129
130         unlock_res_and_lock(lock);
131         LDLM_LOCK_PUT(lock);
132
133         RETURN(0);
134 }
135
136 enum ldlm_mode mdc_lock_match(struct obd_export *exp, __u64 flags,
137                               const struct lu_fid *fid, enum ldlm_type type,
138                               union ldlm_policy_data *policy,
139                               enum ldlm_mode mode, struct lustre_handle *lockh)
140 {
141         struct ldlm_res_id res_id;
142         enum ldlm_mode rc;
143
144         ENTRY;
145         fid_build_reg_res_name(fid, &res_id);
146         /* LU-4405: Clear bits not supported by server */
147         policy->l_inodebits.bits &= exp_connect_ibits(exp);
148         rc = ldlm_lock_match(class_exp2obd(exp)->obd_namespace, flags,
149                              &res_id, type, policy, mode, lockh);
150         RETURN(rc);
151 }
152
153 int mdc_cancel_unused(struct obd_export *exp, const struct lu_fid *fid,
154                       union ldlm_policy_data *policy, enum ldlm_mode mode,
155                       enum ldlm_cancel_flags flags, void *opaque)
156 {
157         struct obd_device *obd = class_exp2obd(exp);
158         struct ldlm_res_id res_id;
159         int rc;
160
161         ENTRY;
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
175         ENTRY;
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",
202                           rc);
203                 LBUG();
204         }
205 }
206
207 /**
208  * Save a large LOV EA into the request buffer so that it is available
209  * for replay.  We don't do this in the initial request because the
210  * original request doesn't need this buffer (at most it sends just the
211  * lov_mds_md) and it is a waste of RAM/bandwidth to send the empty
212  * buffer and may also be difficult to allocate and save a very large
213  * request buffer for each open. (b=5707)
214  *
215  * OOM here may cause recovery failure if lmm is needed (only for the
216  * original open if the MDS crashed just when this client also OOM'd)
217  * but this is incredibly unlikely, and questionable whether the client
218  * could do MDS recovery under OOM anyways...
219  */
220 static int mdc_save_lovea(struct ptlrpc_request *req, void *data, u32 size)
221 {
222         struct req_capsule *pill = &req->rq_pill;
223         void *lovea;
224         int rc = 0;
225
226         if (req_capsule_get_size(pill, &RMF_EADATA, RCL_CLIENT) < size) {
227                 rc = sptlrpc_cli_enlarge_reqbuf(req, &RMF_EADATA, size);
228                 if (rc) {
229                         CERROR("%s: Can't enlarge ea size to %d: rc = %d\n",
230                                req->rq_export->exp_obd->obd_name,
231                                size, rc);
232                         return rc;
233                 }
234         } else {
235                 req_capsule_shrink(pill, &RMF_EADATA, size, RCL_CLIENT);
236         }
237
238         req_capsule_set_size(pill, &RMF_EADATA, RCL_CLIENT, size);
239         lovea = req_capsule_client_get(pill, &RMF_EADATA);
240         if (lovea) {
241                 memcpy(lovea, data, size);
242                 lov_fix_ea_for_replay(lovea);
243         }
244
245         return rc;
246 }
247
248 static struct ptlrpc_request *
249 mdc_intent_open_pack(struct obd_export *exp, struct lookup_intent *it,
250                      struct md_op_data *op_data, __u32 acl_bufsize)
251 {
252         struct ptlrpc_request *req;
253         struct obd_device *obd = class_exp2obd(exp);
254         struct ldlm_intent *lit;
255         const void *lmm = op_data->op_data;
256         __u32 lmmsize = op_data->op_data_size;
257         __u32  mdt_md_capsule_size;
258         LIST_HEAD(cancels);
259         int count = 0;
260         enum ldlm_mode mode;
261         int repsize, repsize_estimate;
262         int rc;
263
264         ENTRY;
265
266         mdt_md_capsule_size = obd->u.cli.cl_default_mds_easize;
267
268         it->it_create_mode = (it->it_create_mode & ~S_IFMT) | S_IFREG;
269
270         /* XXX: openlock is not cancelled for cross-refs. */
271         /* If inode is known, cancel conflicting OPEN locks. */
272         if (fid_is_sane(&op_data->op_fid2)) {
273                 if (it->it_flags & MDS_OPEN_LEASE) { /* try to get lease */
274                         if (it->it_flags & MDS_FMODE_WRITE)
275                                 mode = LCK_EX;
276                         else
277                                 mode = LCK_PR;
278                 } else {
279                         if (it->it_flags & (MDS_FMODE_WRITE | MDS_OPEN_TRUNC))
280                                 mode = LCK_CW;
281 #ifdef FMODE_EXEC
282                         else if (it->it_flags & FMODE_EXEC)
283                                 mode = LCK_PR;
284 #endif
285                         else
286                                 mode = LCK_CR;
287                 }
288                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
289                                                 &cancels, mode,
290                                                 MDS_INODELOCK_OPEN);
291         }
292
293         /* If CREATE, cancel parent's UPDATE lock. */
294         if (it->it_op & IT_CREAT)
295                 mode = LCK_EX;
296         else
297                 mode = LCK_CR;
298         count += mdc_resource_get_unused(exp, &op_data->op_fid1,
299                                          &cancels, mode,
300                                          MDS_INODELOCK_UPDATE);
301
302         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
303                                    &RQF_LDLM_INTENT_OPEN);
304         if (req == NULL) {
305                 ldlm_lock_list_put(&cancels, l_bl_ast, count);
306                 RETURN(ERR_PTR(-ENOMEM));
307         }
308
309         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
310                              op_data->op_namelen + 1);
311         if (cl_is_lov_delay_create(it->it_flags)) {
312                 /* open(O_LOV_DELAY_CREATE) won't pack lmm */
313                 LASSERT(lmmsize == 0);
314                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, 0);
315         } else {
316                 req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT,
317                              max(lmmsize, obd->u.cli.cl_default_mds_easize));
318         }
319
320         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
321                              RCL_CLIENT, op_data->op_file_secctx_name != NULL ?
322                              op_data->op_file_secctx_name_size : 0);
323
324         req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX, RCL_CLIENT,
325                              op_data->op_file_secctx_size);
326
327         req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX, RCL_CLIENT,
328                              op_data->op_file_encctx_size);
329
330         /* get SELinux policy info if any */
331         rc = sptlrpc_get_sepol(req);
332         if (rc < 0) {
333                 ptlrpc_request_free(req);
334                 RETURN(ERR_PTR(rc));
335         }
336         req_capsule_set_size(&req->rq_pill, &RMF_SELINUX_POL, RCL_CLIENT,
337                              strlen(req->rq_sepol) ?
338                              strlen(req->rq_sepol) + 1 : 0);
339
340         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
341         if (rc < 0) {
342                 ptlrpc_request_free(req);
343                 RETURN(ERR_PTR(rc));
344         }
345
346         spin_lock(&req->rq_lock);
347         req->rq_replay = req->rq_import->imp_replayable;
348         spin_unlock(&req->rq_lock);
349
350         /* pack the intent */
351         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
352         lit->opc = (__u64)it->it_op;
353
354         /* pack the intended request */
355         mdc_open_pack(req, op_data, it->it_create_mode, 0, it->it_flags, lmm,
356                       lmmsize);
357
358         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER,
359                              mdt_md_capsule_size);
360         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, acl_bufsize);
361
362         if (!(it->it_op & IT_CREAT) && it->it_op & IT_OPEN &&
363             req_capsule_has_field(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
364                                   RCL_CLIENT) &&
365             op_data->op_file_secctx_name_size > 0 &&
366             op_data->op_file_secctx_name != NULL) {
367                 char *secctx_name;
368
369                 secctx_name = req_capsule_client_get(&req->rq_pill,
370                                                      &RMF_FILE_SECCTX_NAME);
371                 memcpy(secctx_name, op_data->op_file_secctx_name,
372                        op_data->op_file_secctx_name_size);
373                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
374                                      RCL_SERVER,
375                                      obd->u.cli.cl_max_mds_easize);
376
377                 CDEBUG(D_SEC, "packed '%.*s' as security xattr name\n",
378                        op_data->op_file_secctx_name_size,
379                        op_data->op_file_secctx_name);
380
381         } else {
382                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
383                                      RCL_SERVER, 0);
384         }
385
386         if (exp_connect_encrypt(exp) && !(it->it_op & IT_CREAT) &&
387             it->it_op & IT_OPEN)
388                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
389                                      RCL_SERVER,
390                                      obd->u.cli.cl_max_mds_easize);
391         else
392                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
393                                      RCL_SERVER, 0);
394
395         /**
396          * Inline buffer for possible data from Data-on-MDT files.
397          */
398         req_capsule_set_size(&req->rq_pill, &RMF_NIOBUF_INLINE, RCL_SERVER,
399                              sizeof(struct niobuf_remote));
400         ptlrpc_request_set_replen(req);
401
402         /* Get real repbuf allocated size as rounded up power of 2 */
403         repsize = size_roundup_power2(req->rq_replen +
404                                       lustre_msg_early_size());
405         /* Estimate free space for DoM files in repbuf */
406         repsize_estimate = repsize - (req->rq_replen -
407                            mdt_md_capsule_size +
408                            sizeof(struct lov_comp_md_v1) +
409                            sizeof(struct lov_comp_md_entry_v1) +
410                            lov_mds_md_size(0, LOV_MAGIC_V3));
411
412         if (repsize_estimate < obd->u.cli.cl_dom_min_inline_repsize) {
413                 repsize = obd->u.cli.cl_dom_min_inline_repsize -
414                           repsize_estimate + sizeof(struct niobuf_remote);
415                 req_capsule_set_size(&req->rq_pill, &RMF_NIOBUF_INLINE,
416                                      RCL_SERVER,
417                                      sizeof(struct niobuf_remote) + repsize);
418                 ptlrpc_request_set_replen(req);
419                 CDEBUG(D_INFO, "Increase repbuf by %d bytes, total: %d\n",
420                        repsize, req->rq_replen);
421                 repsize = size_roundup_power2(req->rq_replen +
422                                               lustre_msg_early_size());
423         }
424         /* The only way to report real allocated repbuf size to the server
425          * is the lm_repsize but it must be set prior buffer allocation itself
426          * due to security reasons - it is part of buffer used in signature
427          * calculation (see LU-11414). Therefore the saved size is predicted
428          * value as rq_replen rounded to the next higher power of 2.
429          * Such estimation is safe. Though the final allocated buffer might
430          * be even larger, it is not possible to know that at this point.
431          */
432         req->rq_reqmsg->lm_repsize = repsize;
433         RETURN(req);
434 }
435
436 #define GA_DEFAULT_EA_NAME_LEN   20
437 #define GA_DEFAULT_EA_VAL_LEN   250
438 #define GA_DEFAULT_EA_NUM        10
439
440 static struct ptlrpc_request *
441 mdc_intent_getxattr_pack(struct obd_export *exp, struct lookup_intent *it,
442                          struct md_op_data *op_data)
443 {
444         struct ptlrpc_request *req;
445         struct ldlm_intent *lit;
446         int rc, count = 0;
447         LIST_HEAD(cancels);
448         u32 ea_vals_buf_size = GA_DEFAULT_EA_VAL_LEN * GA_DEFAULT_EA_NUM;
449
450         ENTRY;
451         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
452                                         &RQF_LDLM_INTENT_GETXATTR);
453         if (req == NULL)
454                 RETURN(ERR_PTR(-ENOMEM));
455
456         /* get SELinux policy info if any */
457         rc = sptlrpc_get_sepol(req);
458         if (rc < 0) {
459                 ptlrpc_request_free(req);
460                 RETURN(ERR_PTR(rc));
461         }
462         req_capsule_set_size(&req->rq_pill, &RMF_SELINUX_POL, RCL_CLIENT,
463                              strlen(req->rq_sepol) ?
464                              strlen(req->rq_sepol) + 1 : 0);
465
466         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
467         if (rc) {
468                 ptlrpc_request_free(req);
469                 RETURN(ERR_PTR(rc));
470         }
471
472         /* pack the intent */
473         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
474         lit->opc = IT_GETXATTR;
475         /* Message below is checked in sanity-selinux test_20d
476          * and sanity-sec test_49
477          */
478         CDEBUG(D_INFO, "%s: get xattrs for "DFID"\n",
479                exp->exp_obd->obd_name, PFID(&op_data->op_fid1));
480
481 #if LUSTRE_VERSION_CODE < OBD_OCD_VERSION(3, 0, 53, 0)
482         /* If the supplied buffer is too small then the server will return
483          * -ERANGE and llite will fallback to using non cached xattr
484          * operations. On servers before 2.10.1 a (non-cached) listxattr RPC
485          * for an orphan or dead file causes an oops. So let's try to avoid
486          * sending too small a buffer to too old a server. This is effectively
487          * undoing the memory conservation of LU-9417 when it would be *more*
488          * likely to crash the server. See LU-9856.
489          */
490         if (exp->exp_connect_data.ocd_version < OBD_OCD_VERSION(2, 10, 1, 0))
491                 ea_vals_buf_size = max_t(u32, ea_vals_buf_size,
492                                          exp->exp_connect_data.ocd_max_easize);
493 #endif
494
495         /* pack the intended request */
496         mdc_pack_body(req, &op_data->op_fid1, op_data->op_valid,
497                       ea_vals_buf_size, -1, 0);
498
499         /* get SELinux policy info if any */
500         mdc_file_sepol_pack(req);
501
502         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_SERVER,
503                              GA_DEFAULT_EA_NAME_LEN * GA_DEFAULT_EA_NUM);
504
505         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS, RCL_SERVER,
506                              ea_vals_buf_size);
507
508         req_capsule_set_size(&req->rq_pill, &RMF_EAVALS_LENS, RCL_SERVER,
509                              sizeof(u32) * GA_DEFAULT_EA_NUM);
510
511         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, 0);
512
513         ptlrpc_request_set_replen(req);
514
515         RETURN(req);
516 }
517
518 static struct ptlrpc_request *
519 mdc_intent_getattr_pack(struct obd_export *exp, struct lookup_intent *it,
520                         struct md_op_data *op_data, __u32 acl_bufsize)
521 {
522         struct ptlrpc_request *req;
523         struct obd_device *obd = class_exp2obd(exp);
524         u64 valid = OBD_MD_FLGETATTR | OBD_MD_FLEASIZE | OBD_MD_FLMODEASIZE |
525                     OBD_MD_FLDIREA | OBD_MD_MEA | OBD_MD_FLACL |
526                     OBD_MD_DEFAULT_MEA;
527         struct ldlm_intent *lit;
528         __u32 easize;
529         bool have_secctx = false;
530         int rc;
531
532         ENTRY;
533         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
534                                    &RQF_LDLM_INTENT_GETATTR);
535         if (req == NULL)
536                 RETURN(ERR_PTR(-ENOMEM));
537
538         /* send name of security xattr to get upon intent */
539         if (it->it_op & (IT_LOOKUP | IT_GETATTR) &&
540             req_capsule_has_field(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
541                                   RCL_CLIENT) &&
542             op_data->op_file_secctx_name_size > 0 &&
543             op_data->op_file_secctx_name != NULL) {
544                 have_secctx = true;
545                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX_NAME,
546                                      RCL_CLIENT,
547                                      op_data->op_file_secctx_name_size);
548         }
549
550         req_capsule_set_size(&req->rq_pill, &RMF_NAME, RCL_CLIENT,
551                              op_data->op_namelen + 1);
552
553         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
554         if (rc) {
555                 ptlrpc_request_free(req);
556                 RETURN(ERR_PTR(rc));
557         }
558
559         /* pack the intent */
560         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
561         lit->opc = (__u64)it->it_op;
562
563         if (obd->u.cli.cl_default_mds_easize > 0)
564                 easize = obd->u.cli.cl_default_mds_easize;
565         else
566                 easize = obd->u.cli.cl_max_mds_easize;
567
568         /* pack the intended request */
569         mdc_getattr_pack(req, valid, it->it_flags, op_data, easize);
570
571         req_capsule_set_size(&req->rq_pill, &RMF_MDT_MD, RCL_SERVER, easize);
572         req_capsule_set_size(&req->rq_pill, &RMF_ACL, RCL_SERVER, acl_bufsize);
573         req_capsule_set_size(&req->rq_pill, &RMF_DEFAULT_MDT_MD, RCL_SERVER,
574                              sizeof(struct lmv_user_md));
575
576         if (have_secctx) {
577                 char *secctx_name;
578
579                 secctx_name = req_capsule_client_get(&req->rq_pill,
580                                                      &RMF_FILE_SECCTX_NAME);
581                 memcpy(secctx_name, op_data->op_file_secctx_name,
582                        op_data->op_file_secctx_name_size);
583
584                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
585                                      RCL_SERVER, easize);
586
587                 CDEBUG(D_SEC, "packed '%.*s' as security xattr name\n",
588                        op_data->op_file_secctx_name_size,
589                        op_data->op_file_secctx_name);
590         } else {
591                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_SECCTX,
592                                      RCL_SERVER, 0);
593         }
594
595         if (exp_connect_encrypt(exp) && it->it_op & (IT_LOOKUP | IT_GETATTR))
596                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
597                                      RCL_SERVER, easize);
598         else
599                 req_capsule_set_size(&req->rq_pill, &RMF_FILE_ENCCTX,
600                                      RCL_SERVER, 0);
601
602         ptlrpc_request_set_replen(req);
603         RETURN(req);
604 }
605
606 static struct ptlrpc_request *mdc_intent_layout_pack(struct obd_export *exp,
607                                                      struct lookup_intent *it,
608                                                      struct md_op_data *op_data)
609 {
610         struct obd_device *obd = class_exp2obd(exp);
611         struct ptlrpc_request *req;
612         struct ldlm_intent *lit;
613         struct layout_intent *layout;
614         LIST_HEAD(cancels);
615         int count = 0, rc;
616
617         ENTRY;
618         req = ptlrpc_request_alloc(class_exp2cliimp(exp),
619                                 &RQF_LDLM_INTENT_LAYOUT);
620         if (req == NULL)
621                 RETURN(ERR_PTR(-ENOMEM));
622
623         if (fid_is_sane(&op_data->op_fid2) && (it->it_op & IT_LAYOUT) &&
624             (it->it_flags & FMODE_WRITE)) {
625                 count = mdc_resource_get_unused(exp, &op_data->op_fid2,
626                                                 &cancels, LCK_EX,
627                                                 MDS_INODELOCK_LAYOUT);
628         }
629
630         req_capsule_set_size(&req->rq_pill, &RMF_EADATA, RCL_CLIENT, 0);
631         rc = ldlm_prep_enqueue_req(exp, req, &cancels, count);
632         if (rc) {
633                 ptlrpc_request_free(req);
634                 RETURN(ERR_PTR(rc));
635         }
636
637         /* pack the intent */
638         lit = req_capsule_client_get(&req->rq_pill, &RMF_LDLM_INTENT);
639         lit->opc = (__u64)it->it_op;
640
641         /* pack the layout intent request */
642         layout = req_capsule_client_get(&req->rq_pill, &RMF_LAYOUT_INTENT);
643         LASSERT(op_data->op_data != NULL);
644         LASSERT(op_data->op_data_size == sizeof(*layout));
645         memcpy(layout, op_data->op_data, sizeof(*layout));
646
647         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER,
648                              obd->u.cli.cl_default_mds_easize);
649         ptlrpc_request_set_replen(req);
650         RETURN(req);
651 }
652
653 static struct ptlrpc_request *mdc_enqueue_pack(struct obd_export *exp,
654                                                int lvb_len)
655 {
656         struct ptlrpc_request *req;
657         int rc;
658
659         ENTRY;
660         req = ptlrpc_request_alloc(class_exp2cliimp(exp), &RQF_LDLM_ENQUEUE);
661         if (req == NULL)
662                 RETURN(ERR_PTR(-ENOMEM));
663
664         rc = ldlm_prep_enqueue_req(exp, req, NULL, 0);
665         if (rc) {
666                 ptlrpc_request_free(req);
667                 RETURN(ERR_PTR(rc));
668         }
669
670         req_capsule_set_size(&req->rq_pill, &RMF_DLM_LVB, RCL_SERVER, lvb_len);
671         ptlrpc_request_set_replen(req);
672         RETURN(req);
673 }
674
675 static int mdc_finish_enqueue(struct obd_export *exp,
676                               struct ptlrpc_request *req,
677                               struct ldlm_enqueue_info *einfo,
678                               struct lookup_intent *it,
679                               struct lustre_handle *lockh, int rc)
680 {
681         struct req_capsule *pill = &req->rq_pill;
682         struct ldlm_request *lockreq;
683         struct ldlm_reply *lockrep;
684         struct ldlm_lock *lock;
685         struct mdt_body *body = NULL;
686         void *lvb_data = NULL;
687         __u32 lvb_len = 0;
688
689         ENTRY;
690         LASSERT(rc >= 0);
691         /* Similarly, if we're going to replay this request, we don't want to
692          * actually get a lock, just perform the intent.
693          */
694         if (req->rq_transno || req->rq_replay) {
695                 lockreq = req_capsule_client_get(pill, &RMF_DLM_REQ);
696                 lockreq->lock_flags |= ldlm_flags_to_wire(LDLM_FL_INTENT_ONLY);
697         }
698
699         if (rc == ELDLM_LOCK_ABORTED) {
700                 einfo->ei_mode = 0;
701                 memset(lockh, 0, sizeof(*lockh));
702                 rc = 0;
703         } else { /* rc = 0 */
704                 lock = ldlm_handle2lock(lockh);
705                 LASSERT(lock != NULL);
706
707                 /* If server returned a different lock mode, fix up variables */
708                 if (lock->l_req_mode != einfo->ei_mode) {
709                         ldlm_lock_addref(lockh, lock->l_req_mode);
710                         ldlm_lock_decref(lockh, einfo->ei_mode);
711                         einfo->ei_mode = lock->l_req_mode;
712                 }
713                 LDLM_LOCK_PUT(lock);
714         }
715
716         lockrep = req_capsule_server_get(pill, &RMF_DLM_REP);
717         LASSERT(lockrep != NULL); /* checked by ldlm_cli_enqueue() */
718
719         it->it_disposition = (int)lockrep->lock_policy_res1;
720         it->it_status = (int)lockrep->lock_policy_res2;
721         it->it_lock_mode = einfo->ei_mode;
722         it->it_lock_handle = lockh->cookie;
723         it->it_request = req;
724
725         /* Technically speaking rq_transno must already be zero if
726          * it_status is in error, so the check is a bit redundant.
727          */
728         if ((!req->rq_transno || it->it_status < 0) && req->rq_replay)
729                 mdc_clear_replay_flag(req, it->it_status);
730
731         /* If we're doing an IT_OPEN which did not result in an actual
732          * successful open, then we need to remove the bit which saves
733          * this request for unconditional replay.
734          *
735          * It's important that we do this first!  Otherwise we might exit the
736          * function without doing so, and try to replay a failed create.
737          * (b=3440)
738          */
739         if (it->it_op & IT_OPEN && req->rq_replay &&
740             (!it_disposition(it, DISP_OPEN_OPEN) || it->it_status != 0))
741                 mdc_clear_replay_flag(req, it->it_status);
742
743         DEBUG_REQ(D_RPCTRACE, req, "op=%x disposition=%x, status=%d",
744                   it->it_op, it->it_disposition, it->it_status);
745
746         /* We know what to expect, so we do any byte flipping required here */
747         if (it_has_reply_body(it)) {
748                 body = req_capsule_server_get(pill, &RMF_MDT_BODY);
749                 if (body == NULL) {
750                         rc = -EPROTO;
751                         CERROR("%s: cannot swab mdt_body: rc = %d\n",
752                                exp->exp_obd->obd_name, rc);
753                         RETURN(rc);
754                 }
755
756                 if (it_disposition(it, DISP_OPEN_OPEN) &&
757                     !it_open_error(DISP_OPEN_OPEN, it)) {
758                         /*
759                          * If this is a successful OPEN request, we need to set
760                          * replay handler and data early, so that if replay
761                          * happens immediately after swabbing below, new reply
762                          * is swabbed by that handler correctly.
763                          */
764                         mdc_set_open_replay_data(NULL, NULL, it);
765                 }
766
767                 if (it_disposition(it, DISP_OPEN_CREATE) &&
768                     !it_open_error(DISP_OPEN_CREATE, it)) {
769                         lprocfs_counter_incr(exp->exp_obd->obd_md_stats,
770                                              LPROC_MD_CREATE);
771                 }
772
773                 if (body->mbo_valid & (OBD_MD_FLDIREA | OBD_MD_FLEASIZE)) {
774                         void *eadata;
775
776                         mdc_update_max_ea_from_body(exp, body);
777
778                         /*
779                          * The eadata is opaque; just check that it is there.
780                          * Eventually, obd_unpackmd() will check the contents.
781                          */
782                         eadata = req_capsule_server_sized_get(pill, &RMF_MDT_MD,
783                                                         body->mbo_eadatasize);
784                         if (eadata == NULL)
785                                 RETURN(-EPROTO);
786
787                         /* save LVB data and length if for layout lock */
788                         lvb_data = eadata;
789                         lvb_len = body->mbo_eadatasize;
790
791                         /*
792                          * We save the reply LOV EA in case we have to replay a
793                          * create for recovery.  If we didn't allocate a large
794                          * enough request buffer above we need to reallocate it
795                          * here to hold the actual LOV EA.
796                          *
797                          * To not save LOV EA if request is not going to replay
798                          * (for example error one).
799                          */
800                         if ((it->it_op & IT_OPEN) && req->rq_replay) {
801                                 rc = mdc_save_lovea(req, eadata,
802                                                     body->mbo_eadatasize);
803                                 if (rc) {
804                                         body->mbo_valid &= ~OBD_MD_FLEASIZE;
805                                         body->mbo_eadatasize = 0;
806                                         rc = 0;
807                                 }
808                         }
809                 }
810         } else if (it->it_op & IT_LAYOUT) {
811                 /* maybe the lock was granted right away and layout
812                  * is packed into RMF_DLM_LVB of req
813                  */
814                 lvb_len = req_capsule_get_size(pill, &RMF_DLM_LVB, RCL_SERVER);
815                 CDEBUG(D_INFO, "%s: layout return lvb %d transno %lld\n",
816                        class_exp2obd(exp)->obd_name, lvb_len, req->rq_transno);
817                 if (lvb_len > 0) {
818                         lvb_data = req_capsule_server_sized_get(pill,
819                                                         &RMF_DLM_LVB, lvb_len);
820                         if (lvb_data == NULL)
821                                 RETURN(-EPROTO);
822
823                         /**
824                          * save replied layout data to the request buffer for
825                          * recovery consideration (lest MDS reinitialize
826                          * another set of OST objects).
827                          */
828                         if (req->rq_transno)
829                                 (void)mdc_save_lovea(req, lvb_data, lvb_len);
830                 }
831         }
832
833         /* fill in stripe data for layout lock.
834          * LU-6581: trust layout data only if layout lock is granted. The MDT
835          * has stopped sending layout unless the layout lock is granted. The
836          * client still does this checking in case it's talking with an old
837          * server. - Jinshan
838          */
839         lock = ldlm_handle2lock(lockh);
840         if (lock == NULL)
841                 RETURN(rc);
842
843         if (ldlm_has_layout(lock) && lvb_data != NULL &&
844             !(lockrep->lock_flags & LDLM_FL_BLOCKED_MASK)) {
845                 void *lmm;
846
847                 LDLM_DEBUG(lock, "layout lock returned by: %s, lvb_len: %d",
848                         ldlm_it2str(it->it_op), lvb_len);
849
850                 OBD_ALLOC_LARGE(lmm, lvb_len);
851                 if (lmm == NULL)
852                         GOTO(out_lock, rc = -ENOMEM);
853
854                 memcpy(lmm, lvb_data, lvb_len);
855
856                 /* install lvb_data */
857                 lock_res_and_lock(lock);
858                 if (lock->l_lvb_data == NULL) {
859                         lock->l_lvb_type = LVB_T_LAYOUT;
860                         lock->l_lvb_data = lmm;
861                         lock->l_lvb_len = lvb_len;
862                         lmm = NULL;
863                 }
864                 unlock_res_and_lock(lock);
865                 if (lmm != NULL)
866                         OBD_FREE_LARGE(lmm, lvb_len);
867         }
868
869         if (ldlm_has_dom(lock)) {
870                 LASSERT(lock->l_glimpse_ast == mdc_ldlm_glimpse_ast);
871
872                 body = req_capsule_server_get(pill, &RMF_MDT_BODY);
873                 if (!(body->mbo_valid & OBD_MD_DOM_SIZE)) {
874                         LDLM_ERROR(lock, "%s: DoM lock without size.",
875                                    exp->exp_obd->obd_name);
876                         GOTO(out_lock, rc = -EPROTO);
877                 }
878
879                 LDLM_DEBUG(lock, "DoM lock is returned by: %s, size: %llu",
880                            ldlm_it2str(it->it_op), body->mbo_dom_size);
881
882                 rc = mdc_fill_lvb(req, &lock->l_ost_lvb);
883         }
884 out_lock:
885         LDLM_LOCK_PUT(lock);
886
887         RETURN(rc);
888 }
889
890 static inline bool mdc_skip_mod_rpc_slot(const struct lookup_intent *it)
891 {
892         if (it != NULL &&
893             (it->it_op == IT_GETATTR || it->it_op == IT_LOOKUP ||
894              it->it_op == IT_READDIR ||
895              (it->it_op == IT_LAYOUT && !(it->it_flags & MDS_FMODE_WRITE))))
896                 return true;
897         return false;
898 }
899
900 /* We always reserve enough space in the reply packet for a stripe MD, because
901  * we don't know in advance the file type.
902  */
903 static int mdc_enqueue_base(struct obd_export *exp,
904                             struct ldlm_enqueue_info *einfo,
905                             const union ldlm_policy_data *policy,
906                             struct lookup_intent *it,
907                             struct md_op_data *op_data,
908                             struct lustre_handle *lockh,
909                             __u64 extra_lock_flags)
910 {
911         struct obd_device *obd = class_exp2obd(exp);
912         struct ptlrpc_request *req;
913         __u64 flags, saved_flags = extra_lock_flags;
914         struct ldlm_res_id res_id;
915         static const union ldlm_policy_data lookup_policy = {
916                                   .l_inodebits = { MDS_INODELOCK_LOOKUP } };
917         static const union ldlm_policy_data update_policy = {
918                                   .l_inodebits = { MDS_INODELOCK_UPDATE } };
919         static const union ldlm_policy_data layout_policy = {
920                                   .l_inodebits = { MDS_INODELOCK_LAYOUT } };
921         static const union ldlm_policy_data getxattr_policy = {
922                                   .l_inodebits = { MDS_INODELOCK_XATTR } };
923         int generation, resends = 0;
924         struct ldlm_reply *lockrep;
925         struct obd_import *imp = class_exp2cliimp(exp);
926         __u32 acl_bufsize;
927         enum lvb_type lvb_type = 0;
928         int rc;
929
930         ENTRY;
931         LASSERTF(!it || einfo->ei_type == LDLM_IBITS, "lock type %d\n",
932                  einfo->ei_type);
933         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
934
935         if (it != NULL) {
936                 LASSERT(policy == NULL);
937
938                 saved_flags |= LDLM_FL_HAS_INTENT;
939                 if (it->it_op & (IT_GETATTR | IT_READDIR))
940                         policy = &update_policy;
941                 else if (it->it_op & IT_LAYOUT)
942                         policy = &layout_policy;
943                 else if (it->it_op & IT_GETXATTR)
944                         policy = &getxattr_policy;
945                 else
946                         policy = &lookup_policy;
947         }
948
949         generation = obd->u.cli.cl_import->imp_generation;
950         if (!it || (it->it_op & (IT_OPEN | IT_CREAT)))
951                 acl_bufsize = min_t(__u32, imp->imp_connect_data.ocd_max_easize,
952                                     XATTR_SIZE_MAX);
953         else
954                 acl_bufsize = LUSTRE_POSIX_ACL_MAX_SIZE_OLD;
955
956 resend:
957         flags = saved_flags;
958         if (it == NULL) {
959                 /* The only way right now is FLOCK. */
960                 LASSERTF(einfo->ei_type == LDLM_FLOCK, "lock type %d\n",
961                          einfo->ei_type);
962                 res_id.name[3] = LDLM_FLOCK;
963                 req = ldlm_enqueue_pack(exp, 0);
964         } else if (it->it_op & IT_OPEN) {
965                 req = mdc_intent_open_pack(exp, it, op_data, acl_bufsize);
966         } else if (it->it_op & (IT_GETATTR | IT_LOOKUP)) {
967                 req = mdc_intent_getattr_pack(exp, it, op_data, acl_bufsize);
968         } else if (it->it_op & IT_READDIR) {
969                 req = mdc_enqueue_pack(exp, 0);
970         } else if (it->it_op & IT_LAYOUT) {
971                 if (!imp_connect_lvb_type(imp))
972                         RETURN(-EOPNOTSUPP);
973                 req = mdc_intent_layout_pack(exp, it, op_data);
974                 lvb_type = LVB_T_LAYOUT;
975         } else if (it->it_op & IT_GETXATTR) {
976                 req = mdc_intent_getxattr_pack(exp, it, op_data);
977         } else {
978                 LBUG();
979                 RETURN(-EINVAL);
980         }
981
982         if (IS_ERR(req))
983                 RETURN(PTR_ERR(req));
984
985         if (resends) {
986                 req->rq_generation_set = 1;
987                 req->rq_import_generation = generation;
988                 req->rq_sent = ktime_get_real_seconds() + resends;
989         }
990
991         einfo->ei_enq_slot = !mdc_skip_mod_rpc_slot(it);
992
993         /* With Data-on-MDT the glimpse callback is needed too.
994          * It is set here in advance but not in mdc_finish_enqueue()
995          * to avoid possible races. It is safe to have glimpse handler
996          * for non-DOM locks and costs nothing.
997          */
998         if (einfo->ei_cb_gl == NULL)
999                 einfo->ei_cb_gl = mdc_ldlm_glimpse_ast;
1000
1001         rc = ldlm_cli_enqueue(exp, &req, einfo, &res_id, policy, &flags, NULL,
1002                               0, lvb_type, lockh, 0);
1003
1004         if (!it) {
1005                 /* For flock requests we immediatelly return without further
1006                  * delay and let caller deal with the rest, since rest of
1007                  * this function metadata processing makes no sense for flock
1008                  * requests anyway. But in case of problem during comms with
1009                  * server (-ETIMEDOUT) or any signal/kill attempt (-EINTR),
1010                  * we cannot rely on caller and this mainly for F_UNLCKs
1011                  * (explicits or automatically generated by kernel to clean
1012                  * current flocks upon exit) that can't be trashed.
1013                  */
1014                 ptlrpc_req_finished(req);
1015                 if (((rc == -EINTR) || (rc == -ETIMEDOUT)) &&
1016                     (einfo->ei_type == LDLM_FLOCK) &&
1017                     (einfo->ei_mode == LCK_NL))
1018                         goto resend;
1019                 RETURN(rc);
1020         }
1021
1022         if (rc < 0) {
1023                 CDEBUG(D_INFO,
1024                       "%s: ldlm_cli_enqueue "DFID":"DFID"=%s failed: rc = %d\n",
1025                       obd->obd_name, PFID(&op_data->op_fid1),
1026                       PFID(&op_data->op_fid2), op_data->op_name ?: "", rc);
1027
1028                 mdc_clear_replay_flag(req, rc);
1029                 ptlrpc_req_finished(req);
1030                 RETURN(rc);
1031         }
1032
1033         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1034         LASSERT(lockrep != NULL);
1035
1036         lockrep->lock_policy_res2 =
1037                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
1038
1039         /* Retry infinitely when the server returns -EINPROGRESS for the
1040          * intent operation, when server returns -EINPROGRESS for acquiring
1041          * intent lock, we'll retry in after_reply().
1042          */
1043         if (it && (int)lockrep->lock_policy_res2 == -EINPROGRESS) {
1044                 mdc_clear_replay_flag(req, rc);
1045                 ptlrpc_req_finished(req);
1046                 if (generation == obd->u.cli.cl_import->imp_generation) {
1047                         if (signal_pending(current))
1048                                 RETURN(-EINTR);
1049
1050                         resends++;
1051                         CDEBUG(D_HA, "%s: resend:%d op:%d "DFID"/"DFID"\n",
1052                                obd->obd_name, resends, it->it_op,
1053                                PFID(&op_data->op_fid1),
1054                                PFID(&op_data->op_fid2));
1055                         goto resend;
1056                 } else {
1057                         CDEBUG(D_HA, "resend cross eviction\n");
1058                         RETURN(-EIO);
1059                 }
1060         }
1061
1062         if ((int)lockrep->lock_policy_res2 == -ERANGE &&
1063             it->it_op & (IT_OPEN | IT_GETATTR | IT_LOOKUP) &&
1064             acl_bufsize == LUSTRE_POSIX_ACL_MAX_SIZE_OLD) {
1065                 mdc_clear_replay_flag(req, -ERANGE);
1066                 ptlrpc_req_finished(req);
1067                 acl_bufsize = min_t(__u32, imp->imp_connect_data.ocd_max_easize,
1068                                     XATTR_SIZE_MAX);
1069                 goto resend;
1070         }
1071
1072         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
1073         if (rc < 0) {
1074                 if (lustre_handle_is_used(lockh)) {
1075                         ldlm_lock_decref(lockh, einfo->ei_mode);
1076                         memset(lockh, 0, sizeof(*lockh));
1077                 }
1078                 ptlrpc_req_finished(req);
1079
1080                 it->it_lock_handle = 0;
1081                 it->it_lock_mode = 0;
1082                 it->it_request = NULL;
1083         }
1084
1085         RETURN(rc);
1086 }
1087
1088 int mdc_enqueue(struct obd_export *exp, struct ldlm_enqueue_info *einfo,
1089                 const union ldlm_policy_data *policy,
1090                 struct md_op_data *op_data,
1091                 struct lustre_handle *lockh, __u64 extra_lock_flags)
1092 {
1093         return mdc_enqueue_base(exp, einfo, policy, NULL,
1094                                 op_data, lockh, extra_lock_flags);
1095 }
1096
1097 static int mdc_finish_intent_lock(struct obd_export *exp,
1098                                   struct ptlrpc_request *request,
1099                                   struct md_op_data *op_data,
1100                                   struct lookup_intent *it,
1101                                   struct lustre_handle *lockh)
1102 {
1103         struct lustre_handle old_lock;
1104         struct ldlm_lock *lock;
1105         int rc = 0;
1106
1107         ENTRY;
1108         LASSERT(request != NULL);
1109         LASSERT(request != LP_POISON);
1110         LASSERT(request->rq_repmsg != LP_POISON);
1111
1112         if (it->it_op & IT_READDIR)
1113                 RETURN(0);
1114
1115         if (it->it_op & (IT_GETXATTR | IT_LAYOUT)) {
1116                 if (it->it_status != 0)
1117                         GOTO(out, rc = it->it_status);
1118         } else {
1119                 if (!it_disposition(it, DISP_IT_EXECD)) {
1120                         /* The server failed before it even started executing
1121                          * the intent, i.e. because it couldn't unpack the
1122                          * request.
1123                          */
1124                         LASSERT(it->it_status != 0);
1125                         GOTO(out, rc = it->it_status);
1126                 }
1127                 rc = it_open_error(DISP_IT_EXECD, it);
1128                 if (rc)
1129                         GOTO(out, rc);
1130
1131                 rc = it_open_error(DISP_LOOKUP_EXECD, it);
1132                 if (rc)
1133                         GOTO(out, rc);
1134
1135                 /* keep requests around for the multiple phases of the call
1136                  * this shows the DISP_XX must guarantee we make it into the
1137                  * call
1138                  */
1139                 if (!it_disposition(it, DISP_ENQ_CREATE_REF) &&
1140                     it_disposition(it, DISP_OPEN_CREATE) &&
1141                     !it_open_error(DISP_OPEN_CREATE, it)) {
1142                         it_set_disposition(it, DISP_ENQ_CREATE_REF);
1143                         /* balanced in ll_create_node */
1144                         ptlrpc_request_addref(request);
1145                 }
1146                 if (!it_disposition(it, DISP_ENQ_OPEN_REF) &&
1147                     it_disposition(it, DISP_OPEN_OPEN) &&
1148                     !it_open_error(DISP_OPEN_OPEN, it)) {
1149                         it_set_disposition(it, DISP_ENQ_OPEN_REF);
1150                         /* balanced in ll_file_open */
1151                         ptlrpc_request_addref(request);
1152                         /* eviction in middle of open RPC processing b=11546 */
1153                         OBD_FAIL_TIMEOUT(OBD_FAIL_MDC_ENQUEUE_PAUSE,
1154                                          obd_timeout);
1155                 }
1156
1157                 if (it->it_op & IT_CREAT) {
1158                         /* XXX this belongs in ll_create_it */
1159                 } else if (it->it_op == IT_OPEN) {
1160                         LASSERT(!it_disposition(it, DISP_OPEN_CREATE));
1161                 } else {
1162                         LASSERT(it->it_op & (IT_GETATTR | IT_LOOKUP));
1163                 }
1164         }
1165
1166         /* If we already have a matching lock, then cancel the new
1167          * one.  We have to set the data here instead of in
1168          * mdc_enqueue, because we need to use the child's inode as
1169          * the l_ast_data to match, and that's not available until
1170          * intent_finish has performed the iget().
1171          */
1172         lock = ldlm_handle2lock(lockh);
1173         if (lock) {
1174                 union ldlm_policy_data policy = lock->l_policy_data;
1175
1176                 LDLM_DEBUG(lock, "matching against this");
1177
1178                 if (it_has_reply_body(it)) {
1179                         struct mdt_body *body;
1180
1181                         body = req_capsule_server_get(&request->rq_pill,
1182                                                       &RMF_MDT_BODY);
1183                         /* mdc_enqueue checked */
1184                         LASSERT(body != NULL);
1185                         LASSERTF(fid_res_name_eq(&body->mbo_fid1,
1186                                                  &lock->l_resource->lr_name),
1187                                  "Lock res_id: "DLDLMRES", fid: "DFID"\n",
1188                                  PLDLMRES(lock->l_resource),
1189                                  PFID(&body->mbo_fid1));
1190                 }
1191                 LDLM_LOCK_PUT(lock);
1192
1193                 memcpy(&old_lock, lockh, sizeof(*lockh));
1194                 if (ldlm_lock_match(NULL, LDLM_FL_BLOCK_GRANTED, NULL,
1195                                    LDLM_IBITS, &policy, LCK_NL, &old_lock)) {
1196                         ldlm_lock_decref_and_cancel(lockh, it->it_lock_mode);
1197                         memcpy(lockh, &old_lock, sizeof(old_lock));
1198                         it->it_lock_handle = lockh->cookie;
1199                 }
1200         }
1201
1202         EXIT;
1203 out:
1204         CDEBUG(D_DENTRY,
1205                "D_IT dentry=%.*s intent=%s status=%d disp=%x: rc = %d\n",
1206                 (int)op_data->op_namelen, op_data->op_name,
1207                 ldlm_it2str(it->it_op), it->it_status, it->it_disposition, rc);
1208
1209         return rc;
1210 }
1211
1212 int mdc_revalidate_lock(struct obd_export *exp, struct lookup_intent *it,
1213                         struct lu_fid *fid, __u64 *bits)
1214 {
1215         /* We could just return 1 immediately, but as we should only be called
1216          * in revalidate_it if we already have a lock, let's verify that.
1217          */
1218         struct ldlm_res_id res_id;
1219         struct lustre_handle lockh;
1220         union ldlm_policy_data policy;
1221         enum ldlm_mode mode;
1222
1223         ENTRY;
1224         if (it->it_lock_handle) {
1225                 lockh.cookie = it->it_lock_handle;
1226                 mode = ldlm_revalidate_lock_handle(&lockh, bits);
1227         } else {
1228                 fid_build_reg_res_name(fid, &res_id);
1229                 switch (it->it_op) {
1230                 case IT_GETATTR:
1231                         /* File attributes are held under multiple bits:
1232                          * nlink is under lookup lock, size and times are
1233                          * under UPDATE lock and recently we've also got
1234                          * a separate permissions lock for owner/group/acl that
1235                          * were protected by lookup lock before.
1236                          * Getattr must provide all of that information,
1237                          * so we need to ensure we have all of those locks.
1238                          * Unfortunately, if the bits are split across multiple
1239                          * locks, there's no easy way to match all of them here,
1240                          * so an extra RPC would be performed to fetch all
1241                          * of those bits at once for now.
1242                          */
1243                         /* For new MDTs(> 2.4), UPDATE|PERM should be enough,
1244                          * but for old MDTs (< 2.4), permission is covered
1245                          * by LOOKUP lock, so it needs to match all bits here.
1246                          */
1247                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE |
1248                                                   MDS_INODELOCK_PERM;
1249                         break;
1250                 case IT_READDIR:
1251                         policy.l_inodebits.bits = MDS_INODELOCK_UPDATE;
1252                         break;
1253                 case IT_LAYOUT:
1254                         policy.l_inodebits.bits = MDS_INODELOCK_LAYOUT;
1255                         break;
1256                 default:
1257                         policy.l_inodebits.bits = MDS_INODELOCK_LOOKUP;
1258                         break;
1259                 }
1260
1261                 mode = mdc_lock_match(exp, LDLM_FL_BLOCK_GRANTED, fid,
1262                                       LDLM_IBITS, &policy,
1263                                       LCK_CR | LCK_CW | LCK_PR | LCK_PW,
1264                                       &lockh);
1265         }
1266
1267         if (mode) {
1268                 it->it_lock_handle = lockh.cookie;
1269                 it->it_lock_mode = mode;
1270         } else {
1271                 it->it_lock_handle = 0;
1272                 it->it_lock_mode = 0;
1273         }
1274
1275         RETURN(!!mode);
1276 }
1277
1278 /*
1279  * This long block is all about fixing up the lock and request state
1280  * so that it is correct as of the moment _before_ the operation was
1281  * applied; that way, the VFS will think that everything is normal and
1282  * call Lustre's regular VFS methods.
1283  *
1284  * If we're performing a creation, that means that unless the creation
1285  * failed with EEXIST, we should fake up a negative dentry.
1286  *
1287  * For everything else, we want to lookup to succeed.
1288  *
1289  * One additional note: if CREATE or OPEN succeeded, we add an extra
1290  * reference to the request because we need to keep it around until
1291  * ll_create/ll_open gets called.
1292  *
1293  * The server will return to us, in it_disposition, an indication of
1294  * exactly what it_status refers to.
1295  *
1296  * If DISP_OPEN_OPEN is set, then it_status refers to the open() call,
1297  * otherwise if DISP_OPEN_CREATE is set, then it status is the
1298  * creation failure mode.  In either case, one of DISP_LOOKUP_NEG or
1299  * DISP_LOOKUP_POS will be set, indicating whether the child lookup
1300  * was successful.
1301  *
1302  * Else, if DISP_LOOKUP_EXECD then it_status is the rc of the
1303  * child lookup.
1304  */
1305 int mdc_intent_lock(struct obd_export *exp, struct md_op_data *op_data,
1306                     struct lookup_intent *it, struct ptlrpc_request **reqp,
1307                     ldlm_blocking_callback cb_blocking, __u64 extra_lock_flags)
1308 {
1309         struct ldlm_enqueue_info einfo = {
1310                 .ei_type        = LDLM_IBITS,
1311                 .ei_mode        = it_to_lock_mode(it),
1312                 .ei_cb_bl       = cb_blocking,
1313                 .ei_cb_cp       = ldlm_completion_ast,
1314                 .ei_cb_gl       = mdc_ldlm_glimpse_ast,
1315         };
1316         struct lustre_handle lockh;
1317         int rc = 0;
1318
1319         ENTRY;
1320         LASSERT(it);
1321         CDEBUG(D_DLMTRACE, "(name: %.*s,"DFID") in obj "DFID
1322                 ", intent: %s flags %#llo\n", (int)op_data->op_namelen,
1323                 op_data->op_name, PFID(&op_data->op_fid2),
1324                 PFID(&op_data->op_fid1), ldlm_it2str(it->it_op),
1325                 it->it_flags);
1326
1327         lockh.cookie = 0;
1328         if (fid_is_sane(&op_data->op_fid2) &&
1329             (it->it_op & (IT_LOOKUP | IT_GETATTR | IT_READDIR))) {
1330                 /* We could just return 1 immediately, but since we should only
1331                  * be called in revalidate_it if we already have a lock, let's
1332                  * verify that.
1333                  */
1334                 it->it_lock_handle = 0;
1335                 rc = mdc_revalidate_lock(exp, it, &op_data->op_fid2, NULL);
1336                 /* Only return failure if it was not GETATTR by cfid
1337                  * (from inode_revalidate()).
1338                  */
1339                 if (rc || op_data->op_namelen != 0)
1340                         RETURN(rc);
1341         }
1342
1343         /* For case if upper layer did not alloc fid, do it now. */
1344         if (!fid_is_sane(&op_data->op_fid2) && it->it_op & IT_CREAT) {
1345                 rc = mdc_fid_alloc(NULL, exp, &op_data->op_fid2, op_data);
1346                 if (rc < 0) {
1347                         CERROR("%s: cannot allocate new FID: rc=%d\n",
1348                                exp->exp_obd->obd_name, rc);
1349                         RETURN(rc);
1350                 }
1351         }
1352
1353         rc = mdc_enqueue_base(exp, &einfo, NULL, it, op_data, &lockh,
1354                               extra_lock_flags);
1355         if (rc < 0)
1356                 RETURN(rc);
1357
1358         *reqp = it->it_request;
1359         rc = mdc_finish_intent_lock(exp, *reqp, op_data, it, &lockh);
1360         RETURN(rc);
1361 }
1362
1363 static int mdc_intent_getattr_async_interpret(const struct lu_env *env,
1364                                               struct ptlrpc_request *req,
1365                                               void *args, int rc)
1366 {
1367         struct mdc_getattr_args *ga = args;
1368         struct obd_export *exp = ga->ga_exp;
1369         struct md_enqueue_info *minfo = ga->ga_minfo;
1370         struct ldlm_enqueue_info *einfo = &minfo->mi_einfo;
1371         struct lookup_intent *it = &minfo->mi_it;
1372         struct lustre_handle *lockh = &minfo->mi_lockh;
1373         struct ldlm_reply *lockrep;
1374         __u64 flags = LDLM_FL_HAS_INTENT;
1375
1376         ENTRY;
1377         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_GETATTR_ENQUEUE))
1378                 rc = -ETIMEDOUT;
1379
1380         rc = ldlm_cli_enqueue_fini(exp, req, einfo->ei_type, 1, einfo->ei_mode,
1381                                    &flags, NULL, 0, lockh, rc);
1382         if (rc < 0) {
1383                 CERROR("%s: ldlm_cli_enqueue_fini() failed: rc = %d\n",
1384                        exp->exp_obd->obd_name, rc);
1385                 mdc_clear_replay_flag(req, rc);
1386                 GOTO(out, rc);
1387         }
1388
1389         lockrep = req_capsule_server_get(&req->rq_pill, &RMF_DLM_REP);
1390         LASSERT(lockrep != NULL);
1391
1392         lockrep->lock_policy_res2 =
1393                 ptlrpc_status_ntoh(lockrep->lock_policy_res2);
1394
1395         rc = mdc_finish_enqueue(exp, req, einfo, it, lockh, rc);
1396         if (rc)
1397                 GOTO(out, rc);
1398
1399         rc = mdc_finish_intent_lock(exp, req, &minfo->mi_data, it, lockh);
1400         EXIT;
1401
1402 out:
1403         minfo->mi_cb(req, minfo, rc);
1404         return 0;
1405 }
1406
1407 int mdc_intent_getattr_async(struct obd_export *exp,
1408                              struct md_enqueue_info *minfo)
1409 {
1410         struct md_op_data *op_data = &minfo->mi_data;
1411         struct lookup_intent *it = &minfo->mi_it;
1412         struct ptlrpc_request *req;
1413         struct mdc_getattr_args *ga;
1414         struct ldlm_res_id res_id;
1415         union ldlm_policy_data policy = {
1416                 .l_inodebits = { MDS_INODELOCK_LOOKUP | MDS_INODELOCK_UPDATE }
1417         };
1418         __u64 flags = LDLM_FL_HAS_INTENT;
1419         int rc = 0;
1420
1421         ENTRY;
1422         CDEBUG(D_DLMTRACE,
1423                "name: %.*s in inode "DFID", intent: %s flags %#llo\n",
1424                (int)op_data->op_namelen, op_data->op_name,
1425                PFID(&op_data->op_fid1), ldlm_it2str(it->it_op), it->it_flags);
1426
1427         fid_build_reg_res_name(&op_data->op_fid1, &res_id);
1428         /* If the MDT return -ERANGE because of large ACL, then the sponsor
1429          * of the async getattr RPC will handle that by itself.
1430          */
1431         req = mdc_intent_getattr_pack(exp, it, op_data,
1432                                       LUSTRE_POSIX_ACL_MAX_SIZE_OLD);
1433         if (IS_ERR(req))
1434                 RETURN(PTR_ERR(req));
1435
1436         /* With Data-on-MDT the glimpse callback is needed too.
1437          * It is set here in advance but not in mdc_finish_enqueue()
1438          * to avoid possible races. It is safe to have glimpse handler
1439          * for non-DOM locks and costs nothing.
1440          */
1441         if (minfo->mi_einfo.ei_cb_gl == NULL)
1442                 minfo->mi_einfo.ei_cb_gl = mdc_ldlm_glimpse_ast;
1443
1444         rc = ldlm_cli_enqueue(exp, &req, &minfo->mi_einfo, &res_id, &policy,
1445                               &flags, NULL, 0, LVB_T_NONE, &minfo->mi_lockh, 1);
1446         if (rc < 0) {
1447                 ptlrpc_req_finished(req);
1448                 RETURN(rc);
1449         }
1450
1451         ga = ptlrpc_req_async_args(ga, req);
1452         ga->ga_exp = exp;
1453         ga->ga_minfo = minfo;
1454
1455         req->rq_interpret_reply = mdc_intent_getattr_async_interpret;
1456         ptlrpcd_add_req(req);
1457
1458         RETURN(0);
1459 }