轉自 |嵌入式藝術 除了原子操作，中斷屏蔽，自旋鎖以及自旋鎖的衍生鎖之外，在Linux內核中還存在著一些其他同步互斥的手段。

下面我們來理解一下信號量，互斥體，完成量機制。

1、信號量介紹

信號量（Semaphore）是操作系統中最典型的用于同步和互斥的手段，信號量的值可以是0、1或者n。信號量與操作系統中的經典概念PV操作對應。 P（Produce）：

將信號量S的值減1，即S=S-1；

如果S≥0，則該進程繼續執行；否則該進程置為等待狀態，排入等待隊列。

V（Vaporize）：

將信號量S的值加1，即S=S+1；

如果S>0，喚醒隊列中等待信號量的進程。

信號量核心思想： 信號量的操作，更適合去解決生產者和消費者的問題，比如：我做出來一個餅，你才能吃一個餅；如果我沒做出來，你就先釋放CPU去忙其他的事情。

2、信號量的API

struct semaphore sem; // 定義信號量

void sema_init(struct semaphore *sem, int val); // 初始化信號量，并設置信號量sem的值為val。

void down(struct semaphore * sem); // 獲得信號量sem，它會導致睡眠，因此不能在中斷上下文中使用。
int down_interruptible(struct semaphore * sem); // 該函數功能與down類似，不同之處為，因為down（）進入睡眠狀態的進程不能被信號打斷，但因為down_interruptible（）進入睡眠狀態的進程能被信號打斷，信號也會導致該函數返回，這時候函數的返回值非0
int down_trylock(struct semaphore * sem); // 嘗試獲得信號量sem，如果能夠立刻獲得，它就獲得該信號量并返回0，否則，返回非0值。它不會導致調用者睡眠，可以在中斷上下文中使用。

void up(struct semaphore * sem); // 釋放信號量，喚醒等待者。

由于新的Linux內核傾向于直接使用mutex作為互斥手段，信號量用作互斥不再被推薦使用。 信號量也可以用于同步，一個進程A執行down（）等待信號量，另外一個進程B執行up（）釋放信號量，這樣進程A就同步地等待了進程B。

3、API實現

3.1 semaphore

struct semaphore {
raw_spinlock_t lock;
unsigned int count;
struct list_head wait_list;
};

結構體名稱：semaphore

文件位置：include/linux/semaphore.h

主要作用：用于定義一個信號量。

raw_spinlock_t：信號量結構體也使用了自旋鎖，避免互斥。

count：表示信號量的計數器，表示資源的數量

struct list_head wait_list: 這是一個鏈表頭，用于管理等待信號量的線程。當信號量的 count 等于0，即沒有可用資源時，等待信號量的線程會被加入到這個鏈表中，以便在資源可用時進行喚醒。

3.2 sema_init

static inline void sema_init(struct semaphore *sem, int val)
{
static struct lock_class_key __key;
*sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem, val);
lockdep_init_map(&sem->lock.dep_map, "semaphore->lock", &__key, 0);
}

#define __SEMAPHORE_INITIALIZER(name, n)
{
.lock = __RAW_SPIN_LOCK_UNLOCKED((name).lock),
.count = n,
.wait_list = LIST_HEAD_INIT((name).wait_list),
}

#define LIST_HEAD_INIT(name) { &(name), &(name) }

函數名稱：sema_init

文件位置：include/linux/semaphore.h

主要作用：初始化信號量，并設置信號量sem的值為val。

實現流程：

使用__SEMAPHORE_INITIALIZER宏定義來初始化信號量

使用__RAW_SPIN_LOCK_UNLOCKED宏定義來初始化自旋鎖

直接將val賦值給信號量的值count

使用LIST_HEAD_INIT來初始化一個鏈表

#defineLIST_HEAD_INIT(name){&(name),&(name)}

該宏接受一個參數 name，并返回一個結構體對象。這個對象有兩個成員 next 和 prev，分別指向 name 本身。

這樣，當我們使用該宏來初始化鏈表頭節點時，會得到一個擁有 next 和 prev 成員的結構體對象。其中 next 和 prev 成員都指向該結構體對象本身。

這種初始化方式可以用于創建一個空的雙向鏈表，因為在初始狀態下，鏈表頭節點的 next 和 prev 指針都指向自身，表示鏈表為空。

3.3 down

/**

down - acquire the semaphore
@sem: the semaphore to be acquired

Acquires the semaphore. If no more tasks are allowed to acquire the
semaphore, calling this function will put the task to sleep until the
semaphore is released.

Use of this function is deprecated, please use down_interruptible() or
down_killable() instead.
*/
void down(struct semaphore *sem)
{
unsigned long flags;

raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(sem->count > 0))
sem->count--;
else
__down(sem);
raw_spin_unlock_irqrestore(&sem->lock, flags);
}
EXPORT_SYMBOL(down);

static noinline void __sched __down(struct semaphore *sem)
{
__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}

/*

Because this function is inlined, the 'state' parameter will be
constant, and thus optimised away by the compiler. Likewise the
'timeout' parameter for the cases without timeouts.
*/
static inline int __sched __down_common(struct semaphore *sem, long state,
long timeout)
{
struct semaphore_waiter waiter;

list_add_tail(&waiter.list, &sem->wait_list);
waiter.task = current;
waiter.up = false;

for (;;) {
if (signal_pending_state(state, current))
goto interrupted;
if (unlikely(timeout <= 0))
goto timed_out;
__set_current_state(state);
raw_spin_unlock_irq(&sem->lock);
timeout = schedule_timeout(timeout);
raw_spin_lock_irq(&sem->lock);
if (waiter.up)
return 0;
}

timed_out:
list_del(&waiter.list);
return -ETIME;

interrupted:
list_del(&waiter.list);
return -EINTR;
}

函數名稱：down

文件位置：kernel/locking/semaphore.c

主要作用：獲取信號量，如果信號量的值大于0，則消耗一個；如果不存在，則讓線程進入休眠狀態并等待信號量被釋放。

函數調用流程：

down(kernel/locking/semaphore.c)
|--> raw_spin_lock_irqsave // 獲取鎖，并保存中斷信息
||-> sem->count--; // 如果sem->count信號量存在，則消耗一個
|-> __down // 如果sem->count信號量不存在，則進入休眠狀態
|--> __down_common
|--> list_add_tail // 將當前線程添加到信號量的等待鏈表中，表示當前線程正在等待信號量。
|--> __set_current_state// 設置線程為休眠狀態
|--> raw_spin_unlock_irq// 釋放自旋鎖，讓其他線程可用
|--> schedule_timeout // 讓線程進入睡眠狀態，等待信號量釋放或超時。
|--> raw_spin_lock_irq // 重新獲取自旋鎖，繼續執行后續操作
|--> raw_spin_unlock_irqrestore // 釋放鎖，并恢復中斷信息

實現流程：

獲取信號量時，先使用raw_spin_lock_irqsave和raw_spin_unlock_irqrestore將信號量的操作包裹起來，避免競態發生。

然后對sem->count判斷，如果信號量大于0，就消耗一個，否則的話，將當前線程設置為休眠態

調用__down_common接口，默認將當前線程設置為TASK_UNINTERRUPTIBLE中斷不可打斷狀態，并設置最大超時時間MAX_SCHEDULE_TIMEOUT

struct semaphore_waiter waiter：創建waiter結構體，表示當前線程的狀態

調用__set_current_state接口，設置當前線程的狀態信息，即TASK_UNINTERRUPTIBLE

調用schedule_timeout，讓該線程讓出CPU，進入休眠態，并且在前面加上raw_spin_unlock_irq保證其他線程可以正常使用信號量。

當線程時間片到時，獲取CPU，并調用raw_spin_lock_irq，獲取鎖，來防止競態發生。

3.4 up

/**

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;

raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(list_empty(&sem->wait_list)))
sem->count++;
else
__up(sem);
raw_spin_unlock_irqrestore(&sem->lock, flags);
}
EXPORT_SYMBOL(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 = true;
wake_up_process(waiter->task);
}

函數名稱：up

文件位置：kernel/locking/semaphore.c

主要作用：獲取信號量，如果信號量的值大于0，則消耗一個；如果不存在，則讓線程進入休眠狀態并等待信號量被釋放。

實現流程：

相信分析完down后，up也變得很簡單

釋放信號量時，先使用raw_spin_lock_irqsave和raw_spin_unlock_irqrestore將信號量的操作包裹起來，避免競態發生。

然后對sem->wait_list判斷，如果其為空，說明沒有等待的線程，直接將sem->count自增，如果有等待的線程，則喚醒下一個線程。

調用wake_up_process來喚醒線程

4、總結

信號量較為簡單，一般常用來解決生產者和消費者的問題，其主要還是通過自旋鎖來實現互斥的作用，通過鏈表來管理等待隊列的線程信息，通過變量來代表資源的數量。

審核編輯：劉清