Whamcloud - gitweb
LU-13335 ldiskfs: add projid to debug logs
[fs/lustre-release.git] / lustre / llite / file.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) 2002, 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  *
31  * lustre/llite/file.c
32  *
33  * Author: Peter Braam <braam@clusterfs.com>
34  * Author: Phil Schwan <phil@clusterfs.com>
35  * Author: Andreas Dilger <adilger@clusterfs.com>
36  */
37
38 #define DEBUG_SUBSYSTEM S_LLITE
39 #include <lustre_dlm.h>
40 #include <linux/pagemap.h>
41 #include <linux/file.h>
42 #include <linux/sched.h>
43 #include <linux/user_namespace.h>
44 #include <linux/uidgid.h>
45 #include <linux/falloc.h>
46 #include <linux/ktime.h>
47
48 #include <uapi/linux/lustre/lustre_ioctl.h>
49 #include <lustre_swab.h>
50
51 #include "cl_object.h"
52 #include "llite_internal.h"
53 #include "vvp_internal.h"
54
55 struct split_param {
56         struct inode    *sp_inode;
57         __u16           sp_mirror_id;
58 };
59
60 struct pcc_param {
61         __u64   pa_data_version;
62         __u32   pa_archive_id;
63         __u32   pa_layout_gen;
64 };
65
66 static int
67 ll_put_grouplock(struct inode *inode, struct file *file, unsigned long arg);
68
69 static int ll_lease_close(struct obd_client_handle *och, struct inode *inode,
70                           bool *lease_broken);
71
72 static struct ll_file_data *ll_file_data_get(void)
73 {
74         struct ll_file_data *fd;
75
76         OBD_SLAB_ALLOC_PTR_GFP(fd, ll_file_data_slab, GFP_NOFS);
77         if (fd == NULL)
78                 return NULL;
79
80         fd->fd_write_failed = false;
81         pcc_file_init(&fd->fd_pcc_file);
82
83         return fd;
84 }
85
86 static void ll_file_data_put(struct ll_file_data *fd)
87 {
88         if (fd != NULL)
89                 OBD_SLAB_FREE_PTR(fd, ll_file_data_slab);
90 }
91
92 /**
93  * Packs all the attributes into @op_data for the CLOSE rpc.
94  */
95 static void ll_prepare_close(struct inode *inode, struct md_op_data *op_data,
96                              struct obd_client_handle *och)
97 {
98         ENTRY;
99
100         ll_prep_md_op_data(op_data, inode, NULL, NULL,
101                            0, 0, LUSTRE_OPC_ANY, NULL);
102
103         op_data->op_attr.ia_mode = inode->i_mode;
104         op_data->op_attr.ia_atime = inode->i_atime;
105         op_data->op_attr.ia_mtime = inode->i_mtime;
106         op_data->op_attr.ia_ctime = inode->i_ctime;
107         /* In case of encrypted file without the key, visible size was rounded
108          * up to next LUSTRE_ENCRYPTION_UNIT_SIZE, and clear text size was
109          * stored into lli_lazysize in ll_merge_attr(), so set proper file size
110          * now that we are closing.
111          */
112         if (llcrypt_require_key(inode) == -ENOKEY &&
113             ll_i2info(inode)->lli_attr_valid & OBD_MD_FLLAZYSIZE)
114                 op_data->op_attr.ia_size = ll_i2info(inode)->lli_lazysize;
115         else
116                 op_data->op_attr.ia_size = i_size_read(inode);
117         op_data->op_attr.ia_valid |= (ATTR_MODE | ATTR_ATIME | ATTR_ATIME_SET |
118                                       ATTR_MTIME | ATTR_MTIME_SET |
119                                       ATTR_CTIME);
120         op_data->op_xvalid |= OP_XVALID_CTIME_SET;
121         op_data->op_attr_blocks = inode->i_blocks;
122         op_data->op_attr_flags = ll_inode_to_ext_flags(inode->i_flags);
123         if (test_bit(LLIF_PROJECT_INHERIT, &ll_i2info(inode)->lli_flags))
124                 op_data->op_attr_flags |= LUSTRE_PROJINHERIT_FL;
125         op_data->op_open_handle = och->och_open_handle;
126
127         if (och->och_flags & FMODE_WRITE &&
128             test_and_clear_bit(LLIF_DATA_MODIFIED, &ll_i2info(inode)->lli_flags))
129                 /* For HSM: if inode data has been modified, pack it so that
130                  * MDT can set data dirty flag in the archive. */
131                 op_data->op_bias |= MDS_DATA_MODIFIED;
132
133         EXIT;
134 }
135
136 /**
137  * Perform a close, possibly with a bias.
138  * The meaning of "data" depends on the value of "bias".
139  *
140  * If \a bias is MDS_HSM_RELEASE then \a data is a pointer to the data version.
141  * If \a bias is MDS_CLOSE_LAYOUT_SWAP then \a data is a pointer to the inode to
142  * swap layouts with.
143  */
144 static int ll_close_inode_openhandle(struct inode *inode,
145                                      struct obd_client_handle *och,
146                                      enum mds_op_bias bias, void *data)
147 {
148         struct obd_export *md_exp = ll_i2mdexp(inode);
149         const struct ll_inode_info *lli = ll_i2info(inode);
150         struct md_op_data *op_data;
151         struct ptlrpc_request *req = NULL;
152         int rc;
153         ENTRY;
154
155         if (class_exp2obd(md_exp) == NULL) {
156                 CERROR("%s: invalid MDC connection handle closing "DFID"\n",
157                        ll_i2sbi(inode)->ll_fsname, PFID(&lli->lli_fid));
158                 GOTO(out, rc = 0);
159         }
160
161         OBD_ALLOC_PTR(op_data);
162         /* We leak openhandle and request here on error, but not much to be
163          * done in OOM case since app won't retry close on error either. */
164         if (op_data == NULL)
165                 GOTO(out, rc = -ENOMEM);
166
167         ll_prepare_close(inode, op_data, och);
168         switch (bias) {
169         case MDS_CLOSE_LAYOUT_MERGE:
170                 /* merge blocks from the victim inode */
171                 op_data->op_attr_blocks += ((struct inode *)data)->i_blocks;
172                 op_data->op_attr.ia_valid |= ATTR_SIZE;
173                 op_data->op_xvalid |= OP_XVALID_BLOCKS;
174                 fallthrough;
175         case MDS_CLOSE_LAYOUT_SPLIT:
176         case MDS_CLOSE_LAYOUT_SWAP: {
177                 struct split_param *sp = data;
178
179                 LASSERT(data != NULL);
180                 op_data->op_bias |= bias;
181                 op_data->op_data_version = 0;
182                 op_data->op_lease_handle = och->och_lease_handle;
183                 if (bias == MDS_CLOSE_LAYOUT_SPLIT) {
184                         op_data->op_fid2 = *ll_inode2fid(sp->sp_inode);
185                         op_data->op_mirror_id = sp->sp_mirror_id;
186                 } else {
187                         op_data->op_fid2 = *ll_inode2fid(data);
188                 }
189                 break;
190         }
191
192         case MDS_CLOSE_RESYNC_DONE: {
193                 struct ll_ioc_lease *ioc = data;
194
195                 LASSERT(data != NULL);
196                 op_data->op_attr_blocks +=
197                         ioc->lil_count * op_data->op_attr_blocks;
198                 op_data->op_attr.ia_valid |= ATTR_SIZE;
199                 op_data->op_xvalid |= OP_XVALID_BLOCKS;
200                 op_data->op_bias |= MDS_CLOSE_RESYNC_DONE;
201
202                 op_data->op_lease_handle = och->och_lease_handle;
203                 op_data->op_data = &ioc->lil_ids[0];
204                 op_data->op_data_size =
205                         ioc->lil_count * sizeof(ioc->lil_ids[0]);
206                 break;
207         }
208
209         case MDS_PCC_ATTACH: {
210                 struct pcc_param *param = data;
211
212                 LASSERT(data != NULL);
213                 op_data->op_bias |= MDS_HSM_RELEASE | MDS_PCC_ATTACH;
214                 op_data->op_archive_id = param->pa_archive_id;
215                 op_data->op_data_version = param->pa_data_version;
216                 op_data->op_lease_handle = och->och_lease_handle;
217                 break;
218         }
219
220         case MDS_HSM_RELEASE:
221                 LASSERT(data != NULL);
222                 op_data->op_bias |= MDS_HSM_RELEASE;
223                 op_data->op_data_version = *(__u64 *)data;
224                 op_data->op_lease_handle = och->och_lease_handle;
225                 op_data->op_attr.ia_valid |= ATTR_SIZE;
226                 op_data->op_xvalid |= OP_XVALID_BLOCKS;
227                 break;
228
229         default:
230                 LASSERT(data == NULL);
231                 break;
232         }
233
234         if (!(op_data->op_attr.ia_valid & ATTR_SIZE))
235                 op_data->op_xvalid |= OP_XVALID_LAZYSIZE;
236         if (!(op_data->op_xvalid & OP_XVALID_BLOCKS))
237                 op_data->op_xvalid |= OP_XVALID_LAZYBLOCKS;
238
239         rc = md_close(md_exp, op_data, och->och_mod, &req);
240         if (rc != 0 && rc != -EINTR)
241                 CERROR("%s: inode "DFID" mdc close failed: rc = %d\n",
242                        md_exp->exp_obd->obd_name, PFID(&lli->lli_fid), rc);
243
244         if (rc == 0 && op_data->op_bias & bias) {
245                 struct mdt_body *body;
246
247                 body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
248                 if (!(body->mbo_valid & OBD_MD_CLOSE_INTENT_EXECED))
249                         rc = -EBUSY;
250
251                 if (bias & MDS_PCC_ATTACH) {
252                         struct pcc_param *param = data;
253
254                         param->pa_layout_gen = body->mbo_layout_gen;
255                 }
256         }
257
258         ll_finish_md_op_data(op_data);
259         EXIT;
260 out:
261
262         md_clear_open_replay_data(md_exp, och);
263         och->och_open_handle.cookie = DEAD_HANDLE_MAGIC;
264         OBD_FREE_PTR(och);
265
266         ptlrpc_req_finished(req);       /* This is close request */
267         return rc;
268 }
269
270 int ll_md_real_close(struct inode *inode, fmode_t fmode)
271 {
272         struct ll_inode_info *lli = ll_i2info(inode);
273         struct obd_client_handle **och_p;
274         struct obd_client_handle *och;
275         __u64 *och_usecount;
276         int rc = 0;
277         ENTRY;
278
279         if (fmode & FMODE_WRITE) {
280                 och_p = &lli->lli_mds_write_och;
281                 och_usecount = &lli->lli_open_fd_write_count;
282         } else if (fmode & FMODE_EXEC) {
283                 och_p = &lli->lli_mds_exec_och;
284                 och_usecount = &lli->lli_open_fd_exec_count;
285         } else {
286                 LASSERT(fmode & FMODE_READ);
287                 och_p = &lli->lli_mds_read_och;
288                 och_usecount = &lli->lli_open_fd_read_count;
289         }
290
291         mutex_lock(&lli->lli_och_mutex);
292         if (*och_usecount > 0) {
293                 /* There are still users of this handle, so skip
294                  * freeing it. */
295                 mutex_unlock(&lli->lli_och_mutex);
296                 RETURN(0);
297         }
298
299         och = *och_p;
300         *och_p = NULL;
301         mutex_unlock(&lli->lli_och_mutex);
302
303         if (och != NULL) {
304                 /* There might be a race and this handle may already
305                  * be closed. */
306                 rc = ll_close_inode_openhandle(inode, och, 0, NULL);
307         }
308
309         RETURN(rc);
310 }
311
312 static int ll_md_close(struct inode *inode, struct file *file)
313 {
314         union ldlm_policy_data policy = {
315                 .l_inodebits    = { MDS_INODELOCK_OPEN },
316         };
317         __u64 flags = LDLM_FL_BLOCK_GRANTED | LDLM_FL_TEST_LOCK;
318         struct ll_file_data *fd = file->private_data;
319         struct ll_inode_info *lli = ll_i2info(inode);
320         struct lustre_handle lockh;
321         enum ldlm_mode lockmode;
322         int rc = 0;
323         ENTRY;
324
325         /* clear group lock, if present */
326         if (unlikely(fd->fd_flags & LL_FILE_GROUP_LOCKED))
327                 ll_put_grouplock(inode, file, fd->fd_grouplock.lg_gid);
328
329         mutex_lock(&lli->lli_och_mutex);
330         if (fd->fd_lease_och != NULL) {
331                 bool lease_broken;
332                 struct obd_client_handle *lease_och;
333
334                 lease_och = fd->fd_lease_och;
335                 fd->fd_lease_och = NULL;
336                 mutex_unlock(&lli->lli_och_mutex);
337
338                 /* Usually the lease is not released when the
339                  * application crashed, we need to release here. */
340                 rc = ll_lease_close(lease_och, inode, &lease_broken);
341
342                 mutex_lock(&lli->lli_och_mutex);
343
344                 CDEBUG_LIMIT(rc ? D_ERROR : D_INODE,
345                              "Clean up lease "DFID" %d/%d\n",
346                              PFID(&lli->lli_fid), rc, lease_broken);
347         }
348
349         if (fd->fd_och != NULL) {
350                 struct obd_client_handle *och;
351
352                 och = fd->fd_och;
353                 fd->fd_och = NULL;
354                 mutex_unlock(&lli->lli_och_mutex);
355
356                 rc = ll_close_inode_openhandle(inode, och, 0, NULL);
357                 GOTO(out, rc);
358         }
359
360         /* Let's see if we have good enough OPEN lock on the file and if
361            we can skip talking to MDS */
362         if (fd->fd_omode & FMODE_WRITE) {
363                 lockmode = LCK_CW;
364                 LASSERT(lli->lli_open_fd_write_count);
365                 lli->lli_open_fd_write_count--;
366         } else if (fd->fd_omode & FMODE_EXEC) {
367                 lockmode = LCK_PR;
368                 LASSERT(lli->lli_open_fd_exec_count);
369                 lli->lli_open_fd_exec_count--;
370         } else {
371                 lockmode = LCK_CR;
372                 LASSERT(lli->lli_open_fd_read_count);
373                 lli->lli_open_fd_read_count--;
374         }
375         mutex_unlock(&lli->lli_och_mutex);
376
377         /* LU-4398: do not cache write open lock if the file has exec bit */
378         if ((lockmode == LCK_CW && inode->i_mode & S_IXUGO) ||
379             !md_lock_match(ll_i2mdexp(inode), flags, ll_inode2fid(inode),
380                            LDLM_IBITS, &policy, lockmode, &lockh))
381                 rc = ll_md_real_close(inode, fd->fd_omode);
382
383 out:
384         file->private_data = NULL;
385         ll_file_data_put(fd);
386
387         RETURN(rc);
388 }
389
390 /* While this returns an error code, fput() the caller does not, so we need
391  * to make every effort to clean up all of our state here.  Also, applications
392  * rarely check close errors and even if an error is returned they will not
393  * re-try the close call.
394  */
395 int ll_file_release(struct inode *inode, struct file *file)
396 {
397         struct ll_file_data *fd;
398         struct ll_sb_info *sbi = ll_i2sbi(inode);
399         struct ll_inode_info *lli = ll_i2info(inode);
400         ktime_t kstart = ktime_get();
401         int rc;
402
403         ENTRY;
404
405         CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p)\n",
406                PFID(ll_inode2fid(inode)), inode);
407
408         fd = file->private_data;
409         LASSERT(fd != NULL);
410
411         /* The last ref on @file, maybe not the the owner pid of statahead,
412          * because parent and child process can share the same file handle. */
413         if (S_ISDIR(inode->i_mode) && lli->lli_opendir_key == fd)
414                 ll_deauthorize_statahead(inode, fd);
415
416         if (is_root_inode(inode)) {
417                 file->private_data = NULL;
418                 ll_file_data_put(fd);
419                 GOTO(out, rc = 0);
420         }
421
422         pcc_file_release(inode, file);
423
424         if (!S_ISDIR(inode->i_mode)) {
425                 if (lli->lli_clob != NULL)
426                         lov_read_and_clear_async_rc(lli->lli_clob);
427                 lli->lli_async_rc = 0;
428         }
429
430         lli->lli_close_fd_time = ktime_get();
431
432         rc = ll_md_close(inode, file);
433
434         if (CFS_FAIL_TIMEOUT_MS(OBD_FAIL_PTLRPC_DUMP_LOG, cfs_fail_val))
435                 libcfs_debug_dumplog();
436
437 out:
438         if (!rc && !is_root_inode(inode))
439                 ll_stats_ops_tally(sbi, LPROC_LL_RELEASE,
440                                    ktime_us_delta(ktime_get(), kstart));
441         RETURN(rc);
442 }
443
444 static inline int ll_dom_readpage(void *data, struct page *page)
445 {
446         /* since ll_dom_readpage is a page cache helper, it is safe to assume
447          * mapping and host pointers are set here
448          */
449         struct inode *inode;
450         struct niobuf_local *lnb = data;
451         void *kaddr;
452         int rc = 0;
453
454         inode = page2inode(page);
455
456         kaddr = kmap_atomic(page);
457         memcpy(kaddr, lnb->lnb_data, lnb->lnb_len);
458         if (lnb->lnb_len < PAGE_SIZE)
459                 memset(kaddr + lnb->lnb_len, 0,
460                        PAGE_SIZE - lnb->lnb_len);
461         flush_dcache_page(page);
462         SetPageUptodate(page);
463         kunmap_atomic(kaddr);
464
465         if (inode && IS_ENCRYPTED(inode) && S_ISREG(inode->i_mode)) {
466                 if (!llcrypt_has_encryption_key(inode))
467                         CDEBUG(D_SEC, "no enc key for "DFID"\n",
468                                PFID(ll_inode2fid(inode)));
469                 else {
470                         unsigned int offs = 0;
471
472                         while (offs < PAGE_SIZE) {
473                                 /* decrypt only if page is not empty */
474                                 if (memcmp(page_address(page) + offs,
475                                            page_address(ZERO_PAGE(0)),
476                                            LUSTRE_ENCRYPTION_UNIT_SIZE) == 0)
477                                         break;
478
479                                 rc = llcrypt_decrypt_pagecache_blocks(page,
480                                                     LUSTRE_ENCRYPTION_UNIT_SIZE,
481                                                                       offs);
482                                 if (rc)
483                                         break;
484
485                                 offs += LUSTRE_ENCRYPTION_UNIT_SIZE;
486                         }
487                 }
488         }
489         unlock_page(page);
490
491         return rc;
492 }
493
494 void ll_dom_finish_open(struct inode *inode, struct ptlrpc_request *req)
495 {
496         struct lu_env *env;
497         struct cl_io *io;
498         struct ll_inode_info *lli = ll_i2info(inode);
499         struct cl_object *obj = lli->lli_clob;
500         struct address_space *mapping = inode->i_mapping;
501         struct page *vmpage;
502         struct niobuf_remote *rnb;
503         struct mdt_body *body;
504         char *data;
505         unsigned long index, start;
506         struct niobuf_local lnb;
507         __u16 refcheck;
508         int rc;
509
510         ENTRY;
511
512         if (obj == NULL)
513                 RETURN_EXIT;
514
515         if (!req_capsule_field_present(&req->rq_pill, &RMF_NIOBUF_INLINE,
516                                        RCL_SERVER))
517                 RETURN_EXIT;
518
519         rnb = req_capsule_server_get(&req->rq_pill, &RMF_NIOBUF_INLINE);
520         if (rnb == NULL || rnb->rnb_len == 0)
521                 RETURN_EXIT;
522
523         /* LU-11595: Server may return whole file and that is OK always or
524          * it may return just file tail and its offset must be aligned with
525          * client PAGE_SIZE to be used on that client, if server's PAGE_SIZE is
526          * smaller then offset may be not aligned and that data is just ignored.
527          */
528         if (rnb->rnb_offset & ~PAGE_MASK)
529                 RETURN_EXIT;
530
531         /* Server returns whole file or just file tail if it fills in reply
532          * buffer, in both cases total size should be equal to the file size.
533          */
534         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
535         if (rnb->rnb_offset + rnb->rnb_len != body->mbo_dom_size &&
536             !(inode && IS_ENCRYPTED(inode))) {
537                 CERROR("%s: server returns off/len %llu/%u but size %llu\n",
538                        ll_i2sbi(inode)->ll_fsname, rnb->rnb_offset,
539                        rnb->rnb_len, body->mbo_dom_size);
540                 RETURN_EXIT;
541         }
542
543         env = cl_env_get(&refcheck);
544         if (IS_ERR(env))
545                 RETURN_EXIT;
546         io = vvp_env_thread_io(env);
547         io->ci_obj = obj;
548         io->ci_ignore_layout = 1;
549         rc = cl_io_init(env, io, CIT_MISC, obj);
550         if (rc)
551                 GOTO(out_io, rc);
552
553         CDEBUG(D_INFO, "Get data along with open at %llu len %i, size %llu\n",
554                rnb->rnb_offset, rnb->rnb_len, body->mbo_dom_size);
555
556         data = (char *)rnb + sizeof(*rnb);
557
558         lnb.lnb_file_offset = rnb->rnb_offset;
559         start = lnb.lnb_file_offset >> PAGE_SHIFT;
560         index = 0;
561         LASSERT((lnb.lnb_file_offset & ~PAGE_MASK) == 0);
562         lnb.lnb_page_offset = 0;
563         do {
564                 struct cl_page *page;
565
566                 lnb.lnb_data = data + (index << PAGE_SHIFT);
567                 lnb.lnb_len = rnb->rnb_len - (index << PAGE_SHIFT);
568                 if (lnb.lnb_len > PAGE_SIZE)
569                         lnb.lnb_len = PAGE_SIZE;
570
571                 vmpage = read_cache_page(mapping, index + start,
572                                          ll_dom_readpage, &lnb);
573                 if (IS_ERR(vmpage)) {
574                         CWARN("%s: cannot fill page %lu for "DFID
575                               " with data: rc = %li\n",
576                               ll_i2sbi(inode)->ll_fsname, index + start,
577                               PFID(lu_object_fid(&obj->co_lu)),
578                               PTR_ERR(vmpage));
579                         break;
580                 }
581                 lock_page(vmpage);
582                 if (vmpage->mapping == NULL) {
583                         unlock_page(vmpage);
584                         put_page(vmpage);
585                         /* page was truncated */
586                         break;
587                 }
588                 /* attach VM page to CL page cache */
589                 page = cl_page_find(env, obj, vmpage->index, vmpage,
590                                     CPT_CACHEABLE);
591                 if (IS_ERR(page)) {
592                         ClearPageUptodate(vmpage);
593                         unlock_page(vmpage);
594                         put_page(vmpage);
595                         break;
596                 }
597                 SetPageUptodate(vmpage);
598                 cl_page_put(env, page);
599                 unlock_page(vmpage);
600                 put_page(vmpage);
601                 index++;
602         } while (rnb->rnb_len > (index << PAGE_SHIFT));
603
604 out_io:
605         cl_io_fini(env, io);
606         cl_env_put(env, &refcheck);
607
608         EXIT;
609 }
610
611 static int ll_intent_file_open(struct dentry *de, void *lmm, int lmmsize,
612                                 struct lookup_intent *itp)
613 {
614         struct ll_sb_info *sbi = ll_i2sbi(de->d_inode);
615         struct dentry *parent = de->d_parent;
616         char *name = NULL;
617         int len = 0;
618         struct md_op_data *op_data;
619         struct ptlrpc_request *req = NULL;
620         int rc;
621         ENTRY;
622
623         LASSERT(parent != NULL);
624         LASSERT(itp->it_flags & MDS_OPEN_BY_FID);
625
626         /* if server supports open-by-fid, or file name is invalid, don't pack
627          * name in open request */
628         if (OBD_FAIL_CHECK(OBD_FAIL_LLITE_OPEN_BY_NAME) ||
629             !(exp_connect_flags(sbi->ll_md_exp) & OBD_CONNECT_OPEN_BY_FID)) {
630 retry:
631                 len = de->d_name.len;
632                 name = kmalloc(len + 1, GFP_NOFS);
633                 if (!name)
634                         RETURN(-ENOMEM);
635
636                 /* race here */
637                 spin_lock(&de->d_lock);
638                 if (len != de->d_name.len) {
639                         spin_unlock(&de->d_lock);
640                         kfree(name);
641                         goto retry;
642                 }
643                 memcpy(name, de->d_name.name, len);
644                 name[len] = '\0';
645                 spin_unlock(&de->d_lock);
646
647                 if (!lu_name_is_valid_2(name, len)) {
648                         kfree(name);
649                         RETURN(-ESTALE);
650                 }
651         }
652
653         op_data = ll_prep_md_op_data(NULL, parent->d_inode, de->d_inode,
654                                      name, len, 0, LUSTRE_OPC_OPEN, NULL);
655         if (IS_ERR(op_data)) {
656                 kfree(name);
657                 RETURN(PTR_ERR(op_data));
658         }
659         op_data->op_data = lmm;
660         op_data->op_data_size = lmmsize;
661
662         OBD_FAIL_TIMEOUT(OBD_FAIL_LLITE_OPEN_DELAY, cfs_fail_val);
663
664         rc = md_intent_lock(sbi->ll_md_exp, op_data, itp, &req,
665                             &ll_md_blocking_ast, 0);
666         kfree(name);
667         ll_finish_md_op_data(op_data);
668         if (rc == -ESTALE) {
669                 /* reason for keep own exit path - don`t flood log
670                  * with messages with -ESTALE errors.
671                  */
672                 if (!it_disposition(itp, DISP_OPEN_OPEN) ||
673                      it_open_error(DISP_OPEN_OPEN, itp))
674                         GOTO(out, rc);
675                 ll_release_openhandle(de, itp);
676                 GOTO(out, rc);
677         }
678
679         if (it_disposition(itp, DISP_LOOKUP_NEG))
680                 GOTO(out, rc = -ENOENT);
681
682         if (rc != 0 || it_open_error(DISP_OPEN_OPEN, itp)) {
683                 rc = rc ? rc : it_open_error(DISP_OPEN_OPEN, itp);
684                 CDEBUG(D_VFSTRACE, "lock enqueue: err: %d\n", rc);
685                 GOTO(out, rc);
686         }
687
688         rc = ll_prep_inode(&de->d_inode, &req->rq_pill, NULL, itp);
689
690         if (!rc && itp->it_lock_mode) {
691                 __u64 bits = 0;
692
693                 /* If we got a lock back and it has a LOOKUP bit set,
694                  * make sure the dentry is marked as valid so we can find it.
695                  * We don't need to care about actual hashing since other bits
696                  * of kernel will deal with that later.
697                  */
698                 ll_set_lock_data(sbi->ll_md_exp, de->d_inode, itp, &bits);
699                 if (bits & MDS_INODELOCK_LOOKUP) {
700                         d_lustre_revalidate(de);
701                         ll_update_dir_depth(parent->d_inode, de->d_inode);
702                 }
703
704                 /* if DoM bit returned along with LAYOUT bit then there
705                  * can be read-on-open data returned.
706                  */
707                 if (bits & MDS_INODELOCK_DOM && bits & MDS_INODELOCK_LAYOUT)
708                         ll_dom_finish_open(de->d_inode, req);
709         }
710
711 out:
712         ptlrpc_req_finished(req);
713         ll_intent_drop_lock(itp);
714
715         /* We did open by fid, but by the time we got to the server, the object
716          * disappeared.  This is possible if the object was unlinked, but it's
717          * also possible if the object was unlinked by a rename.  In the case
718          * of an object renamed over our existing one, we can't fail this open.
719          * O_CREAT also goes through this path if we had an existing dentry,
720          * and it's obviously wrong to return ENOENT for O_CREAT.
721          *
722          * Instead let's return -ESTALE, and the VFS will retry the open with
723          * LOOKUP_REVAL, which we catch in ll_revalidate_dentry and fail to
724          * revalidate, causing a lookup.  This causes extra lookups in the case
725          * where we had a dentry in cache but the file is being unlinked and we
726          * lose the race with unlink, but this should be very rare.
727          */
728         if (rc == -ENOENT)
729                 rc = -ESTALE;
730
731         RETURN(rc);
732 }
733
734 static int ll_och_fill(struct obd_export *md_exp, struct lookup_intent *it,
735                        struct obd_client_handle *och)
736 {
737         struct mdt_body *body;
738
739         body = req_capsule_server_get(&it->it_request->rq_pill, &RMF_MDT_BODY);
740         och->och_open_handle = body->mbo_open_handle;
741         och->och_fid = body->mbo_fid1;
742         och->och_lease_handle.cookie = it->it_lock_handle;
743         och->och_magic = OBD_CLIENT_HANDLE_MAGIC;
744         och->och_flags = it->it_flags;
745
746         return md_set_open_replay_data(md_exp, och, it);
747 }
748
749 static int ll_local_open(struct file *file, struct lookup_intent *it,
750                          struct ll_file_data *fd, struct obd_client_handle *och)
751 {
752         struct inode *inode = file_inode(file);
753         ENTRY;
754
755         LASSERT(!file->private_data);
756
757         LASSERT(fd != NULL);
758
759         if (och) {
760                 int rc;
761
762                 rc = ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);
763                 if (rc != 0)
764                         RETURN(rc);
765         }
766
767         file->private_data = fd;
768         ll_readahead_init(inode, &fd->fd_ras);
769         fd->fd_omode = it->it_flags & (FMODE_READ | FMODE_WRITE | FMODE_EXEC);
770
771         RETURN(0);
772 }
773
774 void ll_track_file_opens(struct inode *inode)
775 {
776         struct ll_inode_info *lli = ll_i2info(inode);
777         struct ll_sb_info *sbi = ll_i2sbi(inode);
778
779         /* do not skew results with delays from never-opened inodes */
780         if (ktime_to_ns(lli->lli_close_fd_time))
781                 ll_stats_ops_tally(sbi, LPROC_LL_INODE_OPCLTM,
782                            ktime_us_delta(ktime_get(), lli->lli_close_fd_time));
783
784         if (ktime_after(ktime_get(),
785                         ktime_add_ms(lli->lli_close_fd_time,
786                                      sbi->ll_oc_max_ms))) {
787                 lli->lli_open_fd_count = 1;
788                 lli->lli_close_fd_time = ns_to_ktime(0);
789         } else {
790                 lli->lli_open_fd_count++;
791         }
792
793         ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_INODE_OCOUNT,
794                            lli->lli_open_fd_count);
795 }
796
797 /* Open a file, and (for the very first open) create objects on the OSTs at
798  * this time.  If opened with O_LOV_DELAY_CREATE, then we don't do the object
799  * creation or open until ll_lov_setstripe() ioctl is called.
800  *
801  * If we already have the stripe MD locally then we don't request it in
802  * md_open(), by passing a lmm_size = 0.
803  *
804  * It is up to the application to ensure no other processes open this file
805  * in the O_LOV_DELAY_CREATE case, or the default striping pattern will be
806  * used.  We might be able to avoid races of that sort by getting lli_open_sem
807  * before returning in the O_LOV_DELAY_CREATE case and dropping it here
808  * or in ll_file_release(), but I'm not sure that is desirable/necessary.
809  */
810 int ll_file_open(struct inode *inode, struct file *file)
811 {
812         struct ll_inode_info *lli = ll_i2info(inode);
813         struct lookup_intent *it, oit = { .it_op = IT_OPEN,
814                                           .it_flags = file->f_flags };
815         struct obd_client_handle **och_p = NULL;
816         __u64 *och_usecount = NULL;
817         struct ll_file_data *fd;
818         ktime_t kstart = ktime_get();
819         int rc = 0;
820         ENTRY;
821
822         CDEBUG(D_VFSTRACE, "VFS Op:inode="DFID"(%p), flags %o\n",
823                PFID(ll_inode2fid(inode)), inode, file->f_flags);
824
825         it = file->private_data; /* XXX: compat macro */
826         file->private_data = NULL; /* prevent ll_local_open assertion */
827
828         if (S_ISREG(inode->i_mode)) {
829                 rc = ll_file_open_encrypt(inode, file);
830                 if (rc) {
831                         if (it && it->it_disposition)
832                                 ll_release_openhandle(file_dentry(file), it);
833                         GOTO(out_nofiledata, rc);
834                 }
835         }
836
837         fd = ll_file_data_get();
838         if (fd == NULL)
839                 GOTO(out_nofiledata, rc = -ENOMEM);
840
841         fd->fd_file = file;
842         if (S_ISDIR(inode->i_mode))
843                 ll_authorize_statahead(inode, fd);
844
845         ll_track_file_opens(inode);
846         if (is_root_inode(inode)) {
847                 file->private_data = fd;
848                 RETURN(0);
849         }
850
851         if (!it || !it->it_disposition) {
852                 /* Convert f_flags into access mode. We cannot use file->f_mode,
853                  * because everything but O_ACCMODE mask was stripped from
854                  * there */
855                 if ((oit.it_flags + 1) & O_ACCMODE)
856                         oit.it_flags++;
857                 if (file->f_flags & O_TRUNC)
858                         oit.it_flags |= FMODE_WRITE;
859
860                 /* kernel only call f_op->open in dentry_open.  filp_open calls
861                  * dentry_open after call to open_namei that checks permissions.
862                  * Only nfsd_open call dentry_open directly without checking
863                  * permissions and because of that this code below is safe.
864                  */
865                 if (oit.it_flags & (FMODE_WRITE | FMODE_READ))
866                         oit.it_flags |= MDS_OPEN_OWNEROVERRIDE;
867
868                 /* We do not want O_EXCL here, presumably we opened the file
869                  * already? XXX - NFS implications? */
870                 oit.it_flags &= ~O_EXCL;
871
872                 /* bug20584, if "it_flags" contains O_CREAT, the file will be
873                  * created if necessary, then "IT_CREAT" should be set to keep
874                  * consistent with it */
875                 if (oit.it_flags & O_CREAT)
876                         oit.it_op |= IT_CREAT;
877
878                 it = &oit;
879         }
880
881 restart:
882         /* Let's see if we have file open on MDS already. */
883         if (it->it_flags & FMODE_WRITE) {
884                 och_p = &lli->lli_mds_write_och;
885                 och_usecount = &lli->lli_open_fd_write_count;
886         } else if (it->it_flags & FMODE_EXEC) {
887                 och_p = &lli->lli_mds_exec_och;
888                 och_usecount = &lli->lli_open_fd_exec_count;
889         } else {
890                 och_p = &lli->lli_mds_read_och;
891                 och_usecount = &lli->lli_open_fd_read_count;
892         }
893
894         mutex_lock(&lli->lli_och_mutex);
895         if (*och_p) { /* Open handle is present */
896                 if (it_disposition(it, DISP_OPEN_OPEN)) {
897                         /* Well, there's extra open request that we do not need,
898                          * let's close it somehow. This will decref request. */
899                         rc = it_open_error(DISP_OPEN_OPEN, it);
900                         if (rc) {
901                                 mutex_unlock(&lli->lli_och_mutex);
902                                 GOTO(out_openerr, rc);
903                         }
904
905                         ll_release_openhandle(file_dentry(file), it);
906                 }
907                 (*och_usecount)++;
908
909                 rc = ll_local_open(file, it, fd, NULL);
910                 if (rc) {
911                         (*och_usecount)--;
912                         mutex_unlock(&lli->lli_och_mutex);
913                         GOTO(out_openerr, rc);
914                 }
915         } else {
916                 LASSERT(*och_usecount == 0);
917                 if (!it->it_disposition) {
918                         struct dentry *dentry = file_dentry(file);
919                         struct ll_sb_info *sbi = ll_i2sbi(inode);
920                         struct ll_dentry_data *ldd;
921
922                         /* We cannot just request lock handle now, new ELC code
923                          * means that one of other OPEN locks for this file
924                          * could be cancelled, and since blocking ast handler
925                          * would attempt to grab och_mutex as well, that would
926                          * result in a deadlock
927                          */
928                         mutex_unlock(&lli->lli_och_mutex);
929                         /*
930                          * Normally called under two situations:
931                          * 1. NFS export.
932                          * 2. A race/condition on MDS resulting in no open
933                          *    handle to be returned from LOOKUP|OPEN request,
934                          *    for example if the target entry was a symlink.
935                          *
936                          * In NFS path we know there's pathologic behavior
937                          * so we always enable open lock caching when coming
938                          * from there. It's detected by setting a flag in
939                          * ll_iget_for_nfs.
940                          *
941                          * After reaching number of opens of this inode
942                          * we always ask for an open lock on it to handle
943                          * bad userspace actors that open and close files
944                          * in a loop for absolutely no good reason
945                          */
946
947                         ldd = ll_d2d(dentry);
948                         if (filename_is_volatile(dentry->d_name.name,
949                                                  dentry->d_name.len,
950                                                  NULL)) {
951                                 /* There really is nothing here, but this
952                                  * make this more readable I think.
953                                  * We do not want openlock for volatile
954                                  * files under any circumstances
955                                  */
956                         } else if (ldd && ldd->lld_nfs_dentry) {
957                                 /* NFS path. This also happens to catch
958                                  * open by fh files I guess
959                                  */
960                                 it->it_flags |= MDS_OPEN_LOCK;
961                                 /* clear the flag for future lookups */
962                                 ldd->lld_nfs_dentry = 0;
963                         } else if (sbi->ll_oc_thrsh_count > 0) {
964                                 /* Take MDS_OPEN_LOCK with many opens */
965                                 if (lli->lli_open_fd_count >=
966                                     sbi->ll_oc_thrsh_count)
967                                         it->it_flags |= MDS_OPEN_LOCK;
968
969                                 /* If this is open after we just closed */
970                                 else if (ktime_before(ktime_get(),
971                                             ktime_add_ms(lli->lli_close_fd_time,
972                                                          sbi->ll_oc_thrsh_ms)))
973                                         it->it_flags |= MDS_OPEN_LOCK;
974                         }
975
976                         /*
977                          * Always specify MDS_OPEN_BY_FID because we don't want
978                          * to get file with different fid.
979                          */
980                         it->it_flags |= MDS_OPEN_BY_FID;
981                         rc = ll_intent_file_open(dentry, NULL, 0, it);
982                         if (rc)
983                                 GOTO(out_openerr, rc);
984
985                         goto restart;
986                 }
987                 OBD_ALLOC(*och_p, sizeof(struct obd_client_handle));
988                 if (!*och_p)
989                         GOTO(out_och_free, rc = -ENOMEM);
990
991                 (*och_usecount)++;
992
993                 /* md_intent_lock() didn't get a request ref if there was an
994                  * open error, so don't do cleanup on the request here
995                  * (bug 3430) */
996                 /* XXX (green): Should not we bail out on any error here, not
997                  * just open error? */
998                 rc = it_open_error(DISP_OPEN_OPEN, it);
999                 if (rc != 0)
1000                         GOTO(out_och_free, rc);
1001
1002                 LASSERTF(it_disposition(it, DISP_ENQ_OPEN_REF),
1003                          "inode %p: disposition %x, status %d\n", inode,
1004                          it_disposition(it, ~0), it->it_status);
1005
1006                 rc = ll_local_open(file, it, fd, *och_p);
1007                 if (rc)
1008                         GOTO(out_och_free, rc);
1009         }
1010
1011         rc = pcc_file_open(inode, file);
1012         if (rc)
1013                 GOTO(out_och_free, rc);
1014
1015         mutex_unlock(&lli->lli_och_mutex);
1016
1017         fd = NULL;
1018
1019         /* Must do this outside lli_och_mutex lock to prevent deadlock where
1020            different kind of OPEN lock for this same inode gets cancelled
1021            by ldlm_cancel_lru */
1022         if (!S_ISREG(inode->i_mode))
1023                 GOTO(out_och_free, rc);
1024         cl_lov_delay_create_clear(&file->f_flags);
1025         GOTO(out_och_free, rc);
1026
1027 out_och_free:
1028         if (rc) {
1029                 if (och_p && *och_p) {
1030                         OBD_FREE(*och_p, sizeof(struct obd_client_handle));
1031                         *och_p = NULL; /* OBD_FREE writes some magic there */
1032                         (*och_usecount)--;
1033                 }
1034                 mutex_unlock(&lli->lli_och_mutex);
1035
1036 out_openerr:
1037                 if (lli->lli_opendir_key == fd)
1038                         ll_deauthorize_statahead(inode, fd);
1039
1040                 if (fd != NULL)
1041                         ll_file_data_put(fd);
1042         } else {
1043                 ll_stats_ops_tally(ll_i2sbi(inode), LPROC_LL_OPEN,
1044                                    ktime_us_delta(ktime_get(), kstart));
1045         }
1046
1047 out_nofiledata:
1048         if (it && it_disposition(it, DISP_ENQ_OPEN_REF)) {
1049                 ptlrpc_req_finished(it->it_request);
1050                 it_clear_disposition(it, DISP_ENQ_OPEN_REF);
1051         }
1052
1053         return rc;
1054 }
1055
1056 static int ll_md_blocking_lease_ast(struct ldlm_lock *lock,
1057                         struct ldlm_lock_desc *desc, void *data, int flag)
1058 {
1059         int rc;
1060         struct lustre_handle lockh;
1061         ENTRY;
1062
1063         switch (flag) {
1064         case LDLM_CB_BLOCKING:
1065                 ldlm_lock2handle(lock, &lockh);
1066                 rc = ldlm_cli_cancel(&lockh, LCF_ASYNC);
1067                 if (rc < 0) {
1068                         CDEBUG(D_INODE, "ldlm_cli_cancel: %d\n", rc);
1069                         RETURN(rc);
1070                 }
1071                 break;
1072         case LDLM_CB_CANCELING:
1073                 /* do nothing */
1074                 break;
1075         }
1076         RETURN(0);
1077 }
1078
1079 /**
1080  * When setting a lease on a file, we take ownership of the lli_mds_*_och
1081  * and save it as fd->fd_och so as to force client to reopen the file even
1082  * if it has an open lock in cache already.
1083  */
1084 static int ll_lease_och_acquire(struct inode *inode, struct file *file,
1085                                 struct lustre_handle *old_open_handle)
1086 {
1087         struct ll_inode_info *lli = ll_i2info(inode);
1088         struct ll_file_data *fd = file->private_data;
1089         struct obd_client_handle **och_p;
1090         __u64 *och_usecount;
1091         int rc = 0;
1092         ENTRY;
1093
1094         /* Get the openhandle of the file */
1095         mutex_lock(&lli->lli_och_mutex);
1096         if (fd->fd_lease_och != NULL)
1097                 GOTO(out_unlock, rc = -EBUSY);
1098
1099         if (fd->fd_och == NULL) {
1100                 if (file->f_mode & FMODE_WRITE) {
1101                         LASSERT(lli->lli_mds_write_och != NULL);
1102                         och_p = &lli->lli_mds_write_och;
1103                         och_usecount = &lli->lli_open_fd_write_count;
1104                 } else {
1105                         LASSERT(lli->lli_mds_read_och != NULL);
1106                         och_p = &lli->lli_mds_read_och;
1107                         och_usecount = &lli->lli_open_fd_read_count;
1108                 }
1109
1110                 if (*och_usecount > 1)
1111                         GOTO(out_unlock, rc = -EBUSY);
1112
1113                 fd->fd_och = *och_p;
1114                 *och_usecount = 0;
1115                 *och_p = NULL;
1116         }
1117
1118         *old_open_handle = fd->fd_och->och_open_handle;
1119
1120         EXIT;
1121 out_unlock:
1122         mutex_unlock(&lli->lli_och_mutex);
1123         return rc;
1124 }
1125
1126 /**
1127  * Release ownership on lli_mds_*_och when putting back a file lease.
1128  */
1129 static int ll_lease_och_release(struct inode *inode, struct file *file)
1130 {
1131         struct ll_inode_info *lli = ll_i2info(inode);
1132         struct ll_file_data *fd = file->private_data;
1133         struct obd_client_handle **och_p;
1134         struct obd_client_handle *old_och = NULL;
1135         __u64 *och_usecount;
1136         int rc = 0;
1137         ENTRY;
1138
1139         mutex_lock(&lli->lli_och_mutex);
1140         if (file->f_mode & FMODE_WRITE) {
1141                 och_p = &lli->lli_mds_write_och;
1142                 och_usecount = &lli->lli_open_fd_write_count;
1143         } else {
1144                 och_p = &lli->lli_mds_read_och;
1145                 och_usecount = &lli->lli_open_fd_read_count;
1146         }
1147
1148         /* The file may have been open by another process (broken lease) so
1149          * *och_p is not NULL. In this case we should simply increase usecount
1150          * and close fd_och.
1151          */
1152         if (*och_p != NULL) {
1153                 old_och = fd->fd_och;
1154                 (*och_usecount)++;
1155         } else {
1156                 *och_p = fd->fd_och;
1157                 *och_usecount = 1;
1158         }
1159         fd->fd_och = NULL;
1160         mutex_unlock(&lli->lli_och_mutex);
1161
1162         if (old_och != NULL)
1163                 rc = ll_close_inode_openhandle(inode, old_och, 0, NULL);
1164
1165         RETURN(rc);
1166 }
1167
1168 /**
1169  * Acquire a lease and open the file.
1170  */
1171 static struct obd_client_handle *
1172 ll_lease_open(struct inode *inode, struct file *file, fmode_t fmode,
1173               __u64 open_flags)
1174 {
1175         struct lookup_intent it = { .it_op = IT_OPEN };
1176         struct ll_sb_info *sbi = ll_i2sbi(inode);
1177         struct md_op_data *op_data;
1178         struct ptlrpc_request *req = NULL;
1179         struct lustre_handle old_open_handle = { 0 };
1180         struct obd_client_handle *och = NULL;
1181         int rc;
1182         int rc2;
1183         ENTRY;
1184
1185         if (fmode != FMODE_WRITE && fmode != FMODE_READ)
1186                 RETURN(ERR_PTR(-EINVAL));
1187
1188         if (file != NULL) {
1189                 if (!(fmode & file->f_mode) || (file->f_mode & FMODE_EXEC))
1190                         RETURN(ERR_PTR(-EPERM));
1191
1192                 rc = ll_lease_och_acquire(inode, file, &old_open_handle);
1193                 if (rc)
1194                         RETURN(ERR_PTR(rc));
1195         }
1196
1197         OBD_ALLOC_PTR(och);
1198         if (och == NULL)
1199                 RETURN(ERR_PTR(-ENOMEM));
1200
1201         op_data = ll_prep_md_op_data(NULL, inode, inode, NULL, 0, 0,
1202                                         LUSTRE_OPC_ANY, NULL);
1203         if (IS_ERR(op_data))
1204                 GOTO(out, rc = PTR_ERR(op_data));
1205
1206         /* To tell the MDT this openhandle is from the same owner */
1207         op_data->op_open_handle = old_open_handle;
1208
1209         it.it_flags = fmode | open_flags;
1210         it.it_flags |= MDS_OPEN_LOCK | MDS_OPEN_BY_FID | MDS_OPEN_LEASE;
1211         rc = md_intent_lock(sbi->ll_md_exp, op_data, &it, &req,
1212                             &ll_md_blocking_lease_ast,
1213         /* LDLM_FL_NO_LRU: To not put the lease lock into LRU list, otherwise
1214          * it can be cancelled which may mislead applications that the lease is
1215          * broken;
1216          * LDLM_FL_EXCL: Set this flag so that it won't be matched by normal
1217          * open in ll_md_blocking_ast(). Otherwise as ll_md_blocking_lease_ast
1218          * doesn't deal with openhandle, so normal openhandle will be leaked. */
1219                             LDLM_FL_NO_LRU | LDLM_FL_EXCL);
1220         ll_finish_md_op_data(op_data);
1221         ptlrpc_req_finished(req);
1222         if (rc < 0)
1223                 GOTO(out_release_it, rc);
1224
1225         if (it_disposition(&it, DISP_LOOKUP_NEG))
1226                 GOTO(out_release_it, rc = -ENOENT);
1227
1228         rc = it_open_error(DISP_OPEN_OPEN, &it);
1229         if (rc)
1230                 GOTO(out_release_it, rc);
1231
1232         LASSERT(it_disposition(&it, DISP_ENQ_OPEN_REF));
1233         rc = ll_och_fill(sbi->ll_md_exp, &it, och);
1234         if (rc)
1235                 GOTO(out_release_it, rc);
1236
1237         if (!it_disposition(&it, DISP_OPEN_LEASE)) /* old server? */
1238                 GOTO(out_close, rc = -EOPNOTSUPP);
1239
1240         /* already get lease, handle lease lock */
1241         ll_set_lock_data(sbi->ll_md_exp, inode, &it, NULL);
1242         if (!it.it_lock_mode ||
1243             !(it.it_lock_bits & MDS_INODELOCK_OPEN)) {
1244                 /* open lock must return for lease */
1245                 CERROR(DFID "lease granted but no open lock, %d/%llu.\n",
1246                         PFID(ll_inode2fid(inode)), it.it_lock_mode,
1247                         it.it_lock_bits);
1248                 GOTO(out_close, rc = -EPROTO);
1249         }
1250
1251         ll_intent_release(&it);
1252         RETURN(och);
1253
1254 out_close:
1255         /* Cancel open lock */
1256         if (it.it_lock_mode != 0) {
1257                 ldlm_lock_decref_and_cancel(&och->och_lease_handle,
1258                                             it.it_lock_mode);
1259                 it.it_lock_mode = 0;
1260                 och->och_lease_handle.cookie = 0ULL;
1261         }
1262         rc2 = ll_close_inode_openhandle(inode, och, 0, NULL);
1263         if (rc2 < 0)
1264                 CERROR("%s: error closing file "DFID": %d\n",
1265                        sbi->ll_fsname, PFID(&ll_i2info(inode)->lli_fid), rc2);
1266         och = NULL; /* och has been freed in ll_close_inode_openhandle() */
1267 out_release_it:
1268         ll_intent_release(&it);
1269 out:
1270         if (och != NULL)
1271                 OBD_FREE_PTR(och);
1272         RETURN(ERR_PTR(rc));
1273 }
1274
1275 /**
1276  * Check whether a layout swap can be done between two inodes.
1277  *
1278  * \param[in] inode1  First inode to check
1279  * \param[in] inode2  Second inode to check
1280  *
1281  * \retval 0 on success, layout swap can be performed between both inodes
1282  * \retval negative error code if requirements are not met
1283  */
1284 static int ll_check_swap_layouts_validity(struct inode *inode1,
1285                                           struct inode *inode2)
1286 {
1287         if (!S_ISREG(inode1->i_mode) || !S_ISREG(inode2->i_mode))
1288                 return -EINVAL;
1289
1290         if (inode_permission(&init_user_ns, inode1, MAY_WRITE) ||
1291             inode_permission(&init_user_ns, inode2, MAY_WRITE))
1292                 return -EPERM;
1293
1294         if (inode1->i_sb != inode2->i_sb)
1295                 return -EXDEV;
1296
1297         return 0;
1298 }
1299
1300 static int ll_swap_layouts_close(struct obd_client_handle *och,
1301                                  struct inode *inode, struct inode *inode2)
1302 {
1303         const struct lu_fid     *fid1 = ll_inode2fid(inode);
1304         const struct lu_fid     *fid2;
1305         int                      rc;
1306         ENTRY;
1307
1308         CDEBUG(D_INODE, "%s: biased close of file "DFID"\n",
1309                ll_i2sbi(inode)->ll_fsname, PFID(fid1));
1310
1311         rc = ll_check_swap_layouts_validity(inode, inode2);
1312         if (rc < 0)
1313                 GOTO(out_free_och, rc);
1314
1315         /* We now know that inode2 is a lustre inode */
1316         fid2 = ll_inode2fid(inode2);
1317
1318         rc = lu_fid_cmp(fid1, fid2);
1319         if (rc == 0)
1320                 GOTO(out_free_och, rc = -EINVAL);
1321
1322         /* Close the file and {swap,merge} layouts between inode & inode2.
1323          * NB: lease lock handle is released in mdc_close_layout_swap_pack()
1324          * because we still need it to pack l_remote_handle to MDT. */
1325         rc = ll_close_inode_openhandle(inode, och, MDS_CLOSE_LAYOUT_SWAP,
1326                                        inode2);
1327
1328         och = NULL; /* freed in ll_close_inode_openhandle() */
1329
1330 out_free_och:
1331         if (och != NULL)
1332                 OBD_FREE_PTR(och);
1333
1334         RETURN(rc);
1335 }
1336
1337 /**
1338  * Release lease and close the file.
1339  * It will check if the lease has ever broken.
1340  */
1341 static int ll_lease_close_intent(struct obd_client_handle *och,
1342                                  struct inode *inode,
1343                                  bool *lease_broken, enum mds_op_bias bias,
1344                                  void *data)
1345 {
1346         struct ldlm_lock *lock;
1347         bool cancelled = true;
1348         int rc;
1349         ENTRY;
1350
1351         lock = ldlm_handle2lock(&och->och_lease_handle);
1352         if (lock != NULL) {
1353                 lock_res_and_lock(lock);
1354                 cancelled = ldlm_is_cancel(lock);
1355                 unlock_res_and_lock(lock);
1356                 LDLM_LOCK_PUT(lock);
1357         }
1358
1359         CDEBUG(D_INODE, "lease for "DFID" broken? %d, bias: %x\n",
1360                PFID(&ll_i2info(inode)->lli_fid), cancelled, bias);
1361
1362         if (lease_broken != NULL)
1363                 *lease_broken = cancelled;
1364
1365         if (!cancelled && !bias)
1366                 ldlm_cli_cancel(&och->och_lease_handle, 0);
1367
1368         if (cancelled) { /* no need to excute intent */
1369                 bias = 0;
1370                 data = NULL;
1371         }
1372
1373         rc = ll_close_inode_openhandle(inode, och, bias, data);
1374         RETURN(rc);
1375 }
1376
1377 static int ll_lease_close(struct obd_client_handle *och, struct inode *inode,
1378                           bool *lease_broken)
1379 {
1380         return ll_lease_close_intent(och, inode, lease_broken, 0, NULL);
1381 }
1382
1383 /**
1384  * After lease is taken, send the RPC MDS_REINT_RESYNC to the MDT
1385  */
1386 static int ll_lease_file_resync(struct obd_client_handle *och,
1387                                 struct inode *inode, unsigned long arg)
1388 {
1389         struct ll_sb_info *sbi = ll_i2sbi(inode);
1390         struct md_op_data *op_data;
1391         struct ll_ioc_lease_id ioc;
1392         __u64 data_version_unused;
1393         int rc;
1394         ENTRY;
1395
1396         op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
1397                                      LUSTRE_OPC_ANY, NULL);
1398         if (IS_ERR(op_data))
1399                 RETURN(PTR_ERR(op_data));
1400
1401         if (copy_from_user(&ioc, (struct ll_ioc_lease_id __user *)arg,
1402                            sizeof(ioc)))
1403                 RETURN(-EFAULT);
1404
1405         /* before starting file resync, it's necessary to clean up page cache
1406          * in client memory, otherwise once the layout version is increased,
1407          * writing back cached data will be denied the OSTs. */
1408         rc = ll_data_version(inode, &data_version_unused, LL_DV_WR_FLUSH);
1409         if (rc)
1410                 GOTO(out, rc);
1411
1412         op_data->op_lease_handle = och->och_lease_handle;
1413         op_data->op_mirror_id = ioc.lil_mirror_id;
1414         rc = md_file_resync(sbi->ll_md_exp, op_data);
1415         if (rc)
1416                 GOTO(out, rc);
1417
1418         EXIT;
1419 out:
1420         ll_finish_md_op_data(op_data);
1421         return rc;
1422 }
1423
1424 int ll_merge_attr(const struct lu_env *env, struct inode *inode)
1425 {
1426         struct ll_inode_info *lli = ll_i2info(inode);
1427         struct cl_object *obj = lli->lli_clob;
1428         struct cl_attr *attr = vvp_env_thread_attr(env);
1429         s64 atime;
1430         s64 mtime;
1431         s64 ctime;
1432         int rc = 0;
1433
1434         ENTRY;
1435
1436         ll_inode_size_lock(inode);
1437
1438         /* Merge timestamps the most recently obtained from MDS with
1439          * timestamps obtained from OSTs.
1440          *
1441          * Do not overwrite atime of inode because it may be refreshed
1442          * by file_accessed() function. If the read was served by cache
1443          * data, there is no RPC to be sent so that atime may not be
1444          * transferred to OSTs at all. MDT only updates atime at close time
1445          * if it's at least 'mdd.*.atime_diff' older.
1446          * All in all, the atime in Lustre does not strictly comply with
1447          * POSIX. Solving this problem needs to send an RPC to MDT for each
1448          * read, this will hurt performance.
1449          */
1450         if (test_and_clear_bit(LLIF_UPDATE_ATIME, &lli->lli_flags) ||
1451             inode->i_atime.tv_sec < lli->lli_atime)
1452                 inode->i_atime.tv_sec = lli->lli_atime;
1453
1454         inode->i_mtime.tv_sec = lli->lli_mtime;
1455         inode->i_ctime.tv_sec = lli->lli_ctime;
1456
1457         mtime = inode->i_mtime.tv_sec;
1458         atime = inode->i_atime.tv_sec;
1459         ctime = inode->i_ctime.tv_sec;
1460
1461         cl_object_attr_lock(obj);
1462         if (OBD_FAIL_CHECK(OBD_FAIL_MDC_MERGE))
1463                 rc = -EINVAL;
1464         else
1465                 rc = cl_object_attr_get(env, obj, attr);
1466         cl_object_attr_unlock(obj);
1467
1468         if (rc != 0)
1469                 GOTO(out_size_unlock, rc = (rc == -ENODATA ? 0 : rc));
1470
1471         if (atime < attr->cat_atime)
1472                 atime = attr->cat_atime;
1473
1474         if (ctime < attr->cat_ctime)
1475                 ctime = attr->cat_ctime;
1476
1477         if (mtime < attr->cat_mtime)
1478                 mtime = attr->cat_mtime;
1479
1480         CDEBUG(D_VFSTRACE, DFID" updating i_size %llu\n",
1481                PFID(&lli->lli_fid), attr->cat_size);
1482
1483         if (llcrypt_require_key(inode) == -ENOKEY) {
1484                 /* Without the key, round up encrypted file size to next
1485                  * LUSTRE_ENCRYPTION_UNIT_SIZE. Clear text size is put in
1486                  * lli_lazysize for proper file size setting at close time.
1487                  */
1488                 lli->lli_attr_valid |= OBD_MD_FLLAZYSIZE;
1489                 lli->lli_lazysize = attr->cat_size;
1490                 attr->cat_size = round_up(attr->cat_size,
1491                                           LUSTRE_ENCRYPTION_UNIT_SIZE);
1492         }
1493         i_size_write(inode, attr->cat_size);
1494         inode->i_blocks = attr->cat_blocks;
1495
1496         inode->i_mtime.tv_sec = mtime;
1497         inode->i_atime.tv_sec = atime;
1498         inode->i_ctime.tv_sec = ctime;
1499
1500 out_size_unlock:
1501         ll_inode_size_unlock(inode);
1502
1503         RETURN(rc);
1504 }
1505
1506 /**
1507  * Set designated mirror for I/O.
1508  *
1509  * So far only read, write, and truncated can support to issue I/O to
1510  * designated mirror.
1511  */
1512 void ll_io_set_mirror(struct cl_io *io, const struct file *file)
1513 {
1514         struct ll_file_data *fd = file->private_data;
1515
1516         /* clear layout version for generic(non-resync) I/O in case it carries
1517          * stale layout version due to I/O restart */
1518         io->ci_layout_version = 0;
1519
1520         /* FLR: disable non-delay for designated mirror I/O because obviously
1521          * only one mirror is available */
1522         if (fd->fd_designated_mirror > 0) {
1523                 io->ci_ndelay = 0;
1524                 io->ci_designated_mirror = fd->fd_designated_mirror;
1525                 io->ci_layout_version = fd->fd_layout_version;
1526         }
1527
1528         CDEBUG(D_VFSTRACE, "%s: desiginated mirror: %d\n",
1529                file->f_path.dentry->d_name.name, io->ci_designated_mirror);
1530 }
1531
1532 static bool file_is_noatime(const struct file *file)
1533 {
1534         const struct vfsmount *mnt = file->f_path.mnt;
1535         const struct inode *inode = file_inode((struct file *)file);
1536
1537         /* Adapted from file_accessed() and touch_atime().*/
1538         if (file->f_flags & O_NOATIME)
1539                 return true;
1540
1541         if (inode->i_flags & S_NOATIME)
1542                 return true;
1543
1544         if (IS_NOATIME(inode))
1545                 return true;
1546
1547         if (mnt->mnt_flags & (MNT_NOATIME | MNT_READONLY))
1548                 return true;
1549
1550         if ((mnt->mnt_flags & MNT_NODIRATIME) && S_ISDIR(inode->i_mode))
1551                 return true;
1552
1553         if ((inode->i_sb->s_flags & SB_NODIRATIME) && S_ISDIR(inode->i_mode))
1554                 return true;
1555
1556         return false;
1557 }
1558
1559 void ll_io_init(struct cl_io *io, struct file *file, enum cl_io_type iot,
1560                 struct vvp_io_args *args)
1561 {
1562         struct inode *inode = file_inode(file);
1563         struct ll_file_data *fd  = file->private_data;
1564
1565         io->u.ci_rw.crw_nonblock = file->f_flags & O_NONBLOCK;
1566         io->ci_lock_no_expand = fd->ll_lock_no_expand;
1567
1568         if (iot == CIT_WRITE) {
1569                 io->u.ci_wr.wr_append = !!(file->f_flags & O_APPEND);
1570                 io->u.ci_wr.wr_sync   = !!(file->f_flags & O_SYNC ||
1571                                            file->f_flags & O_DIRECT ||
1572                                            IS_SYNC(inode));
1573 #ifdef HAVE_GENERIC_WRITE_SYNC_2ARGS
1574                 io->u.ci_wr.wr_sync  |= !!(args &&
1575                                            (args->u.normal.via_iocb->ki_flags &
1576                                             IOCB_DSYNC));
1577 #endif
1578         }
1579
1580 #ifdef IOCB_NOWAIT
1581         io->ci_iocb_nowait = !!(args &&
1582                                 (args->u.normal.via_iocb->ki_flags &
1583                                  IOCB_NOWAIT));
1584 #endif
1585
1586         io->ci_obj = ll_i2info(inode)->lli_clob;
1587         io->ci_lockreq = CILR_MAYBE;
1588         if (ll_file_nolock(file)) {
1589                 io->ci_lockreq = CILR_NEVER;
1590                 io->ci_no_srvlock = 1;
1591         } else if (file->f_flags & O_APPEND) {
1592                 io->ci_lockreq = CILR_MANDATORY;
1593         }
1594         io->ci_noatime = file_is_noatime(file);
1595         io->ci_async_readahead = false;
1596
1597         /* FLR: only use non-delay I/O for read as there is only one
1598          * avaliable mirror for write. */
1599         io->ci_ndelay = !(iot == CIT_WRITE);
1600
1601         ll_io_set_mirror(io, file);
1602 }
1603
1604 static void ll_heat_add(struct inode *inode, enum cl_io_type iot,
1605                         __u64 count)
1606 {
1607         struct ll_inode_info *lli = ll_i2info(inode);
1608         struct ll_sb_info *sbi = ll_i2sbi(inode);
1609         enum obd_heat_type sample_type;
1610         enum obd_heat_type iobyte_type;
1611         __u64 now = ktime_get_real_seconds();
1612
1613         if (!ll_sbi_has_file_heat(sbi) ||
1614             lli->lli_heat_flags & LU_HEAT_FLAG_OFF)
1615                 return;
1616
1617         if (iot == CIT_READ) {
1618                 sample_type = OBD_HEAT_READSAMPLE;
1619                 iobyte_type = OBD_HEAT_READBYTE;
1620         } else if (iot == CIT_WRITE) {
1621                 sample_type = OBD_HEAT_WRITESAMPLE;
1622                 iobyte_type = OBD_HEAT_WRITEBYTE;
1623         } else {
1624                 return;
1625         }
1626
1627         spin_lock(&lli->lli_heat_lock);
1628         obd_heat_add(&lli->lli_heat_instances[sample_type], now, 1,
1629                      sbi->ll_heat_decay_weight, sbi->ll_heat_period_second);
1630         obd_heat_add(&lli->lli_heat_instances[iobyte_type], now, count,
1631                      sbi->ll_heat_decay_weight, sbi->ll_heat_period_second);
1632         spin_unlock(&lli->lli_heat_lock);
1633 }
1634
1635 static ssize_t
1636 ll_file_io_generic(const struct lu_env *env, struct vvp_io_args *args,
1637                    struct file *file, enum cl_io_type iot,
1638                    loff_t *ppos, size_t count)
1639 {
1640         struct vvp_io *vio = vvp_env_io(env);
1641         struct inode *inode = file_inode(file);
1642         struct ll_inode_info *lli = ll_i2info(inode);
1643         struct ll_sb_info *sbi = ll_i2sbi(inode);
1644         struct ll_file_data *fd  = file->private_data;
1645         struct range_lock range;
1646         bool range_locked = false;
1647         struct cl_io *io;
1648         ssize_t result = 0;
1649         int rc = 0;
1650         int rc2 = 0;
1651         unsigned int retried = 0, dio_lock = 0;
1652         bool is_aio = false;
1653         bool is_parallel_dio = false;
1654         struct cl_dio_aio *ci_aio = NULL;
1655         size_t per_bytes;
1656         bool partial_io = false;
1657         size_t max_io_pages, max_cached_pages;
1658
1659         ENTRY;
1660
1661         CDEBUG(D_VFSTRACE, "%s: %s ppos: %llu, count: %zu\n",
1662                 file_dentry(file)->d_name.name,
1663                 iot == CIT_READ ? "read" : "write", *ppos, count);
1664
1665         max_io_pages = PTLRPC_MAX_BRW_PAGES * OBD_MAX_RIF_DEFAULT;
1666         max_cached_pages = sbi->ll_cache->ccc_lru_max;
1667         if (max_io_pages > (max_cached_pages >> 2))
1668                 max_io_pages = max_cached_pages >> 2;
1669
1670         io = vvp_env_thread_io(env);
1671         if (file->f_flags & O_DIRECT) {
1672                 if (file->f_flags & O_APPEND)
1673                         dio_lock = 1;
1674                 if (!is_sync_kiocb(args->u.normal.via_iocb))
1675                         is_aio = true;
1676
1677                 /* the kernel does not support AIO on pipes, and parallel DIO
1678                  * uses part of the AIO path, so we must not do parallel dio
1679                  * to pipes
1680                  */
1681                 is_parallel_dio = !iov_iter_is_pipe(args->u.normal.via_iter) &&
1682                                !is_aio;
1683
1684                 if (!ll_sbi_has_parallel_dio(sbi))
1685                         is_parallel_dio = false;
1686
1687                 ci_aio = cl_aio_alloc(args->u.normal.via_iocb,
1688                                       ll_i2info(inode)->lli_clob, NULL);
1689                 if (!ci_aio)
1690                         GOTO(out, rc = -ENOMEM);
1691         }
1692
1693 restart:
1694         /**
1695          * IO block size need be aware of cached page limit, otherwise
1696          * if we have small max_cached_mb but large block IO issued, io
1697          * could not be finished and blocked whole client.
1698          */
1699         if (file->f_flags & O_DIRECT)
1700                 per_bytes = count;
1701         else
1702                 per_bytes = min(max_io_pages << PAGE_SHIFT, count);
1703         partial_io = per_bytes < count;
1704         io = vvp_env_thread_io(env);
1705         ll_io_init(io, file, iot, args);
1706         io->ci_aio = ci_aio;
1707         io->ci_dio_lock = dio_lock;
1708         io->ci_ndelay_tried = retried;
1709         io->ci_parallel_dio = is_parallel_dio;
1710
1711         if (cl_io_rw_init(env, io, iot, *ppos, per_bytes) == 0) {
1712                 if (file->f_flags & O_APPEND)
1713                         range_lock_init(&range, 0, LUSTRE_EOF);
1714                 else
1715                         range_lock_init(&range, *ppos, *ppos + per_bytes - 1);
1716
1717                 vio->vui_fd  = file->private_data;
1718                 vio->vui_iter = args->u.normal.via_iter;
1719                 vio->vui_iocb = args->u.normal.via_iocb;
1720                 /* Direct IO reads must also take range lock,
1721                  * or multiple reads will try to work on the same pages
1722                  * See LU-6227 for details.
1723                  */
1724                 if (((iot == CIT_WRITE) ||
1725                     (iot == CIT_READ && (file->f_flags & O_DIRECT))) &&
1726                     !(vio->vui_fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
1727                         CDEBUG(D_VFSTRACE, "Range lock "RL_FMT"\n",
1728                                RL_PARA(&range));
1729                         rc = range_lock(&lli->lli_write_tree, &range);
1730                         if (rc < 0)
1731                                 GOTO(out, rc);
1732
1733                         range_locked = true;
1734                 }
1735
1736                 ll_cl_add(inode, env, io, LCC_RW);
1737                 rc = cl_io_loop(env, io);
1738                 ll_cl_remove(inode, env);
1739
1740                 if (range_locked && !is_parallel_dio) {
1741                         CDEBUG(D_VFSTRACE, "Range unlock "RL_FMT"\n",
1742                                RL_PARA(&range));
1743                         range_unlock(&lli->lli_write_tree, &range);
1744                         range_locked = false;
1745                 }
1746         } else {
1747                 /* cl_io_rw_init() handled IO */
1748                 rc = io->ci_result;
1749         }
1750
1751         /* N/B: parallel DIO may be disabled during i/o submission;
1752          * if that occurs, async RPCs are resolved before we get here, and this
1753          * wait call completes immediately.
1754          */
1755         if (is_parallel_dio) {
1756                 struct cl_sync_io *anchor = &io->ci_aio->cda_sync;
1757
1758                 /* for dio, EIOCBQUEUED is an implementation detail,
1759                  * and we don't return it to userspace
1760                  */
1761                 if (rc == -EIOCBQUEUED)
1762                         rc = 0;
1763
1764                 rc2 = cl_sync_io_wait_recycle(env, anchor, 0, 0);
1765                 if (rc2 < 0)
1766                         rc = rc2;
1767
1768                 if (range_locked) {
1769                         range_unlock(&lli->lli_write_tree, &range);
1770                         range_locked = false;
1771                 }
1772         }
1773
1774         /*
1775          * In order to move forward AIO, ci_nob was increased,
1776          * but that doesn't mean io have been finished, it just
1777          * means io have been submited, we will always return
1778          * EIOCBQUEUED to the caller, So we could only return
1779          * number of bytes in non-AIO case.
1780          */
1781         if (io->ci_nob > 0) {
1782                 if (!is_aio) {
1783                         if (rc2 == 0) {
1784                                 result += io->ci_nob;
1785                                 *ppos = io->u.ci_wr.wr.crw_pos; /* for splice */
1786                         } else if (rc2) {
1787                                 result = 0;
1788                         }
1789                 }
1790                 count -= io->ci_nob;
1791
1792                 /* prepare IO restart */
1793                 if (count > 0)
1794                         args->u.normal.via_iter = vio->vui_iter;
1795
1796                 if (partial_io) {
1797                         /**
1798                          * Reexpand iov count because it was zero
1799                          * after IO finish.
1800                          */
1801                         iov_iter_reexpand(vio->vui_iter, count);
1802                         if (per_bytes == io->ci_nob)
1803                                 io->ci_need_restart = 1;
1804                 }
1805         }
1806 out:
1807         cl_io_fini(env, io);
1808
1809         CDEBUG(D_VFSTRACE,
1810                "%s: %d io complete with rc: %d, result: %zd, restart: %d\n",
1811                file->f_path.dentry->d_name.name,
1812                iot, rc, result, io->ci_need_restart);
1813
1814         if ((rc == 0 || rc == -ENODATA || rc == -ENOLCK) &&
1815             count > 0 && io->ci_need_restart) {
1816                 CDEBUG(D_VFSTRACE,
1817                        "%s: restart %s from %lld, count: %zu, ret: %zd, rc: %d\n",
1818                        file_dentry(file)->d_name.name,
1819                        iot == CIT_READ ? "read" : "write",
1820                        *ppos, count, result, rc);
1821                 /* preserve the tried count for FLR */
1822                 retried = io->ci_ndelay_tried;
1823                 dio_lock = io->ci_dio_lock;
1824                 goto restart;
1825         }
1826
1827         if (io->ci_aio) {
1828                 /*
1829                  * VFS will call aio_complete() if no -EIOCBQUEUED
1830                  * is returned for AIO, so we can not call aio_complete()
1831                  * in our end_io().
1832                  */
1833                 if (rc != -EIOCBQUEUED)
1834                         io->ci_aio->cda_no_aio_complete = 1;
1835                 /**
1836                  * Drop one extra reference so that end_io() could be
1837                  * called for this IO context, we could call it after
1838                  * we make sure all AIO requests have been proceed.
1839                  */
1840                 cl_sync_io_note(env, &io->ci_aio->cda_sync,
1841                                 rc == -EIOCBQUEUED ? 0 : rc);
1842                 if (!is_aio) {
1843                         cl_aio_free(env, io->ci_aio);
1844                         io->ci_aio = NULL;
1845                 }
1846         }
1847
1848         if (iot == CIT_READ) {
1849                 if (result > 0)
1850                         ll_stats_ops_tally(ll_i2sbi(inode),
1851                                            LPROC_LL_READ_BYTES, result);
1852         } else if (iot == CIT_WRITE) {
1853                 if (result > 0) {
1854                         ll_stats_ops_tally(ll_i2sbi(inode),
1855                                            LPROC_LL_WRITE_BYTES, result);
1856                         fd->fd_write_failed = false;
1857                 } else if (result == 0 && rc == 0) {
1858                         rc = io->ci_result;
1859                         if (rc < 0)
1860                                 fd->fd_write_failed = true;
1861                         else
1862                                 fd->fd_write_failed = false;
1863                 } else if (rc != -ERESTARTSYS) {
1864                         fd->fd_write_failed = true;
1865                 }
1866         }
1867
1868         CDEBUG(D_VFSTRACE, "iot: %d, result: %zd\n", iot, result);
1869         if (result > 0)
1870                 ll_heat_add(inode, iot, result);
1871
1872         RETURN(result > 0 ? result : rc);
1873 }
1874
1875 /**
1876  * The purpose of fast read is to overcome per I/O overhead and improve IOPS
1877  * especially for small I/O.
1878  *
1879  * To serve a read request, CLIO has to create and initialize a cl_io and
1880  * then request DLM lock. This has turned out to have siginificant overhead
1881  * and affects the performance of small I/O dramatically.
1882  *
1883  * It's not necessary to create a cl_io for each I/O. Under the help of read
1884  * ahead, most of the pages being read are already in memory cache and we can
1885  * read those pages directly because if the pages exist, the corresponding DLM
1886  * lock must exist so that page content must be valid.
1887  *
1888  * In fast read implementation, the llite speculatively finds and reads pages
1889  * in memory cache. There are three scenarios for fast read:
1890  *   - If the page exists and is uptodate, kernel VM will provide the data and
1891  *     CLIO won't be intervened;
1892  *   - If the page was brought into memory by read ahead, it will be exported
1893  *     and read ahead parameters will be updated;
1894  *   - Otherwise the page is not in memory, we can't do fast read. Therefore,
1895  *     it will go back and invoke normal read, i.e., a cl_io will be created
1896  *     and DLM lock will be requested.
1897  *
1898  * POSIX compliance: posix standard states that read is intended to be atomic.
1899  * Lustre read implementation is in line with Linux kernel read implementation
1900  * and neither of them complies with POSIX standard in this matter. Fast read
1901  * doesn't make the situation worse on single node but it may interleave write
1902  * results from multiple nodes due to short read handling in ll_file_aio_read().
1903  *
1904  * \param env - lu_env
1905  * \param iocb - kiocb from kernel
1906  * \param iter - user space buffers where the data will be copied
1907  *
1908  * \retval - number of bytes have been read, or error code if error occurred.
1909  */
1910 static ssize_t
1911 ll_do_fast_read(struct kiocb *iocb, struct iov_iter *iter)
1912 {
1913         ssize_t result;
1914
1915         if (!ll_sbi_has_fast_read(ll_i2sbi(file_inode(iocb->ki_filp))))
1916                 return 0;
1917
1918         /* NB: we can't do direct IO for fast read because it will need a lock
1919          * to make IO engine happy. */
1920         if (iocb->ki_filp->f_flags & O_DIRECT)
1921                 return 0;
1922
1923         result = generic_file_read_iter(iocb, iter);
1924
1925         /* If the first page is not in cache, generic_file_aio_read() will be
1926          * returned with -ENODATA.
1927          * See corresponding code in ll_readpage(). */
1928         if (result == -ENODATA)
1929                 result = 0;
1930
1931         if (result > 0) {
1932                 ll_heat_add(file_inode(iocb->ki_filp), CIT_READ, result);
1933                 ll_stats_ops_tally(ll_i2sbi(file_inode(iocb->ki_filp)),
1934                                    LPROC_LL_READ_BYTES, result);
1935         }
1936
1937         return result;
1938 }
1939
1940 /*
1941  * Read from a file (through the page cache).
1942  */
1943 static ssize_t ll_file_read_iter(struct kiocb *iocb, struct iov_iter *to)
1944 {
1945         struct lu_env *env;
1946         struct vvp_io_args *args;
1947         struct file *file = iocb->ki_filp;
1948         ssize_t result;
1949         ssize_t rc2;
1950         __u16 refcheck;
1951         ktime_t kstart = ktime_get();
1952         bool cached;
1953
1954         ENTRY;
1955
1956         CDEBUG(D_VFSTRACE|D_IOTRACE, "file %s:"DFID", ppos: %lld, count: %zu\n",
1957                file_dentry(file)->d_name.name,
1958                PFID(ll_inode2fid(file_inode(file))), iocb->ki_pos,
1959                iov_iter_count(to));
1960
1961         if (!iov_iter_count(to))
1962                 RETURN(0);
1963
1964         /**
1965          * Currently when PCC read failed, we do not fall back to the
1966          * normal read path, just return the error.
1967          * The resaon is that: for RW-PCC, the file data may be modified
1968          * in the PCC and inconsistent with the data on OSTs (or file
1969          * data has been removed from the Lustre file system), at this
1970          * time, fallback to the normal read path may read the wrong
1971          * data.
1972          * TODO: for RO-PCC (readonly PCC), fall back to normal read
1973          * path: read data from data copy on OSTs.
1974          */
1975         result = pcc_file_read_iter(iocb, to, &cached);
1976         if (cached)
1977                 GOTO(out, result);
1978
1979         ll_ras_enter(file, iocb->ki_pos, iov_iter_count(to));
1980
1981         result = ll_do_fast_read(iocb, to);
1982         if (result < 0 || iov_iter_count(to) == 0)
1983                 GOTO(out, result);
1984
1985         env = cl_env_get(&refcheck);
1986         if (IS_ERR(env))
1987                 RETURN(PTR_ERR(env));
1988
1989         args = ll_env_args(env);
1990         args->u.normal.via_iter = to;
1991         args->u.normal.via_iocb = iocb;
1992
1993         rc2 = ll_file_io_generic(env, args, file, CIT_READ,
1994                                  &iocb->ki_pos, iov_iter_count(to));
1995         if (rc2 > 0)
1996                 result += rc2;
1997         else if (result == 0)
1998                 result = rc2;
1999
2000         cl_env_put(env, &refcheck);
2001 out:
2002         if (result > 0) {
2003                 ll_rw_stats_tally(ll_i2sbi(file_inode(file)), current->pid,
2004                                   file->private_data, iocb->ki_pos, result,
2005                                   READ);
2006                 ll_stats_ops_tally(ll_i2sbi(file_inode(file)), LPROC_LL_READ,
2007                                    ktime_us_delta(ktime_get(), kstart));
2008         }
2009
2010         CDEBUG(D_IOTRACE,
2011                "COMPLETED: file %s:"DFID", ppos: %lld, count: %zu\n",
2012                file_dentry(file)->d_name.name,
2013                PFID(ll_inode2fid(file_inode(file))), iocb->ki_pos,
2014                iov_iter_count(to));
2015
2016         RETURN(result);
2017 }
2018
2019 /**
2020  * Similar trick to ll_do_fast_read, this improves write speed for tiny writes.
2021  * If a page is already in the page cache and dirty (and some other things -
2022  * See ll_tiny_write_begin for the instantiation of these rules), then we can
2023  * write to it without doing a full I/O, because Lustre already knows about it
2024  * and will write it out.  This saves a lot of processing time.
2025  *
2026  * All writes here are within one page, so exclusion is handled by the page
2027  * lock on the vm page.  We do not do tiny writes for writes which touch
2028  * multiple pages because it's very unlikely multiple sequential pages are
2029  * are already dirty.
2030  *
2031  * We limit these to < PAGE_SIZE because PAGE_SIZE writes are relatively common
2032  * and are unlikely to be to already dirty pages.
2033  *
2034  * Attribute updates are important here, we do them in ll_tiny_write_end.
2035  */
2036 static ssize_t ll_do_tiny_write(struct kiocb *iocb, struct iov_iter *iter)
2037 {
2038         ssize_t count = iov_iter_count(iter);
2039         struct  file *file = iocb->ki_filp;
2040         struct  inode *inode = file_inode(file);
2041         bool    lock_inode = !IS_NOSEC(inode);
2042         ssize_t result = 0;
2043
2044         ENTRY;
2045
2046         /* Restrict writes to single page and < PAGE_SIZE.  See comment at top
2047          * of function for why.
2048          */
2049         if (count >= PAGE_SIZE ||
2050             (iocb->ki_pos & (PAGE_SIZE-1)) + count > PAGE_SIZE)
2051                 RETURN(0);
2052
2053         if (unlikely(lock_inode))
2054                 inode_lock(inode);
2055         result = __generic_file_write_iter(iocb, iter);
2056
2057         if (unlikely(lock_inode))
2058                 inode_unlock(inode);
2059
2060         /* If the page is not already dirty, ll_tiny_write_begin returns
2061          * -ENODATA.  We continue on to normal write.
2062          */
2063         if (result == -ENODATA)
2064                 result = 0;
2065
2066         if (result > 0) {
2067                 ll_heat_add(inode, CIT_WRITE, result);
2068                 set_bit(LLIF_DATA_MODIFIED, &ll_i2info(inode)->lli_flags);
2069         }
2070
2071         CDEBUG(D_VFSTRACE, "result: %zu, original count %zu\n", result, count);
2072
2073         RETURN(result);
2074 }
2075
2076 /*
2077  * Write to a file (through the page cache).
2078  */
2079 static ssize_t ll_file_write_iter(struct kiocb *iocb, struct iov_iter *from)
2080 {
2081         struct vvp_io_args *args;
2082         struct lu_env *env;
2083         ssize_t rc_tiny = 0, rc_normal;
2084         struct file *file = iocb->ki_filp;
2085         __u16 refcheck;
2086         bool cached;
2087         ktime_t kstart = ktime_get();
2088         int result;
2089
2090         ENTRY;
2091
2092         CDEBUG(D_VFSTRACE|D_IOTRACE, "file %s:"DFID", ppos: %lld, count: %zu\n",
2093                file_dentry(file)->d_name.name,
2094                PFID(ll_inode2fid(file_inode(file))), iocb->ki_pos,
2095                iov_iter_count(from));
2096
2097         if (!iov_iter_count(from))
2098                 GOTO(out, rc_normal = 0);
2099
2100         /**
2101          * When PCC write failed, we usually do not fall back to the normal
2102          * write path, just return the error. But there is a special case when
2103          * returned error code is -ENOSPC due to running out of space on PCC HSM
2104          * bakcend. At this time, it will fall back to normal I/O path and
2105          * retry the I/O. As the file is in HSM released state, it will restore
2106          * the file data to OSTs first and redo the write again. And the
2107          * restore process will revoke the layout lock and detach the file
2108          * from PCC cache automatically.
2109          */
2110         result = pcc_file_write_iter(iocb, from, &cached);
2111         if (cached && result != -ENOSPC && result != -EDQUOT)
2112                 GOTO(out, rc_normal = result);
2113
2114         /* NB: we can't do direct IO for tiny writes because they use the page
2115          * cache, we can't do sync writes because tiny writes can't flush
2116          * pages, and we can't do append writes because we can't guarantee the
2117          * required DLM locks are held to protect file size.
2118          */
2119         if (ll_sbi_has_tiny_write(ll_i2sbi(file_inode(file))) &&
2120             !(file->f_flags & (O_DIRECT | O_SYNC | O_APPEND)))
2121                 rc_tiny = ll_do_tiny_write(iocb, from);
2122
2123         /* In case of error, go on and try normal write - Only stop if tiny
2124          * write completed I/O.
2125          */
2126         if (iov_iter_count(from) == 0)
2127                 GOTO(out, rc_normal = rc_tiny);
2128
2129         env = cl_env_get(&refcheck);
2130         if (IS_ERR(env))
2131                 RETURN(PTR_ERR(env));
2132
2133         args = ll_env_args(env);
2134         args->u.normal.via_iter = from;
2135         args->u.normal.via_iocb = iocb;
2136
2137         rc_normal = ll_file_io_generic(env, args, file, CIT_WRITE,
2138                                        &iocb->ki_pos, iov_iter_count(from));
2139
2140         /* On success, combine bytes written. */
2141         if (rc_tiny >= 0 && rc_normal > 0)
2142                 rc_normal += rc_tiny;
2143         /* On error, only return error from normal write if tiny write did not
2144          * write any bytes.  Otherwise return bytes written by tiny write.
2145          */
2146         else if (rc_tiny > 0)
2147                 rc_normal = rc_tiny;
2148
2149         cl_env_put(env, &refcheck);
2150 out:
2151         if (rc_normal > 0) {
2152                 ll_rw_stats_tally(ll_i2sbi(file_inode(file)), current->pid,
2153                                   file->private_data, iocb->ki_pos,
2154                                   rc_normal, WRITE);
2155                 ll_stats_ops_tally(ll_i2sbi(file_inode(file)), LPROC_LL_WRITE,
2156                                    ktime_us_delta(ktime_get(), kstart));
2157         }
2158
2159         CDEBUG(D_IOTRACE,
2160                "COMPLETED: file %s:"DFID", ppos: %lld, count: %zu\n",
2161                file_dentry(file)->d_name.name,
2162                PFID(ll_inode2fid(file_inode(file))), iocb->ki_pos,
2163                iov_iter_count(from));
2164
2165         RETURN(rc_normal);
2166 }
2167
2168 #ifndef HAVE_FILE_OPERATIONS_READ_WRITE_ITER
2169 /*
2170  * XXX: exact copy from kernel code (__generic_file_aio_write_nolock)
2171  */
2172 static int ll_file_get_iov_count(const struct iovec *iov,
2173                                  unsigned long *nr_segs, size_t *count,
2174                                  int access_flags)
2175 {
2176         size_t cnt = 0;
2177         unsigned long seg;
2178
2179         for (seg = 0; seg < *nr_segs; seg++) {
2180                 const struct iovec *iv = &iov[seg];
2181
2182                 /*
2183                  * If any segment has a negative length, or the cumulative
2184                  * length ever wraps negative then return -EINVAL.
2185                  */
2186                 cnt += iv->iov_len;
2187                 if (unlikely((ssize_t)(cnt|iv->iov_len) < 0))
2188                         return -EINVAL;
2189                 if (access_ok(access_flags, iv->iov_base, iv->iov_len))
2190                         continue;
2191                 if (seg == 0)
2192                         return -EFAULT;
2193                 *nr_segs = seg;
2194                 cnt -= iv->iov_len;     /* This segment is no good */
2195                 break;
2196         }
2197         *count = cnt;
2198         return 0;
2199 }
2200
2201 static ssize_t ll_file_aio_read(struct kiocb *iocb, const struct iovec *iov,
2202                                 unsigned long nr_segs, loff_t pos)
2203 {
2204         struct iov_iter to;
2205         size_t iov_count;
2206         ssize_t result;
2207         ENTRY;
2208
2209         result = ll_file_get_iov_count(iov, &nr_segs, &iov_count, VERIFY_READ);
2210         if (result)
2211                 RETURN(result);
2212
2213         if (!iov_count)
2214                 RETURN(0);
2215
2216 # ifdef HAVE_IOV_ITER_INIT_DIRECTION
2217         iov_iter_init(&to, READ, iov, nr_segs, iov_count);
2218 # else /* !HAVE_IOV_ITER_INIT_DIRECTION */
2219         iov_iter_init(&to, iov, nr_segs, iov_count, 0);
2220 # endif /* HAVE_IOV_ITER_INIT_DIRECTION */
2221
2222         result = ll_file_read_iter(iocb, &to);
2223
2224         RETURN(result);
2225 }
2226
2227 static ssize_t ll_file_read(struct file *file, char __user *buf, size_t count,
2228                             loff_t *ppos)
2229 {
2230         struct iovec   iov = { .iov_base = buf, .iov_len = count };
2231         struct kiocb   kiocb;
2232         ssize_t        result;
2233
2234         ENTRY;
2235
2236         if (!count)
2237                 RETURN(0);
2238
2239         init_sync_kiocb(&kiocb, file);
2240         kiocb.ki_pos = *ppos;
2241 #ifdef HAVE_KIOCB_KI_LEFT
2242         kiocb.ki_left = count;
2243 #elif defined(HAVE_KI_NBYTES)
2244         kiocb.i_nbytes = count;
2245 #endif
2246
2247         result = ll_file_aio_read(&kiocb, &iov, 1, kiocb.ki_pos);
2248         *ppos = kiocb.ki_pos;
2249
2250         RETURN(result);
2251 }
2252
2253 /*
2254  * Write to a file (through the page cache).
2255  * AIO stuff
2256  */
2257 static ssize_t ll_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
2258                                  unsigned long nr_segs, loff_t pos)
2259 {
2260         struct iov_iter from;
2261         size_t iov_count;
2262         ssize_t result;
2263         ENTRY;
2264
2265         result = ll_file_get_iov_count(iov, &nr_segs, &iov_count, VERIFY_WRITE);
2266         if (result)
2267                 RETURN(result);
2268
2269         if (!iov_count)
2270                 RETURN(0);
2271
2272 # ifdef HAVE_IOV_ITER_INIT_DIRECTION
2273         iov_iter_init(&from, WRITE, iov, nr_segs, iov_count);
2274 # else /* !HAVE_IOV_ITER_INIT_DIRECTION */
2275         iov_iter_init(&from, iov, nr_segs, iov_count, 0);
2276 # endif /* HAVE_IOV_ITER_INIT_DIRECTION */
2277
2278         result = ll_file_write_iter(iocb, &from);
2279
2280         RETURN(result);
2281 }
2282
2283 static ssize_t ll_file_write(struct file *file, const char __user *buf,
2284                              size_t count, loff_t *ppos)
2285 {
2286         struct iovec   iov = { .iov_base = (void __user *)buf,
2287                                .iov_len = count };
2288         struct kiocb   kiocb;
2289         ssize_t        result;
2290
2291         ENTRY;
2292
2293         if (!count)
2294                 RETURN(0);
2295
2296         init_sync_kiocb(&kiocb, file);
2297         kiocb.ki_pos = *ppos;
2298 #ifdef HAVE_KIOCB_KI_LEFT
2299         kiocb.ki_left = count;
2300 #elif defined(HAVE_KI_NBYTES)
2301         kiocb.ki_nbytes = count;
2302 #endif
2303
2304         result = ll_file_aio_write(&kiocb, &iov, 1, kiocb.ki_pos);
2305         *ppos = kiocb.ki_pos;
2306
2307         RETURN(result);
2308 }
2309 #endif /* !HAVE_FILE_OPERATIONS_READ_WRITE_ITER */
2310
2311 int ll_lov_setstripe_ea_info(struct inode *inode, struct dentry *dentry,
2312                              __u64 flags, struct lov_user_md *lum, int lum_size)
2313 {
2314         struct lookup_intent oit = {
2315                 .it_op = IT_OPEN,
2316                 .it_flags = flags | MDS_OPEN_BY_FID,
2317         };
2318         int rc;
2319         ENTRY;
2320
2321         if ((__swab32(lum->lmm_magic) & le32_to_cpu(LOV_MAGIC_MASK)) ==
2322             le32_to_cpu(LOV_MAGIC_MAGIC)) {
2323                 /* this code will only exist for big-endian systems */
2324                 lustre_swab_lov_user_md(lum, 0);
2325         }
2326
2327         ll_inode_size_lock(inode);
2328         rc = ll_intent_file_open(dentry, lum, lum_size, &oit);
2329         if (rc < 0)
2330                 GOTO(out_unlock, rc);
2331
2332         ll_release_openhandle(dentry, &oit);
2333
2334 out_unlock:
2335         ll_inode_size_unlock(inode);
2336         ll_intent_release(&oit);
2337
2338         RETURN(rc);
2339 }
2340
2341 int ll_lov_getstripe_ea_info(struct inode *inode, const char *filename,
2342                              struct lov_mds_md **lmmp, int *lmm_size,
2343                              struct ptlrpc_request **request)
2344 {
2345         struct ll_sb_info *sbi = ll_i2sbi(inode);
2346         struct mdt_body  *body;
2347         struct lov_mds_md *lmm = NULL;
2348         struct ptlrpc_request *req = NULL;
2349         struct md_op_data *op_data;
2350         int rc, lmmsize;
2351
2352         ENTRY;
2353
2354         rc = ll_get_default_mdsize(sbi, &lmmsize);
2355         if (rc)
2356                 RETURN(rc);
2357
2358         op_data = ll_prep_md_op_data(NULL, inode, NULL, filename,
2359                                      strlen(filename), lmmsize,
2360                                      LUSTRE_OPC_ANY, NULL);
2361         if (IS_ERR(op_data))
2362                 RETURN(PTR_ERR(op_data));
2363
2364         op_data->op_valid = OBD_MD_FLEASIZE | OBD_MD_FLDIREA;
2365         rc = md_getattr_name(sbi->ll_md_exp, op_data, &req);
2366         ll_finish_md_op_data(op_data);
2367         if (rc < 0) {
2368                 CDEBUG(D_INFO, "md_getattr_name failed on %s: rc %d\n",
2369                        filename, rc);
2370                 GOTO(out, rc);
2371         }
2372
2373         body = req_capsule_server_get(&req->rq_pill, &RMF_MDT_BODY);
2374         LASSERT(body != NULL); /* checked by mdc_getattr_name */
2375
2376         lmmsize = body->mbo_eadatasize;
2377
2378         if (!(body->mbo_valid & (OBD_MD_FLEASIZE | OBD_MD_FLDIREA)) ||
2379             lmmsize == 0)
2380                 GOTO(out, rc = -ENODATA);
2381
2382         lmm = req_capsule_server_sized_get(&req->rq_pill, &RMF_MDT_MD, lmmsize);
2383         LASSERT(lmm != NULL);
2384
2385         if (lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V1) &&
2386             lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_V3) &&
2387             lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_COMP_V1) &&
2388             lmm->lmm_magic != cpu_to_le32(LOV_MAGIC_FOREIGN))
2389                 GOTO(out, rc = -EPROTO);
2390
2391         /*
2392          * This is coming from the MDS, so is probably in
2393          * little endian. We convert it to host endian before
2394          * passing it to userspace.
2395          */
2396         if (cpu_to_le32(LOV_MAGIC) != LOV_MAGIC) {
2397                 int stripe_count = 0;
2398
2399                 if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V1) ||
2400                     lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_V3)) {
2401                         stripe_count = le16_to_cpu(lmm->lmm_stripe_count);
2402                         if (le32_to_cpu(lmm->lmm_pattern) &
2403                             LOV_PATTERN_F_RELEASED)
2404                                 stripe_count = 0;
2405                         lustre_swab_lov_user_md((struct lov_user_md *)lmm, 0);
2406
2407                         /* if function called for directory - we should
2408                          * avoid swab not existent lsm objects
2409                          */
2410                         if (lmm->lmm_magic == LOV_MAGIC_V1 &&
2411                             S_ISREG(body->mbo_mode))
2412                                 lustre_swab_lov_user_md_objects(
2413                                 ((struct lov_user_md_v1 *)lmm)->lmm_objects,
2414                                 stripe_count);
2415                         else if (lmm->lmm_magic == LOV_MAGIC_V3 &&
2416                                  S_ISREG(body->mbo_mode))
2417                                 lustre_swab_lov_user_md_objects(
2418                                 ((struct lov_user_md_v3 *)lmm)->lmm_objects,
2419                                 stripe_count);
2420                 } else if (lmm->lmm_magic == cpu_to_le32(LOV_MAGIC_COMP_V1)) {
2421                         lustre_swab_lov_comp_md_v1(
2422                                 (struct lov_comp_md_v1 *)lmm);
2423                 }
2424         }
2425
2426         if (lmm->lmm_magic == LOV_MAGIC_COMP_V1) {
2427                 struct lov_comp_md_v1 *comp_v1 = NULL;
2428                 struct lov_comp_md_entry_v1 *ent;
2429                 struct lov_user_md_v1 *v1;
2430                 __u32 off;
2431                 int i = 0;
2432
2433                 comp_v1 = (struct lov_comp_md_v1 *)lmm;
2434                 /* Dump the striping information */
2435                 for (; i < comp_v1->lcm_entry_count; i++) {
2436                         ent = &comp_v1->lcm_entries[i];
2437                         off = ent->lcme_offset;
2438                         v1 = (struct lov_user_md_v1 *)((char *)lmm + off);
2439                         CDEBUG(D_INFO,
2440                                "comp[%d]: stripe_count=%u, stripe_size=%u\n",
2441                                i, v1->lmm_stripe_count, v1->lmm_stripe_size);
2442                 }
2443
2444                 /**
2445                  * Return valid stripe_count and stripe_size instead of 0 for
2446                  * DoM files to avoid divide-by-zero for older userspace that
2447                  * calls this ioctl, e.g. lustre ADIO driver.
2448                  */
2449                 if (lmm->lmm_stripe_count == 0)
2450                         lmm->lmm_stripe_count = 1;
2451                 if (lmm->lmm_stripe_size == 0) {
2452                         /* Since the first component of the file data is placed
2453                          * on the MDT for faster access, the stripe_size of the
2454                          * second one is always that applications which are
2455                          * doing large IOs.
2456                          */
2457                         if (lmm->lmm_pattern == LOV_PATTERN_MDT)
2458                                 i = comp_v1->lcm_entry_count > 1 ? 1 : 0;
2459                         else
2460                                 i = comp_v1->lcm_entry_count > 1 ?
2461                                     comp_v1->lcm_entry_count - 1 : 0;
2462                         ent = &comp_v1->lcm_entries[i];
2463                         off = ent->lcme_offset;
2464                         v1 = (struct lov_user_md_v1 *)((char *)lmm + off);
2465                         lmm->lmm_stripe_size = v1->lmm_stripe_size;
2466                 }
2467         }
2468 out:
2469         *lmmp = lmm;
2470         *lmm_size = lmmsize;
2471         *request = req;
2472         RETURN(rc);
2473 }
2474
2475 static int ll_lov_setea(struct inode *inode, struct file *file,
2476                         void __user *arg)
2477 {
2478         __u64                    flags = MDS_OPEN_HAS_OBJS | FMODE_WRITE;
2479         struct lov_user_md      *lump;
2480         int                      lum_size = sizeof(struct lov_user_md) +
2481                                             sizeof(struct lov_user_ost_data);
2482         int                      rc;
2483         ENTRY;
2484
2485         if (!capable(CAP_SYS_ADMIN))
2486                 RETURN(-EPERM);
2487
2488         OBD_ALLOC_LARGE(lump, lum_size);
2489         if (lump == NULL)
2490                 RETURN(-ENOMEM);
2491
2492         if (copy_from_user(lump, arg, lum_size))
2493                 GOTO(out_lump, rc = -EFAULT);
2494
2495         rc = ll_lov_setstripe_ea_info(inode, file_dentry(file), flags, lump,
2496                                       lum_size);
2497         cl_lov_delay_create_clear(&file->f_flags);
2498
2499 out_lump:
2500         OBD_FREE_LARGE(lump, lum_size);
2501         RETURN(rc);
2502 }
2503
2504 static int ll_file_getstripe(struct inode *inode, void __user *lum, size_t size)
2505 {
2506         struct lu_env   *env;
2507         __u16           refcheck;
2508         int             rc;
2509         ENTRY;
2510
2511         env = cl_env_get(&refcheck);
2512         if (IS_ERR(env))
2513                 RETURN(PTR_ERR(env));
2514
2515         rc = cl_object_getstripe(env, ll_i2info(inode)->lli_clob, lum, size);
2516         cl_env_put(env, &refcheck);
2517         RETURN(rc);
2518 }
2519
2520 static int ll_lov_setstripe(struct inode *inode, struct file *file,
2521                             void __user *arg)
2522 {
2523         struct lov_user_md __user *lum = (struct lov_user_md __user *)arg;
2524         struct lov_user_md        *klum;
2525         int                        lum_size, rc;
2526         __u64                      flags = FMODE_WRITE;
2527         ENTRY;
2528
2529         rc = ll_copy_user_md(lum, &klum);
2530         if (rc < 0)
2531                 RETURN(rc);
2532
2533         lum_size = rc;
2534         rc = ll_lov_setstripe_ea_info(inode, file_dentry(file), flags, klum,
2535                                       lum_size);
2536         if (!rc) {
2537                 __u32 gen;
2538
2539                 rc = put_user(0, &lum->lmm_stripe_count);
2540                 if (rc)
2541                         GOTO(out, rc);
2542
2543                 rc = ll_layout_refresh(inode, &gen);
2544                 if (rc)
2545                         GOTO(out, rc);
2546
2547                 rc = ll_file_getstripe(inode, arg, lum_size);
2548                 if (S_ISREG(inode->i_mode) && IS_ENCRYPTED(inode) &&
2549                     ll_i2info(inode)->lli_clob) {
2550                         struct iattr attr = { 0 };
2551
2552                         rc = cl_setattr_ost(ll_i2info(inode)->lli_clob, &attr,
2553                                             OP_XVALID_FLAGS, LUSTRE_ENCRYPT_FL);
2554                 }
2555         }
2556         cl_lov_delay_create_clear(&file->f_flags);
2557
2558 out:
2559         OBD_FREE_LARGE(klum, lum_size);
2560         RETURN(rc);
2561 }
2562
2563
2564 static int
2565 ll_get_grouplock(struct inode *inode, struct file *file, unsigned long arg)
2566 {
2567         struct ll_inode_info *lli = ll_i2info(inode);
2568         struct cl_object *obj = lli->lli_clob;
2569         struct ll_file_data *fd = file->private_data;
2570         struct ll_grouplock grouplock;
2571         int rc;
2572         ENTRY;
2573
2574         if (arg == 0) {
2575                 CWARN("group id for group lock must not be 0\n");
2576                 RETURN(-EINVAL);
2577         }
2578
2579         if (ll_file_nolock(file))
2580                 RETURN(-EOPNOTSUPP);
2581 retry:
2582         if (file->f_flags & O_NONBLOCK) {
2583                 if (!mutex_trylock(&lli->lli_group_mutex))
2584                         RETURN(-EAGAIN);
2585         } else
2586                 mutex_lock(&lli->lli_group_mutex);
2587
2588         if (fd->fd_flags & LL_FILE_GROUP_LOCKED) {
2589                 CWARN("group lock already existed with gid %lu\n",
2590                       fd->fd_grouplock.lg_gid);
2591                 GOTO(out, rc = -EINVAL);
2592         }
2593         if (arg != lli->lli_group_gid && lli->lli_group_users != 0) {
2594                 if (file->f_flags & O_NONBLOCK)
2595                         GOTO(out, rc = -EAGAIN);
2596                 mutex_unlock(&lli->lli_group_mutex);
2597                 wait_var_event(&lli->lli_group_users, !lli->lli_group_users);
2598                 GOTO(retry, rc = 0);
2599         }
2600         LASSERT(fd->fd_grouplock.lg_lock == NULL);
2601
2602         /**
2603          * XXX: group lock needs to protect all OST objects while PFL
2604          * can add new OST objects during the IO, so we'd instantiate
2605          * all OST objects before getting its group lock.
2606          */
2607         if (obj) {
2608                 struct lu_env *env;
2609                 __u16 refcheck;
2610                 struct cl_layout cl = {
2611                         .cl_is_composite = false,
2612                 };
2613                 struct lu_extent ext = {
2614                         .e_start = 0,
2615                         .e_end = OBD_OBJECT_EOF,
2616                 };
2617
2618                 env = cl_env_get(&refcheck);
2619                 if (IS_ERR(env))
2620                         GOTO(out, rc = PTR_ERR(env));
2621
2622                 rc = cl_object_layout_get(env, obj, &cl);
2623                 if (rc >= 0 && cl.cl_is_composite)
2624                         rc = ll_layout_write_intent(inode, LAYOUT_INTENT_WRITE,
2625                                                     &ext);
2626
2627                 cl_env_put(env, &refcheck);
2628                 if (rc < 0)
2629                         GOTO(out, rc);
2630         }
2631
2632         rc = cl_get_grouplock(ll_i2info(inode)->lli_clob,
2633                               arg, (file->f_flags & O_NONBLOCK), &grouplock);
2634
2635         if (rc)
2636                 GOTO(out, rc);
2637
2638         fd->fd_flags |= LL_FILE_GROUP_LOCKED;
2639         fd->fd_grouplock = grouplock;
2640         if (lli->lli_group_users == 0)
2641                 lli->lli_group_gid = grouplock.lg_gid;
2642         lli->lli_group_users++;
2643
2644         CDEBUG(D_INFO, "group lock %lu obtained\n", arg);
2645 out:
2646         mutex_unlock(&lli->lli_group_mutex);
2647
2648         RETURN(rc);
2649 }
2650
2651 static int ll_put_grouplock(struct inode *inode, struct file *file,
2652                             unsigned long arg)
2653 {
2654         struct ll_inode_info   *lli = ll_i2info(inode);
2655         struct ll_file_data    *fd = file->private_data;
2656         struct ll_grouplock     grouplock;
2657         int                     rc;
2658         ENTRY;
2659
2660         mutex_lock(&lli->lli_group_mutex);
2661         if (!(fd->fd_flags & LL_FILE_GROUP_LOCKED)) {
2662                 CWARN("no group lock held\n");
2663                 GOTO(out, rc = -EINVAL);
2664         }
2665
2666         LASSERT(fd->fd_grouplock.lg_lock != NULL);
2667
2668         if (fd->fd_grouplock.lg_gid != arg) {
2669                 CWARN("group lock %lu doesn't match current id %lu\n",
2670                       arg, fd->fd_grouplock.lg_gid);
2671                 GOTO(out, rc = -EINVAL);
2672         }
2673
2674         grouplock = fd->fd_grouplock;
2675         memset(&fd->fd_grouplock, 0, sizeof(fd->fd_grouplock));
2676         fd->fd_flags &= ~LL_FILE_GROUP_LOCKED;
2677
2678         cl_put_grouplock(&grouplock);
2679
2680         lli->lli_group_users--;
2681         if (lli->lli_group_users == 0) {
2682                 lli->lli_group_gid = 0;
2683                 wake_up_var(&lli->lli_group_users);
2684         }
2685         CDEBUG(D_INFO, "group lock %lu released\n", arg);
2686         GOTO(out, rc = 0);
2687 out:
2688         mutex_unlock(&lli->lli_group_mutex);
2689
2690         RETURN(rc);
2691 }
2692
2693 /**
2694  * Close inode open handle
2695  *
2696  * \param dentry [in]     dentry which contains the inode
2697  * \param it     [in,out] intent which contains open info and result
2698  *
2699  * \retval 0     success
2700  * \retval <0    failure
2701  */
2702 int ll_release_openhandle(struct dentry *dentry, struct lookup_intent *it)
2703 {
2704         struct inode *inode = dentry->d_inode;
2705         struct obd_client_handle *och;
2706         int rc;
2707         ENTRY;
2708
2709         LASSERT(inode);
2710
2711         /* Root ? Do nothing. */
2712         if (is_root_inode(inode))
2713                 RETURN(0);
2714
2715         /* No open handle to close? Move away */
2716         if (!it_disposition(it, DISP_OPEN_OPEN))
2717                 RETURN(0);
2718
2719         LASSERT(it_open_error(DISP_OPEN_OPEN, it) == 0);
2720
2721         OBD_ALLOC(och, sizeof(*och));
2722         if (!och)
2723                 GOTO(out, rc = -ENOMEM);
2724
2725         rc = ll_och_fill(ll_i2sbi(inode)->ll_md_exp, it, och);
2726         if (rc)
2727                 GOTO(out, rc);
2728
2729         rc = ll_close_inode_openhandle(inode, och, 0, NULL);
2730 out:
2731         /* this one is in place of ll_file_open */
2732         if (it_disposition(it, DISP_ENQ_OPEN_REF)) {
2733                 ptlrpc_req_finished(it->it_request);
2734                 it_clear_disposition(it, DISP_ENQ_OPEN_REF);
2735         }
2736         RETURN(rc);
2737 }
2738
2739 /**
2740  * Get size for inode for which FIEMAP mapping is requested.
2741  * Make the FIEMAP get_info call and returns the result.
2742  * \param fiemap        kernel buffer to hold extens
2743  * \param num_bytes     kernel buffer size
2744  */
2745 static int ll_do_fiemap(struct inode *inode, struct fiemap *fiemap,
2746                         size_t num_bytes)
2747 {
2748         struct lu_env                   *env;
2749         __u16                           refcheck;
2750         int                             rc = 0;
2751         struct ll_fiemap_info_key       fmkey = { .lfik_name = KEY_FIEMAP, };
2752         ENTRY;
2753
2754         /* Checks for fiemap flags */
2755         if (fiemap->fm_flags & ~LUSTRE_FIEMAP_FLAGS_COMPAT) {
2756                 fiemap->fm_flags &= ~LUSTRE_FIEMAP_FLAGS_COMPAT;
2757                 return -EBADR;
2758         }
2759
2760         /* Check for FIEMAP_FLAG_SYNC */
2761         if (fiemap->fm_flags & FIEMAP_FLAG_SYNC) {
2762                 rc = filemap_fdatawrite(inode->i_mapping);
2763                 if (rc)
2764                         return rc;
2765         }
2766
2767         env = cl_env_get(&refcheck);
2768         if (IS_ERR(env))
2769                 RETURN(PTR_ERR(env));
2770
2771         if (i_size_read(inode) == 0) {
2772                 rc = ll_glimpse_size(inode);
2773                 if (rc)
2774                         GOTO(out, rc);
2775         }
2776
2777         fmkey.lfik_oa.o_valid = OBD_MD_FLID | OBD_MD_FLGROUP | OBD_MD_FLPROJID;
2778         obdo_from_inode(&fmkey.lfik_oa, inode, OBD_MD_FLSIZE);
2779         obdo_set_parent_fid(&fmkey.lfik_oa, &ll_i2info(inode)->lli_fid);
2780
2781         /* If filesize is 0, then there would be no objects for mapping */
2782         if (fmkey.lfik_oa.o_size == 0) {
2783                 fiemap->fm_mapped_extents = 0;
2784                 GOTO(out, rc = 0);
2785         }
2786
2787         fmkey.lfik_fiemap = *fiemap;
2788
2789         rc = cl_object_fiemap(env, ll_i2info(inode)->lli_clob,
2790                               &fmkey, fiemap, &num_bytes);
2791 out:
2792         cl_env_put(env, &refcheck);
2793         RETURN(rc);
2794 }
2795
2796 int ll_fid2path(struct inode *inode, void __user *arg)
2797 {
2798         struct obd_export       *exp = ll_i2mdexp(inode);
2799         const struct getinfo_fid2path __user *gfin = arg;
2800         __u32                    pathlen;
2801         struct getinfo_fid2path *gfout;
2802         size_t                   outsize;
2803         int                      rc;
2804
2805         ENTRY;
2806
2807         if (!capable(CAP_DAC_READ_SEARCH) &&
2808             !test_bit(LL_SBI_USER_FID2PATH, ll_i2sbi(inode)->ll_flags))
2809                 RETURN(-EPERM);
2810
2811         /* Only need to get the buflen */
2812         if (get_user(pathlen, &gfin->gf_pathlen))
2813                 RETURN(-EFAULT);
2814
2815         if (pathlen > PATH_MAX)
2816                 RETURN(-EINVAL);
2817
2818         outsize = sizeof(*gfout) + pathlen;
2819         OBD_ALLOC(gfout, outsize);
2820         if (gfout == NULL)
2821                 RETURN(-ENOMEM);
2822
2823         if (copy_from_user(gfout, arg, sizeof(*gfout)))
2824                 GOTO(gf_free, rc = -EFAULT);
2825         /* append root FID after gfout to let MDT know the root FID so that it
2826          * can lookup the correct path, this is mainly for fileset.
2827          * old server without fileset mount support will ignore this. */
2828         *gfout->gf_u.gf_root_fid = *ll_inode2fid(inode);
2829
2830         /* Call mdc_iocontrol */
2831         rc = obd_iocontrol(OBD_IOC_FID2PATH, exp, outsize, gfout, NULL);
2832         if (rc != 0)
2833                 GOTO(gf_free, rc);
2834
2835         if (copy_to_user(arg, gfout, outsize))
2836                 rc = -EFAULT;
2837
2838 gf_free:
2839         OBD_FREE(gfout, outsize);
2840         RETURN(rc);
2841 }
2842
2843 static int
2844 ll_ioc_data_version(struct inode *inode, struct ioc_data_version *ioc)
2845 {
2846         struct cl_object *obj = ll_i2info(inode)->lli_clob;
2847         struct lu_env *env;
2848         struct cl_io *io;
2849         __u16  refcheck;
2850         int result;
2851
2852         ENTRY;
2853
2854         ioc->idv_version = 0;
2855         ioc->idv_layout_version = UINT_MAX;
2856
2857         /* If no file object initialized, we consider its version is 0. */
2858         if (obj == NULL)
2859                 RETURN(0);
2860
2861         env = cl_env_get(&refcheck);
2862         if (IS_ERR(env))
2863                 RETURN(PTR_ERR(env));
2864
2865         io = vvp_env_thread_io(env);
2866         io->ci_obj = obj;
2867         io->u.ci_data_version.dv_data_version = 0;
2868         io->u.ci_data_version.dv_layout_version = UINT_MAX;
2869         io->u.ci_data_version.dv_flags = ioc->idv_flags;
2870
2871 restart:
2872         if (cl_io_init(env, io, CIT_DATA_VERSION, io->ci_obj) == 0)
2873                 result = cl_io_loop(env, io);
2874         else
2875                 result = io->ci_result;
2876
2877         ioc->idv_version = io->u.ci_data_version.dv_data_version;
2878         ioc->idv_layout_version = io->u.ci_data_version.dv_layout_version;
2879
2880         cl_io_fini(env, io);
2881
2882         if (unlikely(io->ci_need_restart))
2883                 goto restart;
2884
2885         cl_env_put(env, &refcheck);
2886
2887         RETURN(result);
2888 }
2889
2890 /*
2891  * Read the data_version for inode.
2892  *
2893  * This value is computed using stripe object version on OST.
2894  * Version is computed using server side locking.
2895  *
2896  * @param flags if do sync on the OST side;
2897  *              0: no sync
2898  *              LL_DV_RD_FLUSH: flush dirty pages, LCK_PR on OSTs
2899  *              LL_DV_WR_FLUSH: drop all caching pages, LCK_PW on OSTs
2900  */
2901 int ll_data_version(struct inode *inode, __u64 *data_version, int flags)
2902 {
2903         struct ioc_data_version ioc = { .idv_flags = flags };
2904         int rc;
2905
2906         rc = ll_ioc_data_version(inode, &ioc);
2907         if (!rc)
2908                 *data_version = ioc.idv_version;
2909
2910         return rc;
2911 }
2912
2913 /*
2914  * Trigger a HSM release request for the provided inode.
2915  */
2916 int ll_hsm_release(struct inode *inode)
2917 {
2918         struct lu_env *env;
2919         struct obd_client_handle *och = NULL;
2920         __u64 data_version = 0;
2921         __u16 refcheck;
2922         int rc;
2923
2924         ENTRY;
2925
2926         CDEBUG(D_INODE, "%s: Releasing file "DFID".\n",
2927                ll_i2sbi(inode)->ll_fsname,
2928                PFID(&ll_i2info(inode)->lli_fid));
2929
2930         och = ll_lease_open(inode, NULL, FMODE_WRITE, MDS_OPEN_RELEASE);
2931         if (IS_ERR(och))
2932                 GOTO(out, rc = PTR_ERR(och));
2933
2934         /* Grab latest data_version and [am]time values */
2935         rc = ll_data_version(inode, &data_version,
2936                              LL_DV_WR_FLUSH | LL_DV_SZ_UPDATE);
2937         if (rc != 0)
2938                 GOTO(out, rc);
2939
2940         env = cl_env_get(&refcheck);
2941         if (IS_ERR(env))
2942                 GOTO(out, rc = PTR_ERR(env));
2943
2944         rc = ll_merge_attr(env, inode);
2945         cl_env_put(env, &refcheck);
2946
2947         /* If error happen, we have the wrong size for a file.
2948          * Don't release it.
2949          */
2950         if (rc != 0)
2951                 GOTO(out, rc);
2952
2953         /* Release the file.
2954          * NB: lease lock handle is released in mdc_hsm_release_pack() because
2955          * we still need it to pack l_remote_handle to MDT. */
2956         rc = ll_close_inode_openhandle(inode, och, MDS_HSM_RELEASE,
2957                                        &data_version);
2958         och = NULL;
2959
2960         EXIT;
2961 out:
2962         if (och != NULL && !IS_ERR(och)) /* close the file */
2963                 ll_lease_close(och, inode, NULL);
2964
2965         return rc;
2966 }
2967
2968 struct ll_swap_stack {
2969         __u64                    dv1;
2970         __u64                    dv2;
2971         struct inode            *inode1;
2972         struct inode            *inode2;
2973         bool                     check_dv1;
2974         bool                     check_dv2;
2975 };
2976
2977 static int ll_swap_layouts(struct file *file1, struct file *file2,
2978                            struct lustre_swap_layouts *lsl)
2979 {
2980         struct mdc_swap_layouts  msl;
2981         struct md_op_data       *op_data;
2982         __u32                    gid;
2983         __u64                    dv;
2984         struct ll_swap_stack    *llss = NULL;
2985         int                      rc;
2986
2987         OBD_ALLOC_PTR(llss);
2988         if (llss == NULL)
2989                 RETURN(-ENOMEM);
2990
2991         llss->inode1 = file_inode(file1);
2992         llss->inode2 = file_inode(file2);
2993
2994         rc = ll_check_swap_layouts_validity(llss->inode1, llss->inode2);
2995         if (rc < 0)
2996                 GOTO(free, rc);
2997
2998         /* we use 2 bool because it is easier to swap than 2 bits */
2999         if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV1)
3000                 llss->check_dv1 = true;
3001
3002         if (lsl->sl_flags & SWAP_LAYOUTS_CHECK_DV2)
3003                 llss->check_dv2 = true;
3004
3005         /* we cannot use lsl->sl_dvX directly because we may swap them */
3006         llss->dv1 = lsl->sl_dv1;
3007         llss->dv2 = lsl->sl_dv2;
3008
3009         rc = lu_fid_cmp(ll_inode2fid(llss->inode1), ll_inode2fid(llss->inode2));
3010         if (rc == 0) /* same file, done! */
3011                 GOTO(free, rc);
3012
3013         if (rc < 0) { /* sequentialize it */
3014                 swap(llss->inode1, llss->inode2);
3015                 swap(file1, file2);
3016                 swap(llss->dv1, llss->dv2);
3017                 swap(llss->check_dv1, llss->check_dv2);
3018         }
3019
3020         gid = lsl->sl_gid;
3021         if (gid != 0) { /* application asks to flush dirty cache */
3022                 rc = ll_get_grouplock(llss->inode1, file1, gid);
3023                 if (rc < 0)
3024                         GOTO(free, rc);
3025
3026                 rc = ll_get_grouplock(llss->inode2, file2, gid);
3027                 if (rc < 0) {
3028                         ll_put_grouplock(llss->inode1, file1, gid);
3029                         GOTO(free, rc);
3030                 }
3031         }
3032
3033         /* ultimate check, before swaping the layouts we check if
3034          * dataversion has changed (if requested) */
3035         if (llss->check_dv1) {
3036                 rc = ll_data_version(llss->inode1, &dv, 0);
3037                 if (rc)
3038                         GOTO(putgl, rc);
3039                 if (dv != llss->dv1)
3040                         GOTO(putgl, rc = -EAGAIN);
3041         }
3042
3043         if (llss->check_dv2) {
3044                 rc = ll_data_version(llss->inode2, &dv, 0);
3045                 if (rc)
3046                         GOTO(putgl, rc);
3047                 if (dv != llss->dv2)
3048                         GOTO(putgl, rc = -EAGAIN);
3049         }
3050
3051         /* struct md_op_data is used to send the swap args to the mdt
3052          * only flags is missing, so we use struct mdc_swap_layouts
3053          * through the md_op_data->op_data */
3054         /* flags from user space have to be converted before they are send to
3055          * server, no flag is sent today, they are only used on the client */
3056         msl.msl_flags = 0;
3057         rc = -ENOMEM;
3058         op_data = ll_prep_md_op_data(NULL, llss->inode1, llss->inode2, NULL, 0,
3059                                      0, LUSTRE_OPC_ANY, &msl);
3060         if (IS_ERR(op_data))
3061                 GOTO(free, rc = PTR_ERR(op_data));
3062
3063         rc = obd_iocontrol(LL_IOC_LOV_SWAP_LAYOUTS, ll_i2mdexp(llss->inode1),
3064                            sizeof(*op_data), op_data, NULL);
3065         ll_finish_md_op_data(op_data);
3066
3067         if (rc < 0)
3068                 GOTO(putgl, rc);
3069
3070 putgl:
3071         if (gid != 0) {
3072                 ll_put_grouplock(llss->inode2, file2, gid);
3073                 ll_put_grouplock(llss->inode1, file1, gid);
3074         }
3075
3076 free:
3077         if (llss != NULL)
3078                 OBD_FREE_PTR(llss);
3079
3080         RETURN(rc);
3081 }
3082
3083 int ll_hsm_state_set(struct inode *inode, struct hsm_state_set *hss)
3084 {
3085         struct obd_export *exp = ll_i2mdexp(inode);
3086         struct md_op_data *op_data;
3087         int rc;
3088         ENTRY;
3089
3090         /* Detect out-of range masks */
3091         if ((hss->hss_setmask | hss->hss_clearmask) & ~HSM_FLAGS_MASK)
3092                 RETURN(-EINVAL);
3093
3094         /* Non-root users are forbidden to set or clear flags which are
3095          * NOT defined in HSM_USER_MASK. */
3096         if (((hss->hss_setmask | hss->hss_clearmask) & ~HSM_USER_MASK) &&
3097             !capable(CAP_SYS_ADMIN))
3098                 RETURN(-EPERM);
3099
3100         if (!exp_connect_archive_id_array(exp)) {
3101                 /* Detect out-of range archive id */
3102                 if ((hss->hss_valid & HSS_ARCHIVE_ID) &&
3103                     (hss->hss_archive_id > LL_HSM_ORIGIN_MAX_ARCHIVE))
3104                         RETURN(-EINVAL);
3105         }
3106
3107         op_data = ll_prep_md_op_data(NULL, inode, NULL, NULL, 0, 0,
3108                                      LUSTRE_OPC_ANY, hss);
3109         if (IS_ERR(op_data))
3110                 RETURN(PTR_ERR(op_data));
3111
3112         rc = obd_iocontrol(LL_IOC_HSM_STATE_SET, exp, sizeof(*op_data),
3113                            op_data, NULL);
3114
3115         ll_finish_md_op_data(op_data);
3116
3117         RETURN(rc);
3118 }
3119
3120 static int ll_hsm_import(struct inode *inode, struct file *file,
3121                          struct hsm_user_import *hui)
3122 {
3123         struct hsm_state_set    *hss = NULL;
3124         struct iattr            *attr = NULL;
3125         int                      rc;
3126         ENTRY;
3127
3128         if (!S_ISREG(inode->i_mode))
3129                 RETURN(-EINVAL);
3130
3131         /* set HSM flags */
3132         OBD_ALLOC_PTR(hss);
3133         if (hss == NULL)
3134                 GOTO(out, rc = -ENOMEM);
3135
3136         hss->hss_valid = HSS_SETMASK | HSS_ARCHIVE_ID;
3137         hss->hss_archive_id = hui->hui_archive_id;
3138         hss->hss_setmask = HS_ARCHIVED | HS_EXISTS | HS_RELEASED;
3139         rc = ll_hsm_state_set(inode, hss);
3140         if (rc != 0)
3141                 GOTO(out, rc);
3142
3143         OBD_ALLOC_PTR(attr);
3144         if (attr == NULL)
3145                 GOTO(out, rc = -ENOMEM);
3146
3147         attr->ia_mode = hui->hui_mode & (S_IRWXU | S_IRWXG | S_IRWXO);
3148         attr->ia_mode |= S_IFREG;
3149         attr->ia_uid = make_kuid(&init_user_ns, hui->hui_uid);
3150         attr->ia_gid = make_kgid(&init_user_ns, hui->hui_gid);
3151         attr->ia_size = hui->hui_size;
3152         attr->ia_mtime.tv_sec = hui->hui_mtime;
3153         attr->ia_mtime.tv_nsec = hui->hui_mtime_ns;
3154         attr->ia_atime.tv_sec = hui->hui_atime;
3155         attr->ia_atime.tv_nsec = hui->hui_atime_ns;
3156
3157         attr->ia_valid = ATTR_SIZE | ATTR_MODE | ATTR_FORCE |
3158                          ATTR_UID | ATTR_GID |
3159                          ATTR_MTIME | ATTR_MTIME_SET |
3160                          ATTR_ATIME | ATTR_ATIME_SET;
3161
3162         inode_lock(inode);
3163
3164         rc = ll_setattr_raw(file_dentry(file), attr, 0, true);
3165         if (rc == -ENODATA)
3166                 rc = 0;
3167
3168         inode_unlock(inode);
3169
3170 out:
3171         if (hss != NULL)
3172                 OBD_FREE_PTR(hss);
3173
3174         if (attr != NULL)
3175                 OBD_FREE_PTR(attr);
3176
3177         RETURN(rc);
3178 }
3179
3180 static inline long ll_lease_type_from_fmode(fmode_t fmode)
3181 {
3182         return ((fmode & FMODE_READ) ? LL_LEASE_RDLCK : 0) |
3183                ((fmode & FMODE_WRITE) ? LL_LEASE_WRLCK : 0);
3184 }
3185
3186 static int ll_file_futimes_3(struct file *file, const struct ll_futimes_3 *lfu)
3187 {
3188         struct inode *inode = file_inode(file);
3189         struct iattr ia = {
3190                 .ia_valid = ATTR_ATIME | ATTR_ATIME_SET |
3191                             ATTR_MTIME | ATTR_MTIME_SET |
3192                             ATTR_CTIME,
3193                 .ia_atime = {
3194                         .tv_sec = lfu->lfu_atime_sec,
3195                         .tv_nsec = lfu->lfu_atime_nsec,
3196                 },
3197                 .ia_mtime = {
3198                         .tv_sec = lfu->lfu_mtime_sec,
3199                         .tv_nsec = lfu->lfu_mtime_nsec,
3200                 },
3201                 .ia_ctime = {
3202                         .tv_sec = lfu->lfu_ctime_sec,
3203                         .tv_nsec = lfu->lfu_ctime_nsec,
3204                 },
3205         };
3206         int rc;
3207         ENTRY;
3208
3209         if (!capable(CAP_SYS_ADMIN))
3210                 RETURN(-EPERM);
3211
3212         if (!S_ISREG(inode->i_mode))
3213                 RETURN(-EINVAL);
3214
3215         inode_lock(inode);
3216         rc = ll_setattr_raw(file_dentry(file), &ia, OP_XVALID_CTIME_SET,
3217                             false);
3218         inode_unlock(inode);
3219
3220         RETURN(rc);
3221 }
3222
3223 static enum cl_lock_mode cl_mode_user_to_kernel(enum lock_mode_user mode)
3224 {
3225         switch (mode) {
3226         case MODE_READ_USER:
3227                 return CLM_READ;
3228         case MODE_WRITE_USER:
3229                 return CLM_WRITE;
3230         default:
3231                 return -EINVAL;
3232         }
3233 }
3234
3235 static const char *const user_lockname[] = LOCK_MODE_NAMES;
3236
3237 /* Used to allow the upper layers of the client to request an LDLM lock
3238  * without doing an actual read or write.
3239  *
3240  * Used for ladvise lockahead to manually request specific locks.
3241  *
3242  * \param[in] file      file this ladvise lock request is on
3243  * \param[in] ladvise   ladvise struct describing this lock request
3244  *
3245  * \retval 0            success, no detailed result available (sync requests
3246  *                      and requests sent to the server [not handled locally]
3247  *                      cannot return detailed results)
3248  * \retval LLA_RESULT_{SAME,DIFFERENT} - detailed result of the lock request,
3249  *                                       see definitions for details.
3250  * \retval negative     negative errno on error
3251  */
3252 int ll_file_lock_ahead(struct file *file, struct llapi_lu_ladvise *ladvise)
3253 {
3254         struct lu_env *env = NULL;
3255         struct cl_io *io  = NULL;
3256         struct cl_lock *lock = NULL;
3257         struct cl_lock_descr *descr = NULL;
3258         struct dentry *dentry = file->f_path.dentry;
3259         struct inode *inode = dentry->d_inode;
3260         enum cl_lock_mode cl_mode;
3261         off_t start = ladvise->lla_start;
3262         off_t end = ladvise->lla_end;
3263         int result;
3264         __u16 refcheck;
3265
3266         ENTRY;
3267
3268         CDEBUG(D_VFSTRACE,
3269                "Lock request: file=%pd, inode=%p, mode=%s start=%llu, end=%llu\n",
3270                dentry, dentry->d_inode,
3271                user_lockname[ladvise->lla_lockahead_mode], (__u64) start,
3272                (__u64) end);
3273
3274         cl_mode = cl_mode_user_to_kernel(ladvise->lla_lockahead_mode);
3275         if (cl_mode < 0)
3276                 GOTO(out, result = cl_mode);
3277
3278         /* Get IO environment */
3279         result = cl_io_get(inode, &env, &io, &refcheck);
3280         if (result <= 0)
3281                 GOTO(out, result);
3282
3283         result = cl_io_init(env, io, CIT_MISC, io->ci_obj);
3284         if (result > 0) {
3285                 /*
3286                  * nothing to do for this io. This currently happens when
3287                  * stripe sub-object's are not yet created.
3288                  */
3289                 result = io->ci_result;
3290         } else if (result == 0) {
3291                 lock = vvp_env_lock(env);
3292                 descr = &lock->cll_descr;
3293
3294                 descr->cld_obj   = io->ci_obj;
3295                 /* Convert byte offsets to pages */
3296                 descr->cld_start = cl_index(io->ci_obj, start);
3297                 descr->cld_end   = cl_index(io->ci_obj, end);
3298                 descr->cld_mode  = cl_mode;
3299                 /* CEF_MUST is used because we do not want to convert a
3300                  * lockahead request to a lockless lock */
3301                 descr->cld_enq_flags = CEF_MUST | CEF_LOCK_NO_EXPAND;
3302
3303                 if (ladvise->lla_peradvice_flags & LF_ASYNC)
3304                         descr->cld_enq_flags |= CEF_SPECULATIVE;
3305
3306                 result = cl_lock_request(env, io, lock);
3307
3308                 /* On success, we need to release the lock */
3309                 if (result >= 0)
3310                         cl_lock_release(env, lock);
3311         }
3312         cl_io_fini(env, io);
3313         cl_env_put(env, &refcheck);
3314
3315         /* -ECANCELED indicates a matching lock with a different extent
3316          * was already present, and -EEXIST indicates a matching&nb