Technologie de contrôle de concurrence dans les pilotes Linux : principes et pratique

Si vous êtes un développeur Linux embarqué, vous pourriez être confronté à cette question : Comment partager en toute sécurité les ressources de l'appareil entre plusieurs tâches ou threads ? Comment éviter les courses aux données et les incohérences ? Comment améliorer les performances et la fiabilité du système ? Ces problèmes impliquent tous la technologie de contrôle de concurrence, c'est-à-dire la manière de coordonner l'accès de plusieurs entités d'exécution aux ressources partagées. Dans cet article, nous présenterons les technologies de contrôle de concurrence couramment utilisées dans les pilotes Linux, notamment les opérations atomiques, les verrous tournants, les sémaphores, les verrous mutex, les verrous en lecture-écriture, les verrous séquentiels et RCU, etc., et donnerons des exemples de leur utilisation et de leur attention. . matière.

Afin de parvenir à une gestion efficace des ressources critiques, les programmes de la couche application disposent de variables atomiques, de variables de condition et de sémaphores pour contrôler la concurrence. Le même problème existe également dans le développement de pilotes. Par exemple, un pilote est appelé par plusieurs programmes de couche application. en même temps.Les variables globales du pilote temporel appartiendront simultanément à l'espace de processus de plusieurs processus de couche d'application. Dans ce cas, certaines technologies doivent également être utilisées pour contrôler la concurrence. Cet article abordera les caractéristiques techniques et les scénarios d'application des technologies de contrôle de concurrence suivantes dans le noyau.

1. Masque d'interruption
2. Opérations atomiques
   1. Opérations de variables atomiques
   2. Opérations sur les bits atomiques
3. verrouillage de la rotation
   1. Verrouillage rotatif traditionnel
   2. Lire et écrire le spin lock
   3. Verrouillage de séquence
   4. RCU
4. signal
   1. Sémaphore traditionnel
   2. Lire et écrire des sémaphores
   3. Montant complété
5. Mutex

Masque d'interruption

Comme son nom l'indique, il bloque toutes les interruptions. Dans les systèmes embarqués, le blindage contre les interruptions peut avoir trois niveaux : 1. Blindage de l'interface matérielle, 2. Blindage matériel GIC, 3. Blindage du CPU (noyau). Si elle est bloquée au niveau de l'interface, l'interruption sera perdue lorsqu'elle arrivera et ne pourra pas être trouvée du tout. S'il est protégé au niveau du GIC, alors si les interruptions irq_1, irq_2, irq_3 surviennent pendant la période de protection, car il n'y a qu'un seul indicateur en attente, l'indicateur en attente sera défini lorsque irq_3 arrivera enfin, puis le blindage sera débloqué, et le Le processeur détecte qu'il y a un bit en attente. Si le bit est activé, il sera toujours traité, mais 1 et 2 seront définitivement perdus. Le blindage au niveau d'ARM, c'est-à-dire le blindage dans le noyau, dépend de la façon dont il est défini. S'il est local_irq_disable, alors s'il est perdu, c'est le même que le blindage au niveau de l'interface. local_irq_save, c'est le même que le second, poursuivant la dernière interruption. Le noyau dispose également d'un mécanisme correspondant pour compter les interruptions et savoir combien d'interruptions sont survenues pendant cette période. Cependant, dans les opérations réelles, dans la plupart des cas, nous ne poursuivrons pas. les interruptions manquées sauf si l'interruption est très importante.

Ce dont nous discutons ici, c'est du masquage des interruptions dans le noyau. Étant donné que de nombreuses opérations importantes dans le noyau reposent sur des interruptions, il est très dangereux de masquer toutes les interruptions. Le code exécuté à l'intérieur doit être aussi rapide que possible. De plus, puisque la planification des processus du noyau est également pilotée par les interruptions, cela est très dangereux. pour masquer toutes les interruptions. Il ne doit y avoir aucun code susceptible de déclencher la mise en veille, sinon il ne peut pas être réveillé. Notez que le masquage d'interruption masque uniquement les interruptions de ce processeur, il ne résout donc pas le problème de concurrence causé par SMP. Habituellement, le masquage d'interruption est utilisé conjointement avec les verrous de rotation pour empêcher l'accès aux sections critiques protégées par les verrous de rotation interrompus. par interruption

Masque d'interruption normale

local_irq_disable();    //屏蔽中断
//或
local_irq_save(flags);  //屏蔽中断并保存目前CPU中的中断位信息

/* 临界区 */

local_irq_enable();     //解除屏蔽
//或
local_irq_restore(flags);   //解除屏蔽并恢复中断位信息

Blindage d'interruption dans la moitié inférieure

local_bh_disable();     //屏蔽中断

/* 临界区 */

local_bh_enable();

---

Opérations atomiques

Les opérations atomiques sont des opérations qui ne peuvent pas être interrompues. Elles sont identiques au concept de la couche application. Le modèle d'opération atomique dans le noyau est le suivant :

Variable atomique entière

//asm/atomic.h
//创建并初始化原子变量
atomic_t tv = ATOMIC_INIT(初值);
//读原子变量
int atomic_read(atomic_t *v);
//写原子变量
void atomic_set(atomic_t *v, int i); 

/**
 *atomic_dec_and_test - 尝试将原子变量-1
 *v:如果-1之后原子变量变为0，返回非0, 否则返回0
 */
int atomic_dec_and_test(volatile atomic_t *v);
int atomic_inc_and_test(volatile atomic_t *v);
int atomic_sub_and_test(int i, volatile atomic_t *v);

//操作并返回
int atomic_add_return(int i, atomic *v);
int atomic_sub_return(int i, atomic *v);
int atomic_inc_return(atomic *v);
int atomic_dev_return(atomic *v);

modèle

static atomic_t tv;
static int demo_open(struct inode *inode, struct file *filp)
{
    if(!atomic_dec_and_test(&tv)){
        atomic_inc(&tv);
        return -EBUSY;
    }
    /* 操作代码 */
    return 0;
}
static int demo_release(struct inode *inode, struct file *filp)
{
    atomic_inc(&tv);
    return 0;
}

static int __init demo_init(void)
{
    // init atomic_t
    atomic_set(&tv, 1);
}

opérations atomiques sur bits

Les opérations atomiques sur les bits sont des opérations sur les bits atomiques. Un grand nombre de "bits" sont utilisés dans le noyau pour enregistrer des informations, telles que les bitmaps. Ces opérations doivent être atomiques.

//设置位
void set_bit(nr,void *addr);

//清除位
void clear_bit(nr,void *addr);

//改变位
void change_bit(nr,void *addr);

//测试位
test_bit(nr, void *addr);

//测试并操作位
int test_and_set_bit(nr, void *addr);
int test_and_clear_bit(nr,void *addr);
int test_and_change_bit(nr,void *addr);

---

Spin Lock

Cela signifie "tourner sur place". Lorsque le verrouillage échoue, tournez et le verrouillage de rotation continuera à occuper le processeur pour les tests de variables. Puisqu'il s'agit d'une opération atomique, l'utilisation du processeur augmentera à 100 %, donc lorsque. en utilisant des verrous tournants, le code dans la section critique doit être très court, sinon cela affectera les performances du système. De plus, en tant que sorte de mécanisme de verrouillage, lorsque vous utilisez des verrous tournants, vous devez également faire attention à l'apparition de blocages. Le fait de ne pas appeler le verrou tournant pendant la période de verrouillage peut entraîner des fonctions de planification de processus, si le processus se bloque après avoir obtenu le verrou tournant, par exemple, copy_from_user(), copy_to_user(), kmalloc(), msleep(), etc., une fois le blocage survenu. , cela peut provoquer un crash du noyau.

Spin Lock peut être utilisé pour résoudre les problèmes de course SMP. Différents types de verrous tournants ont leurs propres mécanismes de traitement, adaptés à différentes situations. Y compris

les verrous de rotation traditionnels, les verrous de rotation en lecture-écriture, les mécanismes RCU, les verrous séquentiels, etc. Les verrous de rotation sont les outils d'implémentation sous-jacents aux sémaphores et aux mutex.

Comparer les types	Verrouillage rotatif traditionnel	Lire et écrire le verrouillage rotatif	Verrouillage de séquence	Mécanisme RCU
Occasions de candidature	Nécessite des ressources exclusives au casier	Nécessite des ressources exclusives à l'écrivain	Il existe peu de ressources capables de lire et d'écrire en même temps	Lisez plus et écrivez moins de ressources
Concurrence lecture+lecture	×	√	√	√
Lecture + écriture simultanées	×	×	√	√
Concurrence d'écriture + écriture	×	×	×	√

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

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就会一直在那忙等，就永远无法等到这个锁，，形成死锁，除非有其他进程将其唤醒(通常都不会有)。

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

Projet	Sémaphore	Verrouillage rotatif
temps de section critique	Le temps de commutation du processus est plus court	Le temps d'exécution des sections critiques est plus court
Contexte du processus	La section critique peut dormir ou programmer	Section critiquene peut pas mettre en veille ou planifier
Contexte d'interruption	Seul down_trylock() peut	Oui

传统信号量

内核的信号量和应用层的信号量的使用方式类似，但没有获取信号量这一步骤，因为内核中中的信号量可以映射到所有调用这个模块的用户进程的内核空间。这些用户进程也就直接共享了一个信号量，所以也就没有获取信号量一说，相关的内容我在”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驱动开发中不可或缺的一部分，它可以保证设备资源的正确性和高效性，也可以提升系统的稳定性和可扩展性。希望本文能够对你有所帮助和启发。

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!