http://www.embexperts.com/viewthread.php?tid=31
兩者最大區別:信號量可以允許多個線程進入臨界區,而互斥體只允許一個線程進入臨界區。本貼將描述信號量與互斥體之間的細微區別以及在實際的代碼設計中如何使用它們。
信號量在2.6.26中的定義如下:
struct semaphore {
spinlock_t lock;
unsigned int count;
struct list_head wait_list;
};
從以上信號量的定義中,可以看到信號量底層使用到了spin lock的鎖定機制,這個spin lock主要用來確保對count成員的原子性的操作(count--)和測試(count > 0)。
信號量的P操作:
(1).void down(struct semaphore *sem);
(2).int down_interruptible(struct semaphore *sem);
(3).int down_trylock(struct semaphore *sem);
其中(1)中的函數根據2.6.26中的代碼注釋,這個函數已經out了(Use of this function is deprecated),所以從實用角度,徹底忘了它吧。其中(3)試圖去獲得一個信號量,如果沒有獲得,函數立刻返回1而不會讓當前進程進入睡眠狀態。
(2)最常用,所以是我們的重點關注:
/**
* down_interruptible - acquire the semaphore unless interrupted
* @sem: the semaphore to be acquired
*
* Attempts to acquire the semaphore. If no more tasks are allowed to
* acquire the semaphore, calling this function will put the task to sleep.
* If the sleep is interrupted by a signal, this function will return -EINTR.
* If the semaphore is successfully acquired, this function returns 0.
*/
int down_interruptible(struct semaphore *sem)
{
unsigned long flags;
int result = 0;
spin_lock_irqsave(&sem->lock, flags);
if (likely(sem->count > 0))
sem->count--;
else
result = __down_interruptible(sem);
spin_unlock_irqrestore(&sem->lock, flags);
return result;
}
這段函數很好理解:在保證原子操作的前提下,先測試count是否大於0,如果是說明可以獲得信號量,這種情況下需要先將count--,以確保別的進程能否獲得該信號量,然后函數返回,其調用者開始進入臨界區。如果沒有獲得信號量,當前進程利用struct semaphore 中wait_list加入等待隊列,開始睡眠。
對於需要休眠的情況,在__down_interruptible()函數中,會構造一個struct semaphore_waiter類型的變量(struct semaphore_waiter定義如下:
struct semaphore_waiter {
struct list_head list;
struct task_struct *task;
int up;
};),將當前進程賦給task,並利用其list成員將該變量的節點加入到以sem中的wait_list為頭部的一個列表中,假設有多個進程在sem上調用down_interruptible,則sem的wait_list上形成的隊列如下圖:
(注:將一個進程阻塞,一般的經過是先把進程放到等待隊列中,接着改變進程的狀態,比如設為TASK_INTERRUPTIBLE,然后調用調度函數schedule(),后者將會把當前進程從cpu的運行隊列中摘下)
信號量的V操作:
void up(struct semaphore *sem);
/**
* up - release the semaphore
* @sem: the semaphore to release
*
* Release the semaphore. Unlike mutexes, up() may be called from any
* context and even by tasks which have never called down().
*/
void up(struct semaphore *sem)
{
unsigned long flags;
spin_lock_irqsave(&sem->lock, flags);
if (likely(list_empty(&sem->wait_list)))
sem->count++;
else
__up(sem);
spin_unlock_irqrestore(&sem->lock, flags);
}
如果沒有其他線程等待在目前即將釋放的信號量上,那么只需將count++即可。如果有其他線程正因為等待該信號量而睡眠,那么調用__up.
__up的定義:
static noinline void __sched __up(struct semaphore *sem)
{
struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
struct semaphore_waiter, list);
list_del(&waiter->list);
waiter->up = 1;
wake_up_process(waiter->task);
}
這個函數首先獲得sem所在的wait_list為頭部的鏈表的第一個有效節點,然后從鏈表中將其刪除,然后喚醒該節點上睡眠的進程。
由此可見,對於sem上的每次down_interruptible調用,都會在sem的wait_list鏈表尾部加入一新的節點。對於sem上的每次up調用,都會刪除掉wait_list鏈表中的第一個有效節點,並喚醒睡眠在該節點上的進程。
互斥體mutex
Linux 2.6.26中mutex的定義:
struct mutex {
/* 1: unlocked, 0: locked, negative: locked, possible waiters */
atomic_t count;
spinlock_t wait_lock;
struct list_head wait_list;
#ifdef CONFIG_DEBUG_MUTEXES
struct thread_info *owner;
const char *name;
void *magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
struct lockdep_map dep_map;
#endif
};
對比1樓中的struct semaphore,struct mutex除了增加了幾個作為debug用途的成員變量外,和semaphore幾乎長得一樣。但是mutex的引入主要是為了提供互斥機制,以避免多個進程同時在一個臨界區中運行。
如果靜態聲明一個count=1的semaphore變量,可以使用DECLARE_MUTEX(name), 宏DECLARE_MUTEX的相關定義如下:
#define __SEMAPHORE_INITIALIZER(name, n) \
{ \
.lock = __SPIN_LOCK_UNLOCKED((name).lock), \
.count = n, \
.wait_list = LIST_HEAD_INIT((name).wait_list), \
}
#define __DECLARE_SEMAPHORE_GENERIC(name, count) \
struct semaphore name = __SEMAPHORE_INITIALIZER(name, count)
#define DECLARE_MUTEX(name) __DECLARE_SEMAPHORE_GENERIC(name, 1)
所以DECLARE_MUTEX(name)實際上是定義一個semaphore,所以它的使用應該對應信號量的P,V函數.
如果要定義一個靜態mutex型變量,應該使用DEFINE_MUTEX:
#define __MUTEX_INITIALIZER(lockname) \
{ .count = ATOMIC_INIT(1) \
, .wait_lock = __SPIN_LOCK_UNLOCKED(lockname.wait_lock) \
, .wait_list = LIST_HEAD_INIT(lockname.wait_list) \
__DEBUG_MUTEX_INITIALIZER(lockname) \
__DEP_MAP_MUTEX_INITIALIZER(lockname) }#define DEFINE_MUTEX(mutexname) \
struct mutex mutexname = __MUTEX_INITIALIZER(mutexname)
如果在程序運行期要初始化一個mutex變量,可以使用mutex_init(mutex),mutex_init是個宏,在該宏定義的內部,會調用__mutex_init函數。
#define mutex_init(mutex) \
do { \
static struct lock_class_key __key; \
\
__mutex_init((mutex), #mutex, &__key); \
} while (0)
__mutex_init定義如下:
/***
* mutex_init - initialize the mutex
* @lock: the mutex to be initialized
*
* Initialize the mutex to unlocked state.
*
* It is not allowed to initialize an already locked mutex.
*/
void
__mutex_init(struct mutex *lock, const char *name, struct lock_class_key *key)
{
atomic_set(&lock->count, 1);
spin_lock_init(&lock->wait_lock);
INIT_LIST_HEAD(&lock->wait_list);
debug_mutex_init(lock, name, key);
}
從__mutex_init的定義可以看出,在使用mutex_init宏來初始化一個mutex變量時,應該使用mutex的指針型。
mutex上的P,V操作:void mutex_lock(struct mutex *lock)和void __sched mutex_unlock(struct mutex *lock)
從原理上講,mutex實際上是count=1情況下的semaphore,所以其PV操作應該和semaphore是一樣的。但是在實際的Linux代碼上,出於性能優化的角度,並非只是單純的重用down_interruptible和up的代碼。以ARM平台的mutex_lock為例,實際上是將mutex_lock分成兩部分實現:fast path和slow path,主要是基於這樣一個事實:在絕大多數情況下,試圖獲得互斥體的代碼總是可以成功獲得。所以Linux的代碼針對這一事實用ARM V6上的LDREX和STREX指令來實現fast path以期獲得最佳的執行性能。metex_lock在ARM平台的代碼定義如下:
void inline __sched mutex_lock(struct mutex *lock)
{
might_sleep();
/*
* The locking fastpath is the 1->0 transition from
* 'unlocked' into 'locked' state.
*/
__mutex_fastpath_lock(&lock->count, __mutex_lock_slowpath);
}
其中的might_sleep主要用於調試目的,這里不考慮。函數的核心是__mutex_fastpath_lock(&lock->count, __mutex_lock_slowpath),也就是上面提到的fast path和slow path。如果在__mutex_fastpath_lock函數中可以成功獲得互斥體,那么__mutex_lock_slowpath將不會調用,否則調用之(使試圖獲得該互斥體而不可得的進程進入休眠狀態)。在絕大多數情況下,__mutex_lock_slowpath都沒機會被調用。__mutex_fastpath_lock主要實現原子級的dec_test這樣的操作序列,可以想見應該是處理器相關的匯編代碼,在ARM V6+,其實現如下:
static inline void
__mutex_fastpath_lock(atomic_t *count, void (*fail_fn)(atomic_t *))
{
int __ex_flag, __res;
__asm__ (
"ldrex %0, [%2] \n\t"
"sub %0, %0, #1 \n\t"
"strex %1, %0, [%2] "
: "=&r" (__res), "=&r" (__ex_flag)
: "r" (&(count)->counter)
: "cc","memory" );
__res |= __ex_flag;
if (unlikely(__res != 0))
fail_fn(count);
}
代碼看起來很簡單,如果count減1之后不為0(說明這之前該互斥體已被別的進程獲得)或者是最后的strex導致代碼中的"%1",也就是__ex_flag不等於0(說明此時有別的進程正在和當前進程競爭該互斥體,而且別的進程較當前進程更早發起對該互斥體的鎖定請求,這種情況下當前進程將不會在本次的競爭中獲勝),這兩種情況都將導致__mutex_lock_slowpath被調用。但是絕大多數情況下,當前進程都會成功獲得互斥體,換句話說,絕大多數情況下,__res都是等於0的(這也是為什么在if語句中使用了unlikely的原因)。
__mutex_lock_slowpath的代碼實現因為加入了debug支持的緣故顯得有點晦澀,本貼不再仔細剖析。如果當前進程在競爭一個互斥體的過程中fast path lock部分失敗,那么在進入low path lock部分,依然有機會重新獲得該互斥體(比如之前獲得該互斥體的進程已經釋放了它),如果依然沒法獲得,那么將當前進程的狀態設置為TASK_INTERRUPTIBLE,調用schedule()使當前進程進入休眠。
從代碼閱讀的角度,似乎一切看起來都很正常。但是有一個問題,在Linux開源社區,經常會聽到有人說:讀懂Linux的代碼並不難,難得是明白代碼背后隱藏的哲學。如果經常去琢磨Linux為什么要象現在看起來的那樣去設計代碼,對提升自己的設計能力無疑是有幫助的。所以,從mutex_lock的設計思想出發,想想為什么semaphore上的down_interruptible是另一幅樣子呢?能否象mutex那樣去實現down_interruptible函數呢?