scoutfs: Add cluster locking to node/file ops

This gives us cluster locking for the overwhelming majority of metadata ops
that scoutfs supports. In particular, we can create and modify file metadata
from one node and immediately see the changes reflected on another node.

In addition to synchonrization the cluster locks here are providing an I/O
endpoint for our item cache, ensuring that it doesn't read stale items.

Readdir and file read/write are notable exception - they require a more
specific approach and will be implemented in a future patch.

Signed-off-by: Mark Fasheh <mfasheh@versity.com>
[fixed iget unlock and truncated commit message summary]
Signed-off-by: Zach Brown <zab@versity.com>

kmod/src/data.c

@@ -31,6 +31,7 @@
 #include "item.h"
 #include "ioctl.h"
 #include "net.h"
+#include "lock.h"
 
 #define EXTF "[off %llu bno %llu bks %llu fl %x]"
 #define EXTA(ne) (ne)->blk_off, (ne)->blkno, (ne)->blocks, (ne)->flags
@@ -1065,7 +1066,7 @@ static int scoutfs_write_begin(struct file *file,
 	flags |= AOP_FLAG_NOFS;
 
 	/* generic write_end updates i_size and calls dirty_inode */
-	ret = scoutfs_dirty_inode_item(inode);
+	ret = scoutfs_dirty_inode_item(inode, NULL);
 	if (ret == 0)
 		ret = block_write_begin(mapping, pos, len, flags, pagep,
 					scoutfs_get_block);
@@ -1116,17 +1117,18 @@ int scoutfs_data_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
 	struct scoutfs_key_buf last;
 	struct scoutfs_key_buf key;
 	struct native_extent ext;
+	struct scoutfs_lock *inode_lock = NULL;
 	u64 logical;
 	u64 blk_off;
 	u64 final;
 	u64 phys;
 	u64 size;
 	u32 flags;
-	int ret = 0;
+	int ret;
 
 	ret = fiemap_check_flags(fieinfo, FIEMAP_FLAG_SYNC);
 	if (ret)
-		goto out;
+		return ret;
 
 	memset(&ext, ~0, sizeof(ext));
 	init_extent_key(&last, last_bytes, &ext, ino, type);
@@ -1139,6 +1141,11 @@ int scoutfs_data_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
 	/* XXX overkill? */
 	mutex_lock(&inode->i_mutex);
 
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_PR, scoutfs_ino(inode),
+				     &inode_lock);
+	if (ret)
+		goto out;
+
 	for (;;) {
 		ext.blk_off = blk_off;
 		ext.blkno = 0;
@@ -1181,8 +1188,10 @@ int scoutfs_data_fiemap(struct inode *inode, struct fiemap_extent_info *fieinfo,
 		blk_off = ext.blk_off + ext.blocks;
 	}
 
-	mutex_unlock(&inode->i_mutex);
+	scoutfs_unlock(sb, inode_lock);
 out:
+	mutex_unlock(&inode->i_mutex);
+
 	return ret;
 }
 

kmod/src/dir.c

@@ -28,6 +28,7 @@
 #include "xattr.h"
 #include "kvec.h"
 #include "item.h"
+#include "lock.h"
 
 /*
  * Directory entries are stored in entries with offsets calculated from
@@ -117,8 +118,16 @@ static void scoutfs_d_release(struct dentry *dentry)
 	}
 }
 
+static int scoutfs_d_revalidate(struct dentry *dentry, unsigned int flags)
+{
+	if (flags & LOOKUP_RCU)
+		return -ECHILD;
+	return 0;/* Always revalidate for now */
+}
+
 static const struct dentry_operations scoutfs_dentry_ops = {
 	.d_release = scoutfs_d_release,
+	.d_revalidate = scoutfs_d_revalidate,
 };
 
 static int alloc_dentry_info(struct dentry *dentry)
@@ -229,6 +238,7 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry,
 	struct super_block *sb = dir->i_sb;
 	struct scoutfs_key_buf *key = NULL;
 	struct scoutfs_dirent dent;
+	struct scoutfs_lock *dir_lock = NULL;
 	SCOUTFS_DECLARE_KVEC(val);
 	struct inode *inode;
 	u64 ino = 0;
@@ -249,9 +259,15 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry,
 		goto out;
 	}
 
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_PR, scoutfs_ino(dir),
+				     &dir_lock);
+	if (ret)
+		goto out;
+
 	scoutfs_kvec_init(val, &dent, sizeof(dent));
 
-	ret = scoutfs_item_lookup_exact(sb, key, val, sizeof(dent), NULL);
+	ret = scoutfs_item_lookup_exact(sb, key, val, sizeof(dent),
+					dir_lock->end);
 	if (ret == -ENOENT) {
 		ino = 0;
 		ret = 0;
@@ -259,7 +275,6 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry,
 		ino = le64_to_cpu(dent.ino);
 		update_dentry_info(dentry, &dent);
 	}
-
 out:
 	if (ret < 0)
 		inode = ERR_PTR(ret);
@@ -268,6 +283,8 @@ out:
 	else
 		inode = scoutfs_iget(sb, ino);
 
+	scoutfs_unlock(sb, dir_lock);
+
 	scoutfs_key_free(sb, key);
 
 	return d_splice_alias(inode, dentry);
@@ -323,6 +340,7 @@ static int scoutfs_readdir(struct file *file, void *dirent, filldir_t filldir)
 	struct scoutfs_key_buf last_key;
 	struct scoutfs_readdir_key rkey;
 	struct scoutfs_readdir_key last_rkey;
+	struct scoutfs_lock *dir_lock;
 	SCOUTFS_DECLARE_KVEC(val);
 	unsigned int item_len;
 	unsigned int name_len;
@@ -332,19 +350,27 @@ static int scoutfs_readdir(struct file *file, void *dirent, filldir_t filldir)
 	if (!dir_emit_dots(file, dirent, filldir))
 		return 0;
 
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_PR, scoutfs_ino(inode),
+				     &dir_lock);
+	if (ret)
+		return ret;
+
 	init_readdir_key(&last_key, &last_rkey, inode, SCOUTFS_DIRENT_LAST_POS);
 
 	item_len = offsetof(struct scoutfs_dirent, name[SCOUTFS_NAME_LEN]);
 	dent = kmalloc(item_len, GFP_KERNEL);
-	if (!dent)
-		return -ENOMEM;
+	if (!dent) {
+		ret = -ENOMEM;
+		goto out;
+	}
 
 	for (;;) {
 		init_readdir_key(&key, &rkey, inode, file->f_pos);
 
 		scoutfs_kvec_init(val, dent, item_len);
 		ret = scoutfs_item_next_same_min(sb, &key, &last_key, val,
-				offsetof(struct scoutfs_dirent, name[1]), NULL);
+				offsetof(struct scoutfs_dirent, name[1]),
+						 dir_lock->end);
 		if (ret < 0) {
 			if (ret == -ENOENT)
 				ret = 0;
@@ -363,12 +389,16 @@ static int scoutfs_readdir(struct file *file, void *dirent, filldir_t filldir)
 		file->f_pos = pos + 1;
 	}
 
+out:
+	scoutfs_unlock(sb, dir_lock);
+
 	kfree(dent);
 	return ret;
 }
 
-static int add_entry_items(struct inode *dir, struct dentry *dentry,
-			   struct inode *inode)
+static int add_entry_items(struct inode *dir, struct scoutfs_lock *dir_lock,
+			   struct dentry *dentry, struct inode *inode,
+			   struct scoutfs_lock *inode_lock)
 {
 	struct scoutfs_inode_info *si = SCOUTFS_I(dir);
 	struct dentry_info *di = dentry->d_fsdata;
@@ -376,6 +406,7 @@ static int add_entry_items(struct inode *dir, struct dentry *dentry,
 	struct scoutfs_key_buf *ent_key = NULL;
 	struct scoutfs_key_buf *lb_key = NULL;
 	struct scoutfs_key_buf *del_keys[3];
+	struct scoutfs_key_buf *end_keys[3];
 	struct scoutfs_key_buf rdir_key;
 	struct scoutfs_readdir_key rkey;
 	struct scoutfs_dirent dent;
@@ -392,7 +423,7 @@ static int add_entry_items(struct inode *dir, struct dentry *dentry,
 	if (dentry->d_name.len > SCOUTFS_NAME_LEN)
 		return -ENAMETOOLONG;
 
-	ret = scoutfs_dirty_inode_item(dir);
+	ret = scoutfs_dirty_inode_item(dir, dir_lock->end);
 	if (ret)
 		return ret;
 
@@ -413,6 +444,7 @@ static int add_entry_items(struct inode *dir, struct dentry *dentry,
 	if (ret)
 		goto out;
 	del_keys[del++] = ent_key;
+	end_keys[del] = dir_lock->end;
 
 	/* readdir item for .. readdir */
 	init_readdir_key(&rdir_key, &rkey, dir, pos);
@@ -423,6 +455,7 @@ static int add_entry_items(struct inode *dir, struct dentry *dentry,
 	if (ret)
 		goto out;
 	del_keys[del++] = &rdir_key;
+	end_keys[del] = dir_lock->end;
 
 	/* link backref item for inode to path resolution */
 	lb_key = alloc_link_backref_key(sb, scoutfs_ino(inode),
@@ -438,12 +471,13 @@ static int add_entry_items(struct inode *dir, struct dentry *dentry,
 	if (ret)
 		goto out;
 	del_keys[del++] = lb_key;
+	end_keys[del] = inode_lock->end;
 
 	update_dentry_info(dentry, &dent);
 	ret = 0;
 out:
 	while (ret < 0 && --del >= 0) {
-		err = scoutfs_item_delete(sb, del_keys[del], NULL);
+		err = scoutfs_item_delete(sb, del_keys[del], end_keys[del]);
 		/* can always delete dirty while holding */
 		BUG_ON(err);
 	}
@@ -459,17 +493,24 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 {
 	struct super_block *sb = dir->i_sb;
 	DECLARE_ITEM_COUNT(cnt);
-	struct inode *inode;
+	struct inode *inode = NULL;
+	struct scoutfs_lock *dir_lock;
+	struct scoutfs_lock *inode_lock = NULL;
 	int ret;
 
 	ret = alloc_dentry_info(dentry);
 	if (ret)
 		return ret;
 
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(dir),
+				     &dir_lock);
+	if (ret)
+		return ret;
+
 	scoutfs_count_mknod(&cnt, dentry->d_name.len);
 	ret = scoutfs_hold_trans(sb, &cnt);
 	if (ret)
-		return ret;
+		goto out_unlock;
 
 	inode = scoutfs_new_inode(sb, dir, mode, rdev);
 	if (IS_ERR(inode)) {
@@ -477,7 +518,13 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 		goto out;
 	}
 
-	ret = add_entry_items(dir, dentry, inode);
+	/* Now that we have ino from scoutfs_new_inode, allocate a lock */
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(inode),
+				     &inode_lock);
+	if (ret)
+		goto out;
+
+	ret = add_entry_items(dir, dir_lock, dentry, inode, inode_lock);
 	if (ret)
 		goto out;
 
@@ -496,10 +543,13 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 	insert_inode_hash(inode);
 	d_instantiate(dentry, inode);
 out:
+	scoutfs_release_trans(sb);
+out_unlock:
+	scoutfs_unlock(sb, dir_lock);
+	scoutfs_unlock(sb, inode_lock);
 	/* XXX delete the inode item here */
 	if (ret && !IS_ERR_OR_NULL(inode))
 		iput(inode);
-	scoutfs_release_trans(sb);
 	return ret;
 }
 
@@ -520,22 +570,34 @@ static int scoutfs_link(struct dentry *old_dentry,
 {
 	struct inode *inode = old_dentry->d_inode;
 	struct super_block *sb = dir->i_sb;
+	struct scoutfs_lock *dir_lock;
+	struct scoutfs_lock *inode_lock = NULL;
 	DECLARE_ITEM_COUNT(cnt);
 	int ret;
 
 	if (inode->i_nlink >= SCOUTFS_LINK_MAX)
 		return -EMLINK;
 
-	ret = alloc_dentry_info(dentry);
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(dir),
+				     &dir_lock);
 	if (ret)
 		return ret;
 
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(inode),
+				     &inode_lock);
+	if (ret)
+		goto out_unlock;
+
+	ret = alloc_dentry_info(dentry);
+	if (ret)
+		goto out_unlock;
+
 	scoutfs_count_link(&cnt, dentry->d_name.len);
 	ret = scoutfs_hold_trans(sb, &cnt);
 	if (ret)
-		return ret;
+		goto out_unlock;
 
-	ret = add_entry_items(dir, dentry, inode);
+	ret = add_entry_items(dir, dir_lock, dentry, inode, inode_lock);
 	if (ret)
 		goto out;
 
@@ -551,6 +613,9 @@ static int scoutfs_link(struct dentry *old_dentry,
 	d_instantiate(dentry, inode);
 out:
 	scoutfs_release_trans(sb);
+out_unlock:
+	scoutfs_unlock(sb, dir_lock);
+	scoutfs_unlock(sb, inode_lock);
 	return ret;
 }
 
@@ -564,21 +629,24 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 	struct inode *inode = dentry->d_inode;
 	struct timespec ts = current_kernel_time();
 	struct scoutfs_key_buf *keys[3] = {NULL,};
+	struct scoutfs_key_buf *ends[3] = {NULL,};
 	struct scoutfs_key_buf rdir_key;
 	struct scoutfs_readdir_key rkey;
 	DECLARE_ITEM_COUNT(cnt);
+	struct scoutfs_lock *dir_lock = NULL;
+	struct scoutfs_lock *inode_lock = NULL;
 	int ret = 0;
 
 	if (S_ISDIR(inode->i_mode) && i_size_read(inode))
 		return -ENOTEMPTY;
 
-	scoutfs_count_unlink(&cnt, dentry->d_name.len);
-	ret = scoutfs_hold_trans(sb, &cnt);
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(dir),
+				     &dir_lock);
 	if (ret)
 		return ret;
 
-	ret = scoutfs_dirty_inode_item(dir) ?:
-	      scoutfs_dirty_inode_item(inode);
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(inode),
+				     &inode_lock);
 	if (ret)
 		goto out;
 
@@ -587,9 +655,11 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 		ret = -ENOMEM;
 		goto out;
 	}
+	ends[0] = dir_lock->end;
 
 	init_readdir_key(&rdir_key, &rkey, dir, dentry_info_pos(dentry));
 	keys[1] = &rdir_key;
+	ends[1] = dir_lock->end;
 
 	keys[2] = alloc_link_backref_key(sb, scoutfs_ino(inode),
 					 scoutfs_ino(dir),
@@ -600,10 +670,20 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 		goto out;
 	}
 
-	ret = scoutfs_item_delete_many(sb, keys, ARRAY_SIZE(keys), NULL);
+	scoutfs_count_unlink(&cnt, dentry->d_name.len);
+	ret = scoutfs_hold_trans(sb, &cnt);
 	if (ret)
 		goto out;
 
+	ret = scoutfs_dirty_inode_item(dir, dir_lock->end) ?:
+		scoutfs_dirty_inode_item(inode, inode_lock->end);
+	if (ret)
+		goto out_trans;
+
+	ret = scoutfs_item_delete_many(sb, keys, ARRAY_SIZE(keys), ends);
+	if (ret)
+		goto out_trans;
+
 	if ((inode->i_nlink == 1) ||
 	    (S_ISDIR(inode->i_mode) && inode->i_nlink == 2)) {
 		/*
@@ -613,7 +693,7 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 		 */
 		ret = scoutfs_orphan_inode(inode);
 		if (ret)
-			goto out;
+			goto out_trans;
 	}
 
 	dir->i_ctime = ts;
@@ -629,10 +709,13 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 	scoutfs_update_inode_item(inode);
 	scoutfs_update_inode_item(dir);
 
+out_trans:
+	scoutfs_release_trans(sb);
 out:
 	scoutfs_key_free(sb, keys[0]);
 	scoutfs_key_free(sb, keys[2]);
-	scoutfs_release_trans(sb);
+	scoutfs_unlock(sb, dir_lock);
+	scoutfs_unlock(sb, inode_lock);
 	return ret;
 }
 
@@ -665,7 +748,8 @@ enum {
 	SYM_DELETE,
 };
 static int symlink_item_ops(struct super_block *sb, int op, u64 ino,
-			    const char *target, size_t size)
+			    struct scoutfs_lock *lock, const char *target,
+			    size_t size)
 {
 	struct scoutfs_symlink_key skey;
 	struct scoutfs_key_buf key;
@@ -690,9 +774,9 @@ static int symlink_item_ops(struct super_block *sb, int op, u64 ino,
 			ret = scoutfs_item_create(sb, &key, val);
 		else if (op == SYM_LOOKUP)
 			ret = scoutfs_item_lookup_exact(sb, &key, val, bytes,
-							NULL);
+							lock->end);
 		else if (op == SYM_DELETE)
-			ret = scoutfs_item_delete(sb, &key, NULL);
+			ret = scoutfs_item_delete(sb, &key, lock->end);
 		if (ret)
 			break;
 
@@ -715,6 +799,7 @@ static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd)
 {
 	struct inode *inode = dentry->d_inode;
 	struct super_block *sb = inode->i_sb;
+	struct scoutfs_lock *inode_lock = NULL;
 	loff_t size = i_size_read(inode);
 	char *path;
 	int ret;
@@ -727,11 +812,19 @@ static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd)
 	if (size > PATH_MAX)
 		return ERR_PTR(-ENAMETOOLONG);
 
-	path = kmalloc(size, GFP_NOFS);
-	if (!path)
-		return ERR_PTR(-ENOMEM);
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_PR, scoutfs_ino(inode),
+				     &inode_lock);
+	if (ret)
+		return ERR_PTR(ret);
 
-	ret = symlink_item_ops(sb, SYM_LOOKUP, scoutfs_ino(inode), path, size);
+	path = kmalloc(size, GFP_NOFS);
+	if (!path) {
+		path = ERR_PTR(-ENOMEM);
+		goto out;
+	}
+
+	ret = symlink_item_ops(sb, SYM_LOOKUP, scoutfs_ino(inode), inode_lock,
+			       path, size);
 
 	/* XXX corruption: missing items or not null term */
 	if (ret == -ENOENT || (ret == 0 && path[size - 1]))
@@ -743,7 +836,8 @@ static void *scoutfs_follow_link(struct dentry *dentry, struct nameidata *nd)
 	} else {
 		nd_set_link(nd, path);
 	}
-
+out:
+	scoutfs_unlock(sb, inode_lock);
 	return path;
 }
 
@@ -774,6 +868,8 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 	struct super_block *sb = dir->i_sb;
 	const int name_len = strlen(symname) + 1;
 	struct inode *inode = NULL;
+	struct scoutfs_lock *dir_lock;
+	struct scoutfs_lock *inode_lock = NULL;
 	DECLARE_ITEM_COUNT(cnt);
 	int ret;
 
@@ -785,10 +881,15 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 	if (ret)
 		return ret;
 
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(dir),
+				     &dir_lock);
+	if (ret)
+		return ret;
+
 	scoutfs_count_symlink(&cnt, dentry->d_name.len, name_len);
 	ret = scoutfs_hold_trans(sb, &cnt);
 	if (ret)
-		return ret;
+		goto out_unlock;
 
 	inode = scoutfs_new_inode(sb, dir, S_IFLNK|S_IRWXUGO, 0);
 	if (IS_ERR(inode)) {
@@ -796,12 +897,17 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 		goto out;
 	}
 
-	ret = symlink_item_ops(sb, SYM_CREATE, scoutfs_ino(inode),
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_EX, scoutfs_ino(inode),
+				     &inode_lock);
+	if (ret)
+		goto out;
+
+	ret = symlink_item_ops(sb, SYM_CREATE, scoutfs_ino(inode), inode_lock,
 			       symname, name_len);
 	if (ret)
 		goto out;
 
-	ret = add_entry_items(dir, dentry, inode);
+	ret = add_entry_items(dir, dir_lock, dentry, inode, inode_lock);
 	if (ret)
 		goto out;
 
@@ -822,19 +928,23 @@ out:
 		if (!IS_ERR_OR_NULL(inode))
 			iput(inode);
 
-		symlink_item_ops(sb, SYM_DELETE, scoutfs_ino(inode),
+		symlink_item_ops(sb, SYM_DELETE, scoutfs_ino(inode), inode_lock,
 				 NULL, name_len);
 	}
 
 	scoutfs_release_trans(sb);
+out_unlock:
+	scoutfs_unlock(sb, dir_lock);
+	scoutfs_unlock(sb, inode_lock);
 	return ret;
 }
 
-int scoutfs_symlink_drop(struct super_block *sb, u64 ino, u64 i_size)
+int scoutfs_symlink_drop(struct super_block *sb, u64 ino,
+			 struct scoutfs_lock *lock, u64 i_size)
 {
 	int ret;
 
-	ret = symlink_item_ops(sb, SYM_DELETE, ino, NULL, i_size);
+	ret = symlink_item_ops(sb, SYM_DELETE, ino, lock, NULL, i_size);
 	if (ret == -ENOENT)
 		ret = 0;
 

kmod/src/dir.h

@@ -2,6 +2,7 @@
 #define _SCOUTFS_DIR_H_
 
 #include "format.h"
+#include "lock.h"
 
 extern const struct file_operations scoutfs_dir_fops;
 extern const struct inode_operations scoutfs_dir_iops;
@@ -19,7 +20,8 @@ int scoutfs_dir_get_backref_path(struct super_block *sb, u64 target_ino,
 void scoutfs_dir_free_backref_path(struct super_block *sb,
 				   struct list_head *list);
 
-int scoutfs_symlink_drop(struct super_block *sb, u64 ino, u64 i_size);
+int scoutfs_symlink_drop(struct super_block *sb, u64 ino,
+			 struct scoutfs_lock *lock, u64 i_size);
 
 int scoutfs_dir_init(void);
 void scoutfs_dir_exit(void);

kmod/src/inode.c

@@ -58,6 +58,8 @@ struct inode_sb_info {
 	struct inode_sb_info *name = SCOUTFS_SB(sb)->inode_sb_info
 
 static struct kmem_cache *scoutfs_inode_cachep;
+static int scoutfs_getattr(struct vfsmount *mnt, struct dentry *dentry,
+			   struct kstat *stat);
 
 /*
  * This is called once before all the allocations and frees of a inode
@@ -141,6 +143,7 @@ void scoutfs_destroy_inode(struct inode *inode)
 }
 
 static const struct inode_operations scoutfs_file_iops = {
+	.getattr	= scoutfs_getattr,
 	.setxattr	= scoutfs_setxattr,
 	.getxattr	= scoutfs_getxattr,
 	.listxattr	= scoutfs_listxattr,
@@ -221,6 +224,25 @@ static void load_inode(struct inode *inode, struct scoutfs_inode *cinode)
 	set_item_info(inode);
 }
 
+static int refresh_inode(struct inode *inode, struct scoutfs_lock *lock)
+{
+	struct super_block *sb = inode->i_sb;
+	struct scoutfs_key_buf key;
+	struct scoutfs_inode_key ikey;
+	struct scoutfs_inode sinode;
+	SCOUTFS_DECLARE_KVEC(val);
+	int ret;
+
+	scoutfs_inode_init_key(&key, &ikey, scoutfs_ino(inode));
+	scoutfs_kvec_init(val, &sinode, sizeof(sinode));
+
+	ret = scoutfs_item_lookup_exact(sb, &key, val, sizeof(sinode), lock->end);
+	if (ret == 0)
+		load_inode(inode, &sinode);
+
+	return ret;
+}
+
 void scoutfs_inode_init_key(struct scoutfs_key_buf *key,
 			    struct scoutfs_inode_key *ikey, u64 ino)
 {
@@ -231,22 +253,24 @@ void scoutfs_inode_init_key(struct scoutfs_key_buf *key,
 	scoutfs_key_init(key, ikey, sizeof(struct scoutfs_inode_key));
 }
 
-static int scoutfs_read_locked_inode(struct inode *inode)
+static int scoutfs_getattr(struct vfsmount *mnt, struct dentry *dentry,
+			   struct kstat *stat)
 {
+	struct inode *inode = dentry->d_inode;
 	struct super_block *sb = inode->i_sb;
-	struct scoutfs_inode_key ikey;
-	struct scoutfs_key_buf key;
-	struct scoutfs_inode sinode;
-	SCOUTFS_DECLARE_KVEC(val);
+	struct scoutfs_lock *lock = NULL;
 	int ret;
 
-	scoutfs_inode_init_key(&key, &ikey, scoutfs_ino(inode));
-	scoutfs_kvec_init(val, &sinode, sizeof(sinode));
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_PR, scoutfs_ino(inode),
+				     &lock);
+	if (ret)
+		return ret;
 
-	ret = scoutfs_item_lookup_exact(sb, &key, val, sizeof(sinode), NULL);
+	ret = refresh_inode(inode, lock);
 	if (ret == 0)
-		load_inode(inode, &sinode);
+		generic_fillattr(inode, stat);
 
+	scoutfs_unlock(sb, lock);
 	return ret;
 }
 
@@ -352,15 +376,22 @@ static int scoutfs_iget_set(struct inode *inode, void *arg)
 struct inode *scoutfs_iget(struct super_block *sb, u64 ino)
 {
 	struct inode *inode;
+	struct scoutfs_lock *lock = NULL;
 	int ret;
 
+	ret = scoutfs_lock_ino_group(sb, DLM_LOCK_PR, ino, &lock);
+	if (ret)
+		return ERR_PTR(ret);
+
 	inode = iget5_locked(sb, ino, scoutfs_iget_test, scoutfs_iget_set,
 			     &ino);
-	if (!inode)
-		return ERR_PTR(-ENOMEM);
+	if (!inode) {
+		inode = ERR_PTR(-ENOMEM);
+		goto out;
+	}
 
 	if (inode->i_state & I_NEW) {
-		ret = scoutfs_read_locked_inode(inode);
+		ret = refresh_inode(inode, lock);
 		if (ret) {
 			iget_failed(inode);
 			inode = ERR_PTR(ret);
@@ -370,6 +401,8 @@ struct inode *scoutfs_iget(struct super_block *sb, u64 ino)
 		}
 	}
 
+out:
+	scoutfs_unlock(sb, lock);
 	return inode;
 }
 
@@ -416,7 +449,7 @@ static void store_inode(struct scoutfs_inode *cinode, struct inode *inode)
  *
  * XXX this will have to do something about variable length inodes
  */
-int scoutfs_dirty_inode_item(struct inode *inode)
+int scoutfs_dirty_inode_item(struct inode *inode, struct scoutfs_key_buf *end)
 {
 	struct super_block *sb = inode->i_sb;
 	struct scoutfs_inode_key ikey;
@@ -428,7 +461,7 @@ int scoutfs_dirty_inode_item(struct inode *inode)
 
 	scoutfs_inode_init_key(&key, &ikey, scoutfs_ino(inode));
 
-	ret = scoutfs_item_dirty(sb, &key, NULL);
+	ret = scoutfs_item_dirty(sb, &key, end);
 	if (!ret)
 		trace_scoutfs_dirty_inode(inode);
 	return ret;
@@ -562,36 +595,6 @@ void scoutfs_update_inode_item(struct inode *inode)
 	trace_scoutfs_update_inode(inode);
 }
 
-/*
- * sop->dirty_inode() can't return failure.  Our use of it has to be
- * careful to pin the inode during a transaction.  The generic write
- * paths pin the inode in write_begin and get called to update the inode
- * in write_end.
- *
- * The caller should have a trans but it's cheap for us to grab it
- * ourselves to make sure.
- *
- * This will holler at us if a caller didn't pin the inode and we
- * couldn't dirty the inode ourselves.
- */
-void scoutfs_dirty_inode(struct inode *inode, int flags)
-{
-	struct super_block *sb = inode->i_sb;
-	DECLARE_ITEM_COUNT(cnt);
-	int ret;
-
-	scoutfs_count_dirty_inode(&cnt);
-	ret = scoutfs_hold_trans(sb, &cnt);
-	if (ret == 0) {
-		ret = scoutfs_dirty_inode_item(inode);
-		if (ret == 0)
-			scoutfs_update_inode_item(inode);
-		scoutfs_release_trans(sb);
-	}
-
-	WARN_ON_ONCE(ret);
-}
-
 /*
  * A quick atomic sample of the last inode number that's been allocated.
  */

kmod/src/inode.h

@@ -46,8 +46,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);
-int scoutfs_dirty_inode_item(struct inode *inode);
-void scoutfs_dirty_inode(struct inode *inode, int flags);
+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);
 struct inode *scoutfs_new_inode(struct super_block *sb, struct inode *dir,

kmod/src/item.c

@@ -1427,13 +1427,13 @@ void scoutfs_item_delete_dirty(struct super_block *sb,
  */
 int scoutfs_item_delete_many(struct super_block *sb,
 			     struct scoutfs_key_buf **keys, unsigned nr,
-			     struct scoutfs_key_buf *end)
+			     struct scoutfs_key_buf **ends)
 {
 	int ret = 0;
 	int i;
 
 	for (i = 0; i < nr; i++) {
-		ret = scoutfs_item_dirty(sb, keys[i], end);
+		ret = scoutfs_item_dirty(sb, keys[i], ends[i]);
 		if (ret)
 			goto out;
 	}

kmod/src/item.h

@@ -38,7 +38,7 @@ void scoutfs_item_delete_dirty(struct super_block *sb,
 			       struct scoutfs_key_buf *key);
 int scoutfs_item_delete_many(struct super_block *sb,
 			     struct scoutfs_key_buf **keys, unsigned nr,
-			     struct scoutfs_key_buf *end);
+			     struct scoutfs_key_buf **ends);
 int scoutfs_item_delete(struct super_block *sb, struct scoutfs_key_buf *key,
 			struct scoutfs_key_buf *end);
 

kmod/src/super.c

@@ -79,7 +79,6 @@ static int scoutfs_statfs(struct dentry *dentry, struct kstatfs *kst)
 
 static const struct super_operations scoutfs_super_ops = {
 	.alloc_inode = scoutfs_alloc_inode,
-	.dirty_inode = scoutfs_dirty_inode,
 	.drop_inode = scoutfs_drop_inode,
 	.evict_inode = scoutfs_evict_inode,
 	.destroy_inode = scoutfs_destroy_inode,

kmod/src/xattr.c

@@ -323,7 +323,7 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 
 	down_write(&si->xattr_rwsem);
 
-	ret = scoutfs_dirty_inode_item(inode) ?:
+	ret = scoutfs_dirty_inode_item(inode, lck->end) ?:
 	      scoutfs_item_set_batch(sb, &list, key, last, sif, lck->end);
 	if (ret == 0) {
 		/* XXX do these want i_mutex or anything? */