[嵌入式]进程间通信概述

进程间通信

进程间通信用来解决：

1. 一个进程传递消息给另外要给进程
2. 确保两个或多个进程在关键活动中不会出现交叉。如两个进程争抢一个资源。
3. 确保正确的顺序。

竞争条件

当两个或多个进程读写某些共享数据时，执行结果取决与进程的执行顺序，称为竞争条件。

临界区

对共享内存进行的访问的程序片段称作临界区域。而保证多个进程中只有一个进程能够对共享内存进行操作则称为互斥。
为了处理好并发，程序应该满足以下四个条件：

1. 任何两个进程不能同时处于临界区域
2. 不对CUP的速度和数量作任何假设。
3. 临界区外运行的程序不得阻塞其他进程。
4. 不得时进程无限期等待进入临界区。

忙等待的互斥

cpu空转导致的互斥。

屏蔽中断

每个进程进入临界区后立刻屏蔽所有中断，屏蔽中断后cpu无法进行调度，因为cpu只会在时钟中断或其他中断时才会进行调度，而屏蔽中断也会屏蔽时钟中断。
特点：

1. 屏蔽中断只能在一个cpu上有效，在多核处理器上，屏蔽一个cpu中断，不会对另外一个有什么影响。
2. 对内核来说，在它更新变量或列表的几条指令期间，屏蔽中断是很有用的。

锁变量

使用一个变量来代表锁。假设锁变量为1代表上锁，则一个进程测试锁为0后，会进入临界区后然后将锁变量置为1，离开临界区后将锁变量置为0。
这种方法并不能保证真正的互斥。因为在测试锁变量为0但没有设置锁变量这个时间段可能有多个进程进入临界区。

严格轮换法

连续测试直到某个条件为正称作忙等待，使用忙等待的锁称为自旋锁。这种方法会浪费cpu的运行时间，只有在等待时间特别短的时候才能使用。
伪代码如下：

while(TRUE)
{
	while(turn != 0) 
		;
	//临界区
	turn = 1;
}
while(TRUE)
{
	while(turn != 1)
		;
	//临界区
	turn = 0;
}

两个进程轮流进入临界区，在其中一个进程很慢的情况的并不是一种很好的方法。

Peterson解法

Petenson算法

#define FALSE	0 
#define TRUE	1
#define N 		2
int turn;
int interested[N]
void enter_region(int process)
{
	int other = 1 - process;
	turn = process;					//TURN用来保证只有一个能进入临界区
	interested[process] = TRUE;		//
    while( turn == process && interested[other] == TRUE )
    	;
}
void leave_region(int process)
{
	interested[process] = FALSE;
}

TSL指令

需要硬件进行支持，由处理器提供一个指令:

	TSL RX,LOCK

该指令作用，RX存入LOCK，LOCK置非0，这些动作为一个原子操作。这个指令称为测试并加锁(test and set lock)。TSL并没有真正的判断操作，它只是保证了读写是一个原子操作。执行TSL指令的CPU将锁住内存总线，以禁止其他CPU在本指令执行结束之前访问内存。

屏蔽中断并不等于锁住内存总线，处理器1屏蔽了中断并不会对处理器2有什么影响。因此，要禁止其他CPU访问内存，需要锁住内存总线。

enter_region:
	TSL REGISTER,LOCK				#LOCK存入REGISTER中，置LOCK为非0.原子操作。
	CMP REGISTER,#0					#REGISTER为非0值，表示有进程在临界区
	JNE enter_region				#若REGISTER非0，跳转到enter_region
	RET
leave_region:						#离开临界区
	MOV LOCK,#0						#设置LOCK为0，其他进程就能进入
	RET

多个进程有各自的寄存器，但是它们共享一个内存块LOCK。在一个进程将LOCK置为1后，其他进程能看到这一结果。TSL实现同步的关键是读和写是原子的，而之前锁变量无法实现同步就是因为读和写不是原子的，从而导致不同的进程读锁变量时可能会得到不一致的状态。使用读写同步能解决该问题。
INTER x86 CUP在低层同步中使用XCHG指令,XCHG原子性的交换两个位置的内容。

enter_region:
	MOVE	REGISTER,#1	
	XCHG 	REGISTER,LOCK			#和TSL类似。REGISTER为非0值，代表有进程在临界区
	CMP		REGISTER,#0
	JNE		enter_region
	RET
leave_region:
	MOVE	LOCK,#0
	RET

两个指令都是，存入一个内存的值，然后置内存的值为非0，且为原子操作。

忙等待的一些问题

忙等待的基本思路：当想要进入临界区时，先检查是否可以进入。如果不能，则进程会原地等待，直到可以进入为止。
这种方法不仅会浪费cpu时间，同时也会存在优先级反转的问题。
假设有两个进程L和H。L的优先级较低,H的优先级较高，并且调度规则规定，当H就绪时，就要调度H。考虑如下情景，L进入临界区后，此时H就绪，准备进入临界区。H想要进入临界区，但临界区中已经被L占据，又由于H的优先级较高，L无法被调度以离开临界区。所以，H会永远的处于忙等待以进去临界区。

信号量

信号量是一种特殊的类型，可以取0或者为正值。它有两种操作，分别时down和up。
down操作会对信号量的值减1，若该值大于0，则直接减去1；若该值等于0，则进程会睡眠，此时down操作并未完成。up让信号量的值加1，若有1个或多个进程在该信号量上睡眠，会随机唤醒一个进程完成down操作，完成down操作后，信号量的值仍为0。检查、修改变量值、睡眠都是原子操作。
初值为1的信号量称作二元信号量，此时只有一个进程能够进到临界区。
生产者与消费者

#define N 100
typedef int semaphore;		//信号量
semaphore mutex = 1;		//控制临界区
semaphoer empty = N;		//空槽数量
semaphoer full  = 0;		//有数据的槽的量
int items       = 0;
void producer(void)
{
	while(true)
	{
		down(empty);
		down(mutex);
		++items;
		up(mutex);
		up(full);
	}
}
void consumer(void)
{
	while(true)
	{
		down(full);
		down(mutex);
		--item;
		up(mutex);
		up(empty);
	}
}

信号量有两种作用：

• 同步。用来协调不同操作的顺序。
• 互斥。用户保证任一时刻仅有一个进程读写缓冲区和相关的变量。

在上面的例子中，full和empty用于控制事件的发生或不发生，而mutex用于保证仅有临界区内只有一个进程。

使用两个信号量
使用两个信号量M,S，一个用于互斥，一个用于阻塞。此外，每个任意信号量还会关联一个计数器，该计数器即任意值信号量。互斥保证只有一个进程能对计数器执行操作，而阻塞则是让进程阻塞于其上。
执行down操作，首先会对M执行down（进入临界区，会修改计数器的值），随后对计数器的值减去一， 接着判断计数器的正负，如果：

• $\ge 0$。对M执行Up。
• $<0$。对M执行Up（离开临界区）。还需要对S执行down。
  执行up操作，对M执行down，将计数器的值加一，判断计数器的值:
• $>0$。对M执行up
• $\le 0$。取出阻塞队列中的一个进程运行，若有。对M执行up，对S执行up。

互斥量

互斥量是信号量的简要版本，它省略了信号量的计数能力，只有两种状态：解锁和加锁。它仅仅使用于管理共享资源或一小段代码。
若互斥量没有被上锁，进程对它加锁会获得一把锁，之后，进程能够顺利进入临界区。否则，进程会阻塞。解锁操作会释放锁，并随机选择一个阻塞的进程允许它获得锁。
使用TSL和XCHG可以在用户空间实现线程包，使用TSL实现的线程包如下：

mutex_lock:
	TSL		REGISTER,LOCK
	CMP		REGISTER,#0
	JZ		ok							#成功获得锁
	CALL	thread_yield				#放弃CPU，运行另外一个线程
	JMP		mutex_lock
ok:	RET
	
mutex_unlock:
	MOVE	LOCK,#0
	RET

在mutex_lock中，当进入临界区失败后会放弃CPU，运行另外一个线程。而在忙等待的enter_region中，进入失败后会不停的循环直到时钟中断。
注意，在用户空间实现的线程并不会有时钟停止运行事件过长的线程。结果是通过忙等待的方式来获得锁的线程会永远循环下去。

futex

如果等待时间较短，则适合用自旋锁，此时使用互斥量会使得内核开销占比大。但如果竞争激烈，则不适合用自旋锁，因为会浪费大量的CPU时间，此时比较适合互斥量。
futex，快速用户空间互斥，它实现了基本的锁，但避免了陷入内核。一个futex包含两部分，分别是用户库和内核服务。
内核服务提供了一个等待队列，阻塞的进程会存于该队列中，将进程存入队列中或解除阻塞需要用到系统调用。
用户库提供了两个操作，减少并检验，增加并检验。

• 减少并检验用来获取锁，如果获取锁失败，则会通过系统调用将该进程放入等待队列中。
• 增加并检验用来释放锁，释放锁后如果有进程阻塞在等待队列中，会通知内核停止阻塞等待队列中的一个或多个进程。

futex检验是否持有锁实现在用户空间，相比实现在内核空间的锁具有更小的内核损耗。

pthread

pthread使用了一个互斥量来保护临界区，并且还提供了条件变量用于同步。
互斥量操作：

条件变量操作：

下列代码是信号量和互斥量的联合使用的示例：

#define MAX 100000000
pthread_mutex_t mutex;
pthread_cond_t	cond;
int buffer = 0;
void *producer(void *args)
{
	for(int i=0; i<MAX; ++i)
	{
		pthread_mutex_lock(&mutex);
		while(buffer != 0)						//buffer是条件
		{
			pthread_cond_wait(&cond, &mutex);	//互斥量与信号量合用
		}
		buffer = i;
		pthread_cond_signal(&cond);
		pthread_mutex_unlock(&mutex);
	}
	pthread_exit(0);
}
void *consumer(void *args)
{
	for(int i=0; i<MAX; ++i)
	{
		pthread_mutex_lock(&mutex);
		while(buffer == 0)
		{
			pthread_cond_wait(&cond, &mutex);
		}
		buffer = 0;
		pthread_cond_signal(&cond);
		pthread_mutex_unlock(&mutex);
	}
	pthread_exit(0);
}
int main(int argc, char **argv)
{
	pthread_t pro, con;
	pthread_mutex_init(&mutex);
	pthread_cond_init(&cond);
	
	pthread_creat(&pro, NULL, producer, NULL);
	pthread_cread(&con, NULL, consumer, NULL );
	
	pthread_join(pro, 0);
	pthread_join(con, 0);
	
	pthread_mutex_destroy(&mutex);
	pthread_cond_destroy(&cond);
	return 0;
}

条件变量用于通知条件变化了，仅此而已。如果没有条件变量，因为条件是全局变量，所以每次判断条件都需要不停的加锁和释放锁，这会增加内核的消耗。有了条件变量之后，线程就能阻塞到条件变量上以等待条件成真，而不用频繁的加锁释放锁。条件的使用肯定要加上锁，因此在wait条件变量时需要知道保护条件的互斥量，因为它要释放锁以让其他线程更新条件。

管程

消息传递

屏障

屏障对于实现多个进程间的同步关系。它保证只有所有进程中的任何一个进程没有到达屏障，则其他所有的进程都会被屏障阻塞。

避免锁：读-复制-更新RCU

RCU是数据同步的一种方式，针对的是链表，目的是提高链表的访问效率。使用加解锁访问链表时，效率低下。RCU允许多个线程不加锁读取链表，然后允许一个线程加锁修改链表。
避免锁主要解决的问题：

1. 多个线程读，一个线程删除结点。当线程删除结点后，原有的读线程可能读到被释放的空间，造成系统崩溃。RCU会在删除结点之前，等待之前读线程操作完毕，这段时间叫做宽限期。
2. 在读取过程中，若插入一个新的结点，当线程读到这个结点时，需要保证该结点的完整性。
3. 保证读取链表的完整性。新增或者删除一个结点，原有的结点依然能够连续的读取到后面的结点。但是RCU不保证能否读到删除/插入结点。
   宽限期

void foo_read(void)
{
	rcu_read_lock();			
	foo *fp = gbl_foo;
	if ( fp != NULL )
			dosomething(fp->a,fp->b,fp->c);
	rcu_read_unlock();
}
 
void foo_update( foo* new_fp )
{
	spin_lock(&foo_mutex);
	foo *old_fp = gbl_foo;
	gbl_foo = new_fp;
	spin_unlock(&foo_mutex);
	synchronize_rcu();
	kfee(old_fp);
}

rcu_read_lock()和 rcu_unlock()标志RCU读过程的开始和结束，其作用是判断宽限期是否结束。synchronize_rcu()会进入宽限期，直到宽限期结束才返回。
订阅-发布机制
编译器对源码产生的指令进行优化，会导致指令的顺序发生变化，从而造成各个线程使用数据不一致的问题。
这里似乎是使用内存屏障来解决该问题。
数据读取的完整性
增加节点：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-DP4QwTIG-1630470696876)(./picture/增加结点.png)]
增加结点X的时候，先让增加结点的指针指向插入位置后面的结点B，之后在改变插入位置前面结点A的指针。
删除节点：
[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-9YS2oCUm-1630470696895)(./picture/删除结点.png)]
删除结点时，要先改变前缀的指针，在等待宽限期结束后，再删除结点。