《Linux4.0設備驅動開發詳解》筆記--第七章：Linux設備中的並發控制

7.1 並發與競態

並發是指多個執行單元同時、並發的被執行，而並發的執行單元對共享資源（硬件資源、軟件上的的全局變量、靜態變量等）的訪問則很容易導致竟態 竟態發生在以下幾種情況
對稱多處理器（SMP）的多個CPU 單CPU內進程與搶占它的進程 中斷與進程 解決方法是保證對共享資源的互斥訪問
訪問共享資源的代碼區域稱為臨界區，臨界區需要以某種互斥機制保護
互斥途徑：中斷屏蔽、原子操作、信號量、自旋鎖、互斥體等

7.2 編譯亂序和執行亂序

編譯亂序是編譯器的問題，而執行亂序是處理器運行時的行為 編譯屏障：在代碼中設置barrier()屏障來阻擋編譯器優化 內存屏障指令：解決多核間一個核的內存行為對另一個核可見的問題
arm處理器的內存屏障指令有
DMB：數據內存屏障 DSB：數據同步屏障 LSB：指令同步屏障 linux的自旋鎖、互斥量的等互斥邏輯需要用到上述指令

7.3 中斷屏蔽

中斷屏蔽的使用方法
local_irp_disable()與local_irp_enable()只能禁止或是能本CPU內的中斷 local_irp_save(flags)與local_irp_restore(flags)：處理屏蔽或使能中斷還能保存目前CPU的中斷信息，對arm就是保存和恢復cpsr local_bh_disable()和loacal_bh_diable()：禁止和使能中斷的底半部

local_irp_disable();
...
critical section /*臨界區*/
...
local_irp_enable();

7.4 原子操作

原子操作可以保證對一個整型數據的修改是排他性的 原子操作函數分為對整型和位的原子操作

7.4.1 對整型的原子操作

設置原子變量的值

//設置原子變量的值為i
void atomic_set(atomic_t *v, int i);
//定義原子變量v並初始化為0
atomic_t v = ATOMIC_INIT(0);

獲取原子變量的值

//返回原子變量的值
atomic_read(atomic_t *v);

原子變量加、減

//原子變量增加1
void atomic_sub(int i, atomic_t *v);

原子變量自增、自減

//原子變量增加1
void atomic_inc(atomic_t *v);
//原子變量減少1
void atomic_dec(atomic_t *v);

操作並測試，下屬操作對原子變量執行自增、自檢和減操作後（沒有加操作）。測試其是否為0，位0返回true，否則返回false

int atomic_inc_and_test(atomic_t *v);
int atomic_dec_and_test(atomic_t *v);
int atomic_sub_and_test(int i, atomic_t *v);

操作並返回，丅述操作是對原子變量的加、減和自增、自減操作，並返回新的值

int atomic_add_return(int i, atomic_t *v);
int atomic_sub_return(int i, atomic_t *v);
int atomic_intc_return(atomic_t *v);
int atomic_dec_return(atomic_t *v);

7.4.2 位原子操作

設置位，丅述設置addr地址的底nr位

void set_bit(nr,void *addr);

清除位

void change_bit(nr, void *addr);

改變位，下述代碼對addr的nr位進行反置

void change_bit(nr, void *addr);

測試位，返回addr的第nr位

test_bit(nr, void *addr)

測試並操作位

int test_and_se_bit(nr, void *addr);
int test_and_clear_bit(nr, void *addr);
int test_and_change_bit(nr, void *addr);

例如：使用原子變量使得文字只能被一個進程打開

static atomic_t xxx_available = ATOMIC_INIT(1);

static int xxx_open(struct inode *inode, struct file *filp)
{
    ...
    if(!atomic_dev_and_test(&xxx_available)){
        atomic_inc(&xxx_available);
        return -EBUSY;//已打開
    }
    ...
    return 0;//成功
}

static int xxx_release(struct inode* inode, struct file *filp){
        atomic_inc(&xxx_available);//釋放設備
        return 0;
    }

7.5 自旋鎖

7.5.1 自旋鎖的使用

自旋鎖是一種典型的對臨界資源進行互斥訪問的手段，其名稱來源於他的工作方式 為了獲取自旋鎖，在某CPU上的運行需要先執行一個原子操作，該操作測試並設置某個內存變量。
如果測試結果表明鎖已經空閒，則程序成功獲取自旋鎖並繼續執行 如果測試結果表明鎖仍在被使用，則程序將在一個小的循環內重復這個“測試並設置”操作，即所謂的自旋 linux系統中對自旋鎖的相關操作
定義自旋鎖：spinlock_t lock; 初始化自旋鎖：spin_lock_init(lock); 獲得自旋鎖：spin_lock(lock);或者spin_locktry(lock); 釋放自旋鎖：spin_unlock(lock); 自旋鎖主要針對SMP或者單CPU單內核可搶占的情況 對於單CPU但是內核不支持搶占的情況，自旋鎖自動轉化為空操作
單CPU系統和可搶占的系統中，自旋鎖持有期間中內核的搶占將被禁止 多核SMP的情況下，任何一個核拿到了自旋鎖，該核上的搶占調度也暫時禁止，但是沒有禁止另外一個核的搶占調度 自旋鎖的臨界區可以被中斷或者底半部影響，因而需要自旋鎖和其他操作的配合
spin_lock_irp() = spin_lock() + local_irq_disable() spin_unlock_irp = spin_unlock() +local_irq_enable(); spin_lock_irqsave() = spin_lock() + local_irq_save(); spin_unlock_irqrestore() = spin_unlock() + local_irq_restore() spin_lock_bh() = spin_lock() + local_bh_disable() spin_unlock_bh() = spin_unlock() + local_bh_enable() 使用自旋鎖應該謹慎，同時需要注意幾個問題
自旋鎖是忙等待，在自旋的時候CPU不做其他的操作，因此只有在占用鎖很短的情況下才使用自旋鎖 可能因為遞歸的使用而使系統死鎖 在自旋期間，不能調用調用可能引起進程調度的函數，可能導致內核崩潰 在單核情況下，也該認為自己的CPU是多核的，以突出驅動的跨平台性

7.5.2 讀寫自旋鎖

自旋鎖不關心臨界區在進行什麼操作，一視同仁 自旋鎖的衍生鎖讀寫自旋鎖允許讀的並發，它保留了自選的概念，在寫方面只能有一個進程，讀方面可以有多個執行單元，但是讀和寫不能同時進行 定義並初始化讀寫自旋鎖:

rwlock_t my_rwlock1 = RW_LOCK_UNLOCKED; //靜態初始化
rwlock_t my_rwlock2;
rwlock_init(&my_rwlock2); //動態初始化

讀鎖定:

void read_lock(rwlock_t* lock);
void read_lock_irqsave(rwlock_t* lock, unsigned long flags);
void read_lock_irq(rwlock_t* lock);
void read_lock_bh(rwlock_t* lock);

讀解鎖:

void read_unlock(rwlock_t* lock);
void read_unlock_irqrestore(rwlock_t* lock, unsigned long flags);
void read_unlock_irq(rwlock_t* lock);
void read_unlock_bh(rwlock_t* lock);

在對共享資源進行讀取之前,應該先調用讀鎖定函數鎖定共享資源,完成之後再調用讀解鎖函數釋放共享資源;

寫鎖定:

void write_lock(rwlock_t* lock);
void write_lock_irqsave(rwlock_t* lock, unsigned long flags);
void write_lock_irq(rwlock_t* lock);
void write_lock_bh(rwlock_t* lock);
void write_trylock(rwlock_t* lock);

寫解鎖:

void write_unlock(rwlock_t* lock);
void write_unlock_irqrestore(rwlock_t* lock);
void write_unlock_irq(rwlock_t* lock);
void write_unlock_bh(rwlock_t* lock);

 - 在對共享資源進行寫操作之前,應該先調用寫鎖定函數鎖定共享資源,完成之後再調用寫解鎖函數釋放共享資源;與spin_trylock()一樣,write_trylock()也只是嘗試獲得寫自旋鎖,不管是否成功,都會立即返回;

讀寫自旋鎖使用套路:

rwlock_t lock;      //定義讀寫自旋鎖
rwlock_init(&lock); //初始化讀寫自旋鎖

read_lock(&lock);   //讀時加鎖
......
//臨界區操作
......
read_unlock(&lock); //讀後解鎖;

write_lock_irqsave(&lock, flags);    //寫時加鎖
......
//臨界區操作
......
write_lock_irqrestore(&lock, flags); //寫後解鎖;

7.5.3 順序鎖

順序鎖相關操作：

void write_seqlock(seqlock_t *s1);
void write_sequnlock(seqlock_t *s1);

//宏調用，相當於：write_seqlock()+local_irq_save()
write_seqlock_irqsave(lock,flags)
write_sequnlock_irqrestore(lock,flags)

//宏調用，相當於：write_seqlock()+local_irq+disable()
write_seqlock_irq(lock)
write_sequnlock_irq(lock)

//宏調用，相當於：write_seqlock()+local_bh_disable()
write_seqlock_bh(lock)
write_sequnlock_bh(lock)

int write_tryseqlock(seqlock_t *s1)，此函數和上面提到的類似。

寫執行單元使用如下一種模式的順序鎖：

write_seqlock(&seqlock);
.........//寫操作代碼塊
write_sequnlock(&seqlock);

讀執行單元涉及如下順序鎖操作：

讀開始：unsigned read_seqbegin(const seqlock_t *s1);
//讀執行單元在訪問共享資源時要，調用該函數，返回鎖s1的順序號
read_seqbegin_irqsave(lock,flags)  
//等同於：local_irq_save()+read_seqbegin()
重讀：int read_seqretry(const seqlock_t *s1,unsigned iv) 
//在讀結束後調用此函數來檢查，是否有寫執行單元對資源進行操作，若有，則重新讀。iv 為鎖的順序號。

7.5.4 讀-復制-更新（RCU）

RCU的讀端沒有鎖、內存屏蔽、原子指令類的開銷，機會可以認為是直接讀 RCU的寫端訪問它的共享資源前首先要復制一個副本，然後對副本進行修改，然後使用一個回調機制在適當的時機把指向原來數據的指針重新指向新的修改數據
這個時機就是所有引用該數據的CPU都退出對共享數據讀操作的時候 等待適當時機的這一時期稱為寬限期 RCU的有點
允許多個讀執行單元同時訪問保護區 允許多個讀寫執行單元同時訪問保護區數據 RCU不能替代讀寫鎖
因為RCU的寫執行單元的同步開銷較大，當寫較多的時候，對讀執行單元的性能提高不能彌補寫執行單元同步導致的損失

7.6 信號量

定義信號量

struct semaphore sem;

初始化：

void sema_init (struct semaphore *sem, int val);
void init_MUTEX (struct semaphore *sem); //將sem的值置為1，表示資源空閒
void init_MUTEX_LOCKED (struct semaphore *sem); //將sem的值置為0，表示資源忙

申請內核信號量所保護的資源：

//可引起睡眠
void down(struct semaphore * sem); 
//down_interruptible能被信號打斷
int down_interruptible(struct semaphore * sem);  
// 非阻塞函數，不會睡眠。無法鎖定資源則馬上返回
int down_trylock(struct semaphore * sem); 

釋放內核信號量所保護的資源：

void up(struct semaphore * sem);

7.7 互斥體

互斥體提供了兩種機制：經典互斥體和實時互斥體 經典互斥體結構體： （會導致無限制優先級反轉問題）

struct mutex {
    /* 1: unlocked, 0: locked, negative: locked, possible waiters */
    atomic_t  count;
    spinlock_t
    wait_lock;
    struct list_head
    wait_list;
    #if defined(CONFIG_DEBUG_MUTEXES) || defined(CONFIG_SMP)
    struct task_struct
    *owner;
    #endif
    #ifdef CONFIG_DEBUG_MUTEXES
    const char 
    *name;
    void  *magic;
    #endif
    #ifdef CONFIG_DEBUG_LOCK_ALLOC
    struct lockdep_map
    dep_map;
    #endif
};

實時互斥體結構體：

struct rt_mutex {
    raw_spinlock_t
    wait_lock;
    struct plist_head
    wait_list;
    struct task_struct
    *owner;
    #ifdef CONFIG_DEBUG_RT_MUTEXES
    int  save_state;
    const char 
    *name, *file;
    int  line;
    void  *magic;
    #endif
};

操作：

struct mutex my_mutex;    
mutex_init(&my_mutex);   
void mutex_lock(struct mutex* lock);                       //獲取互斥體,不可被信號中斷
void mutex_lock_interruptible(struct mutex* lock);     //獲取互斥體,可被信號打斷
int mutex_trylock(struct mutex* lock);                     //嘗試獲取互斥體
void mutex_unlock(struct mutex* lock);                     //釋放互斥體

int mutex_is_locked(struct mutex* lock):
//該函數檢查互斥鎖lock是否處於鎖定狀態。返回1,表示已鎖定;返回0,表示未鎖定;
int mutex_lock_interruptible(struct mutex* lock);  //該函數可被信號打斷
int mutex_lock_killable(struct mutex* lock);       //該函數可被kill信號打斷

用例：

struct mutex my_mutex;   
mutex_init(&my_mutex);   

mutex_lock(&my_mutex);  
...臨界區...                 
mutex_unlock(&my_mutex); 

7.8 完成量

完成量：表示一個執行單元需要等待另一個執行單元完成某事後方可執行。
它是一種輕量級機制，為了完成進程間的同步而設計 使用完成量等待時，調用進程是以獨占睡眠方式進行等待的 不是忙等待 結構體
done變量是完成量要保護的對象 wait則是申請完成量的進程等待隊列

struct completion {
    unsigned int done;
    wait_queue_head_t wait;
};

初始化函數
done變量被初始化為0. 內核代碼中wait_for_common函數其實就是對done變量作判斷，若done變量沒有大於0，則它一直處於while循環中。 complete函數就是對done變量加1。wait_for_common函數便會退出while循環，同時將done減1，表示申請完成量成功。

static inline void init_completion(struct completion *x)
{
x->done = 0;
init_waitqueue_head(&x->wait);
}

操作：

struct completion my_completion;    //定義完成量my_completion

init_completion(&my_completion);    //初始化完成量my_completion

void wait_for_completion(struct completion* comp)
//該函數等待一個完成量被喚醒。該函數會阻塞調用進程,如果所等待的完成量沒有被喚醒,那就一直阻塞下去,而且不會被信號打斷;

int wait_for_completion_interruptible(struct completion* comp)
//該函數等待一個完成量被喚醒。但是它可以被外部信號打斷;

int wait_for_completion_killable(struct completion* comp)
//該函數等待一個完成量被喚醒。但是它可以被kill信號打斷;

unsigned long wait_for_completion_timeout(struct completion* comp, unsigned long timeout)
//該函數等待一個完成量被喚醒。該函數會阻塞調用進程,如果所等待的完成量沒有被喚醒,調用進程也不會一直阻塞下去,而是等待一個指定的超時時間timeout,當超時時間到達時,如果所等待的完成量仍然沒有被喚醒,那就返回;超時時間timeout以系統的時鐘滴答次數jiffies計算

bool try_wait_for_completion(struct completion* comp)
//該函數嘗試等待一個完成量被喚醒。不管所等待的完成量是否被喚醒,該函數都會立即返回

bool completion_done(struct completion* comp)
//該函數用於檢查是否有執行單元阻塞在完成量comp上(是否已經完成),返回0,表示有執行單元被完成量comp阻塞;相當於wait_for_completion_timeout()中的timeout=0

void complete(struct completion* comp)
//該函數只喚醒一個正在等待完成量comp的執行單元

void complete_all(struct completion* comp)
//該函數喚醒所有正在等待同一個完成量comp的執行單元

NORET_TYPE void complete_and_exit(struct completion* comp, long code)
//該函數喚醒一個正在等待完成量comp的執行單元,並退出,code為退出碼

注意:在內核處理完請求之後,必須調用這三個函數中的一個,來喚醒其它正在等待的進程