?  search:
summary | log | graphiclog1 | graphiclog2 | commit | commitdiff | tree | refs | edit | fork
blame | history | raw | HEAD
sh_eth: ensure pm_runtime cannot suspend the device during init
[linux/fpc-iii.git] / fs / ceph / file.c
blob09c7afe32e496c7dfe20d527106c7a184248c3ca
1 #include <linux/ceph/ceph_debug.h>
2
3 #include <linux/module.h>
4 #include <linux/sched.h>
5 #include <linux/slab.h>
6 #include <linux/file.h>
7 #include <linux/mount.h>
8 #include <linux/namei.h>
9 #include <linux/writeback.h>
10 #include <linux/aio.h>
11 #include <linux/falloc.h>
12
13 #include "super.h"
14 #include "mds_client.h"
15 #include "cache.h"
16
17 /*
18 * Ceph file operations
19 *
20 * Implement basic open/close functionality, and implement
21 * read/write.
22 *
23 * We implement three modes of file I/O:
24 * - buffered uses the generic_file_aio_{read,write} helpers
25 *
26 * - synchronous is used when there is multi-client read/write
27 * sharing, avoids the page cache, and synchronously waits for an
28 * ack from the OSD.
29 *
30 * - direct io takes the variant of the sync path that references
31 * user pages directly.
32 *
33 * fsync() flushes and waits on dirty pages, but just queues metadata
34 * for writeback: since the MDS can recover size and mtime there is no
35 * need to wait for MDS acknowledgement.
36 */
37
38
39 /*
40 * Prepare an open request. Preallocate ceph_cap to avoid an
41 * inopportune ENOMEM later.
42 */
43 static struct ceph_mds_request *
44 prepare_open_request(struct super_block *sb, int flags, int create_mode)
45 {
46 struct ceph_fs_client *fsc = ceph_sb_to_client(sb);
47 struct ceph_mds_client *mdsc = fsc->mdsc;
48 struct ceph_mds_request *req;
49 int want_auth = USE_ANY_MDS;
50 int op = (flags & O_CREAT) ? CEPH_MDS_OP_CREATE : CEPH_MDS_OP_OPEN;
51
52 if (flags & (O_WRONLY|O_RDWR|O_CREAT|O_TRUNC))
53 want_auth = USE_AUTH_MDS;
54
55 req = ceph_mdsc_create_request(mdsc, op, want_auth);
56 if (IS_ERR(req))
57 goto out;
58 req->r_fmode = ceph_flags_to_mode(flags);
59 req->r_args.open.flags = cpu_to_le32(flags);
60 req->r_args.open.mode = cpu_to_le32(create_mode);
61 out:
62 return req;
63 }
64
65 /*
66 * initialize private struct file data.
67 * if we fail, clean up by dropping fmode reference on the ceph_inode
68 */
69 static int ceph_init_file(struct inode *inode, struct file *file, int fmode)
70 {
71 struct ceph_file_info *cf;
72 int ret = 0;
73 struct ceph_inode_info *ci = ceph_inode(inode);
74 struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
75 struct ceph_mds_client *mdsc = fsc->mdsc;
76
77 switch (inode->i_mode & S_IFMT) {
78 case S_IFREG:
79 /* First file open request creates the cookie, we want to keep
80 * this cookie around for the filetime of the inode as not to
81 * have to worry about fscache register / revoke / operation
82 * races.
83 *
84 * Also, if we know the operation is going to invalidate data
85 * (non readonly) just nuke the cache right away.
86 */
87 ceph_fscache_register_inode_cookie(mdsc->fsc, ci);
88 if ((fmode & CEPH_FILE_MODE_WR))
89 ceph_fscache_invalidate(inode);
90 case S_IFDIR:
91 dout("init_file %p %p 0%o (regular)\n", inode, file,
92 inode->i_mode);
93 cf = kmem_cache_alloc(ceph_file_cachep, GFP_NOFS | __GFP_ZERO);
94 if (cf == NULL) {
95 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
96 return -ENOMEM;
97 }
98 cf->fmode = fmode;
99 cf->next_offset = 2;
100 file->private_data = cf;
101 BUG_ON(inode->i_fop->release != ceph_release);
102 break;
103
104 case S_IFLNK:
105 dout("init_file %p %p 0%o (symlink)\n", inode, file,
106 inode->i_mode);
107 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
108 break;
109
110 default:
111 dout("init_file %p %p 0%o (special)\n", inode, file,
112 inode->i_mode);
113 /*
114 * we need to drop the open ref now, since we don't
115 * have .release set to ceph_release.
116 */
117 ceph_put_fmode(ceph_inode(inode), fmode); /* clean up */
118 BUG_ON(inode->i_fop->release == ceph_release);
119
120 /* call the proper open fop */
121 ret = inode->i_fop->open(inode, file);
122 }
123 return ret;
124 }
125
126 /*
127 * If we already have the requisite capabilities, we can satisfy
128 * the open request locally (no need to request new caps from the
129 * MDS). We do, however, need to inform the MDS (asynchronously)
130 * if our wanted caps set expands.
131 */
132 int ceph_open(struct inode *inode, struct file *file)
133 {
134 struct ceph_inode_info *ci = ceph_inode(inode);
135 struct ceph_fs_client *fsc = ceph_sb_to_client(inode->i_sb);
136 struct ceph_mds_client *mdsc = fsc->mdsc;
137 struct ceph_mds_request *req;
138 struct ceph_file_info *cf = file->private_data;
139 struct inode *parent_inode = NULL;
140 int err;
141 int flags, fmode, wanted;
142
143 if (cf) {
144 dout("open file %p is already opened\n", file);
145 return 0;
146 }
147
148 /* filter out O_CREAT|O_EXCL; vfs did that already. yuck. */
149 flags = file->f_flags & ~(O_CREAT|O_EXCL);
150 if (S_ISDIR(inode->i_mode))
151 flags = O_DIRECTORY; /* mds likes to know */
152
153 dout("open inode %p ino %llx.%llx file %p flags %d (%d)\n", inode,
154 ceph_vinop(inode), file, flags, file->f_flags);
155 fmode = ceph_flags_to_mode(flags);
156 wanted = ceph_caps_for_mode(fmode);
157
158 /* snapped files are read-only */
159 if (ceph_snap(inode) != CEPH_NOSNAP && (file->f_mode & FMODE_WRITE))
160 return -EROFS;
161
162 /* trivially open snapdir */
163 if (ceph_snap(inode) == CEPH_SNAPDIR) {
164 spin_lock(&ci->i_ceph_lock);
165 __ceph_get_fmode(ci, fmode);
166 spin_unlock(&ci->i_ceph_lock);
167 return ceph_init_file(inode, file, fmode);
168 }
169
170 /*
171 * No need to block if we have caps on the auth MDS (for
172 * write) or any MDS (for read). Update wanted set
173 * asynchronously.
174 */
175 spin_lock(&ci->i_ceph_lock);
176 if (__ceph_is_any_real_caps(ci) &&
177 (((fmode & CEPH_FILE_MODE_WR) == 0) || ci->i_auth_cap)) {
178 int mds_wanted = __ceph_caps_mds_wanted(ci);
179 int issued = __ceph_caps_issued(ci, NULL);
180
181 dout("open %p fmode %d want %s issued %s using existing\n",
182 inode, fmode, ceph_cap_string(wanted),
183 ceph_cap_string(issued));
184 __ceph_get_fmode(ci, fmode);
185 spin_unlock(&ci->i_ceph_lock);
186
187 /* adjust wanted? */
188 if ((issued & wanted) != wanted &&
189 (mds_wanted & wanted) != wanted &&
190 ceph_snap(inode) != CEPH_SNAPDIR)
191 ceph_check_caps(ci, 0, NULL);
192
193 return ceph_init_file(inode, file, fmode);
194 } else if (ceph_snap(inode) != CEPH_NOSNAP &&
195 (ci->i_snap_caps & wanted) == wanted) {
196 __ceph_get_fmode(ci, fmode);
197 spin_unlock(&ci->i_ceph_lock);
198 return ceph_init_file(inode, file, fmode);
199 }
200
201 spin_unlock(&ci->i_ceph_lock);
202
203 dout("open fmode %d wants %s\n", fmode, ceph_cap_string(wanted));
204 req = prepare_open_request(inode->i_sb, flags, 0);
205 if (IS_ERR(req)) {
206 err = PTR_ERR(req);
207 goto out;
208 }
209 req->r_inode = inode;
210 ihold(inode);
211
212 req->r_num_caps = 1;
213 if (flags & (O_CREAT|O_TRUNC))
214 parent_inode = ceph_get_dentry_parent_inode(file->f_dentry);
215 err = ceph_mdsc_do_request(mdsc, parent_inode, req);
216 iput(parent_inode);
217 if (!err)
218 err = ceph_init_file(inode, file, req->r_fmode);
219 ceph_mdsc_put_request(req);
220 dout("open result=%d on %llx.%llx\n", err, ceph_vinop(inode));
221 out:
222 return err;
223 }
224
225
226 /*
227 * Do a lookup + open with a single request. If we get a non-existent
228 * file or symlink, return 1 so the VFS can retry.
229 */
230 int ceph_atomic_open(struct inode *dir, struct dentry *dentry,
231 struct file *file, unsigned flags, umode_t mode,
232 int *opened)
233 {
234 struct ceph_fs_client *fsc = ceph_sb_to_client(dir->i_sb);
235 struct ceph_mds_client *mdsc = fsc->mdsc;
236 struct ceph_mds_request *req;
237 struct dentry *dn;
238 int err;
239
240 dout("atomic_open %p dentry %p '%.*s' %s flags %d mode 0%o\n",
241 dir, dentry, dentry->d_name.len, dentry->d_name.name,
242 d_unhashed(dentry) ? "unhashed" : "hashed", flags, mode);
243
244 if (dentry->d_name.len > NAME_MAX)
245 return -ENAMETOOLONG;
246
247 err = ceph_init_dentry(dentry);
248 if (err < 0)
249 return err;
250
251 /* do the open */
252 req = prepare_open_request(dir->i_sb, flags, mode);
253 if (IS_ERR(req))
254 return PTR_ERR(req);
255 req->r_dentry = dget(dentry);
256 req->r_num_caps = 2;
257 if (flags & O_CREAT) {
258 req->r_dentry_drop = CEPH_CAP_FILE_SHARED;
259 req->r_dentry_unless = CEPH_CAP_FILE_EXCL;
260 }
261 req->r_locked_dir = dir; /* caller holds dir->i_mutex */
262 err = ceph_mdsc_do_request(mdsc,
263 (flags & (O_CREAT|O_TRUNC)) ? dir : NULL,
264 req);
265 if (err)
266 goto out_err;
267
268 err = ceph_handle_snapdir(req, dentry, err);
269 if (err == 0 && (flags & O_CREAT) && !req->r_reply_info.head->is_dentry)
270 err = ceph_handle_notrace_create(dir, dentry);
271
272 if (d_unhashed(dentry)) {
273 dn = ceph_finish_lookup(req, dentry, err);
274 if (IS_ERR(dn))
275 err = PTR_ERR(dn);
276 } else {
277 /* we were given a hashed negative dentry */
278 dn = NULL;
279 }
280 if (err)
281 goto out_err;
282 if (dn || dentry->d_inode == NULL || S_ISLNK(dentry->d_inode->i_mode)) {
283 /* make vfs retry on splice, ENOENT, or symlink */
284 dout("atomic_open finish_no_open on dn %p\n", dn);
285 err = finish_no_open(file, dn);
286 } else {
287 dout("atomic_open finish_open on dn %p\n", dn);
288 if (req->r_op == CEPH_MDS_OP_CREATE && req->r_reply_info.has_create_ino) {
289 ceph_init_acl(dentry, dentry->d_inode, dir);
290 *opened |= FILE_CREATED;
291 }
292 err = finish_open(file, dentry, ceph_open, opened);
293 }
294
295 out_err:
296 ceph_mdsc_put_request(req);
297 dout("atomic_open result=%d\n", err);
298 return err;
299 }
300
301 int ceph_release(struct inode *inode, struct file *file)
302 {
303 struct ceph_inode_info *ci = ceph_inode(inode);
304 struct ceph_file_info *cf = file->private_data;
305
306 dout("release inode %p file %p\n", inode, file);
307 ceph_put_fmode(ci, cf->fmode);
308 if (cf->last_readdir)
309 ceph_mdsc_put_request(cf->last_readdir);
310 kfree(cf->last_name);
311 kfree(cf->dir_info);
312 dput(cf->dentry);
313 kmem_cache_free(ceph_file_cachep, cf);
314
315 /* wake up anyone waiting for caps on this inode */
316 wake_up_all(&ci->i_cap_wq);
317 return 0;
318 }
319
320 /*
321 * Read a range of bytes striped over one or more objects. Iterate over
322 * objects we stripe over. (That's not atomic, but good enough for now.)
323 *
324 * If we get a short result from the OSD, check against i_size; we need to
325 * only return a short read to the caller if we hit EOF.
326 */
327 static int striped_read(struct inode *inode,
328 u64 off, u64 len,
329 struct page **pages, int num_pages,
330 int *checkeof, bool o_direct,
331 unsigned long buf_align)
332 {
333 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
334 struct ceph_inode_info *ci = ceph_inode(inode);
335 u64 pos, this_len, left;
336 int io_align, page_align;
337 int pages_left;
338 int read;
339 struct page **page_pos;
340 int ret;
341 bool hit_stripe, was_short;
342
343 /*
344 * we may need to do multiple reads. not atomic, unfortunately.
345 */
346 pos = off;
347 left = len;
348 page_pos = pages;
349 pages_left = num_pages;
350 read = 0;
351 io_align = off & ~PAGE_MASK;
352
353 more:
354 if (o_direct)
355 page_align = (pos - io_align + buf_align) & ~PAGE_MASK;
356 else
357 page_align = pos & ~PAGE_MASK;
358 this_len = left;
359 ret = ceph_osdc_readpages(&fsc->client->osdc, ceph_vino(inode),
360 &ci->i_layout, pos, &this_len,
361 ci->i_truncate_seq,
362 ci->i_truncate_size,
363 page_pos, pages_left, page_align);
364 if (ret == -ENOENT)
365 ret = 0;
366 hit_stripe = this_len < left;
367 was_short = ret >= 0 && ret < this_len;
368 dout("striped_read %llu~%llu (read %u) got %d%s%s\n", pos, left, read,
369 ret, hit_stripe ? " HITSTRIPE" : "", was_short ? " SHORT" : "");
370
371 if (ret >= 0) {
372 int didpages;
373 if (was_short && (pos + ret < inode->i_size)) {
374 u64 tmp = min(this_len - ret,
375 inode->i_size - pos - ret);
376 dout(" zero gap %llu to %llu\n",
377 pos + ret, pos + ret + tmp);
378 ceph_zero_page_vector_range(page_align + read + ret,
379 tmp, pages);
380 ret += tmp;
381 }
382
383 didpages = (page_align + ret) >> PAGE_CACHE_SHIFT;
384 pos += ret;
385 read = pos - off;
386 left -= ret;
387 page_pos += didpages;
388 pages_left -= didpages;
389
390 /* hit stripe and need continue*/
391 if (left && hit_stripe && pos < inode->i_size)
392 goto more;
393 }
394
395 if (read > 0) {
396 ret = read;
397 /* did we bounce off eof? */
398 if (pos + left > inode->i_size)
399 *checkeof = 1;
400 }
401
402 dout("striped_read returns %d\n", ret);
403 return ret;
404 }
405
406 /*
407 * Completely synchronous read and write methods. Direct from __user
408 * buffer to osd, or directly to user pages (if O_DIRECT).
409 *
410 * If the read spans object boundary, just do multiple reads.
411 */
412 static ssize_t ceph_sync_read(struct kiocb *iocb, struct iov_iter *i,
413 int *checkeof)
414 {
415 struct file *file = iocb->ki_filp;
416 struct inode *inode = file_inode(file);
417 struct page **pages;
418 u64 off = iocb->ki_pos;
419 int num_pages, ret;
420 size_t len = i->count;
421
422 dout("sync_read on file %p %llu~%u %s\n", file, off,
423 (unsigned)len,
424 (file->f_flags & O_DIRECT) ? "O_DIRECT" : "");
425 /*
426 * flush any page cache pages in this range. this
427 * will make concurrent normal and sync io slow,
428 * but it will at least behave sensibly when they are
429 * in sequence.
430 */
431 ret = filemap_write_and_wait_range(inode->i_mapping, off,
432 off + len);
433 if (ret < 0)
434 return ret;
435
436 if (file->f_flags & O_DIRECT) {
437 while (iov_iter_count(i)) {
438 void __user *data = i->iov[0].iov_base + i->iov_offset;
439 size_t len = i->iov[0].iov_len - i->iov_offset;
440
441 num_pages = calc_pages_for((unsigned long)data, len);
442 pages = ceph_get_direct_page_vector(data,
443 num_pages, true);
444 if (IS_ERR(pages))
445 return PTR_ERR(pages);
446
447 ret = striped_read(inode, off, len,
448 pages, num_pages, checkeof,
449 1, (unsigned long)data & ~PAGE_MASK);
450 ceph_put_page_vector(pages, num_pages, true);
451
452 if (ret <= 0)
453 break;
454 off += ret;
455 iov_iter_advance(i, ret);
456 if (ret < len)
457 break;
458 }
459 } else {
460 num_pages = calc_pages_for(off, len);
461 pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
462 if (IS_ERR(pages))
463 return PTR_ERR(pages);
464 ret = striped_read(inode, off, len, pages,
465 num_pages, checkeof, 0, 0);
466 if (ret > 0) {
467 int l, k = 0;
468 size_t left = len = ret;
469
470 while (left) {
471 void __user *data = i->iov[0].iov_base
472 + i->iov_offset;
473 l = min(i->iov[0].iov_len - i->iov_offset,
474 left);
475
476 ret = ceph_copy_page_vector_to_user(&pages[k],
477 data, off,
478 l);
479 if (ret > 0) {
480 iov_iter_advance(i, ret);
481 left -= ret;
482 off += ret;
483 k = calc_pages_for(iocb->ki_pos,
484 len - left + 1) - 1;
485 BUG_ON(k >= num_pages && left);
486 } else
487 break;
488 }
489 }
490 ceph_release_page_vector(pages, num_pages);
491 }
492
493 if (off > iocb->ki_pos) {
494 ret = off - iocb->ki_pos;
495 iocb->ki_pos = off;
496 }
497
498 dout("sync_read result %d\n", ret);
499 return ret;
500 }
501
502 /*
503 * Write commit request unsafe callback, called to tell us when a
504 * request is unsafe (that is, in flight--has been handed to the
505 * messenger to send to its target osd). It is called again when
506 * we've received a response message indicating the request is
507 * "safe" (its CEPH_OSD_FLAG_ONDISK flag is set), or when a request
508 * is completed early (and unsuccessfully) due to a timeout or
509 * interrupt.
510 *
511 * This is used if we requested both an ACK and ONDISK commit reply
512 * from the OSD.
513 */
514 static void ceph_sync_write_unsafe(struct ceph_osd_request *req, bool unsafe)
515 {
516 struct ceph_inode_info *ci = ceph_inode(req->r_inode);
517
518 dout("%s %p tid %llu %ssafe\n", __func__, req, req->r_tid,
519 unsafe ? "un" : "");
520 if (unsafe) {
521 ceph_get_cap_refs(ci, CEPH_CAP_FILE_WR);
522 spin_lock(&ci->i_unsafe_lock);
523 list_add_tail(&req->r_unsafe_item,
524 &ci->i_unsafe_writes);
525 spin_unlock(&ci->i_unsafe_lock);
526 } else {
527 spin_lock(&ci->i_unsafe_lock);
528 list_del_init(&req->r_unsafe_item);
529 spin_unlock(&ci->i_unsafe_lock);
530 ceph_put_cap_refs(ci, CEPH_CAP_FILE_WR);
531 }
532 }
533
534
535 /*
536 * Synchronous write, straight from __user pointer or user pages.
537 *
538 * If write spans object boundary, just do multiple writes. (For a
539 * correct atomic write, we should e.g. take write locks on all
540 * objects, rollback on failure, etc.)
541 */
542 static ssize_t
543 ceph_sync_direct_write(struct kiocb *iocb, const struct iovec *iov,
544 unsigned long nr_segs, size_t count)
545 {
546 struct file *file = iocb->ki_filp;
547 struct inode *inode = file_inode(file);
548 struct ceph_inode_info *ci = ceph_inode(inode);
549 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
550 struct ceph_snap_context *snapc;
551 struct ceph_vino vino;
552 struct ceph_osd_request *req;
553 struct page **pages;
554 int num_pages;
555 int written = 0;
556 int flags;
557 int check_caps = 0;
558 int page_align;
559 int ret;
560 struct timespec mtime = CURRENT_TIME;
561 loff_t pos = iocb->ki_pos;
562 struct iov_iter i;
563
564 if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
565 return -EROFS;
566
567 dout("sync_direct_write on file %p %lld~%u\n", file, pos,
568 (unsigned)count);
569
570 ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
571 if (ret < 0)
572 return ret;
573
574 ret = invalidate_inode_pages2_range(inode->i_mapping,
575 pos >> PAGE_CACHE_SHIFT,
576 (pos + count) >> PAGE_CACHE_SHIFT);
577 if (ret < 0)
578 dout("invalidate_inode_pages2_range returned %d\n", ret);
579
580 flags = CEPH_OSD_FLAG_ORDERSNAP |
581 CEPH_OSD_FLAG_ONDISK |
582 CEPH_OSD_FLAG_WRITE;
583
584 iov_iter_init(&i, iov, nr_segs, count, 0);
585
586 while (iov_iter_count(&i) > 0) {
587 void __user *data = i.iov->iov_base + i.iov_offset;
588 u64 len = i.iov->iov_len - i.iov_offset;
589
590 page_align = (unsigned long)data & ~PAGE_MASK;
591
592 snapc = ci->i_snap_realm->cached_context;
593 vino = ceph_vino(inode);
594 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
595 vino, pos, &len,
596 2,/*include a 'startsync' command*/
597 CEPH_OSD_OP_WRITE, flags, snapc,
598 ci->i_truncate_seq,
599 ci->i_truncate_size,
600 false);
601 if (IS_ERR(req)) {
602 ret = PTR_ERR(req);
603 goto out;
604 }
605
606 num_pages = calc_pages_for(page_align, len);
607 pages = ceph_get_direct_page_vector(data, num_pages, false);
608 if (IS_ERR(pages)) {
609 ret = PTR_ERR(pages);
610 goto out;
611 }
612
613 /*
614 * throw out any page cache pages in this range. this
615 * may block.
616 */
617 truncate_inode_pages_range(inode->i_mapping, pos,
618 (pos+len) | (PAGE_CACHE_SIZE-1));
619 osd_req_op_extent_osd_data_pages(req, 0, pages, len, page_align,
620 false, false);
621
622 /* BUG_ON(vino.snap != CEPH_NOSNAP); */
623 ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
624
625 ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
626 if (!ret)
627 ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
628
629 ceph_put_page_vector(pages, num_pages, false);
630
631 out:
632 ceph_osdc_put_request(req);
633 if (ret == 0) {
634 pos += len;
635 written += len;
636 iov_iter_advance(&i, (size_t)len);
637
638 if (pos > i_size_read(inode)) {
639 check_caps = ceph_inode_set_size(inode, pos);
640 if (check_caps)
641 ceph_check_caps(ceph_inode(inode),
642 CHECK_CAPS_AUTHONLY,
643 NULL);
644 }
645 } else
646 break;
647 }
648
649 if (ret != -EOLDSNAPC && written > 0) {
650 iocb->ki_pos = pos;
651 ret = written;
652 }
653 return ret;
654 }
655
656
657 /*
658 * Synchronous write, straight from __user pointer or user pages.
659 *
660 * If write spans object boundary, just do multiple writes. (For a
661 * correct atomic write, we should e.g. take write locks on all
662 * objects, rollback on failure, etc.)
663 */
664 static ssize_t ceph_sync_write(struct kiocb *iocb, const struct iovec *iov,
665 unsigned long nr_segs, size_t count)
666 {
667 struct file *file = iocb->ki_filp;
668 struct inode *inode = file_inode(file);
669 struct ceph_inode_info *ci = ceph_inode(inode);
670 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
671 struct ceph_snap_context *snapc;
672 struct ceph_vino vino;
673 struct ceph_osd_request *req;
674 struct page **pages;
675 u64 len;
676 int num_pages;
677 int written = 0;
678 int flags;
679 int check_caps = 0;
680 int ret;
681 struct timespec mtime = CURRENT_TIME;
682 loff_t pos = iocb->ki_pos;
683 struct iov_iter i;
684
685 if (ceph_snap(file_inode(file)) != CEPH_NOSNAP)
686 return -EROFS;
687
688 dout("sync_write on file %p %lld~%u\n", file, pos, (unsigned)count);
689
690 ret = filemap_write_and_wait_range(inode->i_mapping, pos, pos + count);
691 if (ret < 0)
692 return ret;
693
694 ret = invalidate_inode_pages2_range(inode->i_mapping,
695 pos >> PAGE_CACHE_SHIFT,
696 (pos + count) >> PAGE_CACHE_SHIFT);
697 if (ret < 0)
698 dout("invalidate_inode_pages2_range returned %d\n", ret);
699
700 flags = CEPH_OSD_FLAG_ORDERSNAP |
701 CEPH_OSD_FLAG_ONDISK |
702 CEPH_OSD_FLAG_WRITE |
703 CEPH_OSD_FLAG_ACK;
704
705 iov_iter_init(&i, iov, nr_segs, count, 0);
706
707 while ((len = iov_iter_count(&i)) > 0) {
708 size_t left;
709 int n;
710
711 snapc = ci->i_snap_realm->cached_context;
712 vino = ceph_vino(inode);
713 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
714 vino, pos, &len, 1,
715 CEPH_OSD_OP_WRITE, flags, snapc,
716 ci->i_truncate_seq,
717 ci->i_truncate_size,
718 false);
719 if (IS_ERR(req)) {
720 ret = PTR_ERR(req);
721 goto out;
722 }
723
724 /*
725 * write from beginning of first page,
726 * regardless of io alignment
727 */
728 num_pages = (len + PAGE_CACHE_SIZE - 1) >> PAGE_CACHE_SHIFT;
729
730 pages = ceph_alloc_page_vector(num_pages, GFP_NOFS);
731 if (IS_ERR(pages)) {
732 ret = PTR_ERR(pages);
733 goto out;
734 }
735
736 left = len;
737 for (n = 0; n < num_pages; n++) {
738 size_t plen = min_t(size_t, left, PAGE_SIZE);
739 ret = iov_iter_copy_from_user(pages[n], &i, 0, plen);
740 if (ret != plen) {
741 ret = -EFAULT;
742 break;
743 }
744 left -= ret;
745 iov_iter_advance(&i, ret);
746 }
747
748 if (ret < 0) {
749 ceph_release_page_vector(pages, num_pages);
750 goto out;
751 }
752
753 /* get a second commit callback */
754 req->r_unsafe_callback = ceph_sync_write_unsafe;
755 req->r_inode = inode;
756
757 osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
758 false, true);
759
760 /* BUG_ON(vino.snap != CEPH_NOSNAP); */
761 ceph_osdc_build_request(req, pos, snapc, vino.snap, &mtime);
762
763 ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
764 if (!ret)
765 ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
766
767 out:
768 ceph_osdc_put_request(req);
769 if (ret == 0) {
770 pos += len;
771 written += len;
772
773 if (pos > i_size_read(inode)) {
774 check_caps = ceph_inode_set_size(inode, pos);
775 if (check_caps)
776 ceph_check_caps(ceph_inode(inode),
777 CHECK_CAPS_AUTHONLY,
778 NULL);
779 }
780 } else
781 break;
782 }
783
784 if (ret != -EOLDSNAPC && written > 0) {
785 ret = written;
786 iocb->ki_pos = pos;
787 }
788 return ret;
789 }
790
791 /*
792 * Wrap generic_file_aio_read with checks for cap bits on the inode.
793 * Atomically grab references, so that those bits are not released
794 * back to the MDS mid-read.
795 *
796 * Hmm, the sync read case isn't actually async... should it be?
797 */
798 static ssize_t ceph_aio_read(struct kiocb *iocb, const struct iovec *iov,
799 unsigned long nr_segs, loff_t pos)
800 {
801 struct file *filp = iocb->ki_filp;
802 struct ceph_file_info *fi = filp->private_data;
803 size_t len = iocb->ki_nbytes;
804 struct inode *inode = file_inode(filp);
805 struct ceph_inode_info *ci = ceph_inode(inode);
806 ssize_t ret;
807 int want, got = 0;
808 int checkeof = 0, read = 0;
809
810 again:
811 dout("aio_read %p %llx.%llx %llu~%u trying to get caps on %p\n",
812 inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len, inode);
813
814 if (fi->fmode & CEPH_FILE_MODE_LAZY)
815 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
816 else
817 want = CEPH_CAP_FILE_CACHE;
818 ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, &got, -1);
819 if (ret < 0)
820 return ret;
821
822 if ((got & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0 ||
823 (iocb->ki_filp->f_flags & O_DIRECT) ||
824 (fi->flags & CEPH_F_SYNC)) {
825 struct iov_iter i;
826
827 dout("aio_sync_read %p %llx.%llx %llu~%u got cap refs on %s\n",
828 inode, ceph_vinop(inode), iocb->ki_pos, (unsigned)len,
829 ceph_cap_string(got));
830
831 if (!read) {
832 ret = generic_segment_checks(iov, &nr_segs,
833 &len, VERIFY_WRITE);
834 if (ret)
835 goto out;
836 }
837
838 iov_iter_init(&i, iov, nr_segs, len, read);
839
840 /* hmm, this isn't really async... */
841 ret = ceph_sync_read(iocb, &i, &checkeof);
842 } else {
843 /*
844 * We can't modify the content of iov,
845 * so we only read from beginning.
846 */
847 if (read) {
848 iocb->ki_pos = pos;
849 len = iocb->ki_nbytes;
850 read = 0;
851 }
852 dout("aio_read %p %llx.%llx %llu~%u got cap refs on %s\n",
853 inode, ceph_vinop(inode), pos, (unsigned)len,
854 ceph_cap_string(got));
855
856 ret = generic_file_aio_read(iocb, iov, nr_segs, pos);
857 }
858 out:
859 dout("aio_read %p %llx.%llx dropping cap refs on %s = %d\n",
860 inode, ceph_vinop(inode), ceph_cap_string(got), (int)ret);
861 ceph_put_cap_refs(ci, got);
862
863 if (checkeof && ret >= 0) {
864 int statret = ceph_do_getattr(inode,
865 CEPH_STAT_CAP_SIZE);
866
867 /* hit EOF or hole? */
868 if (statret == 0 && iocb->ki_pos < inode->i_size &&
869 ret < len) {
870 dout("sync_read hit hole, ppos %lld < size %lld"
871 ", reading more\n", iocb->ki_pos,
872 inode->i_size);
873
874 read += ret;
875 len -= ret;
876 checkeof = 0;
877 goto again;
878 }
879 }
880
881 if (ret >= 0)
882 ret += read;
883
884 return ret;
885 }
886
887 /*
888 * Take cap references to avoid releasing caps to MDS mid-write.
889 *
890 * If we are synchronous, and write with an old snap context, the OSD
891 * may return EOLDSNAPC. In that case, retry the write.. _after_
892 * dropping our cap refs and allowing the pending snap to logically
893 * complete _before_ this write occurs.
894 *
895 * If we are near ENOSPC, write synchronously.
896 */
897 static ssize_t ceph_aio_write(struct kiocb *iocb, const struct iovec *iov,
898 unsigned long nr_segs, loff_t pos)
899 {
900 struct file *file = iocb->ki_filp;
901 struct ceph_file_info *fi = file->private_data;
902 struct inode *inode = file_inode(file);
903 struct ceph_inode_info *ci = ceph_inode(inode);
904 struct ceph_osd_client *osdc =
905 &ceph_sb_to_client(inode->i_sb)->client->osdc;
906 ssize_t count, written = 0;
907 int err, want, got;
908
909 if (ceph_snap(inode) != CEPH_NOSNAP)
910 return -EROFS;
911
912 mutex_lock(&inode->i_mutex);
913
914 err = generic_segment_checks(iov, &nr_segs, &count, VERIFY_READ);
915 if (err)
916 goto out;
917
918 /* We can write back this queue in page reclaim */
919 current->backing_dev_info = file->f_mapping->backing_dev_info;
920
921 err = generic_write_checks(file, &pos, &count, S_ISBLK(inode->i_mode));
922 if (err)
923 goto out;
924
925 if (count == 0)
926 goto out;
927
928 err = file_remove_suid(file);
929 if (err)
930 goto out;
931
932 err = file_update_time(file);
933 if (err)
934 goto out;
935
936 retry_snap:
937 if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL)) {
938 err = -ENOSPC;
939 goto out;
940 }
941
942 dout("aio_write %p %llx.%llx %llu~%zd getting caps. i_size %llu\n",
943 inode, ceph_vinop(inode), pos, count, inode->i_size);
944 if (fi->fmode & CEPH_FILE_MODE_LAZY)
945 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
946 else
947 want = CEPH_CAP_FILE_BUFFER;
948 got = 0;
949 err = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, &got, pos + count);
950 if (err < 0)
951 goto out;
952
953 dout("aio_write %p %llx.%llx %llu~%zd got cap refs on %s\n",
954 inode, ceph_vinop(inode), pos, count, ceph_cap_string(got));
955
956 if ((got & (CEPH_CAP_FILE_BUFFER|CEPH_CAP_FILE_LAZYIO)) == 0 ||
957 (file->f_flags & O_DIRECT) || (fi->flags & CEPH_F_SYNC)) {
958 mutex_unlock(&inode->i_mutex);
959 if (file->f_flags & O_DIRECT)
960 written = ceph_sync_direct_write(iocb, iov,
961 nr_segs, count);
962 else
963 written = ceph_sync_write(iocb, iov, nr_segs, count);
964 if (written == -EOLDSNAPC) {
965 dout("aio_write %p %llx.%llx %llu~%u"
966 "got EOLDSNAPC, retrying\n",
967 inode, ceph_vinop(inode),
968 pos, (unsigned)iov->iov_len);
969 mutex_lock(&inode->i_mutex);
970 goto retry_snap;
971 }
972 } else {
973 /*
974 * No need to acquire the i_truncate_mutex. Because
975 * the MDS revokes Fwb caps before sending truncate
976 * message to us. We can't get Fwb cap while there
977 * are pending vmtruncate. So write and vmtruncate
978 * can not run at the same time
979 */
980 written = generic_file_buffered_write(iocb, iov, nr_segs,
981 pos, &iocb->ki_pos,
982 count, 0);
983 mutex_unlock(&inode->i_mutex);
984 }
985
986 if (written >= 0) {
987 int dirty;
988 spin_lock(&ci->i_ceph_lock);
989 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR);
990 spin_unlock(&ci->i_ceph_lock);
991 if (dirty)
992 __mark_inode_dirty(inode, dirty);
993 }
994
995 dout("aio_write %p %llx.%llx %llu~%u dropping cap refs on %s\n",
996 inode, ceph_vinop(inode), pos, (unsigned)iov->iov_len,
997 ceph_cap_string(got));
998 ceph_put_cap_refs(ci, got);
999
1000 if (written >= 0 &&
1001 ((file->f_flags & O_SYNC) || IS_SYNC(file->f_mapping->host) ||
1002 ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_NEARFULL))) {
1003 err = vfs_fsync_range(file, pos, pos + written - 1, 1);
1004 if (err < 0)
1005 written = err;
1006 }
1007
1008 goto out_unlocked;
1009
1010 out:
1011 mutex_unlock(&inode->i_mutex);
1012 out_unlocked:
1013 current->backing_dev_info = NULL;
1014 return written ? written : err;
1015 }
1016
1017 /*
1018 * llseek. be sure to verify file size on SEEK_END.
1019 */
1020 static loff_t ceph_llseek(struct file *file, loff_t offset, int whence)
1021 {
1022 struct inode *inode = file->f_mapping->host;
1023 int ret;
1024
1025 mutex_lock(&inode->i_mutex);
1026
1027 if (whence == SEEK_END || whence == SEEK_DATA || whence == SEEK_HOLE) {
1028 ret = ceph_do_getattr(inode, CEPH_STAT_CAP_SIZE);
1029 if (ret < 0) {
1030 offset = ret;
1031 goto out;
1032 }
1033 }
1034
1035 switch (whence) {
1036 case SEEK_END:
1037 offset += inode->i_size;
1038 break;
1039 case SEEK_CUR:
1040 /*
1041 * Here we special-case the lseek(fd, 0, SEEK_CUR)
1042 * position-querying operation. Avoid rewriting the "same"
1043 * f_pos value back to the file because a concurrent read(),
1044 * write() or lseek() might have altered it
1045 */
1046 if (offset == 0) {
1047 offset = file->f_pos;
1048 goto out;
1049 }
1050 offset += file->f_pos;
1051 break;
1052 case SEEK_DATA:
1053 if (offset >= inode->i_size) {
1054 ret = -ENXIO;
1055 goto out;
1056 }
1057 break;
1058 case SEEK_HOLE:
1059 if (offset >= inode->i_size) {
1060 ret = -ENXIO;
1061 goto out;
1062 }
1063 offset = inode->i_size;
1064 break;
1065 }
1066
1067 offset = vfs_setpos(file, offset, inode->i_sb->s_maxbytes);
1068
1069 out:
1070 mutex_unlock(&inode->i_mutex);
1071 return offset;
1072 }
1073
1074 static inline void ceph_zero_partial_page(
1075 struct inode *inode, loff_t offset, unsigned size)
1076 {
1077 struct page *page;
1078 pgoff_t index = offset >> PAGE_CACHE_SHIFT;
1079
1080 page = find_lock_page(inode->i_mapping, index);
1081 if (page) {
1082 wait_on_page_writeback(page);
1083 zero_user(page, offset & (PAGE_CACHE_SIZE - 1), size);
1084 unlock_page(page);
1085 page_cache_release(page);
1086 }
1087 }
1088
1089 static void ceph_zero_pagecache_range(struct inode *inode, loff_t offset,
1090 loff_t length)
1091 {
1092 loff_t nearly = round_up(offset, PAGE_CACHE_SIZE);
1093 if (offset < nearly) {
1094 loff_t size = nearly - offset;
1095 if (length < size)
1096 size = length;
1097 ceph_zero_partial_page(inode, offset, size);
1098 offset += size;
1099 length -= size;
1100 }
1101 if (length >= PAGE_CACHE_SIZE) {
1102 loff_t size = round_down(length, PAGE_CACHE_SIZE);
1103 truncate_pagecache_range(inode, offset, offset + size - 1);
1104 offset += size;
1105 length -= size;
1106 }
1107 if (length)
1108 ceph_zero_partial_page(inode, offset, length);
1109 }
1110
1111 static int ceph_zero_partial_object(struct inode *inode,
1112 loff_t offset, loff_t *length)
1113 {
1114 struct ceph_inode_info *ci = ceph_inode(inode);
1115 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1116 struct ceph_osd_request *req;
1117 int ret = 0;
1118 loff_t zero = 0;
1119 int op;
1120
1121 if (!length) {
1122 op = offset ? CEPH_OSD_OP_DELETE : CEPH_OSD_OP_TRUNCATE;
1123 length = &zero;
1124 } else {
1125 op = CEPH_OSD_OP_ZERO;
1126 }
1127
1128 req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout,
1129 ceph_vino(inode),
1130 offset, length,
1131 1, op,
1132 CEPH_OSD_FLAG_WRITE |
1133 CEPH_OSD_FLAG_ONDISK,
1134 NULL, 0, 0, false);
1135 if (IS_ERR(req)) {
1136 ret = PTR_ERR(req);
1137 goto out;
1138 }
1139
1140 ceph_osdc_build_request(req, offset, NULL, ceph_vino(inode).snap,
1141 &inode->i_mtime);
1142
1143 ret = ceph_osdc_start_request(&fsc->client->osdc, req, false);
1144 if (!ret) {
1145 ret = ceph_osdc_wait_request(&fsc->client->osdc, req);
1146 if (ret == -ENOENT)
1147 ret = 0;
1148 }
1149 ceph_osdc_put_request(req);
1150
1151 out:
1152 return ret;
1153 }
1154
1155 static int ceph_zero_objects(struct inode *inode, loff_t offset, loff_t length)
1156 {
1157 int ret = 0;
1158 struct ceph_inode_info *ci = ceph_inode(inode);
1159 s32 stripe_unit = ceph_file_layout_su(ci->i_layout);
1160 s32 stripe_count = ceph_file_layout_stripe_count(ci->i_layout);
1161 s32 object_size = ceph_file_layout_object_size(ci->i_layout);
1162 u64 object_set_size = object_size * stripe_count;
1163 u64 nearly, t;
1164
1165 /* round offset up to next period boundary */
1166 nearly = offset + object_set_size - 1;
1167 t = nearly;
1168 nearly -= do_div(t, object_set_size);
1169
1170 while (length && offset < nearly) {
1171 loff_t size = length;
1172 ret = ceph_zero_partial_object(inode, offset, &size);
1173 if (ret < 0)
1174 return ret;
1175 offset += size;
1176 length -= size;
1177 }
1178 while (length >= object_set_size) {
1179 int i;
1180 loff_t pos = offset;
1181 for (i = 0; i < stripe_count; ++i) {
1182 ret = ceph_zero_partial_object(inode, pos, NULL);
1183 if (ret < 0)
1184 return ret;
1185 pos += stripe_unit;
1186 }
1187 offset += object_set_size;
1188 length -= object_set_size;
1189 }
1190 while (length) {
1191 loff_t size = length;
1192 ret = ceph_zero_partial_object(inode, offset, &size);
1193 if (ret < 0)
1194 return ret;
1195 offset += size;
1196 length -= size;
1197 }
1198 return ret;
1199 }
1200
1201 static long ceph_fallocate(struct file *file, int mode,
1202 loff_t offset, loff_t length)
1203 {
1204 struct ceph_file_info *fi = file->private_data;
1205 struct inode *inode = file_inode(file);
1206 struct ceph_inode_info *ci = ceph_inode(inode);
1207 struct ceph_osd_client *osdc =
1208 &ceph_inode_to_client(inode)->client->osdc;
1209 int want, got = 0;
1210 int dirty;
1211 int ret = 0;
1212 loff_t endoff = 0;
1213 loff_t size;
1214
1215 if (!S_ISREG(inode->i_mode))
1216 return -EOPNOTSUPP;
1217
1218 if (IS_SWAPFILE(inode))
1219 return -ETXTBSY;
1220
1221 mutex_lock(&inode->i_mutex);
1222
1223 if (ceph_snap(inode) != CEPH_NOSNAP) {
1224 ret = -EROFS;
1225 goto unlock;
1226 }
1227
1228 if (ceph_osdmap_flag(osdc->osdmap, CEPH_OSDMAP_FULL) &&
1229 !(mode & FALLOC_FL_PUNCH_HOLE)) {
1230 ret = -ENOSPC;
1231 goto unlock;
1232 }
1233
1234 size = i_size_read(inode);
1235 if (!(mode & FALLOC_FL_KEEP_SIZE))
1236 endoff = offset + length;
1237
1238 if (fi->fmode & CEPH_FILE_MODE_LAZY)
1239 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
1240 else
1241 want = CEPH_CAP_FILE_BUFFER;
1242
1243 ret = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, &got, endoff);
1244 if (ret < 0)
1245 goto unlock;
1246
1247 if (mode & FALLOC_FL_PUNCH_HOLE) {
1248 if (offset < size)
1249 ceph_zero_pagecache_range(inode, offset, length);
1250 ret = ceph_zero_objects(inode, offset, length);
1251 } else if (endoff > size) {
1252 truncate_pagecache_range(inode, size, -1);
1253 if (ceph_inode_set_size(inode, endoff))
1254 ceph_check_caps(ceph_inode(inode),
1255 CHECK_CAPS_AUTHONLY, NULL);
1256 }
1257
1258 if (!ret) {
1259 spin_lock(&ci->i_ceph_lock);
1260 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR);
1261 spin_unlock(&ci->i_ceph_lock);
1262 if (dirty)
1263 __mark_inode_dirty(inode, dirty);
1264 }
1265
1266 ceph_put_cap_refs(ci, got);
1267 unlock:
1268 mutex_unlock(&inode->i_mutex);
1269 return ret;
1270 }
1271
1272 const struct file_operations ceph_file_fops = {
1273 .open = ceph_open,
1274 .release = ceph_release,
1275 .llseek = ceph_llseek,
1276 .read = do_sync_read,
1277 .write = do_sync_write,
1278 .aio_read = ceph_aio_read,
1279 .aio_write = ceph_aio_write,
1280 .mmap = ceph_mmap,
1281 .fsync = ceph_fsync,
1282 .lock = ceph_lock,
1283 .flock = ceph_flock,
1284 .splice_read = generic_file_splice_read,
1285 .splice_write = generic_file_splice_write,
1286 .unlocked_ioctl = ceph_ioctl,
1287 .compat_ioctl = ceph_ioctl,
1288 .fallocate = ceph_fallocate,
1289 };
1290
Linux with added FPC-III support