【源码分析】——信号量

董哥聊技术

发布于 2023-09-28 16:51:24

2780

发布于 2023-09-28 16:51:24

文章被收录于专栏：嵌入式艺术嵌入式艺术

【深入理解Linux内核锁】六、信号量

除了原子操作，中断屏蔽，自旋锁以及自旋锁的衍生锁之外，在Linux内核中还存在着一些其他同步互斥的手段。

下面我们来理解一下信号量，互斥体，完成量机制。

1、信号量介绍

信号量（Semaphore）是操作系统中最典型的用于同步和互斥的手段，信号量的值可以是0、1或者n。信号量与操作系统中的经典概念PV操作对应。

P（Produce）：

将信号量S的值减1，即S=S-1；
如果S≥0，则该进程继续执行；否则该进程置为等待状态，排入等待队列。

V（Vaporize）：

将信号量S的值加1，即S=S+1；
如果S>0，唤醒队列中等待信号量的进程。

信号量核心思想：

信号量的操作，更适合去解决生产者和消费者的问题，比如：我做出来一个饼，你才能吃一个饼；如果我没做出来，你就先释放CPU去忙其他的事情。

2、信号量的API

struct semaphore sem;  // 定义信号量

void sema_init(struct semaphore *sem, int val);  // 初始化信号量，并设置信号量sem的值为val。

void down(struct semaphore * sem);     // 获得信号量sem，它会导致睡眠，因此不能在中断上下文中使用。
int down_interruptible(struct semaphore * sem);  // 该函数功能与down类似，不同之处为，因为down（）进入睡眠状态的进程不能被信号打断，但因为down_interruptible（）进入睡眠状态的进程能被信号打断，信号也会导致该函数返回，这时候函数的返回值非0 
int down_trylock(struct semaphore * sem);   // 尝试获得信号量sem，如果能够立刻获得，它就获得该信号量并返回0，否则，返回非0值。它不会导致调用者睡眠，可以在中断上下文中使用。

void up(struct semaphore * sem);  // 释放信号量，唤醒等待者。

由于新的Linux内核倾向于直接使用mutex作为互斥手段，信号量用作互斥不再被推荐使用。信号量也可以用于同步，一个进程A执行down（）等待信号量，另外一个进程B执行up（）释放信号量，这样进程A就同步地等待了进程B。

3、API实现

3.1 semaphore

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

结构体名称：semaphore

文件位置：include/linux/semaphore.h

主要作用：用于定义一个信号量。

raw_spinlock_t：信号量结构体也使用了自旋锁，避免互斥。
count：表示信号量的计数器，表示资源的数量
struct list_head wait_list: 这是一个链表头，用于管理等待信号量的线程。当信号量的 count 等于0，即没有可用资源时，等待信号量的线程会被加入到这个链表中，以便在资源可用时进行唤醒。

3.2 sema_init

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);
}

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

#define LIST_HEAD_INIT(name) { &(name), &(name) }

函数名称：sema_init

文件位置：include/linux/semaphore.h

主要作用：初始化信号量，并设置信号量sem的值为val。

实现流程：

使用__SEMAPHORE_INITIALIZER宏定义来初始化信号量
- 使用__RAW_SPIN_LOCK_UNLOCKED宏定义来初始化自旋锁
- 直接将val赋值给信号量的值count
- 使用LIST_HEAD_INIT来初始化一个链表

#define LIST_HEAD_INIT(name) { &(name), &(name) }

该宏接受一个参数 name，并返回一个结构体对象。这个对象有两个成员 next 和 prev，分别指向 name 本身。

这样，当我们使用该宏来初始化链表头节点时，会得到一个拥有 next 和 prev 成员的结构体对象。其中 next 和 prev 成员都指向该结构体对象本身。

这种初始化方式可以用于创建一个空的双向链表，因为在初始状态下，链表头节点的 next 和 prev 指针都指向自身，表示链表为空。

3.3 down

/**
 * down - acquire the semaphore
 * @sem: the semaphore to be acquired
 *
 * Acquires the semaphore.  If no more tasks are allowed to acquire the
 * semaphore, calling this function will put the task to sleep until the
 * semaphore is released.
 *
 * Use of this function is deprecated, please use down_interruptible() or
 * down_killable() instead.
 */
void down(struct semaphore *sem)
{
    unsigned long flags;

    raw_spin_lock_irqsave(&sem->lock, flags);
    if (likely(sem->count > 0))
        sem->count--;
    else
        __down(sem);
    raw_spin_unlock_irqrestore(&sem->lock, flags);
}
EXPORT_SYMBOL(down);

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

/*
 * 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;
}

函数名称：down

文件位置：kernel/locking/semaphore.c

主要作用：获取信号量，如果信号量的值大于0，则消耗一个；如果不存在，则让线程进入休眠状态并等待信号量被释放。

函数调用流程：

down(kernel/locking/semaphore.c)
    |--> raw_spin_lock_irqsave      //  获取锁，并保存中断信息
    ||-> sem->count--;              //  如果sem->count信号量存在，则消耗一个
     |-> __down                     //  如果sem->count信号量不存在，则进入休眠状态
        |--> __down_common
            |--> list_add_tail      //  将当前线程添加到信号量的等待链表中，表示当前线程正在等待信号量。
            |--> __set_current_state//  设置线程为休眠状态
            |--> raw_spin_unlock_irq//  释放自旋锁，让其他线程可用
            |--> schedule_timeout   //  让线程进入睡眠状态，等待信号量释放或超时。
            |--> raw_spin_lock_irq  //  重新获取自旋锁，继续执行后续操作
    |--> raw_spin_unlock_irqrestore //  释放锁，并恢复中断信息

实现流程：

获取信号量时，先使用raw_spin_lock_irqsave和raw_spin_unlock_irqrestore将信号量的操作包裹起来，避免竞态发生。
然后对sem->count判断，如果信号量大于0，就消耗一个，否则的话，将当前线程设置为休眠态
调用__down_common接口，默认将当前线程设置为TASK_UNINTERRUPTIBLE中断不可打断状态，并设置最大超时时间MAX_SCHEDULE_TIMEOUT
struct semaphore_waiter waiter：创建waiter结构体，表示当前线程的状态
调用__set_current_state接口，设置当前线程的状态信息，即TASK_UNINTERRUPTIBLE
调用schedule_timeout，让该线程让出CPU，进入休眠态，并且在前面加上raw_spin_unlock_irq保证其他线程可以正常使用信号量。
当线程时间片到时，获取CPU，并调用raw_spin_lock_irq，获取锁，来防止竞态发生。

3.4 up

/**
 * 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);

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

文件位置：kernel/locking/semaphore.c

主要作用：获取信号量，如果信号量的值大于0，则消耗一个；如果不存在，则让线程进入休眠状态并等待信号量被释放。

实现流程：

相信分析完down后，up也变得很简单

释放信号量时，先使用raw_spin_lock_irqsave和raw_spin_unlock_irqrestore将信号量的操作包裹起来，避免竞态发生。
然后对sem->wait_list判断，如果其为空，说明没有等待的线程，直接将sem->count自增，如果有等待的线程，则唤醒下一个线程。
调用wake_up_process来唤醒线程