diff --git a/kmod/src/format.h b/kmod/src/format.h index 3c66f637..70c7d6b9 100644 --- a/kmod/src/format.h +++ b/kmod/src/format.h @@ -184,6 +184,11 @@ struct scoutfs_key { #define skxt_b _sk_second #define skxt_c _sk_third +/* xattr index */ +#define skxi_a _sk_first +#define skxi_b _sk_second +#define skxi_ino _sk_third + /* inode */ #define ski_ino _sk_first @@ -591,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 */ diff --git a/kmod/src/ioctl.c b/kmod/src/ioctl.c index 62e9a9d6..5e9ac6c7 100644 --- a/kmod/src/ioctl.c +++ b/kmod/src/ioctl.c @@ -1481,6 +1481,109 @@ 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)) + +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]; + + if (val_len != 0) + return -EIO; + + ent->a = le64_to_cpu(key->skxi_a); + ent->b = le64_to_cpu(key->skxi_b); + ent->ino = le64_to_cpu(key->skxi_ino); + + 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_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; + + 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.a, rxi.first.b, rxi.first.ino); + scoutfs_xattr_init_indx_key(&last, rxi.last.a, rxi.last.b, rxi.last.ino); + scoutfs_xattr_indx_get_range(&start, &end); + + if (scoutfs_key_compare(&first, &last) > 0) { + ret = -EINVAL; + goto out; + } + + 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; + + if (copy_to_user(&uents[copied], rib->ents, rib->nr * sizeof(rib->ents[0]))) { + ret = -EFAULT; + goto out; + } + + copied += rib->nr; + + ent = &rib->ents[rib->nr - 1]; + scoutfs_xattr_init_indx_key(&first, ent->a, ent->b, ent->ino); + 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) { @@ -1528,6 +1631,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; diff --git a/kmod/src/ioctl.h b/kmod/src/ioctl.h index def67b95..292efa03 100644 --- a/kmod/src/ioctl.h +++ b/kmod/src/ioctl.h @@ -747,4 +747,26 @@ 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 single scoutfs.hide.indx xattr per inode. The xattr sets the + * two index position values, with a being higher significance. + */ +struct scoutfs_ioctl_xattr_index_entry { + __u64 a; + __u64 b; + __u64 ino; +}; + +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 diff --git a/kmod/src/lock.c b/kmod/src/lock.c index 2398f3d9..a6ae9a6e 100644 --- a/kmod/src/lock.c +++ b/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) { diff --git a/kmod/src/lock.h b/kmod/src/lock.h index 1b49e534..07908d62 100644 --- a/kmod/src/lock.h +++ b/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, diff --git a/kmod/src/xattr.c b/kmod/src/xattr.c index fa418b8c..a7f88984 100644 --- a/kmod/src/xattr.c +++ b/kmod/src/xattr.c @@ -82,6 +82,7 @@ static void init_xattr_key(struct scoutfs_key *key, u64 ino, u32 name_hash, #define SCOUTFS_XATTR_PREFIX_LEN (sizeof(SCOUTFS_XATTR_PREFIX) - 1) #define HIDE_TAG "hide." +#define INDX_TAG "indx." #define SRCH_TAG "srch." #define TOTL_TAG "totl." #define TAG_LEN (sizeof(HIDE_TAG) - 1) @@ -103,6 +104,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; @@ -540,47 +544,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) { @@ -607,6 +621,47 @@ 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; +} + +void scoutfs_xattr_init_indx_key(struct scoutfs_key *key, u64 a, u64 b, u64 ino) +{ + scoutfs_key_set_zeros(key); + key->sk_zone = SCOUTFS_XATTR_INDX_ZONE; + key->skxi_a = cpu_to_le64(a); + key->skxi_b = cpu_to_le64(b); + key->skxi_ino = cpu_to_le64(ino); +} + +/* + * indx keys have a restricted name so that there can only be one xattr + * that places in inode at a given position. This lets us emit index + * items under CW cluster locks without reading to see if they exist or + * not. + */ +#define REQUIRED_INDEX_PREFIX "scoutfs.hide.indx" + +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 (!xattr_names_equal(name, ret, REQUIRED_INDEX_PREFIX, sizeof(REQUIRED_INDEX_PREFIX) - 1)) + return -EINVAL; + + scoutfs_xattr_init_indx_key(key, u64s[0], u64s[1], ino); + return 0; +} + /* * The confusing swiss army knife of creating, modifying, and deleting * xattrs. @@ -627,7 +682,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); @@ -635,10 +690,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; @@ -651,7 +707,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 */ @@ -664,10 +721,13 @@ 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 = parse_indx_key(&tag_key, name, name_len, ino)) != 0)) return ret; /* allocate enough to always read an existing xattr's totl */ @@ -718,6 +778,23 @@ 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) @@ -741,6 +818,16 @@ int scoutfs_xattr_set_locked(struct inode *inode, const char *name, size_t name_ le64_add_cpu(&tval.total, total); } + if (tgs->indx) { + 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); @@ -752,7 +839,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; @@ -777,6 +864,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); @@ -785,7 +879,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); } @@ -801,7 +895,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); @@ -816,8 +910,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; } @@ -836,7 +933,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); @@ -845,7 +942,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; } @@ -1055,14 +1152,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; @@ -1108,16 +1206,36 @@ 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) { + if (release) { + scoutfs_release_trans(sb); + release = false; + } + 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); @@ -1140,11 +1258,17 @@ 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) + goto out; + } + scoutfs_release_trans(sb); release = false; @@ -1153,7 +1277,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; diff --git a/kmod/src/xattr.h b/kmod/src/xattr.h index 1cb14566..c94c13e6 100644 --- a/kmod/src/xattr.h +++ b/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,7 @@ 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, u64 a, u64 b, u64 ino); + #endif