int lock = 0; void lock(int * lock){start: if(*lock == 0) *lock = 1; else{ wait(); goto start; }} void unlock(int * lock){ *lock = 0; wakeup();}

可以看到這個鎖的實現(xiàn)邏輯很簡單，定義一個整數(shù)作為鎖，0代表沒人持鎖，1代表有人持鎖。我們先判斷鎖的值，如果是0代表沒人持鎖，我們給鎖賦值1，代表我們獲得了鎖，然后函數(shù)返回，就可以進(jìn)入臨界區(qū)了。如果鎖是1，代表有人已經(jīng)持有了鎖，此時我們就需要等待，等待函數(shù)wait，可以用忙等待，也可以用休眠等待。釋放鎖的時候把鎖設(shè)為0，然后wakeup其他線程或者為空操作。被喚醒的線程從wait中醒來，然后又重走加鎖流程。但是這里面有個問題，就是加鎖操作也是個臨界區(qū)，如果兩個線程在兩個CPU上同時執(zhí)行到加鎖函數(shù)，雙方都檢測到鎖是0，然后都把鎖置為1，都加鎖成功，這不是出問題了嗎。鎖就是用來保護(hù)臨界區(qū)的，但是加鎖本身也是臨界區(qū)，也需要保護(hù)，該怎么辦呢？唯一的方法就是求助于硬件，讓加鎖操作成為原子操作，X86平臺提供了CAS指令來實現(xiàn)這個功能。CAS，Compare and Swap，比較并交換，它的接口邏輯是這樣的：

int cas(int * p, old_value, new_value)

如果p位置的值等于old_value，就把p位置的值設(shè)置為new_value，并返回1，否則返回0。關(guān)鍵就在于cas它是硬件實現(xiàn)的，是原子的。這樣我們就可以用cas指令來實現(xiàn)鎖的邏輯，如下所示：

int lock = 0; void lock(int * lock){start: if(cas(lock, 0, 1)) return; else{ wait(); goto start; }} void unlock(int * lock){ *lock = 0; wakeup();}

4.2 簡單自旋鎖

為什么在這里要先講簡單自旋鎖呢？因為互斥鎖、信號量這些鎖的內(nèi)部實現(xiàn)都用了自旋鎖，所以先把自旋鎖的邏輯講清楚才能繼續(xù)講下去。為什么是簡單自旋鎖呢，因為自旋鎖現(xiàn)在已經(jīng)發(fā)展得很復(fù)雜了，所以這里就是講一下自旋鎖的簡單版本，因為它們的基本邏輯是一致的。自旋鎖由于在加鎖失敗時是忙等待，所以不用考慮等待隊列、睡眠喚醒的問題，所以實現(xiàn)比較簡單，下面是簡單自旋鎖的實現(xiàn)代碼：

int lock = 0; void spin_lock(int * lock){ while(!cas(lock, 0, 1))} void spin_unlock(int * lock){ *lock = 0;}

可以看到簡單自旋鎖的代碼是相當(dāng)簡單，加鎖的時候不停地嘗試加鎖，一直失敗一直加，直到成功才返回。釋放鎖的時候更簡單，直接把鎖置為0就可以了。此時如果有其它自旋鎖在自旋，由于鎖已經(jīng)變成了0，所以就會加鎖成功，結(jié)束自旋。注意這個代碼只是自旋鎖的邏輯演示，并不是真正的自旋鎖實現(xiàn)。

4.3 互斥鎖

互斥鎖是休眠鎖，加鎖失敗時要把自己放入等待隊列，釋放鎖的時候要考慮喚醒等待隊列的線程?；コ怄i的定義如下(刪除了一些配置選項)：

struct mutex { atomic_long_t    owner; raw_spinlock_t    wait_lock; struct list_head  wait_list;};

可以看到互斥鎖的定義有atomic_long_t owner，這個和我們之前定義的int lock是相似的，只不過這里是個加強(qiáng)版，0代表沒加鎖，加鎖的時候是非0，而不是簡單的1，而是記錄的是加鎖的線程。然后是自旋鎖和等待隊列，自旋鎖是用來保護(hù)等待隊列的。這里的自旋鎖為什么要用raw_spinlock_t呢，它和spinlock_t有什么區(qū)別呢？在標(biāo)準(zhǔn)的內(nèi)核版本下是沒有區(qū)別的，如果打了RTLinux補(bǔ)丁之后它們就不一樣了，spinlock_t會轉(zhuǎn)化為阻塞鎖，raw_spinlock_t還是自旋鎖，如果需要在任何情況下都要自旋的話請使用raw_spinlock_t。下面我們看一下它的加鎖操作：

void __sched mutex_lock(struct mutex *lock){ might_sleep();  if (!__mutex_trylock_fast(lock)) __mutex_lock_slowpath(lock);}static __always_inline bool __mutex_trylock_fast(struct mutex *lock){ unsigned long curr = (unsigned long)current; unsigned long zero = 0UL;  if (atomic_long_try_cmpxchg_acquire(&lock->owner, &zero, curr)) return true;  return false;}

可以看到加鎖時先嘗試直接加鎖，用的就是CAS原子指令(x86的CAS指令叫做cmpxchg)。如果owner是0，代表當(dāng)前鎖是開著的，就把owner設(shè)置為自己(也就是當(dāng)前線程，struct task_struct * 強(qiáng)轉(zhuǎn)為 ulong)，代表自己成為這個鎖的主人，也就是自己加鎖成功了，然后返回true。如果owner不為0的話，代表有人已經(jīng)持有鎖了，返回false，后面就要走慢速路徑了，也就是把自己放入等待隊列里休眠等待。下面看看慢速路徑的代碼是怎么實現(xiàn)的：

static noinline void __sched__mutex_lock_slowpath(struct mutex *lock){ __mutex_lock(lock, TASK_UNINTERRUPTIBLE, 0, NULL, _RET_IP_);} static int __sched__mutex_lock(struct mutex *lock, unsigned int state, unsigned int subclass, struct lockdep_map *nest_lock, unsigned long ip){ return __mutex_lock_common(lock, state, subclass, nest_lock, ip, NULL, false);} static __always_inline int __sched__mutex_lock_common(struct mutex *lock, unsigned int state, unsigned int subclass, struct lockdep_map *nest_lock, unsigned long ip, struct ww_acquire_ctx *ww_ctx, const bool use_ww_ctx){ struct mutex_waiter waiter; int ret;  raw_spin_lock(&lock->wait_lock); waiter.task = current; __mutex_add_waiter(lock, &waiter, &lock->wait_list);  set_current_state(state); for (;;) { bool first; if (__mutex_trylock(lock)) goto acquired;  raw_spin_unlock(&lock->wait_lock); schedule_preempt_disabled();  first = __mutex_waiter_is_first(lock, &waiter); set_current_state(state); if (__mutex_trylock_or_handoff(lock, first)) break;  raw_spin_lock(&lock->wait_lock); } acquired: __set_current_state(TASK_RUNNING); __mutex_remove_waiter(lock, &waiter); raw_spin_unlock(&lock->wait_lock);  return ret;} static void__mutex_add_waiter(struct mutex *lock, struct mutex_waiter *waiter, struct list_head *list){ debug_mutex_add_waiter(lock, waiter, current);  list_add_tail(&waiter->list, list); if (__mutex_waiter_is_first(lock, waiter)) __mutex_set_flag(lock, MUTEX_FLAG_WAITERS);}

可以看到慢速路徑最終會調(diào)用函數(shù)__mutex_lock_common，這個函數(shù)本身是很復(fù)雜的，這里進(jìn)行了大量的刪減，只保留了核心邏輯。函數(shù)先加鎖mutex的自旋鎖，然后把自己放到等待隊列上去，然后就在無限for循環(huán)中休眠并等待別人釋放鎖并喚醒自己。For循環(huán)的入口處先嘗試加鎖，如果成功就goto acquired，如果不成功就釋放自旋鎖，并調(diào)用schedule_preempt_disabled，此函數(shù)就是休眠函數(shù)，它會調(diào)度其它進(jìn)程來執(zhí)行，自己就休眠了，直到有人喚醒自己才會醒來繼續(xù)執(zhí)行。別人釋放鎖的時候會喚醒自己，這個我們后面會分析。當(dāng)我們被喚醒之后會去嘗試加鎖，如果加鎖失敗，再次來到for循環(huán)的開頭處，再試一次加鎖，如果不行就再走一次休眠過程。為什么我們加鎖會失敗呢，因為有可能多個線程同時被喚醒來爭奪鎖，我們不一定會搶鎖成功。搶鎖失敗就再去休眠，最后總會搶鎖成功的。

把自己加入等待隊列時會設(shè)置flag MUTEX_FLAG_WAITERS，這個flag是設(shè)置在owner的低位上去，因為task_struct指針至少是L1_CACHE_BYTES對齊的，所以最少有3位可以空出來做flag。

下面我們再來看一下釋放鎖的流程：

void __sched mutex_unlock(struct mutex *lock){  if (__mutex_unlock_fast(lock)) return; __mutex_unlock_slowpath(lock, _RET_IP_);} static __always_inline bool __mutex_unlock_fast(struct mutex *lock){ unsigned long curr = (unsigned long)current;  return atomic_long_try_cmpxchg_release(&lock->owner, &curr, 0UL);}

解鎖的時候先嘗試快速解鎖，快速解鎖的意思是沒有其它在等待隊列里，可以直接釋放鎖。怎么判斷等待隊列里沒有線程在等待呢，這就是前面設(shè)置的MUTEX_FLAG_WAITERS的作用了。如果設(shè)置了這個flag，lock->owner 和 curr就不會相等，直接釋放鎖就會失敗，就要走慢速路徑。慢速路徑的代碼如下：

static noinline void __sched __mutex_unlock_slowpath(struct mutex *lock, unsigned long ip){ struct task_struct *next = NULL; DEFINE_WAKE_Q(wake_q); unsigned long owner;  owner = atomic_long_read(&lock->owner); for (;;) { if (atomic_long_try_cmpxchg_release(&lock->owner, &owner, __owner_flags(owner))) { if (owner & MUTEX_FLAG_WAITERS) break; } }  raw_spin_lock(&lock->wait_lock); if (!list_empty(&lock->wait_list)) { struct mutex_waiter *waiter = list_first_entry(&lock->wait_list, struct mutex_waiter, list); next = waiter->task; wake_q_add(&wake_q, next); } raw_spin_unlock(&lock->wait_lock); wake_up_q(&wake_q);}

上述代碼進(jìn)行了一些刪減?？梢钥闯錾鲜龃a會先釋放鎖，然后喚醒等待隊列里面的第一個等待者。

4.4 信號量

信號量與互斥鎖有很大的不同，互斥鎖代表只有一個線程能同時進(jìn)入臨界區(qū)，而信號量是個整數(shù)計數(shù)，代表著某一類資源有多少個，能同時讓多少個線程訪問這類資源。信號量也沒有加鎖解鎖操作，信號量類似的操作叫做down和up，down代表獲取一個資源，up代表歸還一個資源。

下面我們先看一下信號量的定義：

struct semaphore { raw_spinlock_t    lock; unsigned int    count; struct list_head  wait_list;}; #define __SEMAPHORE_INITIALIZER(name, n)        \{                  \ .lock    = __RAW_SPIN_LOCK_UNLOCKED((name).lock),  \ .count    = n,            \ .wait_list  = LIST_HEAD_INIT((name).wait_list),    \} static inline void sema_init(struct semaphore *sem, int val){ *sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem, val);}

可以看出信號量和互斥鎖的定義很相似，都有一個自旋鎖，一個等待隊列，不同的是信號量沒有owner，取而代之的是count，代表著某一類資源的個數(shù)，而且自旋鎖同時保護(hù)著等待隊列和count。信號量初始化時要指定count的大小。

我們來看一下信號量的down操作(獲取一個資源)：

void down(struct semaphore *sem){ unsigned long flags;  might_sleep(); raw_spin_lock_irqsave(&sem->lock, flags); if (likely(sem->count > 0)) sem->count--; else __down(sem); raw_spin_unlock_irqrestore(&sem->lock, flags);} static noinline void __sched __down(struct semaphore *sem){ __down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);} static inline int __sched __down_common(struct semaphore *sem, long state, long timeout){ struct semaphore_waiter waiter;  list_add_tail(&waiter.list, &sem->wait_list); waiter.task = current; waiter.up = false;  for (;;) { if (signal_pending_state(state, current)) goto interrupted; if (unlikely(timeout <= 0)) goto timed_out; __set_current_state(state); raw_spin_unlock_irq(&sem->lock); timeout = schedule_timeout(timeout); raw_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;}

可以看出我們會先持有自旋鎖，然后看看count是不是大于0，大于0的話代表資源還有剩余，我們直接減1，代表占用一份資源，就可以返回了。如果不大于0的話，代表資源沒有了，我們就進(jìn)去等待隊列等待。

我們再來看看up操作(歸還資源)：

void up(struct semaphore *sem){ unsigned long flags;  raw_spin_lock_irqsave(&sem->lock, flags); if (likely(list_empty(&sem->wait_list))) sem->count++; else __up(sem); raw_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); waiter->up = true; wake_up_process(waiter->task);}

先加鎖自旋鎖，然后看看等待隊列是否為空，如果為空的話直接把count加1就可以了。如果不為空的話，則代表有人在等待資源，資源就不加1了，直接喚醒隊首的線程來獲取。

五、per-CPU變量

typedef struct seqcount { unsigned sequence;} seqcount_t; typedef struct { seqcount_spinlock_t seqcount; spinlock_t lock;} seqlock_t;

 #include pid_t fork(void);

static inline void write_seqlock(seqlock_t *sl){ spin_lock(&sl->lock); do_write_seqcount_begin(&sl->seqcount.seqcount);} static inline void write_sequnlock(seqlock_t *sl){ do_write_seqcount_end(&sl->seqcount.seqcount); spin_unlock(&sl->lock);}

static inline unsigned read_seqbegin(const seqlock_t *sl){ unsigned ret = read_seqcount_begin(&sl->seqcount);  kcsan_atomic_next(0);  /* non-raw usage, assume closing read_seqretry() */ kcsan_flat_atomic_begin(); return ret;} static inline unsigned read_seqretry(const seqlock_t *sl, unsigned start){ /* * Assume not nested: read_seqretry() may be called multiple times when * completing read critical section. */ kcsan_flat_atomic_end();  return read_seqcount_retry(&sl->seqcount, start);}