scoutfs: Cluster coherent read/write

With trylock implemented we can add locking in readpage. After that it's
pretty easy to implement our own read/write functions which at this
point are more or less wrapping the kernel helpers in the correct
cluster locking.

Data invalidation is a bit interesting. If the lock we are invalidating
is an inode group lock, we use the lock boundaries to incrementally
search our inode cache. When an inode struct is found, we sync and
(optionally) truncate pages.

Signed-off-by: Mark Fasheh <mfasheh@versity.com>
[zab: adapted to newer lock call, fixed some error handling]
Signed-off-by: Zach Brown <zab@versity.com>

kmod/src/Makefile

@@ -5,9 +5,9 @@ CFLAGS_super.o = -DSCOUTFS_GIT_DESCRIBE=\"$(SCOUTFS_GIT_DESCRIBE)\"
 CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include
  
 scoutfs-y += alloc.o bio.o btree.o client.o compact.o counters.o data.o dir.o \
-	     dlmglue.o kvec.o inode.o ioctl.o item.o key.o lock.o manifest.o \
-	     msg.o options.o seg.o server.o scoutfs_trace.o sock.o sort_priv.o \
-	     stackglue.o super.o trans.o xattr.o
+	     dlmglue.o file.o kvec.o inode.o ioctl.o item.o key.o lock.o \
+	     manifest.o msg.o options.o seg.o server.o scoutfs_trace.o sock.o \
+	     sort_priv.o stackglue.o super.o trans.o xattr.o
 
 #
 # The raw types aren't available in userspace headers.  Make sure all

kmod/src/data.c

@@ -31,6 +31,7 @@
 #include "ioctl.h"
 #include "client.h"
 #include "lock.h"
+#include "file.h"
 
 #define EXTF "[off %llu bno %llu bks %llu fl %x]"
 #define EXTA(ne) (ne)->blk_off, (ne)->blkno, (ne)->blocks, (ne)->flags
@@ -1070,13 +1071,47 @@ out:
 
 static int scoutfs_readpage(struct file *file, struct page *page)
 {
-	return mpage_readpage(page, scoutfs_get_block);
+	struct inode *inode = file->f_inode;
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
+	int unlock = 1;
+	int ret;
+
+	ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, SCOUTFS_LKF_REFRESH_INODE |
+				 SCOUTFS_LKF_TRYLOCK, inode, &inode_lock);
+	if (ret) {
+		if (ret == -EAGAIN)
+			ret = AOP_TRUNCATED_PAGE;
+		goto out;
+	}
+
+	ret = mpage_readpage(page, scoutfs_get_block);
+	unlock = 0;
+
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR);
+out:
+	if (unlock)
+		unlock_page(page);
+	return ret;
 }
 
 static int scoutfs_readpages(struct file *file, struct address_space *mapping,
 			     struct list_head *pages, unsigned nr_pages)
 {
-	return mpage_readpages(mapping, pages, nr_pages, scoutfs_get_block);
+	struct inode *inode = file->f_inode;
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
+	int ret;
+
+	ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, SCOUTFS_LKF_REFRESH_INODE,
+				 inode, &inode_lock);
+	if (ret)
+		return ret;
+
+	ret = mpage_readpages(mapping, pages, nr_pages, scoutfs_get_block);
+
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR);
+	return ret;
 }
 
 static int scoutfs_writepage(struct page *page, struct writeback_control *wbc)
@@ -1256,8 +1291,8 @@ const struct address_space_operations scoutfs_file_aops = {
 const struct file_operations scoutfs_file_fops = {
 	.read		= do_sync_read,
 	.write		= do_sync_write,
-	.aio_read	= generic_file_aio_read,
-	.aio_write	= generic_file_aio_write,
+	.aio_read	= scoutfs_file_aio_read,
+	.aio_write	= scoutfs_file_aio_write,
 	.unlocked_ioctl	= scoutfs_ioctl,
 	.fsync		= scoutfs_file_fsync,
 	.llseek		= generic_file_llseek,

kmod/src/file.c

@@ -0,0 +1,85 @@
+/*
+ * Copyright (C) 2017 Versity Software, Inc.  All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License v2 as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ */
+
+#include <linux/kernel.h>
+#include <linux/fs.h>
+#include <linux/slab.h>
+#include <linux/pagemap.h>
+#include <linux/mpage.h>
+#include <linux/sched.h>
+#include <linux/aio.h>
+
+#include "format.h"
+#include "super.h"
+#include "data.h"
+#include "scoutfs_trace.h"
+#include "item.h"
+#include "lock.h"
+#include "file.h"
+
+/* TODO: Direct I/O, AIO */
+ssize_t scoutfs_file_aio_read(struct kiocb *iocb, const struct iovec *iov,
+			      unsigned long nr_segs, loff_t pos)
+{
+	struct file *file = iocb->ki_filp;
+	struct inode *inode = file_inode(file);
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
+	int ret;
+
+	ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, SCOUTFS_LKF_REFRESH_INODE,
+				 inode, &inode_lock);
+	if (ret == 0) {
+		ret = generic_file_aio_read(iocb, iov, nr_segs, pos);
+		scoutfs_unlock(sb, inode_lock, DLM_LOCK_PR);
+	}
+
+	return ret;
+}
+
+ssize_t scoutfs_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
+			       unsigned long nr_segs, loff_t pos)
+{
+	struct file *file = iocb->ki_filp;
+	struct inode *inode = file_inode(file);
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
+	int ret;
+
+	if (iocb->ki_left == 0) /* Does this even happen? */
+		return 0;
+
+	mutex_lock(&inode->i_mutex);
+	ret = scoutfs_lock_inode(sb, DLM_LOCK_EX, SCOUTFS_LKF_REFRESH_INODE,
+				 inode, &inode_lock);
+	if (ret)
+		goto out;
+
+	/* XXX: remove SUID bit */
+
+	ret = __generic_file_aio_write(iocb, iov, nr_segs, &iocb->ki_pos);
+
+	scoutfs_unlock(sb, inode_lock, DLM_LOCK_EX);
+out:
+	mutex_unlock(&inode->i_mutex);
+
+	if (ret > 0 || ret == -EIOCBQUEUED) {
+		ssize_t err;
+
+		err = generic_write_sync(file, pos, ret);
+		if (err < 0 && ret > 0)
+			ret = err;
+	}
+
+	return ret;
+}

kmod/src/file.h

@@ -0,0 +1,9 @@
+#ifndef _SCOUTFS_FILE_H_
+#define _SCOUTFS_FILE_H_
+
+ssize_t scoutfs_file_aio_read(struct kiocb *iocb, const struct iovec *iov,
+			      unsigned long nr_segs, loff_t pos);
+ssize_t scoutfs_file_aio_write(struct kiocb *iocb, const struct iovec *iov,
+			       unsigned long nr_segs, loff_t pos);
+
+#endif	/* _SCOUTFS_FILE_H_ */

kmod/src/inode.c

@@ -414,6 +414,11 @@ static int scoutfs_iget_set(struct inode *inode, void *arg)
 	return 0;
 }
 
+struct inode *scoutfs_ilookup(struct super_block *sb, u64 ino)
+{
+	return ilookup5(sb, ino, scoutfs_iget_test, &ino);
+}
+
 struct inode *scoutfs_iget(struct super_block *sb, u64 ino)
 {
 	struct inode *inode;

kmod/src/inode.h

@@ -2,6 +2,7 @@
 #define _SCOUTFS_INODE_H_
 
 #include "key.h"
+#include "lock.h"
 
 struct scoutfs_lock;
 
@@ -59,6 +60,7 @@ void scoutfs_evict_inode(struct inode *inode);
 int scoutfs_orphan_inode(struct inode *inode);
 
 struct inode *scoutfs_iget(struct super_block *sb, u64 ino);
+struct inode *scoutfs_ilookup(struct super_block *sb, u64 ino);
 int scoutfs_dirty_inode_item(struct inode *inode, struct scoutfs_key_buf *end);
 void scoutfs_update_inode_item(struct inode *inode);
 void scoutfs_inode_fill_pool(struct super_block *sb, u64 ino, u64 nr);

kmod/src/lock.c

@@ -15,6 +15,7 @@
 #include <linux/sched.h>
 #include <linux/slab.h>
 #include <linux/dlm.h>
+#include <linux/mm.h>
 
 #include "super.h"
 #include "lock.h"
@@ -65,9 +66,12 @@ static void scoutfs_downconvert_func(struct work_struct *work);
  * to also invalidate all cached overlapping structures.
  */
 static int invalidate_caches(struct super_block *sb, int mode,
-			     struct scoutfs_key_buf *start,
-			     struct scoutfs_key_buf *end)
+			     struct scoutfs_lock *lock)
 {
+	struct scoutfs_key_buf *start = lock->start;
+	struct scoutfs_key_buf *end = lock->end;
+	struct inode *inode;
+	u64 ino, last;
 	int ret;
 
 	trace_scoutfs_lock_invalidate_sb(sb, mode, start, end);
@@ -76,8 +80,23 @@ static int invalidate_caches(struct super_block *sb, int mode,
 	if (ret)
 		return ret;
 
-	if (mode == DLM_LOCK_EX)
+	if (mode == DLM_LOCK_EX) {
+		if (lock->lock_name.zone == SCOUTFS_FS_ZONE) {
+			ino = le64_to_cpu(lock->lock_name.first);
+			last = ino + SCOUTFS_LOCK_INODE_GROUP_NR - 1;
+			while (ino <= last) {
+				inode = scoutfs_ilookup(lock->sb, ino);
+				if (inode && S_ISREG(inode->i_mode))
+					truncate_inode_pages(inode->i_mapping,
+							     0);
+
+				iput(inode);
+				ino++;
+			}
+		}
+
 		ret = scoutfs_item_invalidate(sb, start, end);
+	}
 
 	return ret;
 }
@@ -133,7 +152,7 @@ static int ino_lock_downconvert(struct ocfs2_lock_res *lockres, int blocking)
 	struct scoutfs_lock *lock = lockres->l_priv;
 	struct super_block *sb = lock->sb;
 
-	invalidate_caches(sb, blocking, lock->start, lock->end);
+	invalidate_caches(sb, blocking, lock);
 
 	return UNBLOCK_CONTINUE;
 }
@@ -685,7 +704,7 @@ static void scoutfs_downconvert_func(struct work_struct *work)
 	 * invalidate based on what level we're downconverting to (PR,
 	 * NL).
 	 */
-	invalidate_caches(sb, DLM_LOCK_EX, lock->start, lock->end);
+	invalidate_caches(sb, DLM_LOCK_EX, lock);
 	unlock_range(sb, lock);
 
 	spin_lock(&linfo->lock);