1. 程式人生 > >內核同步機制

內核同步機制

fig 完成 class 加鎖 2.6 線程間通信 非阻塞 隊列 size

內核中用於臨界區保護下的互斥機制,它包括自旋鎖、原子操作和信號量,三者保證了對臨界資源訪問的互斥型。

1.1 內核中的互斥機制

1.1.1 自旋鎖

自旋鎖用在多個CPU系統中。當一個線程在一個CPU上正使用資源,而另一個線程在另一個CPU上正忙等待這個資源的時候,就會用到自旋鎖來保護臨界資源。在單處理器系統中,自旋鎖函數擴展為空。

自旋鎖基於共享變量。函數通過給變量設置一個值來獲得鎖,其他需要鎖的函數就會查詢它,並知道鎖現在不可用,然後在一個忙等待的循環中“自旋”,直到鎖可用為止。

由於使用自旋鎖時,其他CPU被強制等待。因此持有自旋鎖的函數不能花費過長時間。

下面說明操作自旋鎖的宏:

  • spin_loc(spinlock_t *lock):獲得給定的鎖,直到鎖成為可用狀態為止。在spin_lock返回之後,調用函數將擁有該鎖。
  • spin_lock_irq(spinlock_t *lock):類似spin_lock_irqsave,只是不保存當前的中斷狀態。
  • spin_lock_bh(spinlock_t *lock):獲得給定的鎖並且阻止底半部的執行。
  • spin_unlock(spinlock_t *lock):此宏開鎖,它與前面加鎖的宏是配對使用的。spin_unlock解開給定的鎖而不做其他的工作。
  • spin_unlock_irq(spinlock_t *lock):此宏開鎖,它與前面加鎖的宏是配對使用的。spin_unlock_irq無條件地啟動中斷。
  • spin_unlock_bh(spinlock_t *lock):此宏開鎖,它與前面加鎖的宏是配對使用的。spin_unlock_bh重新啟動底半部處理。

linux還有另外一種類型的自旋鎖,稱為“讀者/寫者自旋鎖”。讀者/寫者問題,即如果有多個線程(進程、中斷處理程序、底半部例程)需要以只讀的方式訪問一個臨界區數據,眾多的讀者之間不會彼此幹預,而只有寫者之間會產生競爭。

自旋鎖它是為為實現保護共享資源而提出一種鎖機制。其實,自旋鎖與互斥鎖比較類似,它們都是為了解決對某項資源的互斥使用。無論是互斥鎖,還是自旋鎖,在任何時刻,最多只能有一個保持者,也就說,在任何時刻最多只能有一個執行單元獲得鎖。但是兩者在調度機制上略有不同。對於互斥鎖,如果資源已經被占用,資源申請者只能進入睡眠狀態但是自旋鎖不會引起調用者睡眠,如果自旋鎖已經被別的執行單元保持,調用者就一直循環在那裏看是否該自旋鎖的保持者已經釋放了鎖

,"自旋"一詞就是因此而得名。

自旋鎖一般原理

跟互斥鎖一樣,一個執行單元要想訪問被自旋鎖保護的共享資源,必須先得到鎖,在訪問完共享資源後,必須釋放鎖。如果在獲取自旋鎖時,沒有任何執行單元保持該鎖,那麽將立即得到鎖;如果在獲取自旋鎖時鎖已經有保持者,那麽獲取鎖操作將自旋在那裏,直到該自旋鎖的保持者釋放了鎖。由此我們可以看出,自旋鎖是一種比較低級的保護數據結構或代碼片段的原始方式,這種鎖可能存在兩個問題:死鎖和過多占用cpu資源

自旋鎖適用情況

自旋鎖比較適用於鎖使用者保持鎖時間比較短的情況。正是由於自旋鎖使用者一般保持鎖時間非常短,因此選擇自旋而不是睡眠是非常必要的,自旋鎖的效率遠高於互斥鎖。信號量和讀寫信號量適合於保持時間較長的情況,它們會導致調用者睡眠,因此只能在進程上下文使用,而自旋鎖適合於保持時間非常短的情況,它可以在任何上下文使用。如果被保護的共享資源只在進程上下文訪問,使用信號量保護該共享資源非常合適,如果對共享資源的訪問時間非常短,自旋鎖也可以。但是如果被保護的共享資源需要在中斷上下文訪問(包括底半部即中斷處理句柄和頂半部即軟中斷),就必須使用自旋鎖。自旋鎖保持期間是搶占失效的,而信號量和讀寫信號量保持期間是可以被搶占的。自旋鎖只有在內核可搶占或SMP(多處理器)的情況下才真正需要,在單CPU且不可搶占的內核下,自旋鎖的所有操作都是空操作。另外格外註意一點:自旋鎖不能遞歸使用

關於自旋鎖的定義以及相應的API

自旋鎖定義: linux/Spinlock.h

技術分享圖片
typedef struct spinlock {
          union { //聯合
             struct raw_spinlock rlock;
#ifdef CONFIG_DEBUG_LOCK_ALLOC
# define LOCK_PADSIZE (offsetof(struct raw_spinlock, dep_map))
             struct{
                     u8 __padding[LOCK_PADSIZE];
                     struct lockdep_map dep_map;
             };
#endif
         };
} spinlock_t;
技術分享圖片

定義和初始化

spinlock_t my_lock = SPIN_LOCK_UNLOCKED; 
void spin_lock_init(spinlock_t *lock); 

自旋鎖操作:

技術分享圖片
//加鎖一個自旋鎖函數
void spin_lock(spinlock_t *lock);                                   //獲取指定的自旋鎖
void spin_lock_irq(spinlock_t *lock);                               //禁止本地中斷獲取指定的鎖
void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);      //保存本地中斷的狀態,禁止本地中斷,並獲取指定的鎖
void spin_lock_bh(spinlock_t *lock)                                 //安全地避免死鎖, 而仍然允許硬件中斷被服務


//釋放一個自旋鎖函數
void spin_unlock(spinlock_t *lock);                                 //釋放指定的鎖
void spin_unlock_irq(spinlock_t *lock);                             //釋放指定的鎖,並激活本地中斷
void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags); //釋放指定的鎖,並讓本地中斷恢復到以前的狀態
void spin_unlock_bh(spinlock_t *lock);                              //對應於spin_lock_bh


//非阻塞鎖
int spin_trylock(spinlock_t *lock);                  //試圖獲得某個特定的自旋鎖,如果該鎖已經被爭用,該方法會立刻返回一個非0值,
                                                     //而不會自旋等待鎖被釋放,如果成果獲得了這個鎖,那麽就返回0.
int spin_trylock_bh(spinlock_t *lock);                           
//這些函數成功時返回非零( 獲得了鎖 ), 否則 0. 沒有"try"版本來禁止中斷.

//其他
int spin_is_locked(spinlock_t *lock);               //和try_lock()差不多
技術分享圖片

1.2 原子操作

原子操作指某些操作的執行不可中斷。原子操作分為bitops和atomic_兩類。在原子操作中,常遇到聲明volatile。將變量聲明為volatile時,系統會阻止編譯器對給值進行優化,確保變量使用了用戶定義的精確地址,而不是裝有同一信息的一些別名。

bitops原子操作方式是在一些標誌的設置需要進行原子操作的情況下使用的。原子的位操作是非常快的,使用單條機器指令來完成操作。

atomic_t原子操作方式用於加減之類的運算,這個操作是原子性的,它用單條機器指令來完成操作。

1.3 信號量

進程間對共享資源的互斥訪問是通過信號量機制來實現的。內核中提供了函數down和函數up對信號量進行操作。

信號量和自旋鎖有一定的區別,用信號量的down操作如果無法得到資源,那就會進入等待隊列,通過調度去運行其他進程。而在自旋鎖中,如果無法得到資源,將進入忙等待,直到得到資源。因此,如果資源被占用時間很短,則使用自旋鎖較好,因為它可節約調度時間。如果資源被占用的時間較長,使用信號量較好,因為可讓CPU調度去做其他進程的工作。

信號量的實現包括信號量的初始化及函數up和函數down的實現。

1.3.1信號量

信號量又稱為信號燈,它是用來協調不同進程間的數據對象的,而最主要的應用是共享內存方式的進程間通信。本質上,信號量是一個計數器,它用來記錄對某個資源(如共享內存)的存取狀況。一般說來,為了獲得共享資源,進程需要執行下列操作:
   (1) 測試控制該資源的信號量。
   (2) 若此信號量的值為正,則允許進行使用該資源。進程將信號量減1。
   (3) 若此信號量為0,則該資源目前不可用,進程進入睡眠狀態,直至信號量值大於0,進程被喚醒,轉入步驟(1)。
   (4) 當進程不再使用一個信號量控制的資源時,信號量值加1。如果此時有進程正在睡眠等待此信號量,則喚醒此進程。
    維護信號量狀態的是Linux內核操作系統而不是用戶進程。我們可以從頭文件/usr/src/linux/include/linux/sem.h 中看到內核用來維護信號量狀態的各個結構的定義。信號量是一個數據集合,用戶可以單獨使用這一集合的每個元素。要調用的第一個函數是semget,用以獲得一個信號量ID。Linux2.6.26下定義的信號量結構體:

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

從以上信號量的定義中,可以看到信號量底層使用到了spin lock的鎖定機制,這個spinlock主要用來確保對count成員的原子性的操作(count--)和測試(count > 0)。

1.信號量的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),所以從實用角度,徹底忘了它吧。

(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的運行隊列中摘下)

(3)試圖去獲得一個信號量,如果沒有獲得,函數立刻返回1而不會讓當前進程進入睡眠狀態。

2.信號量的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鏈表中的第一個有效節點,並喚醒睡眠在該節點上的進程。

關於Linux環境下信號量其他API 詳見LKD和ULD

1.4 互斥體

互斥體實現了“互相排斥”(mutual exclusion)同步的簡單形式(所以名為互斥體(mutex))。互斥體禁止多個線程同時進入受保護的代碼“臨界區”(critical section)。因此,在任意時刻,只有一個線程被允許進入這樣的代碼保護區。
  任何線程在進入臨界區之前,必須獲取(acquire)與此區域相關聯的互斥體的所有權。如果已有另一線程擁有了臨界區的互斥體,其他線程就不能再進入其中。這些線程必須等待,直到當前的屬主線程釋放(release)該互斥體。
  什麽時候需要使用互斥體呢?互斥體用於保護共享的易變代碼,也就是,全局或靜態數據。這樣的數據必須通過互斥體進行保護,以防止它們在多個線程同時訪問時損壞

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
};
技術分享圖片

對比前面的struct semaphore,struct mutex除了增加了幾個作為debug用途的成員變量外,和semaphore幾乎長得一樣。但是mutex的引入主要是為了提供互斥機制,以避免多個進程同時在一個臨界區中運行。

如果靜態聲明一個count=1的semaphore變量,可以使用DECLARE_MUTEX(name),DECLARE_MUTEX(name)實際上是定義一個semaphore,所以它的使用應該對應信號量的P,V函數.

如果要定義一個靜態mutex型變量,應該使用DEFINE_MUTEX

如果在程序運行期要初始化一個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以期獲得最佳的執行性能。這裏對於mutex的實現細節,不再多說,如欲深入了解,參考APUE和ULD

1.5 信號量、互斥體和自旋鎖的區別

信號量/互斥體和自旋鎖的區別

信號量/互斥體允許進程睡眠屬於睡眠鎖,自旋鎖則不允許調用者睡眠,而是讓其循環等待,所以有以下區別應用
1)、信號量和讀寫信號量適合於保持時間較長的情況,它們會導致調用者睡眠,因而自旋鎖適合於保持時間非常短的情況
2)、自旋鎖可以用於中斷,不能用於進程上下文(會引起死鎖)。而信號量不允許使用在中斷中,而可以用於進程上下文
3)、自旋鎖保持期間是搶占失效的,自旋鎖被持有時,內核不能被搶占,而信號量和讀寫信號量保持期間是可以被搶占的

另外需要註意的是
1)、信號量鎖保護的臨界區可包含可能引起阻塞的代碼,而自旋鎖則絕對要避免用來保護包含這樣代碼的臨界區,因為阻塞意味著要進行進程的切換,如果進程被切換出去後,另一進程企圖獲取本自旋鎖,死鎖就會發生。
2)、在你占用信號量的同時不能占用自旋鎖,因為在你等待信號量時可能會睡眠,而在持有自旋鎖時是不允許睡眠的。

信號量和互斥體之間的區別

概念上的區別:

信號量:是進程間(線程間)同步用的,一個進程(線程)完成了某一個動作就通過信號量告訴別的進程(線程),別的進程(線程)再進行某些動作。有二值和多值信號量之分。

互斥鎖:是線程間互斥用的,一個線程占用了某一個共享資源,那麽別的線程就無法訪問,直到這個線程離開,其他的線程才開始可以使用這個共享資源。可以把互斥鎖看成二值信號量。

上鎖時:

信號量: 只要信號量的value大於0,其他線程就可以sem_wait成功,成功後信號量的value減一。若value值不大於0,則sem_wait阻塞,直到sem_post釋放後value值加一。一句話,信號量的value>=0。

互斥鎖: 只要被鎖住,其他任何線程都不可以訪問被保護的資源。如果沒有鎖,獲得資源成功,否則進行阻塞等待資源可用。一句話,線程互斥鎖的vlaue可以為負數。

使用場所:

信號量主要適用於進程間通信,當然,也可用於線程間通信。而互斥鎖只能用於線程間通信。

內核同步機制