内核同步

Operation System

内核同步

对于内核，其实有一个很形象的理解：我们可以把内核理解成一个服务器，它为自身和用户提供各种服务。因此它必须要保证每项服务在处理时，不会互相造成影响，也就是解决“并发”的问题。自身的请求，也即中断；客户的请求，也即用户态的系统调用或异常。内核的同步，就是对内核中的任务进行调度，使它们按照正确的方式运行。

内核抢占

这里，“内核抢占”指的是进程A在内核态运行时，被具有更高优先级的进程B取代，也就是发生了进程上下文的切换。而我们知道，中断上下文是不包括进程信息的，不能被调度。所以只要在中断上下文中，就不能进行“进程切换”。因此硬中断和软中断在执行时都不允许内核抢占；只有在内核执行异常处理程序（尤其是系统调用），并且内核抢占没有被显示禁用时，才能进行内核抢占。CPU必须打开本地中断，才能完成内核抢占。

从另一个角度来说，CPU在任何情况下，都处于三种上下文情况之一：

  • 运行在用户空间，执行用户进程；
  • 运行在内核空间，处于进程上下文；
  • 运行在内核空间，处于中断上下文。

在关于中断的博文里，我已经写过，中断上下文是不属于任何进程的，它和current没有任何关系。由于没有任何进程背景，在中断上下文中也不能发生睡眠，否则是不能对它进行调度。因此中断上下文中只能用锁进行同步，中断上下文也叫做原子上下文。而异常和系统调用陷入内核时，是出于进程上下文的，因此可以通过current关联相应的任务。所以在进程上下文中，可以发生睡眠，也可以使用信号量；当然也可以使用锁。

ps：以上说的是内核抢占的情况；用户抢占指的是另一个概念，指的是内核即将返回用户空间的时候，如果need_resched标志被设置，就会调用schedule()，选择一个更为合适的进程运行。

内核不能被抢占的情况有这些：

  • 内核正在进行中断处理。在linux下，进程不能抢占中断（注意，中断是可以抢占、中止其他中断的），中断历程中不允许进行进程调度（schedule()会进行判断，如果在中断中会报错）。这也包括软中断的Bottom half部分。
  • 当前的代码段持有自旋锁、读写锁，这些锁保证SMP系统CPU并发的正确性，此时不能进行抢占。
  • 内核正在执行调度程序时，不应该进行抢占。
  • 内核正在对每CPU数据进行操作。

除此之外的情况，都可以发生内核抢占。

每CPU变量

把内核变量，声明为每个CPU所独有的，它是数组结构的数组，每个CPU对应数组的一个元素，CPU直接不能访问其他CPU对应的数组元素，只能读写自身的元素，因此也不会出现竞争条件。但这同样存在着限制：必须确定CPU上的数据是各自独立的。

但是每CPU变量不能解决内核抢占的问题，他只能解决多CPU的问题，因此在访问时应当禁用抢占。

原子操作

通过保证操作在芯片上是原子级的，保证“读－修改－写”指令不会引发竞争。任何一个这样的操作，都必须以单个指令执行，并且不能中断，避免其他CPU访问这个单元。除了常见的0或1次对齐内存访问的汇编指令、单处理器下的“读－修改－写”指令、前缀为lock的指令也是原子操作指令。

优化和内存屏障

优化屏障主要是用来保证编译时，汇编语言指令按照原顺序来执行，而不进行重排。例如在linux中，barrier()的本质就是asm volatile("":::"memory")。而内存屏障则是保证原语前后的指令执行顺序，也即在执行原语后的指令时，原语前的指令必须已经执行完了。

自旋锁

自旋锁是一类特别广泛使用的同步技术，如果内核控制路径必须访问共享数据结构，或者访问临界区，那么就需要为自己获取一个自旋锁；只有资源是空闲时，获取才能成功；当它释放了锁之后，其他内核控制路径就可以进入房间了。那么自旋锁的意义是什么？它是多处理器环境下一种特殊的锁；如果执行路径发现自旋锁是锁着的，或反复在周围进行“旋转”，反复执行循环，直到锁被释放（忙等）。

自旋锁保护的临界区通常是禁止内核抢占的，如果在单CPU环境下，自旋锁仅仅能够禁止或启用内核抢占，并不能起到锁的作用。当然，忙等时还是可以被抢占的，只有上锁后才会禁止抢占。

ps：阿里巴巴的面试官问过我一个问题，**自旋锁的本质是什么？**我当时猜测了一下，回答了原子操作，但没有能够进一步地进行解释。这里应该结合源码进行说明。可以看到对xadd就是一个标准的源子加操作。linux内核使用了两种实现。其一是“标签自旋锁”，raw_spin_lock最后会调用：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
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);
}
arch_spin_lock(arch_spinlock_t *lock)
{
	register struct __raw_tickets inc = {.tail = TICKET_LOCK_INC};//这个值是0
	inc = xadd(&lock->tickets, inc);	//xadd是原子加，在多CPU时会上锁
										//获取标签，同时把序号＋1  
	if(likely(inc.head == inc.tail))	//标签到自己了，取锁成功了
		goto out;
	for(;;){							//否则就不断循环，直到轮到自己
		unsigned count = SPIN_THRESHOLD;
		do{
			inc.head = READ_ONCE(lock->tickets.head);
			if(__tickets_equal(inc.head, inc.tail))//判断是否到自己的标签了
			goto clear_slowpath;
			cpu_relax();
		}while(--count);
		__ticket_lock_spinning(lock, inc.tail);
	}
	clear_slowpath:
		__ticket_check_and_clear_slowpath(lock, inc.head);
	cout:
		barrier();
}
arch_spinlock_t的结构如下，实际上就是一个u16数
typedef struct arch_spinlock {
         union {
                 __ticketpair_t head_tail;
                 struct __raw_tickets {
                         __ticket_t head, tail;
                 } tickets;
         };
 } arch_spinlock_t;

另一种是一种更加复杂的实现，被称为“排队自旋锁”。排队自旋锁基于每CPU变量实现，其实现比基于标签对实现更公平。

读－拷贝－更新

用来保护在多数情况下，被多个CPU读的数据结构，而设计的另一种同步技术，其特点是允许多个读和写并发执行，并且不使用锁。那么它如何在共享数据读前提下，实现同步呢？RCU只保护被动态分配，并且通过指针引用的数据结构，并且在RCU临界区内，禁止睡眠。RCU的做法是，在写操作时，拷贝一份原来的副本，在副本上进行修改，并且在修改完成后进行更新，将旧的指针更新为新的指针。

信号量

在linux中，有两种信号量，一种是给内核使用的内核信号量，另一种是给用户态进程使用的IPC信号量。这里我们只讨论内核信号量。其实信号量和自旋锁在“上锁”这一点上是类似的，如果锁关闭了，那么就不允许内核控制路径继续执行；只不过它不会像自旋锁一样，在原地“忙等”，而是将相应的进程挂起；只有资源可用了，进程才能继续运行。也正因为“睡眠”的特性，信号量不能用在中断处理程序和延迟处理函数上，只有允许睡眠的情况下，才能够使用信号量。

内核信号量的定义在semaphore.h当中：

1
2
3
4
5
 
struct semaphore {
	     raw_spinlock_t          lock;		//保护信号量的自旋锁
	     unsigned int            count;
	     struct list_head        wait_list;
	};

很神奇的，这里看到了raw_spinlock_t的影子。这其实是一个由Real-time linux引入的命名问题；这里我们只需要明白：尽可能使用spin_lock；绝对不允许被抢占和休眠的地方，使用raw_spin_lock，否则使用spin_lock，信号量的底层，使用了自旋锁来实现。

信号量的后两个域，count和wait_list分别是现有资源数和等待获取资源的进程序列。对于信号量，内核定义了这些API：

1
2
3
4
5
6
 
void down(struct semaphore *sem);
void up(struct semaphore *sem);
int  down_interruptible(struct semaphore *sem);
int  down_killable(struct semaphore *sem);
int  down_trylock(struct semaphore *sem);
int  down_timeout(struct semaphore *sem, long jiffies);

这里看看down函数：

1
2
3
4
5
6
7
8
9
10
 
void down(struct semaphore *sem)
{
       unsigned long flags;
       raw_spin_lock_irqsave(&sem->lock, flags);
       if (likely(sem->count > 0))
               sem->count--;
       else
               __down(sem);
       raw_spin_unlock_irqrestore(&sem->lock, flags);
}

可以看到，这里自旋锁的作用实际上是保证count不被同时操作；而如果count大于0，则可以减少它的值，表示获取了这个锁，否则会__down_common，这个函数在不发生错误大情况下，会调用这样一段函数：

1
2
3
 
raw_spin_unlock_irq(&sem->lock);
timeout = schedule_timeout(timeout);
raw_spin_lock_irq(&sem->lock);

这个函数是在timer.c代码中定义的。schedule_timeout函数将当前的任务置为休眠到设置的超时为止，这也就是信号量和自旋锁不同之处了，它允许进程的休眠。

而对于up函数来说，释放锁，增加count之后，会马上会检查是否有进程在等待资源：

1
2
3
4
5
6
7
8
 
static noinline void __sched __up(struct semaphore *sem)
{
       struct semaphore_waiter *waiter = list_first_entry(&sem->wait_list,
                                               struct semaphore_waiter, list);
       list_del(&waiter->list);
       waiter->up = true;
       wake_up_process(waiter->task);
}

这样看来，其实信号量和自旋锁最大的不同就只有两个：自旋锁的忙等与信号量的休眠，资源的数量。

互斥量

虽然《深入理解linux内核》这本书中没有写，但是内核中也是有互斥量的；实际上它相当于count ＝ 1的信号量。互斥量的定义为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
 
struct mutex {
       atomic_t                count;
       spinlock_t              wait_lock;
       struct list_head        wait_list;
#if defined(CONFIG_DEBUG_MUTEXES) || defined(CONFIG_MUTEX_SPIN_ON_OWNER)
       struct task_struct      *owner;
#endif
#ifdef CONFIG_MUTEX_SPIN_ON_OWNER
       struct optimistic_spin_queue osq;
#endif
#ifdef CONFIG_DEBUG_MUTEXES
       void                    *magic;
#endif
#ifdef CONFIG_DEBUG_LOCK_ALLOC
       struct lockdep_map      dep_map;
#endif
};

可以看到它同样依赖于自旋锁实现，也包含一个进程的等待队列。我们来看看互斥量的上锁操作：

1
2
3
4
5
6
7
8
9
10
11
12
 
void __sched mutex_lock(struct mutex *lock)
{
       might_sleep();
       __mutex_fastpath_lock(&lock->count, __mutex_lock_slowpath);
       mutex_set_owner(lock);
}
这里__mutex_fastpath_lock最终会调用一段汇编代码：
asm_volatile_goto(LOCK_PREFIX "   decl %0\n"
                             "   jns %l[exit]\n"
                             : : "m" (v->counter)
                             : "memory", "cc"
                             : exit);

也就是原子操作，修改mutex的counter，而mutex中的自旋锁，是为了保护wait_list而存在的，只是起到一个辅助作用，这点和信号量不太一样。

读写自旋锁/顺序锁/信号量

为了增加内核到并发能力，操作系统还设置了读写自旋锁。读写自旋锁允许多个内存控制路径，同时读同一个数据结构，但如果相对这个结构进行写操作，那么它必须首先获取读写自旋锁的写锁，写锁能让当前的路径独占访问这个资源。

顺序锁则是允许读者在读的同时进行写操作，因此写操作永远不会等待，但这样读操作有时候必须重复读多次，直到读到有效的副本为止。
读写信号量则和读写自旋锁类似，只不过它以挂起代替自旋。

禁止本地中断/可延迟函数

在前面提到的原语中，很多在实现的时候，都禁止了了本地的中断，这就保证了当前内核控制路径能够继续执行，例如raw_spin_lock_irqsave和raw_spin_lock_irqrestore。不过禁止本地中断不能阻止其他CPU
访问共享数据，因此通常和自旋锁结合使用。

而可延迟函数同样可以禁止和激活，这是由preempt_count字段中的值决定的。



The link of this page is https://blog.nooa.tech/articles/e41a0c57/ . Welcome to reproduce it!

© 2018.02.08 - 2024.05.25 Mengmeng Kuang  保留所有权利！

:D 获取中...

Creative Commons License