Linux 드라이버의 동시성 제어 기술: 원리 및 실제

임베디드 Linux 개발자라면 다음 질문에 직면할 수 있습니다. 여러 작업 또는 스레드 간에 장치 리소스를 안전하게 공유하는 방법은 무엇입니까? 데이터 경합과 불일치를 피하는 방법은 무엇입니까? 시스템 성능과 신뢰성을 향상시키는 방법은 무엇입니까? 이러한 문제는 모두 동시성 제어 기술, 즉 공유 리소스에 대한 여러 실행 엔터티의 액세스를 조정하는 방법과 관련됩니다. 이 기사에서는 원자 연산, 스핀 잠금, 세마포어, 뮤텍스 잠금, 읽기-쓰기 잠금, 순차 잠금 및 RCU 등 Linux 드라이버에서 일반적으로 사용되는 동시성 제어 기술을 소개하고 해당 기술의 사용 및 주의 사항에 대한 예를 제공합니다. . 문제.

중요한 리소스를 효과적으로 관리하기 위해 애플리케이션 계층 프로그램에는 동시성을 제어하는 ​​원자 변수, 조건 변수 및 세마포어가 있습니다. 예를 들어 여러 애플리케이션 계층 프로그램에서 드라이버를 호출하는 경우도 있습니다. 이 경우 타임 드라이버의 전역 변수는 동시에 여러 애플리케이션 계층 프로세스의 프로세스 공간에 속하게 됩니다. 이 경우 동시성을 제어하기 위해 일부 기술도 사용해야 합니다. 이 기사에서는 커널에서 다음과 같은 동시성 제어 기술의 기술적 특성과 응용 시나리오에 대해 설명합니다.

1. 인터럽트 마스크
2. 원자적 연산
   1. 원자변수 연산
   2. 원자 비트 연산
3. 스핀 잠금
   1. 전통적인 스핀락
   2. 스핀 잠금 읽기 및 쓰기
   3. 시퀀스 잠금
   4. RCU
4. 신호
   1. 전통적인 세마포어
   2. 세마포어 읽기 및 쓰기
   3. 완료금액
5. 뮤텍스

방해 마스크

이름에서 알 수 있듯이 모든 인터럽트를 차단합니다. 임베디드 시스템에서 인터럽트 차폐는 1. 하드웨어 인터페이스 차폐, 2. 하드웨어 GIC 차폐, 3. CPU(커널) 차폐의 세 가지 수준을 가질 수 있습니다. 인터페이스에서 차단되면 인터럽트가 올 때 인터럽트가 손실되어 전혀 찾을 수 없습니다. GIC에서 보호된 경우 보호 기간 동안 irq_1, irq_2, irq_3 인터럽트가 발생하면 보류 플래그가 하나만 있기 때문에 irq_3이 최종적으로 오면 보류 플래그가 설정되고 보호가 차단됩니다. CPU는 보류 중인 플래그가 있음을 발견합니다. 비트가 설정되어 있으면 계속 처리되지만 1과 2는 확실히 손실됩니다. ARM의 차폐, 즉 커널의 차폐는 설정 방법에 따라 다릅니다. local_irq_disable인 경우 인터페이스의 차폐와 동일합니다. local_irq_save는 마지막 인터럽트를 추적하는 두 번째 것과 동일합니다. 커널에는 인터럽트를 계산하고 이 기간 동안 발생한 인터럽트 수를 알 수 있는 메커니즘도 있습니다. 그러나 실제 작업에서는 대부분의 경우 추적하지 않습니다. 인터럽트가 매우 중요하지 않은 한 누락된 인터럽트.

여기서 논의하고 있는 것은 커널의 인터럽트 마스킹입니다. 커널의 많은 중요한 작업이 인터럽트에 의존하기 때문에 모든 인터럽트를 마스크하는 것은 매우 위험합니다. 내부에서 실행되는 코드는 가능한 한 빨라야 하며, 커널의 프로세스 스케줄링도 인터럽트에 의해 구동되므로 매우 위험합니다. 모든 인터럽트를 마스크하려면 절전 모드를 트리거할 수 있는 코드가 없어야 합니다. 그렇지 않으면 깨울 수 없습니다. 인터럽트 마스킹은 이 CPU의 인터럽트만 마스크하므로 SMP로 인한 경쟁 문제는 해결되지 않습니다. 일반적으로 인터럽트 마스킹은 스핀 잠금으로 보호되는 중요 섹션에 대한 액세스를 방지하기 위해 스핀 잠금과 함께 사용됩니다. 방해로

일반 인터럽트 마스크

으아악

하반부에 인터럽트 차폐

으아악

---

원자적 연산

원자적 연산은 중단할 수 없는 연산입니다. 이는 애플리케이션 계층의 개념과 동일합니다. 커널의 원자적 연산 템플릿은 다음과 같습니다.

정수 원자 변수 으아악

템플릿 으아악

비트 원자 연산

비트 원자 연산은 비트맵과 같은 정보를 기록하기 위해 커널에서 많은 수의 "비트"를 사용합니다. 커널 API는 다음과 같습니다. 으아악

---

스핀 잠금

"제자리에서 회전한다"는 의미입니다. 잠금에 실패하면 회전하고, 스핀 잠금은 원자적 작업이므로 CPU 사용량이 100%까지 올라가게 됩니다. 스핀 잠금을 사용하면 임계 섹션의 코드가 매우 짧아야 합니다. 그렇지 않으면 시스템 성능에 영향을 미칩니다. 또한 일종의 잠금 메커니즘으로 스핀 잠금을 사용할 때에도 주의해야 합니다. 교착 상태 잠금 기간 동안 스핀 잠금을 호출하지 못하면 프로세스가 스핀 잠금(예: copy_from_user(), copy_to_user(), kmalloc(), msleep() 등)을 획득한 후 차단되는 경우 프로세스 예약 기능이 발생할 수 있습니다. 차단이 발생하면 커널이 충돌할 수 있습니다.

Spin lock을 사용하면 SMP 레이스 문제를 해결할 수 있습니다. 다양한 유형의 스핀 잠금에는 다양한 상황에 적합한 자체 처리 메커니즘이 있습니다.

전통적인 스핀 잠금, 읽기-쓰기 스핀 잠금, RCU 메커니즘, 순차 잠금 등

을 포함합니다. 스핀 잠금은 세마포어 및 뮤텍스를 위한 기본 구현 도구입니다.

유형 비교	전통적인 스핀 잠금	스핀 잠금 읽기 및 쓰기	시퀀스 잠금	RCU 메커니즘
응모상황	라커 전용 리소스가 필요합니다	작가 전용 리소스가 필요합니다	읽기와 쓰기를 동시에 할 수 있는 리소스는 거의 없습니다	더 많이 읽고 리소스를 적게 작성하세요
읽기+읽기 동시성	×	√	√	√
읽기 + 쓰기 동시	×	×	√	√
쓰기+쓰기 동시성	×	×	×	√

和其他锁机制的一样，使用自旋锁保护数据分为抢锁-操作-解锁,下面就是一个典型的使用锁的流程，通过自旋锁实现一个文件只被一个进程打开。

int cnt=0;
lock_t lock;
static int my_open()
{
    lock(&lock);
    if(cnt){
        unlock(&lock)
    }
    cnt++;
    unlock(&lock);
}
static int release()
{
    lock(&lock);
    cnt--;
    unlock(&lock);
}

传统自旋锁

是一种比较粗暴的自旋锁，使用这种锁的时候，被锁定的临界区域不允许其他CPU访问，需要注意的是，尽管获得锁之后执行的临界区操作不会被其他CPU和本CPU内其他抢占进程的打扰，但是仍然会被中断和底半部的影响，所以通常我们会使用下述API中的衍生版本，比如上文中提到的将自旋锁+中断屏蔽来防止使用自旋锁访问临界资源的时候被中断打断，对应的宏函数就是spin_lock_irq和spin_lock_irqsave。

//定义并初始化自旋锁
spinlock_t spinlock
void spin_lock_init(spinlock_t *);

//加锁
//spin_lock - 加锁函数（忙等）
void spin_lock(spinlock_t *lock);
int spin_trylock(spinlock_t *lock);
spin_lock_irq();    //=spin_lock() + local_irq_disable()
spin_lock_irqsave(); //= spin_lock() + lock_irq_save();
spin_lock_bh();     //=spin_lock() + local_bh_disable();

//解锁
void spin_unlock(spinlock_t *lock);
spin_unlock_irq();    //=spin_unlock() + local_irq_enable()
spin_unlock_irqrestore(); //= spin_unlock() + lock_irq_restore();
spin_unlock_bh();     //=spin_unlock() + local_bh_enable();

读写自旋锁

传统的自旋锁粗暴的将临界资源划归给一个CPU，但是很多资源都不会因为读而被破坏，所以我们可以允许多个CPU同时读临界资源，但不允许同时写资源，类似于应用层的文件锁，内核的读写锁机制同样有下述互斥原则:

• 读者 + 读者 不互斥
• 读者 + 写者 互斥
• 写者 + 写者 互斥

//include/linux/rwlock.h
//定义并初始化自旋锁
rwlock_t rwlock;
void rwlock_init(rwlock_t *lock);

//加读锁
void read_lock(rwlock_t *lock);
int read_trylock(rwlock_t *lock);
void read_lock_irqsave(rwlock_t *lock,unsigned long flags);
void read_lock_irq(rwlock_t *lock, unsigned long flags);
void read_lock_bh(rwlock_t *lock);

//解读锁
void read_unlock(rwlock_t *lock)
void read_unlock_irqrestrore(rwlock_t *lock,unsigned long flags);
void read_unlock_irq(rwlock_t *lock, unsigned long flags);
void read_unlock_bh(rwlock_t *lock);

//加写锁
void write_lock(rwlock_t *lock)
int write_trylock(rwlock_t *lock)
void write_lock_irqsave(rwlock_t *lock,unsigned long flags);
void write_lock_irq(rwlock_t *lock, unsigned long flags);
void write_lock_bh(rwlock_t *lock);

//解写锁
void write_unlock(rwlock_t *lock)
void write_unlock_irqrestrore(rwlock_t *lock,unsigned long flags);
void write_unlock_irq(rwlock_t *lock, unsigned long flags);
void write_unlock_bh(rwlock_t *lock);

顺序锁

顺序锁可以看作读写锁的升级版，读写锁不允许同时存在读者和写者，顺序锁对这一情况进行了改良，它允许写者和读者同时访问临界区，不必再向读写锁那样读者要读必须等待写者写完，写者要写必须等待读者读完。不过，使用顺序锁的时候，临界区不能有指针，因为写者可能会修改指针的指向，如果读者去读，就会Oops，此外，如果读者读过的数据被写者改变了，读者需要重新读，以维持数据是最新的，虽然有这两个约束，但是顺序锁提供了比读写锁更大的灵活性。对于写者+写者的情况，顺序锁的机制和读写锁一样，必须等！

• 读者 + 读者 不互斥
• 读者 + 写者 不互斥 , 临界区没有指针+读者需自己注意更新
• 写者 + 写者 互斥

//include/linux/seqlock.h
//定义顺序锁
struct seqlock_t sl;

//获取顺序锁
void write_seqlock(seqlock_t *sl);
void write_tryseqlock(seqlock_t *sl);
void write_seqlock_irqsave(lock,flags); //=local_irq_save() + write_seqlock()
void write_seqlock_irq(seqlock_t *sl);  //=local_irq_disable() + write_seqlock()
void write_seqlock_bh(seqlock_t *sl);   //local_bh_disable() + write_seqlock()

//释放顺序锁
void write_sequnlock(seqlock_t *sl);
void write_sequnlock_irqsave(lock,flags); //=local_irq_restore() + write_sequnlock()
void write_sequnlock_irq(seqlock_t *sl);  //=local_irq_enable() + write_sequnlock()
void write_sequnlock_bh(seqlock_t *sl);   //local_bh_enable() + write_sequnlock()

//读开始
unsigned read_seqbegin(const seqlock_t *sl);
read_seqbegin_irqsave(lock,flags);  //=local_irq_save() + read_seqbegin();

//重读
int read_seqretry(const seqlock_t *sl,unsigned iv);
read_seqretry_irqrestore(lock,iv,flags);    //=local_irq_restore() + read_seqretry();

RCU

RCU即Read-Copy Update，即读者直接读，写者先拷贝再择时更新，是另外一种读写锁的升级版，这种机制在VFS层被大量使用。正如其名，读者访问临界资源不需要锁，从下面的rcu_read_lock的定义即可看出，写者在写之前先将临界资源进行备份，去修改这个副本，等所有的CPU都退出对这块临界区的引用后，再通过回调机制，将引用这块资源的原指针指向已经修改的备份。从中可以看出，在RCU机制下，读者的开销大大降低，也没有顺序锁的指针问题，但是写者的开销很大，所以RCU适用于读多写少的临界资源。如果写操作很多，就有可能将读操作节约的性能抵消掉，得不偿失。

• 读者 + 读者 不互斥
• 读者 + 写者 不互斥 , 读者自己注意更新
• 写者 + 写者 不互斥 ，写者之间自己去同步

内核会为每一个CPU维护两个数据结构-rcu_data和rcu_bh_data，他们用于保存回调函数，函数call_rcu()把回调函数注册到rcu_data，而call_rcu_bh()则把回调函数注册到rcu_bh_data，在每一个数据结构上，回调函数们会组成一个队列。

使用RCU时，读执行单元必须提供一个信号给写执行单元以便写执行单元能够确定数据可以被安全地释放或修改的时机。内核中有一个专门的垃圾收集器来探测读执行单元的信号，一旦所有的读执行单元都已经发送信号告诉收集器自己都没有使用RCU的数据结构，收集器就调用回调函数完成最后的数据释放或修改操作。

读

读即RCU中的R，从下面的宏定义可以看出，读操作其实就是禁止内核的抢占调度，并没有使用一个锁对象。

//读锁定
//include/linux/rcupdate.h
rcu_read_lock();            //preempt_disbale()
rcu_read_lock_bh();         //local_bh_disable()

//读解锁
rcu_read_unlock()           //preempt_enable()
rcu_read_unlock_bh();       //local_bh_enable()

同步

同步即是RCU写操作的最后一个步骤-Update，下面这个接口会则色写执行单元，直到所有的读执行单元已经完成读执行单元临界区，写执行单元才可以继续下一步操作。如果有多个RCU写执行单元调用该函数，他们将在一个grace period(即所有的读执行单元已经完成对临界区的访问)之后全部被唤醒。

synchrosize_rcu()

挂起回调

下面这个接口也由RCU写执行单元调用，它不会使写执行单元阻塞，因而可以在中断上下文或软中断中使用，该函数把func挂接到RCU回调函数链上，然后立即返回。函数sychrosize_rcu()其实也会调用call_rcu()。

void call_rcu(struct rcu_head *head,void (*func)(struct rcu_head *rcu));

下面这个接口会将软中断的完成也当作经历一个quiecent state(静默状态)，因此如果写执行单元调用了该函数，在进程上下文的读执行单元必须使用rcu_read_lock_bh();

void call_rcu_bh(struct rcu_head *head,void (*func)(struct rcu_head *rcu));

RCU机制被大量的运用在内核链表的读写中，下面这些就是内核中使用RCU机制保护的数据结构，函数的功能和非RCU版本一样，具体可以参考内核文件**”include/linux/list.h”**，只不过这里的操作都会使用RCU保护起来。

void list_add_rcu(struct list_head *new, struct list_head *head);
void list_add_tail_rcu(struct list_head *new,struct list_head *head);
void list_del_rcu(struct list_head *entry);
void list_replace_rcu(struct list_head *old,struct list_head *new);
list_for_each_rcu(pos,head);
list_for_each_safe_rcu(pos,n,head);
list_for_each_entry_rcu(pos,head,member);
void hlist_del_rcu(struct hlist_node *n);
void hlist_add_head_rcu(struct hlist_node *n, struct hlist_head *h);
list_for_each_rcu(pos,head);
hlist_for_each_entry_rcu(tpos,pos,head,member);

信号量

自旋锁一节提过，如果一个CPU不能获取临界资源，就会造成”原地自旋”，所以自旋锁保护的临界区的执行时间不能太长，但如果我们的确需要保护一段执行时间比较长的临界区呢？答案就是信号量
，信号量的底层依托于自旋锁来实现其原子性，并进一步将其提高到”进程”的维度，称为一种可以运行在进程上下文的”锁”，正是这种能运行在进程上下文的能力赋予了信号量和自旋锁的诸多不同。

使用信号量，如果试图获取信号量的进程获取失败，内核就会将其调度为睡眠状态，执行其他进程，避免了CPU的忙等。不过，进程上下文的切换也是有成本的，所以通常，信号量在内核中都是只用于保护大块临界区。

此外，一个进程一旦无法获取期待的信号量就会进入睡眠，所以信号量保护的临界区可以有睡眠的代码。在这方面，自旋锁保护的区域是不能睡眠也不能执行schedule()的，因为一旦一个抢到了锁的CPU进行了进程上下文的切换或睡眠，那么其他等待这个自旋锁的CPU就会一直在那忙等，就永远无法等到这个锁，，形成死锁，除非有其他进程将其唤醒(通常都不会有)。

也正是由于信号量操作可能引起阻塞，所以信号量不能用于中断上下文。总结一下刚才罗嗦这一段：

을 할 수 없습니다. 할 수 있습니다.

프로젝트	세마포어	스핀 잠금
중요 섹션 시간	프로세스 전환 시간이 단축됩니다	Critical 섹션 실행 시간이 더 짧아졌습니다
프로세스 컨텍스트	중요 섹션은 잠자기 또는 예약 가능	중요 섹션은 잠자기 또는 일정
중단 컨텍스트	down_trylock()만이	예

传统信号量

内核的信号量和应用层的信号量的使用方式类似，但没有获取信号量这一步骤，因为内核中中的信号量可以映射到所有调用这个模块的用户进程的内核空间。这些用户进程也就直接共享了一个信号量，所以也就没有获取信号量一说，相关的内容我在”Linux IPC System V 信号量”一文中有所讨论。

和应用层的信号量一样，内核信号量也是用于对临界资源的互斥/顺序访问，同样，虽然在使用信号量的时候我们可以初始化为任意值，但实际操作上我们通常只初始化为1或0，下述是Linux内核提供的信号量API。

//include/linux/semaphore.h
//定义并初始化semaphore对象
struct semphore sem;

//初始化信号量
void sem_init(struct semaphore * sem,int val);
init_MUTEX(sem);
init_MUTEX_LOCKED(sem);
DECLARE_MUTEX(sem);
DECLARE_MUTEX_LOCKED(sem);

//P操作
//down()会导致睡眠，不能用于中断上下文
void down(struct semaphore *sem);
//down_interruptible同样会进入休眠，但能被打断
int down_interruptible(struct semaphore *sem);
//down_trylock不能获得锁时会立即返回，不会睡眠，可以用在中断上下文
int down_trylock(struct semaphore *sem);

//V操作
void up(struct semaphore *sem);

读写信号量

读写信号量与信号量的关系 和 读写自旋锁与自旋锁的关系类似，他们的互斥逻辑都是一样的，这里不再赘述

//定义并初始化读写信号量
struct rw_semaphore my_rwsem;
void init_rwsem(struct rw_semaphore *sem);

//P读信号量
void down_read(struct rw_semaphore *sem);
int down_read_trylock(struct rw_semaphore *sem);

//V读信号量
void up_read(struct rw_semaphore *sem);

//P写信号量
void down_write(struct rw_semaphore *sem);
int down_write_trylock(struct rw_semaphore *sem);

//V写信号量
void up_write(struct rw_semaphore *sem);

模板

struct rw_semaphore my_rwsem;
void init_rwsem(&my_rwsem);

//读前获取读信号量
down_read(&my_rwsem);   //若要非阻塞：down_read_trylock(&my_rwsem);

/* 读临界区 */

//读完释放读信号量

up_read(&my_rwsem); 

//写前获取写信号量
down_write(&my_rwsem);  //若要非阻塞：down_write_trylock(&my_rwsem); 

/* 写临界区 */

//写完释放写信号量
up_write(&my_rwsem); 

完成量

完成量用于一个执行单元等待另一个执行单元执行完某事，和传统信号量一样，主要是用来实现队临界区的顺序/互斥访问。但是完成量还提供一种唤醒一个或唤醒所有等待进程的接口，有点类似与应用层的条件变量。

//定义并初始化完成量
struct completion my_completion;
init_completion(&my_completion);
//或
DECLARE_COMPLETION(my_completion)

//等待completion
void wait_for_completion(struct completion *c);

//唤醒completion
void complete(struct completion *c);        //只唤醒一个等待的执行单元
void complete_all(struct completion *c);    //释放所有等待该完成量的执行单元

互斥体

除了信号量，Linux内核还提供了一种专门用于实现互斥的机制-互斥体，相关的内核API如下：

//include/linux/mutex.h
//定义并初始化mutex对象
struct mutex my_mutex;
mutex_init(&my_mutex);

//获取mutex
void mutex_lock(struct mutex *lock);
int mutex_trylock(struct mutex *lock);
int mutex_lock_interruptible(struct mutex *lock);

//释放mutex
void mutex_unlock(struct mutex *lock);

通过本文，我们了解了Linux驱动中的并发控制技术，它们各有优缺点和适用场景。我们应该根据实际需求选择合适的技术，并遵循一些基本原则，如最小化临界区，避免死锁，减少上下文切换等。并发控制技术是Linux驱动开发中不可或缺的一部分，它可以保证设备资源的正确性和高效性，也可以提升系统的稳定性和可扩展性。希望本文能够对你有所帮助和启发。

위 내용은 Linux 드라이버의 동시성 제어 기술: 원리 및 실제의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!