diff --git a/kmod/src/dlmglue.c b/kmod/src/dlmglue.c index edde124d..82611482 100644 --- a/kmod/src/dlmglue.c +++ b/kmod/src/dlmglue.c @@ -419,7 +419,7 @@ static inline int __levels_compat(int lockres_level, int wanted) return level_compat_matrix[wanted + 1][lockres_level + 1]; } -static inline int levels_compat(struct ocfs2_lock_res *lockres, int wanted) +int ocfs2_levels_compat(struct ocfs2_lock_res *lockres, int wanted) { return __levels_compat(lockres->l_level, wanted); } @@ -519,7 +519,7 @@ static inline void ocfs2_generic_handle_downconvert_action(struct ocfs2_lock_res BUG_ON(lockres->l_blocking <= DLM_LOCK_NL); lockres->l_level = lockres->l_requested; - if (levels_compat(lockres, dc_level)) { + if (ocfs2_levels_compat(lockres, dc_level)) { lockres->l_blocking = DLM_LOCK_NL; lockres_clear_flags(lockres, OCFS2_LOCK_BLOCKED); } @@ -1005,16 +1005,6 @@ static inline int ocfs2_may_continue_on_blocked_lock(struct ocfs2_lock_res *lock return wanted <= ocfs2_downconvert_level(lockres, lockres->l_blocking); } -/* the caller doesn't have to wait on a blocked lock if their wanted level - * is compatible with it and there are already holders of the lock */ -static inline int lockres_allow_recursion(struct ocfs2_lock_res *lockres, - int wanted) -{ - return (lockres->l_ops->flags & LOCK_TYPE_RECURSIVE) && - levels_compat(lockres, wanted) && - lockres_has_holders(lockres, H_ANY); -} - static void ocfs2_init_mask_waiter(struct ocfs2_mask_waiter *mw) { INIT_LIST_HEAD(&mw->mw_item); @@ -1153,7 +1143,7 @@ again: * here. If the lock is blocked waiting on a downconvert, * we'll get caught below. */ if (lockres->l_flags & OCFS2_LOCK_BUSY && - !levels_compat(lockres, level)) { + !ocfs2_levels_compat(lockres, level)) { /* is someone sitting in dlm_lock? If so, wait on * them. */ lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0); @@ -1176,12 +1166,11 @@ again: * OCFS2_LOCK_BLOCKED check to ensure that there is no pending * downconvert request. */ - if (levels_compat(lockres, level)) + if (ocfs2_levels_compat(lockres, level)) goto update_holders; } if (lockres->l_flags & OCFS2_LOCK_BLOCKED && - !lockres_allow_recursion(lockres, level) && !ocfs2_may_continue_on_blocked_lock(lockres, level)) { /* is the lock is currently blocked on behalf of * another node */ diff --git a/kmod/src/dlmglue.h b/kmod/src/dlmglue.h index c805b7e8..af138f80 100644 --- a/kmod/src/dlmglue.h +++ b/kmod/src/dlmglue.h @@ -286,20 +286,6 @@ struct ocfs2_lock_res_ops { */ #define LOCK_TYPE_USES_LVB 0x2 -/* - * Tells dlmglue to override fairness considerations when locking this - * lock type - the blocking flag will be ignored when a lock is - * requested and we already have it at the appropriate level and the - * resource is currently held. This allows a process to acquire a - * dlmglue lock on the same resource multiple times in a row without - * deadlocking, even if another node has asked for a competing lock on - * the resource. - * - * Note that lock/unlock calls must always be balanced (1 unlock for - * every lock), even when this flag is set. - */ -#define LOCK_TYPE_RECURSIVE 0x4 - struct ocfs2_lock_holder { struct list_head oh_list; struct pid *oh_owner_pid; @@ -352,6 +338,7 @@ void ocfs2_wake_downconvert_thread(struct ocfs2_super *osb); struct ocfs2_dlm_debug *ocfs2_new_dlm_debug(void); void ocfs2_put_dlm_debug(struct ocfs2_dlm_debug *dlm_debug); +int ocfs2_levels_compat(struct ocfs2_lock_res *lockres, int wanted); #if 0 /* To set the locking protocol on module initialization */ diff --git a/kmod/src/lock.c b/kmod/src/lock.c index f27c2916..85c08a93 100644 --- a/kmod/src/lock.c +++ b/kmod/src/lock.c @@ -61,6 +61,108 @@ struct lock_info { static void scoutfs_lock_reclaim(struct work_struct *work); +struct task_ref { + struct task_struct *task; + struct rb_node node; + int count; + int mode;/* for debugging */ +}; + +static struct task_ref *find_task_ref(struct scoutfs_lock *lock, + struct task_struct *task) +{ + struct rb_node *n; + struct task_ref *tmp; + + spin_lock(&lock->task_refs_lock); + n = lock->task_refs.rb_node; + while (n) { + tmp = rb_entry(n, struct task_ref, node); + + if (tmp->task < task) + n = n->rb_left; + else if (tmp->task > task) + n = n->rb_right; + else { + spin_unlock(&lock->task_refs_lock); + return tmp; + } + } + spin_unlock(&lock->task_refs_lock); + + return NULL; +} + +static struct task_ref *alloc_task_ref(struct task_struct *task, int mode) +{ + struct task_ref *ref = kzalloc(sizeof(*ref), GFP_NOFS); + if (ref) { + ref->task = task; + ref->count = 1; + ref->mode = mode; + RB_CLEAR_NODE(&ref->node); + } + return ref; +} + +static void insert_task_ref(struct scoutfs_lock *lock, struct task_ref *ref) +{ + struct task_ref *tmp; + struct rb_node *parent = NULL; + struct rb_node **p; + + spin_lock(&lock->task_refs_lock); + p = &lock->task_refs.rb_node; + while (*p) { + parent = *p; + + tmp = rb_entry(parent, struct task_ref, node); + + if (tmp->task < ref->task) + p = &(*p)->rb_left; + else if (tmp->task > ref->task) + p = &(*p)->rb_right; + else + BUG(); /* We should never find a duplicate */ + } + + rb_link_node(&ref->node, parent, p); + rb_insert_color(&ref->node, &lock->task_refs); + spin_unlock(&lock->task_refs_lock); +} + +static void get_task_ref(struct task_ref *ref) +{ + ref->count++; +} + +static struct task_ref *new_task_ref(struct scoutfs_lock *lock, + struct task_struct *task, int mode) +{ + struct task_ref *ref = alloc_task_ref(task, mode); + if (ref) + insert_task_ref(lock, ref); + + return ref; +} + +static int put_task_ref(struct scoutfs_lock *lock, struct task_ref *ref) +{ + if (!ref) + return 0; + + ref->count--; + if (ref->count == 0) { + spin_lock(&lock->task_refs_lock); + rb_erase(&ref->node, &lock->task_refs); + spin_unlock(&lock->task_refs_lock); + + kfree(ref); + return 0; + } + return 1; +} + /* * Invalidate caches on this because another node wants a lock * with the a lock with the given mode and range. We always have to @@ -196,7 +298,7 @@ static struct ocfs2_lock_res_ops scoufs_ino_lops = { .downconvert_worker = ino_lock_downconvert, /* XXX: .check_downconvert that queries the item cache for dirty items */ .print = lock_name_string, - .flags = LOCK_TYPE_REQUIRES_REFRESH|LOCK_TYPE_RECURSIVE, + .flags = LOCK_TYPE_REQUIRES_REFRESH, }; static struct ocfs2_lock_res_ops scoufs_ino_index_lops = { @@ -204,7 +306,6 @@ static struct ocfs2_lock_res_ops scoufs_ino_index_lops = { .downconvert_worker = ino_lock_downconvert, /* XXX: .check_downconvert that queries the item cache for dirty items */ .print = lock_name_string, - .flags = LOCK_TYPE_RECURSIVE, }; static struct ocfs2_lock_res_ops scoutfs_global_lops = { @@ -251,6 +352,8 @@ static struct scoutfs_lock *alloc_scoutfs_lock(struct super_block *sb, } } + spin_lock_init(&lock->task_refs_lock); + lock->task_refs = RB_ROOT; RB_CLEAR_NODE(&lock->node); lock->sb = sb; lock->lock_name = *lock_name; @@ -491,6 +594,7 @@ static int lock_name_keys(struct super_block *sb, int mode, int flags, { DECLARE_LOCK_INFO(sb, linfo); struct scoutfs_lock *lock; + struct task_ref *ref = NULL; int lkm_flags; int ret; @@ -506,13 +610,37 @@ static int lock_name_keys(struct super_block *sb, int mode, int flags, trace_scoutfs_lock_resource(sb, lock); + if (!(flags & SCOUTFS_LKF_NO_TASK_REF)) { + ref = find_task_ref(lock, current); + if (ref) { + /* + * We found a ref, which means we have already locked + * this resource. Check that the calling task isn't + * trying to switch modes in the middle of a recursive + * lock request. + */ + BUG_ON(!ocfs2_levels_compat(&lock->lockres, mode)); + get_task_ref(ref); + ret = 0; + goto out; + } + + ref = new_task_ref(lock, current, mode); + if (!ref) { + ret = -ENOMEM; + goto out; + } + } + lkm_flags = DLM_LKF_NOORDER; if (flags & SCOUTFS_LKF_TRYLOCK) lkm_flags |= DLM_LKF_NOQUEUE; /* maybe also NONBLOCK? */ ret = ocfs2_cluster_lock(&linfo->dlmglue, &lock->lockres, mode, lkm_flags, 0); +out: if (ret) { + put_task_ref(lock, ref); dec_lock_users(lock); put_scoutfs_lock(sb, lock); } else { @@ -837,9 +965,10 @@ int scoutfs_lock_node_id(struct super_block *sb, int mode, int flags, &scoutfs_node_id_lops, &start, &end, lock); } -void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, - int level) +void scoutfs_unlock_flags(struct super_block *sb, struct scoutfs_lock *lock, + int level, int flags) { + struct task_ref *ref; DECLARE_LOCK_INFO(sb, linfo); if (!lock) @@ -847,6 +976,13 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, trace_scoutfs_unlock(sb, lock); + if (!(flags & SCOUTFS_LKF_NO_TASK_REF)) { + ref = find_task_ref(lock, current); + BUG_ON(!ref); + if (put_task_ref(lock, ref)) + return; + } + ocfs2_cluster_unlock(&linfo->dlmglue, &lock->lockres, level); dec_lock_users(lock); @@ -854,6 +990,12 @@ void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, put_scoutfs_lock(sb, lock); } +void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, + int level) +{ + scoutfs_unlock_flags(sb, lock, level, 0); +} + /* * The moment this is done we can have other mounts start asking * us to write back and invalidate, so do this very very late. diff --git a/kmod/src/lock.h b/kmod/src/lock.h index 9d5af790..63a914fe 100644 --- a/kmod/src/lock.h +++ b/kmod/src/lock.h @@ -7,6 +7,7 @@ #define SCOUTFS_LKF_REFRESH_INODE 0x01 /* update stale inode from item */ #define SCOUTFS_LKF_TRYLOCK 0x02 /* EAGAIN if contention */ +#define SCOUTFS_LKF_NO_TASK_REF 0x04 /* don't create a task ref */ /* flags for scoutfs_lock->flags */ enum { @@ -30,6 +31,8 @@ struct scoutfs_lock { unsigned int users; /* Tracks active users of this lock */ unsigned long flags; wait_queue_head_t waitq; + struct rb_root task_refs; + spinlock_t task_refs_lock; }; u64 scoutfs_lock_refresh_gen(struct scoutfs_lock *lock); @@ -54,6 +57,8 @@ int scoutfs_lock_node_id(struct super_block *sb, int mode, int flags, u64 node_id, struct scoutfs_lock **lock); void scoutfs_unlock(struct super_block *sb, struct scoutfs_lock *lock, int level); +void scoutfs_unlock_flags(struct super_block *sb, struct scoutfs_lock *lock, + int level, int flags); int scoutfs_lock_setup(struct super_block *sb); void scoutfs_lock_destroy(struct super_block *sb); diff --git a/kmod/src/server.c b/kmod/src/server.c index 007ca7d6..6d178574 100644 --- a/kmod/src/server.c +++ b/kmod/src/server.c @@ -929,7 +929,8 @@ static void scoutfs_server_func(struct work_struct *work) init_waitqueue_head(&waitq); - ret = scoutfs_lock_global(sb, DLM_LOCK_EX, SCOUTFS_LKF_TRYLOCK, + ret = scoutfs_lock_global(sb, DLM_LOCK_EX, + SCOUTFS_LKF_TRYLOCK|SCOUTFS_LKF_NO_TASK_REF, SCOUTFS_LOCK_TYPE_GLOBAL_SERVER, &lock); if (ret) diff --git a/kmod/src/super.c b/kmod/src/super.c index 719640e1..7a21393c 100644 --- a/kmod/src/super.c +++ b/kmod/src/super.c @@ -332,7 +332,8 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent) out: if (ret) { scoutfs_server_destroy(sb); - scoutfs_unlock(sb, sbi->node_id_lock, DLM_LOCK_EX); + scoutfs_unlock_flags(sb, sbi->node_id_lock, DLM_LOCK_EX, + SCOUTFS_LKF_NO_TASK_REF); sbi->node_id_lock = NULL; } return ret; @@ -362,7 +363,8 @@ static void scoutfs_kill_sb(struct super_block *sb) kill_block_super(sb); if (sbi) { - scoutfs_unlock(sb, sbi->node_id_lock, DLM_LOCK_EX); + scoutfs_unlock_flags(sb, sbi->node_id_lock, DLM_LOCK_EX, + SCOUTFS_LKF_NO_TASK_REF); sbi->node_id_lock = NULL; scoutfs_lock_destroy(sb);