5、并发控制

 

一、并发与竞态

并发是指一段时间内有多个程序执行，但任一个时刻点上只有一个程序在运行

并发就会导致一个问题：假设程序A对一个文件写入3000个字符“a”，而另一个程序B对这个文件写入3000个“b”，第三个程序C读取这个文件，会导致读取数据不一定是什么

因为可能在一段时间内先执行了A；当A执行到一半CPU切换到执行B了，这时就会导致数据混乱

 

解决这个问题的途径是保证对共享资源的互斥访问。如程序A向文件中写入字符，那么B就无法访问这个文件

访问共享资源的代码区称为临界区，它需要使用互斥机制保护，途径有中断屏蔽、原子操作、自旋锁、信号量和互斥体等

 

在第二章点亮LED 中简述过的readb()类函数就使用了内存屏蔽指令

 

 

二、中断屏蔽

CPU一般都具备屏蔽中断和打开中断的功能，这项功能可以保证正在执行的内核执行路径不被中断处理程序所抢占，防止某些竞态条件的发生。具体而言，中断屏蔽使得中断与进程之间的并发不再发生，而且，由于Linux内核的进程调度等操作都依赖中断来实现，内核抢占进程之间的并发也得以避免了

使用方法如下：

local_irq_disable() /* 屏蔽中断 */
...
/* 临界区*/
...
local_irq_enable() /* 开中断*/

 

为了保证系统运行，中断屏蔽的时间尽量要少

 

 

三、原子操作

原子操作可以保证对一个整型数据的修改是排他性的。原子操作系列函数使用方法如下：

atomic_t v = ATOMIC_INIT(0);    /* 定义原子变量v = 0 */

atomic_read(&v);    /* 读取原子变量值 */

atomic_add(i, &v);    /* 原子变量 + i */
atomic_sub(i, &v);    /* 原子变量 - i */

atomic_inc(&v);    /* 原子变量 + 1 */
atomic_dec(&v);    /* 原子变量 - 1 */

int atomic_inc_and_test(&v);    /* 原子变量自加并测试是否为0，为0返回1 */
int atomic_dec_and_test(&v);    /* 原子变量自减并测试是否为0 */
int atomic_sub_and_test(i, &v);    /* 原子变量 - i后测试是否为0 */

 

示例如下： 

/* 使用原子变量使设备只能被一个进程打开 */
static atomic_t v = ATOMIC_INIT(1);

static int key_open(struct inode *nodep, struct file *filp)
{
    if (!atomic_dec_and_test(&v)) /* 已经有进程打开 */ {
        atomic_inc(&v);        /* 重新加1 */
        return -EBUSY;
    }
    ...
}

static int key_release(struct inode *nodep, struct file *filp)
{
    atomic_inc(&v);
    ...
}

 

 

四、自旋锁

自旋锁（Spin Lock）是一种典型的对临界资源进行互斥访问的手段，工作原理是测试并设置（Test-And-Set）某个内存变量，如果此变量符合条件则退出

它有以下几个特点：

1. 自旋锁是忙等锁。若锁不可用，CPU会一直循环执行检测，相当于while(1)。因此适用于临界区代码小的情况

2. 可能导致系统死锁。比如递归调用自旋锁

3. 锁定期间不能调用可能导致进程调度的函数。如copy_to_user()和copy_from_user()

 

自旋锁系列函数使用方法如下：

spinlock_t lock;        /* 定义一个自旋锁*/
spin_lock_init(&lock);    /* 初始化一个自旋锁*/
spin_lock(&lock);        /* 获取自旋锁，保护临界区 */

/* 临界区*/

spin_unlock(&lock);        /* 解锁*/

 

示例如下：

/* 使用自旋锁使设备只能被一个进程打开 */
int count = 0;

static int key_open(struct inode *nodep, struct file *filp)
{
    spin_lock(&lock);
    if (count) /* 已经有进程打开 */ {
        spin_unlock(&lock);
        return -EBUSY;
    }
    ++count;
    spin_unlock(&lock);
    ...
}

static int key_release(struct inode *nodep, struct file *filp)
{
    spin_lock(&lock);
    --count;
    spin_unlock(&lock);
    ...
}

static int keys_init(void)
{
    spinlock_t lock;
    spin_lock_init(&lock);
    ...
}

 

 

五、信号量

信号量使用方式类似于原子操作，系列函数使用方法如下：

struct semaphore sem;    /* 定义一个信号量 */

/* 初始化信号量 = val */
void sema_init(struct semaphore *sem, int val);

/* 获得信号量，它会导致休眠，因此不能在中断上下文中使用 */
void down(struct semaphore *sem);

/* 获得信号量，进入休眠状态的进程能被信号打断 */
int down_interruptible(struct semaphore *sem);

/* 尝试获得信号量sem，如果能够立刻获得，它就获得该信号量并返回0，否则返回非0值
 * 它不会导致调用者睡眠，可以在中断上下文中使用
 */
int down_trylock(struct semaphore *sem);

/* 释放信号量 */
void up(struct semaphore *sem);

DEFINE_SEMAPHORE(sem)可以初始化信号量并赋值为1。如果是在Linux2.6.36版本以下，应使用DECLARE_MUTEX(sem)

 

示例如下：

/* 使用信号量使设备只能被一个进程打开 */
static DEFINE_SEMAPHORE(lock);

static int key_open(struct inode *nodep, struct file *filp)
{
    if (!down_trylock(&lock)) /* 已经有进程打开 */ {
        return -EBUSY;
    }
    ...
}

static int key_release(struct inode *nodep, struct file *filp)
{
    up(&lock);
    ...
}

 

 

六、互斥体

互斥体系列函数使用方法如下：

struct mutex lock;   /* 定义一个互斥体 */

mutex_init(&lock);   /* 初始化互斥体 */

/* 获取互斥体 */
void mutex_lock(struct mutex *lock);
int mutex_lock_interruptible(struct mutex *lock);
int mutex_trylock(struct mutex *lock);

/* 释放互斥体 */
void mutex_unlock(struct mutex *lock);

 

key源代码（本示例通过互斥体防止copy_to_user()函数竞态）：

#include <linux/module.h>
#include <linux/fs.h>
#include <linux/init.h>
#include <linux/cdev.h>
#include <linux/slab.h>
#include <linux/device.h>
#include <linux/irq.h>
#include <linux/interrupt.h>
#include <linux/wait.h>
#include <linux/timer.h>
#include <linux/gpio.h>
#include <linux/sched.h>
#include <linux/mutex.h>

#include <asm/uaccess.h>
#include <asm/irq.h>
#include <asm/io.h>

#include <mach/gpio.h>

#define KEY_MAJOR        255

struct pin_desc {
    int gpio;
    int val;
    char *name;
};

struct key_device {
    struct cdev cdev;
    wait_queue_head_t r_head;
    wait_queue_head_t w_head;
    struct mutex lock;
};

static struct pin_desc desc[4] = {
    { EXYNOS4_GPX3(2), 0x01, "KEY0" },
    { EXYNOS4_GPX3(3), 0x02, "KEY1" },
    { EXYNOS4_GPX3(4), 0x03, "KEY2" },
    { EXYNOS4_GPX3(5), 0x04, "KEY3" },
};

static int g_major = KEY_MAJOR;
module_param(g_major, int, S_IRUGO);

static struct key_device*    dev;
static struct class*        scls;
static struct device*        sdev;
static unsigned char        key_val;

static irqreturn_t key_interrupt(int irq, void *dev_id)
{
    struct pin_desc *pindesc = (struct pin_desc *)dev_id;
    unsigned int tmp;

    tmp = gpio_get_value(pindesc->gpio);

    /* active low */
    printk(KERN_DEBUG "KEY %d: %08x\n", pindesc->val, tmp);

    if (tmp)
        key_val = pindesc->val;
    else
        key_val = pindesc->val | 0x80;

    set_current_state(TASK_RUNNING);

    return IRQ_HANDLED;
}

static ssize_t key_read(struct file *filp, char __user *buf, size_t len, loff_t * loff)
{
    struct key_device *dev = filp->private_data;

    // 声明等待队列
    DECLARE_WAITQUEUE(rwait, current);    
    mutex_lock(&dev->lock);        // 上锁
    add_wait_queue(&dev->r_head, &rwait);

    // 休眠
    __set_current_state(TASK_INTERRUPTIBLE);
    mutex_unlock(&dev->lock);    // 解锁
    schedule();

    // 有数据
    mutex_lock(&dev->lock);        // 上锁
    copy_to_user(buf, &key_val, 1);
    mutex_unlock(&dev->lock);    // 解锁

    remove_wait_queue(&dev->r_head, &rwait);
    set_current_state(TASK_RUNNING);

    return 1;
}

static int key_open(struct inode *nodep, struct file *filp)
{
    struct key_device *dev = container_of(nodep->i_cdev, struct key_device, cdev);
    // 放入私有数据中
    filp->private_data = dev;

    int irq;
    int i, err = 0;

    for (i = 0; i < ARRAY_SIZE(desc); i++) {
        if (!desc[i].gpio)
            continue;

        irq = gpio_to_irq(desc[i].gpio);
        err = request_irq(irq, key_interrupt, IRQ_TYPE_EDGE_BOTH, 
                desc[i].name, (void *)&desc[i]);
        if (err)
            break;
    }
    
    if (err) {
        i--;
        for (; i >= 0; i--) {
            if (!desc[i].gpio)
                continue;

            irq = gpio_to_irq(desc[i].gpio);
            free_irq(irq, (void *)&desc[i]);
        }
        return -EBUSY;
    }
    
    init_waitqueue_head(&dev->r_head);
    mutex_init(&dev->lock);        // 初始化

    return 0;
}

static int key_release(struct inode *nodep, struct file *filp)
{
    // 释放中断
    int irq, i;

    for (i = 0; i < ARRAY_SIZE(desc); i++) {
        if (!desc[i].gpio)
            continue;

        irq = gpio_to_irq(desc[i].gpio);
        free_irq(irq, (void *)&desc[i]);
    }

    return 0;
}

static struct file_operations key_fops = {
    .owner    = THIS_MODULE,
    .read    = key_read,
    .open        = key_open,
    .release    = key_release,
};

static int keys_init(void)
{
    int ret;
    int devt;
    if (g_major) {
        devt = MKDEV(g_major, 0);
        ret = register_chrdev_region(devt, 1, "key");
    }
    else {
        ret = alloc_chrdev_region(&devt, 0, 1, "key");
        g_major = MAJOR(devt);
    }

    if (ret)
        return ret;

    dev = kzalloc(sizeof(struct key_device), GFP_KERNEL);
    if (!dev) {
        ret = -ENOMEM;
        goto fail_alloc;
    }

    cdev_init(&dev->cdev, &key_fops);
    ret = cdev_add(&dev->cdev, devt, 1);
    if (ret)
        return ret;

    scls = class_create(THIS_MODULE, "key");
    sdev = device_create(scls, NULL, devt, NULL, "key");

    return 0;

fail_alloc:
    unregister_chrdev_region(devt, 1);

    return ret;
}

static void keys_exit(void)
{
    dev_t devt = MKDEV(g_major, 0);

    device_destroy(scls, devt);
    class_destroy(scls);

    cdev_del(&(dev->cdev));
    kfree(dev);

    unregister_chrdev_region(devt, 1);
}

module_init(keys_init);
module_exit(keys_exit);

MODULE_LICENSE("GPL");

Makefile：

KERN_DIR = /work/tiny4412/tools/linux-3.5

all:
    make -C $(KERN_DIR) M=`pwd` modules 

clean:
    make -C $(KERN_DIR) M=`pwd` modules clean
    rm -rf modules.order

obj-m    += key.o

测试文件：

#include <stdio.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <fcntl.h>
#include <string.h>

int main(int argc, char** argv)
{
    int fd;
    fd = open("/dev/key", O_RDWR);
    if (fd < 0) {
        printf("can't open /dev/key\n");
        return -1;
    }

    unsigned char key_val;

    while (1) {
        read(fd, &key_val, 1);
        printf("key_val = 0x%x\n", key_val);
    }
    
    close(fd);

    return 0;
}

 

 

七、互斥体和自旋锁的选择

通过比较可知，互斥体的作用于自旋锁类似。那么什么时候使用互斥体，什么时候使用自旋锁呢？

1. 若临界区较小，宜使用自旋锁；若临界区较大，宜使用互斥体

2. 互斥体所保护的临界区可以出现包含阻塞（进程切换）的代码，但是自旋锁不可以

3. 互斥体存在于进程上下文；如果要在中断中使用，宜使用自旋锁

 

 

下一章  6、异步通知