Merge pull request #51 from versity/zab/totl_xattr_tag

Zab/totl xattr tag

kmod/src/alloc.c

@@ -1526,8 +1526,8 @@ struct foreach_cb_args {
 	void *cb_arg;
 };
 
-static int alloc_btree_extent_item_cb(struct super_block *sb, struct scoutfs_key *key,
-				      void *val, int val_len, void *arg)
+static int alloc_btree_extent_item_cb(struct super_block *sb, struct scoutfs_key *key, u64 seq,
+				      u8 flags, void *val, int val_len, void *arg)
 {
 	struct foreach_cb_args *cba = arg;
 	struct scoutfs_extent ext;

kmod/src/btree.c

@@ -30,6 +30,7 @@
 #include "avl.h"
 #include "hash.h"
 #include "sort_priv.h"
+#include "forest.h"
 
 #include "scoutfs_trace.h"
 
@@ -502,9 +503,8 @@ static __le16 insert_value(struct scoutfs_btree_block *bt, __le16 item_off,
  * This only consumes free space.  It's safe to use references to block
  * structures after this call.
  */
-static void create_item(struct scoutfs_btree_block *bt,
-			struct scoutfs_key *key, void *val, unsigned val_len,
-			struct scoutfs_avl_node *parent, int cmp)
+static void create_item(struct scoutfs_btree_block *bt, struct scoutfs_key *key, u64 seq, u8 flags,
+			void *val, unsigned val_len, struct scoutfs_avl_node *parent, int cmp)
 {
 	struct scoutfs_btree_item *item;
 
@@ -516,6 +516,8 @@ static void create_item(struct scoutfs_btree_block *bt,
 	item = end_item(bt);
 
 	item->key = *key;
+	item->seq = cpu_to_le64(seq);
+	item->flags = flags;
 
 	scoutfs_avl_insert(&bt->item_root, parent, &item->node, cmp);
 	leaf_item_hash_insert(bt, item_key(item), ptr_off(bt, item));
@@ -558,6 +560,8 @@ static void delete_item(struct scoutfs_btree_block *bt,
 	/* move the final item into the deleted space */
 	if (end != item) {
 		item->key = end->key;
+		item->seq = end->seq;
+		item->flags = end->flags;
 		item->val_off = end->val_off;
 		item->val_len = end->val_len;
 		leaf_item_hash_change(bt, &end->key, ptr_off(bt, item),
@@ -606,8 +610,8 @@ static void move_items(struct scoutfs_btree_block *dst,
 		else
 			next = next_item(src, from);
 
-		create_item(dst, item_key(from), item_val(src, from),
-			    item_val_len(from), par, cmp);
+		create_item(dst, item_key(from), le64_to_cpu(from->seq), from->flags,
+			    item_val(src, from), item_val_len(from), par, cmp);
 
 		if (move_right) {
 			if (par)
@@ -680,7 +684,7 @@ static void create_parent_item(struct scoutfs_btree_block *parent,
 
 	scoutfs_avl_search(&parent->item_root, cmp_key_item, key, &cmp, &par,
 			   NULL, NULL);
-	create_item(parent, key, &ref, sizeof(ref), par, cmp);
+	create_item(parent, key, 0, 0, &ref, sizeof(ref), par, cmp);
 }
 
 /*
@@ -1529,7 +1533,7 @@ int scoutfs_btree_insert(struct super_block *sb,
 			if (node) {
 				ret = -EEXIST;
 			} else {
-				create_item(bt, key, val, val_len, par, cmp);
+				create_item(bt, key, 0, 0, val, val_len, par, cmp);
 				ret = 0;
 			}
 		}
@@ -1630,7 +1634,7 @@ int scoutfs_btree_force(struct super_block *sb,
 		} else {
 			scoutfs_avl_search(&bt->item_root, cmp_key_item, key,
 					   &cmp, &par, NULL, NULL);
-			create_item(bt, key, val, val_len, par, cmp);
+			create_item(bt, key, 0, 0, val, val_len, par, cmp);
 		}
 		ret = 0;
 
@@ -1849,8 +1853,8 @@ int scoutfs_btree_read_items(struct super_block *sb,
 		if (scoutfs_key_compare(&item->key, end) > 0)
 			break;
 
-		ret = cb(sb, item_key(item), item_val(bt, item),
-			 item_val_len(item), arg);
+		ret = cb(sb, item_key(item), le64_to_cpu(item->seq), item->flags,
+			 item_val(bt, item), item_val_len(item), arg);
 		if (ret < 0)
 			break;
 
@@ -1870,6 +1874,10 @@ out:
  * This can make partial progress before returning an error, leaving
  * dirty btree blocks with only some of the caller's items.  It's up to
  * the caller to resolve this.
+ *
+ * This, along with merging, are the only places that seq and flags are
+ * set in btree items.  They're only used for fs items written through
+ * the item cache and forest of log btrees.
  */
 int scoutfs_btree_insert_list(struct super_block *sb,
 			      struct scoutfs_alloc *alloc,
@@ -1895,13 +1903,28 @@ int scoutfs_btree_insert_list(struct super_block *sb,
 		do {
 			item = leaf_item_hash_search(sb, bt, &lst->key);
 			if (item) {
-				update_item_value(bt, item, lst->val,
-						  lst->val_len);
+				/* try to merge delta values, _NULL not deleted; merge will */
+				ret = scoutfs_forest_combine_deltas(&lst->key,
+								    item_val(bt, item),
+								    item_val_len(item),
+								    lst->val, lst->val_len);
+				if (ret < 0) {
+					scoutfs_block_put(sb, bl);
+					goto out;
+				}
+
+				item->seq = cpu_to_le64(lst->seq);
+				item->flags = lst->flags;
+
+				if (ret == 0)
+					update_item_value(bt, item, lst->val, lst->val_len);
+				else
+					ret = 0;
 			} else {
 				scoutfs_avl_search(&bt->item_root,
 						   cmp_key_item, &lst->key,
 						   &cmp, &par, NULL, NULL);
-				create_item(bt, &lst->key, lst->val,
+				create_item(bt, &lst->key, lst->seq, lst->flags, lst->val,
 					    lst->val_len, par, cmp);
 			}
 
@@ -2017,6 +2040,8 @@ struct merge_pos {
 	struct scoutfs_btree_block *bt;
 	struct scoutfs_avl_node *avl;
 	struct scoutfs_key *key;
+	u64 seq;
+	u8 flags;
 	unsigned int val_len;
 	u8 *val;
 };
@@ -2029,14 +2054,23 @@ static struct merge_pos *first_mpos(struct rb_root *root)
 	return NULL;
 }
 
+static struct merge_pos *next_mpos(struct merge_pos *mpos)
+{
+	struct rb_node *node;
+
+	if (mpos && (node = rb_next(&mpos->node)))
+		return container_of(node, struct merge_pos, node);
+	else
+		return NULL;
+}
+
 static void free_mpos(struct super_block *sb, struct merge_pos *mpos)
 {
 	scoutfs_block_put(sb, mpos->bl);
 	kfree(mpos);
 }
 
-static void insert_mpos(struct rb_root *pos_root, struct merge_pos *ins,
-			scoutfs_btree_merge_cmp_t merge_cmp)
+static void insert_mpos(struct rb_root *pos_root, struct merge_pos *ins)
 {
 	struct rb_node **node = &pos_root->rb_node;
 	struct rb_node *parent = NULL;
@@ -2050,7 +2084,7 @@ static void insert_mpos(struct rb_root *pos_root, struct merge_pos *ins,
 
 		/* sort merge items by key then newest to oldest */
 		cmp = scoutfs_key_compare(ins->key, mpos->key) ?:
-		      -merge_cmp(ins->val, ins->val_len, mpos->val, mpos->val_len);
+		      -scoutfs_cmp(ins->seq, mpos->seq);
 
 		if (cmp < 0)
 			node = &(*node)->rb_left;
@@ -2069,8 +2103,7 @@ static void insert_mpos(struct rb_root *pos_root, struct merge_pos *ins,
  * the mpos on error or if there are no more items in the range.
  */
 static int reset_mpos(struct super_block *sb, struct rb_root *pos_root, struct merge_pos *mpos,
-		      struct scoutfs_key *start, struct scoutfs_key *end,
-		      scoutfs_btree_merge_cmp_t merge_cmp)
+		      struct scoutfs_key *start, struct scoutfs_key *end)
 {
 	struct scoutfs_btree_item *item;
 	struct scoutfs_avl_node *next;
@@ -2123,31 +2156,76 @@ static int reset_mpos(struct super_block *sb, struct rb_root *pos_root, struct m
 
 	/* insert the next item within range at its version */
 	mpos->key = item_key(item);
+	mpos->seq = le64_to_cpu(item->seq);
+	mpos->flags = item->flags;
 	mpos->val_len = item_val_len(item);
 	mpos->val = item_val(mpos->bt, item);
 
-	insert_mpos(pos_root, mpos, merge_cmp);
+	insert_mpos(pos_root, mpos);
 	ret = 0;
 out:
 	return ret;
 }
 
+/*
+ * The caller has reset all the merge positions for all the input log
+ * btree roots and wants the next logged item it should try and merge
+ * with the items in the fs_root.
+ *
+ * We look ahead in the logged item stream to see if we should merge any
+ * older logged delta items into one result for the caller.  We also
+ * take this opportunity to skip and reset the mpos for any older
+ * versions of the first item.
+ */
+static int next_resolved_mpos(struct super_block *sb, struct rb_root *pos_root,
+			      struct scoutfs_key *end, struct merge_pos **mpos_ret)
+{
+	struct merge_pos *mpos;
+	struct merge_pos *next;
+	struct scoutfs_key key;
+	int ret = 0;
+
+	while ((mpos = first_mpos(pos_root)) && (next = next_mpos(mpos)) &&
+	       !scoutfs_key_compare(mpos->key, next->key)) {
+
+		ret = scoutfs_forest_combine_deltas(mpos->key, mpos->val, mpos->val_len,
+						    next->val, next->val_len);
+		if (ret < 0)
+			break;
+
+		/* reset advances to the next item */
+		key = *mpos->key;
+		scoutfs_key_inc(&key);
+
+		/* always skip next combined or older version */
+		ret = reset_mpos(sb, pos_root, next, &key, end);
+		if (ret < 0)
+			break;
+
+		if (ret == SCOUTFS_DELTA_COMBINED) {
+			scoutfs_inc_counter(sb, btree_merge_delta_combined);
+		} else if (ret == SCOUTFS_DELTA_COMBINED_NULL) {
+			scoutfs_inc_counter(sb, btree_merge_delta_null);
+			/* if merging resulted in no info, skip current */
+			ret = reset_mpos(sb, pos_root, mpos, &key, end);
+			if (ret < 0)
+				break;
+		}
+	}
+
+	*mpos_ret = mpos;
+	return ret;
+}
+
 /*
  * Merge items from a number of read-only input roots into a writable
  * destination root.  The order of the input roots doesn't matter, the
  * items are merged in sorted key order.
  *
- * The merge_cmp callback determines the order that the input items are
- * merged in.  The is_del callback determines if a merging item should
- * be removed from the destination.
- *
  * subtree indicates that the destination root is in fact one of many
  * parent blocks and shouldn't be split or allowed to fall below the
  * join low water mark.
  *
- * drop_val indicates the initial length of the value that should be
- * dropped when merging items into destination items.
- *
  * -ERANGE is returned if the merge doesn't fully exhaust the range, due
  * to allocators running low or needing to join/split the parent.
  * *next_ret is set to the next key which hasn't been merged so that the
@@ -2161,9 +2239,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 			struct scoutfs_key *next_ret,
 			struct scoutfs_btree_root *root,
 			struct list_head *inputs,
-			scoutfs_btree_merge_cmp_t merge_cmp,
-			scoutfs_btree_merge_is_del_t merge_is_del, bool subtree,
-			int drop_val, int dirty_limit, int alloc_low)
+			bool subtree, int dirty_limit, int alloc_low)
 {
 	struct scoutfs_btree_root_head *rhead;
 	struct rb_root pos_root = RB_ROOT;
@@ -2178,6 +2254,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 	int walk_val_len;
 	int walk_flags;
 	bool is_del;
+	int delta;
 	int cmp;
 	int ret;
 
@@ -2194,7 +2271,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 		RB_CLEAR_NODE(&mpos->node);
 		mpos->root = &rhead->root;
 
-		ret = reset_mpos(sb, &pos_root, mpos, start, end, merge_cmp);
+		ret = reset_mpos(sb, &pos_root, mpos, start, end);
 		if (ret < 0)
 			goto out;
 	}
@@ -2204,7 +2281,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 		walk_flags |= BTW_SUBTREE;
 	walk_val_len = 0;
 
-	while ((mpos = first_mpos(&pos_root))) {
+	while ((ret = next_resolved_mpos(sb, &pos_root, end, &mpos)) == 0 && mpos) {
 
 		if (scoutfs_block_writer_dirty_bytes(sb, wri) >= dirty_limit) {
 			scoutfs_inc_counter(sb, btree_merge_dirty_limit);
@@ -2232,13 +2309,13 @@ int scoutfs_btree_merge(struct super_block *sb,
 		bt = bl->data;
 		scoutfs_inc_counter(sb, btree_merge_walk);
 
-		for (; mpos; mpos = first_mpos(&pos_root)) {
+		/* catch non-root blocks that fell under low, maybe from null deltas */
+		if (root->ref.blkno != bt->hdr.blkno && !total_above_join_low_water(bt)) {
+			walk_flags |= BTW_DELETE;
+			continue;
+		}
 
-			/* val must have at least what we need to drop */
-			if (mpos->val_len < drop_val) {
-				ret = -EIO;
-				goto out;
-			}
+		while ((ret = next_resolved_mpos(sb, &pos_root, end, &mpos)) == 0 && mpos) {
 
 			/* walk to new leaf if we exceed parent ref key */
 			if (scoutfs_key_compare(mpos->key, &kr.end) > 0)
@@ -2246,7 +2323,24 @@ int scoutfs_btree_merge(struct super_block *sb,
 
 			/* see if there's an existing item */
 			item = leaf_item_hash_search(sb, bt, mpos->key);
-			is_del = merge_is_del(mpos->val, mpos->val_len);
+			is_del = !!(mpos->flags & SCOUTFS_ITEM_FLAG_DELETION);
+
+			/* see if we're merging delta items */
+			if (item && !is_del)
+				delta = scoutfs_forest_combine_deltas(mpos->key,
+								      item_val(bt, item),
+								      item_val_len(item),
+								      mpos->val, mpos->val_len);
+			else
+				delta = 0;
+			if (delta < 0) {
+				ret = delta;
+				goto out;
+			} else if (delta == SCOUTFS_DELTA_COMBINED) {
+				scoutfs_inc_counter(sb, btree_merge_delta_combined);
+			} else if (delta == SCOUTFS_DELTA_COMBINED_NULL) {
+				scoutfs_inc_counter(sb, btree_merge_delta_null);
+			}
 
 			trace_scoutfs_btree_merge_items(sb, mpos->root,
 					mpos->key, mpos->val_len,
@@ -2255,7 +2349,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 					item ? item_val_len(item) : 0, is_del);
 
 			/* rewalk and split if ins/update needs room */
-			if (!is_del && !mid_free_item_room(bt, mpos->val_len)) {
+			if (!is_del && !delta && !mid_free_item_room(bt, mpos->val_len)) {
 				walk_flags |= BTW_INSERT;
 				walk_val_len = mpos->val_len;
 				break;
@@ -2266,20 +2360,37 @@ int scoutfs_btree_merge(struct super_block *sb,
 				scoutfs_avl_search(&bt->item_root,
 						   cmp_key_item, mpos->key,
 						   &cmp, &par, NULL, NULL);
-				create_item(bt, mpos->key,
-					    mpos->val + drop_val,
-					    mpos->val_len - drop_val, par, cmp);
+				create_item(bt, mpos->key, mpos->seq, mpos->flags,
+					    mpos->val, mpos->val_len, par, cmp);
 				scoutfs_inc_counter(sb, btree_merge_insert);
 			}
 
 			/* update existing items */
-			if (item && !is_del) {
-				update_item_value(bt, item,
-						  mpos->val + drop_val,
-						  mpos->val_len - drop_val);
+			if (item && !is_del && !delta) {
+				item->seq = cpu_to_le64(mpos->seq);
+				item->flags = mpos->flags;
+				update_item_value(bt, item, mpos->val, mpos->val_len);
 				scoutfs_inc_counter(sb, btree_merge_update);
 			}
 
+			/* update combined delta item seq */
+			if (delta == SCOUTFS_DELTA_COMBINED) {
+				item->seq = cpu_to_le64(mpos->seq);
+			}
+
+			/*
+			 * combined delta items that aren't needed are
+			 * immediately dropped.  We don't back off if
+			 * the deletion would fall under the low water
+			 * mark because we've already modified the
+			 * value, we don't want to retry after a join
+			 * and apply the value a second time.
+			 */
+			if (delta == SCOUTFS_DELTA_COMBINED_NULL) {
+				delete_item(bt, item, NULL);
+				scoutfs_inc_counter(sb, btree_merge_delta_null);
+			}
+
 			/* delete if merge item was deletion */
 			if (item && is_del) {
 				/* rewalk and join if non-root falls under low water mark */
@@ -2299,12 +2410,9 @@ int scoutfs_btree_merge(struct super_block *sb,
 			/* finished with this key, skip any older items */
 			next = *mpos->key;
 			scoutfs_key_inc(&next);
-			while (mpos && scoutfs_key_compare(mpos->key, &next) < 0) {
-				ret = reset_mpos(sb, &pos_root, mpos, &next, end, merge_cmp);
-				if (ret < 0)
-					goto out;
-				mpos = first_mpos(&pos_root);
-			}
+			ret = reset_mpos(sb, &pos_root, mpos, &next, end);
+			if (ret < 0)
+				goto out;
 		}
 	}
 

kmod/src/btree.h

@@ -20,13 +20,15 @@ struct scoutfs_btree_item_ref {
 
 /* caller gives an item to the callback */
 typedef int (*scoutfs_btree_item_cb)(struct super_block *sb,
-				     struct scoutfs_key *key,
+				     struct scoutfs_key *key, u64 seq, u8 flags,
 				     void *val, int val_len, void *arg);
 
 /* simple singly-linked list of items */
 struct scoutfs_btree_item_list {
 	struct scoutfs_btree_item_list *next;
 	struct scoutfs_key key;
+	u64 seq;
+	u8 flags;
 	int val_len;
 	u8 val[0];
 };
@@ -108,14 +110,7 @@ struct scoutfs_btree_root_head {
 	struct list_head head;
 	struct scoutfs_btree_root root;
 };
-/*
- * Compare the values of merge input items whose keys are equal to
- * determine their merge order.
- */
-typedef int (*scoutfs_btree_merge_cmp_t)(void *a_val, int a_val_len,
-					 void *b_val, int b_val_len);
-/* whether merging item should be removed from destination */
-typedef bool (*scoutfs_btree_merge_is_del_t)(void *val, int val_len);
+
 int scoutfs_btree_merge(struct super_block *sb,
 			struct scoutfs_alloc *alloc,
 			struct scoutfs_block_writer *wri,
@@ -124,9 +119,7 @@ int scoutfs_btree_merge(struct super_block *sb,
 			struct scoutfs_key *next_ret,
 			struct scoutfs_btree_root *root,
 			struct list_head *input_list,
-			scoutfs_btree_merge_cmp_t merge_cmp,
-			scoutfs_btree_merge_is_del_t merge_is_del, bool subtree,
-			int drop_val, int dirty_limit, int alloc_low);
+			bool subtree, int dirty_limit, int alloc_low);
 
 int scoutfs_btree_free_blocks(struct super_block *sb,
 			      struct scoutfs_alloc *alloc,

kmod/src/counters.h

@@ -47,6 +47,8 @@
 	EXPAND_COUNTER(btree_merge)				\
 	EXPAND_COUNTER(btree_merge_alloc_low)			\
 	EXPAND_COUNTER(btree_merge_delete)			\
+	EXPAND_COUNTER(btree_merge_delta_combined)		\
+	EXPAND_COUNTER(btree_merge_delta_null)			\
 	EXPAND_COUNTER(btree_merge_dirty_limit)			\
 	EXPAND_COUNTER(btree_merge_drop_old)			\
 	EXPAND_COUNTER(btree_merge_insert)			\
@@ -91,6 +93,8 @@
 	EXPAND_COUNTER(item_clear_dirty)			\
 	EXPAND_COUNTER(item_create)				\
 	EXPAND_COUNTER(item_delete)				\
+	EXPAND_COUNTER(item_delta)				\
+	EXPAND_COUNTER(item_delta_written)			\
 	EXPAND_COUNTER(item_dirty)				\
 	EXPAND_COUNTER(item_invalidate)				\
 	EXPAND_COUNTER(item_invalidate_page)			\
@@ -188,6 +192,11 @@
 	EXPAND_COUNTER(srch_search_xattrs)			\
 	EXPAND_COUNTER(srch_read_stale)				\
 	EXPAND_COUNTER(statfs)					\
+	EXPAND_COUNTER(totl_read_copied)			\
+	EXPAND_COUNTER(totl_read_finalized)			\
+	EXPAND_COUNTER(totl_read_fs)				\
+	EXPAND_COUNTER(totl_read_item)				\
+	EXPAND_COUNTER(totl_read_logged)			\
 	EXPAND_COUNTER(trans_commit_data_alloc_low)		\
 	EXPAND_COUNTER(trans_commit_dirty_meta_full)		\
 	EXPAND_COUNTER(trans_commit_fsync)			\

kmod/src/data.c

@@ -830,6 +830,7 @@ static int scoutfs_write_end(struct file *file, struct address_space *mapping,
 			scoutfs_inode_inc_data_version(inode);
 		}
 
+		inode_inc_iversion(inode);
 		scoutfs_update_inode_item(inode, wbd->lock, &wbd->ind_locks);
 		scoutfs_inode_queue_writeback(inode);
 	}
@@ -1033,6 +1034,7 @@ long scoutfs_fallocate(struct file *file, int mode, loff_t offset, loff_t len)
 				end = offset + len;
 			if (end > i_size_read(inode)) {
 				i_size_write(inode, end);
+				inode_inc_iversion(inode);
 				scoutfs_inode_inc_data_version(inode);
 			}
 		}
@@ -1366,10 +1368,12 @@ int scoutfs_data_move_blocks(struct inode *from, u64 from_off,
 		cur_time = CURRENT_TIME;
 		if (!is_stage) {
 			to->i_ctime = to->i_mtime = cur_time;
+			inode_inc_iversion(to);
 			scoutfs_inode_inc_data_version(to);
 			scoutfs_inode_set_data_seq(to);
 		}
 		from->i_ctime = from->i_mtime = cur_time;
+		inode_inc_iversion(from);
 		scoutfs_inode_inc_data_version(from);
 		scoutfs_inode_set_data_seq(from);
 

kmod/src/dir.c

@@ -135,8 +135,8 @@ static int alloc_dentry_info(struct dentry *dentry)
 {
 	struct dentry_info *di;
 
-	/* XXX read mb? */
-	if (dentry->d_fsdata)
+	smp_rmb();
+	if (dentry->d_op == &scoutfs_dentry_ops)
 		return 0;
 
 	di = kmem_cache_zalloc(dentry_info_cache, GFP_NOFS);
@@ -148,6 +148,7 @@ static int alloc_dentry_info(struct dentry *dentry)
 	spin_lock(&dentry->d_lock);
 	if (!dentry->d_fsdata) {
 		dentry->d_fsdata = di;
+		smp_wmb();
 		d_set_d_op(dentry, &scoutfs_dentry_ops);
 	}
 	spin_unlock(&dentry->d_lock);
@@ -801,6 +802,7 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 	struct inode *inode = NULL;
 	struct scoutfs_lock *dir_lock = NULL;
 	struct scoutfs_lock *inode_lock = NULL;
+	struct scoutfs_inode_info *si;
 	LIST_HEAD(ind_locks);
 	u64 hash;
 	u64 pos;
@@ -814,6 +816,7 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 				 &dir_lock, &inode_lock, NULL, &ind_locks);
 	if (IS_ERR(inode))
 		return PTR_ERR(inode);
+	si = SCOUTFS_I(inode);
 
 	ret = verify_entry(sb, scoutfs_ino(dir), dentry, dir_lock);
 	if (ret < 0)
@@ -833,6 +836,9 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode,
 	i_size_write(dir, i_size_read(dir) + dentry->d_name.len);
 	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
 	inode->i_mtime = inode->i_atime = inode->i_ctime = dir->i_mtime;
+	si->crtime = inode->i_mtime;
+	inode_inc_iversion(dir);
+	inode_inc_iversion(inode);
 
 	if (S_ISDIR(mode)) {
 		inc_nlink(inode);
@@ -898,10 +904,6 @@ static int scoutfs_link(struct dentry *old_dentry,
 	if (ret)
 		return ret;
 
-	ret = verify_entry(sb, scoutfs_ino(dir), dentry, dir_lock);
-	if (ret < 0)
-		goto out_unlock;
-
 	if (inode->i_nlink >= SCOUTFS_LINK_MAX) {
 		ret = -EMLINK;
 		goto out_unlock;
@@ -911,6 +913,10 @@ static int scoutfs_link(struct dentry *old_dentry,
 	if (ret)
 		goto out_unlock;
 
+	ret = verify_entry(sb, scoutfs_ino(dir), dentry, dir_lock);
+	if (ret < 0)
+		goto out_unlock;
+
 	dir_size = i_size_read(dir) + dentry->d_name.len;
 
 	if (inode->i_nlink == 0) {
@@ -958,6 +964,8 @@ retry:
 	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
 	inode->i_ctime = dir->i_mtime;
 	inc_nlink(inode);
+	inode_inc_iversion(dir);
+	inode_inc_iversion(inode);
 
 	scoutfs_update_inode_item(inode, inode_lock, &ind_locks);
 	scoutfs_update_inode_item(dir, dir_lock, &ind_locks);
@@ -1009,6 +1017,10 @@ static int scoutfs_unlink(struct inode *dir, struct dentry *dentry)
 	if (ret)
 		return ret;
 
+	ret = alloc_dentry_info(dentry);
+	if (ret)
+		goto unlock;
+
 	ret = verify_entry(sb, scoutfs_ino(dir), dentry, dir_lock);
 	if (ret < 0)
 		goto unlock;
@@ -1055,6 +1067,8 @@ retry:
 	dir->i_ctime = ts;
 	dir->i_mtime = ts;
 	i_size_write(dir, i_size_read(dir) - dentry->d_name.len);
+	inode_inc_iversion(dir);
+	inode_inc_iversion(inode);
 
 	inode->i_ctime = ts;
 	drop_nlink(inode);
@@ -1247,6 +1261,7 @@ 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;
+	struct scoutfs_inode_info *si;
 	LIST_HEAD(ind_locks);
 	u64 hash;
 	u64 pos;
@@ -1267,6 +1282,7 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 				 &dir_lock, &inode_lock, NULL, &ind_locks);
 	if (IS_ERR(inode))
 		return PTR_ERR(inode);
+	si = SCOUTFS_I(inode);
 
 	ret = verify_entry(sb, scoutfs_ino(dir), dentry, dir_lock);
 	if (ret < 0)
@@ -1290,9 +1306,12 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry,
 
 	i_size_write(dir, i_size_read(dir) + dentry->d_name.len);
 	dir->i_mtime = dir->i_ctime = CURRENT_TIME;
+	inode_inc_iversion(dir);
 
 	inode->i_ctime = dir->i_mtime;
+	si->crtime = inode->i_ctime;
 	i_size_write(inode, name_len);
+	inode_inc_iversion(inode);
 
 	scoutfs_update_inode_item(inode, inode_lock, &ind_locks);
 	scoutfs_update_inode_item(dir, dir_lock, &ind_locks);
@@ -1662,7 +1681,9 @@ static int scoutfs_rename(struct inode *old_dir, struct dentry *old_dentry,
 	}
 
 	/* make sure that the entries assumed by the argument still exist */
-	ret = verify_entry(sb, scoutfs_ino(old_dir), old_dentry, old_dir_lock) ?:
+	ret = alloc_dentry_info(old_dentry) ?:
+	      alloc_dentry_info(new_dentry) ?:
+	      verify_entry(sb, scoutfs_ino(old_dir), old_dentry, old_dir_lock) ?:
 	      verify_entry(sb, scoutfs_ino(new_dir), new_dentry, new_dir_lock);
 	if (ret)
 		goto out_unlock;
@@ -1773,6 +1794,13 @@ retry:
 	if (new_inode)
 		old_inode->i_ctime = now;
 
+	inode_inc_iversion(old_dir);
+	inode_inc_iversion(old_inode);
+	if (new_dir != old_dir)
+		inode_inc_iversion(new_dir);
+	if (new_inode)
+		inode_inc_iversion(new_inode);
+
 	scoutfs_update_inode_item(old_dir, old_dir_lock, &ind_locks);
 	scoutfs_update_inode_item(old_inode, old_inode_lock, &ind_locks);
 	if (new_dir != old_dir)
@@ -1858,6 +1886,7 @@ static int scoutfs_tmpfile(struct inode *dir, struct dentry *dentry, umode_t mod
 	struct scoutfs_lock *dir_lock = NULL;
 	struct scoutfs_lock *inode_lock = NULL;
 	struct scoutfs_lock *orph_lock = NULL;
+	struct scoutfs_inode_info *si;
 	LIST_HEAD(ind_locks);
 	int ret;
 
@@ -1868,6 +1897,7 @@ static int scoutfs_tmpfile(struct inode *dir, struct dentry *dentry, umode_t mod
 				 &dir_lock, &inode_lock, &orph_lock, &ind_locks);
 	if (IS_ERR(inode))
 		return PTR_ERR(inode);
+	si = SCOUTFS_I(inode);
 
 	ret = scoutfs_inode_orphan_create(sb, scoutfs_ino(inode), orph_lock);
 	if (ret < 0) {
@@ -1876,9 +1906,11 @@ static int scoutfs_tmpfile(struct inode *dir, struct dentry *dentry, umode_t mod
 	}
 
 	inode->i_mtime = inode->i_atime = inode->i_ctime = CURRENT_TIME;
+	si->crtime = inode->i_mtime;
 	insert_inode_hash(inode);
 	ihold(inode); /* need to update inode modifications in d_tmpfile */
 	d_tmpfile(dentry, inode);
+	inode_inc_iversion(inode);
 
 	scoutfs_update_inode_item(inode, inode_lock, &ind_locks);
 	scoutfs_update_inode_item(dir, dir_lock, &ind_locks);

kmod/src/forest.c

@@ -26,6 +26,7 @@
 #include "hash.h"
 #include "srch.h"
 #include "counters.h"
+#include "xattr.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -221,25 +222,17 @@ out:
 }
 
 struct forest_read_items_data {
-	bool is_fs;
+	int fic;
 	scoutfs_forest_item_cb cb;
 	void *cb_arg;
 };
 
-static int forest_read_items(struct super_block *sb, struct scoutfs_key *key,
+static int forest_read_items(struct super_block *sb, struct scoutfs_key *key, u64 seq, u8 flags,
 			     void *val, int val_len, void *arg)
 {
 	struct forest_read_items_data *rid = arg;
-	struct scoutfs_log_item_value _liv = {0,};
-	struct scoutfs_log_item_value *liv = &_liv;
 
-	if (!rid->is_fs) {
-		liv = val;
-		val += sizeof(struct scoutfs_log_item_value);
-		val_len -= sizeof(struct scoutfs_log_item_value);
-	}
-
-	return rid->cb(sb, key, liv, val, val_len, rid->cb_arg);
+	return rid->cb(sb, key, seq, flags, val, val_len, rid->fic, rid->cb_arg);
 }
 
 /*
@@ -255,8 +248,8 @@ static int forest_read_items(struct super_block *sb, struct scoutfs_key *key,
  * to reset their state and retry with a newer version of the btrees.
  */
 int scoutfs_forest_read_items(struct super_block *sb,
-			      struct scoutfs_lock *lock,
 			      struct scoutfs_key *key,
+			      struct scoutfs_key *bloom_key,
 			      struct scoutfs_key *start,
 			      struct scoutfs_key *end,
 			      scoutfs_forest_item_cb cb, void *arg)
@@ -272,11 +265,13 @@ int scoutfs_forest_read_items(struct super_block *sb,
 	SCOUTFS_BTREE_ITEM_REF(iref);
 	struct scoutfs_block *bl;
 	struct scoutfs_key ltk;
+	struct scoutfs_key orig_start = *start;
+	struct scoutfs_key orig_end = *end;
 	int ret;
 	int i;
 
 	scoutfs_inc_counter(sb, forest_read_items);
-	calc_bloom_nrs(&bloom, &lock->start);
+	calc_bloom_nrs(&bloom, bloom_key);
 
 	ret = scoutfs_client_get_roots(sb, &roots);
 	if (ret)
@@ -284,16 +279,16 @@ int scoutfs_forest_read_items(struct super_block *sb,
 
 	trace_scoutfs_forest_using_roots(sb, &roots.fs_root, &roots.logs_root);
 
-	*start = lock->start;
-	*end = lock->end;
+	*start = orig_start;
+	*end = orig_end;
 
 	/* start with fs root items */
-	rid.is_fs = true;
+	rid.fic |= FIC_FS_ROOT;
 	ret = scoutfs_btree_read_items(sb, &roots.fs_root, key, start, end,
 				       forest_read_items, &rid);
 	if (ret < 0)
 		goto out;
-	rid.is_fs = false;
+	rid.fic &= ~FIC_FS_ROOT;
 
 	scoutfs_key_init_log_trees(&ltk, 0, 0);
 	for (;; scoutfs_key_inc(&ltk)) {
@@ -338,10 +333,15 @@ int scoutfs_forest_read_items(struct super_block *sb,
 
 		scoutfs_inc_counter(sb, forest_bloom_pass);
 
+		if ((le64_to_cpu(lt.flags) & SCOUTFS_LOG_TREES_FINALIZED))
+			rid.fic |= FIC_FINALIZED;
+
 		ret = scoutfs_btree_read_items(sb, &lt.item_root, key, start,
 					       end, forest_read_items, &rid);
 		if (ret < 0)
 			goto out;
+
+		rid.fic &= ~FIC_FINALIZED;
 	}
 
 	ret = 0;
@@ -349,6 +349,24 @@ out:
 	return ret;
 }
 
+/*
+ * If the items are deltas then combine the src with the destination
+ * value and store the result in the destination.
+ *
+ * Returns:
+ *  -errno: fatal error, no change
+ *  0: not delta items, no change
+ *  +ve: SCOUTFS_DELTA_ values indicating when dst and/or src can be dropped
+ */
+int scoutfs_forest_combine_deltas(struct scoutfs_key *key, void *dst, int dst_len,
+				  void *src, int src_len)
+{
+	if (key->sk_zone == SCOUTFS_XATTR_TOTL_ZONE)
+		return scoutfs_xattr_combine_totl(dst, dst_len, src, src_len);
+
+	return 0;
+}
+
 /*
  * Make sure that the bloom bits for the lock's start key are all set in
  * the current log's bloom block.  We record the nr of our log tree in
@@ -564,26 +582,6 @@ void scoutfs_forest_get_btrees(struct super_block *sb,
 					    &lt->bloom_ref);
 }
 
-/*
- * Compare input items to merge by their log item value seq when their
- * keys match.
- */
-static int merge_cmp(void *a_val, int a_val_len, void *b_val, int b_val_len)
-{
-	struct scoutfs_log_item_value *a = a_val;
-	struct scoutfs_log_item_value *b = b_val;
-
-	/* sort merge item by seq */
-	return scoutfs_cmp(le64_to_cpu(a->seq), le64_to_cpu(b->seq));
-}
-
-static bool merge_is_del(void *val, int val_len)
-{
-	struct scoutfs_log_item_value *liv = val;
-
-	return !!(liv->flags & SCOUTFS_LOG_ITEM_FLAG_DELETION);
-}
-
 #define LOG_MERGE_DELAY_MS (5 * MSEC_PER_SEC)
 
 /*
@@ -673,10 +671,8 @@ static void scoutfs_forest_log_merge_worker(struct work_struct *work)
 	}
 
 	ret = scoutfs_btree_merge(sb, &alloc, &wri, &req.start, &req.end,
-				  &next, &comp.root, &inputs, merge_cmp,
-				  merge_is_del,
+				  &next, &comp.root, &inputs,
 				  !!(req.flags & cpu_to_le64(SCOUTFS_LOG_MERGE_REQUEST_SUBTREE)),
-				  sizeof(struct scoutfs_log_item_value),
 				  SCOUTFS_LOG_MERGE_DIRTY_BYTE_LIMIT, 10);
 	if (ret == -ERANGE) {
 		comp.remain = next;

kmod/src/forest.h

@@ -8,16 +8,18 @@ struct scoutfs_block;
 #include "btree.h"
 
 /* caller gives an item to the callback */
-typedef int (*scoutfs_forest_item_cb)(struct super_block *sb,
-				      struct scoutfs_key *key,
-				      struct scoutfs_log_item_value *liv,
-				      void *val, int val_len, void *arg);
+enum {
+	FIC_FS_ROOT = (1 << 0),
+	FIC_FINALIZED = (1 << 1),
+};
+typedef int (*scoutfs_forest_item_cb)(struct super_block *sb, struct scoutfs_key *key, u64 seq,
+				      u8 flags, void *val, int val_len, int fic, void *arg);
 
 int scoutfs_forest_next_hint(struct super_block *sb, struct scoutfs_key *key,
 			     struct scoutfs_key *next);
 int scoutfs_forest_read_items(struct super_block *sb,
-			      struct scoutfs_lock *lock,
 			      struct scoutfs_key *key,
+			      struct scoutfs_key *bloom_key,
 			      struct scoutfs_key *start,
 			      struct scoutfs_key *end,
 			      scoutfs_forest_item_cb cb, void *arg);
@@ -38,6 +40,12 @@ void scoutfs_forest_init_btrees(struct super_block *sb,
 void scoutfs_forest_get_btrees(struct super_block *sb,
 			       struct scoutfs_log_trees *lt);
 
+/* > 0 error codes */
+#define SCOUTFS_DELTA_COMBINED		1	/* src val was combined, drop src */
+#define SCOUTFS_DELTA_COMBINED_NULL	2	/* combined val has no data, drop both */
+int scoutfs_forest_combine_deltas(struct scoutfs_key *key, void *dst, int dst_len,
+				  void *src, int src_len);
+
 int scoutfs_forest_setup(struct super_block *sb);
 void scoutfs_forest_start(struct super_block *sb);
 void scoutfs_forest_stop(struct super_block *sb);

kmod/src/format.h

@@ -168,6 +168,11 @@ struct scoutfs_key {
 #define sko_rid		_sk_first
 #define sko_ino		_sk_second
 
+/* xattr totl */
+#define skxt_a		_sk_first
+#define skxt_b		_sk_second
+#define skxt_c		_sk_third
+
 /* inode */
 #define ski_ino		_sk_first
 
@@ -244,11 +249,15 @@ struct scoutfs_btree_root {
 struct scoutfs_btree_item {
 	struct scoutfs_avl_node node;
 	struct scoutfs_key key;
+	__le64 seq;
 	__le16 val_off;
 	__le16 val_len;
-	__u8 __pad[4];
+	__u8 flags;
+	__u8 __pad[3];
 };
 
+#define SCOUTFS_ITEM_FLAG_DELETION (1 << 0)
+
 struct scoutfs_btree_block {
 	struct scoutfs_block_header hdr;
 	struct scoutfs_avl_root item_root;
@@ -465,21 +474,8 @@ struct scoutfs_log_trees {
 
 #define SCOUTFS_LOG_TREES_FINALIZED	(1ULL << 0)
 
-struct scoutfs_log_item_value {
-	__le64 seq;
-	__u8 flags;
-	__u8 __pad[7];
-	__u8 data[];
-};
-
-/*
- * FS items are limited by the max btree value length with the log item
- * value header.
- */
-#define SCOUTFS_MAX_VAL_SIZE \
-	(SCOUTFS_BTREE_MAX_VAL_LEN - sizeof(struct scoutfs_log_item_value))
-
-#define SCOUTFS_LOG_ITEM_FLAG_DELETION		(1 << 0)
+/* FS items are limited by the max btree value length */
+#define SCOUTFS_MAX_VAL_SIZE	SCOUTFS_BTREE_MAX_VAL_LEN
 
 struct scoutfs_bloom_block {
 	struct scoutfs_block_header hdr;
@@ -577,8 +573,9 @@ struct scoutfs_log_merge_freeing {
  */
 #define SCOUTFS_INODE_INDEX_ZONE		1
 #define SCOUTFS_ORPHAN_ZONE			2
-#define SCOUTFS_FS_ZONE				3
-#define SCOUTFS_LOCK_ZONE			4
+#define SCOUTFS_XATTR_TOTL_ZONE			3
+#define SCOUTFS_FS_ZONE				4
+#define SCOUTFS_LOCK_ZONE			5
 /* Items only stored in server btrees */
 #define SCOUTFS_LOG_TREES_ZONE			6
 #define SCOUTFS_TRANS_SEQ_ZONE			7
@@ -642,6 +639,17 @@ struct scoutfs_xattr {
 	__u8 name[];
 };
 
+/*
+ * .totl. xattrs are mapped to items.  The dotted u64s in the xattr name
+ * map to the item key.  The item value total is the sum of all the
+ * xattr values.   The item value count records the number of xattrs
+ * contributing to the total and is used when combining logged items to
+ * determine if totals are being created or destroyed.
+ */
+struct scoutfs_xattr_totl_val {
+	__le64 total;
+	__le64 count;
+};
 
 /* XXX does this exist upstream somewhere? */
 #define member_sizeof(TYPE, MEMBER) (sizeof(((TYPE *)0)->MEMBER))
@@ -815,13 +823,6 @@ struct scoutfs_super_block {
  *
  * @offline_blocks: The number of fixed 4k blocks that could be made
  * online by staging.
- *
- * XXX
- *	- otime?
- *	- compat flags?
- *	- version?
- *	- generation?
- *	- be more careful with rdev?
  */
 struct scoutfs_inode {
 	__le64 size;
@@ -832,6 +833,7 @@ struct scoutfs_inode {
 	__le64 offline_blocks;
 	__le64 next_readdir_pos;
 	__le64 next_xattr_id;
+	__le64 version;
 	__le32 nlink;
 	__le32 uid;
 	__le32 gid;
@@ -841,6 +843,7 @@ struct scoutfs_inode {
 	struct scoutfs_timespec atime;
 	struct scoutfs_timespec ctime;
 	struct scoutfs_timespec mtime;
+	struct scoutfs_timespec crtime;
 };
 
 #define SCOUTFS_INO_FLAG_TRUNCATE 0x1
@@ -892,6 +895,7 @@ enum scoutfs_dentry_type {
 #define SCOUTFS_XATTR_MAX_NAME_LEN	255
 #define SCOUTFS_XATTR_MAX_VAL_LEN	65535
 #define SCOUTFS_XATTR_MAX_PART_SIZE	SCOUTFS_MAX_VAL_SIZE
+#define SCOUTFS_XATTR_MAX_TOTL_U64	23 /* octal U64_MAX */
 
 #define SCOUTFS_XATTR_NR_PARTS(name_len, val_len)			\
 	DIV_ROUND_UP(sizeof(struct scoutfs_xattr) + name_len + val_len, \

kmod/src/inode.c

@@ -212,6 +212,7 @@ static void load_inode(struct inode *inode, struct scoutfs_inode *cinode)
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 
 	i_size_write(inode, le64_to_cpu(cinode->size));
+	inode->i_version = le64_to_cpu(cinode->version);
 	set_nlink(inode, le32_to_cpu(cinode->nlink));
 	i_uid_write(inode, le32_to_cpu(cinode->uid));
 	i_gid_write(inode, le32_to_cpu(cinode->gid));
@@ -232,6 +233,8 @@ static void load_inode(struct inode *inode, struct scoutfs_inode *cinode)
 	si->next_readdir_pos = le64_to_cpu(cinode->next_readdir_pos);
 	si->next_xattr_id = le64_to_cpu(cinode->next_xattr_id);
 	si->flags = le32_to_cpu(cinode->flags);
+	si->crtime.tv_sec = le64_to_cpu(cinode->crtime.sec);
+	si->crtime.tv_nsec = le32_to_cpu(cinode->crtime.nsec);
 
 	/*
 	 * i_blocks is initialized from online and offline and is then
@@ -344,6 +347,7 @@ static int set_inode_size(struct inode *inode, struct scoutfs_lock *lock,
 	if (truncate)
 		si->flags |= SCOUTFS_INO_FLAG_TRUNCATE;
 	scoutfs_inode_set_data_seq(inode);
+	inode_inc_iversion(inode);
 	scoutfs_update_inode_item(inode, lock, &ind_locks);
 
 	scoutfs_release_trans(sb);
@@ -479,6 +483,7 @@ retry:
 		goto out;
 
 	setattr_copy(inode, attr);
+	inode_inc_iversion(inode);
 	scoutfs_update_inode_item(inode, lock, &ind_locks);
 
 	scoutfs_release_trans(sb);
@@ -684,6 +689,7 @@ struct inode *scoutfs_iget(struct super_block *sb, u64 ino, int lkf)
 		/* XXX ensure refresh, instead clear in drop_inode? */
 		si = SCOUTFS_I(inode);
 		atomic64_set(&si->last_refreshed, 0);
+		inode->i_version = 0;
 
 		ret = scoutfs_inode_refresh(inode, lock, 0);
 		if (ret == 0)
@@ -711,6 +717,7 @@ static void store_inode(struct scoutfs_inode *cinode, struct inode *inode)
 	scoutfs_inode_get_onoff(inode, &online_blocks, &offline_blocks);
 
 	cinode->size = cpu_to_le64(i_size_read(inode));
+	cinode->version = cpu_to_le64(inode->i_version);
 	cinode->nlink = cpu_to_le32(inode->i_nlink);
 	cinode->uid = cpu_to_le32(i_uid_read(inode));
 	cinode->gid = cpu_to_le32(i_gid_read(inode));
@@ -734,6 +741,9 @@ static void store_inode(struct scoutfs_inode *cinode, struct inode *inode)
 	cinode->next_readdir_pos = cpu_to_le64(si->next_readdir_pos);
 	cinode->next_xattr_id = cpu_to_le64(si->next_xattr_id);
 	cinode->flags = cpu_to_le32(si->flags);
+	cinode->crtime.sec = cpu_to_le64(si->crtime.tv_sec);
+	cinode->crtime.nsec = cpu_to_le32(si->crtime.tv_nsec);
+	memset(cinode->crtime.__pad, 0, sizeof(cinode->crtime.__pad));
 }
 
 /*

kmod/src/inode.h

@@ -20,6 +20,7 @@ struct scoutfs_inode_info {
 	u64 online_blocks;
 	u64 offline_blocks;
 	u32 flags;
+	struct timespec crtime;
 
 	/*
 	 * Protects per-inode extent items, most particularly readers

kmod/src/ioctl.c

@@ -21,6 +21,7 @@
 #include <linux/mm.h>
 #include <linux/sched.h>
 #include <linux/aio.h>
+#include <linux/list_sort.h>
 
 #include "format.h"
 #include "key.h"
@@ -39,6 +40,7 @@
 #include "srch.h"
 #include "alloc.h"
 #include "server.h"
+#include "counters.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -541,6 +543,7 @@ out:
 static long scoutfs_ioc_stat_more(struct file *file, unsigned long arg)
 {
 	struct inode *inode = file_inode(file);
+	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	struct scoutfs_ioctl_stat_more stm;
 
 	if (get_user(stm.valid_bytes, (__u64 __user *)arg))
@@ -552,6 +555,8 @@ static long scoutfs_ioc_stat_more(struct file *file, unsigned long arg)
 	stm.data_seq = scoutfs_inode_data_seq(inode);
 	stm.data_version = scoutfs_inode_data_version(inode);
 	scoutfs_inode_get_onoff(inode, &stm.online_blocks, &stm.offline_blocks);
+	stm.crtime_sec = si->crtime.tv_sec;
+	stm.crtime_nsec = si->crtime.tv_nsec;
 
 	if (copy_to_user((void __user *)arg, &stm, stm.valid_bytes))
 		return -EFAULT;
@@ -617,6 +622,7 @@ static long scoutfs_ioc_data_waiting(struct file *file, unsigned long arg)
 static long scoutfs_ioc_setattr_more(struct file *file, unsigned long arg)
 {
 	struct inode *inode = file->f_inode;
+	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	struct super_block *sb = inode->i_sb;
 	struct scoutfs_ioctl_setattr_more __user *usm = (void __user *)arg;
 	struct scoutfs_ioctl_setattr_more sm;
@@ -685,6 +691,8 @@ static long scoutfs_ioc_setattr_more(struct file *file, unsigned long arg)
 		i_size_write(inode, sm.i_size);
 	inode->i_ctime.tv_sec = sm.ctime_sec;
 	inode->i_ctime.tv_nsec = sm.ctime_nsec;
+	si->crtime.tv_sec = sm.crtime_sec;
+	si->crtime.tv_nsec = sm.crtime_nsec;
 
 	scoutfs_update_inode_item(inode, lock, &ind_locks);
 	ret = 0;
@@ -1035,6 +1043,293 @@ out:
 	return ret;
 }
 
+struct xattr_total_entry {
+	struct rb_node node;
+	struct scoutfs_ioctl_xattr_total xt;
+	u64 fs_seq;
+	u64 fs_total;
+	u64 fs_count;
+	u64 fin_seq;
+	u64 fin_total;
+	s64 fin_count;
+	u64 log_seq;
+	u64 log_total;
+	s64 log_count;
+};
+
+static int cmp_xt_entry_name(const struct xattr_total_entry *a,
+			     const struct xattr_total_entry *b)
+
+{
+	return scoutfs_cmp_u64s(a->xt.name[0], b->xt.name[0]) ?:
+	       scoutfs_cmp_u64s(a->xt.name[1], b->xt.name[1]) ?:
+	       scoutfs_cmp_u64s(a->xt.name[2], b->xt.name[2]);
+}
+
+/*
+ * Record the contribution of the three classes of logged items we can
+ * see: the item in the fs_root, items from finalized log btrees, and
+ * items from active log btrees.  Once we have the full set the caller
+ * can decide which of the items contribute to the total it sends to the
+ * user.
+ */
+static int read_xattr_total_item(struct super_block *sb, struct scoutfs_key *key,
+				 u64 seq, u8 flags, void *val, int val_len, int fic, void *arg)
+{
+	struct scoutfs_xattr_totl_val *tval = val;
+	struct xattr_total_entry *ent;
+	struct xattr_total_entry rd;
+	struct rb_root *root = arg;
+	struct rb_node *parent;
+	struct rb_node **node;
+	int cmp;
+
+	rd.xt.name[0] = le64_to_cpu(key->skxt_a);
+	rd.xt.name[1] = le64_to_cpu(key->skxt_b);
+	rd.xt.name[2] = le64_to_cpu(key->skxt_c);
+
+	/* find entry matching name */
+	node = &root->rb_node;
+	parent = NULL;
+	cmp = -1;
+	while (*node) {
+		parent = *node;
+		ent = container_of(*node, struct xattr_total_entry, node);
+
+		/* sort merge items by key then newest to oldest */
+		cmp = cmp_xt_entry_name(&rd, ent);
+		if (cmp < 0)
+			node = &(*node)->rb_left;
+		else if (cmp > 0)
+			node = &(*node)->rb_right;
+		else
+			break;
+	}
+
+	/* allocate and insert new node if we need to */
+	if (cmp != 0) {
+		ent = kzalloc(sizeof(*ent), GFP_KERNEL);
+		if (!ent)
+			return -ENOMEM;
+
+		memcpy(&ent->xt.name, &rd.xt.name, sizeof(ent->xt.name));
+
+		rb_link_node(&ent->node, parent, node);
+		rb_insert_color(&ent->node, root);
+	}
+
+	if (fic & FIC_FS_ROOT) {
+		ent->fs_seq = seq;
+		ent->fs_total = le64_to_cpu(tval->total);
+		ent->fs_count = le64_to_cpu(tval->count);
+	} else if (fic & FIC_FINALIZED) {
+		ent->fin_seq = seq;
+		ent->fin_total += le64_to_cpu(tval->total);
+		ent->fin_count += le64_to_cpu(tval->count);
+	} else {
+		ent->log_seq = seq;
+		ent->log_total += le64_to_cpu(tval->total);
+		ent->log_count += le64_to_cpu(tval->count);
+	}
+
+	scoutfs_inc_counter(sb, totl_read_item);
+
+	return 0;
+}
+
+/* these are always _safe, node stores next */
+#define for_each_xt_ent(ent, node, root)					\
+	for (node = rb_first(root);						\
+	     node && (ent = rb_entry(node, struct xattr_total_entry, node),	\
+		      node = rb_next(node), 1); )
+
+#define for_each_xt_ent_reverse(ent, node, root)				\
+	for (node = rb_last(root);						\
+	     node && (ent = rb_entry(node, struct xattr_total_entry, node),	\
+		      node = rb_prev(node), 1); )
+
+static void free_xt_ent(struct rb_root *root, struct xattr_total_entry *ent)
+{
+	rb_erase(&ent->node, root);
+	kfree(ent);
+}
+
+static void free_all_xt_ents(struct rb_root *root)
+{
+	struct xattr_total_entry *ent;
+	struct rb_node *node;
+
+	for_each_xt_ent(ent, node, root)
+		free_xt_ent(root, ent);
+}
+
+/*
+ * Starting from the caller's pos_name, copy the names, totals, and
+ * counts for the .totl. tagged xattrs in the system sorted by their
+ * name until the user's buffer is full.  This only sees xattrs that
+ * have been committed.  It doesn't use locking to force commits and
+ * block writers so it can be a little bit out of date with respect to
+ * dirty xattrs in memory across the system.
+ *
+ * Our reader has to be careful because the log btree merging code can
+ * write partial results to the fs_root.  This means that a reader can
+ * see both cases where new finalized logs should be applied to the old
+ * fs items and where old finalized logs have already been applied to
+ * the partially merged fs items.  Currently active logged items are
+ * always applied on top of all cases.
+ *
+ * These cases are differentiated with a combination of sequence numbers
+ * in items, the count of contributing xattrs, and a flag
+ * differentiating finalized and active logged items.  This lets us
+ * recognize all cases, including when finalized logs were merged and
+ * deleted the fs item.
+ *
+ * We're allocating a tracking struct for each totl name we see while
+ * traversing the item btrees.  The forest reader is providing the items
+ * it finds in leaf blocks that contain the search key.  In the worst
+ * case all of these blocks are full and none of the items overlap.  At
+ * most, figure order a thousand names per mount.  But in practice many
+ * of these factors fall away: leaf blocks aren't fill, leaf items
+ * overlap, there aren't finalized log btrees, and not all mounts are
+ * actively changing totals.   We're much more likely to only read a
+ * leaf block's worth of totals that have been long since merged into
+ * the fs_root.
+ */
+static long scoutfs_ioc_read_xattr_totals(struct file *file, unsigned long arg)
+{
+	struct super_block *sb = file_inode(file)->i_sb;
+	struct scoutfs_ioctl_read_xattr_totals __user *urxt = (void __user *)arg;
+	struct scoutfs_ioctl_read_xattr_totals rxt;
+	struct scoutfs_ioctl_xattr_total __user *uxt;
+	struct xattr_total_entry *ent;
+	struct scoutfs_key key;
+	struct scoutfs_key bloom_key;
+	struct scoutfs_key start;
+	struct scoutfs_key end;
+	struct rb_root root = RB_ROOT;
+	struct rb_node *node;
+	int count = 0;
+	int ret;
+
+	if (!(file->f_mode & FMODE_READ)) {
+		ret = -EBADF;
+		goto out;
+	}
+
+	if (!capable(CAP_SYS_ADMIN)) {
+		ret = -EPERM;
+		goto out;
+	}
+
+	if (copy_from_user(&rxt, urxt, sizeof(rxt))) {
+		ret = -EFAULT;
+		goto out;
+	}
+	uxt = (void __user *)rxt.totals_ptr;
+
+	if ((rxt.totals_ptr & (sizeof(__u64) - 1)) ||
+	    (rxt.totals_bytes < sizeof(struct scoutfs_ioctl_xattr_total))) {
+		ret = -EINVAL;
+		goto out;
+	}
+
+	scoutfs_key_set_zeros(&bloom_key);
+	bloom_key.sk_zone = SCOUTFS_XATTR_TOTL_ZONE;
+	scoutfs_xattr_init_totl_key(&start, rxt.pos_name);
+
+	while (rxt.totals_bytes >= sizeof(struct scoutfs_ioctl_xattr_total)) {
+
+		scoutfs_key_set_ones(&end);
+		end.sk_zone = SCOUTFS_XATTR_TOTL_ZONE;
+		if (scoutfs_key_compare(&start, &end) > 0)
+			break;
+
+		key = start;
+		ret = scoutfs_forest_read_items(sb, &key, &bloom_key, &start, &end,
+						read_xattr_total_item, &root);
+		if (ret < 0) {
+			if (ret == -ESTALE) {
+				free_all_xt_ents(&root);
+				continue;
+			}
+			goto out;
+		}
+
+		if (RB_EMPTY_ROOT(&root))
+			break;
+
+		/* trim totals that fall outside of the consistent range */
+		for_each_xt_ent(ent, node, &root) {
+			scoutfs_xattr_init_totl_key(&key, ent->xt.name);
+			if (scoutfs_key_compare(&key, &start) < 0) {
+				free_xt_ent(&root, ent);
+			} else {
+				break;
+			}
+		}
+		for_each_xt_ent_reverse(ent, node, &root) {
+			scoutfs_xattr_init_totl_key(&key, ent->xt.name);
+			if (scoutfs_key_compare(&key, &end) > 0) {
+				free_xt_ent(&root, ent);
+			} else {
+				break;
+			}
+		}
+
+		/* copy resulting unique non-zero totals to userspace */
+		for_each_xt_ent(ent, node, &root) {
+			if (rxt.totals_bytes < sizeof(ent->xt))
+				break;
+
+			/* start with the fs item if we have it */
+			if (ent->fs_seq != 0) {
+				ent->xt.total = ent->fs_total;
+				ent->xt.count = ent->fs_count;
+				scoutfs_inc_counter(sb, totl_read_fs);
+			}
+
+			/* apply finalized logs if they're newer or creating */
+			if (((ent->fs_seq != 0) && (ent->fin_seq > ent->fs_seq)) ||
+			    ((ent->fs_seq == 0) && (ent->fin_count > 0))) {
+				ent->xt.total += ent->fin_total;
+				ent->xt.count += ent->fin_count;
+				scoutfs_inc_counter(sb, totl_read_finalized);
+			}
+
+			/* always apply active logs which must be newer than fs and finalized */
+			if (ent->log_seq > 0) {
+				ent->xt.total += ent->log_total;
+				ent->xt.count += ent->log_count;
+				scoutfs_inc_counter(sb, totl_read_logged);
+			}
+
+			if (ent->xt.total != 0 || ent->xt.count != 0) {
+				if (copy_to_user(uxt, &ent->xt, sizeof(ent->xt))) {
+					ret = -EFAULT;
+					goto out;
+				}
+
+				uxt++;
+				rxt.totals_bytes -= sizeof(ent->xt);
+				count++;
+				scoutfs_inc_counter(sb, totl_read_copied);
+			}
+
+			free_xt_ent(&root, ent);
+		}
+
+		/* continue after the last possible key read */
+		start = end;
+		scoutfs_key_inc(&start);
+	}
+
+	ret = 0;
+out:
+	free_all_xt_ents(&root);
+
+	return ret ?: count;
+}
+
 long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 {
 	switch (cmd) {
@@ -1066,6 +1361,8 @@ long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		return scoutfs_ioc_move_blocks(file, arg);
 	case SCOUTFS_IOC_RESIZE_DEVICES:
 		return scoutfs_ioc_resize_devices(file, arg);
+	case SCOUTFS_IOC_READ_XATTR_TOTALS:
+		return scoutfs_ioc_read_xattr_totals(file, arg);
 	}
 
 	return -ENOTTY;

kmod/src/ioctl.h

@@ -232,6 +232,9 @@ struct scoutfs_ioctl_stat_more {
 	__u64 data_version;
 	__u64 online_blocks;
 	__u64 offline_blocks;
+	__u64 crtime_sec;
+	__u32 crtime_nsec;
+	__u8  _pad[4];
 };
 
 #define SCOUTFS_IOC_STAT_MORE _IOR(SCOUTFS_IOCTL_MAGIC, 5, \
@@ -267,7 +270,8 @@ struct scoutfs_ioctl_data_waiting {
 /*
  * If i_size is set then data_version must be non-zero.  If the offline
  * flag is set then i_size must be set and a offline extent will be
- * created from offset 0 to i_size.
+ * created from offset 0 to i_size.  The time fields are always applied
+ * to the inode.
  */
 struct scoutfs_ioctl_setattr_more {
 	__u64 data_version;
@@ -275,7 +279,8 @@ struct scoutfs_ioctl_setattr_more {
 	__u64 flags;
 	__u64 ctime_sec;
 	__u32 ctime_nsec;
-	__u8 _pad[4];
+	__u32 crtime_nsec;
+	__u64 crtime_sec;
 };
 
 #define SCOUTFS_IOC_SETATTR_MORE_OFFLINE		(1 << 0)
@@ -485,4 +490,55 @@ struct scoutfs_ioctl_resize_devices {
 #define SCOUTFS_IOC_RESIZE_DEVICES \
 	_IOR(SCOUTFS_IOCTL_MAGIC, 14, struct scoutfs_ioctl_resize_devices)
 
+#define SCOUTFs_IOCTL_XATTR_TOTAL_NAME_NR 3
+
+/*
+ * Copy global totals of .totl. xattr value payloads to the user.   This
+ * only sees xattrs which have been committed and this doesn't force
+ * commits of dirty data throughout the system.  This can be out of sync
+ * by the amount of xattrs that can be dirty in open transactions that
+ * are being built throughout the system.
+ *
+ * pos_name: The array name of the first total that can be returned.
+ * The name is derived from the key of the xattrs that contribute to the
+ * total.  For xattrs with a .totl.1.2.3 key, the pos_name[] should be
+ * {1, 2, 3}.
+ *
+ * totals_ptr: An aligned pointer to a buffer that will be filled with
+ * an array of scoutfs_ioctl_xattr_total structs for each total copied.
+ *
+ * totals_bytes: The size of the buffer in bytes.  There must be room
+ * for at least one struct element so that returning 0 can promise that
+ * there were no more totals to copy after the pos_name.
+ *
+ * The number of copied elements is returned and 0 is returned if there
+ * were no more totals to copy after the pos_name.
+ *
+ * In addition to the usual errnos (EIO, EINVAL, EPERM, EFAULT) this
+ * adds:
+ *
+ * EINVAL: The totals_ buffer was not aligned or was not large enough
+ * for a single struct entry.
+ */
+struct scoutfs_ioctl_read_xattr_totals {
+	__u64 pos_name[SCOUTFs_IOCTL_XATTR_TOTAL_NAME_NR];
+	__u64 totals_ptr;
+	__u64 totals_bytes;
+};
+
+/*
+ * An individual total that is given to userspace.   The total is the
+ * sum of all the values in the xattr payloads matching the name.  The
+ * count is the number of xattrs, not number of files, contributing to
+ * the total.
+ */
+struct scoutfs_ioctl_xattr_total {
+	__u64 name[SCOUTFs_IOCTL_XATTR_TOTAL_NAME_NR];
+	__u64 total;
+	__u64 count;
+};
+
+#define SCOUTFS_IOC_READ_XATTR_TOTALS \
+	_IOR(SCOUTFS_IOCTL_MAGIC, 15, struct scoutfs_ioctl_read_xattr_totals)
+
 #endif

kmod/src/item.c

@@ -127,7 +127,7 @@ struct cached_page {
 	unsigned long lru_time;
 	struct list_head dirty_list;
 	struct list_head dirty_head;
-	u64 max_liv_seq;
+	u64 max_seq;
 	struct page *page;
 	unsigned int page_off;
 	unsigned int erased_bytes;
@@ -139,10 +139,11 @@ struct cached_item {
 	struct list_head dirty_head;
 	unsigned int dirty:1,		/* needs to be written */
 		     persistent:1,	/* in btrees, needs deletion item */
-		     deletion:1;	/* negative del item for writing */
+		     deletion:1,	/* negative del item for writing */
+		     delta:1;		/* item vales are combined, freed after write */
 	unsigned int val_len;
 	struct scoutfs_key key;
-	struct scoutfs_log_item_value liv;
+	u64 seq;
 	char val[0];
 };
 
@@ -386,12 +387,10 @@ static void put_pg(struct super_block *sb, struct cached_page *pg)
 	}
 }
 
-static void update_pg_max_liv_seq(struct cached_page *pg, struct cached_item *item)
+static void update_pg_max_seq(struct cached_page *pg, struct cached_item *item)
 {
-	u64 liv_seq = le64_to_cpu(item->liv.seq);
-
-	if (liv_seq > pg->max_liv_seq)
-		pg->max_liv_seq = liv_seq;
+	if (item->seq > pg->max_seq)
+		pg->max_seq = item->seq;
 }
 
 /*
@@ -401,8 +400,7 @@ static void update_pg_max_liv_seq(struct cached_page *pg, struct cached_item *it
  * page or checking the free space first.
  */
 static struct cached_item *alloc_item(struct cached_page *pg,
-				      struct scoutfs_key *key,
-				      struct scoutfs_log_item_value *liv,
+				      struct scoutfs_key *key, u64 seq, bool deletion,
 				      void *val, int val_len)
 {
 	struct cached_item *item;
@@ -417,15 +415,16 @@ static struct cached_item *alloc_item(struct cached_page *pg,
 	INIT_LIST_HEAD(&item->dirty_head);
 	item->dirty = 0;
 	item->persistent = 0;
-	item->deletion = !!(liv->flags & SCOUTFS_LOG_ITEM_FLAG_DELETION);
+	item->deletion = !!deletion;
+	item->delta = 0;
 	item->val_len = val_len;
 	item->key = *key;
-	item->liv = *liv;
+	item->seq = seq;
 
 	if (val_len)
 		memcpy(item->val, val, val_len);
 
-	update_pg_max_liv_seq(pg, item);
+	update_pg_max_seq(pg, item);
 
 	return item;
 }
@@ -634,7 +633,7 @@ static void mark_item_dirty(struct super_block *sb,
 		item->dirty = 1;
 	}
 
-	update_pg_max_liv_seq(pg, item);
+	update_pg_max_seq(pg, item);
 }
 
 static void clear_item_dirty(struct super_block *sb,
@@ -711,7 +710,7 @@ static void move_page_items(struct super_block *sb,
 		if (stop && scoutfs_key_compare(&from->key, stop) >= 0)
 			break;
 
-		to = alloc_item(right, &from->key, &from->liv, from->val,
+		to = alloc_item(right, &from->key, from->seq, from->deletion, from->val,
 				from->val_len);
 		rbtree_insert(&to->node, par, pnode, &right->item_root);
 		par = &to->node;
@@ -723,7 +722,7 @@ static void move_page_items(struct super_block *sb,
 		}
 
 		to->persistent = from->persistent;
-		to->deletion = from->deletion;
+		to->delta = from->delta;
 
 		erase_item(left, from);
 	}
@@ -1356,11 +1355,11 @@ static void del_active_reader(struct item_cache_info *cinf, struct active_reader
  * insert old versions of items into the tree here so that the trees
  * don't have to compare seqs.
  */
-static int read_page_item(struct super_block *sb, struct scoutfs_key *key,
-			  struct scoutfs_log_item_value *liv, void *val,
-			  int val_len, void *arg)
+static int read_page_item(struct super_block *sb, struct scoutfs_key *key, u64 seq, u8 flags,
+			  void *val, int val_len, int fic, void *arg)
 {
 	DECLARE_ITEM_CACHE_INFO(sb, cinf);
+	const bool deletion = !!(flags & SCOUTFS_ITEM_FLAG_DELETION);
 	struct rb_root *root = arg;
 	struct cached_page *right = NULL;
 	struct cached_page *left = NULL;
@@ -1374,7 +1373,7 @@ static int read_page_item(struct super_block *sb, struct scoutfs_key *key,
 
 	pg = page_rbtree_walk(sb, root, key, key, NULL, NULL, &p_par, &p_pnode);
 	found = item_rbtree_walk(&pg->item_root, key, NULL, &par, &pnode);
-	if (found && (le64_to_cpu(found->liv.seq) >= le64_to_cpu(liv->seq)))
+	if (found && (found->seq >= seq))
 		return 0;
 
 	if (!page_has_room(pg, val_len)) {
@@ -1388,7 +1387,7 @@ static int read_page_item(struct super_block *sb, struct scoutfs_key *key,
 					 &pnode);
 	}
 
-	item = alloc_item(pg, key, liv, val, val_len);
+	item = alloc_item(pg, key, seq, deletion, val, val_len);
 	if (!item) {
 		/* simpler split of private pages, no locking/dirty/lru */
 		if (!left)
@@ -1411,7 +1410,7 @@ static int read_page_item(struct super_block *sb, struct scoutfs_key *key,
 		put_pg(sb, pg);
 
 		pg = scoutfs_key_compare(key, &left->end) <= 0 ? left : right;
-		item = alloc_item(pg, key, liv, val, val_len);
+		item = alloc_item(pg, key, seq, deletion, val, val_len);
 		found = item_rbtree_walk(&pg->item_root, key, NULL, &par,
 					 &pnode);
 
@@ -1484,8 +1483,9 @@ static int read_pages(struct super_block *sb, struct item_cache_info *cinf,
 	/* set active reader seq before reading persistent roots */
 	add_active_reader(sb, &active);
 
-	ret = scoutfs_forest_read_items(sb, lock, key, &start, &end,
-				       read_page_item, &root);
+	start = lock->start;
+	end = lock->end;
+	ret = scoutfs_forest_read_items(sb, key, &lock->start, &start, &end, read_page_item, &root);
 	if (ret < 0)
 		goto out;
 
@@ -1824,11 +1824,11 @@ out:
  * to the last stable seq and ensure that all the items in open
  * transactions and granted locks will have greater seqs.
  */
-static __le64 item_seq(struct super_block *sb, struct scoutfs_lock *lock)
+static u64 item_seq(struct super_block *sb, struct scoutfs_lock *lock)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 
-	return cpu_to_le64(max(sbi->trans_seq, lock->write_seq));
+	return max(sbi->trans_seq, lock->write_seq);
 }
 
 /*
@@ -1863,7 +1863,7 @@ int scoutfs_item_dirty(struct super_block *sb, struct scoutfs_key *key,
 	if (!item || item->deletion) {
 		ret = -ENOENT;
 	} else {
-		item->liv.seq = item_seq(sb, lock);
+		item->seq = item_seq(sb, lock);
 		mark_item_dirty(sb, cinf, pg, NULL, item);
 		ret = 0;
 	}
@@ -1883,9 +1883,7 @@ static int item_create(struct super_block *sb, struct scoutfs_key *key,
 		       int mode, bool force)
 {
 	DECLARE_ITEM_CACHE_INFO(sb, cinf);
-	struct scoutfs_log_item_value liv = {
-		.seq = item_seq(sb, lock),
-	};
+	const u64 seq = item_seq(sb, lock);
 	struct cached_item *found;
 	struct cached_item *item;
 	struct cached_page *pg;
@@ -1913,7 +1911,7 @@ static int item_create(struct super_block *sb, struct scoutfs_key *key,
 		goto unlock;
 	}
 
-	item = alloc_item(pg, key, &liv, val, val_len);
+	item = alloc_item(pg, key, seq, false, val, val_len);
 	rbtree_insert(&item->node, par, pnode, &pg->item_root);
 	mark_item_dirty(sb, cinf, pg, NULL, item);
 
@@ -1958,9 +1956,7 @@ int scoutfs_item_update(struct super_block *sb, struct scoutfs_key *key,
 			void *val, int val_len, struct scoutfs_lock *lock)
 {
 	DECLARE_ITEM_CACHE_INFO(sb, cinf);
-	struct scoutfs_log_item_value liv = {
-		.seq = item_seq(sb, lock),
-	};
+	const u64 seq = item_seq(sb, lock);
 	struct cached_item *item;
 	struct cached_item *found;
 	struct cached_page *pg;
@@ -1995,10 +1991,10 @@ int scoutfs_item_update(struct super_block *sb, struct scoutfs_key *key,
 			pg->erased_bytes += item_val_bytes(found->val_len) -
 					    item_val_bytes(val_len);
 		found->val_len = val_len;
-		found->liv.seq = liv.seq;
+		found->seq = seq;
 		mark_item_dirty(sb, cinf, pg, NULL, found);
 	} else {
-		item = alloc_item(pg, key, &liv, val, val_len);
+		item = alloc_item(pg, key, seq, false, val, val_len);
 		item->persistent = found->persistent;
 		rbtree_insert(&item->node, par, pnode, &pg->item_root);
 		mark_item_dirty(sb, cinf, pg, NULL, item);
@@ -2014,6 +2010,77 @@ out:
 	return ret;
 }
 
+/*
+ * Add a delta item.  Delta items are an incremental change relative to
+ * the current persistent delta items.  We never have to read the
+ * current items so the caller always writes with write only locks.  If
+ * combining the current delta item and the caller's item results in a
+ * null we can just drop it, we don't have to emit a deletion item.
+ */
+int scoutfs_item_delta(struct super_block *sb, struct scoutfs_key *key,
+		       void *val, int val_len, struct scoutfs_lock *lock)
+{
+	DECLARE_ITEM_CACHE_INFO(sb, cinf);
+	const u64 seq = item_seq(sb, lock);
+	struct cached_item *item;
+	struct cached_page *pg;
+	struct rb_node **pnode;
+	struct rb_node *par;
+	int ret;
+
+	scoutfs_inc_counter(sb, item_delta);
+
+	if ((ret = lock_safe(lock, key, SCOUTFS_LOCK_WRITE_ONLY)))
+		goto out;
+
+	ret = scoutfs_forest_set_bloom_bits(sb, lock);
+	if (ret < 0)
+		goto out;
+
+	ret = get_cached_page(sb, cinf, lock, key, true, true, val_len, &pg);
+	if (ret < 0)
+		goto out;
+	__acquire(pg->rwlock);
+
+	item = item_rbtree_walk(&pg->item_root, key, NULL, &par, &pnode);
+	if (item) {
+		if (!item->delta) {
+			ret = -EIO;
+			goto unlock;
+		}
+
+		ret = scoutfs_forest_combine_deltas(key, item->val, item->val_len, val, val_len);
+		if (ret <= 0) {
+			if (ret == 0)
+				ret = -EIO;
+			goto unlock;
+		}
+
+		if (ret == SCOUTFS_DELTA_COMBINED) {
+			item->seq = seq;
+			mark_item_dirty(sb, cinf, pg, NULL, item);
+		} else if (ret == SCOUTFS_DELTA_COMBINED_NULL) {
+			clear_item_dirty(sb, cinf, pg, item);
+			erase_item(pg, item);
+		} else {
+			ret = -EIO;
+			goto unlock;
+		}
+		ret = 0;
+	} else {
+		item = alloc_item(pg, key, seq, false, val, val_len);
+		rbtree_insert(&item->node, par, pnode, &pg->item_root);
+		mark_item_dirty(sb, cinf, pg, NULL, item);
+		item->delta = 1;
+		ret = 0;
+	}
+
+unlock:
+	write_unlock(&pg->rwlock);
+out:
+	return ret;
+}
+
 /*
  * Delete an item from the cache.  We can leave behind a dirty deletion
  * item if there is a persistent item that needs to be overwritten.
@@ -2026,9 +2093,7 @@ static int item_delete(struct super_block *sb, struct scoutfs_key *key,
 		       struct scoutfs_lock *lock, int mode, bool force)
 {
 	DECLARE_ITEM_CACHE_INFO(sb, cinf);
-	struct scoutfs_log_item_value liv = {
-		.seq = item_seq(sb, lock),
-	};
+	const u64 seq = item_seq(sb, lock);
 	struct cached_item *item;
 	struct cached_page *pg;
 	struct rb_node **pnode;
@@ -2056,7 +2121,7 @@ static int item_delete(struct super_block *sb, struct scoutfs_key *key,
 	}
 
 	if (!item) {
-		item = alloc_item(pg, key, &liv, NULL, 0);
+		item = alloc_item(pg, key, seq, false, NULL, 0);
 		rbtree_insert(&item->node, par, pnode, &pg->item_root);
 	}
 
@@ -2069,8 +2134,7 @@ static int item_delete(struct super_block *sb, struct scoutfs_key *key,
 		erase_item(pg, item);
 	} else {
 		/* must emit deletion to clobber old persistent item */
-		item->liv.seq = liv.seq;
-		item->liv.flags |= SCOUTFS_LOG_ITEM_FLAG_DELETION;
+		item->seq = seq;
 		item->deletion = 1;
 		pg->erased_bytes += item_val_bytes(item->val_len) -
 				    item_val_bytes(0);
@@ -2157,16 +2221,10 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 	LIST_HEAD(pages);
 	LIST_HEAD(pos);
 	u64 max_seq = 0;
-	int val_len;
 	int bytes;
 	int off;
 	int ret;
 
-	/* we're relying on struct layout to prepend item value headers */
-	BUILD_BUG_ON(offsetof(struct cached_item, val) !=
-		     (offsetof(struct cached_item, liv) +
-		      member_sizeof(struct cached_item, liv)));
-
 	if (atomic_read(&cinf->dirty_pages) == 0)
 		return 0;
 
@@ -2218,10 +2276,9 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 		list_sort(NULL, &pg->dirty_list, cmp_item_key);
 
 		list_for_each_entry(item, &pg->dirty_list, dirty_head) {
-			val_len = sizeof(item->liv) + item->val_len;
 			bytes = offsetof(struct scoutfs_btree_item_list,
-					 val[val_len]);
-			max_seq = max(max_seq, le64_to_cpu(item->liv.seq));
+					 val[item->val_len]);
+			max_seq = max(max_seq, item->seq);
 
 			if (off + bytes > PAGE_SIZE) {
 				page = second;
@@ -2237,8 +2294,10 @@ int scoutfs_item_write_dirty(struct super_block *sb)
 			prev = &lst->next;
 
 			lst->key = item->key;
-			lst->val_len = val_len;
-			memcpy(lst->val, &item->liv, val_len);
+			lst->seq = item->seq;
+			lst->flags = item->deletion ? SCOUTFS_ITEM_FLAG_DELETION : 0;
+			lst->val_len = item->val_len;
+			memcpy(lst->val, item->val, item->val_len);
 		}
 
 		spin_lock(&cinf->dirty_lock);
@@ -2296,8 +2355,11 @@ retry:
 					 dirty_head) {
 			clear_item_dirty(sb, cinf, pg, item);
 
+			if (item->delta)
+				scoutfs_inc_counter(sb, item_delta_written);
+
 			/* free deletion items */
-			if (item->deletion)
+			if (item->deletion || item->delta)
 				erase_item(pg, item);
 			else
 				item->persistent = 1;
@@ -2467,7 +2529,7 @@ static int item_lru_shrink(struct shrinker *shrink,
 
 	list_for_each_entry_safe(pg, tmp, &cinf->lru_list, lru_head) {
 
-		if (first_reader_seq <= pg->max_liv_seq) {
+		if (first_reader_seq <= pg->max_seq) {
 			scoutfs_inc_counter(sb, item_shrink_page_reader);
 			continue;
 		}

kmod/src/item.h

@@ -18,6 +18,8 @@ int scoutfs_item_create_force(struct super_block *sb, struct scoutfs_key *key,
 			      struct scoutfs_lock *lock);
 int scoutfs_item_update(struct super_block *sb, struct scoutfs_key *key,
 			void *val, int val_len, struct scoutfs_lock *lock);
+int scoutfs_item_delta(struct super_block *sb, struct scoutfs_key *key,
+		       void *val, int val_len, struct scoutfs_lock *lock);
 int scoutfs_item_delete(struct super_block *sb, struct scoutfs_key *key,
 			  struct scoutfs_lock *lock);
 int scoutfs_item_delete_force(struct super_block *sb,

kmod/src/lock.c

@@ -1237,6 +1237,20 @@ int scoutfs_lock_orphan(struct super_block *sb, enum scoutfs_lock_mode mode, int
 	return lock_key_range(sb, mode, flags, &start, &end, lock);
 }
 
+int scoutfs_lock_xattr_totl(struct super_block *sb, enum scoutfs_lock_mode mode, int flags,
+			    struct scoutfs_lock **lock)
+{
+	struct scoutfs_key start;
+	struct scoutfs_key end;
+
+	scoutfs_key_set_zeros(&start);
+	start.sk_zone = SCOUTFS_XATTR_TOTL_ZONE;
+	scoutfs_key_set_ones(&end);
+	end.sk_zone = SCOUTFS_XATTR_TOTL_ZONE;
+
+	return lock_key_range(sb, mode, flags, &start, &end, lock);
+}
+
 void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, enum scoutfs_lock_mode mode)
 {
 	DECLARE_LOCK_INFO(sb, linfo);

kmod/src/lock.h

@@ -84,6 +84,8 @@ int scoutfs_lock_rename(struct super_block *sb, enum scoutfs_lock_mode mode, int
 			struct scoutfs_lock **lock);
 int scoutfs_lock_orphan(struct super_block *sb, enum scoutfs_lock_mode mode, int flags,
 		        u64 ino, struct scoutfs_lock **lock);
+int scoutfs_lock_xattr_totl(struct super_block *sb, enum scoutfs_lock_mode mode, int flags,
+			    struct scoutfs_lock **lock);
 void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock,
 		    enum scoutfs_lock_mode mode);
 

kmod/src/srch.c

@@ -2081,7 +2081,7 @@ static int delete_files(struct super_block *sb, struct scoutfs_alloc *alloc,
 		       struct scoutfs_block_writer *wri,
 		       struct scoutfs_srch_compact *sc)
 {
-	int ret;
+	int ret = 0;
 	int i;
 
 	for (i = 0; i < sc->nr; i++) {

kmod/src/super.c

@@ -545,6 +545,7 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 	sb->s_maxbytes = MAX_LFS_FILESIZE;
 	sb->s_op = &scoutfs_super_ops;
 	sb->s_export_op = &scoutfs_export_ops;
+	sb->s_flags |= MS_I_VERSION;
 
 	/* btree blocks use long lived bh->b_data refs */
 	mapping_set_gfp_mask(sb->s_bdev->bd_inode->i_mapping, GFP_NOFS);

kmod/src/xattr.c

@@ -97,6 +97,7 @@ static int unknown_prefix(const char *name)
 
 #define HIDE_TAG	"hide."
 #define SRCH_TAG	"srch."
+#define TOTL_TAG	"totl."
 #define TAG_LEN		(sizeof(HIDE_TAG) - 1)
 
 int scoutfs_xattr_parse_tags(const char *name, unsigned int name_len,
@@ -119,6 +120,9 @@ int scoutfs_xattr_parse_tags(const char *name, unsigned int name_len,
 		} else if (!strncmp(name, SRCH_TAG, TAG_LEN)) {
 			if (++tgs->srch == 0)
 				return -EINVAL;
+		} else if (!strncmp(name, TOTL_TAG, TAG_LEN)) {
+			if (++tgs->totl == 0)
+				return -EINVAL;
 		} else {
 			/* only reason to use scoutfs. is tags */
 			if (!found)
@@ -468,6 +472,100 @@ out:
 	return ret;
 }
 
+void scoutfs_xattr_init_totl_key(struct scoutfs_key *key, u64 *name)
+{
+	scoutfs_key_set_zeros(key);
+	key->sk_zone = SCOUTFS_XATTR_TOTL_ZONE;
+	key->skxt_a = cpu_to_le64(name[0]);
+	key->skxt_b = cpu_to_le64(name[1]);
+	key->skxt_c = cpu_to_le64(name[2]);
+}
+
+/*
+ * Parse a u64 in any base after null terminating it while forbidding
+ * the leading + and trailing \n that kstrotull allows.
+ */
+static int parse_totl_u64(const char *s, int len, u64 *res)
+{
+	char str[SCOUTFS_XATTR_MAX_TOTL_U64 + 1];
+
+	if (len <= 0 || len >= ARRAY_SIZE(str) || s[0] == '+' || s[len - 1] == '\n')
+		return -EINVAL;
+
+	memcpy(str, s, len);
+	str[len] = '\0';
+
+	return kstrtoull(str, 0, res) != 0 ? -EINVAL : 0;
+}
+
+/*
+ * non-destructive relatively quick parse of the last 3 dotted u64s that
+ * make up the name of the xattr total.  -EINVAL is returned if there
+ * are anything but 3 valid u64 encodings between single dots at the end
+ * of the name.
+ */
+static int parse_totl_key(struct scoutfs_key *key, const char *name, int name_len)
+{
+	u64 tot_name[3];
+	int end = name_len;
+	int nr = 0;
+	int len;
+	int ret;
+	int i;
+
+	/* parse name elements in reserve order from end of xattr name string */
+	for (i = name_len - 1; i >= 0 && nr < ARRAY_SIZE(tot_name); i--) {
+		if (name[i] != '.')
+			continue;
+
+		len = end - (i + 1);
+		ret = parse_totl_u64(&name[i + 1], len, &tot_name[nr]);
+		if (ret < 0)
+			goto out;
+
+		end = i;
+		nr++;
+	}
+
+	if (nr == ARRAY_SIZE(tot_name)) {
+		/* swap to account for parsing in reverse */
+		swap(tot_name[0], tot_name[2]);
+		scoutfs_xattr_init_totl_key(key, tot_name);
+		ret = 0;
+	} else {
+		ret = -EINVAL;
+	}
+
+out:
+	return ret;
+}
+
+static int apply_totl_delta(struct super_block *sb, struct scoutfs_key *key,
+			    struct scoutfs_xattr_totl_val *tval, struct scoutfs_lock *lock)
+{
+	if (tval->total == 0 && tval->count == 0)
+		return 0;
+
+	return scoutfs_item_delta(sb, key, tval, sizeof(*tval), lock);
+}
+
+int scoutfs_xattr_combine_totl(void *dst, int dst_len, void *src, int src_len)
+{
+	struct scoutfs_xattr_totl_val *s_tval = src;
+	struct scoutfs_xattr_totl_val *d_tval = dst;
+
+	if (src_len != sizeof(*s_tval) || dst_len != src_len)
+		return -EIO;
+
+	le64_add_cpu(&d_tval->total, le64_to_cpu(s_tval->total));
+	le64_add_cpu(&d_tval->count, le64_to_cpu(s_tval->count));
+
+	if (d_tval->total == 0 && d_tval->count == 0)
+		return SCOUTFS_DELTA_COMBINED_NULL;
+
+	return SCOUTFS_DELTA_COMBINED;
+}
+
 /*
  * The confusing swiss army knife of creating, modifying, and deleting
  * xattrs.
@@ -486,16 +584,22 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	struct super_block *sb = inode->i_sb;
 	const u64 ino = scoutfs_ino(inode);
+	struct scoutfs_xattr_totl_val tval = {0,};
 	struct scoutfs_xattr_prefix_tags tgs;
 	struct scoutfs_xattr *xat = NULL;
 	struct scoutfs_lock *lck = NULL;
+	struct scoutfs_lock *totl_lock = NULL;
 	size_t name_len = strlen(name);
+	struct scoutfs_key totl_key;
 	struct scoutfs_key key;
 	bool undo_srch = false;
+	bool undo_totl = false;
 	LIST_HEAD(ind_locks);
 	u8 found_parts;
 	unsigned int bytes;
+	unsigned int val_len;
 	u64 ind_seq;
+	u64 total;
 	u64 hash = 0;
 	u64 id = 0;
 	int ret;
@@ -519,11 +623,15 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	if (scoutfs_xattr_parse_tags(name, name_len, &tgs) != 0)
 		return -EINVAL;
 
-	if ((tgs.hide || tgs.srch) && !capable(CAP_SYS_ADMIN))
+	if ((tgs.hide | tgs.srch | tgs.totl) && !capable(CAP_SYS_ADMIN))
 		return -EPERM;
 
+	if (tgs.totl && ((ret = parse_totl_key(&totl_key, name, name_len)) != 0))
+		return ret;
+
 	bytes = sizeof(struct scoutfs_xattr) + name_len + size;
-	xat = __vmalloc(bytes, GFP_NOFS, PAGE_KERNEL);
+	/* alloc enough to read old totl value */
+	xat = __vmalloc(bytes + SCOUTFS_XATTR_MAX_TOTL_U64, GFP_NOFS, PAGE_KERNEL);
 	if (!xat) {
 		ret = -ENOMEM;
 		goto out;
@@ -536,9 +644,9 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 
 	down_write(&si->xattr_rwsem);
 
-	/* find an existing xattr to delete */
+	/* find an existing xattr to delete, including possible totl value */
 	ret = get_next_xattr(inode, &key, xat,
-			     sizeof(struct scoutfs_xattr) + name_len,
+			     sizeof(struct scoutfs_xattr) + name_len + SCOUTFS_XATTR_MAX_TOTL_U64,
 			     name, name_len, 0, 0, lck);
 	if (ret < 0 && ret != -ENOENT)
 		goto unlock;
@@ -558,9 +666,23 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 		goto unlock;
 	}
 
+	/* s64 count delta if we create or delete */
+	if (tgs.totl)
+		tval.count = cpu_to_le64((u64)!!(value) - (u64)!!(ret != -ENOENT));
+
 	/* found fields in key will also be used */
 	found_parts = ret >= 0 ? xattr_nr_parts(xat) : 0;
 
+	if (found_parts && tgs.totl) {
+		/* parse old totl value before we clobber xat buf */
+		val_len = ret - offsetof(struct scoutfs_xattr, name[xat->name_len]);
+		ret = parse_totl_u64(&xat->name[xat->name_len], val_len, &total);
+		if (ret < 0)
+			goto unlock;
+
+		le64_add_cpu(&tval.total, -total);
+	}
+
 	/* prepare our xattr */
 	if (value) {
 		if (found_parts)
@@ -572,6 +694,20 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 		memset(xat->__pad, 0, sizeof(xat->__pad));
 		memcpy(xat->name, name, name_len);
 		memcpy(&xat->name[xat->name_len], value, size);
+
+		if (tgs.totl) {
+			ret = parse_totl_u64(value, size, &total);
+			if (ret < 0)
+				goto unlock;
+		}
+
+		le64_add_cpu(&tval.total, total);
+	}
+
+	if (tgs.totl) {
+		ret = scoutfs_lock_xattr_totl(sb, SCOUTFS_LOCK_WRITE_ONLY, 0, &totl_lock);
+		if (ret)
+			goto unlock;
 	}
 
 retry:
@@ -597,6 +733,13 @@ retry:
 		undo_srch = true;
 	}
 
+	if (tgs.totl) {
+		ret = apply_totl_delta(sb, &totl_key, &tval, totl_lock);
+		if (ret < 0)
+			goto release;
+		undo_totl = true;
+	}
+
 	if (found_parts && value)
 		ret = change_xattr_items(inode, id, xat, bytes,
 					 xattr_nr_parts(xat), found_parts, lck);
@@ -620,12 +763,20 @@ release:
 		err = scoutfs_forest_srch_add(sb, hash, ino, id);
 		BUG_ON(err);
 	}
+	if (ret < 0 && undo_totl) {
+		/* _delta() on dirty items shouldn't fail */
+		tval.total = cpu_to_le64(-le64_to_cpu(tval.total));
+		tval.count = cpu_to_le64(-le64_to_cpu(tval.count));
+		err = apply_totl_delta(sb, &totl_key, &tval, totl_lock);
+		BUG_ON(err);
+	}
 
 	scoutfs_release_trans(sb);
 	scoutfs_inode_index_unlock(sb, &ind_locks);
 unlock:
 	up_write(&si->xattr_rwsem);
 	scoutfs_unlock(sb, lck, SCOUTFS_LOCK_WRITE);
+	scoutfs_unlock(sb, totl_lock, SCOUTFS_LOCK_WRITE_ONLY);
 out:
 	vfree(xat);
 
@@ -746,15 +897,22 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 {
 	struct scoutfs_xattr_prefix_tags tgs;
 	struct scoutfs_xattr *xat = NULL;
+	struct scoutfs_lock *totl_lock = NULL;
+	struct scoutfs_xattr_totl_val tval;
+	struct scoutfs_key totl_key;
 	struct scoutfs_key last;
 	struct scoutfs_key key;
 	bool release = false;
 	unsigned int bytes;
+	unsigned int val_len;
+	void *value;
+	u64 total;
 	u64 hash;
 	int ret;
 
-	/* need a buffer large enough for all possible names */
-	bytes = sizeof(struct scoutfs_xattr) + SCOUTFS_XATTR_MAX_NAME_LEN;
+	/* need a buffer large enough for all possible names and totl value */
+	bytes = sizeof(struct scoutfs_xattr) + SCOUTFS_XATTR_MAX_NAME_LEN +
+		SCOUTFS_XATTR_MAX_TOTL_U64;
 	xat = kmalloc(bytes, GFP_NOFS);
 	if (!xat) {
 		ret = -ENOMEM;
@@ -773,11 +931,37 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 			break;
 		}
 
+		if (key.skx_part == 0 && (ret < sizeof(struct scoutfs_xattr) ||
+		    ret < offsetof(struct scoutfs_xattr, name[xat->name_len]))) {
+			ret = -EIO;
+			break;
+		}
+
 		if (key.skx_part != 0 ||
 		    scoutfs_xattr_parse_tags(xat->name, xat->name_len,
 					     &tgs) != 0)
 			memset(&tgs, 0, sizeof(tgs));
 
+		if (tgs.totl) {
+			value = &xat->name[xat->name_len];
+			val_len = ret - offsetof(struct scoutfs_xattr, name[xat->name_len]);
+			if (val_len != le16_to_cpu(xat->val_len)) {
+				ret = -EIO;
+				goto out;
+			}
+
+			ret = parse_totl_key(&totl_key, xat->name, xat->name_len) ?:
+			      parse_totl_u64(value, val_len, &total);
+			if (ret < 0)
+				break;
+		}
+
+		if (tgs.totl && totl_lock == NULL) {
+			ret = scoutfs_lock_xattr_totl(sb, SCOUTFS_LOCK_WRITE_ONLY, 0, &totl_lock);
+			if (ret < 0)
+				break;
+		}
+
 		ret = scoutfs_hold_trans(sb, false);
 		if (ret < 0)
 			break;
@@ -795,6 +979,14 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 			       break;
 		}
 
+		if (tgs.totl) {
+			tval.total = cpu_to_le64(-total);
+			tval.count = cpu_to_le64(-1LL);
+			ret = apply_totl_delta(sb, &totl_key, &tval, totl_lock);
+			if (ret < 0)
+				break;
+		}
+
 		scoutfs_release_trans(sb);
 		release = false;
 
@@ -803,6 +995,7 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 
 	if (release)
 		scoutfs_release_trans(sb);
+	scoutfs_unlock(sb, totl_lock, SCOUTFS_LOCK_WRITE_ONLY);
 	kfree(xat);
 out:
 	return ret;

kmod/src/xattr.h

@@ -16,10 +16,14 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 
 struct scoutfs_xattr_prefix_tags {
 	unsigned long hide:1,
-		      srch:1;
+		      srch:1,
+		      totl:1;
 };
 
 int scoutfs_xattr_parse_tags(const char *name, unsigned int name_len,
 			     struct scoutfs_xattr_prefix_tags *tgs);
 
+void scoutfs_xattr_init_totl_key(struct scoutfs_key *key, u64 *name);
+int scoutfs_xattr_combine_totl(void *dst, int dst_len, void *src, int src_len);
+
 #endif

tests/golden/basic-posix-consistency

@@ -53,3 +53,5 @@ mv: cannot move ‘/mnt/test/test/basic-posix-consistency/dir/c/clobber’ to
 == inode indexes match after syncing existing
 == inode indexes match after copying and syncing
 == inode indexes match after removing and syncing
+== concurrent creates make one file
+one-file

tests/golden/totl-xattr-tag

@@ -0,0 +1,30 @@
+== single file
+1.2.3 = 1, 1
+4.5.6 = 1, 1
+== multiple files add up
+1.2.3 = 2, 2
+4.5.6 = 2, 2
+== removing xattr updates total
+1.2.3 = 2, 2
+4.5.6 = 1, 1
+== updating xattr updates total
+1.2.3 = 11, 2
+4.5.6 = 1, 1
+== removing files update total
+1.2.3 = 10, 1
+== multiple files/names in one transaction
+1.2.3 = 55, 10
+== testing invalid names
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+== testing invalid values
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+setfattr: /mnt/test/test/totl-xattr-tag/invalid: Invalid argument
+== larger population that could merge

tests/sequence

@@ -10,6 +10,7 @@ move-blocks.sh
 enospc.sh
 srch-basic-functionality.sh
 simple-xattr-unit.sh
+totl-xattr-tag.sh
 lock-refleak.sh
 lock-shrink-consistency.sh
 lock-pr-cw-conflict.sh

tests/tests/basic-posix-consistency.sh

@@ -197,4 +197,13 @@ scoutfs walk-inodes -p "$T_M0" -- data_seq 0 -1 > "$T_TMP.0"
 scoutfs walk-inodes -p "$T_M1" -- data_seq 0 -1 > "$T_TMP.1"
 diff -u "$T_TMP.0" "$T_TMP.1"
 
+echo "== concurrent creates make one file"
+mkdir "$T_D0/concurrent"
+for i in $(t_fs_nrs); do
+	eval p="\$T_D${i}/concurrent/one-file"
+	touch "$p" 2>&1 > "$T_TMP.multi-create.$i" &
+done
+wait
+ls "$T_D0/concurrent"
+
 t_pass

tests/tests/totl-xattr-tag.sh

@@ -0,0 +1,126 @@
+t_require_commands touch rm setfattr scoutfs find_xattrs
+
+read_xattr_totals()
+{
+	sync
+	scoutfs read-xattr-totals -p "$T_M0"
+}
+
+echo "== single file"
+touch "$T_D0/file-1"
+setfattr -n scoutfs.totl.test.1.2.3 -v 1 "$T_D0/file-1" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.4.5.6 -v 1 "$T_D0/file-1" 2>&1 | t_filter_fs
+read_xattr_totals
+
+echo "== multiple files add up"
+touch "$T_D0/file-2"
+setfattr -n scoutfs.totl.test.1.2.3 -v 1 "$T_D0/file-2" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.4.5.6 -v 1 "$T_D0/file-2" 2>&1 | t_filter_fs
+read_xattr_totals
+
+echo "== removing xattr updates total"
+setfattr -x scoutfs.totl.test.4.5.6 "$T_D0/file-2" 2>&1 | t_filter_fs
+read_xattr_totals
+
+echo "== updating xattr updates total"
+setfattr -n scoutfs.totl.test.1.2.3 -v 10 "$T_D0/file-2" 2>&1 | t_filter_fs
+read_xattr_totals
+
+echo "== removing files update total"
+rm -f "$T_D0/file-1"
+read_xattr_totals
+rm -f "$T_D0/file-2"
+read_xattr_totals
+
+echo "== multiple files/names in one transaction"
+for a in $(seq 1 10); do 
+	touch "$T_D0/file-$a"
+	setfattr -n scoutfs.totl.test.1.2.3 -v $a "$T_D0/file-$a" 2>&1 | t_filter_fs
+done
+read_xattr_totals
+rm -rf "$T_D0"/file-[0-9]*
+
+echo "== testing invalid names"
+touch "$T_D0/invalid"
+setfattr -n scoutfs.totl.test... -v 10 "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test..2.3 -v 10 "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1..3 -v 10 "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1.2. -v 10 "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1 -v 10 "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1.2 -v 10 "$T_D0/invalid" 2>&1 | t_filter_fs
+
+echo "== testing invalid values"
+setfattr -n scoutfs.totl.test.1.2.3 -v "+1" "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1.2.3 -v "10." "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1.2.3 -v "-" "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1.2.3 -v "junk10" "$T_D0/invalid" 2>&1 | t_filter_fs
+setfattr -n scoutfs.totl.test.1.2.3 -v "10junk" "$T_D0/invalid" 2>&1 | t_filter_fs
+rm -f "$T_D0/invalid"
+
+echo "== larger population that could merge"
+NR=5000
+TOTS=100
+CHECK=100
+PER_DIR=1000
+PER_FILE=10
+
+declare -A totals counts
+LOTS="$T_D0/lots"
+
+for i in $(seq 0 $PER_DIR $NR); do
+	p="$LOTS/$((i / PER_DIR))"
+	mkdir -p $p
+done
+for i in $(seq 0 $PER_FILE $NR); do
+	p="$LOTS/$((i / PER_DIR))/file-$((i / PER_FILE))"
+	touch $p
+done
+
+for phase in create update remove; do
+	for i in $(seq 0 $NR); do
+		p="$LOTS/$((i / PER_DIR))/file-$((i / PER_FILE))"
+
+		t=$((i % TOTS))
+		n="scoutfs.totl.test-$i.$t.0.0"
+
+		case $phase in 
+			create)
+				v="$i"
+				setfattr -n "$n" -v "$v" "$p" 2>&1 >> $T_TMP.sfa
+				((totals[$t]+=$v))
+				((counts[$t]++))
+				;;
+			update)
+				v=$((i * 3))
+				delta=$((i * 2))
+				setfattr -n "$n" -v "$v" "$p" 2>&1 >> $T_TMP.sfa
+				((totals[$t]+=$delta))
+				;;
+			remove)
+				v=$((i * 3))
+				setfattr -x "$n" "$p" 2>&1 >> $T_TMP.sfa
+				((totals[$t]-=$v))
+				((counts[$t]--))
+				;;
+		esac
+
+		if [ "$i" -gt 0 -a "$((i % CHECK))" == "0" ]; then
+			echo "checking $phase $i" > $T_TMP.check_arr
+			echo "checking $phase $i" > $T_TMP.check_read
+
+			( for k in ${!totals[@]}; do
+				echo "$k.0.0 = ${totals[$k]}, ${counts[$k]}"
+			  done ) | grep -v "= 0, 0$" | sort -n >> $T_TMP.check_arr
+
+			sync
+			read_xattr_totals | sort -n >> $T_TMP.check_read
+
+			diff -u $T_TMP.check_arr $T_TMP.check_read || \
+				t_fail "totals read didn't match expected arrays"
+		fi
+	done
+done
+
+rm -rf "$T_D0/merging"
+
+t_pass

utils/man/scoutfs.5

@@ -142,6 +142,62 @@ If the
 file is written to then the server cannot make forward progress and
 shuts down.  The request can similarly enter an errored state if enough
 time passes before userspace completes the request.
+
+.SH EXTENDED ATTRIBUTE TAGS
+
+.B scoutfs
+adds the
+.IB scoutfs.
+extended attribute namespace which uses a system of tags to extend the
+functionality of extended attributes.  Immediately following the
+scoutfs. prefix are a series of tag words seperated by dots.
+Any text starting after the last recognized tag is considered the xattr
+name and is not parsed.
+.sp
+Tags may be combined in any order.   Specifying a tag more than once
+will return an error.  There is no explicit boundary between the end of
+tags and the start of the name so unknown or incorrect tags will be
+successfully parsed as part of the name of the xattr.  Tags can only be
+created, updated, or removed with the CAP_SYS_ADMIN capability.
+
+The following tags are currently supported:
+
+.RS
+.TP
+.B .hide.
+Attributes with the .hide. tag are not visible to the
+.BR listxattr(2)
+system call.  They will instead be included in the output of the
+.IB LISTXATTR_HIDDEN
+ioctl.  This is meant to be used by archival management agents to store
+metadata that is bound to a specific volume and should not be
+transferred with the file by tools that read extended attributes, like
+.BR tar(1) .
+.TP
+.B .srch.
+Attributes with the .srch. tag are indexed so that they can be
+found by the
+.IB SEARCH_XATTRS
+ioctl.   The search ioctl takes an extended attribute name and returns
+the inode number of all the inodes which contain an extended attribute
+with that name.  The indexing structures behind .srch. tags are designed
+to efficiently handle a large number of .srch. attributes per file with
+no limits on the number of indexed files.
+.TP
+.B .totl.
+Attributes with the .totl. flag are used to efficiently maintain counts
+across all files in the system.  The attribute's name must end in three
+64bit values seperated by dots that specify the global total that the
+extended attribute will contribute to.   The value of the extended
+attribute is a string representation of the 64bit quantity which will be
+added to the total.   As attributes are added, updated, or removed (and
+particularly as a file is finally deleted), the corresponding global
+total is also updated by the file system.  All the totals with their
+name, total value, and a count of contributing attributes can be read
+with the
+.IB READ_XATTR_TOTALS
+ioctl.
+.RE
  
 .SH CORRUPTION DETECTION
 A

utils/src/btree.c

@@ -75,6 +75,9 @@ void btree_append_item(struct scoutfs_btree_block *bt,
 	le16_add_cpu(&bt->total_item_bytes, sizeof(struct scoutfs_btree_item));
 
 	item->key = *key;
+	item->seq = cpu_to_le64(1);
+	item->flags = 0;
+
 	leaf_item_hash_insert(bt, &item->key,
 			      cpu_to_le16((void *)item - (void *)bt));
 	if (val_len == 0)

utils/src/print.c

@@ -47,13 +47,14 @@ static void print_inode(struct scoutfs_key *key, void *val, int val_len)
 {
 	struct scoutfs_inode *inode = val;
 
-	printf("    inode: ino %llu size %llu nlink %u\n"
+	printf("    inode: ino %llu size %llu version %llu nlink %u\n"
 	       "      uid %u gid %u mode 0%o rdev 0x%x flags 0x%x\n"
 	       "      next_readdir_pos %llu meta_seq %llu data_seq %llu data_version %llu\n"
 	       "      atime %llu.%08u ctime %llu.%08u\n"
 	       "      mtime %llu.%08u\n",
 	       le64_to_cpu(key->ski_ino),
 	       le64_to_cpu(inode->size),
+	       le64_to_cpu(inode->version),
 	       le32_to_cpu(inode->nlink), le32_to_cpu(inode->uid),
 	       le32_to_cpu(inode->gid), le32_to_cpu(inode->mode),
 	       le32_to_cpu(inode->rdev),
@@ -75,6 +76,17 @@ static void print_orphan(struct scoutfs_key *key, void *val, int val_len)
 	printf("    orphan: ino %llu\n", le64_to_cpu(key->sko_ino));
 }
 
+
+static void print_xattr_totl(struct scoutfs_key *key, void *val, int val_len)
+{
+	struct scoutfs_xattr_totl_val *tval = val;
+
+	printf("    xattr totl: %llu.%llu.%llu = %lld, %lld\n",
+	       le64_to_cpu(key->skxt_a), le64_to_cpu(key->skxt_b),
+	       le64_to_cpu(key->skxt_c), le64_to_cpu(tval->total),
+	       le64_to_cpu(tval->count));
+}
+
 static u8 *global_printable_name(u8 *name, int name_len)
 {
 	static u8 name_buf[SCOUTFS_NAME_LEN + 1];
@@ -163,6 +175,9 @@ static print_func_t find_printer(u8 zone, u8 type)
 			return print_orphan;
 	}
 
+	if (zone == SCOUTFS_XATTR_TOTL_ZONE)
+		return print_xattr_totl;
+
 	if (zone == SCOUTFS_FS_ZONE) {
 		switch(type) {
 			case SCOUTFS_INODE_TYPE: return print_inode;
@@ -178,15 +193,19 @@ static print_func_t find_printer(u8 zone, u8 type)
 	return NULL;
 }
 
-static int print_fs_item(struct scoutfs_key *key, void *val,
+#define flag_char(val, bit, c) \
+	(((val) & (bit)) ? (c) : '-')
+
+static int print_fs_item(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 			 unsigned val_len, void *arg)
 {
 	print_func_t printer;
 
-	printf("    "SK_FMT"\n", SK_ARG(key));
+	printf("    "SK_FMT" %llu %c\n",
+	       SK_ARG(key), seq, flag_char(flags, SCOUTFS_ITEM_FLAG_DELETION, 'd'));
 
 	/* only items in leaf blocks have values */
-	if (val) {
+	if (val != NULL && !(flags & SCOUTFS_ITEM_FLAG_DELETION)) {
 		printer = find_printer(key->sk_zone, key->sk_type);
 		if (printer)
 			printer(key, val, val_len);
@@ -198,37 +217,6 @@ static int print_fs_item(struct scoutfs_key *key, void *val,
 	return 0;
 }
 
-/* same as fs item but with a small header in the value */
-static int print_logs_item(struct scoutfs_key *key, void *val,
-			   unsigned val_len, void *arg)
-{
-	struct scoutfs_log_item_value *liv;
-	print_func_t printer;
-
-	printf("    "SK_FMT"\n", SK_ARG(key));
-
-	/* only items in leaf blocks have values */
-	if (val) {
-		liv = val;
-		printf("    log_item_value: seq %llu flags %x\n",
-		       le64_to_cpu(liv->seq), liv->flags);
-
-		/* deletion items don't have values */
-		if (!(liv->flags & SCOUTFS_LOG_ITEM_FLAG_DELETION)) {
-			printer = find_printer(key->sk_zone,
-					       key->sk_type);
-			if (printer)
-				printer(key, val + sizeof(*liv),
-					val_len - sizeof(*liv));
-			else
-				printf("      (unknown zone %u type %u)\n",
-				       key->sk_zone, key->sk_type);
-		}
-	}
-
-	return 0;
-}
-
 #define BTREF_F \
 	"blkno %llu seq %llu"
 #define BTREF_A(ref) \
@@ -269,7 +257,7 @@ static int print_logs_item(struct scoutfs_key *key, void *val,
 	le64_to_cpu((srf)->ref.seq)
 
 /* same as fs item but with a small header in the value */
-static int print_log_trees_item(struct scoutfs_key *key, void *val,
+static int print_log_trees_item(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 				unsigned val_len, void *arg)
 {
 	struct scoutfs_log_trees *lt = val;
@@ -330,7 +318,7 @@ static int print_log_trees_item(struct scoutfs_key *key, void *val,
 	return 0;
 }
 
-static int print_srch_root_item(struct scoutfs_key *key, void *val,
+static int print_srch_root_item(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 				unsigned val_len, void *arg)
 {
 	struct scoutfs_srch_compact *sc;
@@ -363,7 +351,7 @@ static int print_srch_root_item(struct scoutfs_key *key, void *val,
 	return 0;
 }
 
-static int print_trans_seqs_entry(struct scoutfs_key *key, void *val,
+static int print_trans_seqs_entry(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 				  unsigned val_len, void *arg)
 {
 	printf("    trans_seq %llu rid %016llx\n",
@@ -372,7 +360,7 @@ static int print_trans_seqs_entry(struct scoutfs_key *key, void *val,
 	return 0;
 }
 
-static int print_mounted_client_entry(struct scoutfs_key *key, void *val,
+static int print_mounted_client_entry(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 				      unsigned val_len, void *arg)
 {
 	struct scoutfs_mounted_client_btree_val *mcv = val;
@@ -387,8 +375,8 @@ static int print_mounted_client_entry(struct scoutfs_key *key, void *val,
 	return 0;
 }
 
-static int print_log_merge_item(struct scoutfs_key *key, void *val,
-				      unsigned val_len, void *arg)
+static int print_log_merge_item(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
+				unsigned val_len, void *arg)
 {
 	struct scoutfs_log_merge_status *stat;
 	struct scoutfs_log_merge_range *rng;
@@ -451,7 +439,7 @@ static int print_log_merge_item(struct scoutfs_key *key, void *val,
 	return 0;
 }
 
-static int print_alloc_item(struct scoutfs_key *key, void *val,
+static int print_alloc_item(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 			    unsigned val_len, void *arg)
 {
 	if (key->sk_zone == SCOUTFS_FREE_EXTENT_BLKNO_ZONE)
@@ -469,7 +457,7 @@ static int print_alloc_item(struct scoutfs_key *key, void *val,
 	return 0;
 }
 
-typedef int (*print_item_func)(struct scoutfs_key *key, void *val,
+typedef int (*print_item_func)(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 			       unsigned val_len, void *arg);
 
 static int print_block_ref(struct scoutfs_key *key, void *val,
@@ -477,7 +465,7 @@ static int print_block_ref(struct scoutfs_key *key, void *val,
 {
 	struct scoutfs_block_ref *ref = val;
 
-	func(key, NULL, 0, arg);
+	func(key, 0, 0, NULL, 0, arg);
 	printf("    ref blkno %llu seq %llu\n",
 		le64_to_cpu(ref->blkno), le64_to_cpu(ref->seq));
 
@@ -586,7 +574,7 @@ static int print_btree_block(int fd, struct scoutfs_super_block *super,
 		if (level)
 			print_block_ref(key, val, val_len, func, arg);
 		else
-			func(key, val, val_len, arg);
+			func(key, le64_to_cpu(item->seq), item->flags, val, val_len, arg);
 	}
 
 	free(bt);
@@ -744,8 +732,8 @@ struct print_recursion_args {
 };
 
 /* same as fs item but with a small header in the value */
-static int print_log_trees_roots(struct scoutfs_key *key, void *val,
-				unsigned val_len, void *arg)
+static int print_log_trees_roots(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
+				 unsigned val_len, void *arg)
 {
 	struct scoutfs_log_trees *lt = val;
 	struct print_recursion_args *pa = arg;
@@ -776,14 +764,14 @@ static int print_log_trees_roots(struct scoutfs_key *key, void *val,
 		ret = err;
 
 	err = print_btree(pa->fd, pa->super, "", &lt->item_root,
-			  print_logs_item, NULL);
+			  print_fs_item, NULL);
 	if (err && !ret)
 		ret = err;
 
 	return ret;
 }
 
-static int print_srch_root_files(struct scoutfs_key *key, void *val,
+static int print_srch_root_files(struct scoutfs_key *key, u64 seq, u8 flags, void *val,
 				 unsigned val_len, void *arg)
 {
 	struct print_recursion_args *pa = arg;
@@ -843,7 +831,7 @@ static int print_btree_leaf_items(int fd, struct scoutfs_super_block *super,
 				break;
 			continue;
 		} else {
-			func(key, val, val_len, arg);
+			func(key, le64_to_cpu(item->seq), item->flags, val, val_len, arg);
 		}
 
 		node = avl_next(&bt->item_root, node);

utils/src/read_xattr_totals.c

@@ -0,0 +1,120 @@
+#include <unistd.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/ioctl.h>
+#include <fcntl.h>
+#include <errno.h>
+#include <string.h>
+#include <argp.h>
+
+#include "sparse.h"
+#include "parse.h"
+#include "util.h"
+#include "format.h"
+#include "ioctl.h"
+#include "cmd.h"
+
+struct xattr_args {
+	char *path;
+};
+
+static int do_read_xattr_totals(struct xattr_args *args)
+{
+	struct scoutfs_ioctl_read_xattr_totals rxt;
+	struct scoutfs_ioctl_xattr_total *xts = NULL;
+	struct scoutfs_ioctl_xattr_total *xt;
+	u64 bytes = 1024 * 1024;
+	int fd = -1;
+	int ret;
+	int i;
+
+	xts = malloc(bytes);
+	if (!xts) {
+		fprintf(stderr, "xattr total mem alloc failed\n");
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	fd = get_path(args->path, O_RDONLY);
+	if (fd < 0)
+		return fd;
+
+	memset(&rxt, 0, sizeof(rxt));
+	rxt.totals_ptr = (unsigned long)xts;
+	rxt.totals_bytes = bytes;
+
+	for (;;) {
+		ret = ioctl(fd, SCOUTFS_IOC_READ_XATTR_TOTALS, &rxt);
+		if (ret == 0)
+			break;
+		if (ret < 0) {
+			ret = -errno;
+			fprintf(stderr, "read_xattr_totals ioctl failed: "
+				"%s (%d)\n", strerror(errno), errno);
+			goto out;
+		}
+
+		for (i = 0, xt = xts; i < ret; i++, xt++)
+			printf("%llu.%llu.%llu = %lld, %lld\n",
+				xt->name[0], xt->name[1], xt->name[2], xt->total, xt->count);
+
+		memcpy(&rxt.pos_name, &xts[ret - 1].name, sizeof(rxt.pos_name));
+		if (++rxt.pos_name[2] == 0 && ++rxt.pos_name[1] == 0 && ++rxt.pos_name[0] == 0)
+			break;
+	}
+
+	ret = 0;
+out:
+	if (fd >= 0)
+		close(fd);
+	free(xts);
+
+	return ret;
+};
+
+static int parse_opt(int key, char *arg, struct argp_state *state)
+{
+	struct xattr_args *args = state->input;
+
+	switch (key) {
+	case 'p':
+		args->path = strdup_or_error(state, arg);
+		break;
+	default:
+		break;
+	}
+
+	return 0;
+}
+
+static struct argp_option options[] = {
+	{ "path", 'p', "PATH", 0, "Path to ScoutFS filesystem"},
+	{ NULL }
+};
+
+static struct argp argp = {
+	options,
+	parse_opt,
+	"",
+	"Print global value totals of .totl. xattrs"
+};
+
+static int read_xattr_totals_cmd(int argc, char **argv)
+{
+
+	struct xattr_args xattr_args = {NULL};
+	int ret;
+
+	ret = argp_parse(&argp, argc, argv, 0, NULL, &xattr_args);
+	if (ret)
+		return ret;
+
+	return do_read_xattr_totals(&xattr_args);
+}
+
+static void __attribute__((constructor)) read_xattr_totals_ctor(void)
+{
+	cmd_register_argp("read-xattr-totals", &argp, GROUP_INFO, read_xattr_totals_cmd);
+}

utils/src/setattr.c

@@ -21,6 +21,7 @@
 struct setattr_args {
 	char *filename;
 	struct timespec ctime;
+	struct timespec crtime;
 	u64 data_version;
 	u64 i_size;
 	bool offline;
@@ -42,6 +43,8 @@ static int do_setattr(struct setattr_args *args)
 
 	sm.ctime_sec = args->ctime.tv_sec;
 	sm.ctime_nsec = args->ctime.tv_nsec;
+	sm.crtime_sec = args->crtime.tv_sec;
+	sm.crtime_nsec = args->crtime.tv_nsec;
 	sm.data_version = args->data_version;
 	if (args->offline)
 		sm.flags |= SCOUTFS_IOC_SETATTR_MORE_OFFLINE;
@@ -73,6 +76,11 @@ static int parse_opt(int key, char *arg, struct argp_state *state)
 		if (ret)
 			return ret;
 		break;
+	case 'r': /* timespec */
+		ret = parse_timespec(arg, &args->crtime);
+		if (ret)
+			return ret;
+		break;
 	case 'V': /* data version */
 		ret = parse_u64(arg, &args->data_version);
 		if (ret)
@@ -112,7 +120,8 @@ static int parse_opt(int key, char *arg, struct argp_state *state)
 }
 
 static struct argp_option options[] = {
-	{ "ctime", 't', "TIMESPEC", 0, "Set creation time using \"<seconds-since-epoch>.<nanoseconds>\" format"},
+	{ "ctime", 't', "TIMESPEC", 0, "Set change time using \"<seconds-since-epoch>.<nanoseconds>\" format"},
+	{ "crtime", 'r', "TIMESPEC", 0, "Set creation time using \"<seconds-since-epoch>.<nanoseconds>\" format"},
 	{ "data-version", 'V', "VERSION", 0, "Set data version"},
 	{ "size", 's', "SIZE", 0, "Set file size (bytes or KMGTP units). Requires --data-version"},
 	{ "offline", 'o', NULL, 0, "Set file contents as offline, not sparse. Requires --size"},

utils/src/stat.c

@@ -37,6 +37,7 @@ static struct stat_more_field inode_fields[] = {
 	INODE_FIELD(data_version),
 	INODE_FIELD(online_blocks),
 	INODE_FIELD(offline_blocks),
+	{ .name = "crtime", .offset = INODE_FIELD_OFF(crtime_sec) },
 	{ NULL, }
 };
 
@@ -60,6 +61,9 @@ static void print_inode_field(void *st, size_t off)
 		case INODE_FIELD_OFF(offline_blocks):
 			printf("%llu", stm->offline_blocks);
 			break;
+		case INODE_FIELD_OFF(crtime_sec):
+			printf("%llu.%09u", stm->crtime_sec, stm->crtime_nsec);
+			break;
 	};
 }