scoutfs: add index of inodes by xattr names

Add a .indx. xattr tag which adds the inode to an index of inodes keyed
by the hash of xattr names.  An ioctl is added which then returns all
the inodes which may contain an xattr of the given name.  Dropping all
xattrs now has to parse the name to find out if it also has to delete an
index item.

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

kmod/src/count.h

@@ -205,12 +205,14 @@ static inline const struct scoutfs_item_count SIC_RENAME(unsigned old_len,
  * item with the header and name.  Any previously existing items are
  * deleted which dirties their key but removes their value.  The two
  * sets of items are indexed by different ids so their items don't
- * overlap.
+ * overlap.  If the xattr name is indexed then we modify one xattr index
+ * item.
  */
 static inline const struct scoutfs_item_count SIC_XATTR_SET(unsigned old_parts,
 							    bool creating,
 							    unsigned name_len,
-							    unsigned size)
+							    unsigned size,
+							    bool indexed)
 {
 	struct scoutfs_item_count cnt = {0,};
 	unsigned int new_parts;
@@ -219,6 +221,8 @@ static inline const struct scoutfs_item_count SIC_XATTR_SET(unsigned old_parts,
 
 	if (old_parts)
 		cnt.items += old_parts;
+	if (indexed)
+		cnt.items++;
 
 	if (creating) {
 		new_parts = SCOUTFS_XATTR_NR_PARTS(name_len, size)

kmod/src/format.h

@@ -108,6 +108,11 @@ struct scoutfs_key {
 #define skii_major	_sk_second
 #define skii_ino	_sk_third
 
+/* xattr index */
+#define skxi_hash	_sk_first
+#define skxi_ino	_sk_second
+#define skxi_id		_sk_third
+
 /* node free extent */
 #define sknf_node_id	_sk_first
 #define sknf_major	_sk_second
@@ -351,9 +356,10 @@ struct scoutfs_segment_block {
  * Keys are first sorted by major key zones.
  */
 #define SCOUTFS_INODE_INDEX_ZONE		1
-#define SCOUTFS_NODE_ZONE			2
-#define SCOUTFS_FS_ZONE				3
-#define SCOUTFS_LOCK_ZONE			4
+#define SCOUTFS_XATTR_INDEX_ZONE		2
+#define SCOUTFS_NODE_ZONE			3
+#define SCOUTFS_FS_ZONE				4
+#define SCOUTFS_LOCK_ZONE			5
 #define SCOUTFS_MAX_ZONE			8 /* power of 2 is efficient */
 
 /* inode index zone */
@@ -361,6 +367,9 @@ struct scoutfs_segment_block {
 #define SCOUTFS_INODE_INDEX_DATA_SEQ_TYPE	2
 #define SCOUTFS_INODE_INDEX_NR			3 /* don't forget to update */
 
+/* xattr index zone */
+#define SCOUTFS_XATTR_INDEX_NAME_TYPE		1
+
 /* node zone (also used in server alloc btree) */
 #define SCOUTFS_FREE_EXTENT_BLKNO_TYPE		1
 #define SCOUTFS_FREE_EXTENT_BLOCKS_TYPE		2

kmod/src/ioctl.c

@@ -34,6 +34,7 @@
 #include "manifest.h"
 #include "trans.h"
 #include "xattr.h"
+#include "hash.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -747,6 +748,100 @@ out:
 	return ret ?: total;
 }
 
+/*
+ * Return the inode numbers of inodes which might contain the given
+ * named xattr.  This will only find scoutfs xattrs with the index tag
+ * but we don't check that the callers xattr name contains the tag and
+ * search for it regardless.
+ */
+static long scoutfs_ioc_find_xattrs(struct file *file, unsigned long arg)
+{
+	struct super_block *sb = file_inode(file)->i_sb;
+	struct scoutfs_ioctl_find_xattrs __user *ufx = (void __user *)arg;
+	struct scoutfs_ioctl_find_xattrs fx;
+	struct scoutfs_lock *lock = NULL;
+	struct scoutfs_key last;
+	struct scoutfs_key key;
+	char *name = NULL;
+	int total = 0;
+	u64 hash;
+	u64 ino;
+	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(&fx, ufx, sizeof(fx))) {
+		ret = -EFAULT;
+		goto out;
+	}
+
+	if (fx.name_bytes > SCOUTFS_XATTR_MAX_NAME_LEN) {
+		ret = -EINVAL;
+		goto out;
+	}
+
+	name = kmalloc(fx.name_bytes, GFP_KERNEL);
+	if (!name) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
+	if (copy_from_user(name, (void __user *)fx.name_ptr, fx.name_bytes)) {
+		ret = -EFAULT;
+		goto out;
+	}
+
+	hash = scoutfs_hash64(name, fx.name_bytes);
+	scoutfs_xattr_index_key(&key, hash, fx.next_ino, 0);
+	scoutfs_xattr_index_key(&last, hash, U64_MAX, U64_MAX);
+	ino = 0;
+
+	ret = scoutfs_lock_xattr_index(sb, SCOUTFS_LOCK_READ, 0, hash, &lock);
+	if (ret < 0)
+		goto out;
+
+	while (fx.nr_inodes) {
+
+		ret = scoutfs_item_next(sb, &key, &last, NULL, lock);
+		if (ret < 0) {
+			if (ret == -ENOENT)
+				ret = 0;
+			break;
+		}
+
+		/* xattrs hashes can collide and add multiple entries */
+		if (le64_to_cpu(key.skxi_ino) != ino) {
+			ino = le64_to_cpu(key.skxi_ino);
+			if (put_user(ino, (u64 __user *)fx.inodes_ptr)) {
+				ret = -EFAULT;
+				break;
+			}
+
+			fx.inodes_ptr += sizeof(u64);
+			fx.nr_inodes--;
+			total++;
+			ret = 0;
+		}
+
+		scoutfs_key_inc(&key);
+	}
+
+	scoutfs_unlock(sb, lock, SCOUTFS_LOCK_READ);
+
+out:
+	kfree(name);
+
+	return ret ?: total;
+}
+
 long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 {
 	switch (cmd) {
@@ -768,6 +863,8 @@ long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		return scoutfs_ioc_setattr_more(file, arg);
 	case SCOUTFS_IOC_LISTXATTR_RAW:
 		return scoutfs_ioc_listxattr_raw(file, arg);
+	case SCOUTFS_IOC_FIND_XATTRS:
+		return scoutfs_ioc_find_xattrs(file, arg);
 	}
 
 	return -ENOTTY;

kmod/src/ioctl.h

@@ -281,4 +281,34 @@ struct scoutfs_ioctl_listxattr_raw {
 #define SCOUTFS_IOC_LISTXATTR_RAW _IOW(SCOUTFS_IOCTL_MAGIC, 11, \
 				       struct scoutfs_ioctl_listxattr_raw)
 
+/*
+ * Return the inode numbers of inodes which might contain the given
+ * named xattr.  The inode may not have a set xattr with that name, the
+ * caller must check the returned inodes to see if they match.
+ *
+ * @next_ino: The next inode number that could be returned.  Initialized
+ * to 0 when first searching and set to one past the last inode number
+ * returned to continue searching.
+ * @name_ptr: The address of the name of the xattr to search for.  It does
+ * not need to be null terminated.
+ * @inodes_ptr: The address of the array of uint64_t inode numbers in which
+ * to store inode numbers that may contain the xattr.  EFAULT may be returned
+ * if this address is not naturally aligned.
+ * @name_bytes: The number of non-null bytes found in the name at name_ptr.
+ * @nr_inodes: The number of elements in the array found at inodes_ptr.
+ *
+ * This requires the CAP_SYS_ADMIN capability and will return -EPERM if
+ * it's not granted.
+ */
+struct scoutfs_ioctl_find_xattrs {
+	__u64 next_ino;
+	__u64 name_ptr;
+	__u64 inodes_ptr;
+	__u16 name_bytes;
+	__u16 nr_inodes;
+} __packed;
+
+#define SCOUTFS_IOC_FIND_XATTRS _IOW(SCOUTFS_IOCTL_MAGIC, 12, \
+				       struct scoutfs_ioctl_find_xattrs)
+
 #endif

kmod/src/lock.c

@@ -33,6 +33,7 @@
 #include "tseq.h"
 #include "client.h"
 #include "data.h"
+#include "xattr.h"
 
 /*
  * scoutfs uses a lock service to manage item cache consistency between
@@ -1157,6 +1158,24 @@ int scoutfs_lock_inode_index(struct super_block *sb, int mode,
 	return lock_key_range(sb, mode, 0, &start, &end, ret_lock);
 }
 
+/*
+ * Today we lock a hash value entirely. If we went to finer grained ino
+ * locking as well we'd need to check the manifest to find the next
+ * possible ino to lock so that we didn't try to iterate over all of
+ * them.
+ */
+int scoutfs_lock_xattr_index(struct super_block *sb, int mode, int flags,
+			     u64 hash, struct scoutfs_lock **ret_lock)
+{
+	struct scoutfs_key start;
+	struct scoutfs_key end;
+
+	scoutfs_xattr_index_key(&start, hash, 0, 0);
+	scoutfs_xattr_index_key(&end, hash, U64_MAX, U64_MAX);
+
+	return lock_key_range(sb, mode, flags, &start, &end, ret_lock);
+}
+
 /*
  * The node_id lock protects a mount's private persistent items in the
  * node_id zone.  It's held for the duration of the mount.  It lets the

kmod/src/lock.h

@@ -61,6 +61,8 @@ void scoutfs_lock_get_index_item_range(u8 type, u64 major, u64 ino,
 int scoutfs_lock_inode_index(struct super_block *sb, int mode,
 			     u8 type, u64 major, u64 ino,
 			     struct scoutfs_lock **ret_lock);
+int scoutfs_lock_xattr_index(struct super_block *sb, int mode, int flags,
+			     u64 hash, struct scoutfs_lock **ret_lock);
 int scoutfs_lock_inodes(struct super_block *sb, int mode, int flags,
 			struct inode *a, struct scoutfs_lock **a_lock,
 			struct inode *b, struct scoutfs_lock **b_lock,

kmod/src/xattr.c

@@ -25,6 +25,7 @@
 #include "trans.h"
 #include "xattr.h"
 #include "lock.h"
+#include "hash.h"
 #include "scoutfs_trace.h"
 
 /*
@@ -94,40 +95,58 @@ static int unknown_prefix(const char *name)
 }
 
 struct prefix_tags {
-	unsigned long hide:1;
+	unsigned long hide:1,
+		      indx:1;
 };
 
 #define HIDE_TAG	"hide."
-#define HIDE_TAG_LEN	(sizeof(HIDE_TAG) - 1)
+#define INDX_TAG	"indx."
+#define TAG_LEN		(sizeof(HIDE_TAG) - 1)
 
-static int parse_tags(const char *name, struct prefix_tags *tgs)
+static int parse_tags(const char *name, unsigned int name_len,
+		      struct prefix_tags *tgs)
 {
 	bool found;
 
 	memset(tgs, 0, sizeof(struct prefix_tags));
 
-	if (strncmp(name, SCOUTFS_XATTR_PREFIX, SCOUTFS_XATTR_PREFIX_LEN))
+	if ((name_len < (SCOUTFS_XATTR_PREFIX_LEN + TAG_LEN + 1)) ||
+	    strncmp(name, SCOUTFS_XATTR_PREFIX, SCOUTFS_XATTR_PREFIX_LEN))
 		return 0;
 	name += SCOUTFS_XATTR_PREFIX_LEN;
 
 	found = false;
 	for (;;) {
-		if (!strncmp(name, HIDE_TAG, HIDE_TAG_LEN)) {
+		if (!strncmp(name, HIDE_TAG, TAG_LEN)) {
 			if (++tgs->hide == 0)
 				return -EINVAL;
-			name += HIDE_TAG_LEN;
+		} else if (!strncmp(name, INDX_TAG, TAG_LEN)) {
+			if (++tgs->indx == 0)
+				return -EINVAL;
 		} else {
 			/* only reason to use scoutfs. is tags */
 			if (!found)
 				return -EINVAL;
 			break;
 		}
+		name += TAG_LEN;
 		found = true;
 	}
 
 	return 0;
 }
 
+void scoutfs_xattr_index_key(struct scoutfs_key *key,
+			     u64 hash, u64 ino, u64 id)
+{
+	scoutfs_key_set_zeros(key);
+	key->sk_zone = SCOUTFS_XATTR_INDEX_ZONE;
+	key->skxi_hash = cpu_to_le64(hash);
+	key->sk_type = SCOUTFS_XATTR_INDEX_NAME_TYPE;
+	key->skxi_ino = cpu_to_le64(ino);
+	key->skxi_id = cpu_to_le64(id);
+}
+
 /*
  * Find the next xattr and copy the key, xattr header, and as much of
  * the name and value into the callers buffer as we can.  Returns the
@@ -390,18 +409,24 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	struct inode *inode = dentry->d_inode;
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	struct super_block *sb = inode->i_sb;
+	const u64 ino = scoutfs_ino(inode);
 	struct scoutfs_xattr *xat = NULL;
+	struct scoutfs_lock *indx_lock = NULL;
 	struct scoutfs_lock *lck = NULL;
 	size_t name_len = strlen(name);
+	struct scoutfs_key indx_key;
 	struct scoutfs_key key;
 	struct prefix_tags tgs;
+	bool undo_indx = false;
 	LIST_HEAD(ind_locks);
 	LIST_HEAD(saved);
 	u8 found_parts;
 	unsigned int bytes;
 	u64 ind_seq;
-	u64 id;
+	u64 hash;
+	u64 id = 0;
 	int ret;
+	int err;
 
 	trace_scoutfs_xattr_set(sb, name_len, value, size, flags);
 
@@ -418,10 +443,10 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	if (unknown_prefix(name))
 		return -EOPNOTSUPP;
 
-	if (parse_tags(name, &tgs) != 0)
+	if (parse_tags(name, name_len, &tgs) != 0)
 		return -EINVAL;
 
-	if (tgs.hide && !capable(CAP_SYS_ADMIN))
+	if ((tgs.hide || tgs.indx) && !capable(CAP_SYS_ADMIN))
 		return -EPERM;
 
 	bytes = sizeof(struct scoutfs_xattr) + name_len + size;
@@ -472,13 +497,22 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 		memcpy(&xat->name[xat->name_len], value, size);
 	}
 
+	if (tgs.indx && !(found_parts && value)) {
+		hash = scoutfs_hash64(name, name_len);
+		ret = scoutfs_lock_xattr_index(sb, SCOUTFS_LOCK_WRITE_ONLY, 0,
+					       hash, &indx_lock);
+		if (ret < 0)
+			goto unlock;
+	}
+
 retry:
 	ret = scoutfs_inode_index_start(sb, &ind_seq) ?:
 	      scoutfs_inode_index_prepare(sb, &ind_locks, inode, false) ?:
 	      scoutfs_inode_index_try_lock_hold(sb, &ind_locks, ind_seq,
 						SIC_XATTR_SET(found_parts,
 							      value != NULL,
-							      name_len, size));
+							      name_len, size,
+							      tgs.indx));
 	if (ret > 0)
 		goto retry;
 	if (ret)
@@ -488,6 +522,22 @@ retry:
 	if (ret < 0)
 		goto release;
 
+	if (tgs.indx && !(found_parts && value)) {
+		if (found_parts)
+			id = le64_to_cpu(key.skx_id);
+		hash = scoutfs_hash64(name, name_len);
+		scoutfs_xattr_index_key(&indx_key, hash, ino, id);
+		if (value)
+			ret = scoutfs_item_create_force(sb, &indx_key, NULL,
+							indx_lock);
+		else
+			ret = scoutfs_item_delete_force(sb, &indx_key,
+							indx_lock);
+		if (ret < 0)
+			goto release;
+		undo_indx = true;
+	}
+
 	ret = 0;
 	if (found_parts)
 		ret = delete_xattr_items(inode, le64_to_cpu(key.skx_name_hash),
@@ -508,10 +558,21 @@ retry:
 	ret = 0;
 
 release:
+	if (ret < 0 && undo_indx) {
+		if (value)
+			err = scoutfs_item_delete_force(sb, &indx_key,
+							indx_lock);
+		else
+			err = scoutfs_item_create_force(sb, &indx_key, NULL,
+							indx_lock);
+		BUG_ON(err);
+	}
+
 	scoutfs_release_trans(sb);
 	scoutfs_inode_index_unlock(sb, &ind_locks);
 unlock:
 	up_write(&si->xattr_rwsem);
+	scoutfs_unlock(sb, indx_lock, SCOUTFS_LOCK_WRITE_ONLY);
 	scoutfs_unlock(sb, lck, SCOUTFS_LOCK_WRITE);
 out:
 	kfree(xat);
@@ -577,7 +638,9 @@ ssize_t scoutfs_list_xattrs(struct inode *inode, char *buffer,
 			break;
 		}
 
-		if (hidden || parse_tags(xat->name, &tgs) != 0 || !tgs.hide) {
+		if (hidden ||
+		    parse_tags(xat->name, xat->name_len, &tgs) != 0 ||
+		    !tgs.hide) {
 
 			if (size) {
 				if ((total + xat->name_len + 1) > size) {
@@ -624,54 +687,86 @@ ssize_t scoutfs_listxattr(struct dentry *dentry, char *buffer, size_t size)
 /*
  * Delete all the xattr items associated with this inode.  The inode is
  * dead so we don't need the xattr rwsem.
- *
- * XXX This isn't great because it reads in all the items so that it can
- * create deletion items for each.  It would be better to have the
- * caller create range deletion items for all the items covered by the
- * inode.  That wouldn't require reading at all.
  */
 int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 		       struct scoutfs_lock *lock)
 {
+	struct scoutfs_lock *indx_lock = NULL;
+	struct scoutfs_xattr *xat = NULL;
+	struct scoutfs_key indx_key;
 	struct scoutfs_key last;
 	struct scoutfs_key key;
-	unsigned int items = 16;
-	bool holding = false;
+	struct prefix_tags tgs;
+	bool release = false;
+	unsigned int bytes;
+	struct kvec val;
+	u64 hash;
 	int ret;
 
+	/* need a buffer large enough for all possible names */
+	bytes = sizeof(struct scoutfs_xattr) + SCOUTFS_XATTR_MAX_NAME_LEN;
+	xat = kmalloc(bytes, GFP_NOFS);
+	if (!xat) {
+		ret = -ENOMEM;
+		goto out;
+	}
+
 	init_xattr_key(&key, ino, 0, 0);
 	init_xattr_key(&last, ino, U32_MAX, U64_MAX);
 
 	for (;;) {
-		ret = scoutfs_item_next(sb, &key, &last, NULL, lock);
+		kvec_init(&val, (void *)xat, bytes);
+		ret = scoutfs_item_next(sb, &key, &last, &val, lock);
 		if (ret < 0) {
 			if (ret == -ENOENT)
 				ret = 0;
 			break;
 		}
 
-		if (!holding) {
-			ret = scoutfs_hold_trans(sb, SIC_EXACT(items, 0));
-			if (ret)
+		if (key.skx_part != 0 ||
+		    parse_tags(xat->name, xat->name_len, &tgs) != 0)
+			memset(&tgs, 0, sizeof(tgs));
+
+		if (tgs.indx) {
+			hash = scoutfs_hash64(xat->name, xat->name_len);
+			scoutfs_xattr_index_key(&indx_key, hash, ino,
+						le64_to_cpu(key.skx_id));
+			ret = scoutfs_lock_xattr_index(sb,
+						      SCOUTFS_LOCK_WRITE_ONLY,
+						      0, hash, &indx_lock);
+			if (ret < 0)
 				break;
-			holding = true;
 		}
 
+		ret = scoutfs_hold_trans(sb, SIC_EXACT(2, 0));
+		if (ret < 0)
+			break;
+		release = true;
+
 		ret = scoutfs_item_delete(sb, &key, lock);
-		if (ret)
+		if (ret < 0)
 			break;
 
-		if (--items == 0) {
-			scoutfs_release_trans(sb);
-			holding = false;
-			items = 16;
+		if (tgs.indx) {
+		       ret = scoutfs_item_delete_force(sb, &indx_key,
+						       indx_lock);
+		       if (ret < 0)
+			       break;
 		}
 
+		scoutfs_release_trans(sb);
+		release = false;
+
+		scoutfs_unlock(sb, indx_lock, SCOUTFS_LOCK_WRITE_ONLY);
+		indx_lock = NULL;
+
 		/* don't need to inc, next won't see deleted item */
 	}
 
-	if (holding)
+	if (release)
 		scoutfs_release_trans(sb);
-
+	scoutfs_unlock(sb, indx_lock, SCOUTFS_LOCK_WRITE_ONLY);
+	kfree(xat);
+out:
 	return ret;
 }

kmod/src/xattr.h

@@ -14,4 +14,7 @@ ssize_t scoutfs_list_xattrs(struct inode *inode, char *buffer,
 int scoutfs_xattr_drop(struct super_block *sb, u64 ino,
 		       struct scoutfs_lock *lock);
 
+void scoutfs_xattr_index_key(struct scoutfs_key *key,
+			     u64 hash, u64 ino, u64 id);
+
 #endif