学步园

用户类进程之间使用信号量（semaphore）进行同步，内核线程之间也使用了信号量。信号量与自旋锁类似，保护临界区代码。但信号量与自旋锁有一定的区别，信号量在无法得到资源时，内核线程处于睡眠阻塞状态，而自旋锁处于忙等待状态。因此，如果资源被占用时间很短时，使用自旋锁较好，因为它可节约调度时间。如果资源被占用的时间较长，使用信号量较好，因为可让CPU调度去做其它进程的工作。
操作信号量的API函数说明如表6。
表6 信号量API函数功能说明

样例：信号量的使用
下面函数do_utimes利用信号量防止多个线程对文件系统节点inode同时进行访问。其列出如下（在fs/open.c中）：

long do_utimes(char __user * filename, struct timeval *times)
{
    struct inode * inode;
    ……
    down(&inode->i_sem); //获取信号量
    error = notify_change(nd.dentry, &newattrs);//修改inode中值
    up(&inode->i_sem); //释放信号量
    ……
}

下面说明信号量API函数。

（1）信号量结构semaphore
信号量用结构semaphore描述，它在自旋锁的基础上改进而成，它包括一个自旋锁、信号量计数器和一个等待队列。用户程序只能调用信号量API函数，而不能直接访问信号量结构，其列出如下（在include/linux/semaphore.h中）：

struct semaphore {
    spinlock_t lock;
    unsigned int count;
    struct list_head wait_list;
};

（2）初始化函数sema_init
函数sema_init初始化信号量，将信号量值初始化为n，其列出如下：

static inline void sema_init(struct semaphore *sem, intval)
{
    static struct lock_class_key __key;
    *sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem,val);
    /*初始化一个锁的实例，用于调试中获取信号量的调试信息*/
    lockdep_init_map(&sem->lock.dep_map,"semaphore->lock", &__key, 0);
}
#define __SEMAPHORE_INITIALIZER(name, n) /
{ /
    .lock = __SPIN_LOCK_UNLOCKED((name).lock), / //初始化自旋锁
    .count = n, / //将信号量计数器赋值为n
    .wait_list = LIST_HEAD_INIT((name).wait_list), / //初始化等待队列
}

（3）可中断获取信号量函数down_interruptible

函数down_interruptible获取信号量，存放在参数sem中。它尝试获取信号量，如果其他线程被允许尝试获取此信号量，则将本线程睡眠等待。如果有一个信号中断睡眠，则它返回错误-EINTR。如果成功获取信号量，函数返回0。
函数down_interruptible列出如下（在kernel/semaphore.c中）：

int down_interruptible(struct semaphore *sem)
{
    unsigned long flags;
    int result = 0;

    spin_lock_irqsave(&sem->lock, flags);//获取自旋锁，关闭中断，将状态寄存器值存放在flags
    /*如果信号量计数器值大于0，说明有多个空闲资源可访问，可以成功获取信号量了*/
    if (likely(sem->count > 0)) //likely表示成功获取的概率大，通知编译器进行分支预测优化
        sem->count--;
    else
        result = __down_interruptible(sem);//进入睡眠等待
    spin_unlock_irqrestore(&sem->lock, flags);

    return result;
}

static noinline int __sched __down_interruptible(structsemaphore *sem)
{
    return __down_common(sem, TASK_INTERRUPTIBLE,MAX_SCHEDULE_TIMEOUT);
}

函数__down_common进入睡眠等待，其列出如下：

static inline int __sched __down_common(struct semaphore*sem, long state, long timeout)
{
    struct task_struct *task = current;
    struct semaphore_waiter waiter;

    list_add_tail(&waiter.list,&sem->wait_list); //加入到等待队列
    waiter.task = task;
    waiter.up = 0;

    for (;;) {
        if (state == TASK_INTERRUPTIBLE&& signal_pending(task))
            gotointerrupted;
        if (state == TASK_KILLABLE&& fatal_signal_pending(task))
            gotointerrupted;
        if (timeout <= 0)
            gototimed_out;
        __set_task_state(task, state);
        spin_unlock_irq(&sem->lock);
        timeout = schedule_timeout(timeout);//调度
        spin_lock_irq(&sem->lock);
        if (waiter.up)
            return 0;
    }

timed_out:
    list_del(&waiter.list);
    return -ETIME;

interrupted:
    list_del(&waiter.list);
    return -EINTR;
}

（3）释放信号量函数up

函数up在没有其他线程等待使用信号量的情况下释放信号量，否则，唤醒其他等待线程。其列出如下：

void up(struct semaphore *sem)
{
    unsigned long flags;

    spin_lock_irqsave(&sem->lock,flags);
    /*判断是否有线程等待在此信号量上，即判断等待队列是否为空*/
    if (likely(list_empty(&sem->wait_list))) 
        /*没有线程等待此信号量，释放信号量，将信号量计数器加1，表示增加了1个空闲资源*/
        sem->count++;
    else
        __up(sem); /*将本线程从等待队列删除，唤醒等待此信号量的其他线程*/
    spin_unlock_irqrestore(&sem->lock, flags);
}

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 = 1;
    wake_up_process(waiter->task); //唤醒等待此信号量的其他线程
}

互斥锁

信号量的初始值表示可以有多少个任务可同时访问的共享资源，如果初始值为1，表示只有1个任务可以访问，信号量变成互斥锁（Mutex）。可见互斥锁是信号量的特例。
互斥锁（mutex）是在原子操作API的基础上实现的信号量行为。互斥锁不能进行递归锁定或解锁，能用于交互上下文，同一时间只能有一个任务持有互斥锁。
互斥锁功能上基本上与信号量一样，互斥锁占用空间比信号量小，运行效率比信号量高。互斥锁的API函数功能说明如表1。

表1
互斥锁的API函数功能说明

互斥锁用结构mutex描述，它含有信号量计数和等待队列成员，信号量的值为1或0或负数。其列出如下（在include/linux/mutex.h中）：

struct mutex {
    /* 1：表示解锁，0：表示锁住，负数：表示锁住，可能有等待者*/
    atomic_t count;
    spinlock_t wait_lock; /*操作等待队列的自旋锁*/
    struct list_head wait_list; /*等待队列*/
    /*省略了用于调试的结构成员*/
};

读/写信号量

读/写信号量适于在读多写少的情况下使用。如果一个任务需要读和写操作时，它将被看作写者，在不需要写操作的情况下可降级为读者。任意多个读者可同时拥有一个读/写信号量，对临界区代码进行操作。
在没有写者操作时，任何读者都可成功获得读/写信号量进行读操作。如果有写者在操作时，读者必须被挂起等待直到写者释放该信号量。在没有写者或读者操作时，写者必须等待前面的写者或读者释放该信号量后，才能访问临界区。写者独占临界区，排斥其他的写者和读者，而读者只排斥写者。
读/写信号量可通过依赖硬件架构或纯软件代码两种方式实现。下面只说明纯软件代码实现方式。

（1）API说明

用户可通过调用读/写信号量API实现读/写操作的同步。读/写信号量API说明如表1。

表1
读/写信号量API函数功能说明

（2）读/写信号量结构rw_semaphore

读/写信号量结构rw_semaphore描述了读/写信号量的值和等待队列，其列出如下（在include/linux/rwsem-spinlock.h中）：
struct rw_semaphore {
    /*读/写信号量定义：
    * - 如果activity为0，那么没有激活的读者或写者。
    * - 如果activity为+ve，那么将有ve个激活的读者。
    * - 如果activity为-1，那么将有1个激活的写者。
*/
    __s32 activity; /*信号量值*/
    spinlock_t wait_lock; /*用于锁等待队列wait_list*/
    struct list_head wait_list; /*如果非空，表示有进程等待该信号量*/
#ifdef CONFIG_DEBUG_LOCK_ALLOC /*用于锁调试*/
    struct lockdep_map dep_map;
#endif
};

（3）读者加锁/解锁操作实现分析

1）加读者锁操作
读者加锁函数down_read用于加读者锁，如果没有写者操作时，等待队列为空，读者可以加读者锁，将信号量的读者计数加1。如果有写在操作时，等待队列非空，读者需要等待写者操作完成。函数down_read列出如下（在kernel/rwsem.c中）：

void __sched down_read(struct rw_semaphore *sem)
{
    might_sleep(); /*用于调试自旋锁睡眠*/
    rwsem_acquire_read(&sem->dep_map, 0, 0, _RET_IP_); /*确认获得锁，用于调试*/
    /*跟踪锁状态信息（如：锁深度），用于调试*/
    LOCK_CONTENDED(sem, __down_read_trylock, __down_read);
}

函数__down_read
完成加读者的具体操作，其列出如下（在lib/rwsem-spinlock.c中）：
void __sched __down_read(struct rw_semaphore *sem)
{
    struct rwsem_waiter waiter;
    struct task_struct *tsk;

    spin_lock_irq(&sem->wait_lock);
    /*如果有0或多个读者，并且等待队列为空，就可以获取sem*/
    if (sem->activity >= 0 &&list_empty(&sem->wait_list)) {
        /* 获得sem */
        sem->activity++; /*读者计数加1*/
       spin_unlock_irq(&sem->wait_lock);
        goto out;
    }

    /*运行到这里，说明不能获取sem，将当前进程加入等待队列进行等待*/
    tsk = current;
    set_task_state(tsk, TASK_UNINTERRUPTIBLE);

    /*
建立等待队列成员*/
    waiter.task = tsk;
    waiter.flags = RWSEM_WAITING_FOR_READ; /*表示等待读操作*/
    get_task_struct(tsk); /*进程使用计数加1*/

    list_add_tail(&waiter.list,&sem->wait_list); /*将等待成员加到等待队列尾*/

    /*
不再需要访问等待队列，因此，这里解锁*/
    spin_unlock_irq(&sem->wait_lock);

    /*
读者等待获取sem */
    for (;;) {
        if (!waiter.task)
            break;
        schedule();
        set_task_state(tsk,TASK_UNINTERRUPTIBLE);
    }
    /*运行这里，退出等待，说明可以获取sem了*/
    tsk->state = TASK_RUNNING;
out:
    ;
}

2）解读者锁操作
函数up_read释放读者锁，如果等待队列非空，说明有写者在等待，就从等待队列唤醒一个写者。其列出如下（在kernel/rwsem.c中）：

void up_read(struct rw_semaphore *sem)
{
    rwsem_release(&sem->dep_map, 1, _RET_IP_); /*获取解锁信息，用于调试*/

    __up_read(sem);
}

函数__up_read是释放读者锁的具体操作函数，其列出如下：

void __up_read(struct rw_semaphore *sem)
{
    unsigned long flags;

   spin_lock_irqsave(&sem->wait_lock, flags);
    /*如果所有读者完成读操作，并且有写者等待，那么唤醒一个写者*/
    if (--sem->activity == 0&& !list_empty(&sem->wait_list))
        sem = __rwsem_wake_one_writer(sem);

   spin_unlock_irqrestore(&sem->wait_lock, flags);
}

/*唤醒一个写者*/
static inline struct rw_semaphore *__rwsem_wake_one_writer(struct rw_semaphore*sem)
{
    struct rwsem_waiter *waiter;
    struct task_struct *tsk;

    sem->activity = -1; /*表示有一个写者正在写操作*/

    /*获取一个等待者*/
    waiter = list_entry(sem->wait_list.next, structrwsem_waiter, list);
    list_del(&waiter->list); /*将该等待者从等待队列删除*/

    tsk = waiter->task;
    smp_mb(); /*加内存屏障，确保完成上面的指针引用操作*/
    waiter->task = NULL;
    wake_up_process(tsk); /*唤醒进程*/
    put_task_struct(tsk); /*进程上下文使用计数减1*/
    return sem;
}

（3）写者加锁/解锁操作实现分析

1）加写者锁操作
函数down_write完成加写者锁操作，其列出如下：
void __sched down_write(struct rw_semaphore *sem)
{
    might_sleep();
    rwsem_acquire(&sem->dep_map, 0, 0, _RET_IP_);

    LOCK_CONTENDED(sem,__down_write_trylock, __down_write);
}

void __sched __down_write(struct rw_semaphore *sem)
{
    __down_write_nested(sem, 0);
}

函数__down_write_nested完成加写者锁的具体操作。当没有读者或写者操作时，写者才可以获取写者锁。写者锁是独占的。如果有其他写者或读者操作时，写者必须等待。其列出如下：

void __sched __down_write_nested(struct rw_semaphore*sem, int subclass)
{
    struct rwsem_waiter waiter;
    struct task_struct *tsk;

    spin_lock_irq(&sem->wait_lock);
    /*如果没有读者，并且等待队列为空（说明没有写者）时，写者才能获取写者锁*/
    if (sem->activity == 0 && list_empty(&sem->wait_list)){
        /* 获取写者锁*/
        sem->activity = -1;
       spin_unlock_irq(&sem->wait_lock);
        goto out;
    }

    /*运行到这里，说明有读者或写者在操作，需要等待*/
    tsk = current;
    set_task_state(tsk, TASK_UNINTERRUPTIBLE);

    /*
建立等待队列成员*/
    waiter.task = tsk;
    waiter.flags = RWSEM_WAITING_FOR_WRITE; /*标识为等待写操作*/
    get_task_struct(tsk); /*进程上下文使用计数加1*/

    list_add_tail(&waiter.list,&sem->wait_list); /*加到等待队列尾*/
    spin_unlock_irq(&sem->wait_lock);

    /*
进行等待*/
    for (;;) {
        if (!waiter.task)
            break;
       schedule();
        set_task_state(tsk,TASK_UNINTERRUPTIBLE);
    }
    /*被唤醒*/
    tsk->state = TASK_RUNNING;
out:
    ;
}

2）解写者锁操作

函数up_write释放写者锁，将读者计数设置为0，其列出如下：
void up_write(struct rw_semaphore *sem)
{
    rwsem_release(&sem->dep_map, 1, _RET_IP_);

    __up_write(sem);
}

void __up_write(struct rw_semaphore *sem)
{
    unsigned long flags;

   spin_lock_irqsave(&sem->wait_lock, flags);

    sem->activity = 0; /*表示有0个读者*/
    if (!list_empty(&sem->wait_list))
        sem = __rwsem_do_wake(sem, 1); /*唤醒等待者*/

   spin_unlock_irqrestore(&sem->wait_lock, flags);
}