linux 内核并发同步 2

信号量semaphore

信号量是一种允许进程进入睡眠的同步机制，信号量是一个计数器，支持两种原语即P 和V操作，也就是down 和up 操作，

/* Please don't access any members of this structure directly */
struct semaphore {
    raw_spinlock_t        lock;//用于对count以及wait_list成员的保护
    unsigned int        count;//表示允许进入临界区的控制路径
    struct list_head    wait_list;//用于管理所有在该信号上睡眠的进程
};
#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);
}

信号量的操作：

 void down(struct semaphore *sem);
 int __must_check down_interruptible(struct semaphore *sem);
 int __must_check down_killable(struct semaphore *sem);
 int __must_check down_trylock(struct semaphore *sem);
 int __must_check down_timeout(struct semaphore *sem, long jiffies);
 void up(struct semaphore *sem); 1 /** 2  * down_interruptible - acquire the semaphore unless interrupted 3  * @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;

    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;
}
static noinline int __sched __down_interruptible(struct semaphore *sem)
{
return __down_common(sem, TASK_INTERRUPTIBLE, 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 task_struct *task = current;
    struct semaphore_waiter waiter;

    list_add_tail(&waiter.list, &sem->wait_list);//添加到末尾 fifo 先进先出
    waiter.task = task;
    waiter.up = false;

    for (;;) {
        if (signal_pending_state(state, task))//收到信号且不是sigkill
            goto interrupted;
        if (unlikely(timeout <= 0))//超时唤醒
            goto timed_out;
        __set_task_state(task, state);
        raw_spin_unlock_irq(&sem->lock);
        timeout = schedule_timeout(timeout);
        raw_spin_lock_irq(&sem->lock);
        if (waiter.up)//up 为true 表示被唤醒切获取到自旋锁

            return 0;
    }

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

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

可知16-19行为临界区，涉及到count 的计数 会使用spinlock 来保护，由于这些信号量可能在中断函数里面使用，所以需要关闭本地cpu 中断，这里采用raw_spin_lock_irqsave，如果count值小于0 表示当前进程无法获取该信号量使用down_interrupt来执行睡眠等待操作，

49-50行：主动让出cpu之前需要释放该锁，

/**
 * 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);
}
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);
}
/**
 * wake_up_process - Wake up a specific process
 * @p: The process to be woken up.
 *
 * Attempt to wake up the nominated process and move it to the set of runnable
 * processes.
 *
 * Return: 1 if the process was woken up, 0 if it was already running.
 *
 * It may be assumed that this function implies a write memory barrier before
 * changing the task state if and only if any tasks are woken up.
 */
int wake_up_process(struct task_struct *p)
{
    return try_to_wake_up(p, TASK_NORMAL, 0);
}

可知：如果信号量上的等待队列为空，则说明没有进程等待该变量，那么只需要将count++，如果不为空则有进程在等待队列休眠，需要up叫醒他们。up函数中唤醒进程是先进先出。

MUTEX 互斥体

mutex是阻塞锁，当某线程无法获取互斥量时，该线程会被直接挂起，该线程不再消耗CPU时间，当其他线程释放互斥量后，操作系统会激活那个被挂起的线程，让其投入运行.

问题：mutex 互斥体解决了什么问题？，mutex 和信号量的区别？

信号量可以根据初始化的count值大小分为计数信号量和互斥信号量。mutex 相当于count计数等于1的信号量，为什么还有mutex呢？

猜测原因：应该是在锁竞争激烈的场景下，mutex 执行更快吧，

作者：匿名用户
链接：https://www.zhihu.com/question/47704079/answer/216430116
来源：知乎
著作权归作者所有。商业转载请联系作者获得授权，非商业转载请注明出处。

A mutex
is essentially the same thing as a binary semaphore and sometimes uses
the same basic implementation. The differences between them are in how
they are used. While a binary semaphore may be used as a mutex, a mutex
is a more specific use-case, in that only the thread that locked the
mutex is supposed to unlock it. This constraint makes it possible to
implement some additional features in mutexes:

1. Since only the thread that locked the mutex is supposed to unlock
   it, a mutex may store the id of thread that locked it and verify the
   same thread unlocks it.
2. Mutexes may provide priority inversion
   safety. If the mutex knows who locked it and is supposed to unlock it,
   it is possible to promote the priority of that thread whenever a
   higher-priority task starts waiting on the mutex.
3. Mutexes may also provide deletion safety, where the thread holding the mutex cannot be accidentally deleted.
4. Alternately, if the thread holding the mutex is deleted (perhaps due
   to an unrecoverable error), the mutex can be automatically released.
5. A mutex may be recursive: a thread is allowed to lock it multiple times without causing a deadlock.

/*
 * Simple, straightforward mutexes with strict semantics:
 *
 * - only one task can hold the mutex at a time
 * - only the owner can unlock the mutex
 * - multiple unlocks are not permitted
 * - recursive locking is not permitted
 * - a mutex object must be initialized via the API
 * - a mutex object must not be initialized via memset or copying
 * - task may not exit with mutex held
 * - memory areas where held locks reside must not be freed
 * - held mutexes must not be reinitialized
 * - mutexes may not be used in hardware or software interrupt
 *   contexts such as tasklets and timers
 *
 * These semantics are fully enforced when DEBUG_MUTEXES is
 * enabled. Furthermore, besides enforcing the above rules, the mutex
 * debugging code also implements a number of additional features
 * that make lock debugging easier and faster:
 *
 * - uses symbolic names of mutexes, whenever they are printed in debug output
 * - point-of-acquire tracking, symbolic lookup of function names
 * - list of all locks held in the system, printout of them
 * - owner tracking
 * - detects self-recursing locks and prints out all relevant info
 * - detects multi-task circular deadlocks and prints out all affected
 *   locks and tasks (and only those tasks)
 */
struct mutex {
    /* 1: unlocked, 0: locked, negative: locked, possible waiters */
    atomic_t        count;//原子计数 1表示没有人持有锁，0 表示被持有
    spinlock_t        wait_lock; //用于保护wait_list睡眠等待队列
    struct list_head    wait_list;
#if defined(CONFIG_DEBUG_MUTEXES) || defined(CONFIG_MUTEX_SPIN_ON_OWNER)
    struct task_struct    *owner;
#endif
#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
    struct optimistic_spin_queue osq; /* Spinner MCS lock */
#endif
#ifdef CONFIG_DEBUG_MUTEXES
    void            *magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
    struct lockdep_map    dep_map;
#endif
};

Spinner MCS lock ： mcs 一种spinlock的优化方案

mutex 实现：

/**
 * mutex_lock - acquire the mutex
 * @lock: the mutex to be acquired
 *
 * Lock the mutex exclusively for this task. If the mutex is not
 * available right now, it will sleep until it can get it.
 *
 * The mutex must later on be released by the same task that
 * acquired it. Recursive locking is not allowed. The task
 * may not exit without first unlocking the mutex. Also, kernel
 * memory where the mutex resides must not be freed with
 * the mutex still locked. The mutex must first be initialized
 * (or statically defined) before it can be locked. memset()-ing
 * the mutex to 0 is not allowed.
 *
 * ( The CONFIG_DEBUG_MUTEXES .config option turns on debugging
 *   checks that will enforce the restrictions and will also do
 *   deadlock debugging. )
 *
 * This function is similar to (but not equivalent to) down().
 */
void __sched mutex_lock(struct mutex *lock)
{
    might_sleep();
    /*
     * The locking fastpath is the 1->0 transition from
     * 'unlocked' into 'locked' state.
     */
    __mutex_fastpath_lock(&lock->count, __mutex_lock_slowpath);
    mutex_set_owner(lock);//成功持有锁后 lock->owner 只想当前进程的task_struct 当前线程的thread_info可以通过sp低13位清零获取到

}

eg:如果count计数减1后小于0，表示该锁已经被人持有，这进入slowpath，  1 __mutex_lock_slowpath(atomic_t *lock_count)

{
    struct mutex *lock = container_of(lock_count, struct mutex, count);

    __mutex_lock_common(lock, TASK_UNINTERRUPTIBLE, 0,
                NULL, _RET_IP_, NULL, 0);
}

/*
 * Lock a mutex (possibly interruptible), slowpath:
 */
static __always_inline int __sched
__mutex_lock_common(struct mutex *lock, long state, unsigned int subclass,
            struct lockdep_map *nest_lock, unsigned long ip,
            struct ww_acquire_ctx *ww_ctx, const bool use_ww_ctx)
{
    struct task_struct *task = current;// 通过sp低13位清零获取到thread_info，再通过thread_info获取到task指针

    struct mutex_waiter waiter;
    unsigned long flags;
    int ret;

    if (use_ww_ctx) {
        struct ww_mutex *ww = container_of(lock, struct ww_mutex, base);
        if (unlikely(ww_ctx == READ_ONCE(ww->ctx)))
            return -EALREADY;
    }

    preempt_disable();
    mutex_acquire_nest(&lock->dep_map, subclass, 0, nest_lock, ip);

    if (mutex_optimistic_spin(lock, ww_ctx, use_ww_ctx)) {
        /* got the lock, yay! */
        preempt_enable();
        return 0;
    }

    spin_lock_mutex(&lock->wait_lock, flags);// 获取lock->wait_lock自旋锁，直到获取成功为止，因为接下来的代码要对lock的成员变量进行读写，需要互斥访问

    /*
     * Once more, try to acquire the lock. Only try-lock the mutex if
     * it is unlocked to reduce unnecessary xchg() operations.
     */ // 循环读取互斥锁计数器count的值，直到count为1，表示互斥锁可用，此处读取互斥锁计数器的同时也将互斥锁的值设置为0了
    if (!mutex_is_locked(lock) &&
        (atomic_xchg_acquire(&lock->count, 0) == 1))
        goto skip_wait;

    debug_mutex_lock_common(lock, &waiter);
    debug_mutex_add_waiter(lock, &waiter, task_thread_info(task));

    /* add waiting tasks to the end of the waitqueue (FIFO): */
    list_add_tail(&waiter.list, &lock->wait_list);
    waiter.task = task;

    lock_contended(&lock->dep_map, ip);

    for (;;) {
        /*
         * Lets try to take the lock again - this is needed even if
         * we get here for the first time (shortly after failing to
         * acquire the lock), to make sure that we get a wakeup once
         * it's unlocked. Later on, if we sleep, this is the
         * operation that gives us the lock. We xchg it to -1, so
         * that when we release the lock, we properly wake up the
         * other waiters. We only attempt the xchg if the count is
         * non-negative in order to avoid unnecessary xchg operations:
         */
        if (atomic_read(&lock->count) >= 0 &&
            (atomic_xchg_acquire(&lock->count, -1) == 1))
            break;

        /*
         * got a signal? (This code gets eliminated in the
         * TASK_UNINTERRUPTIBLE case.)
         */
        if (unlikely(signal_pending_state(state, task))) {
            ret = -EINTR;
            goto err;
        }

        if (use_ww_ctx && ww_ctx->acquired > 0) {
            ret = __ww_mutex_lock_check_stamp(lock, ww_ctx);
            if (ret)
                goto err;
        }
// 设置当前任务状态TASK_UNINTERRUPTIBLE

        __set_task_state(task, state);

        /* didn't get the lock, go to sleep: */
        spin_unlock_mutex(&lock->wait_lock, flags); // 释放自旋锁(此处对lock的成员变量修改读写都已经完成，其他任务可读写修改
        schedule_preempt_disabled();/ 执行一次调度，主动切换到其他任务，等其他任务释放互斥锁是会唤醒当前任务，继续执行下面的函数
        spin_lock_mutex(&lock->wait_lock, flags);// 任务被唤醒，重新获取自旋锁
    }
    __set_task_state(task, TASK_RUNNING);

    mutex_remove_waiter(lock, &waiter, current_thread_info());
    /* set it to 0 if there are no waiters left: */
    if (likely(list_empty(&lock->wait_list)))
        atomic_set(&lock->count, 0);
    debug_mutex_free_waiter(&waiter);

skip_wait:
    /* got the lock - cleanup and rejoice! */
    lock_acquired(&lock->dep_map, ip);
    mutex_set_owner(lock);

    if (use_ww_ctx) {
        struct ww_mutex *ww = container_of(lock, struct ww_mutex, base);
        ww_mutex_set_context_slowpath(ww, ww_ctx);
    }

    spin_unlock_mutex(&lock->wait_lock, flags);
    preempt_enable();
    return 0;

err:
    mutex_remove_waiter(lock, &waiter, task_thread_info(task));
    spin_unlock_mutex(&lock->wait_lock, flags);
    debug_mutex_free_waiter(&waiter);
    mutex_release(&lock->dep_map, 1, ip);
    preempt_enable();
    return ret;
}

就不详细看了 ；

主要是mutex 和semaphore使用区别：一个是互斥一个 计数信号量

以生活现象为例：超市最多允许同时进100个人，一开始100人顺利进入，后面出一个进一个

// 通过sp低13位清零获取到thread_info，再通过thread_info获取到task指针