// SPDX-License-Identifier: GPL-2.0#include <linux/ceph/ceph_debug.h> #include <linux/backing-dev.h>#include <linux/fs.h>#include <linux/mm.h>#include <linux/swap.h>#include <linux/pagemap.h>#include <linux/slab.h>#include <linux/folio_batch.h>#include <linux/task_io_accounting_ops.h>#include <linux/signal.h>#include <linux/iversion.h>#include <linux/ktime.h>#include <linux/netfs.h>#include <trace/events/netfs.h> #include "super.h"#include "mds_client.h"#include "cache.h"#include "metric.h"#include "subvolume_metrics.h"#include "crypto.h"#include <linux/ceph/osd_client.h>#include <linux/ceph/striper.h> /* * Ceph address space ops. * * There are a few funny things going on here. * * The page->private field is used to reference a struct * ceph_snap_context for _every_ dirty page. This indicates which * snapshot the page was logically dirtied in, and thus which snap * context needs to be associated with the osd write during writeback. * * Similarly, struct ceph_inode_info maintains a set of counters to * count dirty pages on the inode. In the absence of snapshots, * i_wrbuffer_ref == i_wrbuffer_ref_head == the dirty page count. * * When a snapshot is taken (that is, when the client receives * notification that a snapshot was taken), each inode with caps and * with dirty pages (dirty pages implies there is a cap) gets a new * ceph_cap_snap in the i_cap_snaps list (which is sorted in ascending * order, new snaps go to the tail). The i_wrbuffer_ref_head count is * moved to capsnap->dirty. (Unless a sync write is currently in * progress. In that case, the capsnap is said to be "pending", new * writes cannot start, and the capsnap isn't "finalized" until the * write completes (or fails) and a final size/mtime for the inode for * that snap can be settled upon.) i_wrbuffer_ref_head is reset to 0. * * On writeback, we must submit writes to the osd IN SNAP ORDER. So, * we look for the first capsnap in i_cap_snaps and write out pages in * that snap context _only_. Then we move on to the next capsnap, * eventually reaching the "live" or "head" context (i.e., pages that * are not yet snapped) and are writing the most recently dirtied * pages. * * Invalidate and so forth must take care to ensure the dirty page * accounting is preserved. */ #define CONGESTION_ON_THRESH(congestion_kb) (congestion_kb >> (PAGE_SHIFT-10))#define CONGESTION_OFF_THRESH(congestion_kb) \ (CONGESTION_ON_THRESH(congestion_kb) - \ (CONGESTION_ON_THRESH(congestion_kb) >> 2)) static int ceph_netfs_check_write_begin(struct file *file, loff_t pos, unsigned int len, struct folio **foliop, void **_fsdata); static inline struct ceph_snap_context *page_snap_context(struct page *page){ if (PagePrivate(page)) return (void *)page->private; return NULL;} /* * Dirty a page. Optimistically adjust accounting, on the assumption * that we won't race with invalidate. If we do, readjust. */static bool ceph_dirty_folio(struct address_space *mapping, struct folio *folio){ struct inode *inode = mapping->host; struct ceph_client *cl = ceph_inode_to_client(inode); struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); struct ceph_inode_info *ci; struct ceph_snap_context *snapc; if (folio_test_dirty(folio)) { doutc(cl, "%llx.%llx %p idx %lu -- already dirty\n", ceph_vinop(inode), folio, folio->index); VM_BUG_ON_FOLIO(!folio_test_private(folio), folio); return false; } atomic64_inc(&mdsc->dirty_folios); ci = ceph_inode(inode); /* dirty the head */ spin_lock(&ci->i_ceph_lock); if (__ceph_have_pending_cap_snap(ci)) { struct ceph_cap_snap *capsnap = list_last_entry(&ci->i_cap_snaps, struct ceph_cap_snap, ci_item); snapc = ceph_get_snap_context(capsnap->context); capsnap->dirty_pages++; } else { BUG_ON(!ci->i_head_snapc); snapc = ceph_get_snap_context(ci->i_head_snapc); ++ci->i_wrbuffer_ref_head; } if (ci->i_wrbuffer_ref == 0) ihold(inode); ++ci->i_wrbuffer_ref; doutc(cl, "%llx.%llx %p idx %lu head %d/%d -> %d/%d " "snapc %p seq %lld (%d snaps)\n", ceph_vinop(inode), folio, folio->index, ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref_head-1, ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head, snapc, snapc->seq, snapc->num_snaps); spin_unlock(&ci->i_ceph_lock); /* * Reference snap context in folio->private. Also set * PagePrivate so that we get invalidate_folio callback. */ VM_WARN_ON_FOLIO(folio->private, folio); folio_attach_private(folio, snapc); return ceph_fscache_dirty_folio(mapping, folio);} /* * If we are truncating the full folio (i.e. offset == 0), adjust the * dirty folio counters appropriately. Only called if there is private * data on the folio. */static void ceph_invalidate_folio(struct folio *folio, size_t offset, size_t length){ struct inode *inode = folio->mapping->host; struct ceph_client *cl = ceph_inode_to_client(inode); struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_snap_context *snapc; if (offset != 0 || length != folio_size(folio)) { doutc(cl, "%llx.%llx idx %lu partial dirty page %zu~%zu\n", ceph_vinop(inode), folio->index, offset, length); return; } WARN_ON(!folio_test_locked(folio)); if (folio_test_private(folio)) { doutc(cl, "%llx.%llx idx %lu full dirty page\n", ceph_vinop(inode), folio->index); snapc = folio_detach_private(folio); ceph_put_wrbuffer_cap_refs(ci, 1, snapc); ceph_put_snap_context(snapc); } netfs_invalidate_folio(folio, offset, length);} static void ceph_netfs_expand_readahead(struct netfs_io_request *rreq){ struct inode *inode = rreq->inode; struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_file_layout *lo = &ci->i_layout; unsigned long max_pages = inode->i_sb->s_bdi->ra_pages; loff_t end = rreq->start + rreq->len, new_end; struct ceph_netfs_request_data *priv = rreq->netfs_priv; unsigned long max_len; u32 blockoff; if (priv) { /* Readahead is disabled by posix_fadvise POSIX_FADV_RANDOM */ if (priv->file_ra_disabled) max_pages = 0; else max_pages = priv->file_ra_pages; } /* Readahead is disabled */ if (!max_pages) return; max_len = max_pages << PAGE_SHIFT; /* * Try to expand the length forward by rounding up it to the next * block, but do not exceed the file size, unless the original * request already exceeds it. */ new_end = umin(round_up(end, lo->stripe_unit), rreq->i_size); if (new_end > end && new_end <= rreq->start + max_len) rreq->len = new_end - rreq->start; /* Try to expand the start downward */ div_u64_rem(rreq->start, lo->stripe_unit, &blockoff); if (rreq->len + blockoff <= max_len) { rreq->start -= blockoff; rreq->len += blockoff; }} static void finish_netfs_read(struct ceph_osd_request *req){ struct inode *inode = req->r_inode; struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_client *cl = fsc->client; struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0); struct netfs_io_subrequest *subreq = req->r_priv; struct ceph_osd_req_op *op = &req->r_ops[0]; int err = req->r_result; bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ); ceph_update_read_metrics(&fsc->mdsc->metric, req->r_start_latency, req->r_end_latency, osd_data->length, err); doutc(cl, "result %d subreq->len=%zu i_size=%lld\n", req->r_result, subreq->len, i_size_read(req->r_inode)); /* no object means success but no data */ if (err == -ENOENT) { __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); __set_bit(NETFS_SREQ_MADE_PROGRESS, &subreq->flags); err = 0; } else if (err == -EBLOCKLISTED) { fsc->blocklisted = true; } if (err >= 0) { if (sparse && err > 0) err = ceph_sparse_ext_map_end(op); if (err < subreq->len && subreq->rreq->origin != NETFS_UNBUFFERED_READ && subreq->rreq->origin != NETFS_DIO_READ) __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); if (IS_ENCRYPTED(inode) && err > 0) { err = ceph_fscrypt_decrypt_extents(inode, osd_data->pages, subreq->start, op->extent.sparse_ext, op->extent.sparse_ext_cnt); if (err > subreq->len) err = subreq->len; } if (err > 0) __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); } if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) { int num_pages = calc_pages_for(osd_data->alignment, osd_data->length); for (int i = 0; i < num_pages; i++) put_page(osd_data->pages[i]); kvfree(osd_data->pages); } if (err > 0) { ceph_subvolume_metrics_record_io(fsc->mdsc, ceph_inode(inode), false, err, req->r_start_latency, req->r_end_latency); subreq->transferred = err; err = 0; } subreq->error = err; trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress); netfs_read_subreq_terminated(subreq); iput(req->r_inode); ceph_dec_osd_stopping_blocker(fsc->mdsc);} static bool ceph_netfs_issue_op_inline(struct netfs_io_subrequest *subreq){ struct netfs_io_request *rreq = subreq->rreq; struct inode *inode = rreq->inode; struct ceph_mds_reply_info_parsed *rinfo; struct ceph_mds_reply_info_in *iinfo; struct ceph_mds_request *req; struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); struct ceph_inode_info *ci = ceph_inode(inode); ssize_t err = 0; size_t len; int mode; if (rreq->origin != NETFS_UNBUFFERED_READ && rreq->origin != NETFS_DIO_READ) __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); __clear_bit(NETFS_SREQ_COPY_TO_CACHE, &subreq->flags); if (subreq->start >= inode->i_size) goto out; /* We need to fetch the inline data. */ mode = ceph_try_to_choose_auth_mds(inode, CEPH_STAT_CAP_INLINE_DATA); req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, mode); if (IS_ERR(req)) { err = PTR_ERR(req); goto out; } req->r_ino1 = ci->i_vino; req->r_args.getattr.mask = cpu_to_le32(CEPH_STAT_CAP_INLINE_DATA); req->r_num_caps = 2; trace_netfs_sreq(subreq, netfs_sreq_trace_submit); err = ceph_mdsc_do_request(mdsc, NULL, req); if (err < 0) goto out; rinfo = &req->r_reply_info; iinfo = &rinfo->targeti; if (iinfo->inline_version == CEPH_INLINE_NONE) { /* The data got uninlined */ ceph_mdsc_put_request(req); return false; } len = min_t(size_t, iinfo->inline_len - subreq->start, subreq->len); err = copy_to_iter(iinfo->inline_data + subreq->start, len, &subreq->io_iter); if (err == 0) { err = -EFAULT; } else { subreq->transferred += err; err = 0; } ceph_mdsc_put_request(req);out: subreq->error = err; trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress); netfs_read_subreq_terminated(subreq); return true;} static int ceph_netfs_prepare_read(struct netfs_io_subrequest *subreq){ struct netfs_io_request *rreq = subreq->rreq; struct inode *inode = rreq->inode; struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); u64 objno, objoff; u32 xlen; /* Truncate the extent at the end of the current block */ ceph_calc_file_object_mapping(&ci->i_layout, subreq->start, subreq->len, &objno, &objoff, &xlen); rreq->io_streams[0].sreq_max_len = umin(xlen, fsc->mount_options->rsize); return 0;} static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq){ struct netfs_io_request *rreq = subreq->rreq; struct inode *inode = rreq->inode; struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_client *cl = fsc->client; struct ceph_osd_request *req = NULL; struct ceph_vino vino = ceph_vino(inode); int err; u64 len; bool sparse = IS_ENCRYPTED(inode) || ceph_test_mount_opt(fsc, SPARSEREAD); u64 off = subreq->start; int extent_cnt; if (ceph_inode_is_shutdown(inode)) { err = -EIO; goto out; } if (ceph_has_inline_data(ci) && ceph_netfs_issue_op_inline(subreq)) return; // TODO: This rounding here is slightly dodgy. It *should* work, for // now, as the cache only deals in blocks that are a multiple of // PAGE_SIZE and fscrypt blocks are at most PAGE_SIZE. What needs to // happen is for the fscrypt driving to be moved into netfslib and the // data in the cache also to be stored encrypted. len = subreq->len; ceph_fscrypt_adjust_off_and_len(inode, &off, &len); req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, vino, off, &len, 0, 1, sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ, NULL, ci->i_truncate_seq, ci->i_truncate_size, false); if (IS_ERR(req)) { err = PTR_ERR(req); req = NULL; goto out; } if (sparse) { extent_cnt = __ceph_sparse_read_ext_count(inode, len); err = ceph_alloc_sparse_ext_map(&req->r_ops[0], extent_cnt); if (err) goto out; } doutc(cl, "%llx.%llx pos=%llu orig_len=%zu len=%llu\n", ceph_vinop(inode), subreq->start, subreq->len, len); /* * FIXME: For now, use CEPH_OSD_DATA_TYPE_PAGES instead of _ITER for * encrypted inodes. We'd need infrastructure that handles an iov_iter * instead of page arrays, and we don't have that as of yet. Once the * dust settles on the write helpers and encrypt/decrypt routines for * netfs, we should be able to rework this. */ if (IS_ENCRYPTED(inode)) { struct page **pages; size_t page_off; /* * FIXME: io_iter.count needs to be corrected to aligned * length. Otherwise, iov_iter_get_pages_alloc2() operates * with the initial unaligned length value. As a result, * ceph_msg_data_cursor_init() triggers BUG_ON() in the case * if msg->sparse_read_total > msg->data_length. */ subreq->io_iter.count = len; err = iov_iter_get_pages_alloc2(&subreq->io_iter, &pages, len, &page_off); if (err < 0) { doutc(cl, "%llx.%llx failed to allocate pages, %d\n", ceph_vinop(inode), err); goto out; } /* should always give us a page-aligned read */ WARN_ON_ONCE(page_off); len = err; err = 0; osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false, false); } else { osd_req_op_extent_osd_iter(req, 0, &subreq->io_iter); } if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) { err = -EIO; goto out; } req->r_callback = finish_netfs_read; req->r_priv = subreq; req->r_inode = inode; ihold(inode); trace_netfs_sreq(subreq, netfs_sreq_trace_submit); ceph_osdc_start_request(req->r_osdc, req);out: ceph_osdc_put_request(req); if (err) { subreq->error = err; netfs_read_subreq_terminated(subreq); } doutc(cl, "%llx.%llx result %d\n", ceph_vinop(inode), err);} static int ceph_init_request(struct netfs_io_request *rreq, struct file *file){ struct inode *inode = rreq->inode; struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_client *cl = ceph_inode_to_client(inode); int got = 0, want = CEPH_CAP_FILE_CACHE; struct ceph_netfs_request_data *priv; int ret = 0; /* [DEPRECATED] Use PG_private_2 to mark folio being written to the cache. */ __set_bit(NETFS_RREQ_USE_PGPRIV2, &rreq->flags); if (rreq->origin != NETFS_READAHEAD) return 0; priv = kzalloc_obj(*priv, GFP_NOFS); if (!priv) return -ENOMEM; if (file) { struct ceph_rw_context *rw_ctx; struct ceph_file_info *fi = file->private_data; priv->file_ra_pages = file->f_ra.ra_pages; priv->file_ra_disabled = file->f_mode & FMODE_RANDOM; rw_ctx = ceph_find_rw_context(fi); if (rw_ctx) { rreq->netfs_priv = priv; return 0; } } /* * readahead callers do not necessarily hold Fcb capsLines 1–500 of 2658. Load more
// SPDX-License-Identifier: GPL-2.0 #include <linux/ceph/ceph_debug.h> #include <linux/backing-dev.h> #include <linux/fs.h> #include <linux/mm.h> #include <linux/swap.h> #include <linux/pagemap.h> #include <linux/slab.h> #include <linux/folio_batch.h> #include <linux/task_io_accounting_ops.h> #include <linux/signal.h> #include <linux/iversion.h> #include <linux/ktime.h> #include <linux/netfs.h> #include <trace/events/netfs.h> #include "super.h" #include "mds_client.h" #include "cache.h" #include "metric.h" #include "subvolume_metrics.h" #include "crypto.h" #include <linux/ceph/osd_client.h> #include <linux/ceph/striper.h> /* * Ceph address space ops. * * There are a few funny things going on here. * * The page->private field is used to reference a struct * ceph_snap_context for _every_ dirty page. This indicates which * snapshot the page was logically dirtied in, and thus which snap * context needs to be associated with the osd write during writeback. * * Similarly, struct ceph_inode_info maintains a set of counters to * count dirty pages on the inode. In the absence of snapshots, * i_wrbuffer_ref == i_wrbuffer_ref_head == the dirty page count. * * When a snapshot is taken (that is, when the client receives * notification that a snapshot was taken), each inode with caps and * with dirty pages (dirty pages implies there is a cap) gets a new * ceph_cap_snap in the i_cap_snaps list (which is sorted in ascending * order, new snaps go to the tail). The i_wrbuffer_ref_head count is * moved to capsnap->dirty. (Unless a sync write is currently in * progress. In that case, the capsnap is said to be "pending", new * writes cannot start, and the capsnap isn't "finalized" until the * write completes (or fails) and a final size/mtime for the inode for * that snap can be settled upon.) i_wrbuffer_ref_head is reset to 0. * * On writeback, we must submit writes to the osd IN SNAP ORDER. So, * we look for the first capsnap in i_cap_snaps and write out pages in * that snap context _only_. Then we move on to the next capsnap, * eventually reaching the "live" or "head" context (i.e., pages that * are not yet snapped) and are writing the most recently dirtied * pages. * * Invalidate and so forth must take care to ensure the dirty page * accounting is preserved. */ #define CONGESTION_ON_THRESH(congestion_kb) (congestion_kb >> (PAGE_SHIFT-10)) #define CONGESTION_OFF_THRESH(congestion_kb) \ (CONGESTION_ON_THRESH(congestion_kb) - \ (CONGESTION_ON_THRESH(congestion_kb) >> 2)) static int ceph_netfs_check_write_begin(struct file *file, loff_t pos, unsigned int len, struct folio **foliop, void **_fsdata); static inline struct ceph_snap_context *page_snap_context(struct page *page) { if (PagePrivate(page)) return (void *)page->private; return NULL; } /* * Dirty a page. Optimistically adjust accounting, on the assumption * that we won't race with invalidate. If we do, readjust. */ static bool ceph_dirty_folio(struct address_space *mapping, struct folio *folio) { struct inode *inode = mapping->host; struct ceph_client *cl = ceph_inode_to_client(inode); struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); struct ceph_inode_info *ci; struct ceph_snap_context *snapc; if (folio_test_dirty(folio)) { doutc(cl, "%llx.%llx %p idx %lu -- already dirty\n", ceph_vinop(inode), folio, folio->index); VM_BUG_ON_FOLIO(!folio_test_private(folio), folio); return false; } atomic64_inc(&mdsc->dirty_folios); ci = ceph_inode(inode); /* dirty the head */ spin_lock(&ci->i_ceph_lock); if (__ceph_have_pending_cap_snap(ci)) { struct ceph_cap_snap *capsnap = list_last_entry(&ci->i_cap_snaps, struct ceph_cap_snap, ci_item); snapc = ceph_get_snap_context(capsnap->context); capsnap->dirty_pages++; } else { BUG_ON(!ci->i_head_snapc); snapc = ceph_get_snap_context(ci->i_head_snapc); ++ci->i_wrbuffer_ref_head; } if (ci->i_wrbuffer_ref == 0) ihold(inode); ++ci->i_wrbuffer_ref; doutc(cl, "%llx.%llx %p idx %lu head %d/%d -> %d/%d " "snapc %p seq %lld (%d snaps)\n", ceph_vinop(inode), folio, folio->index, ci->i_wrbuffer_ref-1, ci->i_wrbuffer_ref_head-1, ci->i_wrbuffer_ref, ci->i_wrbuffer_ref_head, snapc, snapc->seq, snapc->num_snaps); spin_unlock(&ci->i_ceph_lock); /* * Reference snap context in folio->private. Also set * PagePrivate so that we get invalidate_folio callback. */ VM_WARN_ON_FOLIO(folio->private, folio); folio_attach_private(folio, snapc); return ceph_fscache_dirty_folio(mapping, folio); } /* * If we are truncating the full folio (i.e. offset == 0), adjust the * dirty folio counters appropriately. Only called if there is private * data on the folio. */ static void ceph_invalidate_folio(struct folio *folio, size_t offset, size_t length) { struct inode *inode = folio->mapping->host; struct ceph_client *cl = ceph_inode_to_client(inode); struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_snap_context *snapc; if (offset != 0 || length != folio_size(folio)) { doutc(cl, "%llx.%llx idx %lu partial dirty page %zu~%zu\n", ceph_vinop(inode), folio->index, offset, length); return; } WARN_ON(!folio_test_locked(folio)); if (folio_test_private(folio)) { doutc(cl, "%llx.%llx idx %lu full dirty page\n", ceph_vinop(inode), folio->index); snapc = folio_detach_private(folio); ceph_put_wrbuffer_cap_refs(ci, 1, snapc); ceph_put_snap_context(snapc); } netfs_invalidate_folio(folio, offset, length); } static void ceph_netfs_expand_readahead(struct netfs_io_request *rreq) { struct inode *inode = rreq->inode; struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_file_layout *lo = &ci->i_layout; unsigned long max_pages = inode->i_sb->s_bdi->ra_pages; loff_t end = rreq->start + rreq->len, new_end; struct ceph_netfs_request_data *priv = rreq->netfs_priv; unsigned long max_len; u32 blockoff; if (priv) { /* Readahead is disabled by posix_fadvise POSIX_FADV_RANDOM */ if (priv->file_ra_disabled) max_pages = 0; else max_pages = priv->file_ra_pages; } /* Readahead is disabled */ if (!max_pages) return; max_len = max_pages << PAGE_SHIFT; /* * Try to expand the length forward by rounding up it to the next * block, but do not exceed the file size, unless the original * request already exceeds it. */ new_end = umin(round_up(end, lo->stripe_unit), rreq->i_size); if (new_end > end && new_end <= rreq->start + max_len) rreq->len = new_end - rreq->start; /* Try to expand the start downward */ div_u64_rem(rreq->start, lo->stripe_unit, &blockoff); if (rreq->len + blockoff <= max_len) { rreq->start -= blockoff; rreq->len += blockoff; } } static void finish_netfs_read(struct ceph_osd_request *req) { struct inode *inode = req->r_inode; struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_client *cl = fsc->client; struct ceph_osd_data *osd_data = osd_req_op_extent_osd_data(req, 0); struct netfs_io_subrequest *subreq = req->r_priv; struct ceph_osd_req_op *op = &req->r_ops[0]; int err = req->r_result; bool sparse = (op->op == CEPH_OSD_OP_SPARSE_READ); ceph_update_read_metrics(&fsc->mdsc->metric, req->r_start_latency, req->r_end_latency, osd_data->length, err); doutc(cl, "result %d subreq->len=%zu i_size=%lld\n", req->r_result, subreq->len, i_size_read(req->r_inode)); /* no object means success but no data */ if (err == -ENOENT) { __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); __set_bit(NETFS_SREQ_MADE_PROGRESS, &subreq->flags); err = 0; } else if (err == -EBLOCKLISTED) { fsc->blocklisted = true; } if (err >= 0) { if (sparse && err > 0) err = ceph_sparse_ext_map_end(op); if (err < subreq->len && subreq->rreq->origin != NETFS_UNBUFFERED_READ && subreq->rreq->origin != NETFS_DIO_READ) __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); if (IS_ENCRYPTED(inode) && err > 0) { err = ceph_fscrypt_decrypt_extents(inode, osd_data->pages, subreq->start, op->extent.sparse_ext, op->extent.sparse_ext_cnt); if (err > subreq->len) err = subreq->len; } if (err > 0) __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); } if (osd_data->type == CEPH_OSD_DATA_TYPE_PAGES) { int num_pages = calc_pages_for(osd_data->alignment, osd_data->length); for (int i = 0; i < num_pages; i++) put_page(osd_data->pages[i]); kvfree(osd_data->pages); } if (err > 0) { ceph_subvolume_metrics_record_io(fsc->mdsc, ceph_inode(inode), false, err, req->r_start_latency, req->r_end_latency); subreq->transferred = err; err = 0; } subreq->error = err; trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress); netfs_read_subreq_terminated(subreq); iput(req->r_inode); ceph_dec_osd_stopping_blocker(fsc->mdsc); } static bool ceph_netfs_issue_op_inline(struct netfs_io_subrequest *subreq) { struct netfs_io_request *rreq = subreq->rreq; struct inode *inode = rreq->inode; struct ceph_mds_reply_info_parsed *rinfo; struct ceph_mds_reply_info_in *iinfo; struct ceph_mds_request *req; struct ceph_mds_client *mdsc = ceph_sb_to_mdsc(inode->i_sb); struct ceph_inode_info *ci = ceph_inode(inode); ssize_t err = 0; size_t len; int mode; if (rreq->origin != NETFS_UNBUFFERED_READ && rreq->origin != NETFS_DIO_READ) __set_bit(NETFS_SREQ_CLEAR_TAIL, &subreq->flags); __clear_bit(NETFS_SREQ_COPY_TO_CACHE, &subreq->flags); if (subreq->start >= inode->i_size) goto out; /* We need to fetch the inline data. */ mode = ceph_try_to_choose_auth_mds(inode, CEPH_STAT_CAP_INLINE_DATA); req = ceph_mdsc_create_request(mdsc, CEPH_MDS_OP_GETATTR, mode); if (IS_ERR(req)) { err = PTR_ERR(req); goto out; } req->r_ino1 = ci->i_vino; req->r_args.getattr.mask = cpu_to_le32(CEPH_STAT_CAP_INLINE_DATA); req->r_num_caps = 2; trace_netfs_sreq(subreq, netfs_sreq_trace_submit); err = ceph_mdsc_do_request(mdsc, NULL, req); if (err < 0) goto out; rinfo = &req->r_reply_info; iinfo = &rinfo->targeti; if (iinfo->inline_version == CEPH_INLINE_NONE) { /* The data got uninlined */ ceph_mdsc_put_request(req); return false; } len = min_t(size_t, iinfo->inline_len - subreq->start, subreq->len); err = copy_to_iter(iinfo->inline_data + subreq->start, len, &subreq->io_iter); if (err == 0) { err = -EFAULT; } else { subreq->transferred += err; err = 0; } ceph_mdsc_put_request(req); out: subreq->error = err; trace_netfs_sreq(subreq, netfs_sreq_trace_io_progress); netfs_read_subreq_terminated(subreq); return true; } static int ceph_netfs_prepare_read(struct netfs_io_subrequest *subreq) { struct netfs_io_request *rreq = subreq->rreq; struct inode *inode = rreq->inode; struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); u64 objno, objoff; u32 xlen; /* Truncate the extent at the end of the current block */ ceph_calc_file_object_mapping(&ci->i_layout, subreq->start, subreq->len, &objno, &objoff, &xlen); rreq->io_streams[0].sreq_max_len = umin(xlen, fsc->mount_options->rsize); return 0; } static void ceph_netfs_issue_read(struct netfs_io_subrequest *subreq) { struct netfs_io_request *rreq = subreq->rreq; struct inode *inode = rreq->inode; struct ceph_inode_info *ci = ceph_inode(inode); struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_client *cl = fsc->client; struct ceph_osd_request *req = NULL; struct ceph_vino vino = ceph_vino(inode); int err; u64 len; bool sparse = IS_ENCRYPTED(inode) || ceph_test_mount_opt(fsc, SPARSEREAD); u64 off = subreq->start; int extent_cnt; if (ceph_inode_is_shutdown(inode)) { err = -EIO; goto out; } if (ceph_has_inline_data(ci) && ceph_netfs_issue_op_inline(subreq)) return; // TODO: This rounding here is slightly dodgy. It *should* work, for // now, as the cache only deals in blocks that are a multiple of // PAGE_SIZE and fscrypt blocks are at most PAGE_SIZE. What needs to // happen is for the fscrypt driving to be moved into netfslib and the // data in the cache also to be stored encrypted. len = subreq->len; ceph_fscrypt_adjust_off_and_len(inode, &off, &len); req = ceph_osdc_new_request(&fsc->client->osdc, &ci->i_layout, vino, off, &len, 0, 1, sparse ? CEPH_OSD_OP_SPARSE_READ : CEPH_OSD_OP_READ, CEPH_OSD_FLAG_READ, NULL, ci->i_truncate_seq, ci->i_truncate_size, false); if (IS_ERR(req)) { err = PTR_ERR(req); req = NULL; goto out; } if (sparse) { extent_cnt = __ceph_sparse_read_ext_count(inode, len); err = ceph_alloc_sparse_ext_map(&req->r_ops[0], extent_cnt); if (err) goto out; } doutc(cl, "%llx.%llx pos=%llu orig_len=%zu len=%llu\n", ceph_vinop(inode), subreq->start, subreq->len, len); /* * FIXME: For now, use CEPH_OSD_DATA_TYPE_PAGES instead of _ITER for * encrypted inodes. We'd need infrastructure that handles an iov_iter * instead of page arrays, and we don't have that as of yet. Once the * dust settles on the write helpers and encrypt/decrypt routines for * netfs, we should be able to rework this. */ if (IS_ENCRYPTED(inode)) { struct page **pages; size_t page_off; /* * FIXME: io_iter.count needs to be corrected to aligned * length. Otherwise, iov_iter_get_pages_alloc2() operates * with the initial unaligned length value. As a result, * ceph_msg_data_cursor_init() triggers BUG_ON() in the case * if msg->sparse_read_total > msg->data_length. */ subreq->io_iter.count = len; err = iov_iter_get_pages_alloc2(&subreq->io_iter, &pages, len, &page_off); if (err < 0) { doutc(cl, "%llx.%llx failed to allocate pages, %d\n", ceph_vinop(inode), err); goto out; } /* should always give us a page-aligned read */ WARN_ON_ONCE(page_off); len = err; err = 0; osd_req_op_extent_osd_data_pages(req, 0, pages, len, 0, false, false); } else { osd_req_op_extent_osd_iter(req, 0, &subreq->io_iter); } if (!ceph_inc_osd_stopping_blocker(fsc->mdsc)) { err = -EIO; goto out; } req->r_callback = finish_netfs_read; req->r_priv = subreq; req->r_inode = inode; ihold(inode); trace_netfs_sreq(subreq, netfs_sreq_trace_submit); ceph_osdc_start_request(req->r_osdc, req); out: ceph_osdc_put_request(req); if (err) { subreq->error = err; netfs_read_subreq_terminated(subreq); } doutc(cl, "%llx.%llx result %d\n", ceph_vinop(inode), err); } static int ceph_init_request(struct netfs_io_request *rreq, struct file *file) { struct inode *inode = rreq->inode; struct ceph_fs_client *fsc = ceph_inode_to_fs_client(inode); struct ceph_client *cl = ceph_inode_to_client(inode); int got = 0, want = CEPH_CAP_FILE_CACHE; struct ceph_netfs_request_data *priv; int ret = 0; /* [DEPRECATED] Use PG_private_2 to mark folio being written to the cache. */ __set_bit(NETFS_RREQ_USE_PGPRIV2, &rreq->flags); if (rreq->origin != NETFS_READAHEAD) return 0; priv = kzalloc_obj(*priv, GFP_NOFS); if (!priv) return -ENOMEM; if (file) { struct ceph_rw_context *rw_ctx; struct ceph_file_info *fi = file->private_data; priv->file_ra_pages = file->f_ra.ra_pages; priv->file_ra_disabled = file->f_mode & FMODE_RANDOM; rw_ctx = ceph_find_rw_context(fi); if (rw_ctx) { rreq->netfs_priv = priv; return 0; } } /* * readahead callers do not necessarily hold Fcb caps