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);