CPU乱序执行

超、凢脫俗 2022-09-17 01:51 184阅读 0赞

# CPU乱序执行 #

### 文章目录 ###

*  CPU乱序执行
 *   *  前言
     *  CPU乱序执行
     *  编译器乱序优化
     *  GCC编译器优化屏障和内存屏障
     *  C语言中的内存屏障
     *  CPU存储模型
     *  实际遇到的问题

## 前言 ##

**乱序优化包括：**

*  CPU乱序执行优化
 *  编译器乱序优化

**对应的限制乱序优化的方式：**

*  内存屏障
 *  优化屏障

## CPU乱序执行 ##

CPU在保证结果一致的情况下，把原来有序的指令列表，按照指令依赖关系和指令执行周期，重 新安排执行顺序。

// 原代码 
    int a = 10; 
    int b = a; 
    int c = 20; 
    int d = c; 
    
    // 实际CPU执行的结果
     int a = 10; 
     int c = 20; 
     int b = a; 
     int d = c;

CPU乱序优化在一定程度上可以提高程序的运行速度。在多核情况下，由于CPU内部的高速缓存， 乱序执行对访问指令的影响可能导致对数据的影响不能及时的反映到主存上，从而导致结果错误。

**我们在一个核上执行写入数据的操作，并在最后写一个标记来表示之前的数据已经准备好，然后另 外一个核上通过判断标志来确定数据是否准备好. 这种做法存在风险：标志位先被写入，但是之前 的数据操作并未完成（可能未计算完成，也可能是数据没有从CPU缓存善刷新到主存），最终导 致了另外一个核使用了错误的数据。**

处理器的**分支预测**单元有可能直接把两条分支指令预取过来并发执行，等到分支判断的结果出来 后,再丢弃掉错误的数据。在下面的例子中，代码的本意是先计算a的结果，后面才能继续运算。实 际上CPU直接把三个运算同时计算，最后直接挑选正确的p值。

a = b + c; 
    if (a > 0) {  
    	p = x + y; 
    } else {  
    	p = x - y; 
    }

CPU乱序的本质原因是CPU为了效率，将长费时的操作“异步”执行，排在后面的指令不等前面的指 令执行完毕就开始执行后面的指令。而且允许排在前面的长费时指令后于排在后面的指令执行完。 如在CPU0上执行下面两句话：

a = 1; 
    b = 2;

在以下情况下b=2会先于a=1执行完：a没有缓存于CPU0的cache上，而b缓存于CPU0的cache 上，且处于Exclusive状态。

在一个CPU上写入没有缓存的变量，CPU0不能仅仅在它的cache里写入a=1，它还要告诉缓存a所 在的CPU：你上面的a缓存过期（Invalidate）了! 等CPU0收到响应（Invalidate Ack）后，才能写 入。

这一通信过程是需要耗费时间，而且距离收到ack的时间是不确定的，这限制于总线的繁忙程度以 及CPU1是否在执行优先级高的任务等等。所以CPU0不能干等着，它要向后继续执行指令：b=2， 而b位于本cache上且处于Exclusive状态，可以直接修改b的值（b变为Modified状态）。此时b=2 已经执行完毕，而a=1还没有执行完毕！从时序上来讲，这就是乱序执行。 CPU 执行乱序主要有以下几种：

*  写写乱序(store store)：a=1;b=2 -> b=2;a=1;
 *  写读乱序(store load)：a=1;load(b); -> load(b);a=1;
 *  读读乱序(load load)：load(a);load(b); -> load(b);load(a);
 *  读写乱序(load store)：load(a);b=2; -> b=2;load(a);

## 编译器乱序优化 ##

受到处理器预取单元的能力限制，处理器每次只能分析一小块指令的并发性，如果指令相隔比较远 就无能为力了。但是从编译器的角度来看，编译器能够对很大一个范围的代码进行分析，能够从更 大的范围内分辨出可以并发的指令，并将其尽量靠近排列让处理器更容预取和并发执行，充分利用 处理器的乱序并发功能。所以现代的高性能编译器在目标码优化上都具备对指令进行乱序优化的能 力，并且可以对访存的指令进行进一步的乱序，减少逻辑上不必要的访问主存，以及尽量提高 Cache命中率和CPU的LSU（load/store unit）的工作效率。所以在打开编译器优化以后，看到生 成的汇编码并不严格按照代码的逻辑顺序是正常的。比如：

int *p, *q; 
    // ... 
    *p = 1;
     *p = 2; 
     *q = *p;

这样，编译器通常会优化掉前面一个对p的写入（逻辑上冗余），仅对p写入2。而对q赋值的时 候，编译器认为此时q的结果就应该是上次p的值，会优化掉从p取数的过程，直接把在寄存器中保 存的p的值给q：

// ... 
    *p = 2; 
    *q = 2; 
    // ...

内存屏障  
大多数现代计算机为了提高性能而采取乱序执行，这使得内存屏障成为必须。

内存屏障，也称内存栅栏，内存栅障，屏障指令等， 是一类同步屏障指令，是CPU或编译器在对 内存随机访问的操作中的一个同步点，使得此点之前的所有读写操作都执行后才可以开始执行此点 之后的操作。

语义上，内存屏障之前的所有写操作都要写入内存；内存屏障之后的读操作都可以获得同步屏障之 前的写操作的结果。因此，对于敏感的程序块，写操作之后、读操作之前可以插入内存屏障。

大多数处理器提供了内存屏障指令:

*  完全内存屏障（full memory barrier）：保障了早于屏障的内存读写操作的结果提交到内存 之后，再执行晚于屏障的读写操作。
 *  内存读屏障（read memory barrier）：仅确保了内存读操作。
 *  内存写屏障（write memory barrier）：仅保证了内存写操作。  
    X86指令集中的内存屏障指令是：
 *  sfence：写屏障store fence，串行化发生在SFENCE指令之前的写操作但是不影响读操作， 即在sfence指令前的写操作当必须在sfence指令后的写操作前完成。
 *  lfence：读屏障load fence，串行化发生在SFENCE指令之前的读操作但是不影响写操作，即 在lfence指令前的读操作当必须在lfence指令后的读操作前完成。
 *  mfence：读写屏障，串行化发生在MFENCE指令之前的读写操作，即在mfence指令前的读 写操作当必须在mfence指令后的读写操作前完成。

ARM指令集中的内存屏障指令（隔离指令）：

*  dmb：数据存储器隔离，数据内存屏障指令，Data Memory Barrier。DMB指令保证： 仅当 所有在它前面的存储器访问操作都执行完毕后，才提交(commit)在它后面的存储器访问操 作。其它数据处理指令等可以越过 DMB 屏障乱序执行。
 *  dsb：数据同步隔离，数据同步屏障指令，Data Synchronization Barrier。比DMB严格： 仅 当所有在它前面的存储器访问操作都执行完毕后，才执行在它后面的指令（亦即任何指令都 要等待存储器访问操作——译者注）
 *  isb：指令同步隔离，指令同步屏障指令，Instruction Synchronization Barrier。最严格：它 会清洗流水线，以保证所有它前面的指令都执行完毕之后，才执行它后面的指令。即ISB 屏 障之前的指令保证执行完，屏障之后的指令直接flush掉再重新从Memroy中取指。

存储器也提供了另一套语义的内存屏障指令:

*  acquire semantics: 该操作结果可利用要早于代码中后续的所有操作的结果。
 *  release semantics: 该操作结果可利用要晚于代码中之前的所有操作的结果。
 *  fence semantics: acquire与release两种语义的共同有效。即该操作结果可利用要晚于代码 中之前的所有操作的结果，且该操作结果可利用要早于代码中后续的所有操作的结果。

## GCC编译器优化屏障和内存屏障 ##

编译器会对生成的可执行代码做一定优化，造成乱序执行甚至省略（不执行）。gcc编译器在遇到 内嵌汇编语句：

asm volatile("" ::: "memory");

将以此作为一条内存屏障，重排序内存操作。即此语句之前的各种编译优化将不会持续到此语句之 后。

优化屏障告知编译器:

*  内存信息已经修改，屏障后的寄存器的值必须从内存中重新获取
 *  必须按照代码顺序产生汇编代码，不得越过屏障

另外，GCC提供了Built-in的原子操作函数可以使用，GCC 4以后的版本也提供了Built-in的屏障函 数\_\_sync\_synchronize()，这个屏障函数既是编译屏障又是内存屏障，代码插入这个函数的地方会 被安插一条mfence/dmb指令。

## C语言中的内存屏障 ##

C与C++语言中，**volatile**关键字意图允许内存映射的I/O操作。这要求编译器对此的数据读写按照 程序中的先后顺序执行，不能对volatile内存的读写重排序。因此关键字volatile并不保证是一个内 存屏障。

**C/C++的**volatile**关键字也能起到优化限制的作用，但是和Java中的volatile（Java 5之后）不 同，C/C++中的volatile不提供任何防止乱序的功能，也并不保证访存的原子性。**

POSIX规范列出了必须为“synchronize memory with respect to other threads”的函数,其中包括 pthread\_mutex\_lock()和pthread\_mutex\_unlock()之类的函数. 在Appendix A.4.11中,明确说明了“同步内存”的功能：

*  必须由高级编译系统识别,以使内存操作和对这些函数的调用不会因优化而重新排序；
 *  根据特定的计算机,可能必须添加内存同步指令.

## CPU存储模型 ##

现代CPU架构的存储模型

现代CPU架构图 
    +-------------+            +-------------+ 
    |     CPU0    |            |    CPU1     | 
    +-------------+            +-------------+ 
    ^        |                      |        ^ 
    |        V                      V        | 
    |    +--------+            +--------+    | 
    |    | Store  |            |  Store |    | 
    |<-->| Buffer |            | Buffer |<-->|
    |    +--------+            +--------+    |
    |        |                      |        | 
    |        V                      V        | 
    +-------------+            +-------------+ 
    |    Cache    |            |    Cache    | 
    +-------------+            +-------------+ 
    |                                        | 
    +------------+              +------------+ 
    | Invalidate |              | Invalidate | 
    |    Queue   |              |   Queue    | 
    +------------+              +------------+ 
    |              Interconnect              | 
    +--------------------+-------------------+                  
                         | 
    +--------------------+-------------------+ 
    |                 Memory                 | 
    +----------------------------------------+

简单地分析一下，为什么CPU中会有Store Buffer和Invalidate Queue。

*  Store Buffer

在没有store buffer时，CPU 写入一个量，有以下情况。  
1）量不在该CPU缓存中，则需要发送read invalidate信号，再等待此信号返回，之后再写入 量到缓存中。  
2）量在该CPU缓存中，如果该量的状态是exclusive则直接更改。而如果是shared则需要发 送invalidate消息让其它CPU感知到这一更改后再更改。

这些情况中，很有可能会触发该CPU与其它CPU进行通讯，接着需要等待它们回复。这会浪 费大量的时钟周期！为了提高效率，可以使用异步的方式去处理：先将值写入到一个buffer 中，再发送通讯的信号，等到信号被响应，再应用到cache中。并且，此buffer能够接受该 CPU读值。这个buffer就是store buffer。而不等对某个量的赋值指令的完成，继续下一条指 令，去store buffer中读该量的值，这种优化叫store forwarding。

*  Invalidate Queue  
    同理，解决了主动发送信号端的效率问题，那么，接受端CPU接受到invalidate信号后如果立 即采取相应行动（去其它CPU同步值），再返回响应信号，则时钟周期也太长了，此处也可 优化。接受端CPU接受到信号后不是立即采取行动，而是将invalidate信号插入到一个queue 中，立即作出响应。等到合适的时机，再去处理这个queue中的invalidate，作相应处理。这 个queue就是invalidate queue。

下面说明内存屏障中读屏障和写屏障的用法。  
在不使用内存栅栏的情况下：  
CPU0执行

a = 1;
    b = 1;

CPU1 执行

while (b != 1); 
    assert (a == 1);

CPU1的assert有可能会失败：只要a不位于CPU0的缓存上，b不位于CPU1的缓存上。  
CPU0需要发出read invalidate的消息去改写a的值，此消息被CPU1接受到后立即响应： invalidate ack并在invalidate queue中加入这条消息，但CPU1并没有实时处理此消息，而是等到 合适的时机再去处理。CPU1需要发出read的消息去获得b的值，此消息被CPU0接受到后立即响应 b的值。  
一种导致assert失败的执行顺序如下：

1.  CPU0的cache line：b=0；CPU1的cache line：a=0。
2.  CPU0和CPU1同时开始执行a=1和while(b!=1)。CPU0将a=1写入store buffer,发送read invalidate消息，CPU1发送(remote)read消息。
3.  CPU0的cache line：b=0，store buffer：a=1；CPU1的cache line：a=0。
4.  CPU0执行完b=1后收到read消息,CPU1收到read invalidate消息。CPU0直接写入CPU1的 cache line: b=1，CPU1在invalidate queue中标记a，表示CPU1中的a值无效，返回 invalidate ack。CPU0的cache line：b=1，store buffer：a=1；CPU1的cache line：a=0。
5.  CPU0的cache line：b=1，store buffer：a=1；CPU1的cache line：a=0 b=1，invalidate queue：a。
6.  CPU0执行完毕；CPU1取cache line中的a=0执行assert(a==1)，失败。

## 实际遇到的问题 ##

在使用共享内存+信号量的方式实现环形缓冲区时，遇到CPU乱序执行导致数据出错的问题，下面是一个 简化后的示例代码。

在测试代码中，创建了4个生产者（写）线程和1个消费者（读）线程。消费者线程调度策略设置为RR， 并设置线程亲和CPU0。生产者线程为默认调度策略，设置线程亲和除CPU0之外的其他CPU核。实测设 置调度策略和CPU亲和性之后问题复现概率提高。

在写线程对fifo写入数据的时候，需要先写入数据，再写入可读标志。

// 如果没有加内存屏障，可能出现CPU乱序执行，即先赋值可读标志，再写数据 
      *(pwrite + 1) = write_data;    // 写入数据 
       __sync_synchronize();        
       *pwrite = 1;                  // 写入可读标志

由于写进程对可读标志变量进行了两次赋值，在互斥量lock范围内先赋值为0，然后在unlock之后再赋值 为1。因此当数据变量不在缓存中，而标志变量存在缓存中的时候，会出现先执行写入可读标志后执行写 入数据的情况。这时，由于读写线程在不同的CPU上执行，就会出现读线程先读到可读标志为1，而读出 数据出错的现象。

在写入数据和写入可读标志之间加入屏障函数\_\_sync\_synchronize，即可保证先写入数据后写入标志的 执行顺行。

另外，如果把互斥锁加到写入数据和写入标志之后，也不会出现数据出错问题。这是因为互斥锁中会强 制执行内存同步。  
**通过反汇编，屏障函数\_\_sync\_synchronize对应的汇编指令是：dmb ish**

#include <stdio.h> 
    #include <errno.h> 
    #include <unistd.h> 
    #include <stdint.h> 
    #include <stdbool.h> 
    #include <string.h>
    #define __USE_GNU 
    #include <sched.h> 
    #include <pthread.h>
    static uint8_t fifo_buffer[4096]; 
    static uint32_t fifo_total; 
    static uint32_t fifo_free; 
    static uint32_t read_index; 
    static uint32_t write_index;
    static pthread_mutex_t mutex; 
    static pthread_cond_t cond_not_empty; 
    static pthread_cond_t cond_not_full;
    static uint32_t write_count = 0;
    
    static int fifo_init(void) 
    {     
    	fifo_total = sizeof(fifo_buffer);    
    	fifo_free = fifo_total;   
    	memset(fifo_buffer, 0, fifo_total);   
    	read_index = write_index = 0;
    	pthread_mutex_init(&mutex, NULL);    
    	pthread_cond_init(&cond_not_empty, NULL);    
    	pthread_cond_init(&cond_not_full, NULL); 
    }
    
    static void* reader(void *arg)
    {     
    	uint32_t read_data = 0;    
    	uint32_t read_count = 0;    
    	uint32_t error_count = 0;    
    	uint32_t count = 0;    
    	uint32_t *pread = NULL;
    	 // 设置线程亲和CPU0 
    	cpu_set_t mask;    
    	CPU_ZERO(&mask);    
    	CPU_SET(0, &mask);    
    	if (pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask) < 0) {         
    		perror("pthread_setaffinity_np");    
    	}
    			
    	while (1) {         
    		pthread_mutex_lock(&mutex);        
    		if (fifo_free == fifo_total) {             
    			pthread_cond_wait(&cond_not_empty, &mutex);        
    		}
    
    		pread = (uint32_t *)(fifo_buffer + read_index);
    
    		// 判断可读标志 
    		count = 0;        
    		while (1) {             
    			if (*pread == 1) {                 
    				break;            
    		}            
    		usleep(1000);            
    
    		count++;            
    		if (count >= 1000) { 
    			count = 0;                
    			printf("wait date readable\n");            
    		}        
    	}
    
    		
    		read_data = *(pread + 1);        
    		read_count++;        
    		if (read_count != read_data) {             
    			printf("read data error: count=%u data=%u\n", read_count, read_data);            
    			error_count++;
    			// 出错后，sleep后可以读到正确写入的值，可以证明是CPU乱序执行导致前面出错 
    			usleep(10000);            
    			read_data = *(pread + 1);            
    			printf(" re-read data: %u\n", read_data);        
    		}
    		
    		read_index += 8;        
    		read_index %= fifo_total;        
    		fifo_free += 8;
    		if (read_count % 10000 == 0) {             
    			printf("read data: %u, error count: %u\n", read_count, error_count);      
    		}
    		
    		pthread_mutex_unlock(&mutex);       
    		pthread_cond_signal(&cond_not_full);   
    	} 
    }
    
    
    static void* writer(void *arg) 
    {     
    	uint32_t write_data = 0;    
    	uint32_t *pwrite = NULL;
    
    	// 设置线程亲和除CPU0之外的其他CPU 
    	cpu_set_t mask;    
    	if (pthread_getaffinity_np(pthread_self(), sizeof(mask), &mask) < 0) {         
    		perror("pthread_getaffinity_np");    }    
    		CPU_CLR(0, &mask);    
    
    	if (pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask) < 0) {         
    		perror("pthread_setaffinity_np");    
    	}
    	
    	while (1) {         
    		pthread_mutex_lock(&mutex);        
    		if (fifo_free == 0) {             
    			pthread_cond_wait(&cond_not_full, &mutex);        
    		}
    		
    		pwrite = (uint32_t *)(fifo_buffer + write_index);        
    		*pwrite = 0;
    		write_count++;       
    		write_data = write_count;
    		write_index += 8;        
    		write_index %= fifo_total;       
    		fifo_free -= 8;
    		
    		pthread_mutex_unlock(&mutex);        
    		pthread_cond_signal(&cond_not_empty);
    
    		// 如果没有加内存屏障，可能出现CPU乱序执行，即先赋值可读标志，再写数据 
    		*(pwrite + 1) = write_data;   // 写入数据 
    		__sync_synchronize();        
    		*pwrite = 1;                  // 写入可读标志
    
    		// 把互斥锁加在这里也可以解决 
    		//pthread_cond_signal(&cond_not_empty); 
    		//pthread_mutex_unlock(&mutex);
    		usleep(1000);    
    	} 
    }
    
    int main(int argc, char *argv[]) 
    {     
    	int ret, i;    
    	int writers = 4;   
    	pthread_t thread_handle;
    	printf("barrier test\n");
    	ret = fifo_init();    
    	if (ret) {         
    		printf("create thread reader ERROR!\n");        
    		return -1;    
    	}
    	
    	pthread_attr_t attr;    
    	struct sched_param param;    
    	pthread_attr_init(&attr);    
    	pthread_attr_setschedpolicy(&attr, SCHED_RR);    
    	param.__sched_priority = 11;    
    	pthread_attr_setschedparam(&attr, &param);    
    	pthread_attr_setscope(&attr, PTHREAD_SCOPE_SYSTEM);    
    	pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);    
    	
    	ret = pthread_create(&thread_handle, &attr, reader, NULL);    
    	if (ret) {         
    		printf("create thread reader ERROR!\n");        
    		return -1;    
    	}    
    	
    	pthread_attr_destroy(&attr);
    	for (i = 0; i < writers; i++) {         
    		ret = pthread_create(&thread_handle, NULL, writer, NULL);        
    		if (ret) {             
    			printf("create thread reader ERROR!\n");           
    	 		return -1;        
    		}    
    	}
     
    	while (1) {         
    		sleep(60);    
    	}
     
    	return 0; 
    }