Linux並發與同步專題 (2)spinlock
關鍵詞：wfe、FIFO ticket-based、spin_lock/spin_trylock/spin_unlock、spin_lock_irq/spin_lock_bh/spin_lock_irqsave。
《Linux並發與同步專題 (1)原子操作和內存屏障》
《Linux並發與同步專題 (2)spinlock》
《Linux並發與同步專題 (3) 信號量》
《Linux並發與同步專題 (4) Mutex互斥量》
《Linux並發與同步專題 (5) 讀寫鎖》
《Linux並發與同步專題 (6) RCU》
《Linux並發與同步專題 (7) 內存管理中的鎖》
《Linux並發與同步專題 (8) 最新更新與展望》
spinlock同一時刻只能被一個內核代碼路徑持有，如果有另外一個內核代碼路徑試圖獲取一個已經被持有的spinlock，那麽該內核代碼路徑需要一直自旋忙等待，直到鎖持有者釋放了該鎖。
spinlock鎖的特性如下：
- spinlock屬於忙等待機制，當無法獲取spinlock鎖時會不斷嘗試，直到獲取鎖為止。
- 同一時刻只能有一個內核代碼路徑可以獲得所。
- 要求spinlock鎖持有者盡快完成臨界區的執行任務。如果臨界區執行時間過長，在鎖外面忙等待的CPU比較浪費，特別是spinlock臨界區裏不能睡眠。
- spinlock鎖可以在中斷上下文中使用。
1. spinlock實現
使用spinlock鎖的重要原則是：擁有spinlock鎖的臨界區代碼必須是原子執行，不能休眠和主動調度。
1.1 spinlock數據結構
spinlock的數據結構spinlock_t考慮了不同體系結構和實時性內核的要求。如果打上了RT-patch，就變成了使用rt_mutex。
如果沒有使用RT-patch，那麽arch_spinlock_t在ARM32體系結構下就對應如下。
typedef struct spinlock { union { struct raw_spinlock rlock; }; } spinlock_t; typedef struct raw_spinlock { arch_spinlock_t raw_lock; } raw_spinlock_t;#define TICKET_SHIFT 16 typedef struct { union { u32 slock;----------------------------在Linux 2.6.25之前，spinlock數據結構是一個簡單無符號類型變量，slock為1表示鎖未被持有，為0或者負數表示鎖被持有。 struct __raw_tickets { #ifdef __ARMEB__ u16 next; u16 owner; #else u16 owner; u16 next; #endif } tickets; }; } arch_spinlock_t;
在Linux 2.6.25內核之後，spinlock實現了一套名為“FIFO ticket-based”算法的spinlock機制。
ticket-based算法的shipinlock仍然使用原來數據結構，但slocl被拆分為兩部分。
owner表示鎖持有者的等號牌，next表示外面排隊隊列中末尾者的等號牌。
1.2 spin_lock()/spin_unlock()/spin_trylock()
spin_lock()函數最終調用__raw_spin_lock()函數來實現。
首先關閉內核搶占，然後調用架構相關的arch_spin_lock()。
static inline void spin_lock(spinlock_t *lock) { raw_spin_lock(&lock->rlock); } #define raw_spin_lock(lock) _raw_spin_lock(lock) void __lockfunc _raw_spin_lock(raw_spinlock_t *lock) { __raw_spin_lock(lock); } static inline void __raw_spin_lock(raw_spinlock_t *lock) { preempt_disable();--------------------------------------------禁止搶占，這裏受CONFIG_PREEMPT_COUNT和CONFIG_PREEMPT開關而不同。 spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);-----------------如果相關調試選項沒有打開，是一個空函數。 LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock);--在沒有定義CONFIG_LOCK_STAT的情況下，就是直接調用do_raw_spin_lock()。 }
static inline void do_raw_spin_lock(raw_spinlock_t *lock) __acquires(lock)
{
__acquire(lock);
arch_spin_lock(&lock->raw_lock);------------------------------架構相關的0spin lock處理。
}
spin_unlock()和spin_lock()對應，最終調用__raw_spin_unlock()。
分別調用架構相關的arch_spin_unlock()，然後打開搶占。
static inline void spin_unlock(spinlock_t *lock) { raw_spin_unlock(&lock->rlock); } #define raw_spin_unlock(lock) _raw_spin_unlock(lock) void __lockfunc _raw_spin_unlock(raw_spinlock_t *lock) { __raw_spin_unlock(lock); } static inline void __raw_spin_unlock(raw_spinlock_t *lock) { spin_release(&lock->dep_map, 1, _RET_IP_);---------------------和spin_acquire()對應。 do_raw_spin_unlock(lock); preempt_enable();----------------------------------------------打開搶占。 } static inline void do_raw_spin_unlock(raw_spinlock_t *lock) __releases(lock) { arch_spin_unlock(&lock->raw_lock);-----------------------------架構相關的spin unlock處理。 __release(lock); }
spin_trylock()是spin_lock的變種，spin_trylock()盡力獲得自旋鎖lock，如果能立即獲得鎖，則獲得鎖並返回真；如果不能，則返回假。
spin_trylock()不會等待lock被釋放。
static inline int spin_trylock(spinlock_t *lock) { return raw_spin_trylock(&lock->rlock); } #define raw_spin_trylock(lock) __cond_lock(lock, _raw_spin_trylock(lock)) int __lockfunc _raw_spin_trylock(raw_spinlock_t *lock) { return __raw_spin_trylock(lock); } static inline int __raw_spin_trylock(raw_spinlock_t *lock) { preempt_disable(); if (do_raw_spin_trylock(lock)) { spin_acquire(&lock->dep_map, 0, 1, _RET_IP_); return 1; } preempt_enable(); return 0; } static inline int do_raw_spin_trylock(raw_spinlock_t *lock) { return arch_spin_trylock(&(lock)->raw_lock); }
1.2.1 spinlock臨界區為什麽不允許發生搶占?
如果spinlock臨界區中允許搶占，那麽如果臨界區內發生中斷，中斷返回時會去檢查搶占調度。
這裏有兩個問題，一是搶占調度相當於持有鎖的進程睡眠，違背了spinlock鎖不能睡眠和快速執行完成的設計；
二是搶占調度進程也有可能會去申請spinlock鎖，那麽會導致發生死鎖。
1.3 arch_spin_lock()/arch_spin_unlock()/arch_spin_trylock()
通過上面的分析可知，spin_lock()/spin_unlock()的核心是架構相關的部分。
static inline void arch_spin_lock(arch_spinlock_t *lock) { unsigned long tmp; u32 newval; arch_spinlock_t lockval; prefetchw(&lock->slock); __asm__ __volatile__( "1: ldrex %0, [%3]\n" " add %1, %0, %4\n" " strex %2, %1, [%3]\n" " teq %2, #0\n" " bne 1b" : "=&r" (lockval), "=&r" (newval), "=&r" (tmp) : "r" (&lock->slock), "I" (1 << TICKET_SHIFT)-----------------------------TICKET_SHIFT為16，也即lock->tickets.next++。 : "cc"); while (lockval.tickets.next != lockval.tickets.owner) {-------------------next是當前進程對應鎖的等號牌，判斷owner和next是否相等。如相等表示CPU成功獲取了spinlock鎖，arch_spin_lock()返回；如不等，則調用wfe讓CPU進入等待狀態。 wfe(); lockval.tickets.owner = ACCESS_ONCE(lock->tickets.owner);-------------更新整個系統鎖持有者的等號牌,lockval.tickets.owner一直在變，而lockval.tickets.next是當前位置持鎖等號牌，不變。 } smp_mb();-----------------------------------------------------------------內存屏障 }
arch_spin_unlock()更新owner域的值，並且發送指令喚醒CPU。
static inline void arch_spin_unlock(arch_spinlock_t *lock) { smp_mb(); lock->tickets.owner++;---------------------------------------------------更新owner，表示所得持有者轉到下一個臨界區。 dsb_sev(); }
可以看出arch_spin_trylock()並沒有類似arch_spin_lock()一樣執行wfe指令，所以會直接返回1或者0.
所以使用spin_trylock()的代碼路徑不會阻塞，它會導致其他欲持有該鎖的代碼路徑阻塞。
在使用spin_trylock()成功獲取鎖之後，仍然需要使用spin_unlock()解鎖。
static inline int arch_spin_trylock(arch_spinlock_t *lock) { unsigned long contended, res; u32 slock; prefetchw(&lock->slock); do { __asm__ __volatile__( " ldrex %0, [%3]\n" " mov %2, #0\n" " subs %1, %0, %0, ror #16\n" " addeq %0, %0, %4\n" " strexeq %2, %0, [%3]" : "=&r" (slock), "=&r" (contended), "=&r" (res) : "r" (&lock->slock), "I" (1 << TICKET_SHIFT) : "cc"); } while (res); if (!contended) { smp_mb(); return 1; } else { return 0; } }
1.3.1 wfe和wfi對比
wfi和wfe指令都是讓ARM核進入standby睡眠模式。wfi是直到有wfi喚醒事件發生才會喚醒CPU，wfe是直到wfe喚醒事件發生，這兩類事件大部分相同。
唯一不同之處在於wfr可以被其他CPU上的sev指令喚醒，sec指令用於修改event寄存器的指令。
1.3.2 smp_mb和dsb_sev
smp_mb調用dmb指令來保證把調用函數之前所有的訪問內存指令都執行完成。
dsb_sev()包含兩個指令dsb和sev。dsb指令保證之前的owner與已經寫入內存中；sev指令來喚醒通過wfe指令進入睡眠狀態的CPU。
此處arch_spin_lock()的wfe()等待處被喚醒，進行owner更新以及和next比較。
2. spinlock變種
上面是spinlock最常見的形式spin_lock()/spin_trylock()/spin_unlock()。
spinlock還有其他一些變種。
2.1 irq變種
2.1.1 為什麽需要irq變種
假設臨界區有鏈表操作需要spinlock來保護。
當處於臨界區發生了外部硬件中斷，此時系統暫停當前進程的執行轉而去處理該中斷。
假設中斷處理程序恰巧也要操作該鏈表，鏈表的操作是一個臨界區，所以在操作之前要調用spin_lock()函數來對該鏈表進行保護。
中斷處理函數試圖去獲取該spinlock，但因為它已經被別人持有了，於是導致中斷處理函數進入忙等待狀態或者wfe睡眠狀態。
在中斷上下文出現忙等待或者睡眠狀態是致命的，中斷處理程序要求“短“和”快“，鎖的持有者因為被中斷打斷而不能盡快釋放鎖，而中斷處理程序一直在忙等待鎖，從而導致死鎖的發生。
Linux內核spin_lock_irq()函數在獲取spinlock時關閉本地CPU中斷，可以解決該問題。
2.1.2 spin_lock_irq()/spin_trylock_irq()/spin_unlock_irq()
從下面三個函數的調用關系可以知道，最終跟spin_lock()/spin_trylock()/spin_unlock()相似，只是中間增加了local_irq_disable()/local_irq_enable()。
local_irq_disable()用於關閉本地處理器中斷，這樣在獲取spinlock瑣時可以確保不會發生中斷，從而避免發生死鎖問題。
因此spin_lock_irq()主要防止本地中斷處理程序和持有鎖者之間存在鎖的爭用。
spin_lock_irq()代碼路徑如下：
static inline void spin_lock_irq(spinlock_t *lock) { raw_spin_lock_irq(&lock->rlock); } #define raw_spin_lock_irq(lock) _raw_spin_lock_irq(lock) void __lockfunc _raw_spin_lock_irq(raw_spinlock_t *lock) { __raw_spin_lock_irq(lock); } static inline void __raw_spin_lock_irq(raw_spinlock_t *lock) { local_irq_disable();-------------------------------------------------------關本地處理器中斷 preempt_disable();---------------------------------------------------------禁止搶占 spin_acquire(&lock->dep_map, 0, 0, _RET_IP_); LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock); }
下面是spin_unlock_irq()代碼路徑，可以看出在spin_lock_irq()和spin_unlock_irq()之間的部分是關本地中斷和禁止搶占的。
static inline void spin_unlock_irq(spinlock_t *lock) { raw_spin_unlock_irq(&lock->rlock); } #define raw_spin_unlock_irq(lock) _raw_spin_unlock_irq(lock) void __lockfunc _raw_spin_unlock_irq(raw_spinlock_t *lock) { __raw_spin_unlock_irq(lock); } static inline void __raw_spin_unlock_irq(raw_spinlock_t *lock) { spin_release(&lock->dep_map, 1, _RET_IP_); do_raw_spin_unlock(lock); local_irq_enable();---------------------------------------------------------打開本地處理器中斷 preempt_enable();-----------------------------------------------------------允許搶占 }
spin_trylock_irq()主要通過raw_spin_trylock()，所以這裏和spin_trylock()的區別也僅僅是關中斷/開中斷。
如果獲取鎖成功，那麽後面會有spin_unlock_irq()來開中斷；如果獲取失敗，直接開中斷。
static inline int spin_trylock_irq(spinlock_t *lock) { return raw_spin_trylock_irq(&lock->rlock); } #define raw_spin_trylock_irq(lock) \ ({ local_irq_disable(); \-----------------------------------------------------增加禁止本地處理器中斷 raw_spin_trylock(lock) ? 1 : ({ local_irq_enable(); 0; }); \---------------------------------------如果獲取鎖不成功，打開本地處理器中斷 })
2.2 bh變種
bh變種可以對照原生的spinlock，區別在於原生spinlock使用preempt_enable()/preempt_disable()來禁止搶占；而bh變種不但表示了當前處於軟中斷上下文，還可能會禁止搶占，這取決於SOFTIRQ_LOCK_OFFSET的定義。
spin_lock_bh()：
static inline void spin_lock_bh(spinlock_t *lock) { raw_spin_lock_bh(&lock->rlock); } #define raw_spin_lock_bh(lock) _raw_spin_lock_bh(lock) void __lockfunc _raw_spin_lock_bh(raw_spinlock_t *lock) { __raw_spin_lock_bh(lock); } static inline void __raw_spin_lock_bh(raw_spinlock_t *lock) { __local_bh_disable_ip(_RET_IP_, SOFTIRQ_LOCK_OFFSET); spin_acquire(&lock->dep_map, 0, 0, _RET_IP_); LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock); }
spin_trylock_bh()：
static inline int spin_trylock_bh(spinlock_t *lock) { return raw_spin_trylock_bh(&lock->rlock); } #define raw_spin_trylock_bh(lock) \ __cond_lock(lock, _raw_spin_trylock_bh(lock)) int __lockfunc _raw_spin_trylock_bh(raw_spinlock_t *lock) { return __raw_spin_trylock_bh(lock); } static inline int __raw_spin_trylock_bh(raw_spinlock_t *lock) { __local_bh_disable_ip(_RET_IP_, SOFTIRQ_LOCK_OFFSET); if (do_raw_spin_trylock(lock)) { spin_acquire(&lock->dep_map, 0, 1, _RET_IP_); return 1; } __local_bh_enable_ip(_RET_IP_, SOFTIRQ_LOCK_OFFSET); return 0; }
spin_unlock_bh()：
static inline void spin_unlock_bh(spinlock_t *lock) { raw_spin_unlock_bh(&lock->rlock); } #define raw_spin_unlock_bh(lock) _raw_spin_unlock_bh(lock) void __lockfunc _raw_spin_unlock_bh(raw_spinlock_t *lock) { __raw_spin_unlock_bh(lock); } static inline void __raw_spin_unlock_bh(raw_spinlock_t *lock) { spin_release(&lock->dep_map, 1, _RET_IP_); do_raw_spin_unlock(lock); __local_bh_enable_ip(_RET_IP_, SOFTIRQ_LOCK_OFFSET); }
2.3 irqsave變種
下面是irqsave的三個變種函數，這些函數和irq變種基本上一致。
只是local_irq_enable()/local_irq_disable()換成了local_irq_save()/local_irq_restore()。
主要是保存和恢復CPSR寄存器內容，這樣做的目的是防止破壞掉中斷響應的狀態。
spin_lock_irqsave()函數：
#define spin_lock_irqsave(lock, flags) do { raw_spin_lock_irqsave(spinlock_check(lock), flags); } while (0) #define raw_spin_lock_irqsave(lock, flags) do { typecheck(unsigned long, flags); flags = _raw_spin_lock_irqsave(lock); } while (0) unsigned long __lockfunc _raw_spin_lock_irqsave(raw_spinlock_t *lock) { return __raw_spin_lock_irqsave(lock); } static inline unsigned long __raw_spin_lock_irqsave(raw_spinlock_t *lock) { unsigned long flags; local_irq_save(flags);------------------------------------------------保存中斷轉改CPSR寄存器到flags，然後關閉本地CPU中斷。 preempt_disable();----------------------------------------------------關閉搶占。 spin_acquire(&lock->dep_map, 0, 0, _RET_IP_); /* * On lockdep we dont want the hand-coded irq-enable of * do_raw_spin_lock_flags() code, because lockdep assumes * that interrupts are not re-enabled during lock-acquire: */ do_raw_spin_lock_flags(lock, &flags); return flags; } static inline void do_raw_spin_lock_flags(raw_spinlock_t *lock, unsigned long *flags) __acquires(lock) { __acquire(lock); arch_spin_lock_flags(&lock->raw_lock, *flags); } #define arch_spin_lock_flags(lock, flags) arch_spin_lock(lock)
spin_trylock_irqsave()函數：
#define spin_trylock_irqsave(lock, flags) \ ({ raw_spin_trylock_irqsave(spinlock_check(lock), flags); }) #define raw_spin_trylock_irqsave(lock, flags) \ ({ local_irq_save(flags); raw_spin_trylock(lock) ? 1 : ({ local_irq_restore(flags); 0; }); })
spin_unlock_irqrestore()函數：
static inline void spin_unlock_irqrestore(spinlock_t *lock, unsigned long flags) { raw_spin_unlock_irqrestore(&lock->rlock, flags); } #define raw_spin_unlock_irqrestore(lock, flags) do { typecheck(unsigned long, flags); _raw_spin_unlock_irqrestore(lock, flags); } while (0) void __lockfunc _raw_spin_unlock_irqrestore(raw_spinlock_t *lock, unsigned long flags) { __raw_spin_unlock_irqrestore(lock, flags); } static inline void __raw_spin_unlock_irqrestore(raw_spinlock_t *lock, unsigned long flags) { spin_release(&lock->dep_map, 1, _RET_IP_); do_raw_spin_unlock(lock); local_irq_restore(flags);-----------------------------------------恢復flags到寄存器中，並打開處理器中斷相應。 preempt_enable();-------------------------------------------------打開搶占。 }
2.4 preempt_enable()/preempt_disable()和__local_bh_enable_ip()/__local_bh_disable_ip()
這四個函數都是改變進程preempt_count計數的，其中preempt_enable()/preempt_disable()只修改preempt域，表示是否允許搶占。
#define preempt_enable() do { barrier(); preempt_count_dec(); } while (0) #define preempt_count_dec() preempt_count_sub(1) #define preempt_count_sub(val) __preempt_count_sub(val) #define preempt_disable() do { preempt_count_inc(); barrier(); } while (0) #define preempt_count_inc() preempt_count_add(1) #define preempt_count_add(val) __preempt_count_add(val)
而__local_bh_enable_ip()/__local_bh_disable_ip()修改的內容則是可變的，具體到bh變種中對應的是SOFTIRQ_LOCK_OFFSET。
#define SOFTIRQ_LOCK_OFFSET (SOFTIRQ_DISABLE_OFFSET + PREEMPT_CHECK_OFFSET)
由上面的定義可知，這裏可能會禁止軟中斷和搶占。
void __local_bh_enable_ip(unsigned long ip, unsigned int cnt) { ... preempt_count_sub(cnt - 1); if (unlikely(!in_interrupt() && local_softirq_pending())) { /* * Run softirq if any pending. And do it in its own stack * as we may be calling this deep in a task call stack already. */ do_softirq(); } preempt_count_dec(); ... } static __always_inline void __local_bh_disable_ip(unsigned long ip, unsigned int cnt) { preempt_count_add(cnt); barrier(); }
2.5 local_irq_enable()/local_irq_disable()、local_irq_save()/local_irq_restore()
這四個函數最終都通過架構相關的函數實現。
#define local_irq_enable() do { raw_local_irq_enable(); } while (0) #define local_irq_disable() do { raw_local_irq_disable(); } while (0) #define local_irq_save(flags) do { raw_local_irq_save(flags); } while (0) #define local_irq_restore(flags) do { raw_local_irq_restore(flags); } while (0) #define raw_local_irq_disable() arch_local_irq_disable() #define raw_local_irq_enable() arch_local_irq_enable() #define raw_local_irq_save(flags) do { typecheck(unsigned long, flags); flags = arch_local_irq_save(); } while (0) #define raw_local_irq_restore(flags) do { typecheck(unsigned long, flags); arch_local_irq_restore(flags); } while (0) /* * Disable IRQs */ static inline void arch_local_irq_disable(void) { unsigned long temp; asm volatile( " mrs %0, cpsr @ arch_local_irq_disable\n" " orr %0, %0, #128\n" " msr cpsr_c, %0" : "=r" (temp) : : "memory", "cc"); } /* * Enable IRQs */ static inline void arch_local_irq_enable(void) { unsigned long temp; asm volatile( " mrs %0, cpsr @ arch_local_irq_enable\n" " bic %0, %0, #128\n" " msr cpsr_c, %0" : "=r" (temp) : : "memory", "cc"); } /* * Save the current interrupt enable state & disable IRQs */ static inline unsigned long arch_local_irq_save(void) { unsigned long flags, temp; asm volatile( " mrs %0, cpsr @ arch_local_irq_save\n" " orr %1, %0, #128\n" " msr cpsr_c, %1" : "=r" (flags), "=r" (temp) : : "memory", "cc"); return flags; } static inline void arch_local_irq_restore(unsigned long flags) { asm volatile( " msr " IRQMASK_REG_NAME_W ", %0 @ local_irq_restore" : : "r" (flags) : "memory", "cc"); }
3. spin_lock()和raw_spin_lock()
對於沒有打RT-patch的平臺，spin_lock()直接調用raw_spin_lock()。
在傳統的include/linux/spinlock.h中定義了spin_lock()。
和RT-pathc後的內核差異在spin_lock()的定義轉到spinlock_rt.h中去了。
#ifdef CONFIG_PREEMPT_RT_FULL # include <linux/spinlock_rt.h> #else /* PREEMPT_RT_FULL */ ...static __always_inline void spin_lock(spinlock_t *lock) { raw_spin_lock(&lock->rlock); } ... #endif /* !PREEMPT_RT_FULL */
Linux的RT-patch旨在提升Linux內核的實時性，它允許在spinlock鎖的臨界區內被強占，且臨界區內允許進程睡眠等到，這樣會導致spinlock語義被修改。
當打開CONFIG_PREEMPT_RT_FULL選項後，spin_lock()的工作就從raw_spin_lock()變成了rt_spin_lock()。
rt_spin_lock()依賴於rt_mutex，其中允許睡眠和搶占。
#define spin_lock(lock) rt_spin_lock(lock) void __lockfunc rt_spin_lock(spinlock_t *lock) { rt_spin_lock_fastlock(&lock->lock, rt_spin_lock_slowlock, true); spin_acquire(&lock->dep_map, 0, 0, _RET_IP_); } static inline void rt_spin_lock_fastlock(struct rt_mutex *lock, void (*slowfn)(struct rt_mutex *lock, bool mg_off), bool do_mig_dis) { might_sleep_no_state_check(); if (do_mig_dis) migrate_disable(); if (likely(rt_mutex_cmpxchg_acquire(lock, NULL, current))) rt_mutex_deadlock_account_lock(lock, current); else slowfn(lock, do_mig_dis); } static void noinline __sched rt_spin_lock_slowlock(struct rt_mutex *lock, bool mg_off) { ... }
內核中真正不允許睡眠的地方修改為使用raw_spin_lock()，其它地方使用spin_lock()。
spin_lock()和raw_spin_lock()的區別在於：在絕對不允許被強占和睡眠的臨界區，應該使用raw_spin_lock()，否則使用spin_lock()。
Linux並發與同步專題 (2)spinlock