Add indx xattr tag support

Add support for the indx xattr tag which lets xattrs determine the sort
order of by their inode number in a global index.

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

kmod/src/format.h

@@ -14,6 +14,7 @@
 #define SCOUTFS_FORMAT_VERSION_FEAT_RETENTION	2
 #define SCOUTFS_FORMAT_VERSION_FEAT_PROJECT_ID	2
 #define SCOUTFS_FORMAT_VERSION_FEAT_QUOTA	2
+#define SCOUTFS_FORMAT_VERSION_FEAT_INDX_TAG	2
 
 /* statfs(2) f_type */
 #define SCOUTFS_SUPER_MAGIC	0x554f4353		/* "SCOU" */
@@ -595,6 +596,7 @@ struct scoutfs_log_merge_freeing {
 #define SCOUTFS_ORPHAN_ZONE			8
 #define SCOUTFS_QUOTA_ZONE			10
 #define SCOUTFS_XATTR_TOTL_ZONE			12
+#define SCOUTFS_XATTR_INDX_ZONE			14
 #define SCOUTFS_FS_ZONE				16
 #define SCOUTFS_LOCK_ZONE			20
 /* Items only stored in server btrees */

kmod/src/ioctl.c

@@ -1461,6 +1461,130 @@ static long scoutfs_ioc_mod_quota_rule(struct file *file, unsigned long arg, boo
 	return scoutfs_quota_mod_rule(sb, is_add, &irule);
 }
 
+struct read_index_buf {
+	int nr;
+	int size;
+	struct scoutfs_ioctl_xattr_index_entry ents[0];
+};
+
+#define READ_INDEX_BUF_MAX_ENTS \
+	((PAGE_SIZE - sizeof(struct read_index_buf)) / \
+		sizeof(struct scoutfs_ioctl_xattr_index_entry))
+
+/*
+ * This doesn't filter out duplicates, the caller filters them out to
+ * catch duplicates between iteration calls.
+ */
+static int read_index_cb(struct scoutfs_key *key, void *val, unsigned int val_len, void *cb_arg)
+{
+	struct read_index_buf *rib = cb_arg;
+	struct scoutfs_ioctl_xattr_index_entry *ent = &rib->ents[rib->nr];
+	u64 xid;
+
+	if (val_len != 0)
+		return -EIO;
+
+	/* discard the xid, they're not exposed to ioctl callers */
+	scoutfs_xattr_get_indx_key(key, &ent->major, &ent->minor, &ent->ino, &xid);
+
+	if (++rib->nr == rib->size)
+		return rib->nr;
+
+	return -EAGAIN;
+}
+
+static long scoutfs_ioc_read_xattr_index(struct file *file, unsigned long arg)
+{
+	struct super_block *sb = file_inode(file)->i_sb;
+	struct scoutfs_ioctl_read_xattr_index __user *urxi = (void __user *)arg;
+	struct scoutfs_ioctl_xattr_index_entry __user *uents;
+	struct scoutfs_ioctl_xattr_index_entry *ent;
+	struct scoutfs_ioctl_xattr_index_entry prev;
+	struct scoutfs_ioctl_read_xattr_index rxi;
+	struct read_index_buf *rib;
+	struct page *page = NULL;
+	struct scoutfs_key first;
+	struct scoutfs_key last;
+	struct scoutfs_key start;
+	struct scoutfs_key end;
+	int copied = 0;
+	int ret;
+	int i;
+
+	if (!capable(CAP_SYS_ADMIN)) {
+		ret = -EPERM;
+		goto out;
+	}
+
+	if (copy_from_user(&rxi, urxi, sizeof(rxi))) {
+		ret = -EFAULT;
+		goto out;
+	}
+	uents = (void __user *)rxi.entries_ptr;
+	rxi.entries_nr = min_t(u64, rxi.entries_nr, INT_MAX);
+
+	page = alloc_page(GFP_KERNEL);
+	if (!page) {
+		ret = -ENOMEM;
+		goto out;
+	}
+	rib = page_address(page);
+
+	scoutfs_xattr_init_indx_key(&first, rxi.first.major, rxi.first.minor, rxi.first.ino, 0);
+	scoutfs_xattr_init_indx_key(&last, rxi.last.major, rxi.last.minor, rxi.last.ino, U64_MAX);
+	scoutfs_xattr_indx_get_range(&start, &end);
+
+	if (scoutfs_key_compare(&first, &last) > 0) {
+		ret = -EINVAL;
+		goto out;
+	}
+
+	/* 0 ino doesn't exist, can't ever match entry to return */
+	memset(&prev, 0, sizeof(prev));
+
+	while (copied < rxi.entries_nr) {
+		rib->nr = 0;
+		rib->size = min_t(u64, rxi.entries_nr - copied, READ_INDEX_BUF_MAX_ENTS);
+		ret = scoutfs_wkic_iterate(sb, &first, &last, &start, &end,
+					   read_index_cb, rib);
+		if (ret < 0)
+			goto out;
+		if (rib->nr == 0)
+			break;
+
+		/*
+		 * Copy entries to userspace, skipping duplicate entries
+		 * that can result from multiple xattrs indexing an
+		 * inode at the same position and which can span
+		 * multiple cache iterations.  (Comparing in order of
+		 * most likely to change to fail fast.)
+		 */
+		for (i = 0, ent = rib->ents; i < rib->nr; i++, ent++) {
+			if (ent->ino == prev.ino && ent->minor == prev.minor &&
+			    ent->major == prev.major)
+				continue;
+
+			if (copy_to_user(&uents[copied], ent, sizeof(*ent))) {
+				ret = -EFAULT;
+				goto out;
+			}
+
+			prev = *ent;
+			copied++;
+		}
+
+		scoutfs_xattr_init_indx_key(&first, prev.major, prev.minor, prev.ino, U64_MAX);
+		scoutfs_key_inc(&first);
+	}
+
+	ret = copied;
+out:
+	if (page)
+		__free_page(page);
+
+	return ret;
+}
+
 long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 {
 	switch (cmd) {
@@ -1508,6 +1632,8 @@ long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		return scoutfs_ioc_mod_quota_rule(file, arg, true);
 	case SCOUTFS_IOC_DEL_QUOTA_RULE:
 		return scoutfs_ioc_mod_quota_rule(file, arg, false);
+	case SCOUTFS_IOC_READ_XATTR_INDEX:
+		return scoutfs_ioc_read_xattr_index(file, arg);
 	}
 
 	return -ENOTTY;

kmod/src/ioctl.h

@@ -819,4 +819,28 @@ struct scoutfs_ioctl_get_quota_rules {
 #define SCOUTFS_IOC_DEL_QUOTA_RULE \
 	_IOW(SCOUTFS_IOCTL_MAGIC, 22, struct scoutfs_ioctl_quota_rule)
 
+/*
+ * Inodes can be indexed in a global key space at a position determined
+ * by a .indx. tagged xattr.  The xattr name specifies the two index
+ * position values, with major having the more significant comparison
+ * order.
+ */
+struct scoutfs_ioctl_xattr_index_entry {
+	__u64 minor;
+	__u64 ino;
+	__u8 major;
+	__u8 _pad[7];
+};
+
+struct scoutfs_ioctl_read_xattr_index {
+	__u64 flags;
+	struct scoutfs_ioctl_xattr_index_entry first;
+	struct scoutfs_ioctl_xattr_index_entry last;
+	__u64 entries_ptr;
+	__u64 entries_nr;
+};
+
+#define SCOUTFS_IOC_READ_XATTR_INDEX \
+	_IOR(SCOUTFS_IOCTL_MAGIC, 23, struct scoutfs_ioctl_read_xattr_index)
+
 #endif

kmod/src/lock.c

@@ -1254,6 +1254,17 @@ int scoutfs_lock_xattr_totl(struct super_block *sb, enum scoutfs_lock_mode mode,
 	return lock_key_range(sb, mode, flags, &start, &end, lock);
 }
 
+int scoutfs_lock_xattr_indx(struct super_block *sb, enum scoutfs_lock_mode mode, int flags,
+			    struct scoutfs_lock **lock)
+{
+	struct scoutfs_key start;
+	struct scoutfs_key end;
+
+	scoutfs_xattr_indx_get_range(&start, &end);
+
+	return lock_key_range(sb, mode, flags, &start, &end, lock);
+}
+
 int scoutfs_lock_quota(struct super_block *sb, enum scoutfs_lock_mode mode, int flags,
 		       struct scoutfs_lock **lock)
 {

kmod/src/lock.h

@@ -86,6 +86,8 @@ int scoutfs_lock_orphan(struct super_block *sb, enum scoutfs_lock_mode mode, int
 		        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);
+int scoutfs_lock_xattr_indx(struct super_block *sb, enum scoutfs_lock_mode mode, int flags,
+			    struct scoutfs_lock **lock);
 int scoutfs_lock_quota(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,

kmod/src/xattr.c

@@ -94,6 +94,7 @@ static inline bool is_user(const char *name)
 }
 
 #define HIDE_TAG	"hide."
+#define INDX_TAG	"indx."
 #define SRCH_TAG	"srch."
 #define TOTL_TAG	"totl."
 #define TAG_LEN		(sizeof(HIDE_TAG) - 1)
@@ -115,6 +116,9 @@ int scoutfs_xattr_parse_tags(const char *name, unsigned int name_len,
 		if (!strncmp(name, HIDE_TAG, TAG_LEN)) {
 			if (++tgs->hide == 0)
 				return -EINVAL;
+		} else if (!strncmp(name, INDX_TAG, TAG_LEN)) {
+			if (++tgs->indx == 0)
+				return -EINVAL;
 		} else if (!strncmp(name, SRCH_TAG, TAG_LEN)) {
 			if (++tgs->srch == 0)
 				return -EINVAL;
@@ -552,47 +556,57 @@ static int parse_totl_u64(const char *s, int len, u64 *res)
 }
 
 /*
- * 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.
+ * non-destructive relatively quick parse of final dotted u64s in an
+ * xattr name.  If the required number of values are found then we
+ * return the number of bytes in the name that are not the final dotted
+ * u64s with their dots.  -EINVAL is returned if we didn't find the
+ * required number of values.
  */
-static int parse_totl_key(struct scoutfs_key *key, const char *name, int name_len)
+static int parse_dotted_u64s(u64 *u64s, int nr, const char *name, int name_len)
 {
-	u64 tot_name[3];
 	int end = name_len;
-	int nr = 0;
 	int len;
 	int ret;
 	int i;
+	int u;
 
 	/* 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--) {
+	for (u = nr - 1, i = name_len - 1; u >= 0 && i >= 0; i--) {
 		if (name[i] != '.')
 			continue;
 
 		len = end - (i + 1);
-		ret = parse_totl_u64(&name[i + 1], len, &tot_name[nr]);
+		ret = parse_totl_u64(&name[i + 1], len, &u64s[u]);
 		if (ret < 0)
 			goto out;
 
 		end = i;
-		nr++;
+		u--;
 	}
 
-	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 {
+	if (u == -1)
+		ret = end;
+	else
 		ret = -EINVAL;
-	}
 
 out:
 	return ret;
 }
 
+static int parse_totl_key(struct scoutfs_key *key, const char *name, int name_len)
+{
+	u64 u64s[3];
+	int ret;
+
+	ret = parse_dotted_u64s(u64s, ARRAY_SIZE(u64s), name, name_len);
+	if (ret >= 0) {
+		scoutfs_xattr_init_totl_key(key, u64s);
+		ret = 0;
+	}
+
+	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)
 {
@@ -619,6 +633,72 @@ int scoutfs_xattr_combine_totl(void *dst, int dst_len, void *src, int src_len)
 	return SCOUTFS_DELTA_COMBINED;
 }
 
+void scoutfs_xattr_indx_get_range(struct scoutfs_key *start, struct scoutfs_key *end)
+{
+	scoutfs_key_set_zeros(start);
+	start->sk_zone = SCOUTFS_XATTR_INDX_ZONE;
+	scoutfs_key_set_ones(end);
+	end->sk_zone = SCOUTFS_XATTR_INDX_ZONE;
+}
+
+/*
+ * .indx. keys are a bit funny because we're iterating over index keys
+ * by major:minor:inode:xattr_id.  That doesn't map nicely to the
+ * comparison precedence of the key fields.  We have to mess around a
+ * little bit to get the major into the most significant key bits and
+ * the low bits of xattr id into the least significant key bits.
+ */
+void scoutfs_xattr_init_indx_key(struct scoutfs_key *key, u8 major, u64 minor, u64 ino, u64 xid)
+{
+	scoutfs_key_set_zeros(key);
+	key->sk_zone = SCOUTFS_XATTR_INDX_ZONE;
+
+	key->_sk_first = cpu_to_le64(((u64)major << 56) | (minor >> 8));
+	key->_sk_second = cpu_to_le64((minor << 56) | (ino >> 8));
+	key->_sk_third = cpu_to_le64((ino << 56) | (xid >> 8));
+	key->_sk_fourth = xid & 0xff;
+}
+
+void scoutfs_xattr_get_indx_key(struct scoutfs_key *key, u8 *major, u64 *minor, u64 *ino, u64 *xid)
+{
+	*major = le64_to_cpu(key->_sk_first) >> 56;
+	*minor = (le64_to_cpu(key->_sk_first) << 8) | (le64_to_cpu(key->_sk_second) >> 56);
+	*ino = (le64_to_cpu(key->_sk_second) << 8) | (le64_to_cpu(key->_sk_third) >> 56);
+	*xid = (le64_to_cpu(key->_sk_third) << 8) | key->_sk_fourth;
+}
+
+void scoutfs_xattr_set_indx_key_xid(struct scoutfs_key *key, u64 xid)
+{
+	u8 major;
+	u64 minor;
+	u64 ino;
+	u64 dummy;
+
+	scoutfs_xattr_get_indx_key(key, &major, &minor, &ino, &dummy);
+	scoutfs_xattr_init_indx_key(key, major, minor, ino, xid);
+}
+
+/*
+ * This initial parsing of the name doesn't yet have access to an xattr
+ * id to put in the key.  That's added later as the existing xattr is
+ * found or a new xattr's id is allocated.
+ */
+static int parse_indx_key(struct scoutfs_key *key, const char *name, int name_len, u64 ino)
+{
+	u64 u64s[2];
+	int ret;
+
+	ret = parse_dotted_u64s(u64s, ARRAY_SIZE(u64s), name, name_len);
+	if (ret < 0)
+		return ret;
+
+	if (u64s[0] > U8_MAX)
+		return -EINVAL;
+
+	scoutfs_xattr_init_indx_key(key, u64s[0], u64s[1], ino, 0);
+	return 0;
+}
+
 /*
  * The confusing swiss army knife of creating, modifying, and deleting
  * xattrs.
@@ -639,7 +719,7 @@ int scoutfs_xattr_combine_totl(void *dst, int dst_len, void *src, int src_len)
 int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_len,
 			     const void *value, size_t size, int flags,
 			     const struct scoutfs_xattr_prefix_tags *tgs,
-			     struct scoutfs_lock *lck, struct scoutfs_lock *totl_lock,
+			     struct scoutfs_lock *lck, struct scoutfs_lock *tag_lock,
 			     struct list_head *ind_locks)
 {
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
@@ -647,10 +727,11 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 	const u64 ino = scoutfs_ino(inode);
 	struct scoutfs_xattr_totl_val tval = {0,};
 	struct scoutfs_xattr *xat = NULL;
-	struct scoutfs_key totl_key;
+	struct scoutfs_key tag_key;
 	struct scoutfs_key key;
 	bool undo_srch = false;
 	bool undo_totl = false;
+	bool undo_indx = false;
 	u8 found_parts;
 	unsigned int xat_bytes_totl;
 	unsigned int xat_bytes;
@@ -663,7 +744,8 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 
 	trace_scoutfs_xattr_set(sb, name_len, value, size, flags);
 
-	if (WARN_ON_ONCE(tgs->totl && !totl_lock))
+	if (WARN_ON_ONCE(tgs->totl && tgs->indx) ||
+	    WARN_ON_ONCE((tgs->totl | tgs->indx) && !tag_lock))
 		return -EINVAL;
 
 	/* mirror the syscall's errors for large names and values */
@@ -676,10 +758,17 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 	    (flags & ~(XATTR_CREATE | XATTR_REPLACE)))
 		return -EINVAL;
 
-	if ((tgs->hide | tgs->srch | tgs->totl) && !capable(CAP_SYS_ADMIN))
+	if ((tgs->hide | tgs->indx | tgs->srch | tgs->totl) && !capable(CAP_SYS_ADMIN))
 		return -EPERM;
 
-	if (tgs->totl && ((ret = parse_totl_key(&totl_key, name, name_len)) != 0))
+	if (tgs->totl && ((ret = parse_totl_key(&tag_key, name, name_len)) != 0))
+		return ret;
+
+	if (tgs->indx &&
+	    (ret = scoutfs_fmt_vers_unsupported(sb, SCOUTFS_FORMAT_VERSION_FEAT_INDX_TAG)))
+		return ret;
+
+	if (tgs->indx && ((ret = parse_indx_key(&tag_key, name, name_len, ino)) != 0))
 		return ret;
 
 	/* retention blocks user. xattr modification, all else allowed */
@@ -725,6 +814,12 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 	/* found fields in key will also be used */
 	found_parts = ret >= 0 ? xattr_nr_parts(xat) : 0;
 
+	/* use existing xattr's id or allocate new when creating */
+	if (found_parts)
+		id = le64_to_cpu(key.skx_id);
+	else if (value)
+		id = si->next_xattr_id++;
+
 	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]);
@@ -735,12 +830,25 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 		le64_add_cpu(&tval.total, -total);
 	}
 
+	/*
+	 * indx xattrs don't have a value.  After returning an error for
+	 * non-zero val length or short circuiting modifying with the
+	 * same 0 length, all we're left with is creating or deleting
+	 * the xattr.
+	 */
+	if (tgs->indx) {
+		if (size != 0) {
+			ret = -EINVAL;
+			goto out;
+		}
+		if (found_parts && value) {
+			ret = 0;
+			goto out;
+		}
+	}
+
 	/* prepare the xattr header, name, and start of value in first item */
 	if (value) {
-		if (found_parts)
-			id = le64_to_cpu(key.skx_id);
-		else
-			id = si->next_xattr_id++;
 		xat->name_len = name_len;
 		xat->val_len = cpu_to_le16(size);
 		memset(xat->__pad, 0, sizeof(xat->__pad));
@@ -758,9 +866,18 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 		le64_add_cpu(&tval.total, total);
 	}
 
+	if (tgs->indx) {
+		scoutfs_xattr_set_indx_key_xid(&tag_key, id);
+		if (value)
+			ret = scoutfs_item_create_force(sb, &tag_key, NULL, 0, tag_lock, NULL);
+		else
+			ret = scoutfs_item_delete_force(sb, &tag_key, tag_lock, NULL);
+		if (ret < 0)
+			goto out;
+		undo_indx = true;
+	}
+
 	if (tgs->srch && !(found_parts && value)) {
-		if (found_parts)
-			id = le64_to_cpu(key.skx_id);
 		hash = scoutfs_hash64(name, name_len);
 		ret = scoutfs_forest_srch_add(sb, hash, ino, id);
 		if (ret < 0)
@@ -769,7 +886,7 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 	}
 
 	if (tgs->totl) {
-		ret = apply_totl_delta(sb, &totl_key, &tval, totl_lock);
+		ret = apply_totl_delta(sb, &tag_key, &tval, tag_lock);
 		if (ret < 0)
 			goto out;
 		undo_totl = true;
@@ -794,6 +911,13 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_
 	ret = 0;
 
 out:
+	if (ret < 0 && undo_indx) {
+		if (value)
+			err = scoutfs_item_delete_force(sb, &tag_key, tag_lock, NULL);
+		else
+			err = scoutfs_item_create_force(sb, &tag_key, NULL, 0, tag_lock, NULL);
+		BUG_ON(err); /* inconsistent */
+	}
 	if (ret < 0 && undo_srch) {
 		err = scoutfs_forest_srch_add(sb, hash, ino, id);
 		BUG_ON(err);
@@ -802,7 +926,7 @@ out:
 		/* _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);
+		err = apply_totl_delta(sb, &tag_key, &tval, tag_lock);
 		BUG_ON(err);
 	}
 
@@ -818,7 +942,7 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name, const void
 	struct inode *inode = dentry->d_inode;
 	struct super_block *sb = inode->i_sb;
 	struct scoutfs_xattr_prefix_tags tgs;
-	struct scoutfs_lock *totl_lock = NULL;
+	struct scoutfs_lock *tag_lock = NULL;
 	struct scoutfs_lock *lck = NULL;
 	size_t name_len = strlen(name);
 	LIST_HEAD(ind_locks);
@@ -833,8 +957,11 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name, const void
 	if (ret)
 		goto unlock;
 
-	if (tgs.totl) {
-		ret = scoutfs_lock_xattr_totl(sb, SCOUTFS_LOCK_WRITE_ONLY, 0, &totl_lock);
+	if (tgs.totl || tgs.indx) {
+		if (tgs.totl)
+			ret = scoutfs_lock_xattr_totl(sb, SCOUTFS_LOCK_WRITE_ONLY, 0, &tag_lock);
+		else
+			ret = scoutfs_lock_xattr_indx(sb, SCOUTFS_LOCK_WRITE_ONLY, 0, &tag_lock);
 		if (ret)
 			goto unlock;
 	}
@@ -853,7 +980,7 @@ retry:
 		goto release;
 
 	ret = scoutfs_xattr_set_locked(dentry->d_inode, name, name_len, value, size, flags, &tgs,
-				       lck, totl_lock, &ind_locks);
+				       lck, tag_lock, &ind_locks);
 	if (ret == 0)
 		scoutfs_update_inode_item(inode, lck, &ind_locks);
 
@@ -862,7 +989,7 @@ release:
 	scoutfs_inode_index_unlock(sb, &ind_locks);
 unlock:
 	scoutfs_unlock(sb, lck, SCOUTFS_LOCK_WRITE);
-	scoutfs_unlock(sb, totl_lock, SCOUTFS_LOCK_WRITE_ONLY);
+	scoutfs_unlock(sb, tag_lock, SCOUTFS_LOCK_WRITE_ONLY);
 
 	return ret;
 }
@@ -1072,14 +1199,15 @@ 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_lock *tag_lock = NULL;
 	struct scoutfs_xattr_totl_val tval;
-	struct scoutfs_key totl_key;
+	struct scoutfs_key tag_key;
 	struct scoutfs_key last;
 	struct scoutfs_key key;
 	bool release = false;
 	unsigned int bytes;
 	unsigned int val_len;
+	u8 locked_zone = 0;
 	void *value;
 	u64 total;
 	u64 hash;
@@ -1125,16 +1253,32 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 				goto out;
 			}
 
-			ret = parse_totl_key(&totl_key, xat->name, xat->name_len) ?:
+			ret = parse_totl_key(&tag_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 (tgs.indx) {
+			ret = parse_indx_key(&tag_key, xat->name, xat->name_len, ino);
+			if (ret < 0)
+				goto out;
+		}
+
+		if ((tgs.totl || tgs.indx) && locked_zone != tag_key.sk_zone) {
+			if (tag_lock) {
+				scoutfs_unlock(sb, tag_lock, SCOUTFS_LOCK_WRITE_ONLY);
+				tag_lock = NULL;
+			}
+			if (tgs.totl)
+				ret = scoutfs_lock_xattr_totl(sb, SCOUTFS_LOCK_WRITE_ONLY, 0,
+							      &tag_lock);
+			else
+				ret = scoutfs_lock_xattr_indx(sb, SCOUTFS_LOCK_WRITE_ONLY, 0,
+							      &tag_lock);
 			if (ret < 0)
 				break;
+			locked_zone = tag_key.sk_zone;
 		}
 
 		ret = scoutfs_hold_trans(sb, false);
@@ -1157,7 +1301,13 @@ int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 		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);
+			ret = apply_totl_delta(sb, &tag_key, &tval, tag_lock);
+			if (ret < 0)
+				break;
+		}
+
+		if (tgs.indx) {
+			ret = scoutfs_item_delete_force(sb, &tag_key, tag_lock, NULL);
 			if (ret < 0)
 				break;
 		}
@@ -1170,7 +1320,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);
+	scoutfs_unlock(sb, tag_lock, SCOUTFS_LOCK_WRITE_ONLY);
 	kfree(xat);
 out:
 	return ret;

kmod/src/xattr.h

@@ -3,6 +3,7 @@
 
 struct scoutfs_xattr_prefix_tags {
 	unsigned long hide:1,
+		      indx:1,
 		      srch:1,
 		      totl:1;
 };
@@ -30,4 +31,9 @@ int scoutfs_xattr_parse_tags(const char *name, unsigned int name_len,
 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);
 
+void scoutfs_xattr_indx_get_range(struct scoutfs_key *start, struct scoutfs_key *end);
+void scoutfs_xattr_init_indx_key(struct scoutfs_key *key, u8 major, u64 minor, u64 ino, u64 xid);
+void scoutfs_xattr_get_indx_key(struct scoutfs_key *key, u8 *major, u64 *minor, u64 *ino, u64 *xid);
+void scoutfs_xattr_set_indx_key_xid(struct scoutfs_key *key, u64 xid);
+
 #endif