Linux内核同步原语之信号量（Semaphore）

在Linux内核代码中，信号量被定义成semaphore结构体（代码位于include/linux/semaphore.h中）：

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

这个结构体由三部分组成：

lock：用来保护这个信号量结构体的自旋锁。
count：信号量用来保护的共享资源的数量。如果该值大于0，表示资源是空闲的，调用者可以立即获得这个信号量，进而访问被保护的共享资源；如果该值等于0，表示资源是忙的，但是没有别的进程在等待这个信号量被释放；如果这个值小于0，表示资源是忙的，且有至少一个别的进程正在等待该信号量被释放。
wait_list：在这个信号量上等待的所有进程链表。

所有插入wait_list等待链表的进程节点由semaphore_waiter结构体表示：

struct semaphore_waiter {struct list_head list;struct task_struct *task;bool up;
};

这个结构体也由三部分组成：

list：插入链表的节点。
task：指向等待进程的task_struct结构体。
up：真表示该等待进程是因为信号量释放被唤醒的，否则都是假。

初始化信号量

信号量在使用之前要对其进行初始化，一般有两种方法：

DEFINE_SEMAPHORE(sem1);struct semaphore sem2;
sema_init(&lock2);

第一种方法是用宏直接定义并且初始化一个顺序锁变量：

#define __SEMAPHORE_INITIALIZER(name, n)             \
{                                   \.lock      = __RAW_SPIN_LOCK_UNLOCKED((name).lock),   \.count     = n,                       \.wait_list = LIST_HEAD_INIT((name).wait_list),        \
}#define DEFINE_SEMAPHORE(name) \struct semaphore name = __SEMAPHORE_INITIALIZER(name, 1)

所以，直接通过宏定义初始化就是定义了一个semaphore结构体变量，将其内部的自旋锁lock初始化为未加锁，将表示共享资源数量的count变量初始化为1，将等待进程链表初始化为空链表。count的值为1，表示这个信号量只能同时由一个进程持有，也就是说被保护的共享资源只能被互斥的访问，这种特殊的信号量也称作二值（Binary）信号量。但是，通常大家用的都是这种二值信号量，用法如下：

down(&sem);
/* 临界区 */
up(&sem);

第二种方法是自己定义一个semaphore结构体变量，然后调用sema_init函数将其初始化：

static inline void sema_init(struct semaphore *sem, int val)
{......*sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem, val);......
}

最终也是通过__SEMAPHORE_INITIALIZER对信号量进行的初始化。但是，通过这种方式初始化可以指定count的值，而不像前者默认设置成1。

获取信号量

要想获取一个信号量，通常是通过调用down函数：

void down(struct semaphore *sem)
{unsigned long flags;/* 获得自旋锁并关中断 */raw_spin_lock_irqsave(&sem->lock, flags);/* 如果信号量的count大于0 */if (likely(sem->count > 0))/* 直接获得该信号量并将count值递减 */sem->count--;else__down(sem);/* 释放自旋锁并开中断 */raw_spin_unlock_irqrestore(&sem->lock, flags);
}
EXPORT_SYMBOL(down);

如果信号量的count值大于0，则直接将其递减然后返回，调用的进程直接获得该信号量。如果小于或等于0，则接着调用__down函数：

static noinline void __sched __down(struct semaphore *sem)
{__down_common(sem, TASK_UNINTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}

__down函数直接调用了__down_common函数。传入的第二个参数是TASK_UNINTERRUPTIBLE，表示；传入的第三个参数是MAX_SCHEDULE_TIMEOUT，表示会一直等待该信号量，直到获得为止，没有到期时间。

除了最基本的down函数外，还可以通过下面几个函数来获得信号量：

down_interruptible：基本功能和down相同，区别是如果无法获得信号量，会将该进程置于TASK_INTERRUPTIBLE状态。因此，在进程睡眠时可以通过信号（Signal）将其唤醒。
down_killable：如果无法获得信号量，会将该进程置于TASK_KILLABLE状态。
down_timeout：不会一直等待该信号量，而是有一个到期时间，时间到了后即使没有获得信号量也会返回。等待的时候也和down函数一样，会将进程置于TASK_UNINTERRUPTIBLE状态。

所有获得信号量的方法最终都是调用了__down_common函数，只不过传入的第二个和第三个参数不一样。

static noinline int __sched __down_interruptible(struct semaphore *sem)
{return __down_common(sem, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}static noinline int __sched __down_killable(struct semaphore *sem)
{return __down_common(sem, TASK_KILLABLE, MAX_SCHEDULE_TIMEOUT);
}static noinline int __sched __down_timeout(struct semaphore *sem, long timeout)
{return __down_common(sem, TASK_UNINTERRUPTIBLE, timeout);
}

我们接着回来看__down_common函数的实现：

static inline int __sched __down_common(struct semaphore *sem, long state,long timeout)
{struct semaphore_waiter waiter;/* 将表示等待进程的节点插入信号量的等待列表中 */list_add_tail(&waiter.list, &sem->wait_list);waiter.task = current;waiter.up = false;for (;;) {/* 如果有待处理的信号则跳到interrupted标签处 */if (signal_pending_state(state, current))goto interrupted;/* 如果超时了则跳到timed_out标签处 */if (unlikely(timeout <= 0))goto timed_out;/* 设置当前进程的状态 */__set_current_state(state);/* 释放自旋锁 */raw_spin_unlock_irq(&sem->lock);/* 当前进程睡眠 */timeout = schedule_timeout(timeout);/* 再次获得自旋锁 */raw_spin_lock_irq(&sem->lock);/* 如果是因为自旋锁释放而被唤醒的则返回0 */if (waiter.up)return 0;}timed_out:list_del(&waiter.list);return -ETIME;interrupted:list_del(&waiter.list);return -EINTR;
}

该函数先将表示等待进程的节点插入信号量的等待列表中，该节点的task变量指向表示当前进程的task_struct结构体，up变量被初始化为否。然后会进入一个大的循环中，循环的退出条件有三个：

当前进程有待处理的信号。
当前进程等待信号量超时了。
当前进程被另一个释放信号量的进程唤醒。

signal_pending_state函数用来判断当前进程是否有待处理的信号（代码位于include/linux/sched/signal.h中）：

static inline int signal_pending_state(long state, struct task_struct *p)
{if (!(state & (TASK_INTERRUPTIBLE | TASK_WAKEKILL)))return 0;if (!signal_pending(p))return 0;return (state & TASK_INTERRUPTIBLE) || __fatal_signal_pending(p);
}

所以，当前进程的状态必须包含TASK_INTERRUPTIBLE或TASK_WAKEKILL才行。也就是说，只有通过调用down_interruptible或down_killable函数获得信号量时，signal_pending_state函数才有可能返回真。

schedule_timeout用来将当前进程休眠，同时设置到期时间（代码位于kernel/time/timer.c中）：

signed long __sched schedule_timeout(signed long timeout)
{struct process_timer timer;unsigned long expire;switch (timeout){case MAX_SCHEDULE_TIMEOUT:/* 没有到期时间间隔直接睡眠 */schedule();goto out;default:if (timeout < 0) {printk(KERN_ERR "schedule_timeout: wrong timeout ""value %lx\n", timeout);dump_stack();current->state = TASK_RUNNING;goto out;}}/* 计算到期时间 */expire = timeout + jiffies;timer.task = current;/* 设置定时器到期处理函数为process_timeout */timer_setup_on_stack(&timer.timer, process_timeout, 0);/* 设置定时器 */__mod_timer(&timer.timer, expire, 0);/* 休眠 */schedule();/* 删除定时器 */del_singleshot_timer_sync(&timer.timer);....../* 计算距离到期时间还剩多少时间 */timeout = expire - jiffies;out:return timeout < 0 ? 0 : timeout;
}
EXPORT_SYMBOL(schedule_timeout);

如果传入的到期时间间隔被设置成了MAX_SCHEDULE_TIMEOUT，表示当前进程在获得信号量时没有设置超时，因此直接调用schedule调度另一个进程执行，本进程进入休眠。如果设置了一个有效的到期时间间隔，那么在调用schedule之前必须要先添加一个定时器，让其在到期时间点被触发。但是，表示定时器本身的结构并不包含由哪个进程设置它的信息，所以还需要定义一个新的结构体process_timer：

struct process_timer {struct timer_list timer;struct task_struct *task;
};

注意，表示定时器的结构体timer_list是在process_timer结构体中的第一个变量。这样，如果我们有了一个指向表示定时器的timer_list结构体的指针，那么它其实也是指向process_timer结构体的。通过前面的分析可以看到，定时器的到期函数被设置成了process_timeout，它的参数就是一个指向timer_list结构体的指针：

static void process_timeout(struct timer_list *t)
{/* 从timer_list指针获得包含它的process_timer */struct process_timer *timeout = from_timer(timeout, t, timer);/* 唤醒到期进程 */wake_up_process(timeout->task);
}

就是通过指向定时器结构体timer_list的指针获得包含它的process_timer结构体指针，从而获得设置该定时器的进程，然后唤醒它。

释放信号量

要想释放一个信号量，只能通过调用up函数：

void up(struct semaphore *sem)
{unsigned long flags;/* 获得自旋锁并关中断 */raw_spin_lock_irqsave(&sem->lock, flags);/* 如果等待进程链表为空 */if (likely(list_empty(&sem->wait_list)))/* 直接将count递增 */sem->count++;else__up(sem);/* 释放自旋锁并开中断 */raw_spin_unlock_irqrestore(&sem->lock, flags);
}
EXPORT_SYMBOL(up);

如果释放的时候发现等待进程链表是空的，表示当前没有被的进程在等这个信号量，直接递增count值就行了；如果不为空，表示有别的进程在等这个信号量被释放，那么当前进程需要接着调用__up函数试着将某个等待进程唤醒。

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 = true;/* 将该等待进程唤醒 */wake_up_process(waiter->task);
}

功能很简单，获得等待链表中的第一个节点，然后唤醒该节点表示的进程。需要把该节点的up字段置为真，让被唤醒的进程知道其是因为信号量被释放才被唤醒的。

使用场景

要使用信号量，一般要满足以下的一些使用场景或条件：

信号量适合于保护较长的临界区。它不应该用来保护较短的临界区，因为竞争信号量时有可能使进程睡眠和切换，然后被再次唤醒，代价很高，这种场景下应该使用自旋锁。
如果被保护的共享资源有多份，并不只是互斥访问的，那非常适合使用信号量。
只有允许睡眠的场景下才能使用内核信号量，也就是说在中断处理程序和可延迟函数中都不能使用信号量。