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、異步通知