之所以寫這篇文章當然還是在面試中涉及了對本文標題的相關問題-互斥鎖和自旋鎖的區別。聽到這個問題的時候,我是比較忐忑的。互斥鎖我還能簡單說一些,但是對於自旋鎖的了解幾乎為零。為此,將總結Linux下的相關鎖-那些“鎖”事兒。知之為知之,不知為不知,是知也。不懂的地方,盡快查漏補缺!
我們曉得在Linux內核中,同步機制是一大特性。比較經典的有原子操作、spin_lock(自旋鎖)、mutex(互斥鎖)、semaphore(信號量)等。
原子操作,也是數據庫事務的一大特性。就是該操作絕不會在執行完之前被任何任務或者事件終止,要不全部執行,要不全部不執行。它是最小的執行單位,原子操作需要硬件的支持,因此是架構相關的。它的API和原子類型定義都在內核源碼樹的include/asm/atomic.h文件中,都是用匯編語言實現。
原子操作主要用於實現資源計數,很多引用計數就是通過原子操作實現的。
原子類型定義:
typedef struct{ volatile int counter; }atomic_t;
volatile關鍵字修飾的字段可以通知gcc不要對該類型的數據做優化處理,對它的訪問都是對內存的訪問,而不是對寄存器的訪問。
原子操作的API:
1)atomic_read(atomic_t *v);
作用:對原子類型的變量進行原子讀操作,它返回原子類型的變量v的值。
2)atomic_set(atomic_t *v,int i);
作用:設置原子類型的變量v的值為i。
3)atomic_add(int i,atomic_t *v);
作用:給原子類型的變量v增加值i。
4)atomic_sub(int i,atomic_t *v);
作用:從原子類型的變量v中減去i。
5)atomic_sub_and_test(int i, atomic_t *v);
作用:從原子類型的變量v中減去i,並判斷結果是否為0,如果為0,返回真,否則返回假。
6)atomic_inc(atomic_t *v);
作用:對原子類型變量v原子地增加1。
7)atomic_dec(atomic_t *v);
作用:對原子類型的變量v原子地減1。
8)atomic_dec_and_test(atomic_t *v);
作用:對原子類型的變量v原子地減1,並判斷結果是否為0,如果為0,返回真,否則返回假。
9)int atomic_inc_and_test(atomic_t *v);
作用:對原子類型的變量v原子地增加1,並判斷結果是否為0,如果為0,返回真,否則返回假。
10)atomic_add_negative(int i, atomic_t*v);
作用:對原子類型的變量v原子地增加i,並判斷結果是否為負數,如果是,返回真,否則返回假。
11)atomic_add_return(int i, atomic_t *v);
作用:對原子類型的變量v原子地增加i,並且返回指向v的指針。
12)int atomic_sub_return(int i, atomic_t *v);
作用:從原子類型的變量v中減去i,並且返回指向v的指針。
13)int atomic_inc_return(atomic_t * v);
作用:對原子類型的變量v原子地增加1並且返回指向v的指針。
14)int atomic_dec_return(atomic_t * v);
作用:對原子類型的變量v原子地減1並且返回指向v的指針。
原子操作通常用於實現資源的引用計數,在TCP/IP協議棧的IP碎片處理中,就使用了引用計數,碎片隊列結構structipq描述了一個IP碎片,字段refcnt就是引用計數器,它的類型為atomic_t,當創建IP碎片時(在函數ip_frag_create中),使用atomic_set函數把它設置為1,當引用該IP碎片時,就使用函數atomic_inc把引用計數加1,當不需要引用該IP碎片時,就使用函數ipq_put來釋放該IP碎片,ipq_put使用函數atomic_dec_and_test把引用計數減1並判斷引用計數是否為0,如果是就釋放Ip碎片。函數ipq_kill把IP碎片從ipq隊列中刪除,並把該刪除的IP碎片的引用計數減1(通過使用函數atomic_dec實現)。
信號量的本質也是一個計數器,用來記錄對某個資源(如共享內存)的存取狀況。用來協調不同進程間的數據對象,最主要的應用是共享內存方式的進程間通信。
一般情況,為了獲取共享資源,進程需要執行如下步驟:
1)測試控制該資源的信號量;
2)如果該信號量為正,就允許使用該資源,進程將型號量減一;
3)如果為0,則該資源目前不可用,進程sleep,知道信號量值大於0,才能被喚醒,從步驟1)開始執行;
4)當進程不再使用某信號量控制的資源時,信號量值加1,。如果此時有進程在sleep並等待此信號量,則可以喚醒該進程。
信號量的定義在頭文件/usr/src/linux/include/linux/sem.h 中,信號量是一個數據集合,用戶可以單獨使用這一集合中的每個元素。
Linux2.6.26下定義的信號量結構體:
struct semaphore { spinlock_t lock; unsigned int count; struct list_head wait_list; };
從以上信號量的定義中,可以看到信號量底層使用到了spinlock的鎖定機制,這個spinlock主要用來確保對count成員的原子性的操作(count--)和測試(count > 0)。
(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而不會讓當前進程進入睡眠狀態。
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鏈表中的第一個有效節點,並喚醒睡眠在該節點上的進程。
1)間接相互制約關系(互斥)
若某一進程要求使用某種資源,而該資源正好被另一進程使用,並且該資源不允許兩個進程同時使用,那麼該進程只好等待已占有的資源的進程釋放資源後再使用。這種制約關系可以用“進程-資源-進程”的形式表示。例如,打印機資源,進程互斥經典問題中生產者-生產者問題。
2)直接相互制約關系(同步)
某一進程若收不到另一進程提供的必要信息就不能繼續運行下去,表明了兩個進程之間在某些點上要交換信息,相互交流運行情況。這種制約關系的進本形式是“進程-進程”。例如生產者與消費者問題,生產者生產產品並放入緩沖池,消費者從緩沖池取走產品進行消費。這兩者就是同步關系。
區分互斥和同步只需記住,同類進程即為互斥關系,不同類進程即為同步關系。
臨界資源:同時只允許一個進程使用的資源。
臨界區:進程中用於訪問臨界資源的代碼段,又稱臨界段。
每個進程的臨界區代碼可以不同,臨界區代碼由於要訪問臨界資源,因此要在進入臨界區之前進行檢查,至於每個進程對臨界資源進行怎樣的操作,這和臨界資源及互斥同步管理是無關的。
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 。
自旋鎖也是實現保護共享資源的一種鎖機制,與互斥鎖比較類似,都是為了解決對某資源的互斥使用。無論是互斥鎖還是自旋鎖,在任何時刻最多只有一個保持者。也就是說,任何時刻最多只有一個執行單元獲得鎖。兩者的不同之處是,對於互斥鎖而言,如果資源已經被占用,其它的資源申請進程只能進入sleep狀態。但是自旋鎖不會引起調用者sleep,如果自旋鎖已經被別的執行單元保持,調用者就一直循環在等待該自旋鎖的保持者是否釋放該鎖。
跟互斥鎖一樣,一個執行單元要想訪問被自旋鎖保護的共享資源,必須先得到鎖,在訪問完共享資源後,必須釋放鎖。如果在獲取自旋鎖時,沒有任何執行單元保持該鎖,那麼將立即得到鎖;如果在獲取自旋鎖時鎖已經有保持者,那麼獲取鎖操作將自旋在那裡,直到該自旋鎖的保持者釋放了鎖。由此我們可以看出,自旋鎖是一種比較低級的保護數據結構或代碼片段的原始方式,這種鎖可能存在兩個問題:死鎖和過多占用cpu資源。
自旋鎖比較適用於鎖使用者保持鎖時間比較短的情況,正是由於自旋鎖使用者一般保持較短的鎖時間,因此選擇自選而不是睡眠是非常必要的,因為自旋鎖的效率遠高於互斥鎖。信號量和讀寫信號量適用於保持時間較長的情況,它們會導致調用者sleep,因此只能在進程上下文使用。而自旋鎖適合於保持時間非常短的情況,它可以再任何上下文使用。如果被保護的共享資源只在進程上下文訪問,使用信號量保護該共享資源非常合適,如果對共享資源的訪問時間非常短,自旋鎖也可以。但是如果被保護的共享資源需要在中斷上下文訪問(包括底半部即中斷處理句柄和頂半部即軟中斷),就必須使用自旋鎖。自旋鎖保持期間是搶占失效的,而信號量和讀寫信號量保持期間是可以被搶占的。自旋鎖只有在內核可搶占或SMP(多處理器)的情況下才真正需要,在單CPU且不可搶占的內核下,自旋鎖的所有操作都是空操作。另外格外注意一點:自旋鎖不能遞歸使用。
自旋鎖定義的文件(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()差不多
信號量。互斥鎖允許進程sleep屬於睡眠鎖,自旋鎖不允許調用者sleep,而是讓其循環等待,所以有以下區別應用:
另外需要注意的是:
信號量和互斥鎖的區別
1、概念上的區別:
信號量:是進程間(線程間)同步用的,一個進程(線程)完成了某一個動作就通過信號量告訴別的進程(線程),別的進程(線程)再進行某些動作。有二值和多值信號量之分;
互斥鎖:是線程間互斥用的,一個線程占用了某一個共享資源,那麼別的線程就無法訪問,直到這個線程離開,其他的線程才開始可以使用這個共享資源。可以把互斥鎖看成二值信號量。
2、上鎖時:
信號量: 只要信號量的value大於0,其他線程就可以sem_wait成功,成功後信號量的value減一。若value值不大於0,則sem_wait阻塞,直到sem_post釋放後value值加一。一句話,信號量的value>=0。
互斥鎖: 只要被鎖住,其他任何線程都不可以訪問被保護的資源。如果沒有鎖,獲得資源成功,否則進行阻塞等待資源可用。一句話,線程互斥鎖的vlaue可以為負數。
3、使用場所:
信號量主要適用於進程間通信,當然,也可用於線程間通信。而互斥鎖只能用於線程間通信。
華山大師兄-信號量、互斥體和自旋鎖
Linux鎖機制
linux 自旋鎖和信號量