|
|
用戶類進(jìn)程之間使用信號(hào)量（semaphore）進(jìn)行同步，內(nèi)核線程之間也使用了信號(hào)量。信號(hào)量與自旋鎖類似，保護(hù)臨界區(qū)代碼。但信號(hào)量與自旋鎖有一定的區(qū)別，信號(hào)量在無(wú)法得到資源時(shí)，內(nèi)核線程處于睡眠阻塞狀態(tài)，而自旋鎖處于忙等待狀態(tài)。因此，如果資源被占用時(shí)間很短時(shí)，使用自旋鎖較好，因?yàn)樗晒?jié)約調(diào)度時(shí)間。如果資源被占用的時(shí)間較長(zhǎng)，使用信號(hào)量較好，因?yàn)榭勺孋PU調(diào)度去做其它進(jìn)程的工作。
操作信號(hào)量的API函數(shù)說(shuō)明如表6。
表6 信號(hào)量API函數(shù)功能說(shuō)明
| 函數(shù)定義 |
功能說(shuō)明 |
| sema_init(struct semaphore *sem, int val) |
初始化信號(hào)量，將信號(hào)量計(jì)數(shù)器值設(shè)置val。 |
| down(struct semaphore *sem) |
獲取信號(hào)量，不建議使用此函數(shù)。 |
| down_interruptible(struct semaphore *sem) |
可被中斷地獲取信號(hào)量，如果睡眠被信號(hào)中斷，返回錯(cuò)誤-EINTR。 |
| down_killable (struct semaphore *sem) |
可被殺死地獲取信號(hào)量。如果睡眠被致命信號(hào)中斷，返回錯(cuò)誤-EINTR。 |
| down_trylock(struct semaphore *sem) |
嘗試原子地獲取信號(hào)量，如果成功獲取，返回0，不能獲取，返回1。
|
| down_timeout(struct semaphore *sem, long jiffies) |
在指定的時(shí)間jiffies內(nèi)獲取信號(hào)量，若超時(shí)未獲取，返回錯(cuò)誤-ETIME。 |
| up(struct semaphore *sem) |
釋放信號(hào)量sem。 |
樣例：信號(hào)量的使用
下面函數(shù)do_utimes利用信號(hào)量防止多個(gè)線程對(duì)文件系統(tǒng)節(jié)點(diǎn)inode同時(shí)進(jìn)行訪問。其列出如下（在fs/open.c中）：
long do_utimes(char __user * filename, struct timeval * times)
{
struct inode * inode;
……
down(&inode->i_sem); //獲取信號(hào)量
error = notify_change(nd.dentry, &newattrs);//修改inode中值
up(&inode->i_sem); //釋放信號(hào)量
……
}
下面說(shuō)明信號(hào)量API函數(shù)。
（1）信號(hào)量結(jié)構(gòu)semaphore
信號(hào)量用結(jié)構(gòu)semaphore描述，它在自旋鎖的基礎(chǔ)上改進(jìn)而成，它包括一個(gè)自旋鎖、信號(hào)量計(jì)數(shù)器和一個(gè)等待隊(duì)列。用戶程序只能調(diào)用信號(hào)量API函數(shù)，而不能直接訪問信號(hào)量結(jié)構(gòu)，其列出如下（在include/linux/semaphore.h中）：
struct semaphore {
spinlock_t lock;
unsigned int count;
struct list_head wait_list;
};
（2）初始化函數(shù)sema_init
函數(shù)sema_init初始化信號(hào)量，將信號(hào)量值初始化為n，其列出如下：
static inline void sema_init(struct semaphore *sem, int val)
{
static struct lock_class_key __key;
*sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem, val);
/*初始化一個(gè)鎖的實(shí)例，用于調(diào)試中獲取信號(hào)量的調(diào)試信息*/
lockdep_init_map(&sem->lock.dep_map, "semaphore->lock", &__key, 0);
}
#define __SEMAPHORE_INITIALIZER(name, n) { .lock = __SPIN_LOCK_UNLOCKED?lock), \ //初始化自旋鎖
.count = n, \ //將信號(hào)量計(jì)數(shù)器賦值為n
.wait_list = LIST_HEAD_INIT((name).wait_list), \ //初始化等待隊(duì)列
}
（3）可中斷獲取信號(hào)量函數(shù)down_interruptible
函數(shù)down_interruptible獲取信號(hào)量，存放在參數(shù)sem中。它嘗試獲取信號(hào)量，如果其他線程被允許嘗試獲取此信號(hào)量，則將本線程睡眠等待。如果有一個(gè)信號(hào)中斷睡眠，則它返回錯(cuò)誤-EINTR。如果成功獲取信號(hào)量，函數(shù)返回0。
函數(shù)down_interruptible列出如下（在kernel/semaphore.c中）：
int down_interruptible(struct semaphore *sem)
{
unsigned long flags;
int result = 0;
spin_lock_irqsave(&sem->lock, flags); //獲取自旋鎖，關(guān)閉中斷，將狀態(tài)寄存器值存放在flags
/*如果信號(hào)量計(jì)數(shù)器值大于0，說(shuō)明有多個(gè)空閑資源可訪問，可以成功獲取信號(hào)量了*/
if (likely(sem->count > 0)) //likely表示成功獲取的概率大，通知編譯器進(jìn)行分支預(yù)測(cè)優(yōu)化
sem->count--;
else
result = __down_interruptible(sem); //進(jìn)入睡眠等待
spin_unlock_irqrestore(&sem->lock, flags);
return result;
}
static noinline int __sched __down_interruptible(struct semaphore *sem)
{
return __down_common(sem, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}
函數(shù)__down_common進(jìn)入睡眠等待，其列出如下：
static inline int __sched __down_common(struct semaphore *sem, long state,
long timeout)
{
struct task_struct *task = current;
struct semaphore_waiter waiter;
list_add_tail(&waiter.list, &sem->wait_list); //加入到等待隊(duì)列
waiter.task = task;
waiter.up = 0;
for (;;) {
if (state == TASK_INTERRUPTIBLE && signal_pending(task))
goto interrupted;
if (state == TASK_KILLABLE && fatal_signal_pending(task))
goto interrupted;
if (timeout <= 0)
goto timed_out;
__set_task_state(task, state);
spin_unlock_irq(&sem->lock);
timeout = schedule_timeout(timeout); //調(diào)度
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;
}
（3）釋放信號(hào)量函數(shù)up
函數(shù)up在沒有其他線程等待使用信號(hào)量的情況下釋放信號(hào)量，否則，喚醒其他等待線程。其列出如下：
void up(struct semaphore *sem)
{
unsigned long flags;
spin_lock_irqsave(&sem->lock, flags);
/*判斷是否有線程等待在此信號(hào)量上，即判斷等待隊(duì)列是否為空*/
if (likely(list_empty(&sem->wait_list)))
/*沒有線程等待此信號(hào)量，釋放信號(hào)量，將信號(hào)量計(jì)數(shù)器加1，表示增加了1個(gè)空閑資源*/
sem->count++;
else
__up(sem); /*將本線程從等待隊(duì)列刪除，喚醒等待此信號(hào)量的其他線程*/
spin_unlock_irqrestore(&sem->lock, flags);
}
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); //將本線程從等待隊(duì)列刪除
waiter->up = 1;
wake_up_process(waiter->task); //喚醒等待此信號(hào)量的其他線程
}
互斥鎖
信號(hào)量的初始值表示可以有多少個(gè)任務(wù)可同時(shí)訪問的共享資源，如果初始值為1，表示只有1個(gè)任務(wù)可以訪問，信號(hào)量變成互斥鎖（Mutex）。可見互斥鎖是信號(hào)量的特例。
互斥鎖（mutex）是在原子操作API的基礎(chǔ)上實(shí)現(xiàn)的信號(hào)量行為。互斥鎖不能進(jìn)行遞歸鎖定或解鎖，能用于交互上下文，同一時(shí)間只能有一個(gè)任務(wù)持有互斥鎖。
互斥鎖功能上基本上與信號(hào)量一樣，互斥鎖占用空間比信號(hào)量小，運(yùn)行效率比信號(hào)量高?；コ怄i的API函數(shù)功能說(shuō)明如表1。
表1 互斥鎖的API函數(shù)功能說(shuō)明
| API函數(shù) |
功能說(shuō)明 |
| DEFINE_MUTEX(mutexname) |
創(chuàng)建和初始化互斥鎖。 |
| void mutex_lock(struct mutex *lock); |
加鎖。 |
| void mutex_unlock(struct mutex *lock); |
解鎖。 |
| int mutex_trylock(struct mutex *lock); |
嘗試加鎖。 |
互斥鎖用結(jié)構(gòu)mutex描述，它含有信號(hào)量計(jì)數(shù)和等待隊(duì)列成員，信號(hào)量的值為1或0或負(fù)數(shù)。其列出如下（在include/linux/mutex.h中）：
struct mutex {
/* 1：表示解鎖，0：表示鎖住，負(fù)數(shù)：表示鎖住，可能有等待者*/
atomic_t count;
spinlock_t wait_lock; /*操作等待隊(duì)列的自旋鎖*/
struct list_head wait_list; /*等待隊(duì)列*/
/*省略了用于調(diào)試的結(jié)構(gòu)成員*/
};
讀/寫信號(hào)量
讀/寫信號(hào)量適于在讀多寫少的情況下使用。如果一個(gè)任務(wù)需要讀和寫操作時(shí)，它將被看作寫者，在不需要寫操作的情況下可降級(jí)為讀者。任意多個(gè)讀者可同時(shí)擁有一個(gè)讀/寫信號(hào)量，對(duì)臨界區(qū)代碼進(jìn)行操作。
在沒有寫者操作時(shí)，任何讀者都可成功獲得讀/寫信號(hào)量進(jìn)行讀操作。如果有寫者在操作時(shí)，讀者必須被掛起等待直到寫者釋放該信號(hào)量。在沒有寫者或讀者操作時(shí)，寫者必須等待前面的寫者或讀者釋放該信號(hào)量后，才能訪問臨界區(qū)。寫者獨(dú)占臨界區(qū)，排斥其他的寫者和讀者，而讀者只排斥寫者。
讀/寫信號(hào)量可通過依賴硬件架構(gòu)或純軟件代碼兩種方式實(shí)現(xiàn)。下面只說(shuō)明純軟件代碼實(shí)現(xiàn)方式。
（1）API說(shuō)明
用戶可通過調(diào)用讀/寫信號(hào)量API實(shí)現(xiàn)讀/寫操作的同步。讀/寫信號(hào)量API說(shuō)明如表1。
表1 讀/寫信號(hào)量API函數(shù)功能說(shuō)明
| API函數(shù)定義 |
功能說(shuō)明 |
| DECLARE_RWSEM(name) |
聲明名為name的讀寫信號(hào)量，并初始化它。 |
| void init_rwsem(struct rw_semaphore *sem); |
對(duì)讀寫信號(hào)量sem進(jìn)行初始化。 |
| void down_read(struct rw_semaphore *sem); |
讀者用來(lái)獲取sem，若沒獲得時(shí)，則調(diào)用者睡眠等待。 |
| void up_read(struct rw_semaphore *sem); |
讀者釋放sem。 |
| int down_read_trylock(struct rw_semaphore *sem); |
讀者嘗試獲取sem，如果獲得返回1，如果沒有獲得返回0?？稍谥袛嗌舷挛氖褂谩? |
| void down_write(struct rw_semaphore *sem); |
寫者用來(lái)獲取sem，若沒獲得時(shí)，則調(diào)用者睡眠等待。 |
| int down_write_trylock(struct rw_semaphore *sem); |
寫者嘗試獲取sem，如果獲得返回1，如果沒有獲得返回0?？稍谥袛嗌舷挛氖褂? |
| void up_write(struct rw_semaphore *sem); |
寫者釋放sem。 |
| void downgrade_write(struct rw_semaphore *sem); |
把寫者降級(jí)為讀者。 |
（2）讀/寫信號(hào)量結(jié)構(gòu)rw_semaphore
讀/寫信號(hào)量結(jié)構(gòu)rw_semaphore描述了讀/寫信號(hào)量的值和等待隊(duì)列，其列出如下（在include/linux/rwsem-spinlock.h中）：
struct rw_semaphore {
/*讀/寫信號(hào)量定義：
* - 如果activity為0，那么沒有激活的讀者或?qū)懻摺? * - 如果activity為+ve，那么將有ve個(gè)激活的讀者。
* - 如果activity為-1，那么將有1個(gè)激活的寫者。 */
__s32 activity; /*信號(hào)量值*/
spinlock_t wait_lock; /*用于鎖等待隊(duì)列wait_list*/
struct list_head wait_list; /*如果非空，表示有進(jìn)程等待該信號(hào)量*/
#ifdef CONFIG_DEBUG_LOCK_ALLOC /*用于鎖調(diào)試*/
struct lockdep_map dep_map;
#endif
};
（3）讀者加鎖/解鎖操作實(shí)現(xiàn)分析
加讀者鎖操作
讀者加鎖函數(shù)down_read用于加讀者鎖，如果沒有寫者操作時(shí)，等待隊(duì)列為空，讀者可以加讀者鎖，將信號(hào)量的讀者計(jì)數(shù)加1。如果有寫在操作時(shí)，等待隊(duì)列非空，讀者需要等待寫者操作完成。函數(shù)down_read列出如下（在kernel/rwsem.c中）：
void __sched down_read(struct rw_semaphore *sem)
{
might_sleep(); /*用于調(diào)試自旋鎖睡眠*/
rwsem_acquire_read(&sem->dep_map, 0, 0, _RET_IP_); /*確認(rèn)獲得鎖，用于調(diào)試*/
/*跟蹤鎖狀態(tài)信息（如：鎖深度），用于調(diào)試*/
LOCK_CONTENDED(sem, __down_read_trylock, __down_read);
}
函數(shù)__down_read 完成加讀者的具體操作，其列出如下（在lib/rwsem-spinlock.c中）：
void __sched __down_read(struct rw_semaphore *sem)
{
struct rwsem_waiter waiter;
struct task_struct *tsk;
spin_lock_irq(&sem->wait_lock);
/*如果有0或多個(gè)讀者，并且等待隊(duì)列為空，就可以獲取sem*/
if (sem->activity >= 0 && list_empty(&sem->wait_list)) {
/* 獲得sem */
sem->activity++; /*讀者計(jì)數(shù)加1*/
spin_unlock_irq(&sem->wait_lock);
goto out;
}
/*運(yùn)行到這里，說(shuō)明不能獲取sem，將當(dāng)前進(jìn)程加入等待隊(duì)列進(jìn)行等待*/
tsk = current;
set_task_state(tsk, TASK_UNINTERRUPTIBLE);
/* 建立等待隊(duì)列成員*/
waiter.task = tsk;
waiter.flags = RWSEM_WAITING_FOR_READ; /*表示等待讀操作*/
get_task_struct(tsk); /*進(jìn)程使用計(jì)數(shù)加1*/
list_add_tail(&waiter.list, &sem->wait_list); /*將等待成員加到等待隊(duì)列尾*/
/* 不再需要訪問等待隊(duì)列，因此，這里解鎖*/
spin_unlock_irq(&sem->wait_lock);
/* 讀者等待獲取sem */
for (;;) {
if (!waiter.task)
break;
schedule();
set_task_state(tsk, TASK_UNINTERRUPTIBLE);
}
/*運(yùn)行這里，退出等待，說(shuō)明可以獲取sem了*/
tsk->state = TASK_RUNNING;
out:
;
}
解讀者鎖操作
函數(shù)up_read釋放讀者鎖，如果等待隊(duì)列非空，說(shuō)明有寫者在等待，就從等待隊(duì)列喚醒一個(gè)寫者。其列出如下（在kernel/rwsem.c中）：
void up_read(struct rw_semaphore *sem)
{
rwsem_release(&sem->dep_map, 1, _RET_IP_); /*獲取解鎖信息，用于調(diào)試*/
__up_read(sem);
}
函數(shù)__up_read是釋放讀者鎖的具體操作函數(shù)，其列出如下：
void __up_read(struct rw_semaphore *sem)
{
unsigned long flags;
spin_lock_irqsave(&sem->wait_lock, flags);
/*如果所有讀者完成讀操作，并且有寫者等待，那么喚醒一個(gè)寫者*/
if (--sem->activity == 0 && !list_empty(&sem->wait_list))
sem = __rwsem_wake_one_writer(sem);
spin_unlock_irqrestore(&sem->wait_lock, flags);
}
/*喚醒一個(gè)寫者*/
static inline struct rw_semaphore *
__rwsem_wake_one_writer(struct rw_semaphore *sem)
{
struct rwsem_waiter *waiter;
struct task_struct *tsk;
sem->activity = -1; /*表示有一個(gè)寫者正在寫操作*/
/*獲取一個(gè)等待者*/
waiter = list_entry(sem->wait_list.next, struct rwsem_waiter, list);
list_del(&waiter->list); /*將該等待者從等待隊(duì)列刪除*/
tsk = waiter->task;
smp_mb(); /*加內(nèi)存屏障，確保完成上面的指針引用操作*/
waiter->task = NULL;
wake_up_process(tsk); /*喚醒進(jìn)程*/
put_task_struct(tsk); /*進(jìn)程上下文使用計(jì)數(shù)減1*/
return sem;
}
（3）寫者加鎖/解鎖操作實(shí)現(xiàn)分析
加寫者鎖操作
函數(shù)down_write完成加寫者鎖操作，其列出如下：
void __sched down_write(struct rw_semaphore *sem)
{
might_sleep();
rwsem_acquire(&sem->dep_map, 0, 0, _RET_IP_);
LOCK_CONTENDED(sem, __down_write_trylock, __down_write);
}
void __sched __down_write(struct rw_semaphore *sem)
{
__down_write_nested(sem, 0);
}
函數(shù)__down_write_nested完成加寫者鎖的具體操作。當(dāng)沒有讀者或?qū)懻卟僮鲿r(shí)，寫者才可以獲取寫者鎖。寫者鎖是獨(dú)占的。如果有其他寫者或讀者操作時(shí)，寫者必須等待。其列出如下：
void __sched __down_write_nested(struct rw_semaphore *sem, int subclass)
{
struct rwsem_waiter waiter;
struct task_struct *tsk;
spin_lock_irq(&sem->wait_lock);
/*如果沒有讀者，并且等待隊(duì)列為空（說(shuō)明沒有寫者）時(shí)，寫者才能獲取寫者鎖*/
if (sem->activity == 0 && list_empty(&sem->wait_list)) {
/* 獲取寫者鎖*/
sem->activity = -1;
spin_unlock_irq(&sem->wait_lock);
goto out;
}
/*運(yùn)行到這里，說(shuō)明有讀者或?qū)懻咴诓僮?，需要等?/
tsk = current;
set_task_state(tsk, TASK_UNINTERRUPTIBLE);
/* 建立等待隊(duì)列成員*/
waiter.task = tsk;
waiter.flags = RWSEM_WAITING_FOR_WRITE; /*標(biāo)識(shí)為等待寫操作*/
get_task_struct(tsk); /*進(jìn)程上下文使用計(jì)數(shù)加1*/
list_add_tail(&waiter.list, &sem->wait_list); /*加到等待隊(duì)列尾*/
spin_unlock_irq(&sem->wait_lock);
/* 進(jìn)行等待*/
for (;;) {
if (!waiter.task)
break;
schedule();
set_task_state(tsk, TASK_UNINTERRUPTIBLE);
}
/*被喚醒*/
tsk->state = TASK_RUNNING;
out:
;
}
解寫者鎖操作
函數(shù)up_write釋放寫者鎖，將讀者計(jì)數(shù)設(shè)置為0，其列出如下：
void up_write(struct rw_semaphore *sem)
{
rwsem_release(&sem->dep_map, 1, _RET_IP_);
__up_write(sem);
}
void __up_write(struct rw_semaphore *sem)
{
unsigned long flags;
spin_lock_irqsave(&sem->wait_lock, flags);
sem->activity = 0; /*表示有0個(gè)讀者*/
if (!list_empty(&sem->wait_list))
sem = __rwsem_do_wake(sem, 1); /*喚醒等待者*/
spin_unlock_irqrestore(&sem->wait_lock, flags);
}
|
|
|
