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

可以看到这个锁的实现逻辑很简单，定义一个整数作为锁，0代表没人持锁，1代表有人持锁。我们先判断锁的值，如果是0代表没人持锁，我们给锁赋值1，代表我们获得了锁，然后函数返回，就可以进入临界区了。如果锁是1，代表有人已经持有了锁，此时我们就需要等待，等待函数wait，可以用忙等待，也可以用休眠等待。释放锁的时候把锁设为0，然后wakeup其他线程或者为空操作。被唤醒的线程从wait中醒来，然后又重走加锁流程。但是这里面有个问题，就是加锁操作也是个临界区，如果两个线程在两个CPU上同时执行到加锁函数，双方都检测到锁是0，然后都把锁置为1，都加锁成功，这不是出问题了吗。锁就是用来保护临界区的，但是加锁本身也是临界区，也需要保护，该怎么办呢？唯一的方法就是求助于硬件，让加锁操作成为原子操作，X86平台提供了CAS指令来实现这个功能。CAS，Compare and Swap，比较并交换，它的接口逻辑是这样的：

int cas(int * p, old_value, new_value)

如果p位置的值等于old_value，就把p位置的值设置为new_value，并返回1，否则返回0。关键就在于cas它是硬件实现的，是原子的。这样我们就可以用cas指令来实现锁的逻辑，如下所示：

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 简单自旋锁

为什么在这里要先讲简单自旋锁呢？因为互斥锁、信号量这些锁的内部实现都用了自旋锁，所以先把自旋锁的逻辑讲清楚才能继续讲下去。为什么是简单自旋锁呢，因为自旋锁现在已经发展得很复杂了，所以这里就是讲一下自旋锁的简单版本，因为它们的基本逻辑是一致的。自旋锁由于在加锁失败时是忙等待，所以不用考虑等待队列、睡眠唤醒的问题，所以实现比较简单，下面是简单自旋锁的实现代码：

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

可以看到简单自旋锁的代码是相当简单，加锁的时候不停地尝试加锁，一直失败一直加，直到成功才返回。释放锁的时候更简单，直接把锁置为0就可以了。此时如果有其它自旋锁在自旋，由于锁已经变成了0，所以就会加锁成功，结束自旋。注意这个代码只是自旋锁的逻辑演示，并不是真正的自旋锁实现。

4.3 互斥锁

互斥锁是休眠锁，加锁失败时要把自己放入等待队列，释放锁的时候要考虑唤醒等待队列的线程。互斥锁的定义如下(删除了一些配置选项)：

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

可以看到互斥锁的定义有atomic_long_t owner，这个和我们之前定义的int lock是相似的，只不过这里是个加强版，0代表没加锁，加锁的时候是非0，而不是简单的1，而是记录的是加锁的线程。然后是自旋锁和等待队列，自旋锁是用来保护等待队列的。这里的自旋锁为什么要用raw_spinlock_t呢，它和spinlock_t有什么区别呢？在标准的内核版本下是没有区别的，如果打了RTLinux补丁之后它们就不一样了，spinlock_t会转化为阻塞锁，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，代表当前锁是开着的，就把owner设置为自己(也就是当前线程，struct task_struct * 强转为 ulong)，代表自己成为这个锁的主人，也就是自己加锁成功了，然后返回true。如果owner不为0的话，代表有人已经持有锁了，返回false，后面就要走慢速路径了，也就是把自己放入等待队列里休眠等待。下面看看慢速路径的代码是怎么实现的：

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);}

可以看到慢速路径最终会调用函数__mutex_lock_common，这个函数本身是很复杂的，这里进行了大量的删减，只保留了核心逻辑。函数先加锁mutex的自旋锁，然后把自己放到等待队列上去，然后就在无限for循环中休眠并等待别人释放锁并唤醒自己。For循环的入口处先尝试加锁，如果成功就goto acquired，如果不成功就释放自旋锁，并调用schedule_preempt_disabled，此函数就是休眠函数，它会调度其它进程来执行，自己就休眠了，直到有人唤醒自己才会醒来继续执行。别人释放锁的时候会唤醒自己，这个我们后面会分析。当我们被唤醒之后会去尝试加锁，如果加锁失败，再次来到for循环的开头处，再试一次加锁，如果不行就再走一次休眠过程。为什么我们加锁会失败呢，因为有可能多个线程同时被唤醒来争夺锁，我们不一定会抢锁成功。抢锁失败就再去休眠，最后总会抢锁成功的。

把自己加入等待队列时会设置flag MUTEX_FLAG_WAITERS，这个flag是设置在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);}

解锁的时候先尝试快速解锁，快速解锁的意思是没有其它在等待队列里，可以直接释放锁。怎么判断等待队列里没有线程在等待呢，这就是前面设置的MUTEX_FLAG_WAITERS的作用了。如果设置了这个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);}

上述代码进行了一些删减。可以看出上述代码会先释放锁，然后唤醒等待队列里面的第一个等待者。

4.4 信号量

信号量与互斥锁有很大的不同，互斥锁代表只有一个线程能同时进入临界区，而信号量是个整数计数，代表着某一类资源有多少个，能同时让多少个线程访问这类资源。信号量也没有加锁解锁操作，信号量类似的操作叫做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，代表着某一类资源的个数，而且自旋锁同时保护着等待队列和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的话，代表资源没有了，我们就进去等待队列等待。

我们再来看看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);}