Linux 并发与同步(二)信号量

原图

信号量

信号量( semaphore)是操作系统中最常用的同步原语之一。自旋锁是一种实现忙等待的锁,而信号量则允许进程进入睡眠状态。简单来说,信号量是一个计数器,它支持两个操作原语,即 P 和 V 操作。

信号量实现

semaphore 数据结构的定义如下。

1
2
3
4
5
6
<include/linux/semaphore.h>
struct semaphore {
raw_spinlock_t lock;
unsigned int count;
struct list_head wait_list;
};
  • lock 是自旋锁变量,用于保护 semaphore 数据结构里的 count 和 wait_list 成员
  • count 用于表示允许进入临界区的内核执行路径个数
  • wait_list 链表用于管理所有在该信号量上睡眠的进程,没有成功获取锁的进程会在这个链表上睡眠

通常通过 sema_init() 函数进行信号量的初始化,其中 __SEMAPHORE_INITIALIZER() 完成对 semaphore 数椐结构的填充,val 值通常设为 1。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
#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)

static inline void sema_init(struct semaphore *sem, int val)
{
static struct lock_class_key __key;
*sem = (struct semaphore) __SEMAPHORE_INITIALIZER(*sem, val);
lockdep_init_map(&sem->lock.dep_map, "semaphore->lock", &__key, 0);
}

下面来看 down() 函数。down() 函数有如下一些变体。其中 down() 函数和 down_interruptible() 函数的区别在于, down_interruptible() 函数在争用信号量失败时进入可中断的睡眠状态,而 down() 函数进入不可中断的睡眠状态。若 down_trylock() 函数返回 0,表示成功获取锁:若返回 1 获取锁失败。

1
2
3
4
5
6
extern void down(struct semaphore *sem);
extern int __must_check down_interruptible(struct semaphore *sem);
extern int __must_check down_killable(struct semaphore *sem);
extern int __must_check down_trylock(struct semaphore *sem);
extern int __must_check down_timeout(struct semaphore *sem, long jiffies);
extern void up(struct semaphore *sem);

接下来看 down_interruptible() 函数的实现:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
/**
* down_interruptible - acquire the semaphore unless interrupted
* @sem: the semaphore to be acquired
*
* Attempts to acquire the semaphore. If no more tasks are allowed to
* acquire the semaphore, calling this function will put the task to sleep.
* If the sleep is interrupted by a signal, this function will return -EINTR.
* If the semaphore is successfully acquired, this function returns 0.
*/
int down_interruptible(struct semaphore *sem)
{
unsigned long flags;
int result = 0;

might_sleep();
raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(sem->count > 0))
sem->count--;
else
result = __down_interruptible(sem);
raw_spin_unlock_irqrestore(&sem->lock, flags);

return result;
}
EXPORT_SYMBOL(down_interruptible);

首先,第 6~9 行代码判断是否进入自旋锁的临界区。注意,后面的操作会临时打开自旋锁,若涉及对信号量中最重要的 cout 的操作,需要自旋锁来保护,并且在某些中断处理函数中也可能会操作该信号量。由于需要关闭本地 CPU 中断,因此这里采用 raw_spin_lock_irqsave() 函数。当成功进入自旋锁的临界区之后,首先判断 sem->count 是否大于 0。如果大于 0,则表明当前进程可以成功地获得信号量,并将 sem->count 值减 1,然后退出。如果 sem->count 小于或等于 0,表明当前进程无法获得该信号量,则调用 __down_interruptible() 函数来执行睡眠操作。

1
2
3
4
static noinline int __sched __down_interruptible(struct semaphore *sem)
{
return __down_common(sem, TASK_INTERRUPTIBLE, MAX_SCHEDULE_TIMEOUT);
}

__down_interruptible() 函数内部调用__down_common() 函数来实现 state 参数为 TASK_INTERRUPTIBLE。 timeout 参数为 MAX_SCHEDULE_TIMEOUT,是一个很大的 LONG_MAX 值。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
<down_interruptible()->__down_interruptible()->down_common()>
/*
* Because this function is inlined, the 'state' parameter will be
* constant, and thus optimised away by the compiler. Likewise the
* 'timeout' parameter for the cases without timeouts.
*/
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 (;;) {
if (signal_pending_state(state, current))
goto interrupted;
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);
if (waiter.up)
return 0;
}

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

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

semaphore_waiter 数据结构用于描述获取信号量失败的进程,每个进程会有一个 semaphore_waiter 数据结构,并且把当前进程放到信号量 sem 的成员变量 wait_list 链表中,接下来的 for 循环将当前进程的 task_struct、状态设置成 TASK_INTERRUPTIBLE,然后调用 schedule_timeout() 函数主动让出 CPU,相当于当前进程睡眠。注意 schedule_timeou() 函数的参数是 MAX_SCHEDULE_TIMEOUT,它并没有实际等待 MAX_SCHEDULE_TIMEOUT 的时间。当进程再次被调度回来执行时, schedule_timeout() 函数返回并判断再次被调度的原因,当 waiter.up 为 true 时,说明睡眠在 wait_list 队列中的进程被该信号量的 UP 操作唤醒,进程可以获得该信号量。如果进程被其它 CPU 发送的信号或者由于超时等而唤醒,则跳转到 timed_out 或 interrupted 标签处并且返回错误代码。

down_interruptible() 函数中,在调用 __down_interruptible() 函数时加入 sem->lock 的自旋锁,这是自旋锁的一个临界区。前面提到,自旋锁临界区中绝对不能睡眠,难道这是例外?仔细阅读__down_common() 函数,会发现 for 循环在调用 schedule_timeout() 函数主动让出 CPU 时,先调用 raw_spin_unlock_irq() 函数释放了该锁,即调用 schedule_timeout() 函数时已经没有自旋锁了,可以让进程先睡眠,“醒来时”再补加一个锁,这是内核编程的常用技巧。

下面来看与 dowm() 函数对应的 up() 函数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
/**
* up - release the semaphore
* @sem: the semaphore to release
*
* Release the semaphore. Unlike mutexes, up() may be called from any
* context and even by tasks which have never called down().
*/
void up(struct semaphore *sem)
{
unsigned long flags;

raw_spin_lock_irqsave(&sem->lock, flags);
if (likely(list_empty(&sem->wait_list)))
sem->count++;
else
__up(sem);
raw_spin_unlock_irqrestore(&sem->lock, flags);
}
EXPORT_SYMBOL(up);

如果信号量上的等待队列(sem->wait_list)为空,则说明没有进程在等待该信号量,直接把 sem->count 加 1 即可,如果不为空,则说明有进程在等待队列里睡民,需要调用__up() 唤醒它们。

1
2
3
4
5
6
7
8
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);
}

首先来看 sm->wait_list 中第一个成员 waiter,这个等待队列是先进先出队列,在 dowm() 函数中通过 list_add_tail() 函数添加到等待队列尾部。把 waiter->up 设置为 true, 把然后调用 wake_up_process() 函数唤醒 wite->task 进程。在 down() 函数中, walter->task 进程醒来后会判断 waiter->up 变量是否为 true 如果为 true 则直接回 0,表示该过得成功获取信号量。

读写信号量

rw_semaphore 数据结构

rw_semaphore 数据结构定义如下:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
struct rw_semaphore {
atomic_long_t count;
/*
* Write owner or one of the read owners as well flags regarding
* the current state of the rwsem. Can be used as a speculative
* check to see if the write owner is running on the cpu.
*/
atomic_long_t owner;
#ifdef CONFIG_RWSEM_SPIN_ON_OWNER
struct optimistic_spin_queue osq; /* spinner MCS lock */
#endif
raw_spinlock_t wait_lock;
struct list_head wait_list;
#ifdef CONFIG_DEBUG_RWSEMS
void *magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
struct lockdep_map dep_map;
#endif
};
  • count 用于表示读写信号量的计数。以前读写信号量的实现用 activity 来表示。若 activity 为 0,表示没有读者和写者:若 activity 为ー 1,表示有写者,若 activity 大于 0,表示有读者。现在 count 的计数方法已经发生了变化
  • wait_list 链表用于管理所有在该信号量上睡眠的进程,没有成功获取锁的进程会睡眠在这个链表上
  • wait_lock 是一个自旋锁变量,用于实现对 rw_semaphore 数据结构中 count 成员的原子操作和保护
  • osq:MCS 锁
  • owner:当写者成功获取锁时, owner 指向锁持有者的 task_struct 数据结构

  • 若 count 初始化为 0,表示没有读者也没有写者
  • 若 count 为正数,表示有 count 个读者
  • 当有写者申请锁时,count 值要加上 RWSEM_ACTIVE_WIRTE_BIAS
  • 当有读者申请锁时,若 count 值要加上 RWSEM_ACTIVE_READ_BIAS,即 count 值加一
  • 当有多个写者中请时,判断 count 值是否等于 RWSEM_ACTIVE_WIRTE_BIAS(-65535)若不相等,说明已经有写者抢先持有锁,要自旋等待或者睡眠
  • 当读者申请锁时,若 count 值加上 RWSEM_ACTIVE_READ_BIAS(1)后还小于 0,说明已经有个写者已经成功申请锁,只能等待写者释放锁

把 count 值当作十六进制或者十进制数不是开发人员的原本设计意图,其实应该把 count 值分成两个字段:Bit[0:31] 为低字段,表示正在持有锁的读者或者写者的个数;Bit[32:63] 为高字段,通常为负数,表示有一个正在持有或者处于 pending 状态的写者,以及等待队列中有读写者在等待。因此 count 值可以看作一个二元数,含义如下

  • RWSEM_ACTIVE_READ_BIAS=0x000 0001 =[0,1],表示有一个读者
  • RWSEM_ACTIVE_WRITE_BIAS=0xFFFF 0000 =[-1,1],表示当前只有一个活跃的写者
  • RWSEM_WAITING_BIAS=0xFFFF 0000=[-1,0],表示睡眠等待队列中有读写者在睡眠等待

kernel/locking/rwsem-xadd.c 代码中有如下一段关于 count 值含义的比较全面的介绍。

  • 0x0000 0000 初始化值,表示没有读者和写者
  • 0x0000 000X:表示有 X 个活跃的读者或者正在申请的读者,没有写者干扰
  • 0xFFFF 000X:或者表示可能有 X 个活跃读者,还有写者正在等待;或者表示有二个写者持有锁,还有多个读者正在等待
  • 0xFFFF 0001:或者表示当前只有一个活跃的写者;或者表示一个活跃或者申请中的读者,还有写者正在睡眠等待
  • 0xFFFF 0000 表示 WAITING_BIAS,有读者或者写者正在等待,但是它们都还没成功获取锁。

申请读者类型信号量

下面来看 down_read() 函数的实现。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
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);
}
...
#define rwsem_acquire_read(l, s, t, i) lock_acquire_shared(l, s, t, NULL, i)
#define lock_acquire_shared(l, s, t, n, i) lock_acquire(l, s, t, 1, 1, n, i)
...
/*
* We are not always called with irqs disabled - do that here,
* and also avoid lockdep recursion:
*/
void lock_acquire(struct lockdep_map *lock, unsigned int subclass,
int trylock, int read, int check,
struct lockdep_map *nest_lock, unsigned long ip)
{
unsigned long flags;

trace_lock_acquire(lock, subclass, trylock, read, check, nest_lock, ip);

if (!debug_locks)
return;

if (unlikely(!lockdep_enabled())) {
/* XXX allow trylock from NMI ?!? */
if (lockdep_nmi() && !trylock) {
struct held_lock hlock;

hlock.acquire_ip = ip;
hlock.instance = lock;
hlock.nest_lock = nest_lock;
hlock.irq_context = 2; // XXX
hlock.trylock = trylock;
hlock.read = read;
hlock.check = check;
hlock.hardirqs_off = true;
hlock.references = 0;

verify_lock_unused(lock, &hlock, subclass);
}
return;
}

raw_local_irq_save(flags);
check_flags(flags);

lockdep_recursion_inc();
__lock_acquire(lock, subclass, trylock, read, check,
irqs_disabled_flags(flags), nest_lock, ip, 0, 0);
lockdep_recursion_finish();
raw_local_irq_restore(flags);
}
EXPORT_SYMBOL_GPL(lock_acquire);

申请读者锁流程如下:

释放读者类型信号量

1
2
3
4
int up_read(struct rw_semaphore *sem)
{
return perf_singlethreaded ? 0 : pthread_rwlock_unlock(&sem->lock);
}

关于写信号量这里不做详细分析了。

小结

读写信号量在内核中应用广泛,特别是在内存管理中,除了前面介绍的 mm->mmap_sem 还有 RMAP 系统中的的 anon_vma->rwsem、 address_space 数据结构中的 i_mmap_rwsem 等。 总结读写信号量的重要特性:

  • down_read():如果一个进程持有读者锁,那么允许继续申请多个读者锁,申请的写者锁则要等待
  • down_write():如果一个进程持有写者锁,那么第二个进程申请该写者锁要自旋等待申请读者锁则要等待
  • up_write() up_read():如果等待队列中第一个成员是写者,那么唤醒该写者;否则唤醒排在等待队列中最前面连续的几个读者

总结

信号量有一个有趣的特点,它可以同时允许任意数量的锁持有者。信号量初始化函数为 sema_init(struct semaphore *sem, int val),其中 val 的值可以大于或等于 1。当 val 大于 1 时,表示允许在同一时刻至多有 val 个锁持有者,这种信号量叫作计数信号量( counting semaphore);当 val 等于 1 时,同一时刻仅允许一个 CPU 持有锁,这种信号量叫作互斥信号量或者二进制信号量( binary semaphore)。在 Linux 内核中,大多使用 val 值为 1 的信号量。相比自旋锁,信号量是一个允许睡眠的锁。

参考文献

《奔跑吧 Linux 内核》