From f9e282048f60a191fdcaf0dcee9474d038c12b7b Mon Sep 17 00:00:00 2001 From: Zach Brown Date: Tue, 27 Feb 2018 15:34:25 -0800 Subject: [PATCH] scoutfs: revalidate dentries by checking items Initially we had d_revalidate always return that the dentry was invalid. This avoids dentry cache consistency problems across the cluster by always performing lookups. That's slow by itself, but it turns out that the dentry invalidation that happens on revalidation failure is very expensive if you have lots of dentries. So we switched to forcefully dropping dirents as we revoked their lock. That avoided the cost of revalidation failure but it adds the problem that dentries are unhashed when their locks are dropped. This causes paths like getcwd() to return errors when they see unhashed dentries instead of trying to revalidate them. This implements a d_revalidate which actually does work to determine if the dentry is still valid. When we populate dentries under a lock we add them to a list on the lock. As we drop the lock we remove them from the list. But the dentry is not modified. This lets paths like getcwd() still work. Then we implement revalidation that does the actual item lookups if the dentry's lock has been dropped. This lets revalidation return success and avoid the terrible invalidation costs from returning failure and then calling lookup to populate a new dentry. This brings us more in line with the revalidation behaviour of other systems that maintain multi-node dcache consistency. Signed-off-by: Zach Brown --- kmod/src/counters.h | 7 ++ kmod/src/dir.c | 162 +++++++++++++++++++++++++++++++++++++------- kmod/src/lock.c | 111 ++++++++++++++++++++++-------- kmod/src/lock.h | 18 +++++ 4 files changed, 245 insertions(+), 53 deletions(-) diff --git a/kmod/src/counters.h b/kmod/src/counters.h index 57cc8c6c..61397277 100644 --- a/kmod/src/counters.h +++ b/kmod/src/counters.h @@ -29,6 +29,13 @@ EXPAND_COUNTER(data_write_begin) \ EXPAND_COUNTER(data_write_end) \ EXPAND_COUNTER(data_writepage) \ + EXPAND_COUNTER(dentry_revalidate_error) \ + EXPAND_COUNTER(dentry_revalidate_invalid) \ + EXPAND_COUNTER(dentry_revalidate_locked) \ + EXPAND_COUNTER(dentry_revalidate_orphan) \ + EXPAND_COUNTER(dentry_revalidate_rcu) \ + EXPAND_COUNTER(dentry_revalidate_root) \ + EXPAND_COUNTER(dentry_revalidate_valid) \ EXPAND_COUNTER(item_alloc) \ EXPAND_COUNTER(item_create) \ EXPAND_COUNTER(item_delete) \ diff --git a/kmod/src/dir.c b/kmod/src/dir.c index 85b9143e..7cbc9bb6 100644 --- a/kmod/src/dir.c +++ b/kmod/src/dir.c @@ -31,6 +31,7 @@ #include "kvec.h" #include "item.h" #include "lock.h" +#include "counters.h" #include "scoutfs_trace.h" /* @@ -101,28 +102,35 @@ static unsigned int dentry_type(unsigned int type) } /* - * Each dentry stores the values that are needed to build the keys of - * the items that are removed on unlink so that we don't to search - * through items on unlink. + * @readdir_pos lets us remove items on final unlink without having to + * look them up. + * + * @lock_cov tells revalidation that the dentry is still locked and valid. */ struct dentry_info { u64 readdir_pos; + struct scoutfs_lock_coverage lock_cov; }; static struct kmem_cache *dentry_info_cache; static void scoutfs_d_release(struct dentry *dentry) { + struct super_block *sb = dentry->d_sb; struct dentry_info *di = dentry->d_fsdata; if (di) { + scoutfs_lock_del_coverage(sb, &di->lock_cov); kmem_cache_free(dentry_info_cache, di); dentry->d_fsdata = NULL; } } +static int scoutfs_d_revalidate(struct dentry *dentry, unsigned int flags); + static const struct dentry_operations scoutfs_dentry_ops = { .d_release = scoutfs_d_release, + .d_revalidate = scoutfs_d_revalidate, }; static int alloc_dentry_info(struct dentry *dentry) @@ -137,6 +145,8 @@ static int alloc_dentry_info(struct dentry *dentry) if (!di) return -ENOMEM; + scoutfs_lock_init_coverage(&di->lock_cov); + spin_lock(&dentry->d_lock); if (!dentry->d_fsdata) { dentry->d_fsdata = di; @@ -150,7 +160,8 @@ static int alloc_dentry_info(struct dentry *dentry) return 0; } -static void update_dentry_info(struct dentry *dentry, u64 pos) +static void update_dentry_info(struct super_block *sb, struct dentry *dentry, + u64 pos, struct scoutfs_lock *lock) { struct dentry_info *di = dentry->d_fsdata; @@ -158,6 +169,7 @@ static void update_dentry_info(struct dentry *dentry, u64 pos) return; di->readdir_pos = pos; + scoutfs_lock_add_coverage(sb, lock, &di->lock_cov); } static u64 dentry_info_pos(struct dentry *dentry) @@ -225,6 +237,116 @@ static struct scoutfs_key_buf *alloc_link_backref_key(struct super_block *sb, return key; } +/* + * Looks for the dirent item and fills the caller's dirent if it finds + * it. Returns item lookup errors including -ENOENT if it's not found. + */ +static int lookup_dirent(struct super_block *sb, struct inode *dir, + const char *name, unsigned name_len, + struct scoutfs_dirent *dent, + struct scoutfs_lock *lock) +{ + struct scoutfs_key_buf *key = NULL; + SCOUTFS_DECLARE_KVEC(val); + int ret; + + key = alloc_dirent_key(sb, scoutfs_ino(dir), name, name_len); + if (!key) { + ret = -ENOMEM; + goto out; + } + + scoutfs_kvec_init(val, dent, sizeof(struct scoutfs_dirent)); + + ret = scoutfs_item_lookup_exact(sb, key, val, + sizeof(struct scoutfs_dirent), lock); +out: + scoutfs_key_free(sb, key); + return ret; +} + +static int scoutfs_d_revalidate(struct dentry *dentry, unsigned int flags) +{ + struct super_block *sb = dentry->d_sb; + struct dentry_info *di = dentry->d_fsdata; + struct scoutfs_lock *lock = NULL; + struct scoutfs_dirent dent; + struct dentry *parent = NULL; + struct inode *dir; + u64 dentry_ino; + int ret; + + /* don't think this happens but we can find out */ + if (IS_ROOT(dentry)) { + scoutfs_inc_counter(sb, dentry_revalidate_root); + if (!dentry->d_inode || + (scoutfs_ino(dentry->d_inode) != SCOUTFS_ROOT_INO)) { + ret = -EIO; + } else { + ret = 1; + } + goto out; + } + + /* XXX what are the rules for _RCU? */ + if (flags & LOOKUP_RCU) { + scoutfs_inc_counter(sb, dentry_revalidate_rcu); + ret = -ECHILD; + goto out; + } + + if (WARN_ON_ONCE(di == NULL)) { + ret = 0; + goto out; + } + + if (scoutfs_lock_is_covered(sb, &di->lock_cov)) { + scoutfs_inc_counter(sb, dentry_revalidate_locked); + ret = 1; + goto out; + } + + parent = dget_parent(dentry); + if (!parent || !parent->d_inode) { + scoutfs_inc_counter(sb, dentry_revalidate_orphan); + ret = 0; + goto out; + } + dir = parent->d_inode; + + ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, 0, dir, &lock); + if (ret) + goto out; + + ret = lookup_dirent(sb, dir, dentry->d_name.name, dentry->d_name.len, + &dent, lock); + if (ret == -ENOENT) + dent.ino = 0; + else if (ret < 0) + goto out; + + dentry_ino = dentry->d_inode ? scoutfs_ino(dentry->d_inode) : 0; + + if ((dentry_ino == le64_to_cpu(dent.ino))) { + update_dentry_info(sb, dentry, le64_to_cpu(dent.readdir_pos), + lock); + scoutfs_inc_counter(sb, dentry_revalidate_valid); + ret = 1; + } else { + scoutfs_inc_counter(sb, dentry_revalidate_invalid); + ret = 0; + } + +out: + dput(parent); + scoutfs_unlock(sb, lock, DLM_LOCK_PR); + + if (ret < 0 && ret != -ECHILD) + scoutfs_inc_counter(sb, dentry_revalidate_error); + + return ret; +} + /* * Because of rename, locks are ordered by inode number. To hold the * dir lock while calling iget, we might have to already hold a lesser @@ -240,10 +362,8 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry, unsigned int flags) { struct super_block *sb = dir->i_sb; - struct scoutfs_key_buf *key = NULL; - struct scoutfs_dirent dent; struct scoutfs_lock *dir_lock = NULL; - SCOUTFS_DECLARE_KVEC(val); + struct scoutfs_dirent dent; struct inode *inode; u64 ino = 0; int ret; @@ -257,28 +377,22 @@ static struct dentry *scoutfs_lookup(struct inode *dir, struct dentry *dentry, if (ret) goto out; - key = alloc_dirent_key(sb, scoutfs_ino(dir), - dentry->d_name.name, dentry->d_name.len); - if (!key) { - ret = -ENOMEM; - goto out; - } - ret = scoutfs_lock_inode(sb, DLM_LOCK_PR, 0, dir, &dir_lock); if (ret) goto out; - scoutfs_kvec_init(val, &dent, sizeof(dent)); - - ret = scoutfs_item_lookup_exact(sb, key, val, sizeof(dent), dir_lock); - scoutfs_unlock(sb, dir_lock, DLM_LOCK_PR); + ret = lookup_dirent(sb, dir, dentry->d_name.name, dentry->d_name.len, + &dent, dir_lock); if (ret == -ENOENT) { ino = 0; ret = 0; } else if (ret == 0) { ino = le64_to_cpu(dent.ino); - update_dentry_info(dentry, le64_to_cpu(dent.readdir_pos)); + update_dentry_info(sb, dentry, le64_to_cpu(dent.readdir_pos), + dir_lock); } + scoutfs_unlock(sb, dir_lock, DLM_LOCK_PR); + out: if (ret < 0) inode = ERR_PTR(ret); @@ -287,8 +401,6 @@ out: else inode = scoutfs_iget(sb, ino); - scoutfs_key_free(sb, key); - return d_splice_alias(inode, dentry); } @@ -625,7 +737,7 @@ static int scoutfs_mknod(struct inode *dir, struct dentry *dentry, umode_t mode, if (ret) goto out; - update_dentry_info(dentry, pos); + update_dentry_info(sb, dentry, pos, dir_lock); i_size_write(dir, i_size_read(dir) + dentry->d_name.len); dir->i_mtime = dir->i_ctime = CURRENT_TIME; @@ -720,7 +832,7 @@ retry: inode->i_mode, dir_lock, inode_lock); if (ret) goto out; - update_dentry_info(dentry, pos); + update_dentry_info(sb, dentry, pos, dir_lock); i_size_write(dir, dir_size); dir->i_mtime = dir->i_ctime = CURRENT_TIME; @@ -1021,7 +1133,7 @@ static int scoutfs_symlink(struct inode *dir, struct dentry *dentry, if (ret) goto out; - update_dentry_info(dentry, pos); + update_dentry_info(sb, dentry, pos, dir_lock); i_size_write(dir, i_size_read(dir) + dentry->d_name.len); dir->i_mtime = dir->i_ctime = CURRENT_TIME; @@ -1507,7 +1619,7 @@ retry: /* won't fail from here on out, update all the vfs structs */ /* the caller will use d_move to move the old_dentry into place */ - update_dentry_info(old_dentry, new_pos); + update_dentry_info(sb, old_dentry, new_pos, new_dir_lock); i_size_write(old_dir, i_size_read(old_dir) - old_dentry->d_name.len); if (!new_inode) diff --git a/kmod/src/lock.c b/kmod/src/lock.c index 3d6e868e..e3178bc8 100644 --- a/kmod/src/lock.c +++ b/kmod/src/lock.c @@ -92,41 +92,17 @@ static void scoutfs_lock_grace_work(struct work_struct *work); /* * invalidate cached data associated with an inode whose lock is going * away. - * - * Our inode granular locks mean that we have to invalidate all the - * child dentries of a dir so that they can't satisfy lookup after we - * re-acquire the lock. We're invalidating the lock so there can't be - * active users that could modify the entries in the dcache (lookup, - * create, rename, unlink). We have to make it through all the child - * entries and remove them from the hash so that lookup can't find them. */ static void invalidate_inode(struct super_block *sb, u64 ino) { struct inode *inode; - struct dentry *parent; - struct dentry *child; inode = scoutfs_ilookup(sb, ino); - if (!inode) - return; - - if (S_ISREG(inode->i_mode)) - truncate_inode_pages(inode->i_mapping, 0); - - if (S_ISDIR(inode->i_mode) && (parent = d_find_alias(inode))) { - - spin_lock(&parent->d_lock); - list_for_each_entry(child, &parent->d_subdirs, d_u.d_child){ - spin_lock_nested(&child->d_lock, DENTRY_D_LOCK_NESTED); - __d_drop(child); - spin_unlock(&child->d_lock); - } - spin_unlock(&parent->d_lock); - - dput(parent); + if (inode) { + if (S_ISREG(inode->i_mode)) + truncate_inode_pages(inode->i_mapping, 0); + iput(inode); } - - iput(inode); } /* @@ -138,6 +114,8 @@ static int lock_invalidate(struct super_block *sb, struct scoutfs_lock *lock, { struct scoutfs_key_buf *start = lock->start; struct scoutfs_key_buf *end = lock->end; + struct scoutfs_lock_coverage *cov; + struct scoutfs_lock_coverage *tmp; u64 ino, last; int ret; @@ -156,6 +134,21 @@ static int lock_invalidate(struct super_block *sb, struct scoutfs_lock *lock, if (prev == DLM_LOCK_CW || (prev == DLM_LOCK_PR && mode != DLM_LOCK_EX) || (prev == DLM_LOCK_EX && mode != DLM_LOCK_PR)) { + +retry: + spin_lock(&lock->cov_list_lock); + list_for_each_entry_safe(cov, tmp, &lock->cov_list, head) { + if (!spin_trylock(&cov->cov_lock)) { + spin_unlock(&lock->cov_list_lock); + cpu_relax(); + goto retry; + } + list_del_init(&cov->head); + cov->lock = NULL; + spin_unlock(&cov->cov_lock); + } + spin_unlock(&lock->cov_list_lock); + if (lock->name.zone == SCOUTFS_FS_ZONE) { ino = le64_to_cpu(lock->name.first); last = ino + SCOUTFS_LOCK_INODE_GROUP_NR - 1; @@ -237,6 +230,9 @@ static struct scoutfs_lock *lock_alloc(struct super_block *sb, RB_CLEAR_NODE(&lock->range_node); INIT_LIST_HEAD(&lock->lru_head); + spin_lock_init(&lock->cov_list_lock); + INIT_LIST_HEAD(&lock->cov_list); + if (start) { lock->start = scoutfs_key_dup(sb, start); lock->end = scoutfs_key_dup(sb, end); @@ -1136,6 +1132,65 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, int mode) spin_unlock(&linfo->lock); } +void scoutfs_lock_init_coverage(struct scoutfs_lock_coverage *cov) +{ + spin_lock_init(&cov->cov_lock); + cov->lock = NULL; + INIT_LIST_HEAD(&cov->head); +} + +/* + * Record that the given coverage struct is protected by the given lock. + * Once the lock is dropped the coverage list head will be removed and + * callers can use that to see that the cov isn't covered any more. The + * cov might be on another lock so we're careful to remove it. + */ +void scoutfs_lock_add_coverage(struct super_block *sb, + struct scoutfs_lock *lock, + struct scoutfs_lock_coverage *cov) +{ + spin_lock(&cov->cov_lock); + + if (cov->lock) { + spin_lock(&cov->lock->cov_list_lock); + list_del_init(&cov->head); + spin_unlock(&cov->lock->cov_list_lock); + cov->lock = NULL; + } + + cov->lock = lock; + spin_lock(&cov->lock->cov_list_lock); + list_add(&cov->head, &lock->cov_list); + spin_unlock(&cov->lock->cov_list_lock); + + spin_unlock(&cov->cov_lock); +} + +bool scoutfs_lock_is_covered(struct super_block *sb, + struct scoutfs_lock_coverage *cov) +{ + bool covered; + + spin_lock(&cov->cov_lock); + covered = !list_empty_careful(&cov->head); + spin_unlock(&cov->cov_lock); + + return covered; +} + +void scoutfs_lock_del_coverage(struct super_block *sb, + struct scoutfs_lock_coverage *cov) +{ + spin_lock(&cov->cov_lock); + if (cov->lock) { + spin_lock(&cov->lock->cov_list_lock); + list_del_init(&cov->head); + spin_unlock(&cov->lock->cov_list_lock); + cov->lock = NULL; + } + spin_unlock(&cov->cov_lock); +} + static int scoutfs_lock_shrink(struct shrinker *shrink, struct shrink_control *sc) { diff --git a/kmod/src/lock.h b/kmod/src/lock.h index 1cabe132..12e8c610 100644 --- a/kmod/src/lock.h +++ b/kmod/src/lock.h @@ -30,6 +30,9 @@ struct scoutfs_lock { struct delayed_work grace_work; bool grace_pending; + spinlock_t cov_list_lock; + struct list_head cov_list; + int error; int granted_mode; int bast_mode; @@ -39,6 +42,12 @@ struct scoutfs_lock { unsigned int users[SCOUTFS_LOCK_NR_MODES]; }; +struct scoutfs_lock_coverage { + spinlock_t cov_lock; + struct scoutfs_lock *lock; + struct list_head head; +}; + int scoutfs_lock_inode(struct super_block *sb, int mode, int flags, struct inode *inode, struct scoutfs_lock **ret_lock); int scoutfs_lock_ino(struct super_block *sb, int mode, int flags, u64 ino, @@ -63,6 +72,15 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, void scoutfs_unlock_flags(struct super_block *sb, struct scoutfs_lock *lock, int level, int flags); +void scoutfs_lock_init_coverage(struct scoutfs_lock_coverage *cov); +void scoutfs_lock_add_coverage(struct super_block *sb, + struct scoutfs_lock *lock, + struct scoutfs_lock_coverage *cov); +bool scoutfs_lock_is_covered(struct super_block *sb, + struct scoutfs_lock_coverage *cov); +void scoutfs_lock_del_coverage(struct super_block *sb, + struct scoutfs_lock_coverage *cov); + void scoutfs_free_unused_locks(struct super_block *sb, unsigned long nr); int scoutfs_lock_setup(struct super_block *sb);