linux驱动移植-进程同步之自旋锁

2022-01-01 18:51:56

一、自旋锁(spinlock)

1.1 什么是自旋锁

自旋锁(spinlock)是一种典型的对临界资源进行互斥访问的手段，其名称来源于它的工作方式。

为了获得一个自旋锁，在某CPU上运行的代码需先执行一个原子操作，该操作测试并设置（Test-AndSet）某个内存变量。由于它是原子操作，所以在该操作完成之前其他执行单元不可能访问这个内存变量。

如果测试结果表明锁已经空闲，则程序获得这个自旋锁并继续执行；
如果测试结果表明锁仍被占用，程序将在一个小的循环内重复这个“测试并设置”操作，即进行所谓的“自旋”，通俗地说就是“在原地打转”。

当自旋锁的持有者通过重置该变量释放这个自旋锁后，某个等待的“测试并设置”操作向其调用者报告锁已释放。

驱动程序在持有自旋锁时绝对不能进入睡眠，而在拥有信号量时就可以：自旋锁禁止内核抢占；而信号量不禁止内核抢占。基于这个原因：

如果自旋锁在锁住以后进入睡眠，由于不能进行内核抢占，其他系统进程将都不能获得CPU而运行，因此不能唤醒睡眠的自旋锁，因此系统将不响应任何操作（除了中断或多核的情况）；
而信号量在临界区睡眠后，其他进程可以用内核抢占的方式继续运行，从而可以进程调用up函数释放信号量而使得睡眠的信号量程序由于获得了等待的资源而被唤醒，从而恢复了正常的代码运行；

当然，自旋锁的睡眠的情况包含考虑多核CPU和中断的因素。自旋锁睡眠时，只是当前CPU的睡眠以及当前CPU的禁止内核抢占，所以，如果存在多个CPU，那么其他活动的CPU可以继续运行使操作系统功能正常，并有可能完成相应工作而唤醒睡眠了的自旋锁，从而没有造成系统死机；自旋锁睡眠时，如果允许中断处理，那么中断的代码是可以正常运行的，但是中断通常不会唤醒睡眠的自旋锁，因此系统仍然运行不正常。

1.2 自旋锁具有的特点

spinlock是一种死等的锁机制；
semaphore可以允许多个执行单元进入，spinlock不行，一次只能有一个执行单元获取锁并进入临界区，其他的执行单元都是在门口不停的死等；
执行时间短，由于spinlock死等这种特性，如果临界区执行时间太长，那么不断的在临界区门口“死等”的那些执行单元会浪费CPU；
可以在中断上下文中执行，由于不休眠，因此spinlock可以在中断上下文中适用；

1.3 自旋锁禁止内核抢占

自旋锁禁止内核抢占这是为什么呢？

如果内核可抢占, 单CPU：process1通过系统调用进入内核态，如果其需要访问临界区，则在进入临界区前获得锁，上锁，然后进入临界区。如果process1在内核态执行临界区代码的过程中发生了一个外部中断，当中断处理函数返回时，因为内核的可抢占性，此时将会出现一个调度点，如果CPU的运行队列中出现了一个比当前被中断进程process1优先级更高的进程process2，那么被中断的进程将会被换出处理器，即便此时它正运行于内核态；如果process2也通过系统调用进入内核态，且要访问相同的临界区，由于拿不到锁，则会一致等待，形成死锁(由于process2优先级高，拥有锁的Process1永远没有机会再运行从而释放锁）；这也是和信号量不同的地方，信号量获取不不到锁，会将当前进程睡眠，并加入信号量的等待队列，然后主动执行CPU调度；
如果内核可抢占，多CPU：CPU1上的process1通过系统调用进入内核态，如果其需要访问临界区，则在进入临界区前获得锁，上锁，然后进入临界区；如果process1在内核态执行临界区代码的过程中发生了一个外部中断，当中断处理函数返回时，因为内核的可抢占性，此时将会出现一个调度点，如果CPU1的运行队列中出现了一个比当前被中断进程process1优先级更高的进程process2，那么被中断的进程process1将会被换出处理器，即便此时它正运行于内核态；如果CPU2上的process3也通过系统调用进入内核态，且要访问相同的临界区，也一样形成死锁；

为了防止系统进入死锁状态，需要在真正上锁前，调用preempt_disable来关闭内核抢占。

1.4 自旋锁的使用

定义自旋锁：

spinlock_t lock;

初始化自旋锁：

spin_lock_init(&lock);

获得自旋锁：

spin_lock(&lock);

该宏用于获得自旋锁lock，如果能够立即获得锁，它就马上返回，否则，它将自旋在那里，直到该自旋锁的保持者释放；

spin_trylock(&lock)

该宏尝试获得自旋锁lock，如果能立即获得锁，它获得锁并返回非0值，否则返回0，实际上不再"在原地打转"；

释放自旋锁：

spin_unlock(&lock);

该函数释放自旋锁lock，它与spin_trylock或spin_lock配对使用。

1.5 中断情况下自旋锁的使用

尽管用了自旋锁可以保证临界区不受别的CPU和本CPU内的内核抢占打扰，但是得到锁的代码路径在执行临界区的时候，还可能受到中断和底半部的影响。为了防止这种影响，所以与中断屏蔽联系使用。

spin_lock /spin_unlock是自旋锁机制的基础，它们和：

关中断local_irq_disable/开中断local_irq_enable
关底半部local_bh_disable/开底半部local_bh_enable
关中断并保存状态字local_irq_save/开中断并恢复状态字local_irq_restore

结合就形成了整套自旋锁机制，关系如下：

spin_lock_irq() = spin_lock() + local_irq_disable()
spin_unlock_irq() = spin_unlock() + local_irq_enable()
spin_lock_irqsave() = spin_lock() + local_irq_save()
spin_unlock_irqrestore() = spin_unlock() + local_irq_restore()
spin_lock_bh() = spin_lock() + local_bh_disable()
spin_unlock_bh() = spin_unlock() + local_bh_enable()

在多核编程的时候，如果进程和中断可能访问同一片临界资源，我们一般需要在进程上下文中调用spin_lock_irqsave /spin_unlock_irqrestore，在中断上下文中调用spin_lock/spin_unlock。

例如，在CPU0上，无论是进程上下文，还是中断上下文获得了自旋锁，此后，如果CPU1无论是进程上下文，还是中断上下文，想获得同一自旋锁，都必须忙等待，这避免一切核间并发的可能性。同时，由于每个核的进程上下文持有锁的时候用的是spin_lock_irqsave，所以该核上的中断是不可能进入的，这避免了核内并发的可能性。

1.6

二、自旋锁的源码实现

2.1 spinlock_t结构头

spinlonk_t结构体定义位于include/linux/spinlock_types.h文件中:

typedef struct spinlock {
        union {
                struct raw_spinlock rlock;
        };
} spinlock_t;

在该文件，定位到struct raw_spinlock结构体：

typedef struct raw_spinlock {
        arch_spinlock_t raw_lock;
} raw_spinlock_t;

最后定位到arch_spinlock_t，该函数也是和硬件体系相关的函数，位于arch/arm/include/asm/spinlock_types.h:

typedef struct {
        union {
                u32 slock;
                struct __raw_tickets {
#ifdef __ARMEB__                        // 大端 高字节保存在低位
                        u16 next;
                        u16 owner;
#else
                        u16 owner;
                        u16 next;
#endif
                } tickets;
        };
} arch_spinlock_t;

owner表示持有这个数字的进程可以获取自旋锁；

next表示如果后续再有进程请求获取这个自旋锁，就给它分配这个数字；

2.2 spin_lock_init

宏spin_lock_init位于include/linux/spinlock.h文件中：

#define spin_lock_init(_lock)                           \
do {                                                    \
        spinlock_check(_lock);                          \
        raw_spin_lock_init(&(_lock)->rlock);            \
} while (0)

在当前文件定位到宏raw_spin_lock_init：

# define raw_spin_lock_init(lock)                               \
        do { *(lock) = __RAW_SPIN_LOCK_UNLOCKED(lock); } while (0)
#endif

再次定位到宏__RAW_SPIN_LOCK_UNLOCKED，该宏位于include/linux/spinlock_types.h：

#define __RAW_SPIN_LOCK_INITIALIZER(lockname)   \
        {                                       \
        .raw_lock = __ARCH_SPIN_LOCK_UNLOCKED,  \
        SPIN_DEBUG_INIT(lockname)               \
        SPIN_DEP_MAP_INIT(lockname) }

#define __RAW_SPIN_LOCK_UNLOCKED(lockname)      \
        (raw_spinlock_t) __RAW_SPIN_LOCK_INITIALIZER(lockname)

这里使用__ARCH_SPIN_LOCK_UNLOCKED初始化结构体成员raw_lock，该宏位于arch/arm/include/asm/spinlock_types.h:

#define __ARCH_SPIN_LOCK_UNLOCKED       { { 0 } }

这样owner、next都被初始化为0。

2.3 spin_lock

我们再来看一下获取自旋锁宏spin_lock，位于include/linux/spinlock.h：

static __always_inline void spin_lock(spinlock_t *lock)
{
        raw_spin_lock(&lock->rlock);
}

定位到当前文件宏raw_spin_lock：

#define raw_spin_lock(lock)     _raw_spin_lock(lock)

_raw_spin_lock有两个实现：

位于include/linux/spinlock_api_up.h 单核CPU
位于kernel/locking/spinlock.c 多核CPU

先介绍include/linux/spinlock_api_up.h中的实现:

#define _raw_spin_lock(lock)                    __LOCK(lock)
#define ___LOCK(lock) \
  do { __acquire(lock); (void)(lock); } while (0)
#define __LOCK(lock) \
  do { preempt_disable(); ___LOCK(lock); } while (0)

这里___LOCK函数啥也没做，所以我们重点关注preempt_disable，这个函数是会禁止内核抢占。

然后再来看kernel/locking/spinlock.c中的实现：

void __lockfunc _raw_spin_lock(raw_spinlock_t *lock)
{
        __raw_spin_lock(lock);
}

__raw_spin_lock定义在include/linux/spinlock_api_smp.h中：

static inline void __raw_spin_lock(raw_spinlock_t *lock)
{
        preempt_disable();
        spin_acquire(&lock->dep_map, 0, 0, _RET_IP_);
        LOCK_CONTENDED(lock, do_raw_spin_trylock, do_raw_spin_lock);
}

首先禁止内核抢占，然后执行spin_acquire，该函数位于include/linux/lockdep.h:

#define lock_acquire_exclusive(l, s, t, n, i)           lock_acquire(l, s, t, 0, 1, n, i)
#define spin_acquire(l, s, t, i)                lock_acquire_exclusive(l, s, t, NULL, i)
# define lock_acquire(l, s, t, r, c, n, i) do { } while (0)

可以看到这个函数啥也没做，我们最来到LOCK_CONTENDED，也是位于include/linux/lockdep.h:

#define LOCK_CONTENDED(_lock, try, lock)                        \
do {                                                            \
        if (!try(_lock)) {                                      \
                lock_contended(&(_lock)->dep_map, _RET_IP_);    \
                lock(_lock);                                    \
        }                                                       \
        lock_acquired(&(_lock)->dep_map, _RET_IP_);                     \
} while (0)

第三个参数为do_raw_spin_lock，位于kernel/locking/spinlock_debug.c：

/*
 * We are now relying on the NMI watchdog to detect lockup instead of doing
 * the detection here with an unfair lock which can cause problem of its own.
 */
void do_raw_spin_lock(raw_spinlock_t *lock)
{
        debug_spin_lock_before(lock);
        arch_spin_lock(&lock->raw_lock);
        mmiowb_spin_lock();
        debug_spin_lock_after(lock);
}

定位到arm体系架构代码，arch/arm/include/asm/spinlock.h：

/*
 * ARMv6 ticket-based spin-locking.
 *
 * A memory barrier is required after we get a lock, and before we
 * release it, because V6 CPUs are assumed to have weakly ordered
 * memory.
 */

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]\n"
"       add     %1, %0, %4\n"
"       strex   %2, %1, [%3]\n"
"       teq     %2, #0\n"
"       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 = READ_ONCE(lock->tickets.owner);
        }

        smp_mb();
}

这里我们就不具体分析这个汇编代码了，这里汇编代码本质上还是利用CPU的独占访问指令实现对slock值的修改，大致介绍一下：

这里我们仍然以R0表示%0，R1表示%1，R2表示%2，R3表示%3（指向slock的地址）；
独占访问指令，取出slock中的值到R0;
R1 = R0 + (1 << TICKET_SHIFT)，TICKET_SHIFT值为16;
独占访问指令，执行成功，[R3]=R1，R2=0；
测试R2=0?
如果R2=0，执行完成，否则继续循环执行；

执行成功之后等价于执行如下指令：

lockval=lock->slock;   // 保存旧值
lock->slock += 1<<16;  // 修改后的新值 slock是一个union，由next和owner组成，等价于next++;
newlock = lock->slock;

然后再来看一下C代码：

判断next是否和ower相等，如果不相等，则进程一直等待；
如果相等，获取到锁；

举个例子假设有三个进程执行这段代码，lock->slock初始值为0：

指令	进程1	进程2	进程3	影响
1		ldrex R0, [R3]		设置独占标记 R0=[R3 ]
2		add R0, R0, R4		R0=R0+(1<<16)
3	ldrex R0, [R3]			设置独占标记 R0=[R3 ]
4			ldrex R0, [R3]	设置独占标记 R0=[R3 ]
5			add R0, R0, R4	R0=R0+(1<<16)
6	add R0, R0, R4			R0=R0+(1<<16)
7	strex R2, R1, [R3]			执行成功，R1写回[R3]，清除独占标记
8		strex R1, R0, [R3]		没有独占标记，执行失败
9	teq R2, #0			相等 R2=0
10			strex R2, R1, [R3]	没有独占标记，执行失败
11		teq R2, #0		不相等 R2=1
12		b 1b		跳转
13			teq R2, #0	不相等 R2=1
14			b 1b	跳转
15	汇编执行完毕			lockval=R1=0 当前保存进程值 lock->slock=R3=1<<16 lock->tickets.next=1
16	lockval.tickets.next != lockval.tickets.owner			条件不满足，获取spinlock，进入临界区
17		再次经历上面步骤，不过此时lock->slock初始值为1<<16
18		....
19			...
20		汇编执行完毕		lockval=R1=1<<16 当前保存进程值 lock->slock=R3=2<<16 lock->tickets.next=2
21			汇编执行完毕	lockval=R1=2<<16 当前保存进程值 lock->slock=R3=3<<16 lock->tickets.next=3
22		lockval.tickets.next != lockval.tickets.owner		lockval=1<<16 ，条件成立开始死等
23		lockval.tickets.owner = READ_ONCE(lock->tickets.owner)		lockval=1<<16
24			lockval.tickets.next != lockval.tickets.owner	lockval=2<<16 ，条件成立开始死等
25			lockval.tickets.owner = READ_ONCE(lock->tickets.owner)	lockval不变
26	spin_unlock			lock->slock=3<<16+1 lock->tickets.owner=1
27		lockval.tickets.next != lockval.tickets.owner		成立开始死等
28		lockval.tickets.owner = READ_ONCE(lock->tickets.owner)		lockval=1<<16 +1
		lockval.tickets.next != lockval.tickets.owner		条件不满足，获取spinlock，进入临界区

代码大致流程如下，三个进程同时修改lock->slock(联合体，或者说lock->tickets)，这个变量是三个进程共享的：

刚开始lock->tickets.owner=0，lock->tickets.next=0；
进程1获得独占锁，成功修改lock->slock，此时当前进程局部变量locakval.tickets.owner=0,locakval.tickets.next=0，lock->tickets.owner=0，lock->tickets.next=1；
进程2获取独占锁，成功修改lock-slock，此时当前进程局部变量locakval.tickets.owner=0,locakval.tickets.next=1，lock->tickets.owner=0，lock->tickets.next=2；
进程2满足条件lockval.tickets.next != lockval.tickets.owner，开始死等；
进程3获取独占锁，成功修改lock-slock，此时当前进程局部变量locakval.tickets.owner=0,locakval.tickets.next=2，lock->tickets.owner=0，lock->tickets.next=3；
进程3满足条件lockval.tickets.next != lockval.tickets.owner，开始死等；
此时第一个进程释放spinlock，则执行lock->tickets.owner++，lock->tickets.owner=2;
虽然此时第二个进程和第三个进程都在等待spinlock，但是因为第二个进程的修改了lockval.tickets.owner，满足了lockval.tickets.next != lockval.tickets.owner，所以第二个进程可以获取到spinlock，第三个进程则继续等待；

这样保证了spinlock的唤醒机制是先到先唤醒，后到后唤醒，保证了公平性；

看完这个我们再来通俗解释一下自旋锁的实现：

通过strex和ldrex指令实现对独占资源lock->slock值的互斥访问，该值初始为0；
当第一个进程获取到修改lock->slock权限时，进程使用局部变量locklval保存修改前的值，locklval=0，然后修改lock->slock.next++;
进程lockval值next部分可以理解为分配给这个进程的数字为0，lock->slock.owner部分可以理解为持有数字0的进程可以获得锁；两者相等，因此第一个进程直接获取spinlock锁；
当第二个进程获取到修改lock->slock权限时，进程使用局部变量locklval保存修改前的值，locklval=1<<16，然后修改lock->slock.next++;
进程lockval值next部分可以理解为分配给这个进程的数字为1，lock->slock.owner部分可以理解为持有数字0的进程可以获得锁；两者不相等，因此第二个进程死等；
当第一个进程释放spinlock锁，修改lock->slock.owner++;此时可以理解为持有数字1进程可以获得锁；
此时第二个进程获取spinlick锁；

三、自旋锁示例程序

3.1 使用注意事项

自旋锁实际上是忙等锁，当锁不可用时，CPU一直循环执行“测试并设置”该锁直到可用而取得该锁，CPU在等待自旋锁时不做任何有用的工作，仅仅是等待。因此，只有在占用锁的时间极短的情况下，使用自旋锁才是合理的。当临界区很大，或有共享设备的时候，需要较长时间占用锁，使用自旋锁会降低系统的性能；
自旋锁可能导致系统死锁。引发这个问题最常见的情况是递归使用一个自旋锁，即如果一个已经拥有某个自旋锁的CPU想第二次获得这个自旋锁，则该CPU将死锁；
在自旋锁锁定期间不能调用可能引起进程调度的函数。如果进程获得自旋锁之后再阻塞，如调用copy_from_user 、 copy_to_user、kmalloc 和msleep等函数，则可能导致内核的崩溃；
在单核情况下编程的时候，也应该认为自己的CPU是多核的，驱动特别强调跨平台的概念。比如，在单CPU的情况下，若中断和进程可能访问同一临界区，进程里调用spin_lock_irqsave是安全的，在中断里其实不调用spin_lock也没有问题，因为spin_lock_irqsave可以保证这个CPU的中断服务程序不可能执行。但是，若CPU变多核，spin_lock_irqsave不能屏蔽另外一个核的中断，所以另外一个核就可能造成并发问题。因此，无论如何，我们在中断服务程序里也应该调用spin_lock;

3.2 示例程序

修改信号量示例里面的驱动程序：

在驱动程序中首先定义并初始化一个自旋锁：
增加驱动程序里的open函数和close函数里对自旋锁的操作：

#include <linux/module.h>
#include <linux/cdev.h>
#include <linux/fs.h>

#define OK   (0)
#define ERROR  (-1)

/* 自旋锁 */
static spinlock_t lock;
static int count = 0;

int hello_open(struct inode *p, struct file *f)
{
    /* 获取自旋锁 */
    spin_lock(&lock);
    if(count >= 1){
        spin_unlock(&lock);        
        printk("device busy,hello_open failed");
        return ERROR;
    }
    count++;
    spin_unlock(&lock);        
    printk("hello_open\n");
    return 0;
}

ssize_t hello_write(struct file *f, const char __user *u, size_t s, loff_t *l)
{
    printk("hello_write\n");
    return 0;
}

ssize_t hello_read(struct file *f, char __user *u, size_t s, loff_t *l)
{
    printk("hello_read\n");
    return 0;
}

int hello_close(struct inode *inode, struct file *file)
{
     /* 获取自旋锁 */
    spin_lock(&lock);
    count--;
    spin_unlock(&lock);        
    return 0;
}

struct file_operations hello_fops = {
    .owner   =   THIS_MODULE,
    .open    =   hello_open,
    .read    =   hello_read,
    .write   =   hello_write,
    .release =   hello_close,
};

dev_t devid;                      // 起始设备编号
struct cdev hello_cdev;          // 保存操作结构体的字符设备 
struct class *hello_cls;

int hello_init(void)
{
    
    /* 动态分配字符设备: (major,0) */
    if(OK == alloc_chrdev_region(&devid, 0, 1,"hello")){   // ls /proc/devices看到的名字
        printk("register_chrdev_region ok\n");
    }else {
        printk("register_chrdev_region error\n");
        return ERROR;
    }
    
     cdev_init(&hello_cdev, &hello_fops);
     cdev_add(&hello_cdev, devid, 1);


    /* 创建类,它会在sys目录下创建/sys/class/hello这个类  */
     hello_cls = class_create(THIS_MODULE, "hello");
     if(IS_ERR(hello_cls)){
         printk("can't create class\n");
         return ERROR;
     }
    /* 在/sys/class/hello下创建hellos设备，然后mdev通过这个自动创建/dev/hello这个设备节点 */
     device_create(hello_cls, NULL, devid, NULL, "hello"); 

     /* 初始化自旋锁 */
     spin_lock_init(&lock);
     return 0;
}

void __exit hello_exit(void)
{
    printk("hello driver exit\n");
    /* 注销类、以及类设备 /sys/class/hello会被移除*/
    device_destroy(hello_cls, devid);
    class_destroy(hello_cls);

    cdev_del(&hello_cdev);
    unregister_chrdev_region(devid, 1);
    return;
}


module_init(hello_init);
module_exit(hello_exit);
MODULE_LICENSE("GPL");