scoutfs: make trans item count const rhs

The item count estimate functions didn't obviously differentiate between
adding to a count and resetting it.  Most callers initialized the count
struct to 0 on the stack, incremented their estimate once, and passed it
in.  The problem is that those same functions that increment once in
callers are also used in other estimates to build counts based on
multiple operations.

This tripped up the data truncate path.  It looped and kept incrementing
its count while truncating a file with lots of extents until the count
got so large that it didn't fit in a segment by itself and blocked
forever.

This cleans up the item count code so that it's much harder to get
wrong.  We differentiate between the SIC_*() high level count estimates
that are meant to be passed in to _hold_trans(), and the internal
__count_*() functions which are used to add up the item counts that make
up an aggregate operation.

With this fix the only way to use the count in extent truncation is to
correctly reset it for the item count for each transacation.

Signed-off-by: Zach Brown <zab@versity.com>

kmod/src/count.h

@@ -1,19 +1,34 @@
 #ifndef _SCOUTFS_COUNT_H_
 #define _SCOUTFS_COUNT_H_
 
+/*
+ * Our estimate of the space consumed while dirtying items isn't a
+ * single value.  We're packing items into segments which have different
+ * overheads for items (header overhead), keys (block aligned), and
+ * values (can span blocks, not aligned).
+ *
+ * The estimate is still a read-only input to entering the transaction.
+ * We'd like to use it as a clean rhs arg to hold_trans.  We define SIC_
+ * functions which return the count struct.  This lets us have a single
+ * arg and avoid bugs in initializing and passing in struct pointers
+ * from callers.  The internal __count functions are used compose an
+ * estimate out of the sets of items it manipulates.  We program in much
+ * clearer C instead of in the preprocessor.
+ *
+ * Compilers are able to collapse the inlines into constants for the
+ * constant estimates.
+ */
+
 struct scoutfs_item_count {
 	signed items;
 	signed keys;
 	signed vals;
 };
 
-#define DECLARE_ITEM_COUNT(name) \
-	struct scoutfs_item_count name = { 0, }
-
 /*
  * Allocating an inode creates a new set of indexed items.
  */
-static inline void scoutfs_count_alloc_inode(struct scoutfs_item_count *cnt)
+static inline void __count_alloc_inode(struct scoutfs_item_count *cnt)
 {
 	const int nr_indices = SCOUTFS_INODE_INDEX_NR;
 
@@ -27,7 +42,7 @@ static inline void scoutfs_count_alloc_inode(struct scoutfs_item_count *cnt)
  * Dirtying an inode dirties the inode item and can delete and create
  * the full set of indexed items.
  */
-static inline void scoutfs_count_dirty_inode(struct scoutfs_item_count *cnt)
+static inline void __count_dirty_inode(struct scoutfs_item_count *cnt)
 {
 	const int nr_indices = 2 * SCOUTFS_INODE_INDEX_NR;
 
@@ -37,11 +52,29 @@ static inline void scoutfs_count_dirty_inode(struct scoutfs_item_count *cnt)
 	cnt->vals += sizeof(struct scoutfs_inode);
 }
 
+static inline const struct scoutfs_item_count SIC_ALLOC_INODE(void)
+{
+	struct scoutfs_item_count cnt = {0,};
+
+	__count_alloc_inode(&cnt);
+
+	return cnt;
+}
+
+static inline const struct scoutfs_item_count SIC_DIRTY_INODE(void)
+{
+	struct scoutfs_item_count cnt = {0,};
+
+	__count_dirty_inode(&cnt);
+
+	return cnt;
+}
+
 /*
  * Adding a dirent adds the entry key, readdir key, and backref.
  */
-static inline void scoutfs_count_dirents(struct scoutfs_item_count *cnt,
-					   unsigned name_len)
+static inline void __count_dirents(struct scoutfs_item_count *cnt,
+				   unsigned name_len)
 {
 
 	cnt->items += 3;
@@ -51,8 +84,8 @@ static inline void scoutfs_count_dirents(struct scoutfs_item_count *cnt,
 	cnt->vals += 2 * offsetof(struct scoutfs_dirent, name[name_len]);
 }
 
-static inline void scoutfs_count_sym_target(struct scoutfs_item_count *cnt,
-					      unsigned size)
+static inline void __count_sym_target(struct scoutfs_item_count *cnt,
+				      unsigned size)
 {
 	unsigned nr = DIV_ROUND_UP(size, SCOUTFS_MAX_VAL_SIZE);
 
@@ -61,46 +94,65 @@ static inline void scoutfs_count_sym_target(struct scoutfs_item_count *cnt,
 	cnt->vals += size;
 }
 
-static inline void scoutfs_count_orphan(struct scoutfs_item_count *cnt)
+static inline void __count_orphan(struct scoutfs_item_count *cnt)
 {
 
 	cnt->items += 1;
 	cnt->keys += sizeof(struct scoutfs_orphan_key);
 }
 
-static inline void scoutfs_count_mknod(struct scoutfs_item_count *cnt,
-					 unsigned name_len)
+static inline void __count_mknod(struct scoutfs_item_count *cnt,
+				 unsigned name_len)
 {
-	scoutfs_count_alloc_inode(cnt);
-	scoutfs_count_dirents(cnt, name_len);
-	scoutfs_count_dirty_inode(cnt);
+	__count_alloc_inode(cnt);
+	__count_dirents(cnt, name_len);
+	__count_dirty_inode(cnt);
 }
 
-static inline void scoutfs_count_link(struct scoutfs_item_count *cnt,
-					unsigned name_len)
+static inline const struct scoutfs_item_count SIC_MKNOD(unsigned name_len)
 {
-	scoutfs_count_dirents(cnt, name_len);
-	scoutfs_count_dirty_inode(cnt);
-	scoutfs_count_dirty_inode(cnt);
+	struct scoutfs_item_count cnt = {0,};
+
+	__count_mknod(&cnt, name_len);
+
+	return cnt;
+}
+
+static inline const struct scoutfs_item_count SIC_LINK(unsigned name_len)
+{
+	struct scoutfs_item_count cnt = {0,};
+
+	__count_dirents(&cnt, name_len);
+	__count_dirty_inode(&cnt);
+	__count_dirty_inode(&cnt);
+
+	return cnt;
 }
 
 /*
  * Unlink can add orphan items.
  */
-static inline void scoutfs_count_unlink(struct scoutfs_item_count *cnt,
-					  unsigned name_len)
+static inline const struct scoutfs_item_count SIC_UNLINK(unsigned name_len)
 {
-	scoutfs_count_dirents(cnt, name_len);
-	scoutfs_count_dirty_inode(cnt);
-	scoutfs_count_dirty_inode(cnt);
-	scoutfs_count_orphan(cnt);
+	struct scoutfs_item_count cnt = {0,};
+
+	__count_dirents(&cnt, name_len);
+	__count_dirty_inode(&cnt);
+	__count_dirty_inode(&cnt);
+	__count_orphan(&cnt);
+
+	return cnt;
 }
 
-static inline void scoutfs_count_symlink(struct scoutfs_item_count *cnt,
-					   unsigned name_len, unsigned size)
+static inline const struct scoutfs_item_count SIC_SYMLINK(unsigned name_len,
+							  unsigned size)
 {
-	scoutfs_count_mknod(cnt, name_len);
-	scoutfs_count_sym_target(cnt, size);
+	struct scoutfs_item_count cnt = {0,};
+
+	__count_mknod(&cnt, name_len);
+	__count_sym_target(&cnt, size);
+
+	return cnt;
 }
 
 /*
@@ -108,22 +160,26 @@ static inline void scoutfs_count_symlink(struct scoutfs_item_count *cnt,
  * unlinks an existing target.  That'll be worse than the common case
  * by a few hundred bytes.
  */
-static inline void scoutfs_count_rename(struct scoutfs_item_count *cnt,
-					unsigned old_len, unsigned new_len)
+static inline const struct scoutfs_item_count SIC_RENAME(unsigned old_len,
+							 unsigned new_len)
 {
+	struct scoutfs_item_count cnt = {0,};
+
 	/* dirty dirs and inodes */
-	scoutfs_count_dirty_inode(cnt);
-	scoutfs_count_dirty_inode(cnt);
-	scoutfs_count_dirty_inode(cnt);
-	scoutfs_count_dirty_inode(cnt);
+	__count_dirty_inode(&cnt);
+	__count_dirty_inode(&cnt);
+	__count_dirty_inode(&cnt);
+	__count_dirty_inode(&cnt);
 
 	/* unlink old and new, link new */
-	scoutfs_count_dirents(cnt, old_len);
-	scoutfs_count_dirents(cnt, new_len);
-	scoutfs_count_dirents(cnt, new_len);
+	__count_dirents(&cnt, old_len);
+	__count_dirents(&cnt, new_len);
+	__count_dirents(&cnt, new_len);
 
 	/* orphan the existing target */
-	scoutfs_count_orphan(cnt);
+	__count_orphan(&cnt);
+
+	return cnt;
 }
 
 /*
@@ -131,19 +187,22 @@ static inline void scoutfs_count_rename(struct scoutfs_item_count *cnt,
  * max name and length.  Any existing items will be dirtied rather than
  * deleted so we won't have more items than a max xattr's worth.
  */
-static inline void scoutfs_count_xattr_set(struct scoutfs_item_count *cnt,
-					     unsigned name_len, unsigned size)
+static inline const struct scoutfs_item_count SIC_XATTR_SET(unsigned name_len,
+							    unsigned size)
 {
+	struct scoutfs_item_count cnt = {0,};
 	unsigned parts = DIV_ROUND_UP(size, SCOUTFS_XATTR_PART_SIZE);
 
-	scoutfs_count_dirty_inode(cnt);
+	__count_dirty_inode(&cnt);
 
-	cnt->items += parts;
-	cnt->keys += parts * (offsetof(struct scoutfs_xattr_key,
+	cnt.items += parts;
+	cnt.keys += parts * (offsetof(struct scoutfs_xattr_key,
 					name[name_len]) +
 			       sizeof(struct scoutfs_xattr_key_footer));
-	cnt->vals += parts * (sizeof(struct scoutfs_xattr_val_header) +
+	cnt.vals += parts * (sizeof(struct scoutfs_xattr_val_header) +
 			       SCOUTFS_XATTR_PART_SIZE);
+
+	return cnt;
 }
 
 /*
@@ -152,10 +211,9 @@ static inline void scoutfs_count_xattr_set(struct scoutfs_item_count *cnt,
  * third new extent and removal can delete an existing extent and create
  * two new remaining extents.
  */
-static inline void scoutfs_count_extents(struct scoutfs_item_count *cnt,
-					   unsigned nr_mod, unsigned sz)
+static inline void __count_extents(struct scoutfs_item_count *cnt,
+				   unsigned nr_mod, unsigned sz)
 {
-
 	cnt->items += nr_mod * 3;
 	cnt->keys += (nr_mod * 3) * sz;
 }
@@ -165,29 +223,35 @@ static inline void scoutfs_count_extents(struct scoutfs_item_count *cnt,
  * alloc an block, delete an offline mapping, and insert the new allocated
  * mapping.
  */
-static inline void scoutfs_count_write_begin(struct scoutfs_item_count *cnt)
+static inline const struct scoutfs_item_count SIC_WRITE_BEGIN(void)
 {
+	struct scoutfs_item_count cnt = {0,};
+
 	BUILD_BUG_ON(sizeof(struct scoutfs_free_extent_blkno_key) !=
 		     sizeof(struct scoutfs_free_extent_blocks_key));
 
-	scoutfs_count_dirty_inode(cnt);
+	__count_dirty_inode(&cnt);
 
-	scoutfs_count_extents(cnt, 2 * (SCOUTFS_BULK_ALLOC_COUNT + 1),
-			        sizeof(struct scoutfs_free_extent_blkno_key));
-	scoutfs_count_extents(cnt, 2,
-			        sizeof(struct scoutfs_file_extent_key));
+	__count_extents(&cnt, 2 * (SCOUTFS_BULK_ALLOC_COUNT + 1),
+			sizeof(struct scoutfs_free_extent_blkno_key));
+	__count_extents(&cnt, 2, sizeof(struct scoutfs_file_extent_key));
+
+	return cnt;
 }
 
 /*
  * Truncating a block can free an allocated block, delete an online
  * mapping, and create an offline mapping.
  */
-static inline void scoutfs_count_trunc_block(struct scoutfs_item_count *cnt)
+static inline const struct scoutfs_item_count SIC_TRUNC_BLOCK(void)
 {
-	scoutfs_count_extents(cnt, 2 * 1,
-			        sizeof(struct scoutfs_free_extent_blkno_key));
-	scoutfs_count_extents(cnt, 2,
-			        sizeof(struct scoutfs_file_extent_key));
+	struct scoutfs_item_count cnt = {0,};
+
+	__count_extents(&cnt, 2 * 1,
+			sizeof(struct scoutfs_free_extent_blkno_key));
+	__count_extents(&cnt, 2, sizeof(struct scoutfs_file_extent_key));
+
+	return cnt;
 }
 
 #endif

kmod/src/data.c

@@ -536,7 +536,6 @@ int scoutfs_data_truncate_items(struct super_block *sb, u64 ino, u64 iblock,
 	struct native_extent ext;
 	struct native_extent ofl;
 	struct native_extent fr;
-	DECLARE_ITEM_COUNT(cnt);
 	bool rem_fr = false;
 	bool ins_ext = false;
 	bool holding = false;
@@ -602,8 +601,7 @@ int scoutfs_data_truncate_items(struct super_block *sb, u64 ino, u64 iblock,
 		if (offline && (ext.flags & SCOUTFS_FILE_EXTENT_OFFLINE))
 			continue;
 
-		scoutfs_count_trunc_block(&cnt);
-		ret = scoutfs_hold_trans(sb, &cnt);
+		ret = scoutfs_hold_trans(sb, SIC_TRUNC_BLOCK());
 		if (ret)
 			break;
 		holding = true;
@@ -1125,14 +1123,12 @@ static int scoutfs_write_begin(struct file *file,
 {
 	struct inode *inode = mapping->host;
 	struct super_block *sb = inode->i_sb;
-	DECLARE_ITEM_COUNT(cnt);
 	int ret;
 
 	trace_printk("ino %llu pos %llu len %u\n",
 		     scoutfs_ino(inode), (u64)pos, len);
 
-	scoutfs_count_write_begin(&cnt);
-	ret = scoutfs_hold_trans(sb, &cnt);
+	ret = scoutfs_hold_trans(sb, SIC_WRITE_BEGIN());
 	if (ret)
 		goto out;
 

kmod/src/dir.c

@@ -537,7 +537,7 @@ out:
  */
 static struct inode *lock_hold_create(struct inode *dir, struct dentry *dentry,
 				      umode_t mode, dev_t rdev,
-				      struct scoutfs_item_count *cnt,
+				      const struct scoutfs_item_count cnt,
 				      struct scoutfs_lock **dir_lock,
 				      struct scoutfs_lock **inode_lock)
 {
@@ -599,7 +599,6 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 		       dev_t rdev)
 {
 	struct super_block *sb = dir->i_sb;
-	DECLARE_ITEM_COUNT(cnt);
 	struct inode *inode = NULL;
 	struct scoutfs_lock *dir_lock = NULL;
 	struct scoutfs_lock *inode_lock = NULL;
@@ -609,9 +608,9 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 	if (dentry->d_name.len > SCOUTFS_NAME_LEN)
 		return -ENAMETOOLONG;
 
-	scoutfs_count_mknod(&cnt, dentry->d_name.len);
 
-	inode = lock_hold_create(dir, dentry, mode, rdev, &cnt,
+	inode = lock_hold_create(dir, dentry, mode, rdev,
+				 SIC_MKNOD(dentry->d_name.len),
 				 &dir_lock, &inode_lock);
 	if (IS_ERR(inode))
 		return PTR_ERR(inode);
@@ -670,7 +669,6 @@ static int scoutfs_link(struct dentry *old_dentry,
 	struct super_block *sb = dir->i_sb;
 	struct scoutfs_lock *dir_lock;
 	struct scoutfs_lock *inode_lock = NULL;
-	DECLARE_ITEM_COUNT(cnt);
 	u64 pos;
 	int ret;
 
@@ -692,8 +690,7 @@ static int scoutfs_link(struct dentry *old_dentry,
 	if (ret)
 		goto out_unlock;
 
-	scoutfs_count_link(&cnt, dentry->d_name.len);
-	ret = scoutfs_hold_trans(sb, &cnt);
+	ret = scoutfs_hold_trans(sb, SIC_LINK(dentry->d_name.len));
 	if (ret)
 		goto out_unlock;
 
@@ -750,7 +747,6 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 	struct timespec ts = current_kernel_time();
 	struct scoutfs_lock *inode_lock = NULL;
 	struct scoutfs_lock *dir_lock = NULL;
-	DECLARE_ITEM_COUNT(cnt);
 	int ret = 0;
 
 	ret = scoutfs_lock_inodes(sb, DLM_LOCK_EX, SCOUTFS_LKF_REFRESH_INODE,
@@ -764,8 +760,7 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 		goto unlock;
 	}
 
-	scoutfs_count_unlink(&cnt, dentry->d_name.len);
-	ret = scoutfs_hold_trans(sb, &cnt);
+	ret = scoutfs_hold_trans(sb, SIC_UNLINK(dentry->d_name.len));
 	if (ret)
 		goto unlock;
 
@@ -966,7 +961,6 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 	struct inode *inode = NULL;
 	struct scoutfs_lock *dir_lock = NULL;
 	struct scoutfs_lock *inode_lock = NULL;
-	DECLARE_ITEM_COUNT(cnt);
 	u64 pos;
 	int ret;
 
@@ -979,8 +973,8 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 	if (ret)
 		return ret;
 
-	scoutfs_count_symlink(&cnt, dentry->d_name.len, name_len);
-	inode = lock_hold_create(dir, dentry, S_IFLNK|S_IRWXUGO, 0, &cnt,
+	inode = lock_hold_create(dir, dentry, S_IFLNK|S_IRWXUGO, 0,
+				 SIC_SYMLINK(dentry->d_name.len, name_len),
 				 &dir_lock, &inode_lock);
 	if (IS_ERR(inode))
 		return PTR_ERR(inode);
@@ -1328,7 +1322,6 @@ static int scoutfs_rename(struct inode *old_dir, struct dentry *old_dentry,
 	struct scoutfs_lock *old_inode_lock = NULL;
 	struct scoutfs_lock *new_inode_lock = NULL;
 	struct timespec now;
-	DECLARE_ITEM_COUNT(cnt);
 	bool ins_new = false;
 	bool del_new = false;
 	bool ins_old = false;
@@ -1379,9 +1372,8 @@ static int scoutfs_rename(struct inode *old_dir, struct dentry *old_dentry,
 	if (ret)
 		goto out_unlock;
 
-	scoutfs_count_rename(&cnt, old_dentry->d_name.len,
-				   new_dentry->d_name.len);
-	ret = scoutfs_hold_trans(sb, &cnt);
+	ret = scoutfs_hold_trans(sb, SIC_RENAME(old_dentry->d_name.len,
+						new_dentry->d_name.len));
 	if (ret)
 		goto out_unlock;
 

kmod/src/inode.c

@@ -892,7 +892,6 @@ static int delete_inode_items(struct super_block *sb, u64 ino)
 	struct scoutfs_inode sinode;
 	struct scoutfs_key_buf key;
 	SCOUTFS_DECLARE_KVEC(val);
-	DECLARE_ITEM_COUNT(cnt);
 	bool release = false;
 	umode_t mode;
 	int ret;
@@ -917,8 +916,7 @@ static int delete_inode_items(struct super_block *sb, u64 ino)
 	trace_delete_inode(sb, ino, mode);
 
 	/* XXX this is obviously not done yet :) */
-	scoutfs_count_dirty_inode(&cnt);
-	ret = scoutfs_hold_trans(sb, &cnt);
+	ret = scoutfs_hold_trans(sb, SIC_DIRTY_INODE());
 	if (ret)
 		goto out;
 	release = true;

kmod/src/trans.c

@@ -282,7 +282,7 @@ struct scoutfs_reservation {
  */
 static bool acquired_hold(struct super_block *sb,
 			  struct scoutfs_reservation *rsv,
-			  struct scoutfs_item_count *cnt)
+			  const struct scoutfs_item_count *cnt)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	DECLARE_TRANS_INFO(sb, tri);
@@ -340,7 +340,8 @@ out:
 	return acquired;
 }
 
-int scoutfs_hold_trans(struct super_block *sb, struct scoutfs_item_count *cnt)
+int scoutfs_hold_trans(struct super_block *sb,
+		       const struct scoutfs_item_count cnt)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_reservation *rsv;
@@ -350,9 +351,9 @@ int scoutfs_hold_trans(struct super_block *sb, struct scoutfs_item_count *cnt)
 	 * Caller shouldn't provide garbage counts, nor counts that
 	 * can't fit in segments by themselves.
 	 */
-	if (WARN_ON_ONCE(cnt->items <= 0 || cnt->keys < 0 || cnt->vals < 0) ||
-	    WARN_ON_ONCE(!scoutfs_seg_fits_single(cnt->items, cnt->keys,
-						  cnt->vals)))
+	if (WARN_ON_ONCE(cnt.items <= 0 || cnt.keys < 0 || cnt.vals < 0) ||
+	    WARN_ON_ONCE(!scoutfs_seg_fits_single(cnt.items, cnt.keys,
+						  cnt.vals)))
 		return -EINVAL;
 
 	if (current == sbi->trans_task)
@@ -371,7 +372,7 @@ int scoutfs_hold_trans(struct super_block *sb, struct scoutfs_item_count *cnt)
 	BUG_ON(rsv->magic != SCOUTFS_RESERVATION_MAGIC);
 
 	ret = wait_event_interruptible(sbi->trans_hold_wq,
-				       acquired_hold(sb, rsv, cnt));
+				       acquired_hold(sb, rsv, &cnt));
 	if (ret && rsv->holders == 0) {
 		current->journal_info = NULL;
 		kfree(rsv);

kmod/src/trans.h

@@ -9,7 +9,8 @@ int scoutfs_file_fsync(struct file *file, loff_t start, loff_t end,
 		       int datasync);
 void scoutfs_trans_restart_sync_deadline(struct super_block *sb);
 
-int scoutfs_hold_trans(struct super_block *sb, struct scoutfs_item_count *cnt);
+int scoutfs_hold_trans(struct super_block *sb,
+		       const struct scoutfs_item_count cnt);
 bool scoutfs_trans_held(void);
 void scoutfs_release_trans(struct super_block *sb);
 void scoutfs_trans_track_item(struct super_block *sb, signed items,

kmod/src/xattr.c

@@ -263,7 +263,6 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	struct scoutfs_xattr_val_header vh;
 	size_t name_len = strlen(name);
 	SCOUTFS_DECLARE_KVEC(val);
-	DECLARE_ITEM_COUNT(cnt);
 	struct scoutfs_lock *lck;
 	unsigned int bytes;
 	unsigned int off;
@@ -316,8 +315,7 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	else
 		sif = 0;
 
-	scoutfs_count_xattr_set(&cnt, name_len, size);
-	ret = scoutfs_hold_trans(sb, &cnt);
+	ret = scoutfs_hold_trans(sb, SIC_XATTR_SET(name_len, size));
 	if (ret)
 		goto unlock;