1#include <linux/ceph/ceph_debug.h>
2
3#include <linux/backing-dev.h>
4#include <linux/fs.h>
5#include <linux/mm.h>
6#include <linux/pagemap.h>
7#include <linux/writeback.h>
8#include <linux/slab.h>
9#include <linux/pagevec.h>
10#include <linux/task_io_accounting_ops.h>
11
12#include "super.h"
13#include "mds_client.h"
14#include "cache.h"
15#include <linux/ceph/osd_client.h>
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53#define CONGESTION_ON_THRESH(congestion_kb) (congestion_kb >> (PAGE_SHIFT-10))
54#define CONGESTION_OFF_THRESH(congestion_kb) \
55 (CONGESTION_ON_THRESH(congestion_kb) - \
56 (CONGESTION_ON_THRESH(congestion_kb) >> 2))
57
58static inline struct ceph_snap_context *page_snap_context(struct page *page)
59{
60 if (PagePrivate(page))
61 return (void *)page->private;
62 return NULL;
63}
64
65
66
67
68
69static int ceph_set_page_dirty(struct page *page)
70{
71 struct address_space *mapping = page->mapping;
72 struct inode *inode;
73 struct ceph_inode_info *ci;
74 struct ceph_snap_context *snapc;
75 int ret;
76
77 if (unlikely(!mapping))
78 return !TestSetPageDirty(page);
79
80 if (PageDirty(page)) {
81 dout("%p set_page_dirty %p idx %lu -- already dirty\n",
82 mapping->host, page, page->index);
83 BUG_ON(!PagePrivate(page));
84 return 0;
85 }
86
87 inode = mapping->host;
88 ci = ceph_inode(inode);
89
90
91
92
93
94 snapc = ceph_get_snap_context(ci->i_snap_realm->cached_context);
95
96
97 spin_lock(&ci->i_ceph_lock);
98 if (ci->i_head_snapc == NULL)
99 ci->i_head_snapc = ceph_get_snap_context(snapc);
100 ++ci->i_wrbuffer_ref_head;
101 if (ci->i_wrbuffer_ref == 0)
102 ihold(inode);
103 ++ci->i_wrbuffer_ref;
104 dout("%p set_page_dirty %p idx %lu head %d/%d -> %d/%d "
105 "snapc %p seq %lld (%d snaps)\n",
106 mapping->host, page, page->index,
107 ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref_head-1,
108 ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head,
109 snapc, snapc->seq, snapc->num_snaps);
110 spin_unlock(&ci->i_ceph_lock);
111
112
113
114
115
116 BUG_ON(PagePrivate(page));
117 page->private = (unsigned long)snapc;
118 SetPagePrivate(page);
119
120 ret = __set_page_dirty_nobuffers(page);
121 WARN_ON(!PageLocked(page));
122 WARN_ON(!page->mapping);
123
124 return ret;
125}
126
127
128
129
130
131
132static void ceph_invalidatepage(struct page *page, unsigned int offset,
133 unsigned int length)
134{
135 struct inode *inode;
136 struct ceph_inode_info *ci;
137 struct ceph_snap_context *snapc = page_snap_context(page);
138
139 inode = page->mapping->host;
140 ci = ceph_inode(inode);
141
142 if (offset != 0 || length != PAGE_CACHE_SIZE) {
143 dout("%p invalidatepage %p idx %lu partial dirty page %u~%u\n",
144 inode, page, page->index, offset, length);
145 return;
146 }
147
148 ceph_invalidate_fscache_page(inode, page);
149
150 if (!PagePrivate(page))
151 return;
152
153
154
155
156
157
158 if (!PageDirty(page))
159 pr_err("%p invalidatepage %p page not dirty\n", inode, page);
160
161 ClearPageChecked(page);
162
163 dout("%p invalidatepage %p idx %lu full dirty page\n",
164 inode, page, page->index);
165
166 ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
167 ceph_put_snap_context(snapc);
168 page->private = 0;
169 ClearPagePrivate(page);
170}
171
172static int ceph_releasepage(struct page *page, gfp_t g)
173{
174 struct inode *inode = page->mapping ? page->mapping->host : NULL;
175 dout("%p releasepage %p idx %lu\n", inode, page, page->index);
176 WARN_ON(PageDirty(page));
177
178
179 if (!ceph_release_fscache_page(page, g))
180 return 0;
181
182 return !PagePrivate(page);
183}
184
185
186
187
188static int readpage_nounlock(struct file *filp, struct page *page)
189{
190 struct inode *inode = file_inode(filp);
191 struct ceph_inode_info *ci = ceph_inode(inode);
192 struct ceph_osd_client *osdc =
193 &ceph_inode_to_client(inode)->client->osdc;
194 int err = 0;
195 u64 len = PAGE_CACHE_SIZE;
196
197 err = ceph_readpage_from_fscache(inode, page);
198
199 if (err == 0)
200 goto out;
201
202 dout("readpage inode %p file %p page %p index %lu\n",
203 inode, filp, page, page->index);
204 err = ceph_osdc_readpages(osdc, ceph_vino(inode), &ci->i_layout,
205 (u64) page_offset(page), &len,
206 ci->i_truncate_seq, ci->i_truncate_size,
207 &page, 1, 0);
208 if (err == -ENOENT)
209 err = 0;
210 if (err < 0) {
211 SetPageError(page);
212 ceph_fscache_readpage_cancel(inode, page);
213 goto out;
214 }
215 if (err < PAGE_CACHE_SIZE)
216
217 zero_user_segment(page, err, PAGE_CACHE_SIZE);
218 else
219 flush_dcache_page(page);
220
221 SetPageUptodate(page);
222 ceph_readpage_to_fscache(inode, page);
223
224out:
225 return err < 0 ? err : 0;
226}
227
228static int ceph_readpage(struct file *filp, struct page *page)
229{
230 int r = readpage_nounlock(filp, page);
231 unlock_page(page);
232 return r;
233}
234
235
236
237
238static void finish_read(struct ceph_osd_request *req, struct ceph_msg *msg)
239{
240 struct inode *inode = req->r_inode;
241 struct ceph_osd_data *osd_data;
242 int rc = req->r_result;
243 int bytes = le32_to_cpu(msg->hdr.data_len);
244 int num_pages;
245 int i;
246
247 dout("finish_read %p req %p rc %d bytes %d\n", inode, req, rc, bytes);
248
249
250 osd_data = osd_req_op_extent_osd_data(req, 0);
251 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
252 num_pages = calc_pages_for((u64)osd_data->alignment,
253 (u64)osd_data->length);
254 for (i = 0; i < num_pages; i++) {
255 struct page *page = osd_data->pages[i];
256
257 if (rc < 0)
258 goto unlock;
259 if (bytes < (int)PAGE_CACHE_SIZE) {
260
261 int s = bytes < 0 ? 0 : bytes;
262 zero_user_segment(page, s, PAGE_CACHE_SIZE);
263 }
264 dout("finish_read %p uptodate %p idx %lu\n", inode, page,
265 page->index);
266 flush_dcache_page(page);
267 SetPageUptodate(page);
268 ceph_readpage_to_fscache(inode, page);
269unlock:
270 unlock_page(page);
271 page_cache_release(page);
272 bytes -= PAGE_CACHE_SIZE;
273 }
274 kfree(osd_data->pages);
275}
276
277static void ceph_unlock_page_vector(struct page **pages, int num_pages)
278{
279 int i;
280
281 for (i = 0; i < num_pages; i++)
282 unlock_page(pages[i]);
283}
284
285
286
287
288
289static int start_read(struct inode *inode, struct list_head *page_list, int max)
290{
291 struct ceph_osd_client *osdc =
292 &ceph_inode_to_client(inode)->client->osdc;
293 struct ceph_inode_info *ci = ceph_inode(inode);
294 struct page *page = list_entry(page_list->prev, struct page, lru);
295 struct ceph_vino vino;
296 struct ceph_osd_request *req;
297 u64 off;
298 u64 len;
299 int i;
300 struct page **pages;
301 pgoff_t next_index;
302 int nr_pages = 0;
303 int ret;
304
305 off = (u64) page_offset(page);
306
307
308 next_index = page->index;
309 list_for_each_entry_reverse(page, page_list, lru) {
310 if (page->index != next_index)
311 break;
312 nr_pages++;
313 next_index++;
314 if (max && nr_pages == max)
315 break;
316 }
317 len = nr_pages << PAGE_CACHE_SHIFT;
318 dout("start_read %p nr_pages %d is %lld~%lld\n", inode, nr_pages,
319 off, len);
320 vino = ceph_vino(inode);
321 req = ceph_osdc_new_request(osdc, &ci->i_layout, vino, off, &len,
322 1, CEPH_OSD_OP_READ,
323 CEPH_OSD_FLAG_READ, NULL,
324 ci->i_truncate_seq, ci->i_truncate_size,
325 false);
326 if (IS_ERR(req))
327 return PTR_ERR(req);
328
329
330 nr_pages = calc_pages_for(0, len);
331 pages = kmalloc(sizeof(*pages) * nr_pages, GFP_NOFS);
332 ret = -ENOMEM;
333 if (!pages)
334 goto out;
335 for (i = 0; i < nr_pages; ++i) {
336 page = list_entry(page_list->prev, struct page, lru);
337 BUG_ON(PageLocked(page));
338 list_del(&page->lru);
339
340 dout("start_read %p adding %p idx %lu\n", inode, page,
341 page->index);
342 if (add_to_page_cache_lru(page, &inode->i_data, page->index,
343 GFP_NOFS)) {
344 ceph_fscache_uncache_page(inode, page);
345 page_cache_release(page);
346 dout("start_read %p add_to_page_cache failed %p\n",
347 inode, page);
348 nr_pages = i;
349 goto out_pages;
350 }
351 pages[i] = page;
352 }
353 osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false, false);
354 req->r_callback = finish_read;
355 req->r_inode = inode;
356
357 ceph_osdc_build_request(req, off, NULL, vino.snap, NULL);
358
359 dout("start_read %p starting %p %lld~%lld\n", inode, req, off, len);
360 ret = ceph_osdc_start_request(osdc, req, false);
361 if (ret < 0)
362 goto out_pages;
363 ceph_osdc_put_request(req);
364 return nr_pages;
365
366out_pages:
367 ceph_unlock_page_vector(pages, nr_pages);
368 ceph_release_page_vector(pages, nr_pages);
369out:
370 ceph_osdc_put_request(req);
371 return ret;
372}
373
374
375
376
377
378
379static int ceph_readpages(struct file *file, struct address_space *mapping,
380 struct list_head *page_list, unsigned nr_pages)
381{
382 struct inode *inode = file_inode(file);
383 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
384 int rc = 0;
385 int max = 0;
386
387 rc = ceph_readpages_from_fscache(mapping->host, mapping, page_list,
388 &nr_pages);
389
390 if (rc == 0)
391 goto out;
392
393 if (fsc->mount_options->rsize >= PAGE_CACHE_SIZE)
394 max = (fsc->mount_options->rsize + PAGE_CACHE_SIZE - 1)
395 >> PAGE_SHIFT;
396
397 dout("readpages %p file %p nr_pages %d max %d\n", inode,
398 file, nr_pages,
399 max);
400 while (!list_empty(page_list)) {
401 rc = start_read(inode, page_list, max);
402 if (rc < 0)
403 goto out;
404 BUG_ON(rc == 0);
405 }
406out:
407 ceph_fscache_readpages_cancel(inode, page_list);
408
409 dout("readpages %p file %p ret %d\n", inode, file, rc);
410 return rc;
411}
412
413
414
415
416
417static struct ceph_snap_context *get_oldest_context(struct inode *inode,
418 u64 *snap_size)
419{
420 struct ceph_inode_info *ci = ceph_inode(inode);
421 struct ceph_snap_context *snapc = NULL;
422 struct ceph_cap_snap *capsnap = NULL;
423
424 spin_lock(&ci->i_ceph_lock);
425 list_for_each_entry(capsnap, &ci->i_cap_snaps, ci_item) {
426 dout(" cap_snap %p snapc %p has %d dirty pages\n", capsnap,
427 capsnap->context, capsnap->dirty_pages);
428 if (capsnap->dirty_pages) {
429 snapc = ceph_get_snap_context(capsnap->context);
430 if (snap_size)
431 *snap_size = capsnap->size;
432 break;
433 }
434 }
435 if (!snapc && ci->i_wrbuffer_ref_head) {
436 snapc = ceph_get_snap_context(ci->i_head_snapc);
437 dout(" head snapc %p has %d dirty pages\n",
438 snapc, ci->i_wrbuffer_ref_head);
439 }
440 spin_unlock(&ci->i_ceph_lock);
441 return snapc;
442}
443
444
445
446
447
448
449
450static int writepage_nounlock(struct page *page, struct writeback_control *wbc)
451{
452 struct inode *inode;
453 struct ceph_inode_info *ci;
454 struct ceph_fs_client *fsc;
455 struct ceph_osd_client *osdc;
456 struct ceph_snap_context *snapc, *oldest;
457 loff_t page_off = page_offset(page);
458 long writeback_stat;
459 u64 truncate_size, snap_size = 0;
460 u32 truncate_seq;
461 int err = 0, len = PAGE_CACHE_SIZE;
462
463 dout("writepage %p idx %lu\n", page, page->index);
464
465 if (!page->mapping || !page->mapping->host) {
466 dout("writepage %p - no mapping\n", page);
467 return -EFAULT;
468 }
469 inode = page->mapping->host;
470 ci = ceph_inode(inode);
471 fsc = ceph_inode_to_client(inode);
472 osdc = &fsc->client->osdc;
473
474
475 snapc = page_snap_context(page);
476 if (snapc == NULL) {
477 dout("writepage %p page %p not dirty?\n", inode, page);
478 goto out;
479 }
480 oldest = get_oldest_context(inode, &snap_size);
481 if (snapc->seq > oldest->seq) {
482 dout("writepage %p page %p snapc %p not writeable - noop\n",
483 inode, page, snapc);
484
485 WARN_ON((current->flags & PF_MEMALLOC) == 0);
486 ceph_put_snap_context(oldest);
487 goto out;
488 }
489 ceph_put_snap_context(oldest);
490
491 spin_lock(&ci->i_ceph_lock);
492 truncate_seq = ci->i_truncate_seq;
493 truncate_size = ci->i_truncate_size;
494 if (!snap_size)
495 snap_size = i_size_read(inode);
496 spin_unlock(&ci->i_ceph_lock);
497
498
499 if (page_off >= snap_size) {
500 dout("%p page eof %llu\n", page, snap_size);
501 goto out;
502 }
503 if (snap_size < page_off + len)
504 len = snap_size - page_off;
505
506 dout("writepage %p page %p index %lu on %llu~%u snapc %p\n",
507 inode, page, page->index, page_off, len, snapc);
508
509 writeback_stat = atomic_long_inc_return(&fsc->writeback_count);
510 if (writeback_stat >
511 CONGESTION_ON_THRESH(fsc->mount_options->congestion_kb))
512 set_bdi_congested(&fsc->backing_dev_info, BLK_RW_ASYNC);
513
514 ceph_readpage_to_fscache(inode, page);
515
516 set_page_writeback(page);
517 err = ceph_osdc_writepages(osdc, ceph_vino(inode),
518 &ci->i_layout, snapc,
519 page_off, len,
520 truncate_seq, truncate_size,
521 &inode->i_mtime, &page, 1);
522 if (err < 0) {
523 dout("writepage setting page/mapping error %d %p\n", err, page);
524 SetPageError(page);
525 mapping_set_error(&inode->i_data, err);
526 if (wbc)
527 wbc->pages_skipped++;
528 } else {
529 dout("writepage cleaned page %p\n", page);
530 err = 0;
531 }
532 page->private = 0;
533 ClearPagePrivate(page);
534 end_page_writeback(page);
535 ceph_put_wrbuffer_cap_refs(ci, 1, snapc);
536 ceph_put_snap_context(snapc);
537out:
538 return err;
539}
540
541static int ceph_writepage(struct page *page, struct writeback_control *wbc)
542{
543 int err;
544 struct inode *inode = page->mapping->host;
545 BUG_ON(!inode);
546 ihold(inode);
547 err = writepage_nounlock(page, wbc);
548 unlock_page(page);
549 iput(inode);
550 return err;
551}
552
553
554
555
556
557
558static void ceph_release_pages(struct page **pages, int num)
559{
560 struct pagevec pvec;
561 int i;
562
563 pagevec_init(&pvec, 0);
564 for (i = 0; i < num; i++) {
565 if (pagevec_add(&pvec, pages[i]) == 0)
566 pagevec_release(&pvec);
567 }
568 pagevec_release(&pvec);
569}
570
571
572
573
574
575
576
577static void writepages_finish(struct ceph_osd_request *req,
578 struct ceph_msg *msg)
579{
580 struct inode *inode = req->r_inode;
581 struct ceph_inode_info *ci = ceph_inode(inode);
582 struct ceph_osd_data *osd_data;
583 unsigned wrote;
584 struct page *page;
585 int num_pages;
586 int i;
587 struct ceph_snap_context *snapc = req->r_snapc;
588 struct address_space *mapping = inode->i_mapping;
589 int rc = req->r_result;
590 u64 bytes = req->r_ops[0].extent.length;
591 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
592 long writeback_stat;
593 unsigned issued = ceph_caps_issued(ci);
594
595 osd_data = osd_req_op_extent_osd_data(req, 0);
596 BUG_ON(osd_data->type != CEPH_OSD_DATA_TYPE_PAGES);
597 num_pages = calc_pages_for((u64)osd_data->alignment,
598 (u64)osd_data->length);
599 if (rc >= 0) {
600
601
602
603
604
605
606 wrote = num_pages;
607 } else {
608 wrote = 0;
609 mapping_set_error(mapping, rc);
610 }
611 dout("writepages_finish %p rc %d bytes %llu wrote %d (pages)\n",
612 inode, rc, bytes, wrote);
613
614
615 for (i = 0; i < num_pages; i++) {
616 page = osd_data->pages[i];
617 BUG_ON(!page);
618 WARN_ON(!PageUptodate(page));
619
620 writeback_stat =
621 atomic_long_dec_return(&fsc->writeback_count);
622 if (writeback_stat <
623 CONGESTION_OFF_THRESH(fsc->mount_options->congestion_kb))
624 clear_bdi_congested(&fsc->backing_dev_info,
625 BLK_RW_ASYNC);
626
627 ceph_put_snap_context(page_snap_context(page));
628 page->private = 0;
629 ClearPagePrivate(page);
630 dout("unlocking %d %p\n", i, page);
631 end_page_writeback(page);
632
633
634
635
636
637
638
639 if ((issued & (CEPH_CAP_FILE_CACHE|CEPH_CAP_FILE_LAZYIO)) == 0)
640 generic_error_remove_page(inode->i_mapping, page);
641
642 unlock_page(page);
643 }
644 dout("%p wrote+cleaned %d pages\n", inode, wrote);
645 ceph_put_wrbuffer_cap_refs(ci, num_pages, snapc);
646
647 ceph_release_pages(osd_data->pages, num_pages);
648 if (osd_data->pages_from_pool)
649 mempool_free(osd_data->pages,
650 ceph_sb_to_client(inode->i_sb)->wb_pagevec_pool);
651 else
652 kfree(osd_data->pages);
653 ceph_osdc_put_request(req);
654}
655
656
657
658
659static int ceph_writepages_start(struct address_space *mapping,
660 struct writeback_control *wbc)
661{
662 struct inode *inode = mapping->host;
663 struct ceph_inode_info *ci = ceph_inode(inode);
664 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
665 struct ceph_vino vino = ceph_vino(inode);
666 pgoff_t index, start, end;
667 int range_whole = 0;
668 int should_loop = 1;
669 pgoff_t max_pages = 0, max_pages_ever = 0;
670 struct ceph_snap_context *snapc = NULL, *last_snapc = NULL, *pgsnapc;
671 struct pagevec pvec;
672 int done = 0;
673 int rc = 0;
674 unsigned wsize = 1 << inode->i_blkbits;
675 struct ceph_osd_request *req = NULL;
676 int do_sync;
677 u64 truncate_size, snap_size;
678 u32 truncate_seq;
679
680
681
682
683
684
685 if ((wbc->sync_mode == WB_SYNC_ALL) ||
686 ceph_caps_revoking(ci, CEPH_CAP_FILE_BUFFER))
687 do_sync = 1;
688 dout("writepages_start %p dosync=%d (mode=%s)\n",
689 inode, do_sync,
690 wbc->sync_mode == WB_SYNC_NONE ? "NONE" :
691 (wbc->sync_mode == WB_SYNC_ALL ? "ALL" : "HOLD"));
692
693 if (fsc->mount_state == CEPH_MOUNT_SHUTDOWN) {
694 pr_warn("writepage_start %p on forced umount\n", inode);
695 return -EIO;
696 }
697 if (fsc->mount_options->wsize && fsc->mount_options->wsize < wsize)
698 wsize = fsc->mount_options->wsize;
699 if (wsize < PAGE_CACHE_SIZE)
700 wsize = PAGE_CACHE_SIZE;
701 max_pages_ever = wsize >> PAGE_CACHE_SHIFT;
702
703 pagevec_init(&pvec, 0);
704
705
706 if (wbc->range_cyclic) {
707 start = mapping->writeback_index;
708 end = -1;
709 dout(" cyclic, start at %lu\n", start);
710 } else {
711 start = wbc->range_start >> PAGE_CACHE_SHIFT;
712 end = wbc->range_end >> PAGE_CACHE_SHIFT;
713 if (wbc->range_start == 0 && wbc->range_end == LLONG_MAX)
714 range_whole = 1;
715 should_loop = 0;
716 dout(" not cyclic, %lu to %lu\n", start, end);
717 }
718 index = start;
719
720retry:
721
722 ceph_put_snap_context(snapc);
723 snap_size = 0;
724 snapc = get_oldest_context(inode, &snap_size);
725 if (!snapc) {
726
727
728 dout(" no snap context with dirty data?\n");
729 goto out;
730 }
731 if (snap_size == 0)
732 snap_size = i_size_read(inode);
733 dout(" oldest snapc is %p seq %lld (%d snaps)\n",
734 snapc, snapc->seq, snapc->num_snaps);
735
736 spin_lock(&ci->i_ceph_lock);
737 truncate_seq = ci->i_truncate_seq;
738 truncate_size = ci->i_truncate_size;
739 if (!snap_size)
740 snap_size = i_size_read(inode);
741 spin_unlock(&ci->i_ceph_lock);
742
743 if (last_snapc && snapc != last_snapc) {
744
745
746 dout(" snapc differs from last pass, restarting at %lu\n",
747 index);
748 index = start;
749 }
750 last_snapc = snapc;
751
752 while (!done && index <= end) {
753 int num_ops = do_sync ? 2 : 1;
754 unsigned i;
755 int first;
756 pgoff_t next;
757 int pvec_pages, locked_pages;
758 struct page **pages = NULL;
759 mempool_t *pool = NULL;
760 struct page *page;
761 int want;
762 u64 offset, len;
763 long writeback_stat;
764
765 next = 0;
766 locked_pages = 0;
767 max_pages = max_pages_ever;
768
769get_more_pages:
770 first = -1;
771 want = min(end - index,
772 min((pgoff_t)PAGEVEC_SIZE,
773 max_pages - (pgoff_t)locked_pages) - 1)
774 + 1;
775 pvec_pages = pagevec_lookup_tag(&pvec, mapping, &index,
776 PAGECACHE_TAG_DIRTY,
777 want);
778 dout("pagevec_lookup_tag got %d\n", pvec_pages);
779 if (!pvec_pages && !locked_pages)
780 break;
781 for (i = 0; i < pvec_pages && locked_pages < max_pages; i++) {
782 page = pvec.pages[i];
783 dout("? %p idx %lu\n", page, page->index);
784 if (locked_pages == 0)
785 lock_page(page);
786 else if (!trylock_page(page))
787 break;
788
789
790 if (unlikely(!PageDirty(page)) ||
791 unlikely(page->mapping != mapping)) {
792 dout("!dirty or !mapping %p\n", page);
793 unlock_page(page);
794 break;
795 }
796 if (!wbc->range_cyclic && page->index > end) {
797 dout("end of range %p\n", page);
798 done = 1;
799 unlock_page(page);
800 break;
801 }
802 if (next && (page->index != next)) {
803 dout("not consecutive %p\n", page);
804 unlock_page(page);
805 break;
806 }
807 if (wbc->sync_mode != WB_SYNC_NONE) {
808 dout("waiting on writeback %p\n", page);
809 wait_on_page_writeback(page);
810 }
811 if (page_offset(page) >= snap_size) {
812 dout("%p page eof %llu\n", page, snap_size);
813 done = 1;
814 unlock_page(page);
815 break;
816 }
817 if (PageWriteback(page)) {
818 dout("%p under writeback\n", page);
819 unlock_page(page);
820 break;
821 }
822
823
824 pgsnapc = page_snap_context(page);
825 if (pgsnapc->seq > snapc->seq) {
826 dout("page snapc %p %lld > oldest %p %lld\n",
827 pgsnapc, pgsnapc->seq, snapc, snapc->seq);
828 unlock_page(page);
829 if (!locked_pages)
830 continue;
831 break;
832 }
833
834 if (!clear_page_dirty_for_io(page)) {
835 dout("%p !clear_page_dirty_for_io\n", page);
836 unlock_page(page);
837 break;
838 }
839
840
841
842
843
844
845
846 if (locked_pages == 0) {
847 BUG_ON(pages);
848
849 offset = (u64)page_offset(page);
850 len = wsize;
851 req = ceph_osdc_new_request(&fsc->client->osdc,
852 &ci->i_layout, vino,
853 offset, &len, num_ops,
854 CEPH_OSD_OP_WRITE,
855 CEPH_OSD_FLAG_WRITE |
856 CEPH_OSD_FLAG_ONDISK,
857 snapc, truncate_seq,
858 truncate_size, true);
859 if (IS_ERR(req)) {
860 rc = PTR_ERR(req);
861 unlock_page(page);
862 break;
863 }
864
865 req->r_callback = writepages_finish;
866 req->r_inode = inode;
867
868 max_pages = calc_pages_for(0, (u64)len);
869 pages = kmalloc(max_pages * sizeof (*pages),
870 GFP_NOFS);
871 if (!pages) {
872 pool = fsc->wb_pagevec_pool;
873 pages = mempool_alloc(pool, GFP_NOFS);
874 BUG_ON(!pages);
875 }
876 }
877
878
879 if (first < 0)
880 first = i;
881 dout("%p will write page %p idx %lu\n",
882 inode, page, page->index);
883
884 writeback_stat =
885 atomic_long_inc_return(&fsc->writeback_count);
886 if (writeback_stat > CONGESTION_ON_THRESH(
887 fsc->mount_options->congestion_kb)) {
888 set_bdi_congested(&fsc->backing_dev_info,
889 BLK_RW_ASYNC);
890 }
891
892 set_page_writeback(page);
893 pages[locked_pages] = page;
894 locked_pages++;
895 next = page->index + 1;
896 }
897
898
899 if (!locked_pages)
900 goto release_pvec_pages;
901 if (i) {
902 int j;
903 BUG_ON(!locked_pages || first < 0);
904
905 if (pvec_pages && i == pvec_pages &&
906 locked_pages < max_pages) {
907 dout("reached end pvec, trying for more\n");
908 pagevec_reinit(&pvec);
909 goto get_more_pages;
910 }
911
912
913
914 for (j = i; j < pvec_pages; j++) {
915 dout(" pvec leftover page %p\n",
916 pvec.pages[j]);
917 pvec.pages[j-i+first] = pvec.pages[j];
918 }
919 pvec.nr -= i-first;
920 }
921
922
923
924 offset = page_offset(pages[0]);
925 len = min(snap_size - offset,
926 (u64)locked_pages << PAGE_CACHE_SHIFT);
927 dout("writepages got %d pages at %llu~%llu\n",
928 locked_pages, offset, len);
929
930 osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0,
931 !!pool, false);
932
933 pages = NULL;
934 pool = NULL;
935
936
937
938 osd_req_op_extent_update(req, 0, len);
939
940 vino = ceph_vino(inode);
941 ceph_osdc_build_request(req, offset, snapc, vino.snap,
942 &inode->i_mtime);
943
944 rc = ceph_osdc_start_request(&fsc->client->osdc, req, true);
945 BUG_ON(rc);
946 req = NULL;
947
948
949 index = next;
950 wbc->nr_to_write -= locked_pages;
951 if (wbc->nr_to_write <= 0)
952 done = 1;
953
954release_pvec_pages:
955 dout("pagevec_release on %d pages (%p)\n", (int)pvec.nr,
956 pvec.nr ? pvec.pages[0] : NULL);
957 pagevec_release(&pvec);
958
959 if (locked_pages && !done)
960 goto retry;
961 }
962
963 if (should_loop && !done) {
964
965 dout("writepages looping back to beginning of file\n");
966 should_loop = 0;
967 index = 0;
968 goto retry;
969 }
970
971 if (wbc->range_cyclic || (range_whole && wbc->nr_to_write > 0))
972 mapping->writeback_index = index;
973
974out:
975 if (req)
976 ceph_osdc_put_request(req);
977 ceph_put_snap_context(snapc);
978 dout("writepages done, rc = %d\n", rc);
979 return rc;
980}
981
982
983
984
985
986
987static int context_is_writeable_or_written(struct inode *inode,
988 struct ceph_snap_context *snapc)
989{
990 struct ceph_snap_context *oldest = get_oldest_context(inode, NULL);
991 int ret = !oldest || snapc->seq <= oldest->seq;
992
993 ceph_put_snap_context(oldest);
994 return ret;
995}
996
997
998
999
1000
1001
1002
1003
1004
1005static int ceph_update_writeable_page(struct file *file,
1006 loff_t pos, unsigned len,
1007 struct page *page)
1008{
1009 struct inode *inode = file_inode(file);
1010 struct ceph_inode_info *ci = ceph_inode(inode);
1011 struct ceph_mds_client *mdsc = ceph_inode_to_client(inode)->mdsc;
1012 loff_t page_off = pos & PAGE_CACHE_MASK;
1013 int pos_in_page = pos & ~PAGE_CACHE_MASK;
1014 int end_in_page = pos_in_page + len;
1015 loff_t i_size;
1016 int r;
1017 struct ceph_snap_context *snapc, *oldest;
1018
1019retry_locked:
1020
1021 wait_on_page_writeback(page);
1022
1023
1024 BUG_ON(!ci->i_snap_realm);
1025 down_read(&mdsc->snap_rwsem);
1026 BUG_ON(!ci->i_snap_realm->cached_context);
1027 snapc = page_snap_context(page);
1028 if (snapc && snapc != ci->i_head_snapc) {
1029
1030
1031
1032
1033 oldest = get_oldest_context(inode, NULL);
1034 up_read(&mdsc->snap_rwsem);
1035
1036 if (snapc->seq > oldest->seq) {
1037 ceph_put_snap_context(oldest);
1038 dout(" page %p snapc %p not current or oldest\n",
1039 page, snapc);
1040
1041
1042
1043
1044 snapc = ceph_get_snap_context(snapc);
1045 unlock_page(page);
1046 ceph_queue_writeback(inode);
1047 r = wait_event_interruptible(ci->i_cap_wq,
1048 context_is_writeable_or_written(inode, snapc));
1049 ceph_put_snap_context(snapc);
1050 if (r == -ERESTARTSYS)
1051 return r;
1052 return -EAGAIN;
1053 }
1054 ceph_put_snap_context(oldest);
1055
1056
1057 dout(" page %p snapc %p not current, but oldest\n",
1058 page, snapc);
1059 if (!clear_page_dirty_for_io(page))
1060 goto retry_locked;
1061 r = writepage_nounlock(page, NULL);
1062 if (r < 0)
1063 goto fail_nosnap;
1064 goto retry_locked;
1065 }
1066
1067 if (PageUptodate(page)) {
1068 dout(" page %p already uptodate\n", page);
1069 return 0;
1070 }
1071
1072
1073 if (pos_in_page == 0 && len == PAGE_CACHE_SIZE)
1074 return 0;
1075
1076
1077 i_size = inode->i_size;
1078
1079 if (page_off >= i_size ||
1080 (pos_in_page == 0 && (pos+len) >= i_size &&
1081 end_in_page - pos_in_page != PAGE_CACHE_SIZE)) {
1082 dout(" zeroing %p 0 - %d and %d - %d\n",
1083 page, pos_in_page, end_in_page, (int)PAGE_CACHE_SIZE);
1084 zero_user_segments(page,
1085 0, pos_in_page,
1086 end_in_page, PAGE_CACHE_SIZE);
1087 return 0;
1088 }
1089
1090
1091 up_read(&mdsc->snap_rwsem);
1092 r = readpage_nounlock(file, page);
1093 if (r < 0)
1094 goto fail_nosnap;
1095 goto retry_locked;
1096fail_nosnap:
1097 unlock_page(page);
1098 return r;
1099}
1100
1101
1102
1103
1104
1105static int ceph_write_begin(struct file *file, struct address_space *mapping,
1106 loff_t pos, unsigned len, unsigned flags,
1107 struct page **pagep, void **fsdata)
1108{
1109 struct inode *inode = file_inode(file);
1110 struct page *page;
1111 pgoff_t index = pos >> PAGE_CACHE_SHIFT;
1112 int r;
1113
1114 do {
1115
1116 page = grab_cache_page_write_begin(mapping, index, 0);
1117 if (!page)
1118 return -ENOMEM;
1119 *pagep = page;
1120
1121 dout("write_begin file %p inode %p page %p %d~%d\n", file,
1122 inode, page, (int)pos, (int)len);
1123
1124 r = ceph_update_writeable_page(file, pos, len, page);
1125 } while (r == -EAGAIN);
1126
1127 return r;
1128}
1129
1130
1131
1132
1133
1134
1135static int ceph_write_end(struct file *file, struct address_space *mapping,
1136 loff_t pos, unsigned len, unsigned copied,
1137 struct page *page, void *fsdata)
1138{
1139 struct inode *inode = file_inode(file);
1140 struct ceph_fs_client *fsc = ceph_inode_to_client(inode);
1141 struct ceph_mds_client *mdsc = fsc->mdsc;
1142 unsigned from = pos & (PAGE_CACHE_SIZE - 1);
1143 int check_cap = 0;
1144
1145 dout("write_end file %p inode %p page %p %d~%d (%d)\n", file,
1146 inode, page, (int)pos, (int)copied, (int)len);
1147
1148
1149 if (copied < len)
1150 zero_user_segment(page, from+copied, len);
1151
1152
1153
1154 if (pos+copied > inode->i_size)
1155 check_cap = ceph_inode_set_size(inode, pos+copied);
1156
1157 if (!PageUptodate(page))
1158 SetPageUptodate(page);
1159
1160 set_page_dirty(page);
1161
1162 unlock_page(page);
1163 up_read(&mdsc->snap_rwsem);
1164 page_cache_release(page);
1165
1166 if (check_cap)
1167 ceph_check_caps(ceph_inode(inode), CHECK_CAPS_AUTHONLY, NULL);
1168
1169 return copied;
1170}
1171
1172
1173
1174
1175
1176
1177static ssize_t ceph_direct_io(int rw, struct kiocb *iocb,
1178 struct iov_iter *iter,
1179 loff_t pos)
1180{
1181 WARN_ON(1);
1182 return -EINVAL;
1183}
1184
1185const struct address_space_operations ceph_aops = {
1186 .readpage = ceph_readpage,
1187 .readpages = ceph_readpages,
1188 .writepage = ceph_writepage,
1189 .writepages = ceph_writepages_start,
1190 .write_begin = ceph_write_begin,
1191 .write_end = ceph_write_end,
1192 .set_page_dirty = ceph_set_page_dirty,
1193 .invalidatepage = ceph_invalidatepage,
1194 .releasepage = ceph_releasepage,
1195 .direct_IO = ceph_direct_io,
1196};
1197
1198
1199
1200
1201
1202static int ceph_filemap_fault(struct vm_area_struct *vma, struct vm_fault *vmf)
1203{
1204 struct inode *inode = file_inode(vma->vm_file);
1205 struct ceph_inode_info *ci = ceph_inode(inode);
1206 struct ceph_file_info *fi = vma->vm_file->private_data;
1207 loff_t off = vmf->pgoff << PAGE_CACHE_SHIFT;
1208 int want, got, ret;
1209
1210 dout("filemap_fault %p %llx.%llx %llu~%zd trying to get caps\n",
1211 inode, ceph_vinop(inode), off, (size_t)PAGE_CACHE_SIZE);
1212 if (fi->fmode & CEPH_FILE_MODE_LAZY)
1213 want = CEPH_CAP_FILE_CACHE | CEPH_CAP_FILE_LAZYIO;
1214 else
1215 want = CEPH_CAP_FILE_CACHE;
1216 while (1) {
1217 got = 0;
1218 ret = ceph_get_caps(ci, CEPH_CAP_FILE_RD, want, &got, -1);
1219 if (ret == 0)
1220 break;
1221 if (ret != -ERESTARTSYS) {
1222 WARN_ON(1);
1223 return VM_FAULT_SIGBUS;
1224 }
1225 }
1226 dout("filemap_fault %p %llu~%zd got cap refs on %s\n",
1227 inode, off, (size_t)PAGE_CACHE_SIZE, ceph_cap_string(got));
1228
1229 ret = filemap_fault(vma, vmf);
1230
1231 dout("filemap_fault %p %llu~%zd dropping cap refs on %s ret %d\n",
1232 inode, off, (size_t)PAGE_CACHE_SIZE, ceph_cap_string(got), ret);
1233 ceph_put_cap_refs(ci, got);
1234
1235 return ret;
1236}
1237
1238
1239
1240
1241static int ceph_page_mkwrite(struct vm_area_struct *vma, struct vm_fault *vmf)
1242{
1243 struct inode *inode = file_inode(vma->vm_file);
1244 struct ceph_inode_info *ci = ceph_inode(inode);
1245 struct ceph_file_info *fi = vma->vm_file->private_data;
1246 struct ceph_mds_client *mdsc = ceph_inode_to_client(inode)->mdsc;
1247 struct page *page = vmf->page;
1248 loff_t off = page_offset(page);
1249 loff_t size = i_size_read(inode);
1250 size_t len;
1251 int want, got, ret;
1252
1253 if (off + PAGE_CACHE_SIZE <= size)
1254 len = PAGE_CACHE_SIZE;
1255 else
1256 len = size & ~PAGE_CACHE_MASK;
1257
1258 dout("page_mkwrite %p %llx.%llx %llu~%zd getting caps i_size %llu\n",
1259 inode, ceph_vinop(inode), off, len, size);
1260 if (fi->fmode & CEPH_FILE_MODE_LAZY)
1261 want = CEPH_CAP_FILE_BUFFER | CEPH_CAP_FILE_LAZYIO;
1262 else
1263 want = CEPH_CAP_FILE_BUFFER;
1264 while (1) {
1265 got = 0;
1266 ret = ceph_get_caps(ci, CEPH_CAP_FILE_WR, want, &got, off + len);
1267 if (ret == 0)
1268 break;
1269 if (ret != -ERESTARTSYS) {
1270 WARN_ON(1);
1271 return VM_FAULT_SIGBUS;
1272 }
1273 }
1274 dout("page_mkwrite %p %llu~%zd got cap refs on %s\n",
1275 inode, off, len, ceph_cap_string(got));
1276
1277
1278 file_update_time(vma->vm_file);
1279
1280 lock_page(page);
1281
1282 ret = VM_FAULT_NOPAGE;
1283 if ((off > size) ||
1284 (page->mapping != inode->i_mapping))
1285 goto out;
1286
1287 ret = ceph_update_writeable_page(vma->vm_file, off, len, page);
1288 if (ret == 0) {
1289
1290 set_page_dirty(page);
1291 up_read(&mdsc->snap_rwsem);
1292 ret = VM_FAULT_LOCKED;
1293 } else {
1294 if (ret == -ENOMEM)
1295 ret = VM_FAULT_OOM;
1296 else
1297 ret = VM_FAULT_SIGBUS;
1298 }
1299out:
1300 if (ret != VM_FAULT_LOCKED) {
1301 unlock_page(page);
1302 } else {
1303 int dirty;
1304 spin_lock(&ci->i_ceph_lock);
1305 dirty = __ceph_mark_dirty_caps(ci, CEPH_CAP_FILE_WR);
1306 spin_unlock(&ci->i_ceph_lock);
1307 if (dirty)
1308 __mark_inode_dirty(inode, dirty);
1309 }
1310
1311 dout("page_mkwrite %p %llu~%zd dropping cap refs on %s ret %d\n",
1312 inode, off, len, ceph_cap_string(got), ret);
1313 ceph_put_cap_refs(ci, got);
1314
1315 return ret;
1316}
1317
1318static struct vm_operations_struct ceph_vmops = {
1319 .fault = ceph_filemap_fault,
1320 .page_mkwrite = ceph_page_mkwrite,
1321 .remap_pages = generic_file_remap_pages,
1322};
1323
1324int ceph_mmap(struct file *file, struct vm_area_struct *vma)
1325{
1326 struct address_space *mapping = file->f_mapping;
1327
1328 if (!mapping->a_ops->readpage)
1329 return -ENOEXEC;
1330 file_accessed(file);
1331 vma->vm_ops = &ceph_vmops;
1332 return 0;
1333}
1334