(转)linux的同步与互斥

liebo

临界区：（critical region）
所谓临界区就是访问和操作共享数据的代码段。
并发有伪并发（单处理器）和真并发（多处理器）之分，但是都会造成竞争条件。
同步：（synchronization）
避免并发（多个执行线程并发访问同一个资源）和防止竞争条件（两个执行线程处于同一临界区）被称为同步。
用户空间之所以需要同步，是因为用户程序会被调度程序抢占和重新调度。
造成并发执行的原因：
1 中断
2 软中断和tasklet
3 内核抢占——任务的优先级
4 睡眠及用户空间的同步
5 SMP
在编写代码的开始阶段就应该设计恰当的锁
我们要给数据而不是代码加锁。大多数内核数据结构都需要加锁！而局部自动变量，动态分配的数据结构不需要加锁。
Linux信号量的实现
0.概念：
信号量: 一个信号量本质上是一个整数值, 它和一对函数联合使用, 这一对函数通常成为P和V, 也就是我们所说的P/V操作.
· 当一个进程希望进入临界区时, 对临界区的信号量执行P操作:
· 如果信号量的值大于0, 则该值会减1, 而进程可以继续;
· 如果信号量的值等于0(或更小), 进程必须等待直到其他人释放该信号量.
· 当一个进程完成对临界区的操作时, 对临界区信号量执行V操作:
将信号量的值加1, 并在必要时唤醒等待的进程.
· 在这种模式下, 一个信号量有时也称为一个"互斥体"(mutex), 它是互斥的简称.
规则:
· 当信号量用于互斥时(即避免多个进程同时操作一个临界区), 信号量的值应该初始化为1.
· 信号量在任何时刻只能由单个进程或线程拥有.
1. 定义:
头文件:
数据类型: struct semaphore
直接创建:
void sema_init(struct semaphore *sem, int val); /* 其中val是信号量的初始值 */
辅助宏:
DECLARE_MUTEX(name);                 /* 把一个称为name的信号量变量初始化为1 */
DECLARE_MUTEX_LOCKED(name); /* 把一个称为name的信号量变量初始化为0 */
动态分配:
/* 用于运行时的初始化 */
void init_MUTEX(struct semaphore *sem);
void init_MUTEX_LOCKED(struct semaphore *sem);
在Linux世界中, P函数被称为down, 指的是该函数减小了信号量的值, 它也许会将调用者置于休眠状态, 然后等待信号量变得可用, 之后再授予调用者对被保护资源的访问权限. down函数有三个版本:
void down(struct semaphore *sem);     /* 减小信号量的值, 并在必要时一直等待。无法获得信号量时，会导致调用进程休眠*/
作为通常规则, 我们不应该使用非中断版本. 非中断操作是建立不可杀进程的好方法（ps输出中的"D state")
void down_interruptible(struct semephore *sem);  /* 可中断版本, 常用。 无法获得信号量时，会导致调用进程休眠*/
可中断版本几乎是我们始终要使用的版本。
使用该函数, 如果得到，返回0；如果操作被中断, 该函数会返回非0值(-EINTR), 而调用者不会拥有该信号量.
因此对该函数的正确使用需要始终检查返回值, 并做出相应的响应.
void down_trylock(struct semaphore *sem);
/* 永远不会休眠, 如信号量在调用时不可获得, 立即返回非0值 */
后来的内核又增加了：
int down_killable(struct semaphore *sem);
/*无法获得信号量时，会导致调用进程休眠*，操作可被fatal信号中断, 函数会返回非0值(-EINTR), 而调用者不会拥有该信号量. */
int down_timeout(struct semaphore *sem, long jiffies);
/*在指定时间内获取该信号量，会导致调用进程休眠，无法获取时返回-ETIME*/
当一个线程成功调用down函数的某个版本之后, 就称为该线程拥有了该信号量, 可以访问被该信号量保护的临界区. 当互斥操作完成后, 必须释放该信号量.
Linux的V函数是up:
/* 调用up之后, 调用者不再拥有该信号量 */
void up(struct semaphore *sem);
2. 举例:
我们拥有一个共享数据结构:
struct st_data
{
    char name[32];
    char data[128];
    int data_len;
};
这个数据结构被多个进程同时访问.
为了避免这些进程在访问该结构时产生竞态, 我们在该结构的底部为其加上信号量:
struct st_data
{
    char name[32];              /* name */
    char data[128];              /* data */
    int data_len;                  /* data length */
    struct semaphore sem; /* semaphore */
};
信号量在使用前必须进行初始化, 而且是在共享数据其他部分可用前初始化. 因此, 我们在其他数据赋值之前调用init_MUTEX, 否则会建立一个竞态, 即在信号量准备好之前, 有代码可能会访问它们.
st_data data;
init_MUTEX(&data->sem);
setup_data(&data); /* 初始化数据 */
...
接下来, 我们必须仔细检查代码, 确保在不拥有该信号量的时候不会访问data数据结构. 例如, 在data_write的开始处加入:
if (down_interruptible(&data->sem))   
     return -ERESTARTSYS;
这是检查down_interruptible的返回值, 如果返回非0值, 说明操作被中断. 这种情况下, 通常要做的工作是返回-ERESTARTSYS. 在得到这个返回值后, 内核会从头重新启动该调用, 或者将该错误返回给用户.
如果我们返回-ERESTARTSYS, 则必须首先撤销已经做出的修改, 这样, 系统调用才可正确重试. 如果无法撤销这些操作, 则应该返回-EINTR, 表明中断.
不管data_write能否完成其他工作, 它都必须释放信号量:
out:
    up(&data->sem);
    return retval;
在data_write中有几个地方可能会产生错误, 包括内存分配失败等. 在这些情况下, 代码会执行goto out, 确保正确的完成信号量的释放工作.
读取/写入信号量
信号量对所有调用者执行互斥操作, 而不管线程想做什么(读/写). 正如这样, 我们可以把任务划分为两种类型: 读取和写入. 多个并发的读取应该是被允许的, 因为只读任务可并行完成他们的工作, 这样做可以大大提高性能.
如此, 便有了Linux内核提供的一种特殊信号量类型, 称为"rwsem"(reader/writer semaphore). 虽然在驱动程序中使用rwsem的机会相对比较少, 但偶尔也比较有用.
头文件:
数据类型: struct rw_semaphore
一个rwsem对象必须用一下函数显式地初始化:
void init_rwsem(struct rw_semaphore *sem);
对受保护资源的只读访问, 可和其他读取者并发地访问. 可用的接口有:
void down_read(struct rw_semaphore *sem);/* 可能会将调用进程置于不可中断的休眠 */
int down_read_trylock(struct rw_semaphore *sem);/* 不会阻塞等待，授予访问时返回非0, 其他情况返回0  注意他的返回值用法和其它函数的不同*/
void up_read(struct rw_semaphore *sem); /* 释放rw_sem对象 */
针对写入者的接口类似于读取者:
void down_write(struct rw_semaphore *sem);
int down_write_trylock(struct rw_semaphore *sem);
void up_write(struct rw_semaphore *sem);
这3个函数与读取者的对应函数行为相同, 他们提供的是写入访问.
/* 当某个快速改变获得写入者锁,
  *其后有更长时间的只读访问的时候,
  *我们可以调用该函数来允许其他读取者访问 */
void downgrade_write(struct rw_semaphore *sem);
读取/写入信号量特点:
· 一个rwsem可以允许一个写入者或无限多个读取者拥有该信号量.
· 写入者具有更高的优先级, 某个写入者试图进入临界区, 在其完成工作之前, 不会允许读取者获得访问.
· 如果大量写入者竞争该信号量, 会导致读取者"饿死".
· 最好在很少需要写访问且写入者只会短期拥有信号量什使用rwsem.
Completion（完成量）
有时, 由于信号量轻松up操作之后, 会产生过长时间的操作, 为了避免信号量执行down操作导致的长时间阻塞, 内核提供了一组completion(完成)接口解决这种问题.
Completion是一种轻量级的机制, 它允许一个线程告诉另一个线程某个工作已经完成.
头文件:
数据类型: struct completion
一个completion对象可以通过两种方法创建和初始化:
DECLARE_COMPLETION(my_completion);
或者, 如果必须动态, 则使用:
struct completion my_completion;
/* ... */
init_completion(&my_completion);
对completion操作的接口有:
等待completion的操作可以用:
/* 执行一个非中断等待 */
void wait_for_completion(struct completion *c);
实际的completion事件调用可以用:
/* 只会唤醒一个等待线程 */
void complete(struct completion *c);
/* 唤醒所有等待线程 */
void complete_all(struct completion *c);
但在大多数情况下, 只会有一个等待者, 此时两个函数效果相同.
一个completion通常是一个单次(one-shot)设备, 它只会被使用一次然后被丢弃.触发事件明确时, 如果没有使用complete_all, 可以重复使用一个completion结构.如果使用了complete_all, 则必须在重复使用该结构之前重新初始化它.
可以用下面的宏进行重新初始化Completion
INIT_COMPLETION(struct completion *c) ;
Completion机制的典型是模块退出时的内核线程终止. 这时, 驱动程序内部由一个while(1)循环完成, 当内核准备清楚该模块时, exit函数会令线程退出并等待completion. 此时, 内核可调用一个特殊函数:
void complete_and_exit(struct completion *c, long retval);
/* 其中retval是exit的返回值, 可用于错误代码检查 */
linux中自旋锁（spinlock）的实现
信号量对互斥来讲是非常有用的, 但它并不是内核提供的唯一工具.
大多数锁定通过称为"自旋锁"的机制实现. 和信号量不同, 自旋锁可以在不能休眠的代码中使用, 比如: 中断处理例程. 在正确使用的情况下, 自旋锁通常可以提供比信号量更高的性能, 但它也有一组不同的使用限制.
1. 概念:
自旋锁在概念上非常简单, 一个自旋锁就是一个互斥设备, 它只有两个值: "锁定"和"解锁". 它通常实现为某个整数值中的单个位.
· 希望获得某特定锁的代码, 测试相关的位. (原子操作)
· 如果锁可用, 则"锁定"位被设置, 而代码继续进入临界区.
· 相反, 如果锁被其他人获得, 则代码进入忙循环并重复检查这个锁, 直到该锁可用为止.
这个循环就是自旋锁的"自旋"部分. 当存在自旋锁时, 等待执行忙循环的处理器做不了任何有用的工作.
核心原则：任何拥有自旋锁的代码都必须是原子的。它不能休眠。
另外一个重要原则：自旋锁必须在可能的最短时间内拥有。
在拥有锁时，要注意调用的每一个函数是否会引起睡眠。是否会有中断服务程序获得相同的锁（需禁止本地CPU中断）
2. 自旋锁基本API:
头文件:
数据类型: spinlock_t
初始化, 以下两种方法:
      静态（编译时）: spinlock_t my_lock = SPIN_LOCK_UNLOCKED;
      动态（运行时）: void spin_lock_init(spinlock_t *lock);
获得锁: void spin_lock(spinlock_t *lock);
释放锁: void spin_unlock(spinlock_t *lock);
3. 自旋锁其他API:
锁定一个自旋锁的函数实际有4个:
void spin_lock(spinlock_t *lock);     /* 基本函数 */
void spin_lock_irqsave(spinlock_t *lock, unsigned long flags);/* 会在获得自旋锁之前禁止本地中断, 而先前的中断状态保存在flags中 */
void spin_lock_irq(spinlock_t *lock);/* 同上, 但无跟踪标志 */
void spin_lock_bh(spinlock_t *lock);/* 在获得锁之前禁止软件中断 */
对应的, 也有4个unlock函数:
void spin_unlock(spinlock_t *lock);
/* 这里面的flags必须和传递给spinlock_lock_irqsave的是同一个变量,
* 还必须在同一函数中调用, 否则代码可能在某些架构上出现问题 */
void spin_unlock_irqstore(spinlock_t *lock, unsigned long flags);
void spin_unlock_irq(spinlock_t *lock);
void spin_unlock_bh(spinlock_t *lock);
同时, 还有2个非阻塞的自旋锁操作:
int spin_trylock(spinlock_t *lock);      /*成功获得返回非0，否则返回0*/
int spin_trylock_bh(spinlock_t *lock); /*成功获得返回非0，否则返回0*/
对禁止中断的情况, 没有对应的"try"版本.
注意：《linux内核设计与实现》9.2节 自旋锁 中论述：
自旋锁更多的是为多处理器提供防止并发访问所需要的保护机制。
在单处理器机器上，编译的时候并不会加入自旋锁。它仅仅被当作一个设置内核抢占机制是否被启用的开关。如果禁止内核抢占，那么在编译时自旋锁会被完全剔除出内核。
观察内核代码发现：在自旋锁与CONFIG_SMP和CONFIG_PREEMPT两个配置选项有很大的关联。
而在自旋锁的各个操作函数中都会调用preempt_disable()来禁止抢占（若CONFIG_PREEMPT选项关闭，则preempt_disable为空操作）
内核抢占代码使用自旋锁作为非抢占区域的标记。
Linux 2.4.x及以前的版本都是非抢占式内核方式，如果编译成单处理器系统，在同一时间只有一个进程在执行，除非它自己放弃，不然只有通过"中断"才能中断其执行。因此，在单处理器非抢占式内核中，如果需要修改某个重要的数据结构，或者执行某些关键代码，只需要禁止中断。
但是在对称多处理器，仅仅禁止某个CPU的中断是不够的，当然我们也可以将所有CPU的中断都禁止，但这样做开销很大，整个系统的性能会明显下降。
此外，即使在单处理器上，如果内核是抢占式的，也可能出现不同进程上下文同时进入临界区的情况。为此，Linux内核中提供了"自旋锁（spinlock）"的同步机制。
自旋锁和信号量的比较
需求
建议的加锁方法
低开销加锁
优先使用自旋锁
短期加锁
优先使用自旋锁
长期加锁
优先使用信号量
中断上下文加锁
使用自旋锁
持有锁需要睡眠
使用信号量
读取/写入自旋锁
头文件:
数据类型: rwlock_t
初始化, 以下两种方法:
rwlock_t my_lock = RW_LOCK_UNLOCKED;    /*Static way*/
rwlock_t my_lock;
rw_lock_init(&my_lock);  /*Dynamic way*/
读取相关的自旋锁的函数:
void read_lock(rwlock_t *lock);
void read_lock_irqsave(rwlock_t *lock, unsigned long flags);
void read_lock_irq(rwlock_t *lock);/* 同上, 但无跟踪标志 */
void read_lock_bh(rwlock_t *lock);/* 在获得锁之前禁止软件中断 */
void read_unlock(rwlock_t *lock);
voidread_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
voidread_unlock_irq(rwlock_t *lock);
voidread_unlock_bh(rwlock_t *lock);
没有read_trylock函数可用。
写入相关的自旋锁的函数:
void write_lock(rwlock_t *lock);
voidwrite_lock_irqsave(rwlock_t *lock, unsigned long flags);
voidwrite_lock_irq(rwlock_t *lock);/* 同上, 但无跟踪标志 */
voidwrite_lock_bh(rwlock_t *lock);/* 在获得锁之前禁止软件中断 */
voidwrite_unlock(rwlock_t *lock);
voidwrite_unlock_irqrestore(rwlock_t *lock, unsigned long flags);
voidwrite_unlock_irq(rwlock_t *lock);
void write_unlock_bh(rwlock_t *lock);
排队自旋锁(FIFO Ticket Spinlock)
排队自旋锁(FIFO Ticket Spinlock)是 Linux 内核 2.6.25 版本中引入的一种新型自旋锁，它解决了传统自旋锁由于无序竞争导致的“公平性”问题。
参考：
Linux 内核的排队自旋锁(FIFO Ticket Spinlock)
内核处理方式我的理解如下：
将自旋锁的计数整数分成两个域：Next域和Owner域，两个域的位数相同。分别保存未来锁申请者和当前锁持有者的票据序号(Ticket Number)，初始时Next = Owner = 0。
当申请者申请排队自旋锁时，它会获得一个票据号，是当前Next值，然后Next增1；然后比较申请者的票据号和Owner值，若相等，则申请者可以获得该锁，否则忙等待；
当申请者处理完后释放排队自旋锁时，会使该锁的Owner值增1。等待的下一个申请者会发现这一变化，从忙等待状态中退出，并获得该锁。
线程将严格地按照申请顺序拿到Next票据号，然后随着Owner的增加，票据号 = Owner的申请者依次获取排队自旋锁，从而完全解决了“不公平”问题。
排序自旋锁和银行的拿号排队的原理是相同的。
死锁
产生死锁的原因：资源竞争以及进程推进顺序非法
产生死锁的4个必要条件：
· 互斥条件
· 请求保持条件
· 不可剥夺条件
· 环路条件
死锁的预防：预先静态分配法 和 资源有序分配法
免锁方法
1、循环缓冲区——免锁算法
在设备驱动程序中使用相当普遍，特别是网络适配器（处理交换数据）
一个生产者将数据放入数组的结尾，消费者从数组的另一端移走数据，当到达数组尾部时，生产者绕回到数组头部。没有多个生产者或消费者的情况下，循环缓冲区不需要加锁。
内核中有一个通用的循环缓冲区实现，参阅
参考：
Linux设备驱动程序学习（3-补）-Linux中的循环缓冲区
2、原子变量
有时，共享的资源可能恰好是个简单的整数值。
内核提供了一种原子的整数类型，称为：atomic_t
定义在：
这是一个与CPU架构有关的变量。
atomic_t数据项必须只能通过下面的函数来访问
初始化：
void atomic_set(atomic_t *v, int i);
atomic_t v = ATOMIC_INIT(0);
读取变量值：
int atomic_read(atomic_t *v);
增减操作：（注意不返回操作后的结果）
void atomic_add(int i, atomic_t *v);  /* v=+i */
void atomic_sub(int i, atomic_t *v); /* v=-i */
void atomic_inc(atomic_t *v);   /* v++ */
void atomic_dec(atomic_t *v);  /* v-- */
附加操作：
操作后并测试原子值，若为0，则返回true，否则返回false
int atomic_inc_and_test(atomic_t *v);  
int atomic_dec_and_test(atomic_t *v);
int atomic_sub_and_test(int i, atomic_t *v);
操作后并测试原子值，若为负，则返回true，否则返回false
int atomic_add_negative(int i, atomic_t *v);
操作后并返回原子值给调用者
int atomic_add_return(int i, atomic_t *v);
int atomic_sub_return(int i, atomic_t *v);
int atomic_inc_return(atomic_t *v);
int atomic_dec_return(atomic_t *v);
3、位操作
内核提供了一组可原子地修改和测试单个位的函数。
原子位操作只要硬件允许，就可以使用单个机器指令来执行，并且不需要禁止中断。所以位操作也是依赖于具体的架构，在中声明
nr通常定义为int型或unsigned long；addr通常是unsigned long* 或 void*
void set_bit(nr, void *addr);   /*设置addr指向的数据项的第nr位*/
void clear_bit(nr, void *addr); /*清除addr指向的数据项的第nr位*/
void change_bit(nr, void *addr);/*切换addr指向的数据项的第nr位*/
test_bit(nr, void *addr);/*返回指定位的当前值，唯一不必原子方式实现的位操作函数*/
int test_and_set_bit(nr, void *addr);  /*设定指定位，并返回指定位的先前值*/
int test_and_clear_bit(nr, void *addr);/*清除指定位，并返回指定位的先前值*/
int test_and_change_bit(nr, void *addr);/*切换指定位，并返回指定位的先前值*/
在新代码中应该优先使用自旋锁，因为自旋锁已经很好的调试过，并且易读性比位操作更好。
4、顺序锁（seqlock）
当要保护的资源很小、很简单、会频繁被访问而且写入访问很少发生且必须快速时，就可以使用seqlock。
seqlock通常不能用于保护包含有指针的数据结构。
seqlock在中定义。
seqlock_t lock1 = SEQLOCK_UNLOCKED;
seqlock_t lock2;
seqlock_init(&lock2);
seqlock允许读取者对资源的自由访问，但需要读取者检查是否和写入着发生冲突，冲突发生时，需要重试访问资源。读取者的代码如下：
unsigned int seq;
do {
       seq = read_seqbegin(&the_lock);    /* 获得一个（无符号的）整数顺序值seq*/
      /* Do what you need to do */
} while read_seqretry(&the_lock, seq);  /*退出时，seq和当前值比较，若不等则重试*/
在中断处理例程中，使用IRQ安全版本：
unsigned int read_seqbegin_irqsave(seqlock_t *lock, unsigned long flags);
int read_seqretry_irqrestore(seqlock_t *lock, unsigned int seq, unsigned long flags);
写入者在进入由seqlock保护的临界区时获得一个互斥锁。（写入锁使用自旋锁实现）
void write_seqlock(seqlock_t *lock);
释放锁：
void write_sequnlock(seqlock_t *lock);
常用变种：
void write_seqlock_irqsave(seqlock_t *lock, unsigned long flags);
void write_seqlock_irq(seqlock_t *lock);
void write_seqlock_bh(seqlock_t *lock);
void write_sequnlock_irqrestore(seqlock_t *lock, unsigned long flags);
void write_sequnlock_irq(seqlock_t *lock);
void write_sequnlock_bh(seqlock_t *lock);
5、读取-复制-更新（read-copy-update，RCU）
一种高级的互斥机制，它针对经常发生读取而很少写入的情形的优化。被保护的资源应该通过指针访问，而对这些资源的访问应该通过指针进行，而对这些资源的引用必须仅有原子代码拥有。在需要修改该数据结构时，写入线程首先复制，然后修改副本，子后用新的版本替代相关指针。当内核确信老的版本上没有其他引用时，可以释放老的版本。
参见RCU的白皮书
(
http://www.rdrop.com/users/paulmck/rclock/intro/rclock_intro.html
).
以及内核头文件
应用实例参考网络路由表
参考文献：
1.Linux Device Driver 3rd
2.Linux Kernel Develoment 2nd
3.
Linux内核的同步机制


本文来自ChinaUnix博客，如果查看原文请点：http://blog.chinaunix.net/u3/105293/showart_2132341.html