HDFS pipeline写 -- datanode

站在DataNode的视角，看看pipeline写的流程，本文不分析客户端部分，从客户端写数据以前拿到了3个可写的block位置提及。

每一个datanode会建立一个线程DataXceiverServer，接收上游过来的TCP链接，对于每一个新建的TCP链接,都会建立一个叫作DataXceiver的线程处理这个链接. 这个线程不断的从TCP链接中读op，而后调用processOp(op)处理这个op，这里以write block 这个op为例.

对于datanode来讲，write block操做由DataXceiver的writeBlock函数实现.

大致步骤以下：

1. new 一个BlockReceiver对象，随后用于接收上游(client或者datanode)的block数据.

2. 根据传进来的DatanodeInfo数组，向数组的第一个元素表明的datanode创建TCP链接,targets参数是从上游的TCP链接中解析出来的，逻辑在Receiver的opWriteBlock方法中，Receiver是DataXceiver的基类.而后调用Sender的writeBlock方法给下游datanode发送write block相关元信息，包括DatanodeInfo数组(刨去第一个元素)，clientname,block的当前gs,minBytesRcvd,maxBytesRcvd(对于append,recovery操做有用)等。而后读取下游的回复封装在BlockOpResponseProto对象中，能够经过内部成员firstBadLink知道建pipeline中第一个失败的datanode节点。接着将BlockOpResponseProto回复给上游
   (datanode或者client)，最后调用第一步new的BlockReceiver的receiveBlock方法用于接收一个完整的block.以下：

   receiveBlock内部根据clientname发现是一个客户端在写block，建立一个PacketResponder线程用于处理下游datanode对packet的ack.PacketResponder后面分析。接着，不断的调用receivePacket()方法从上游(datanode或者client)接收一个个的packet，接收一个完整的packet的逻辑是由内部的PacketReceiver来处理的.
   对于一个接收到的packet，写入block file文件，同时checksum信息写meta文件，而后放入PacketResponder的ack queue队列，而后将packet写给下游的datanode。最后调用PacketResponder的 close方法，这个方法会等到ack queue为空，即全部packet都已经从下游收到，而且已经给上游ack.

3. receiveBlock()结束后，关掉和上下游的链接.

清空ack queue的逻辑由专门处理下游ack包的PacketResponder线程处理,逻辑以下：

1. 若是datanode是pipeline的中间node（经过PacketResponder的type属性来决定，LAST_IN_PIPELINE和HAS_DOWNSTREAM_IN_PIPELINE）,
   那么从下游读一个PipelineAck，从ack中拿到seqno，而后从ack queue中get(不删除)第一个packet，拿出seqno，记做expected_seq_no，而后比较是否相等，若是不相等，说明写出错. 若是seqno相同，往下.

2. 若是从ack queue中get的packet是block的最后一个packet，说明一个block接收完成.那么调用finalizeBlock方法.finalizeBlock方法逻辑以下：

   关闭block file和meta file文件，调用FsDatasetImpl的finalizeBlock(block)将block文件以及对应的meta文件移动到对应的block pool下的finalized目录下，而后生成一个FinalizedReplica对象，将bpid->FinalizedReplica的映射关系记录在内存中的volumnMap中,对象位于FsDatasetImpl下的ReplicaMap volumnMap(从ReplicaMap中定位一个ReplicaInfo，须要拿着bpid和block id去找)最后调用datanode的closeBlock()方法,将block回报给namenode,该方法逻辑以下：

   拿着block的bpid从BlockPoolManager中拿到相应的BPOfferService，通知namenode这个block。在data node这边，data node和每一个namenode的接口由一
   个BPServiceActor来承担，这是一个线程, 这个线程会向namenode汇报received block或者指示namenode去删除block.最后调用DatanodeProtocolClientSideTranslatorPB bpNamenode的blockReceivedAndDeleted()将block信息汇报上去.

3. 给从下游接收的ack回复给上游。

4. 将packet从ack queue的头部删除。

能够看出，一个block的写操做对于每一个data node来讲，由两个线程参与，一个是DataXceiver，用于接收上游的数据，一个是PacketResponder，用于处理下游回来的ack。尚未接收到下游的ack而且没有给上游回复ack的packet都存在在ack queue中。

参考资料

hadoop-hdfs-2.4.1.jar