scoutfs: add find xattr ioctls

Add ioctls that return the inode numbers that probably contain the given
xattr name or value.  To support these we add items that index inodes by
the presence of xattr items whose names or values hash to a give hash
value.

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

kmod/src/format.h

@@ -99,9 +99,11 @@ struct scoutfs_key {
  */
 #define SCOUTFS_INODE_KEY		1
 #define SCOUTFS_XATTR_KEY		2
-#define SCOUTFS_DIRENT_KEY		3
-#define SCOUTFS_LINK_BACKREF_KEY	4
-#define SCOUTFS_BMAP_KEY		5
+#define SCOUTFS_XATTR_NAME_HASH_KEY	3
+#define SCOUTFS_XATTR_VAL_HASH_KEY	4
+#define SCOUTFS_DIRENT_KEY		5
+#define SCOUTFS_LINK_BACKREF_KEY	6
+#define SCOUTFS_BMAP_KEY		7
 
 #define SCOUTFS_MAX_ITEM_LEN 512
 
@@ -246,9 +248,8 @@ enum {
 	SCOUTFS_DT_WHT,
 };
 
-#define SCOUTFS_MAX_XATTR_NAME_LEN 255
-#define SCOUTFS_MAX_XATTR_VALUE_LEN 255
-#define SCOUTFS_XATTR_HASH_MASK 7ULL
+#define SCOUTFS_MAX_XATTR_LEN 255
+#define SCOUTFS_XATTR_NAME_HASH_MASK 7ULL
 
 struct scoutfs_xattr {
 	__u8 name_len;

kmod/src/ioctl.c

@@ -21,6 +21,7 @@
 #include "btree.h"
 #include "key.h"
 #include "dir.h"
+#include "name.h"
 #include "ioctl.h"
 
 /*
@@ -200,6 +201,101 @@ out:
 	return ret;
 }
 
+/*
+ * Find inodes that might contain a given xattr name or value.
+ *
+ * The inodes are filled in sorted order from the first to the last
+ * inode.  The number of found inodes is returned.  If an error is hit
+ * it can return the number of inodes found before the error.
+ *
+ * The search can be continued from the next inode after the last
+ * returned.
+ */
+static long scoutfs_ioc_find_xattr(struct file *file, unsigned long arg,
+				   bool find_name)
+{
+	struct super_block *sb = file_inode(file)->i_sb;
+	struct scoutfs_ioctl_find_xattr args;
+	DECLARE_SCOUTFS_BTREE_CURSOR(curs);
+	struct scoutfs_key first;
+	struct scoutfs_key last;
+	char __user *ustr;
+	u64 __user *uino;
+	u64 inos[32];
+	char *str;
+	int nr_inos = 0;
+	int copied = 0;
+	int ret;
+	u8 type;
+	u64 h;
+
+	if (copy_from_user(&args, (void __user *)arg, sizeof(args)))
+		return -EFAULT;
+
+	if (args.str_len > SCOUTFS_MAX_XATTR_LEN || args.ino_count > INT_MAX)
+		return -EINVAL;
+
+	if (args.ino_count == 0)
+		return 0;
+
+	ustr = (void __user *)(unsigned long)args.str_ptr;
+	uino = (void __user *)(unsigned long)args.ino_ptr;
+
+	str = kmalloc(args.str_len, GFP_KERNEL);
+	if (!str)
+		return -ENOMEM;
+
+	if (copy_from_user(str, ustr, args.str_len)) {
+		ret = -EFAULT;
+		goto out;
+	}
+
+	h = scoutfs_name_hash(str, args.str_len);
+
+	if (find_name) {
+		h &= ~SCOUTFS_XATTR_NAME_HASH_MASK;
+		type = SCOUTFS_XATTR_NAME_HASH_KEY;
+	} else {
+		type = SCOUTFS_XATTR_VAL_HASH_KEY;
+	}
+
+	scoutfs_set_key(&first, h, type, args.first_ino);
+	scoutfs_set_key(&last, h, type, args.last_ino);
+
+	while (copied < args.ino_count) {
+
+		while ((ret = scoutfs_btree_next(sb, &first, &last,
+						 &curs)) > 0) {
+			inos[nr_inos++] = scoutfs_key_offset(curs.key);
+
+			first = *curs.key;
+			scoutfs_inc_key(&first);
+
+			if (nr_inos == ARRAY_SIZE(inos) ||
+			    (nr_inos + copied) == args.ino_count) {
+				scoutfs_btree_release(&curs);
+				ret = 0;
+				break;
+			}
+		}
+		if (ret < 0 || nr_inos == 0)
+			break;
+
+		if (copy_to_user(uino, inos, nr_inos * sizeof(u64))) {
+			ret = -EFAULT;
+			break;
+		}
+
+		uino += nr_inos;
+		copied += nr_inos;
+		nr_inos = 0;
+	}
+
+out:
+	kfree(str);
+	return copied ?: ret;
+}
+
 long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 {
 	switch (cmd) {
@@ -207,6 +303,10 @@ long scoutfs_ioctl(struct file *file, unsigned int cmd, unsigned long arg)
 		return scoutfs_ioc_inodes_since(file, arg);
 	case SCOUTFS_IOC_INODE_PATHS:
 		return scoutfs_ioc_inode_paths(file, arg);
+	case SCOUTFS_IOC_FIND_XATTR_NAME:
+		return scoutfs_ioc_find_xattr(file, arg, true);
+	case SCOUTFS_IOC_FIND_XATTR_VAL:
+		return scoutfs_ioc_find_xattr(file, arg, false);
 	}
 
 	return -ENOTTY;

kmod/src/ioctl.h

@@ -39,4 +39,19 @@ struct scoutfs_ioctl_inode_paths {
 #define SCOUTFS_IOC_INODE_PATHS _IOW(SCOUTFS_IOCTL_MAGIC, 2, \
 				      struct scoutfs_ioctl_inode_paths)
 
+/* XXX might as well include a seq?  0 for current behaviour? */
+struct scoutfs_ioctl_find_xattr {
+	__u64 first_ino;
+	__u64 last_ino;
+	__u64 str_ptr;
+	__u32 str_len;
+	__u64 ino_ptr;
+	__u32 ino_count;
+} __packed;
+
+#define SCOUTFS_IOC_FIND_XATTR_NAME _IOW(SCOUTFS_IOCTL_MAGIC, 3, \
+				      struct scoutfs_ioctl_find_xattr)
+#define SCOUTFS_IOC_FIND_XATTR_VAL _IOW(SCOUTFS_IOCTL_MAGIC, 4, \
+				      struct scoutfs_ioctl_find_xattr)
+
 #endif

kmod/src/xattr.c

@@ -40,6 +40,14 @@
  * multiple versions of an xattr in the btree.  So we add an inode rw
  * semaphore around xattr operations.
  *
+ * We support ioctls which find inodes that may contain xattrs with
+ * either a given name or value.  A name hash item is created for a
+ * given hash value with no collision bits as long as there are any
+ * names at that hash value.  A value hash item is created but it
+ * contains a refcount in its value to track the number of values with
+ * that hash value because we can't use the xattr keys to determine if
+ * there are matching values or not.
+ *
  * XXX
  *  - add acl support and call generic xattr->handlers for SYSTEM
  *  - remove all xattrs on unlink
@@ -56,92 +64,221 @@ static unsigned int xat_bytes(unsigned int name_len, unsigned int value_len)
 	return offsetof(struct scoutfs_xattr, name[name_len + value_len]);
 }
 
+static void set_xattr_keys(struct inode *inode, struct scoutfs_key *first,
+			   struct scoutfs_key *last, const char *name,
+			   unsigned int name_len)
+{
+	u64 h = scoutfs_name_hash(name, name_len) &
+		~SCOUTFS_XATTR_NAME_HASH_MASK;
+
+	scoutfs_set_key(first, scoutfs_ino(inode), SCOUTFS_XATTR_KEY, h);
+	scoutfs_set_key(last, scoutfs_ino(inode), SCOUTFS_XATTR_KEY,
+			h | SCOUTFS_XATTR_NAME_HASH_MASK);
+}
+
+static void set_name_val_keys(struct scoutfs_key *name_key,
+			      struct scoutfs_key *val_key,
+			      struct scoutfs_key *key, u64 val_hash)
+{
+	u64 h = scoutfs_key_offset(key) & ~SCOUTFS_XATTR_NAME_HASH_MASK;
+
+	scoutfs_set_key(name_key, h, SCOUTFS_XATTR_NAME_HASH_KEY,
+			scoutfs_key_inode(key));
+
+	scoutfs_set_key(val_key, val_hash, SCOUTFS_XATTR_VAL_HASH_KEY,
+			scoutfs_key_inode(key));
+}
+
 /*
- * The caller provides an initialized cursor.
+ * Before insertion we perform a pretty through search of the xattr
+ * items whose offset collides with the name to be inserted.
  *
- * If we return > 0 then the cursor points to an xattr with the given
- * name and the caller must clean up the cursor.
- *
- * Returns 0 when no matching xattr is found or -errno on error.
+ * We try to find the item with the matching item so it can be removed.
+ * We notice if there are other colliding names so that the caller can
+ * correctly maintain the name hash items.  We calculate the value hash
+ * of the existing item so that the caller can maintain the value hash
+ * items.  And we notice if there are any free colliding items that are
+ * available for new item insertion.
  */
-static int lookup_xattr(struct inode *inode, const char *name,
-			unsigned int name_len,
-			struct scoutfs_btree_cursor *curs)
+struct xattr_search_results {
+	bool found;
+	bool other_coll;
+	struct scoutfs_key key;
+	u64 val_hash;
+	bool found_hole;
+	struct scoutfs_key hole_key;
+};
+
+static int search_xattr_items(struct inode *inode, const char *name,
+			      unsigned int name_len,
+			      struct xattr_search_results *res)
 {
 	struct super_block *sb = inode->i_sb;
+	DECLARE_SCOUTFS_BTREE_CURSOR(curs);
 	struct scoutfs_key first;
 	struct scoutfs_key last;
 	struct scoutfs_xattr *xat;
 	int ret;
-	u64 h;
 
-	if (name_len > SCOUTFS_MAX_XATTR_NAME_LEN)
-		return -EINVAL;
+	set_xattr_keys(inode, &first, &last, name, name_len);
 
-	/* XXX could be a lookup helper? */
-	h = scoutfs_name_hash(name, name_len) & ~SCOUTFS_XATTR_HASH_MASK;
+	res->found = false;
+	res->other_coll = false;
+	res->found_hole = false;
+	res->hole_key = first;
 
-	scoutfs_set_key(&first, scoutfs_ino(inode), SCOUTFS_XATTR_KEY, h);
-	scoutfs_set_key(&last, scoutfs_ino(inode), SCOUTFS_XATTR_KEY,
-			h | SCOUTFS_XATTR_HASH_MASK);
+	while ((ret = scoutfs_btree_next(sb, &first, &last, &curs)) > 0) {
+		xat = curs.val;
 
-	while ((ret = scoutfs_btree_next(sb, &first, &last, curs)) > 0) {
-		xat = curs->val;
+		/* found a hole when we skip past next expected key */
+		if (!res->found_hole &&
+		    scoutfs_key_cmp(&res->hole_key, curs.key) < 0)
+			res->found_hole = true;
 
-		if (scoutfs_names_equal(name, name_len, xat->name,
-					xat->name_len))
+		/* keep searching for a hole past this cursor key */
+		if (!res->found_hole) {
+			res->hole_key = *curs.key;
+			scoutfs_inc_key(&res->hole_key);
+		}
+
+		/* only compare the names until we find our given name */
+		if (!res->found &&
+		    scoutfs_names_equal(name, name_len, xat->name,
+				        xat->name_len)) {
+			res->found = true;
+			res->key = *curs.key;
+			res->val_hash = scoutfs_name_hash(xat_value(xat),
+							  xat->value_len);
+		} else {
+			res->other_coll = true;
+		}
+
+		/* finished once we have all the caller needs */
+		if (res->found && res->other_coll && res->found_hole) {
+			ret = 0;
+			scoutfs_btree_release(&curs);
 			break;
+		}
 	}
 
-	if (ret <= 0)
-		scoutfs_btree_release(curs);
-
 	return ret;
 }
 
 /*
- * Insert a new xattr and set the caller's key to the key that we used.
- * The caller is responsible for managing transactions and locking.
+ * Inset a new xattr item, updating the name and value hash items as
+ * needed.  The caller is responsible for managing transactions and
+ * locking.  If this returns an error then no changes will have been
+ * made.
  */
 static int insert_xattr(struct inode *inode, const char *name,
 			unsigned int name_len, const void *value, size_t size,
-			struct scoutfs_key *key)
+			struct scoutfs_key *key, bool other_coll,
+			u64 val_hash)
 {
 	struct super_block *sb = inode->i_sb;
 	DECLARE_SCOUTFS_BTREE_CURSOR(curs);
+	bool inserted_name_hash_item = false;
+	__le64 * __packed refcount;
+	struct scoutfs_key name_key;
+	struct scoutfs_key val_key;
 	struct scoutfs_xattr *xat;
-	struct scoutfs_key first;
-	struct scoutfs_key last;
 	int ret;
-	u64 h;
 
-	if (name_len > SCOUTFS_MAX_XATTR_NAME_LEN ||
-	    size > SCOUTFS_MAX_XATTR_NAME_LEN)
-		return -EINVAL;
+	set_name_val_keys(&name_key, &val_key, key, val_hash);
 
-	/* XXX could be a lookup helper? */
-	h = scoutfs_name_hash(name, name_len) & ~SCOUTFS_XATTR_HASH_MASK;
-
-	scoutfs_set_key(&first, scoutfs_ino(inode), SCOUTFS_XATTR_KEY, h);
-	scoutfs_set_key(&last, scoutfs_ino(inode), SCOUTFS_XATTR_KEY,
-			h | SCOUTFS_XATTR_HASH_MASK);
-
-	/* find the first unoccupied key offset after the hashed name */
-	ret = scoutfs_btree_hole(sb, &first, &last, key);
+	ret = scoutfs_btree_insert(sb, key, xat_bytes(name_len, size), &curs);
 	if (ret)
 		return ret;
 
-	ret = scoutfs_btree_insert(sb, key, xat_bytes(name_len, size), &curs);
-	if (!ret) {
-		xat = curs.val;
-		xat->name_len = name_len;
-		xat->value_len = size;
-		memcpy(xat->name, name, name_len);
-		memcpy(xat_value(xat), value, size);
+	xat = curs.val;
+	xat->name_len = name_len;
+	xat->value_len = size;
+	memcpy(xat->name, name, name_len);
+	memcpy(xat_value(xat), value, size);
 
+	scoutfs_btree_release(&curs);
+
+	/* insert the name hash item for find_xattr if we're first */
+	if (!other_coll) {
+		ret = scoutfs_btree_insert(sb, &name_key, 0, &curs);
+		/* XXX eexist would be corruption */
+		if (ret)
+			goto out;
+		scoutfs_btree_release(&curs);
+		inserted_name_hash_item = true;
+	}
+
+	/* increment the val hash item for find_xattr, inserting if first */
+	ret = scoutfs_btree_update(sb, &val_key, &curs);
+	if (ret == -ENOENT) {
+		ret = scoutfs_btree_insert(sb, &val_key, sizeof(*refcount),
+					   &curs);
+		if (ret == 0) {
+			/* XXX test sane item size */
+			refcount = curs.val;
+			*refcount = 0;
+		}
+	}
+	if (ret == 0) {
+		refcount = curs.val;
+		le64_add_cpu(refcount, 1);
 		scoutfs_btree_release(&curs);
 	}
 
+out:
+	if (ret) {
+		scoutfs_btree_delete(sb, key);
+		if (inserted_name_hash_item)
+			scoutfs_btree_delete(sb, &name_key);
+	}
+	return ret;
+}
+
+/*
+ * Remove an xattr.  Remove the name hash item if there are no more xattrs
+ * in the inode that hash to the name's hash value.  Remove the value hash
+ * item if there are no more xattr values in the inode with this value
+ * hash.
+ */
+static int delete_xattr(struct super_block *sb, struct scoutfs_key *key,
+			bool other_coll, u64 val_hash)
+{
+	DECLARE_SCOUTFS_BTREE_CURSOR(curs);
+	struct scoutfs_key name_key;
+	struct scoutfs_key val_key;
+	__le64 * __packed refcount;
+	bool del_val = false;
+	int ret;
+
+	set_name_val_keys(&name_key, &val_key, key, val_hash);
+
+	if (!other_coll) {
+		ret = scoutfs_btree_dirty(sb, &name_key);
+		if (ret)
+			goto out;
+	}
+	ret = scoutfs_btree_dirty(sb, &val_key);
+	if (ret)
+		goto out;
+
+	ret = scoutfs_btree_delete(sb, key);
+	if (ret)
+		goto out;
+
+	if (!other_coll)
+		scoutfs_btree_delete(sb, &name_key);
+
+	scoutfs_btree_update(sb, &val_key, &curs);
+	refcount = curs.val;
+	le64_add_cpu(refcount, -1ULL);
+	if (*refcount == 0)
+		del_val = true;
+	scoutfs_btree_release(&curs);
+
+	if (del_val)
+		scoutfs_btree_delete(sb, &val_key);
+	ret = 0;
+out:
 	return ret;
 }
 
@@ -158,23 +295,32 @@ ssize_t scoutfs_getxattr(struct dentry *dentry, const char *name, void *buffer,
 			 size_t size)
 {
 	struct inode *inode = dentry->d_inode;
+	struct super_block *sb = inode->i_sb;
 	struct scoutfs_inode_info *si = SCOUTFS_I(inode);
 	DECLARE_SCOUTFS_BTREE_CURSOR(curs);
 	size_t name_len = strlen(name);
 	struct scoutfs_xattr *xat;
+	struct scoutfs_key first;
+	struct scoutfs_key last;
 	int ret;
 
 	if (unknown_prefix(name))
 		return -EOPNOTSUPP;
 
+	set_xattr_keys(inode, &first, &last, name, name_len);
+
 	down_read(&si->xattr_rwsem);
 
-	ret = lookup_xattr(inode, name, name_len, &curs);
-	if (ret == 0) {
-		ret = -ENODATA;
-	} else if (ret > 0) {
+	ret = -ENODATA;
+	while ((ret = scoutfs_btree_next(sb, &first, &last, &curs)) > 0) {
 		xat = curs.val;
 
+		if (!scoutfs_names_equal(name, name_len, xat->name,
+					 xat->name_len)) {
+			ret = -ENODATA;
+			continue;
+		}
+
 		ret = xat->value_len;
 		if (buffer) {
 			if (ret <= size)
@@ -183,6 +329,7 @@ ssize_t scoutfs_getxattr(struct dentry *dentry, const char *name, void *buffer,
 				ret = -ERANGE;
 		}
 		scoutfs_btree_release(&curs);
+		break;
 	}
 
 	up_read(&si->xattr_rwsem);
@@ -191,9 +338,16 @@ ssize_t scoutfs_getxattr(struct dentry *dentry, const char *name, void *buffer,
 }
 
 /*
- * Set the xattr with the given name to the given value.  The value can
- * have a size of 0.  A null value pointer indicates that we should
- * delete the xattr.
+ * The confusing swiss army knife of creating, modifying, and deleting
+ * xattrs.
+ *
+ * If the value pointer is non-null then we always create a new item.  The
+ * value can have a size of 0.  We create a new item before possibly
+ * deleting an old item.
+ *
+ * We always delete the old xattr item.  If we have a null value then we're
+ * deleting the xattr.  If there's a value then we're effectively updating
+ * the xattr by deleting old and creating new.
  */
 static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 			     const void *value, size_t size, int flags)
@@ -202,13 +356,15 @@ 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;
-	DECLARE_SCOUTFS_BTREE_CURSOR(curs);
+	struct xattr_search_results old = {0,};
 	size_t name_len = strlen(name);
-	struct scoutfs_key old_key;
-	struct scoutfs_key new_key;
-	bool old;
+	u64 new_val_hash = 0;
 	int ret;
 
+	if (name_len > SCOUTFS_MAX_XATTR_LEN ||
+	    (value && size > SCOUTFS_MAX_XATTR_LEN))
+		return -EINVAL;
+
 	if (unknown_prefix(name))
 		return -EOPNOTSUPP;
 
@@ -220,39 +376,48 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 	if (ret)
 		goto out;
 
+	/* might as well do this outside locking */
+	if (value)
+		new_val_hash = scoutfs_name_hash(value, size);
+
 	down_write(&si->xattr_rwsem);
 
-	ret = lookup_xattr(inode, name, name_len, &curs);
-	if (ret > 0) {
-		old = true;
-		old_key = *curs.key;
-		scoutfs_btree_release(&curs);
-	} else if (ret == 0) {
-		old = false;
-	} else {
+	/*
+	 * The presence of other colliding names is a little tricky.
+	 * Searching will set it if there are other non-matching names.
+	 * It will be false if we only found the old matching name. That
+	 * old match is also considered a collision for later insertion.
+	 * Then *that* insertion is considered a collision for deletion
+	 * of the existing old matching name.
+	 */
+	ret = search_xattr_items(inode, name, name_len, &old);
+	if (ret)
 		goto out;
-	}
 
-	if (old && (flags & XATTR_CREATE)) {
+	if (old.found && (flags & XATTR_CREATE)) {
 		ret = -EEXIST;
 		goto out;
 	}
-	if (!old && (flags & XATTR_REPLACE)) {
+	if (!old.found && (flags & XATTR_REPLACE)) {
 		ret = -ENODATA;
 		goto out;
 	}
 
 	if (value) {
 		ret = insert_xattr(inode, name, name_len, value, size,
-				   &new_key);
+				   &old.hole_key, old.other_coll || old.found,
+				   new_val_hash);
 		if (ret)
 			goto out;
 	}
 
-	if (old) {
-		ret = scoutfs_btree_delete(sb, &old_key);
+	if (old.found) {
+		ret = delete_xattr(sb, &old.key, old.other_coll || value,
+				   old.val_hash);
 		if (ret) {
-			scoutfs_btree_delete(sb, &new_key);
+			if (value)
+				delete_xattr(sb, &old.hole_key, true,
+					     new_val_hash);
 			goto out;
 		}
 	}
@@ -264,6 +429,7 @@ static int scoutfs_xattr_set(struct dentry *dentry, const char *name,
 out:
 	up_write(&si->xattr_rwsem);
 	scoutfs_release_trans(sb);
+
 	return ret;
 }