并发1.0中已经介绍过操作系统中并发的常见处理方式，不过那是用户态代码。Android作为多处理器系统，现在都有8个核了，更需要一些同步机制来限制各执行单元对共享的数据和代码的访问。
信号量为1 ≈ 互斥锁 ≈ 允许睡眠的自旋锁
原子操作
顾名思义，要么执行完，要么不执行，不存在执行到一半的状态。常用作实现资源计数、引用计数。
API定义位于<arch/arm64/include/asm/atomic.h>，通过汇编语言实现，内核源码根目录下的include/asm-generic/atomic.h则抽象封装了API，该API最后分派的实现来自于arch目录下对应的代码。
常用API
atomic_long_t 就等于 atomic64_t
int atomic_read(atomic_t * v);
该函数对原子类型的变量进行原子读操作，它返回原子类型的变量v的int值。void atomic_set(atomic_t * v, int i);
该函数设置原子类型的变量v的值为i。void atomic_add(int i, atomic_t *v);
该函数给原子类型的变量v增加值i。atomic_sub(int i, atomic_t *v);
该函数从原子类型的变量v中减去i。int atomic_sub_and_test(int i, atomic_t *v);
该函数从原子类型的变量v中减去i，并判断结果是否为0，如果为0，返回真，否则返回假。void atomic_inc(atomic_t *v);
该函数对原子类型变量v原子地增加1。void atomic_dec(atomic_t *v);
该函数对原子类型的变量v原子地减1。int atomic_dec_and_test(atomic_t *v);
该函数对原子类型的变量v原子地减1，并判断结果是否为0，如果为0，返回真，否则返回假。int atomic_inc_and_test(atomic_t *v);
该函数对原子类型的变量v原子地增加1，并判断结果是否为0，如果为0，返回真，否则返回假。int atomic_add_negative(int i, atomic_t *v);
该函数对原子类型的变量v原子地增加i，并判断结果是否为负数，如果是，返回真，否则返回假。int atomic_add_return(int i, atomic_t *v);
该函数对原子类型的变量v原子地增加i，并且返回指向v的指针。int atomic_sub_return(int i, atomic_t *v);
该函数从原子类型的变量v中减去i，并且返回指向v的指针。int atomic_inc_return(atomic_t * v);
该函数对原子类型的变量v原子地增加1并且返回指向v的指针。int atomic_dec_return(atomic_t * v);
该函数对原子类型的变量v原子地减1并且返回指向v的指针。
自旋锁
若自旋锁已被别的执行者保持，调用者就会原地循环等待并检查该锁的持有者是否已经释放锁（即进入自旋状态），若释放则调用者开始持有该锁。自旋锁持有期间不可被抢占。适用于代码较短，切换上下文时间还不如自旋等待。
常用API
- spin_lock_init 动态初始化spin_lock；常用
//初始化结构体中的锁
spin_lock_init(&zone-lock);
- DEFINE_SPINLOCK 声明并初始化spin_lock；常用
mm/page_alloc.c
static DEFINE_SPINLOCK(early_pfn_lock);
spin_lock(&early_pfn_lock);
- SPIN_LOCK_UNLOCKED 静态初始化spin_lock，都没见过用的
- spin_lock 上锁
- spin_unlock 释放锁
- spin_trylock 尝试获得锁，不能就立即返回，不自旋；用的很少
- spin_is_locked 判断自旋锁是否已被持有；用得更少
spin_lock可以防止多处理器多进程的非同步访问。那么对于同一CPU上，如何防止因为中断导致的并发问题呢？
- spin_lock_irqsave (lock,flags) / spin_unlock_irqrestore (lock,flags)
获取锁时保存中断寄存器的到flags中，关中断，释放锁反之。如果中断前，是开的中断，之后还是开；中断前关的中断，之后还是关。这个就是和不保存寄存器的区别。
<page_alloc.c>
unsigned long flags;
spin_lock_irqsave(&zone->lock, flags);
...
spin_unlock_irqrestore(&zone->lock, flags);
spin_lock_irq(lock) / spin_unlock_irq(lock)
获取锁时，关中断，释放时把中断打开。如果中断前，中断是关的，这个时候就会打开。spin_trylock_irqsave (lock,flags)
能够获得锁时保存中断寄存器的到flags中，关中断，否则立刻返回。
static bool compact_lock_irqsave(spinlock_t *lock, unsigned long *flags,
struct compact_control *cc)
{
/* Track if the lock is contended in async mode */
if (cc->mode == MIGRATE_ASYNC && !cc->contended) {
if (spin_trylock_irqsave(lock, *flags))
return true;
cc->contended = true;
}
spin_lock_irqsave(lock, *flags);
return true;
}
- spin_trylock_bh
能够获得锁时，关软中断，否则立刻返回
性能上，spin_lock > spin_lock_bh > spin_lock_irq > spin_lock_irqsave。
安全上，spin_lock_irqsave > spin_lock_irq > spin_lock_bh >spin_lock。
信号量
看下定义，自旋锁+计数变量+等待队列
struct semaphore {
raw_spinlock_t lock;
unsigned int count;
struct list_head wait_list;
};
再看下PV操作，这是down的，就是取锁：
<include/linux/semaphore.h>
extern void down(struct semaphore *sem);
void down(struct semaphore *sem)
{
unsigned long flags;
raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(sem->count > 0)) /*如果计数器大于1，可以进入*/
sem->count--;
else
/*小于0就将task加入等待队列，并进入等待队列，直至其被__up唤醒，或者因超时以被移除等待队列*/
__down(sem);
raw_spin_unlock_irqrestore(&sem->lock, flags);
}
EXPORT_SYMBOL(down);
static noinline void __sched __down(struct semaphore *sem)
{
__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}
struct semaphore_waiter {
struct list_head list;
struct task_struct *task;
bool up;
};
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); //避免和up线程，抢占sem->lock
timeout = schedule_timeout(timeout);
raw_spin_lock_irq(&sem->lock);
if (waiter.up) /*等这个被置位1*/
return 0;
}
/*超时移除等待队列*/
timed_out:
list_del(&waiter.list);
return -ETIME;
interrupted:
list_del(&waiter.list);
return -EINTR;
}
再看下放锁：
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);
}
EXPORT_SYMBOL(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 = true; /*设置为true*/
wake_up_process(waiter->task); /*执行到shed/core.c的，唤醒该进程*/
}
常用API
- DEFINE_SEMAPHORE(name) 声明信号量并初始化为1
/kernel/printk/printk.c
static DEFINE_SEMAPHORE(console_sem)
- void sema_init(struct semaphore *sem, int val) 声明信号量并初始化为val
- void down(&name) 获得信号量，task不可被中断
- int down_interruptible(struct semaphore *sem) 获得信号量，task可被中断，成功返回0
static noinline int __sched __down_interruptible(struct semaphore *sem)
{
return __down_common(sem, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}
- int down_killable(struct semaphore *sem) 获得信号量，task可被kill，成功返回0
static noinline int __sched __down_killable(struct semaphore *sem)
{
return __down_common(sem, TASK_KILLABLE, MAX_SCHEDULE_TIMEOUT);
}
- int down_trylock(struct semaphore *sem) 尝试获取信号量，count--，否则直接返回，不加入等待队列
int down_trylock(struct semaphore *sem)
{
unsigned long flags;
int count;
raw_spin_lock_irqsave(&sem->lock, flags);
count = sem->count - 1;
if (likely(count >= 0))
sem->count = count;
raw_spin_unlock_irqrestore(&sem->lock, flags);
return (count < 0);
}
EXPORT_SYMBOL(down_trylock);
- void up(struct semaphore *sem) 释放信号量
Mutex锁-互斥锁
在互斥锁中同时只能有一个线程可以访问该锁保护的共享资源，且释放锁和获得锁的调用方必须一致。因此在互斥锁中，除了对锁本身进行同步，对调用方（或称持有者）必须也进行同步。当互斥锁无法获得时，task会加入等待队列，直至可获得锁为止。
结构体：
struct mutex {
atomic_long_t owner;
spinlock_t wait_lock;
#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
/*加入了自旋锁功能，常用算法是OSQ算法，看了下俺手机是打开的*/
struct optimistic_spin_queue osq; /* Spinner MCS lock */
#endif
struct list_head wait_list;
#ifdef CONFIG_DEBUG_MUTEXES
void *magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
struct lockdep_map dep_map;
#endif
};
看下上锁操作的代（相当于down）：
void __sched mutex_lock(struct mutex *lock)
{
might_sleep(); /*可能会睡眠，所以不能用于中断处理和下半部处理*/
if (!__mutex_trylock_fast(lock)) /*进程号为0的时候可以走快速路径*/
__mutex_lock_slowpath(lock); /*核心！*/
}
EXPORT_SYMBOL(mutex_lock);
__mutex_lock_slowpath(struct mutex *lock)
{
/*是不是很眼熟，是的，和down很像，说明又可以给进程更换state了，还有killable啥的可选*/
__mutex_lock(lock, TASK_UNINTERRUPTIBLE, 0, NULL, _RET_IP_);
}
static int __sched
__mutex_lock(struct mutex *lock, long 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, long 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;
bool first = false;
struct ww_mutex *ww;
int ret;
might_sleep();
#ifdef CONFIG_DEBUG_MUTEXES
DEBUG_LOCKS_WARN_ON(lock->magic != lock);
#endif
ww = container_of(lock, struct ww_mutex, base);
if (use_ww_ctx && ww_ctx) {
if (unlikely(ww_ctx == READ_ONCE(ww->ctx)))
return -EALREADY;
/*
* Reset the wounded flag after a kill. No other process can
* race and wound us here since they can't have a valid owner
* pointer if we don't have any locks held.
*/
if (ww_ctx->acquired == 0)
ww_ctx->wounded = 0;
}
preempt_disable();
mutex_acquire_nest(&lock->dep_map, subclass, 0, nest_lock, ip);
if (__mutex_trylock(lock) ||
mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx, NULL)) {
/* got the lock, yay! */
lock_acquired(&lock->dep_map, ip);
if (use_ww_ctx && ww_ctx)
ww_mutex_set_context_fastpath(ww, ww_ctx);
preempt_enable();
return 0;
}
spin_lock(&lock->wait_lock);
/*
* After waiting to acquire the wait_lock, try again.
*/
if (__mutex_trylock(lock)) {
if (use_ww_ctx && ww_ctx)
__ww_mutex_check_waiters(lock, ww_ctx);
goto skip_wait;
}
debug_mutex_lock_common(lock, &waiter);
lock_contended(&lock->dep_map, ip);
if (!use_ww_ctx) {
/* add waiting tasks to the end of the waitqueue (FIFO): */
__mutex_add_waiter(lock, &waiter, &lock->wait_list);
#ifdef CONFIG_DEBUG_MUTEXES
waiter.ww_ctx = MUTEX_POISON_WW_CTX;
#endif
} else {
/*
* Add in stamp order, waking up waiters that must kill
* themselves.
*/
ret = __ww_mutex_add_waiter(&waiter, lock, ww_ctx);
if (ret)
goto err_early_kill;
waiter.ww_ctx = ww_ctx;
}
waiter.task = current;
trace_android_vh_mutex_wait_start(lock); // google kernel-common - set inherit ux
set_current_state(state);
for (;;) {
/*
* Once we hold wait_lock, we're serialized against
* mutex_unlock() handing the lock off to us, do a trylock
* before testing the error conditions to make sure we pick up
* the handoff.
*/
if (__mutex_trylock(lock))
goto acquired;
/*
* Check for signals and kill conditions while holding
* wait_lock. This ensures the lock cancellation is ordered
* against mutex_unlock() and wake-ups do not go missing.
*/
if (unlikely(signal_pending_state(state, current))) {
ret = -EINTR;
goto err;
}
if (use_ww_ctx && ww_ctx) {
ret = __ww_mutex_check_kill(lock, &waiter, ww_ctx);
if (ret)
goto err;
}
spin_unlock(&lock->wait_lock);
schedule_preempt_disabled();
/*
* ww_mutex needs to always recheck its position since its waiter
* list is not FIFO ordered.
*/
if ((use_ww_ctx && ww_ctx) || !first) {
first = __mutex_waiter_is_first(lock, &waiter);
if (first)
__mutex_set_flag(lock, MUTEX_FLAG_HANDOFF);
}
set_current_state(state);
/*
* Here we order against unlock; we must either see it change
* state back to RUNNING and fall through the next schedule(),
* or we must see its unlock and acquire.
*/
if (__mutex_trylock(lock) ||
(first && mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx, &waiter)))
break;
spin_lock(&lock->wait_lock);
}
spin_lock(&lock->wait_lock);
acquired:
__set_current_state(TASK_RUNNING);
trace_android_vh_mutex_wait_finish(lock); // google kernel-common - unset inherit ux
if (use_ww_ctx && ww_ctx) {
/*
* Wound-Wait; we stole the lock (!first_waiter), check the
* waiters as anyone might want to wound us.
*/
if (!ww_ctx->is_wait_die &&
!__mutex_waiter_is_first(lock, &waiter))
__ww_mutex_check_waiters(lock, ww_ctx);
}
mutex_remove_waiter(lock, &waiter, current);
if (likely(list_empty(&lock->wait_list)))
__mutex_clear_flag(lock, MUTEX_FLAGS);
debug_mutex_free_waiter(&waiter);
skip_wait:
/* got the lock - cleanup and rejoice! */
lock_acquired(&lock->dep_map, ip);
if (use_ww_ctx && ww_ctx)
ww_mutex_lock_acquired(ww, ww_ctx);
spin_unlock(&lock->wait_lock);
preempt_enable();
return 0;
err:
__set_current_state(TASK_RUNNING);
trace_android_vh_mutex_wait_finish(lock);
mutex_remove_waiter(lock, &waiter, current);
err_early_kill:
spin_unlock(&lock->wait_lock);
debug_mutex_free_waiter(&waiter);
mutex_release(&lock->dep_map, 1, ip);
preempt_enable();
return ret;
}
常用API：
- DEFINE_MUTEX(name) 静态声明互斥量并初始化解锁状态
- mutex_init(&name) 动态声明互斥量并初始化解锁状态
- bool mutex_is_locked(struct mutex *lock) 判断互斥量是否被锁住
- void mutex_lock(&name) 获得锁，task除了kill信号不可中断
- void mutex_unlock(&name) 解锁
- int mutex_trylock(&name) 尝试获得锁，成功返回1，不能加锁就立刻返回0
- mutex_lock_interruptible 获得锁，task可以被中断
- mutex_lock_killable 获得锁，task可以被kill信号中断
