Linux驱动程序设计之四---并发和竞态

http://stephanxu.blog.163.com/blog/static/17007707920101082833917/

UNIT 3  并发与竟态

1.  并发与竞争条件概述 2.  信号量与互斥体 3.  Completion 4.  原子操作 5.  Spinlock 6.  信号量 VS spinlock 7.  避免死锁的规则

 

1． 并发与竞争条件概述

在现代 Linux 系统, 有非常多的并发源, 并且因此而来的可能竞争情况.

ü  多个用户空间进程在运行, 它们可能以令人惊讶的方式组合存取你的代码.

ü  SMP 系统能够同时在不同处理器上执行你的代码.

ü  内核代码是可抢占的; 你的驱动代码可能在任何时间失去处理器, 代替它的进程可能也在你的驱动中运行.

ü  设备中断是能够导致你的代码并发执行的异步事件.

ü   内核也提供各种延迟代码执行的机制, 例如 workqueue, tasklet, 以及定时器,

这些能够使你的代码在任何时间以一种与当前进程在做的事情无关的方式运行. 在现代的, 热插拔的世界中, 你的设备可能在你使用它们的时候轻易地消失.

避免竞争情况可能是一个令人害怕的工作. 在一个任何时候可能发生任何事的世界, 驱动程序员如何避免产生绝对的混乱? 事实证明, 大部分竞争情况可以避免, 通过一些想法, 内核并发控制原语, 以及几个基本原则的应用. 我们会先从原则开始, 接着进入如何使用它们的细节中竞争情况来自对资源的共享存取的结果. 当 2 个执行的线路有机会操作同一个数据结构(或者硬件资源), 混合的可能性就一直存在.

因此第一个经验法则是在你设计驱动时在任何可能的时候记住避免共享的资源. 如果没有并发存取, 就没有竞争情况. 因此小心编写的内核代码应当有最小的共享. 这个想法的最明显应用是避免使用全局变量. 如果你将一个资源放在多个执行线路能够找到它的地方, 应当有一个很强的理由这样做.

事实是, 然而, 这样的共享常常是需要的. 硬件资源是, 由于它们的特性, 共享的, 软件资源也必须常常共享给多个线程. 也要记住全局变量远远不是共享数据的唯一方式; 任何时候你的代码传递一个指针给内核的其他部分, 潜在地它创造了一个新的共享情形. 共享是生活的事实.

这是资源共享的硬规则: 任何时候一个硬件或软件资源被超出一个单个执行线程共享, 并且可能存在一个线程看到那个资源的不一致时, 你必须明确地管理对那个资源的存取. 存取管理的常用技术是加锁或者互斥 -- 确保在任何时间只有一个执行线程可以操作一个共享资源.

然而, 首先, 我们必须简短考虑一下另一个重要规则. 当内核代码创建一个会被内核其他部分共享的对象时, 这个对象必须一直存在(并且功能正常)到它知道没有对它的外部引用存在为止.

遵照上面的规则需要计划和对细节小心注意. 容易被对资源的并发存取而吃惊, 你事先并没有认识到被共享. 通过一些努力, 然而, 大部分竞争情况能够在它们咬到你或者你的用户前被消灭.

我们对前面的写的ctest程序作一些修改，如下:

-----------------------------------------------------------------------

 //. . . . . . . 以上保持不变

static ssize_t ctest_write(struct file *file,char __user *buf,

size_t count,loff_t *offst)

{

  static char ctestbuf[256];   //仅仅是把ctestbuf变成了static的缓冲区

  int cnt;

  memset(ctestbuf,0,256);

  if(count<256)

    cnt = count;

  else

    cnt = 255;

  if(!copy_from_user(ctestbuf,buf,cnt)){ 

    printk(“%s\n”,ctestbuf); //考虑一下，如果在这一行执行前发生进程间切换，如何？

    return cnt;

}else{

  return -1;

}

}

// . . . . . . 以下保持不变

-----------------------------------------------------------------------

大家仔细分析上面的代码，我们只是将char ctestbuf[256]改称了static char ctestbuf[256], 这样这个空间将不是存储在函数的栈上，而是存储在了静态全局区，这样ctestbuf就变成了一个共享的内存资源。

当有多个进程对这个设备进行write操作的时候，比如P1向/dev/ctest设备写”hello,kernel”,P2向/dev/ctest写”hello,ctest”,请问最后的结果，printk是输出什么呢？

 

大家可以写下面的一段程序来测试:

-----------------------------------------------------------------------

#include

#include

#include

#include

#include

int main()

{

   int fd = fopen(“/dev/ctest”,O_RDWR);

   if(fd<0){

     perror(“open /dev/ctest failed\n”)

     exit(1);

}

if(fork()==0){     //子进程

   int cnt = 0xfffff;

   while(--cnt){

     write(fd,”hello,kernel”,13);

   }

   close(fd)

}else{            //父进程

  int cnt = 0xfffff;

  while(--cnt){

    write(fd,”hello,ctest”,12);

}

close(fd);

wait(NULL);

return 0;

}

}

 

-----------------------------------------------------------------------

    这里的答案实际上是不确定的，如果内核不支持抢占的话，P1进程将是让printk输出”hello,kernel”, P2进程将是让printk()输出”hello,ctest”.

但是如果支持抢占的话，那么结果就很难确定了，考虑一下极端情况，如果P1进程刚刚完成了数据从user buffer 到ctestbuf的拷贝工作，此时，进行了一次进程调度(如代码中指示的位置)，调度的结果如果是让P2进程执行，并且完成了write的操作，那么ctestbuf里面的数据将会是”hello,ctest”.这样当控制权再次回到P1的时候，printk的输出结果将是”hello,ctest”.同理P2进程也可能输出”hello,kernel”.这就是我们一直在讨论的竞争条件的情况。

那怎么样来解决这个问题，使P1总是输出”hello,kernel”,P2总是输出”hello,ctest”呢？请回顾我们在操作系统课程中学过的一些知识，下面我们就看一下Linux内核里面是如何实现的。

2．信号量和互斥体

    在学习操作系统理论的时候，我们提出了竞态的解决方案称之为临界区,临界区有三个特性：它的本质是一段代码，其二这段代码应该访问共享资源，其三，这段代码必须是互斥的。也就是说，一次只能有一个进程处在临界区中。Dijkstra提出了使用semaphore实现互斥的方法。

    Semaphore本质上是一个整形，可以执行两个操作，也就是通常所说的PV原语(原语的意思是说，P操作和V操作本身都是原子操作，原子操作意思是不可分割的操作，要么全部执行，要么全不执行)，P操作的意思是对信号量的值减一，如果结果是小于零的，那么表明资源被占用，进程应该进入blocked queue等待资源的释放，否则就继续执行。V操作是对信号量的值加一，如果结果小于等于零，表明有进程在此blocked queue中等待此资源，应该唤醒等待进程,把在此blocked队列等待的进程移到ready queue，否则表明没有进程在等待此资源。

    当信号量用于互斥时，有两种方式，一种是二元信号量及只有0,1之值。一种是初始值为1的信号量，并且总是成对使用。这种信号量在任何给定时刻只能由单个进程或线程拥有。在这种使用模式下，一个信号量有时也称为一个“互斥体(mutex)”，它是互斥(mutual exclusion)的简称。Linux内核中几乎所有的信号量均用于互斥。

    要使用一个信号量，首先应该#include  或者是include , 不同的体系结构使用的头文件的位置可能是不同的。我们使用struct semphore这个类型来表示信号量，有几种声明和初始化的方式，如下：

struct semaphore sem; void sema_init(struct  semaphore *sem,int val)	操作步骤： 1.  定义一个信号量 2.  使用sema_init函数初始化它，val的值代表初始值
DECLARE_MUTEX(name); DECLARE_MUTEX_LOCKED(name)	这里定义的就是一个互斥体，这两个宏是由内核提供的，在内核里面常见这样的方式，不光定义了struct semaphore name;还给出了初始值，他们的初始值是多少？
void init_MUTEX(struct  semaphore *sem); void init_MUTEX_LOCKED(structsemaphore *sem)	这里很明显是对于前面定义的sem进行初始化，而且是将之当成互斥体初始化，也就是说信号量的初始值非0既1。

 

在Linux的世界中，P操作对应的函数为down()，意思很明显，将信号量的值下降一点.在Linux里面，获得一个信号量的方法有三种，其意义差别很大：

void down(struct semaphore *)

int down_interruptible(struct semaphore *)

    都执行P操作，将信号量的值减1，如果信号量的值小于0，它将进入blocked 状态，只是这里需要注意：第二个函数比第一个函数多了interruptible,很明显，如果使用第二个函数进入的blocked状态,那么将是一个可中断的blocked状态，那么down()进入的就是一个不可中断的状态。可中断状态与不可中断状态的区别在于对信号的处理方式不同。

如果进程是因为down()进入的blocked状态，那么将不能被杀死，因为杀死进程就是向其发送一个信号，比如kill -9 PID.

使用down_interruptible需要额外小心，如果操作被中断，该函数会返回非零值，而调用者不会拥有该信号量。对down_interruptible的正确使用需要始终检查返回值，并作出相应的响应，如果返回值为非0，通常立即返回-ERESTARTSYS,如：

if(down_interruptible(&sem)){

    return –ERESTARTSYS;

}

 

int down_trylock(struct semaphore *)

此函数一样是执行P操作，只是这里带有try的意思，那也就是尝试获得，换句话说，它可能会获得信号量，也可能信号量此时不可用。当信号量不可用的时候，调用此函数的进程将不会等待，也就是说不会进入blocked 状态，而是恢复信号量的原有值，并继续执行。

既然此函数不会导致进程等待，而是继续执行，那么它的返回值就应该告知此函数是否获得了信号量。当返回为0时，表明获得信号量，所以在退出临界区的时候需要释放，否则将无须此操作。

if(!down_trylock(&sem)){

  ...

  up(&sem);

}

 

当一个线程成功调用上述down的某个版本之后，就称为该线程“拥有”(或“拿到”、“获取”)了该信号量。这样，该线程就被赋予访问由该信号量保护的临界区的权利。

不管你用何种方式获得了semaphore,释放的操作都是一样的，这里释放的操作也就对应sempahore的V操作。

void up(struct semaphore *sem)；

调用up之后，调用者不再拥有该信号量。

 

如读者所料，任何拿到信号量的线程都必须通过一次(只有一次)对up的调用而释放该信号量。在出现错误的情况下，经常需要特别小心；如果在拥有一个信号量时发生错误，必须在将错误状态返回给调用者之前释放该信号量。我们很容易犯忘记释放信号量的错误，而其结果(进程在某些无关位置处被挂起)很难复现和跟踪。信号量一般这样被使用，如下所示：

//定义信号量

DECLARE_MUTEX(sem)；

down(&sem);//获取信号量，保护临界区

…                         

Critical section//临界区

…                         

up(&sem);//释放信号量

    这样，我们可以使用semaphore让我们开篇的代码，不管是否是抢占式，还是非抢占式内核下都可安全的运行:

-----------------------------------------------------------------------

DECLARE_MUTEX(sem);

 //. . . . . . . 以上保持不变

static ssize_t ctest_write(struct file *file,char __user *buf,

size_t count,loff_t *offst)

{

  static char ctestbuf[256];   //仅仅是把ctestbuf变成了static的缓冲区

  int cnt;

  if(down_interruptible(&sem)){

  return –ERESTARTSYS;

}

  memset(ctestbuf,0,256);

  if(count<256)

    cnt = count;

  else

    cnt = 255;

  if(!copy_from_user(ctestbuf,buf,cnt)){ 

    printk(“%s\n”,ctestbuf); //考虑一下，如果在这一行执行前发生进程间切换，如何？

    up(&sem);

return cnt;

}else{

  up(&sem);

  return -1;

}

}

// . . . . . . 以下保持不变

-----------------------------------------------------------------------

读写信号量

    Linux内核提供了一套机制，直接解决了我们在学习操作系统的时候提到的读者/写者模型。Linux设计了一个struct rw_semaphore的类型。当我们需要使用读写信号量的时候，需要#include ;一般说来在驱动中较少使用：

初始化：

struct rw_semaphore rwsem;

void init_rwsem(struct rw_semaphore *);

此函数用来初始化一个读写信号量。

 

读锁操作：

void down_read(struct rw_semaphore *sem);

int down_read_trylock(struct rw_semaphore *sem);

对down_read的调用提供了对受保护资源的只读访问，可和其他读取者并发地访问。注意，down_read可能会将调用进程置于不可中断的休眠。down_read_trylock不会在读取访问不可获得时等待；它在授予访问时返回非零，其他情况下返回零。注意，

down_read_trylock的用法和其他大多数内核函数不同，其他函数会在成功时返回零。由down_read获得的rwsem对象最终必须通过up_read被释放。

 

void up_read(struct rw_semaphore *sem);

用于读者程序，用来来释放读锁

写锁操作：

void down_write(struct rw_semaphore *sem);

int down_write_trylock(struct rw_semaphore *sem);

void up_write(struct rw_semaphore *sem);

void downgrade_write(struct rw_semaphore *sem);

down_write、down_write_trylock和up_write与读取者的对应函数行为相同，当然，它们提供的是写入访问。当某个快速改变获得了写入者锁，而其后是更长时间的只读访问的话，我们可以在结束修改之后调用downgrade_write，来允许其他读取者的访问。

 

 

一个rwsem可允许一个写入者或无限多个读取者拥有该信号量。写入者具有更高的优先级；当某个给定写入者试图进入临界区时，在所有写入者完成其工作之前，不会允许读取者获得访问。如果有大量的写入者竞争该信号量，则这种实现会导致读取者“饿死”，即可能会长期拒绝读取者的访问。为此，最好在很少需要写访问且写入者只会短期拥有信号量的时候使用rwsem。

读写信号量一般可以如下方法使用：

rw_semaphore_t rw_sem; //定义读写信号量

init_rwsem(&rw_sem);    //初始化读写信号量

down_read(&rw_sem); //读时获取信号量

… //临界资源

up_read(&rw_sem);

down_write(&rw_sem);  //写时获取信号量

… //临界资源

up_write(&rw_sem);

 

 

3．Completion

    内核驱动是被所有的用户程序所共享的，而用户程序之间可能经常需要做同步操作，比如我们以前学过的管道就是如此，当父进程向管道中写数据时，如果管道已经写满了，那么父进程应该阻塞，或者当子进程读取管道时，管道的数据已经被读空了，那么子进程也应该阻塞。

    我们在这里稍微修改一下上面的例子，当写进程把数据都写到缓冲区之后，读进程才能开始读取数据，换句话说，当写进程的数据没有发送完之前，读进程应该处于等待状态，在这种情况下，completion就显得特别有用，并且使用简单明了。

    细心的同学肯定能够发现，当发生这种情况的时候，semaphore 也能够完成这个任务，确实如此，所以completion的机制，又被称为轻量级semaphore.更详细的解释还请参考<>的第五章节。

    下面我们解释一下，如何使用completion, 必须#include ,这里和semaphore一样，提供了一个特别的类型:struct completion

初始化:

struct completion xxx_completion;

init_completion(&xxx_completion)

或者采用宏，静态的初始化：

DECLARE_COMPLETION(xxx_completion);

当需要等待某个进程完成的时候可调用:

wait_for_completion(struct completion *)

注意，这也不是一个interruptible版本的等待。

当此事件完成的时候，可以对其发送完成的信号：

complete(struct completion *);

还有另外一个也用来表示完成:complete_all(struct completion *)，读者可自行分析。

示例如下:在unit4中，我们可以用来实现具体的实验：

-----------------------------------------------------------------------DECLARE_COMPLETION(xxx_comp)；

ssize_t complete_read(struct file *filp，

char __user *buf，

size_t count,loff_t *pos)

{

    wait_for_completion(&xxx_comp)；

    return 0;

}

ssize_t complete_write(struct file *filp，

const char __user *buf，

size_t count，loff_t *pos)

{

    complete(&xxx_comp)；

return count；

}

-----------------------------------------------------------------------

 

4. 原子操作

    原子操作即不可分割的操作，也就是说在执行这个原子操作期间是忽略中断的，这里需要特别注意:中断在计算机系统中的意义非凡，一般情况下是不能关中断执行的，至少不能长时间的忽略中断。那从这里也就可以得出: 此时的原子操作都是非常短暂的执行操作(注：这里的原子操作是真正意义上的原子操作，而不仅仅是临界区的概念)。

    Linux的原子操作分为两类，一类是对整形操作，一类是对位操作：

整形原子操作:

有时，共享的资源可能恰好是一个简单的整数值。假定我们的驱动程序维护着一个共享变量count，该变量的值表明有多少个设备操作正在并发地执行。通常，即使下面的简单操作也需要锁定：

count ++;

某些处理器可以以原子的方式执行这类增加，但我们不能指望它。但话又说回来，完整的锁机制对一个简单的整数来讲却显得有些浪费。针对这种情况，内核提供了一种原子的整数类型，称为atomic_t，定义在中。

一个atomic_t变量在所有内核支持的架构上保存一个int值。但是，由于某些处理器上这种数据类型的工作方式有些限制，因此不能使用完整的整数范围；也就是说，在atomic_t变量中不能记录大于24位的整数。

下面针对这种类型的操作在SMP计算机的所有处理器上都确保是原子的。这种操作的速度非常快，因为只要可能，它们就会被编译成单个机器指令。

void atomic_set(atomic_t *v，int i)； atomic_t v = ATOMIC_INIT(0)；	将原子变量v的值设置为整数值i。也可以在编译时利用ATOMIC_INIT宏来初始化原子变量的值。
int atomic_read(atomic_t *v)；	返回v的当前值。
void atomic_add(int i，atomic_t *v)；	将i累加到v指向的原子变量。返回值是void，这是因为返回新的值将带来额外的成本，而大多数情况下没有必要知道累加后的值。
void atomic_sub(int i，atomic_t *v)；	从*v中减去i。
void atomic_inc(atomic_t *v)； void atomic_dec(atomic_t *v)；	增加或缩减一个原子变量。
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)；	执行特定的操作并测试结果；如果在操作结束后，原子值为0，则返回值为true；否则返回值为false。注意，不存在atomic_add_and_test函数。
int atomic_add_negative(int i， atomic_t *v);	将整数变量i累加到v。返回值在结果为负时为true，否则为false。
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)；	类似于atomic_add及其变种，例外之处在于这些函数会将新的值返回给调用者。

 

先前说过，atomic_t数据项必须只能通过上述函数来访问。如果读者将原子变量传递给了需要整型参数的函数，则会遇到编译错误。

还要记住，只有原子变量的数目是原子的，atomic_t变量才能工作。需要多个atomic_t变量的操作，仍然需要某种类型的锁。考虑下面的代码：

atomic_sub(amount，&first_atomic)；

atomic_add(amount，&second_atomic)；

在amount已经从第一个原子值中减去，到还没有增加到第二个原子值之间，会有一小段时间。如果可能在这两个操作之间运行的代码会导致问题的发生，则必须使用某种形式的锁。

 

原子位操作:

void set_bit(nr, void *addr);	设置位： 上述操作设置addr地址的第nr位，所谓设置位即将位写为1.
void clear_bit(nr, void *addr);	清除位： 上述操作清除addr地址的第nr位，所谓清除位即将位写为0.
void change_bit(nr, void *addr);	改变位： 上述操作对addr地址的第nr位进行反置。
test_bit(nr, void *addr);	测试位： 上述操作返回addr地址的第nr位。
int test_and_set_bit(nr, void *addr); int test_and_clear_bit(nr, void *addr); int test_and_change_bit(nr, void *addr);	测试并操作位： 上述test_and_xxx_bit(nr, void *addr)操作等同于test_bit(nr, void *addr)后再执行xxx_bit(nr, void *addr)。