背景
由于在多处理器环境中某些资源的有限性,有时需要互斥访问(mutual exclusion),这时候就需要引入锁的概念,只有获取了锁的任务才能够对资源进行访问,由于多线程的核心是CPU的时间分片,所以同一时刻只能有一个任务获取到锁。
内核当发生访问资源冲突的时候,通常有两种处理方式:
一个是原地等待
一个是挂起当前进程,调度其他进程执行(睡眠)
自旋锁
Spinlock 是内核中提供的一种比较常见的锁机制,自旋锁是“原地等待”的方式解决资源冲突的。即,一个线程获取了一个自旋锁后,另外一个线程期望获取该自旋锁,获取不到,只能够原地“打转”(忙等待)。
由于自旋锁的这个忙等待的特性,注定了它使用场景上的限制 —— 自旋锁不应该被长时间的持有(消耗 CPU 资源)。
自旋锁的优点
自旋锁不会使线程状态发生切换,一直处于用户态,即线程一直都是active的;不会使线程进入阻塞状态,减少了不必要的上下文切换,执行速度快。
非自旋锁在获取不到锁的时候会进入阻塞状态,从而进入内核态,当获取到锁的时候需要从内核态恢复,需要线程上下文切换。(线程被阻塞后便进入内核(Linux)调度状态,这个会导致系统在用户态与内核态之间来回切换,严重影响锁的性能)。
自旋锁的使用
在linux kernel的实现中,经常会遇到这样的场景:共享数据被中断上下文和进程上下文访问,该如何保护呢?
如果只有进程上下文的访问,那么可以考虑使用semaphore或者mutex的锁机制,但是现在中断上下文也掺和进来,那些可以导致睡眠的lock就不能使用了,这时候,可以考虑使用spin lock。
在中断上下文,是不允许睡眠的,所以,这里需要的是一个不会导致睡眠的锁——spinlock。
换言之,中断上下文要用锁,首选 spinlock。
使用自旋锁,有两种方式定义一个锁:
动态的:
spinlock_t lock;spin_lock_init (&lock);
静态的:
DEFINE_SPINLOCK(lock);
使用步骤
spinlock的使用很简单:
我们要访问临界资源需要首先申请自旋锁;
获取不到锁就自旋,如果能获得锁就进入临界区;
当自旋锁释放后,自旋在这个锁的任务即可获得锁并进入临界区,退出临界区的任务必须释放自旋锁。
使用实例
static spinlock_t lock;static int flage = 1;spin_lock_init(&lock);static int hello_open (struct inode *inode, struct file *filep){ spin_lock(&lock); if(flage !=1) { spin_unlock(&lock); return -EBUSY; } flage =0; spin_unlock(&lock); return 0;}static int hello_release (struct inode *inode, struct file *filep){ flage = 1; return 0;}
补充
中断上下文不能睡眠的原因是:
中断处理的时候,不应该发生进程切换,因为在中断context中,唯一能打断当前中断handler的只有更高优先级的中断,它不会被进程打断,如果在 中断context中休眠,则没有办法唤醒它,因为所有的wake_up_xxx都是针对某个进程而言的,而在中断context中,没有进程的概念,没 有一个task_struct(这点对于softirq和tasklet一样),因此真的休眠了,比如调用了会导致block的例程,内核几乎肯定会死。
schedule()在切换进程时,保存当前的进程上下文(CPU寄存器的值、进程的状态以及堆栈中的内容),以便以后恢复此进程运行。中断发生后,内核会先保存当前被中断的进程上下文(在调用中断处理程序后恢复);
但在中断处理程序里,CPU寄存器的值肯定已经变化了吧(最重要的程序计数器PC、堆栈SP等),如果此时因为睡眠或阻塞操作调用了schedule(),则保存的进程上下文就不是当前的进程context了。所以不可以在中断处理程序中调用schedule()。
内核中schedule()函数本身在进来的时候判断是否处于中断上下文:
if(unlikely(in_interrupt())) BUG();
因此,强行调用schedule()的结果就是内核BUG。
中断handler会使用被中断的进程内核堆栈,但不会对它有任何影响,因为handler使用完后会完全清除它使用的那部分堆栈,恢复被中断前的原貌。
处于中断context时候,内核是不可抢占的。因此,如果休眠,则内核一定挂起。
自旋锁的死锁
自旋锁不可递归,自己等待自己已经获取的锁,会导致死锁。
自旋锁可以在中断上下文中使用,但是试想一个场景:一个线程获取了一个锁,但是被中断处理程序打断,中断处理程序也获取了这个锁(但是之前已经被锁住了,无法获取到,只能自旋),中断无法退出,导致线程中后面释放锁的代码无法被执行,导致死锁。(如果确认中断中不会访问和线程中同一个锁,其实无所谓)。
一、考虑下面的场景(内核抢占场景):
(1)进程A在某个系统调用过程中访问了共享资源 R
(2)进程B在某个系统调用过程中也访问了共享资源 R
会不会造成冲突呢?
假设在A访问共享资源R的过程中发生了中断,中断唤醒了沉睡中的,优先级更高的B,在中断返回现场的时候,发生进程切换,B启动执行,并通过系统调用访问了R,如果没有锁保护,则会出现两个thread进入临界区,导致程序执行不正确。OK,我们加上spin lock看看如何:A在进入临界区之前获取了spin lock,同样的,在A访问共享资源R的过程中发生了中断,中断唤醒了沉睡中的,优先级更高的B,B在访问临界区之前仍然会试图获取spin lock,这时候由于A进程持有spin lock而导致B进程进入了永久的spin……怎么破?linux的kernel很简单,在A进程获取spin lock的时候,禁止本CPU上的抢占(上面的永久spin的场合仅仅在本CPU的进程抢占本CPU的当前进程这样的场景中发生)。如果A和B运行在不同的CPU上,那么情况会简单一些:A进程虽然持有spin lock而导致B进程进入spin状态,不过由于运行在不同的CPU上,A进程会持续执行并会很快释放spin lock,解除B进程的spin状态。
二、再考虑下面的场景(中断上下文场景):
运行在CPU0上的进程A在某个系统调用过程中访问了共享资源 R
运行在CPU1上的进程B在某个系统调用过程中也访问了共享资源 R
外设P的中断handler中也会访问共享资源 R
在这样的场景下,使用spin lock可以保护访问共享资源R的临界区吗?
我们假设CPU0上的进程A持有spin lock进入临界区,这时候,外设P发生了中断事件,并且调度到了CPU1上执行,看起来没有什么问题,执行在CPU1上的handler会稍微等待一会CPU0上的进程A,等它立刻临界区就会释放spin lock的,但是,如果外设P的中断事件被调度到了CPU0上执行会怎么样?CPU0上的进程A在持有spin lock的状态下被中断上下文抢占,而抢占它的CPU0上的handler在进入临界区之前仍然会试图获取spin lock,悲剧发生了,CPU0上的P外设的中断handler永远的进入spin状态,这时候,CPU1上的进程B也不可避免在试图持有spin lock的时候失败而导致进入spin状态。为了解决这样的问题,linux kernel采用了这样的办法:如果涉及到中断上下文的访问,spin lock需要和禁止本 CPU 上的中断联合使用。
三、再考虑下面的场景(底半部场景)
linux kernel中提供了丰富的bottom half的机制,虽然同属中断上下文,不过还是稍有不同。我们可以把上面的场景简单修改一下:外设P不是中断handler中访问共享资源R,而是在的bottom half中访问。使用spin lock+禁止本地中断当然是可以达到保护共享资源的效果,但是使用牛刀来杀鸡似乎有点小题大做,这时候disable bottom half就可以了。
四、中断上下文之间的竞争
同一种中断handler之间在uni core和multi core上都不会并行执行,这是linux kernel的特性。如果不同中断handler需要使用spin lock保护共享资源,对于新的内核(不区分fast handler和slow handler),所有handler都是关闭中断的,因此使用spin lock不需要关闭中断的配合。bottom half又分成softirq和tasklet,同一种softirq会在不同的CPU上并发执行,因此如果某个驱动中的softirq的handler中会访问某个全局变量,对该全局变量是需要使用spin lock保护的,不用配合disable CPU中断或者bottom half。tasklet更简单,因为同一种tasklet不会多个CPU上并发。
自旋锁的实现原理
数据结构
首先定义一个 spinlock_t 的数据类型,其本质上是一个整数值(对该数值的操作需要保证原子性),该数值表示spin lock是否可用。初始化的时候被设定为1。当thread想要持有锁的时候调用spin_lock函数,该函数将spin lock那个整数值减去1,然后进行判断,如果等于0,表示可以获取spin lock,如果是负数,则说明其他thread的持有该锁,本thread需要spin。
内核中的spinlock_t的数据类型定义如下:
typedef struct spinlock { struct raw_spinlock rlock; } spinlock_t;typedef struct raw_spinlock { arch_spinlock_t raw_lock;} raw_spinlock_t;
通用(适用于各种arch)的spin lock使用spinlock_t这样的type name,各种arch定义自己的struct raw_spinlock。听起来不错的主意和命名方式,直到linux realtime tree(PREEMPT_RT)提出对spinlock的挑战。
spin lock的命名规范定义如下:
spinlock,在rt linux(配置了PREEMPT_RT)的时候可能会被抢占(实际底层可能是使用支持PI(优先级翻转)的mutext)。
raw_spinlock,即便是配置了PREEMPT_RT也要顽强的spin
arch_spinlock,spin lock是和architecture相关的,
加锁
同样的,这里也只是选择一个典型的API来分析,其他的大家可以自行学习。我们选择的是 arch_spin_lock,其ARM32的代码如下:
static inline void arch_spin_lock(arch_spinlock_t *lock){ unsigned long tmp; u32 newval; arch_spinlock_t lockval; prefetchw(&lock-》slock);---------(0) __asm__ __volatile__(“1: ldrex %0, [%3]\n”---------(1)“ add %1, %0, %4\n” ----------(2)“ strex %2, %1,[%3]\n”---------(3)“ teq %2, #0\n”-------------(4)“ bne 1b” : “=&r” (lockval), “=&r” (newval), “=&r” (tmp) : “r” (&lock-》slock), “I” (1 《《 TICKET_SHIFT) : “cc”); while (lockval.tickets.next != lockval.tickets.owner) {----(5) wfe();------------(6) lockval.tickets.owner = ACCESS_ONCE(lock-》tickets.owner);----(7) } smp_mb();-----------(8)}
(0)和preloading cache相关的操作,主要是为了性能考虑(1)lockval = lock-》slock (如果lock-》slock没有被其他处理器独占,则标记当前执行处理器对lock-》slock地址的独占访问;否则不影响)(2)newval = lockval + (1 《《 TICKET_SHIFT)(3)strex tmp, newval, [&lock-》slock] (如果当前执行处理器没有独占lock-》slock地址的访问,不进行存储,返回1给temp;如果当前处理器已经独占lock-》slock内存访问,则对内存进行写,返回0给temp,清除独占标记) lock-》tickets.next = lock-》tickets.next + 1(4)检查是否写入成功 lockval.tickets.next(5)初始化时lock-》tickets.owner、lock-》tickets.next都为0,假设第一次执行arch_spin_lock,lockval = *lock,lock-》tickets.next++,lockval.tickets.next 等于 lockval.tickets.owner,获取到自旋锁;自旋锁未释放,第二次执行的时候,lock-》tickets.owner = 0, lock-》tickets.next = 1,拷贝到lockval后,lockval.tickets.next != lockval.tickets.owner,会执行wfe等待被自旋锁释放被唤醒,自旋锁释放时会执行 lock-》tickets.owner++,lockval.tickets.owner重新赋值(6)暂时中断挂起执行。如果当前spin lock的状态是locked,那么调用wfe进入等待状态。更具体的细节请参考ARM WFI和WFE指令中的描述。(7)其他的CPU唤醒了本cpu的执行,说明owner发生了变化,该新的own赋给lockval,然后继续判断spin lock的状态,也就是回到step 5。(8)memory barrier的操作,具体可以参考memory barrier中的描述。
释放锁
static inline void arch_spin_unlock(arch_spinlock_t *lock){ smp_mb(); lock-》tickets.owner++; ---------------------- (0) dsb_sev(); ---------------------------------- (1)}
(0)lock-》tickets.owner增加1,下一个被唤醒的处理器会检查该值是否与自己的lockval.tickets.next相等,lock-》tickets.owner代表可以获取的自旋锁的处理器,lock-》tickets.next你一个可以获取的自旋锁的owner;处理器获取自旋锁时,会先读取lock-》tickets.next用于与lock-》tickets.owner比较并且对lock-》tickets.next加1,下一个处理器获取到的lock-》tickets.next就与当前处理器不一致了,两个处理器都与lock-》tickets.owner比较,肯定只有一个处理器会相等,自旋锁释放时时对lock-》tickets.owner加1计算,因此,先申请自旋锁多处理器lock-》tickets.next值更新,自然先获取到自旋锁
(1)执行sev指令,唤醒wfe等待的处理器
自旋锁导致死锁实例
死锁的2种情况
1)拥有自旋锁的进程A在内核态阻塞了,内核调度B进程,碰巧B进程也要获得自旋锁,此时B只能自旋转。而此时抢占已经关闭,不会调度A进程了,B永远自旋,产生死锁。
2)进程A拥有自旋锁,中断到来,CPU执行中断函数,中断处理函数,中断处理函数需要获得自旋锁,访问共享资源,此时无法获得锁,只能自旋,产生死锁。
如何避免死锁
如果中断处理函数中也要获得自旋锁,那么驱动程序需要在拥有自旋锁时禁止中断;
自旋锁必须在可能的最短时间内拥有;
避免某个获得锁的函数调用其他同样试图获取这个锁的函数,否则代码就会死锁;不论是信号量还是自旋锁,都不允许锁拥有者第二次获得这个锁,如果试图这么做,系统将挂起;
锁的顺序规则 按同样的顺序获得锁; 如果必须获得一个局部锁和一个属于内核更中心位置的锁,则应该首先获取自己的局部锁 ; 如果我们拥有信号量和自旋锁的组合,则必须首先获得信号量;在拥有自旋锁时调用down(可导致休眠)是个严重的错误的。
死锁举例
因为自旋锁持有时间非常短,没有直观的现象,下面举一个会导致死锁的实例。
运行条件
虚拟机:vmware
OS :Ubuntu 14
配置 :将虚拟机的处理个数设置为1,否则不会死锁
原理
针对单CPU,拥有自旋锁的任务不应该调度会引起休眠的函数,否则会导致死锁。
步骤:
进程A在open()字符设备后,对应的内核函数会申请自旋锁,此时自旋锁空闲,申请到自旋锁,进程A随即进入执行sleep()函数进入休眠;
在进程A 处于sleep期间,自旋锁一直属于进程A所有;
运行进程B,进程B执行open函数,对应的内核函数也会申请自旋锁,此时自旋锁归进程A所有,所以进程B进入自旋状态;
因为此时抢占已经关闭,系统死锁。
驱动代码如下:
#include 《linux/init.h》#include 《linux/module.h》#include 《linux/kdev_t.h》#include 《linux/fs.h》#include 《linux/cdev.h》#include 《linux/device.h》#include 《linux/spinlock.h》static int major = 250;static int minor = 0;static dev_t devno;static struct cdev cdev;static struct class *cls;static struct device *test_device;static spinlock_t lock;static int flage = 1;#define DEAD 1static int hello_open (struct inode *inode, struct file *filep){ spin_lock(&lock); if(flage !=1) { spin_unlock(&lock); return -EBUSY; } flage =0; #if DEAD #elif spin_unlock(&lock); #endif return 0;}static int hello_release (struct inode *inode, struct file *filep){ flage = 1; #if DEAD spin_unlock(&lock); #endif return 0;}static struct file_operations hello_ops ={ .open = hello_open, .release = hello_release,};static int hello_init(void){ int result; int error; printk(“hello_init \n”); result = register_chrdev( major, “hello”, &hello_ops); if(result 《 0) { printk(“register_chrdev fail \n”); return result; } devno = MKDEV(major,minor); cls = class_create(THIS_MODULE,“helloclass”); if(IS_ERR(cls)) { unregister_chrdev(major,“hello”); return result; } test_device = device_create(cls,NULL,devno,NULL,“test”); if(IS_ERR(test_device )) { class_destroy(cls); unregister_chrdev(major,“hello”); return result; } spin_lock_init(&lock); return 0;}static void hello_exit(void){ printk(“hello_exit \n”); device_destroy(cls,devno); class_destroy(cls); unregister_chrdev(major,“hello”); return;}module_init(hello_init);module_exit(hello_exit);MODULE_LICENSE(“GPL”);
测试程序如下:
#include 《stdio.h》#include 《sys/types.h》#include 《sys/stat.h》#include 《fcntl.h》main(){ int fd; fd = open(“/dev/test”,O_RDWR); if(fd《0) { perror(“open fail \n”); return; } sleep(20); close(fd); printf(“open ok \n ”);}
测试步骤:
编译加载内核
makeinsmod hello.ko
运行进程A
gcc test.c -o a./a
打开一个新的终端,运行进程B
gcc test.c -o b./b
注意,一定要在进程A没有退出的时候运行进程B。
-
Linux
+关注
关注
87文章
11291浏览量
209309 -
锁存
+关注
关注
0文章
21浏览量
20630
发布评论请先 登录
相关推荐
评论