奔跑吧Linux内核--ch4并发与同步

目录

内核中4种并发源

  • 中断和异常
  • 软中断和tasklet
  • 内核抢占
  • 多核处理器并发执行

保护资源或数据,而非保护代码

原子操作和内存屏障

[include/linux/types.h]

typedef struct{
    int counter;
} atomic_t;

在ARM处理器中,如何实现独占访问内存

LL/SC操作:Load-Link(LL)操作返回一个内存地址上当前存储的值,后面的Store-Conditional(SC)操作,会向这个内存地址写入一个新值,但是只有在这个内存地址上存储的值,从上个LL操作开始直到现在都没有发生改变的情况下,写入操作才能成功,否则都会失败。

对于ARM平台来说,在硬件层面上提供了对LL/SC的支持,LL操作用的是LDREX指令,SC操作用的是STREX指令。

  • LDREX Rx, [Ry] 读取寄存器Ry指向的4字节内存值,将其保存到Rx寄存器中,同时标记对Ry指向内存区域的独占访问。
    如果执行LDREX指令的时候发现已经被标记为独占访问了,并不会对指令的执行产生影响。

  • STREX Rx, Ry, [Rz] STREX在更新内存数值时,会检查该段内存是否已经被标记为独占访问,并以此来决定是否更新内存中的值。如果执行这条指令的时候发现已经被标记为独占访问了,则将寄存器Ry中的值更新到寄存器Rz指向的内存,并将寄存器Rx设置成0。指令执行成功后,会将独占访问标记位清除。而如果执行这条指令的时候发现没有设置独占标记,则不会更新内存,且将寄存器Rx的值设置成1。一旦某条STREX指令执行成功后,以后再对同一段内存尝试使用STREX指令更新的时候,会发现独占标记已经被清空了,就不能再更新了,从而实现独占访问的机制

ARM有2种独占监视器(Exclusive Monito):每一个处理器内部都有一个本地监视器(Local Monitor),且在整个系统范围内还有一个全局监视器(Global Monitor)

atomic_cmpxchg()和atomic_xchg()分别表示什么含义

[include/asm-generic/atomic.h]

#define atomic_cmpxchg(v, old, new) 比较old和原子变量v的值,如果相等则把new赋值给v,返回原子变量v的旧值
#define atomic_xchg(v, new) 把new赋值给原子变量v,返回原子变量v的旧值

spinlock

为什么spinlock的临界区不能睡眠(不考虑RT-Linux)

spinlock是忙等待锁,锁持有者应尽快完成临界区的执行任务。

Linux内核中经典spinlock的实现有什么缺点?

[include/linux/spinlock_types.h]

typedef struct spinlock { 
        struct raw_spinlock rlock;  
} spinlock_t;
 
typedef struct raw_spinlock { 
    arch_spinlock_t raw_lock; 
} raw_spinlock_t;

2.6.23版本的内核中,和arm平台相关的spinlock数据结构(这时候还没有arch_spinlock_t):

typedef struct { 
    volatile unsigned int lock; 
} raw_spinlock_t;

0表示unlocked,1表示locked。这种实现的缺点是不公平,因为刚刚释放锁的CPU的L1 cache存储了该锁,它比别的CPU更快的获得锁。

为什么spinlock临界区不允许发生抢占?

如果允许抢占,那么在临界区中发生中断时,中断返回会检查抢占调度。这有两个问题:

一、抢占调度相当于持有锁的进程睡眠,违背了spinlock的设计;

二、抢占调度进程也可能会申请spinlock锁,发生死锁

Ticket-based的spinlock机制是如何实现的?

[arch/arm/include/asm/spinlock_types.h]

typedef struct { 
    union { 
        u32 slock; 
        struct __raw_tickets { 
            u16 owner; 
            u16 next; 
        } tickets; 
    }; 
} arch_spinlock_t;

owner表示持有者号牌,next表示排队队列末尾者的号牌。owner等于next的获得锁,并++next。这时其他线程想获得锁时,++next作为自己的号牌。持有锁的线程释放锁时++owner,让下一个next等于owner的线程获得锁。实现"FIFO ticket-based"。

加锁

static inline void arch_spin_lock(arch_spinlock_t *lock) 
{ 
    unsigned long tmp; 
    u32 newval; 
    arch_spinlock_t lockval;
 
    prefetchw(&lock->slock);
    __asm__ __volatile__( 
"1:    ldrex    %0, [%3]
" 
"    add    %1, %0, %4
" )
"    strex    %2, %1, [%3]
"
"    teq    %2, #0
"
"    bne    1b" 
    : "=&r" (lockval), "=&r" (newval), "=&r" (tmp) 
    : "r" (&lock->slock), "I" (1 << TICKET_SHIFT) 
    : "cc");
 
    while (lockval.tickets.next != lockval.tickets.owner) {
        wfe();
        lockval.tickets.owner = ACCESS_ONCE(lock->tickets.owner);
    }
 
    smp_mb();
}

和原子变量一样,使用ldrex和strex指令,将next+1.之后检查next和owner域是否相等,不相等时用wfe指令让CPU进入等待状态。被唤醒时owner必然被改变了,更新owner值再次比较。

释放锁

static inline void arch_spin_unlock(arch_spinlock_t *lock)
{
  smp_mb();  
  lock->tickets.owner++;
  dsb_sev(); 
}

smp_mb屏障保证对内存的访问指令执行完成,然后owner加1,调用dsb_sev,保证owner写入内存,且执行SEV指令唤醒因WFE指令进入睡眠状态的CPU

如果再spin_lock()和spin_unlock()的临界区中发生了中断,并且中断处理程序也恰巧修改了该临界资源,那么会发生什么后果?该如何避免呢?

死锁,中断处理程序进入忙等或者WFE睡眠状态。可以使用spin_lock_irq()在获取spinlock时关闭本地CPU中断以避免这种情况。

信号量

与spinlock相比,信号量有哪些特点?

允许睡眠;在同一时刻允许有多个锁持有者

请简述信号量是如何实现的。

include/linux/semaphore.h

struct semaphore{
    raw_spinlock_t lock;
    unsigned int count; // 允许进入临界区的内核执行路径个数
    struct list_head wait_list;  //用于管理所有在该信号量上睡眠的进程
};

down操作

int down_interruptible(struct semaphore *sem) {
  unsigned long flags;
  int result = 0;
  spin_lock_irqsave(&sem->lock, flags);  // 关闭本地CPU中断,对自旋锁上锁
  if (likely(sem->count > 0))
    sem->count--;
  else
    result = __down_interruptible(sem);  // 见下
  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);
}

static inline int __sched __down_common(struct semaphore *sem, longstate,
                                        longtimeout) {
  struct task_struct *task = current;  //当前进程
  struct semaphore_waiterwaiter;
  list_add_tail(&waiter.list, &sem->wait_list);  //加到信号量的wait_list中
  waiter.task = task;
  waiter.up = false;
  for (;;) {
    if (signal_pending_state(state, task))  // 进程被其他人发送的信号唤醒
      goto interrupted;
    if (unlikely(timeout) <= 0)
      goto timed_out;
    __set_task_state(task, state);
    raw_spin_unlock_irq(&sem->lock);  // 下面要睡眠了,spinlock不允许睡眠,这里释放锁,被唤醒后在获得锁
    timeout = schedule_timeout(timeout);  //主动让出CPU,相当于让当前进程睡眠
    spin_lock_irq(&sem->lock);
    if (waiter.up) return 0;  //当waiter.up为true时,说明睡眠在wait_list中的进程被信号量UP操作唤醒
  }
timed_out:
  list_del(&waiter.list);
  return -ETIME;
interrupted:
  list_del(&waiter.list);
  return -EINTR;
}

up操作

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

  spin_lock_irqsave(&sem->lock, flags);
  if (likely(list_empty(&sem->wait_list)))
    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, structsemaphore_waiter, list);
  list_del(&waiter->list);
  waiter->up = true;
  wake_up_process(waiter->task);
}

Mutex互斥体

Linux内核已经实现了信号量机制,为何要单独设置一个Mutex机制呢?

Mutex语义比信号量简单,在锁争用激烈的测试场景下,Mutex比信号量执行速度更快,可扩展性更好。
Mutex数据结构定义比信号量小。

include/linux/mutex.h

struct mutex {
  atomic_t count;
  spinlock_t wait_lock;
  struct list_head wait_list;
#if defined(CONFIG_MUTEX_SPIN_ON_OWNER)
  struct task_struct *owner;  // 指向锁持有者的task_struct
#endif
#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
  struct optimistic_spin_queue osq;  // Spinner MCS lock
#endif
}

请简述MCS锁机制的实现原理

自旋等待机制(optimistic spinning):锁持有者在临界区且没有其他优先级更高的进程要被调度时,当前进程乐观的认为锁持有者会很快释放锁,与其睡眠不如自旋等待锁释放,以减少睡眠唤醒的开销。

MCS锁:内核通过MCS锁机制保证只有一个人自旋等待锁持有者释放锁。

Linux 2.6.25 spinlock排队自旋,由于所有线程都在一个共享变量上自旋、修改, 在锁争用激烈的情况,可能因cacahe一致性原理导致cacheline bouncing,使CPU的cacheline反复失效。MCS的设计可以减少这种情况,让锁申请者只在本地CPU变量上自旋。

OSQ锁是MCS锁的一种具体实现:

include/linux/osq_lock.h

/*
 * An MCS like lock especially tailored for optimistic spinning for sleeping
 * lock implementations (mutex, rwsem, etc).
 */
struct optimistic_spin_node {
  struct optimistic_spin_node *next, *prev;
  int locked; /* 1 if lock acquired */
  int cpu;    /* encoded CPU # + 1 value */
};

struct optimistic_spin_queue {
  /*
   * Stores an encoded value of the CPU # of the tail node in the queue.
   * If the queue is empty, then it's set to OSQ_UNLOCKED_VAL.
   */
  atomic_t tail;
};

kernel/locking/osq_lock.c


在编写内核代码时,该如何选择信号量和Mutex?

中断上下文中选择spinlock,如果临界区有睡眠,隐含睡眠的动作及内核API,应避免spinlock。
除非代码场景不符合上述Mutex约束的某一条,否则应优先使用Mutex

  • 同一时刻只有一个线程可以持有Mutex
  • 只有持有者可以解锁
  • 不允许递归加锁和解锁
  • 当进程持有Mutex时,不可退出
  • Mutex可以睡眠,所以不允许在中断处理程序或者中断下半部中使用,如tasklet、定时器

读写锁

什么时候使用读者信号量,什么时候使用写者信号量,由什么来判断?

对临界区资源读用读者信号量,可以有多个读者同时存在临界区;
对临界区资源写用写者信号量,只有一个写者信号量能在临界区。

读写信号量使用的自旋等待机制是如何实现的?

注:读写信号量定义:

[include/linux/rwsem.h]

struct rw_semaphore {
  long count;                       // 读、写者信息,见下
  struct list_head wait_list;       //未获取到读写信号量,在上面睡眠的进程
  raw_spinlock_t wait_lock;         //保护count
#ifdef CONFIG_RWSEM_SPIN_ON_OWNER
  struct ptimistic_spin_queue osq;  //MCS锁
  struct task_struct *owner;        //写者获取到锁时,指向写者
#endif
};

cout值含义

  1. 0x0000_0000 初始化值
  2. 0x0000_000X X个活跃的读锁
  3. 0xffff_000X X个活跃读锁,写锁睡眠;或有一个写锁活跃,X个读锁睡眠
  4. 0xffff_0000 WAITING_BIAS,读者写者都没有获取到锁

RCU

RCU相比读写锁有哪些优势?

请解释Quiescent State 和 Grace Period。

请简述RCU实现的基本原理

在大型系统中,经典RCU遇到了什么问题?Tree RCU又是如何解决该问题的?

在RCU实现中,为什么要使用ULONG_CMP_GE()和ULONG_CMP_LT()宏来比较两个数的大小,而不直接使用大于号或小于号来比较?

请简述一个Grace Period的生命周期及其状态机的变化

内存管理中的锁

请总结原子操作、spinlock、信号量、读写信号量、Mutex和RCU等Linux内核常用锁的特点和使用规则

在KSM中扫描某个VMA寻找有效的匿名页面,假设此VMA恰巧被其他CPU销毁了,会不会有问题呢?

请简述页面锁PG_locked的常用使用方法

在mm/rmap.c文件中的page_get_anon_vma()函数中,为什么要使用rcu_read_lock()?什么时候注册RCU回调函数呢?

在mm/oom_kill.c的select_bad_process()函数中,为什么要使用rcu_read_lock()?什么时候注册RCU回调函数呢?

原文地址:https://www.cnblogs.com/pusidun/p/13925309.html