[PATCH 8/8] rpdfs: add read_folio, dirty_folio, write_begin, write_end

This commit completes basic file data read/write support. Currently it
will deadlock with multiple clients, so much more work is needed.

Signed-off-by: Valerie Aurora <[email protected]>
---
 fs/rpdfs/data.c  | 190 +++++++++++++++++++++++++++++++++++++++++++++++
 fs/rpdfs/inode.c |   1 +
 2 files changed, 191 insertions(+)

diff --git a/fs/rpdfs/data.c b/fs/rpdfs/data.c
index 3a6e9560b32f..42fe5c8c3c34 100644
--- a/fs/rpdfs/data.c
+++ b/fs/rpdfs/data.c
@@ -308,3 +308,193 @@ static int get_or_alloc_file_block(struct rpdfs_fs_info *rfi, struct rpdfs_trans
 		  ret, rpdfs_inode_ino(inode), lblk, bnr, *hnd_ret);
 	return ret;
 }
+
+/*
+ * We have to avoid a potential deadlock between the kernel's lock on
+ * each page and our cache consistency algorithm. The order of
+ * acquisition on a read or write or similar operation is:
+ *
+ * 1. kernel grabs page lock, calls file system routine
+ * 2. local node attempts to get access to a block
+ *
+ * But invalidation of a page has this order:
+ *
+ * 1. remote node requests exclusive access to a block
+ * 2. local node receives cache invalidate message and tries to get page lock
+ *
+ * If the local node is trying to get access to a block while another
+ * node tries to get exclusive access, we could end up with:
+ *
+ * 1. local node holds page lock, can't get access to block
+ * 2. local node attempts to service invalidate request but can't get page lock
+ *
+ * The solution is to drop the page lock while getting access to blocks,
+ * then re-acquire the lock and handle any changes that occurred while
+ * it was unlocked. Generally this looks like returning
+ * AOP_TRUNCATED_PAGE if the page is gone.
+ *
+ * Note that readahead() should satisfy most read requests
+ * asynchronously, leaving us to do any remaining synchronous requests
+ * in read_folio().
+ */
+
+/*
+ * TODO: cannot block on block acquisition while holding the folio lock.
+ */
+static int rpdfs_read_folio(struct file *file, struct folio *folio)
+{
+	struct inode *inode = folio->mapping->host;
+	struct rpdfs_fs_info *rfi = RPDFS_INODE_FS(inode);
+	loff_t pos = folio_pos(folio);
+	size_t len = folio_size(folio);
+	struct rpdfs_block_handle *inode_hnd = NULL;
+	struct rpdfs_block_handle *blk_hnd = NULL;
+	u64 lblk;
+	int ret;
+
+	lblk = lblk_from_offset(pos);
+
+	rpdfs_prd("ino %llu lblk %llu pos %lld len %lu",
+		  rpdfs_inode_ino(inode), lblk, pos, len);
+
+	/* we turn off atime always, inode will not be written */
+	ret = rpdfs_inode_acquire(rfi, NULL, inode, &inode_hnd, 0);
+	if (ret < 0)
+		goto out;
+
+	ret = get_or_alloc_file_block(rfi, NULL, inode, inode_hnd, lblk, 0, &blk_hnd);
+	if (ret < 0)
+		goto out;
+
+	/* copy the data into the actual block */
+	if (blk_hnd)
+		memcpy(folio_address(folio), blk_hnd->data, len);
+	else
+		memset(folio_address(folio), 0, len);
+
+	folio_mark_uptodate(folio);
+out:
+	folio_unlock(folio);
+
+	rpdfs_block_release(rfi, &inode_hnd);
+	rpdfs_block_release(rfi, &blk_hnd);
+
+	rpdfs_prd("ret %d", ret);
+
+	return ret;
+}
+
+static bool rpdfs_dirty_folio(struct address_space *mapping, struct folio *folio)
+{
+	rpdfs_prd("ino %lu index %lu", mapping->host->i_ino, folio->index);
+
+	return filemap_dirty_folio(folio_mapping(folio), folio);
+}
+
+/*
+ * Do whatever preparation is necessary before the VFS copies the data
+ * from the user buffer into the folio pages. Must return a folio.
+ *
+ * Called before a write starts with the page lock held. Due to this, we
+ * can't do anything that blocks on acquisition of a block. Right now it
+ * just gets the folio. In the future it might check quotas, file system
+ * error state, etc.
+ */
+static int rpdfs_write_begin(struct file *file, struct address_space *mapping,
+			     loff_t pos, unsigned len,
+			     struct folio **foliop, void **fsdata)
+{
+	struct folio *folio;
+	int ret;
+
+	rpdfs_prd("ino %lu pos %lld len %u", mapping->host->i_ino, pos, len);
+
+	folio = __filemap_get_folio(mapping, pos >> PAGE_SHIFT, FGP_WRITEBEGIN,
+				    mapping_gfp_mask(mapping));
+	if (IS_ERR(folio)) {
+		ret = PTR_ERR(folio);
+		goto out;
+	}
+
+	*foliop = folio;
+	ret = 0;
+out:
+	return ret;
+}
+
+/*
+ * Called when the data for a write has been copied in memory, with page
+ * lock held. Here we actually do the write.
+ *
+ * To avoid deadlocking with another thread/node that has exclusive
+ * access to a block we need and is also attempting to get the page lock
+ * on this page, we first do a non-blocking lookup of the blocks we
+ * need. If that fails, we drop the page lock, do a blocking
+ * acquisition, and when that succeeds, return AOP_TRUNCATE_PAGE to
+ * request a retry.
+ */
+static int rpdfs_write_end(struct file *file, struct address_space *mapping,
+			   loff_t pos, unsigned len, unsigned copied,
+			   struct folio *folio, void *fsdata)
+{
+	struct inode *inode = folio->mapping->host;
+	struct rpdfs_fs_info *rfi = RPDFS_INODE_FS(inode);
+	struct rpdfs_inode_info *ri = RPDFS_I(inode);
+	struct rpdfs_block_handle *inode_hnd = NULL;
+	struct rpdfs_block_handle *blk_hnd = NULL;
+	struct rpdfs_transaction txn = RPDFS_INIT_TXN;
+	rbaf_t rbaf;
+	u64 lblk;
+	int ret;
+	unsigned offset = offset_in_folio(folio, pos);
+
+	rpdfs_prd("ino %llu pos %lld len %u offset %u", ri->ig.ino, pos, len, offset);
+
+	lblk = lblk_from_offset(pos);
+
+	/* XXX use pos/len to figure out when it is an overwrite */
+	rbaf = RBAF_WRITE;
+
+	ret = rpdfs_inode_acquire(rfi, &txn, inode, &inode_hnd, RBAF_WRITE);
+	if (ret < 0)
+		goto out;
+
+	ret = get_or_alloc_file_block(rfi, &txn, inode, inode_hnd, lblk, rbaf, &blk_hnd);
+	if (ret < 0)
+		goto out;
+
+	/* copy the data into the actual block */
+	memcpy(blk_hnd->data + offset, folio_address(folio) + offset, len);
+
+	rpdfs_prd("copied %c len %d to %c", *((char *) folio_address(folio) + offset), len,
+		  *((char *) blk_hnd->data + offset));
+
+	if ((pos + len) > inode->i_size)
+		i_size_write(inode, pos + len);
+
+	folio_mark_uptodate(folio);
+	folio_mark_dirty(folio);
+	folio_unlock(folio);
+
+	file_update_time(file);
+	mark_inode_dirty(inode); /* TODO: not always necessary */
+
+	/* finalize changes to the inode and block */
+	rpdfs_inode_update(rfi, inode, inode_hnd);
+	rpdfs_block_release(rfi, &inode_hnd);
+	rpdfs_block_release(rfi, &blk_hnd);
+
+	ret = len;
+out:
+	rpdfs_txn_finish(rfi, &txn);
+
+	rpdfs_prd("i_size %lld ret %d", i_size_read(inode), ret);
+	return ret;
+}
+
+const struct address_space_operations rpdfs_aops = {
+	.read_folio =		rpdfs_read_folio,
+	.dirty_folio =		rpdfs_dirty_folio,
+	.write_begin =		rpdfs_write_begin,
+	.write_end =		rpdfs_write_end,
+};
diff --git a/fs/rpdfs/inode.c b/fs/rpdfs/inode.c
index 947a8f1703a3..79265d2716c5 100644
--- a/fs/rpdfs/inode.c
+++ b/fs/rpdfs/inode.c
@@ -164,6 +164,7 @@ void rpdfs_inode_init_ops(struct inode *inode)
 	case S_IFREG:
 		inode->i_op = &rpdfs_file_iops;
 		inode->i_fop = &rpdfs_file_fops;
+		inode->i_mapping->a_ops = &rpdfs_aops;
 		break;
 	case S_IFDIR:
 		inode->i_op = &rpdfs_dir_iops;
-- 
2.49.0