Linux内核原语（九）——互斥体（mutex）

Linux内核原语（九）——互斥体（mutex）

小狼@http://blog.csdn.net/xiaolangyangyang

互斥体是一种睡眠锁，他是一种简单的睡眠锁，其行为和 count 为 1 的信号量类似。（关于信号量参考：Linux内核原语（八）——信号量（semaphore））。

互斥体简洁高效，但是相比信号量，有更多的限制，因此对于互斥体的使用条件更加严格：

任何时刻，只有一个指定的任务允许持有 mutex，也就是说，mutex 的计数永远是 1；
给 mutex 上锁这，必须负责给他解锁，也就是不允许在一个上下文中上锁，在另外一个上下文中解锁。这个限制注定了 mutex 无法承担内核和用户空间同步的复杂场景。常用的方式是在一个上下文中进行上锁/解锁。
递归的调用上锁和解锁是不允许的。也就是说，不能递归的去持有同一个锁，也不能够递归的解开一个已经解开的锁。
当持有 mutex 的进程，不允许退出
mutex 不允许在中断上下文和软中断上下文中使用过，即便是mutex_trylock 也不行
mutex 只能使用内核提供的 APIs操作，不允许拷贝，手动初始化和重复初始化

信号量和互斥体

他们两者很相似，除非是 mutex 的限制妨碍到逻辑，否则这两者之间，首选 mutex

自旋锁和互斥体

多数情况，很好区分。中断中只能考虑自旋锁，任务睡眠使用互斥体。如果都可以的的情况下，低开销或者短时间的锁，选择自旋锁，长期加锁的话，使用互斥体。

互斥体的使用

函数定义	功能说明
mutex_lock(struct mutex *lock)	加锁，如果不可用，则睡眠（UNINTERRUPTIBLE）
mutex_lock_interruptible(struct mutex *lock);	加锁，如果不可用，则睡眠（TASK_INTERRUPTIBLE）
mutex_unlock(struct mutex *lock)	解锁
mutex_trylock(struct mutex *lock)	试图获取指定的 mutex，或得到返回1，否则返回 0
mutex_is_locked(struct mutex *lock)	如果 mutex 被占用返回1，否则返回 0

互斥体的实现

互斥体的定义在：include/linux/mutex.h

1. mutex 的结构

struct mutex {atomic_long_t        owner;spinlock_t        wait_lock;
#ifdef CONFIG_MUTEX_SPIN_ON_OWNERstruct optimistic_spin_queue osq; /* Spinner MCS lock */
#endifstruct list_head    wait_list;
#ifdef CONFIG_DEBUG_MUTEXESvoid            *magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOCstruct lockdep_map    dep_map;
#endif
};

2. mutex 初始化

void
__mutex_init(struct mutex *lock, const char *name, struct lock_class_key *key)
{atomic_long_set(&lock->owner, 0);spin_lock_init(&lock->wait_lock);INIT_LIST_HEAD(&lock->wait_list);
#ifdef CONFIG_MUTEX_SPIN_ON_OWNERosq_lock_init(&lock->osq);
#endifdebug_mutex_init(lock, name, key);
}
EXPORT_SYMBOL(__mutex_init);

3. mutex 加锁

void __sched mutex_lock(struct mutex *lock)
{might_sleep();if (!__mutex_trylock_fast(lock))__mutex_lock_slowpath(lock);
}

首先 check 是否能够获得锁，否则调用到 __mutex_lock_slowpath：

static noinline void __sched
__mutex_lock_slowpath(struct mutex *lock)
{__mutex_lock(lock, TASK_UNINTERRUPTIBLE, 0, NULL, _RET_IP_);
}

static int __sched
__mutex_lock(struct mutex *lock, long state, unsigned int subclass,struct lockdep_map *nest_lock, unsigned long ip)
{return __mutex_lock_common(lock, state, subclass, nest_lock, ip, NULL, false);
}

所以调用到了 __mutex_lock_common 函数：

static __always_inline int __sched
__mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,struct lockdep_map *nest_lock, unsigned long ip,struct ww_acquire_ctx *ww_ctx, const bool use_ww_ctx)
{struct mutex_waiter waiter;bool first = false;struct ww_mutex *ww;int ret;might_sleep();ww = container_of(lock, struct ww_mutex, base);if (use_ww_ctx && ww_ctx) {if (unlikely(ww_ctx == READ_ONCE(ww->ctx)))return -EALREADY;/** Reset the wounded flag after a kill. No other process can* race and wound us here since they can't have a valid owner* pointer if we don't have any locks held.*/if (ww_ctx->acquired == 0)ww_ctx->wounded = 0;}preempt_disable();mutex_acquire_nest(&lock->dep_map, subclass, 0, nest_lock, ip);if (__mutex_trylock(lock) ||mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx, NULL)) {/* got the lock, yay! */lock_acquired(&lock->dep_map, ip);if (use_ww_ctx && ww_ctx)ww_mutex_set_context_fastpath(ww, ww_ctx);preempt_enable();return 0;}spin_lock(&lock->wait_lock);/** After waiting to acquire the wait_lock, try again.*/if (__mutex_trylock(lock)) {if (use_ww_ctx && ww_ctx)__ww_mutex_check_waiters(lock, ww_ctx);goto skip_wait;}debug_mutex_lock_common(lock, &waiter);lock_contended(&lock->dep_map, ip);if (!use_ww_ctx) {/* add waiting tasks to the end of the waitqueue (FIFO): */__mutex_add_waiter(lock, &waiter, &lock->wait_list);#ifdef CONFIG_DEBUG_MUTEXESwaiter.ww_ctx = MUTEX_POISON_WW_CTX;
#endif} else {/** Add in stamp order, waking up waiters that must kill* themselves.*/ret = __ww_mutex_add_waiter(&waiter, lock, ww_ctx);if (ret)goto err_early_kill;waiter.ww_ctx = ww_ctx;}waiter.task = current;set_current_state(state);for (;;) {/** Once we hold wait_lock, we're serialized against* mutex_unlock() handing the lock off to us, do a trylock* before testing the error conditions to make sure we pick up* the handoff.*/if (__mutex_trylock(lock))goto acquired;/** Check for signals and kill conditions while holding* wait_lock. This ensures the lock cancellation is ordered* against mutex_unlock() and wake-ups do not go missing.*/if (unlikely(signal_pending_state(state, current))) {ret = -EINTR;goto err;}if (use_ww_ctx && ww_ctx) {ret = __ww_mutex_check_kill(lock, &waiter, ww_ctx);if (ret)goto err;}spin_unlock(&lock->wait_lock);schedule_preempt_disabled();/** ww_mutex needs to always recheck its position since its waiter* list is not FIFO ordered.*/if ((use_ww_ctx && ww_ctx) || !first) {first = __mutex_waiter_is_first(lock, &waiter);if (first)__mutex_set_flag(lock, MUTEX_FLAG_HANDOFF);}set_current_state(state);/** Here we order against unlock; we must either see it change* state back to RUNNING and fall through the next schedule(),* or we must see its unlock and acquire.*/if (__mutex_trylock(lock) ||(first && mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx, &waiter)))break;spin_lock(&lock->wait_lock);}spin_lock(&lock->wait_lock);
acquired:__set_current_state(TASK_RUNNING);if (use_ww_ctx && ww_ctx) {/** Wound-Wait; we stole the lock (!first_waiter), check the* waiters as anyone might want to wound us.*/if (!ww_ctx->is_wait_die &&!__mutex_waiter_is_first(lock, &waiter))__ww_mutex_check_waiters(lock, ww_ctx);}mutex_remove_waiter(lock, &waiter, current);if (likely(list_empty(&lock->wait_list)))__mutex_clear_flag(lock, MUTEX_FLAGS);debug_mutex_free_waiter(&waiter);skip_wait:/* got the lock - cleanup and rejoice! */lock_acquired(&lock->dep_map, ip);if (use_ww_ctx && ww_ctx)ww_mutex_lock_acquired(ww, ww_ctx);spin_unlock(&lock->wait_lock);preempt_enable();return 0;err:__set_current_state(TASK_RUNNING);mutex_remove_waiter(lock, &waiter, current);
err_early_kill:spin_unlock(&lock->wait_lock);debug_mutex_free_waiter(&waiter);mutex_release(&lock->dep_map, 1, ip);preempt_enable();return ret;
}

进入等待队列。

4. mutex 解锁

void __sched mutex_unlock(struct mutex *lock)
{
#ifndef CONFIG_DEBUG_LOCK_ALLOCif (__mutex_unlock_fast(lock))return;
#endif__mutex_unlock_slowpath(lock, _RET_IP_);
}
EXPORT_SYMBOL(mutex_unlock);

调用到 __mutex_unlock_slowpath ：

static noinline void __sched __mutex_unlock_slowpath(struct mutex *lock, unsigned long ip)
{struct task_struct *next = NULL;DEFINE_WAKE_Q(wake_q);unsigned long owner;mutex_release(&lock->dep_map, 1, ip);/** Release the lock before (potentially) taking the spinlock such that* other contenders can get on with things ASAP.** Except when HANDOFF, in that case we must not clear the owner field,* but instead set it to the top waiter.*/owner = atomic_long_read(&lock->owner);for (;;) {unsigned long old;#ifdef CONFIG_DEBUG_MUTEXESDEBUG_LOCKS_WARN_ON(__owner_task(owner) != current);DEBUG_LOCKS_WARN_ON(owner & MUTEX_FLAG_PICKUP);
#endifif (owner & MUTEX_FLAG_HANDOFF)break;old = atomic_long_cmpxchg_release(&lock->owner, owner,__owner_flags(owner));if (old == owner) {if (owner & MUTEX_FLAG_WAITERS)break;return;}owner = old;}spin_lock(&lock->wait_lock);debug_mutex_unlock(lock);if (!list_empty(&lock->wait_list)) {/* get the first entry from the wait-list: */struct mutex_waiter *waiter =list_first_entry(&lock->wait_list,struct mutex_waiter, list);next = waiter->task;debug_mutex_wake_waiter(lock, waiter);wake_q_add(&wake_q, next);}if (owner & MUTEX_FLAG_HANDOFF)__mutex_handoff(lock, next);spin_unlock(&lock->wait_lock);wake_up_q(&wake_q);
}

做唤醒操作。