HDFS3源码分析之Datanode写流程(四)

云存储
hdfs, 写入, 分布式, 文件系统

上一篇介绍了HDFS普通写在Namenode的实现逻辑, 最后来看看在Datanode(DN)的实现. 这块内容同样很多, 所以先抓核心要素, 有些细节可以先放一下

0x00. 前置

先来回顾一下, 从Client到DN端传输数据, 它本质通过socket的方式传输IO流, 而在发数据之前, 是如何创建这个pipeline(传输管道)的呢?

还记得之前画过的那辆小火车吧, 火车头(Sender)和到站点(Reciver)大体框架如下所示:

graph LR
Client --发送--> Sender
Reciver --解析--> datanode

subgraph DataTransferProtocol
Sender
Reciver
end

io[pipeline]
Sender --> io
io --> Reciver

这里要注意一下, Sender和Reciver都是一个大的载体, 里面包含了读写DN的多个关键的方法,它核心有:

  1. readBlock

  2. writeBlock

  3. transferBlock

  4. 本地读相关操作

    我们写数据调用的其实是Sender.writeBlock() , 然后来看看writeBlock时封装了哪些信息

注: 发送的信息结构从低版到高版本HDFS做过几次拆分调整, 下图是低版本整体结构: (高版本有不小区别. 暂没单独画)

sendHeader00

重点关注一下第二行的信息:

  • 比如其中的OP 代表操作码. 常见的写(80), 读(81)
  • 然后高版本加了一个重要的stage 参数, 标示当前pipeline处于什么状态, 就无需单独携带recoveryFlag了

所以这个图仅供参考, 实际信息以Sender.writeBlock()为准, 差别比较大. (大部分参数高版中都被整合或换名了, 切不可照套)

0x01. 接收Client请求

之前提到客户端在创建pipeline的时候, 会发送一个火车头(Sender)给第一个DN, 调用它的writeBlock()方法序列化一系列的DN信息, 然后再DN这边, 则通过启动之后就通过Receiver的具体实现DataXceiver的线程来接收信息了, 但是一个DN可能会有许多个Client同时发送读写请求, 这里DN就提供了一个控制层来处理(如图所示):

dxServer01

每个DN在启动的时候, 都会初始化一个DataXceiverServer (DXServer)线程来监听Socket的流式接口, 然后再转发给一个单独的DataXceiver (DX)线程, 过程类似于你通过滴滴打车, 请求是先提交给平台(DXServer), 然后平台再分派给指定的司机(DX)执行, 所以书中原图也挺清晰, 下面看看它的代码

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
38
39
40
41
@Override
  public void run() {
    Peer peer = null; // 封装了socket
    while (datanode.shouldRun && !datanode.shutdownForUpgrade) {
      try {
        // 1.接受请求
        peer = peerServer.accept();

        // 确认当前工作线程未超过4096个(默认值)
        if (datanode.getXceiverCount() > maxXceiverCount) 
           throw new IOException("Xceiver count exceeds the limit");
        
        // 2.直接创建并启动对应的DataXceiver线程
        new Daemon(datanode.threadGroup, DataXceiver.create(peer, datanode, this)).start();
          
      } catch (SocketTimeoutException ignored) { // 忽略socket超时
      } catch (AsynchronousCloseException ace) { // 这种异常比较特殊
        if (datanode.shouldRun && !datanode.shutdownForUpgrade)
          LOG.warn(datanode.getDisplayName() + ":DataXceiverServer: ", ace);
      } catch (IOException ie) {
        IOUtils.cleanup(null, peer);
      } catch (OutOfMemoryError ie) {
        IOUtils.cleanup(null, peer);
        try {
          Thread.sleep(30 * 1000);
        } catch (InterruptedException e) {}
      } catch (Throwable te) {
        datanode.shouldRun = false; // 其他异常直接关闭DN
      }
    } // end-while

    // 3. 如果跳出循环, 先关闭peerServer
    try {
      peerServer.close();
      closed = true;
    } catch (IOException ie) {logWarn(ie);}
    // 升级重启DN相关,暂略
      
    // 4. 关闭当前server的所有worker
    closeAllPeers();
  }

然后显然我们更关心的是Worker线程 —- DataXceiver的核心逻辑, 简单说它就是在解析OP参数跳转到对应的读/写/复制副本块等逻辑.

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
38
39
40
41
42
43
44
45
46
47
48
49
@Override
  public void run() {
    int opsProcessed = 0;
    Op op = null;

    try {
      // 1.通过DXServer转发客户端的输入输出流, 然后完成初始化
      synchronized(this) {xceiver = Thread.currentThread();}
      dataXceiverServer.addPeer(peer, Thread.currentThread(), this);
      peer.setWriteTimeout(datanode.getDnConf().socketWriteTimeout);
      InputStream input = socketIn;
      try {
        IOStreamPair saslStreams = datanode.saslServer.receive(params..);
        input = new BufferedInputStream(saslStreams.in, smallBufferSize);
        socketOut = saslStreams.out;
      } catch (InvalidMagicNumberException imne) {return;} // 严重异常
      super.initialize(new DataInputStream(input));
      
      // 2. 根据操作码循环执行. 然后等待
      do {
        try {
          if (opsProcessed != 0) {
            peer.setReadTimeout(dnConf.socketKeepaliveTimeout);
          } else {
            peer.setReadTimeout(dnConf.socketTimeout);
          }
          op = readOp();
        } catch (InterruptedIOException ignored) {break;}
          catch (EOFException | ClosedChannelException e) {break;}
          catch (IOException err) {
          incrDatanodeNetworkErrors();
          throw err;
        }
        if (opsProcessed != 0) peer.setReadTimeout(dnConf.socketTimeout);

        opStartTime = monotonicNow();
        processOp(op); // 执行具体的读/写操作
        ++opsProcessed;
      } while (!peer.isClosed() && dnConf.socketKeepaliveTimeout > 0);
    
    // 3. 如果worker连接关闭, 则进入最后的清理
    } catch (Throwable t) {
      // 具体的异常判断, 然后记一下log, 略
    } finally {
      collectThreadLocalStates();
      if (peer != null) {
        dataXceiverServer.closePeer(peer);
        IOUtils.closeStream(in);
}}}

所以关键就是看processOP()调用实际由DataXceiver实现的writeBlock() 方法处理客户端发送来的数据了.

0x02. 写数据整体

先来看看写入操作在DN端整体的时序图: (核心)

sequenceDiagram
    participant A as Client
    participant B as Datanode1
    participant C as Datanode2
    participant D as Datanode3
    %%participant E as DataNode

    A->>B: Sender发送写块请求
    B->>C: 建立连接 & 转发
    C->>D: 建立连接 & 转发
    D-->>C: 返回ACK
    C-->>B: 返回ACK
    B-->>A: 返回ACK

    %%rect rgb(182, 237, 227)
    rect rgb(208, 222, 247)
    %%rect rgb(208, 189, 230)
    alt Success
    A->>B: 发送Packet1
    end
    B->>C: 处理Packet1 & 转发
    A->>B: 发送Packet2~n
    C->>D: 处理Packet1 & 转发
    Note right of D: DN3进行校验
    Note left of A: 异步的发送packet
    D-->>C: 返回p1-ack   
    C-->>B: 返回p1-ack
    B->>C: 处理Packet2~n & 转发
    B-->>A: 返回p1-ack
    C->>D: 处理Packet2~n & 转发
    end

    alt 数据包发完
    A->>B: 发送空尾包
    end
    B->>C: 处理空尾包 & 转发
    C->>D: 处理空尾包 & 转发
    Note right of D: 确认收尾
    D-->>C: 返回ACK
    C-->>B: 返回ACK
    B-->>A: 返回ACK
    Note left of A: 文件写入完成

然后对于每个DN来说, 都需要维护两组输入/输出流 (最后一个DN除外, 它无下游):

stateDiagram

state IO数据流向 {
上游节点 --> 当前Datanode: in
当前Datanode --> 下游节点: mirrorOut
下游节点 --> 当前Datanode: mirrorIn
当前Datanode --> 上游节点: out
}

这里为了方便记忆, 把replyOut简写为out与in对应, 这是一对I/O流, 然后另一对就是用于转发给下一个DN的mirror I/O流 (镜像)

最后再来看上图中”处理Packet和转发“的具体实现, 方法比较长, 也按”总-分“的模式来看, (日志和目前无关的信息略去)

  1. 初始化一对IO流, 以及核心的BlockReceiver对象
  2. 连接到下游的Datanode节点, 初始化mirror_IO流
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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
 @Override
 public void writeBlock(params) {
   // 1. 初始化IO流, 以及关键的BlockReceiver对象
   previousOpClientName = clientname;
   // isDatanode标记当前的操作是否是Client发起. 初始值false表示Client发起..
   final boolean isDatanode = clientname.length() == 0;
   final boolean isClient = !isDatanode; // 初始值true
   final boolean isTransfer = stage == TRANSFER_RBW || stage == TRANSFER_FINALIZED; // 初始值false
   long size = 0;
   final DataOutputStream replyOut = getBufferedOutputStream(); // 对应上图的out

   int nst = targetStorageTypes.length; // 存储类型, 异构的内容
   StorageType[] storageTypes = new StorageType[nst + 1];
   storageTypes[0] = storageType;
   if (targetStorageTypes.length > 0) arraycopy(targetStorageTypes, 0, storageTypes, 1, nst);

   // 为了支持低版本client, 不跳过空的block_storageIds
   final int nsi = targetStorageIds.length;
   final String[] storageIds;
   if (nsi > 0) {
     storageIds = new String[nsi + 1];
     storageIds[0] = storageId;
     if (targetStorageTypes.length > 0) System.arraycopy(targetStorageIds, 0, storageIds, 1, nsi);
   } else {
     storageIds = new String[0];
   }

   // block的长度和时间戳之后会被改变, 现在先记录一份初始副本以便于转发给下游DN
   final ExtendedBlock originalBlock = new ExtendedBlock(block);

   DataOutputStream mirrorOut = null;  // 发给下一个DN的输出流
   DataInputStream mirrorIn = null;    // 下一个DN的返回输入流
   Socket mirrorSock = null;           
   String mirrorNode = null;           
   String firstBadLink = "";           // 建立pipeline时第一个异常的DN
   Status mirrorInStatus = SUCCESS;
   final String storageUuid; // 当前block在dn存储时的随机id?
   final boolean isOnTransientStorage;
   try {
     final Replica replica;
     if (isDatanode || stage != PIPELINE_CLOSE_RECOVERY) {
       // 初始化BlockReceiver对象, 这里改成get/set方法是因为br对象会被不同线程访问, 所以加了锁(?)
       setCurrentBlockReceiver(new BlockReceiver(params...));
       replica = blockReceiver.getReplica();
     } else { // 
       replica = datanode.data.recoverClose(block, latestGenerationStamp, minBytesRcvd);
     }
     storageUuid = replica.getStorageUuid();
     isOnTransientStorage = replica.isOnTransientStorage();

   
     // 2. 连接到下游的DN节点, 初始化mirrorI/O (大量细节, 可略)
     if (targets.length > 0) {
       InetSocketAddress mirrorTarget;
       mirrorNode = targets[0].getXferAddr(connectToDnViaHostname);
       mirrorTarget = NetUtils.createSocketAddr(mirrorNode);
       mirrorSock = datanode.newSocket();
       try {
         DataNodeFaultInjector.get().failMirrorConnection();
         int timeoutValue = dnConf.socketTimeout +(READ_TIMEOUT_EXTENSION * targets.length);
         int writeTimeout = dnConf.socketWriteTimeout +(WRITE_TIMEOUT_EXTENSION * targets.length);
         NetUtils.connect(mirrorSock, mirrorTarget, timeoutValue);
         mirrorSock.setTcpNoDelay(dnConf.getDataTransferServerTcpNoDelay());
         mirrorSock.setSoTimeout(timeoutValue);
         mirrorSock.setKeepAlive(true);
         if (dnConf.getTransferSocketSendBufferSize() > 0)
           mirrorSock.setSendBufferSize(dnConf.getTransferSocketSendBufferSize());

         OutputStream unbufMirrorOut = NetUtils.getOutputStream(mirrorSock, writeTimeout);
         InputStream unbufMirrorIn = NetUtils.getInputStream(mirrorSock);
         DataEncryptionKeyFactory keyFactory = datanode.getDataEncryptionKeyFactoryForBlock(block);
         IOStreamPair saslStreams = datanode.saslClient.socketSend(params.., targets[0]);
         unbufMirrorOut = saslStreams.out;
         unbufMirrorIn = saslStreams.in;
         mirrorOut = new DataOutputStream(new BufferedOutputStream(unbufMirrorOut,smallBufferSize));
         mirrorIn = new DataInputStream(unbufMirrorIn);

         String targetStorageId = null;
         if (targetStorageIds.length > 0) targetStorageId = targetStorageIds[0];

           // 2.2 写操作(转发火车头给下一个DN)
           // 参数targetPinnings表示块副本是否固定在DN上, 是则balance时不会被搬运, 默认空
           if (targetPinnings != null && targetPinnings.length > 0) {
            new Sender(mirrorOut).writeBlock(targetPinnings[0], targetParams..);
           } else {
               new Sender(mirrorOut).writeBlock(false, targetParams..);
           }

         mirrorOut.flush(); //刷给下游DN

         // 只有是客户端发起写操作,才需要读ACK
         if (isClient) {
             BlockOpResponseProto connectAck = parseFrom(mirrorIn); // 简写, 读取ack信息
             mirrorInStatus = connectAck.getStatus();
             firstBadLink = connectAck.getFirstBadLink();
         }

       } catch (IOException e) { // 如果在连接下一个DN时出现了异常
         if (isClient) { // 返回异常信息, 然后关闭到下游的socket和mirror_I/O
             Text.writeString(replyOut, targets[0].getXferAddr());
             replyOut.flush();
         }
         closeAndSetNull(mirrorOut, mirrorIn, mirrorSock)
         if (isClient) {
           LOG.error("Exception transfering block xx to mirror xx");
           throw e;
         } else {
           incrDatanodeNetworkErrors();
     }}}

     // 3. 封装创建pipeline结果返回给Client, 错误DN携带在badLink中(client查此)
     if (isClient && !isTransfer) {
         BlockOpResponseProto.newBuilder().setStatus(mirrorInStatus)
                                          .setFirstBadLink(firstBadLink)
                                          .build().writeDelimitedTo(replyOut);
       }
       replyOut.flush();
     }

     // 4. DN写数据并转发给下游DN (核心)
     if (blockReceiver != null) {
       String mirrorAddr = (mirrorSock == null) ? null : mirrorNode;
       // 写数据和转发数据都在此
       blockReceiver.receiveBlock(mirrorOut, mirrorIn, replyOut, targets, params..);
         
       if (isTransfer) writeResponse(SUCCESS, null, replyOut);// 块复制无需转发和ack, 直接返回
     }

     // 5. 更新当前block的时间戳和大小
     if (isClient && stage == PIPELINE_CLOSE_RECOVERY) {
       block.setGenerationStamp(latestGenerationStamp);
       block.setNumBytes(minBytesRcvd);
     }
     
     // 如果是pipeline的关闭恢复操作, 那么finalize当前副本, 汇报给NN
     if (isDatanode || stage == PIPELINE_CLOSE_RECOVERY) 
       datanode.closeBlock(block, null, storageUuid, isOnTransientStorage);

     if(isClient) size = block.getNumBytes();
   } catch (IOException ioe) {
     incrDatanodeNetworkErrors();
     throw ioe;
   } finally { 
     closeAllStreamAndSocket(params) // 最后, 依次关闭所有IO流和socket
     setCurrentBlockReceiver(null);
}}

在writeBlock的流程里, 有个关键的地方是初始化BlockReceiver, 它在客户端第一次新写block时, 也就是pipeline处于SETUP_CREATE 状态时, 它会执行创建RBW副本的操作. 主要做的是在rbw文件夹中创建一个新的副本, 初始状态RBW (副本的状态转变比较复杂, 详见此图, 在租约恢复篇再细说)

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
38
39
40
41
42
43
44
45
@Override // FsDatasetSpi
 public ReplicaHandler createRbw(params...) {
   try (AutoCloseableLock lock = datasetLock.acquire()) {
     ReplicaInfo replicaInfo = volumeMap.get(b.getBlockPoolId(), b.getBlockId());
     if (replicaInfo != null) {
       throw new ReplicaAlreadyExistsException(
       "Block x already exists in state x and can't be created.");
     }
     FsVolumeReference ref = null; //准备创建新RBW副本块

     // 仅当开启延时持久化, 且blockSize是OS的页整数倍时, 会使用内存(可显著降低临时资源占用)
     if (allowLazyPersist && lazyWriter != null &&
         b.getNumBytes() % cacheManager.getOsPageSize() == 0 &&
         reserveLockedMemory(b.getNumBytes())) {
       try {
         // 先尝试把副本块存放在临时存储中(比如内存)
         ref = volumes.getNextTransientVolume(b.getNumBytes());
         datanode.getMetrics().incrRamDiskBlocksWrite();
       } catch (DiskOutOfSpaceException de) {
         // 如果放内存失败, 则之后会直接放磁盘, 因此这种异常可以忽略
         LOG.warn("transient volume is not enough, fall back to persistent storage");
       } finally {
         if (ref == null) cacheManager.release(b.getNumBytes());
  }}

     // 创建RBW文件存放副本数据
     if (ref == null) ref = volumes.getNextVolume(params...);
     FsVolumeImpl v = (FsVolumeImpl) ref.getVolume();

     ReplicaInPipeline newReplicaInfo;
     try {
       // 这里底下会在块池中新建一个RBW文件, 调用层级比较深(细节暂略)
       newReplicaInfo = v.createRbw(b);
       if (newReplicaInfo.getReplicaInfo().getState() != ReplicaState.RBW)
         throw new IOException("Replica was not RBW. xxx");
     } catch (IOException e) {
       IOUtils.cleanup(null, ref);
       throw e;
     }

     // 最后在映射关系中记录新的副本
     volumeMap.add(b.getBlockPoolId(), newReplicaInfo.getReplicaInfo());
     return new ReplicaHandler(newReplicaInfo, ref);
   }
 }

0x03. 写入细节

上面整体的步骤过完了, 下面接着看正常写DN时BlockReceiver类的receiveBlock()方法, 同样先来看看它内部的整体流程: (参考书中图)

blockReceiverST00

它和我们在Client端看到的DataStreamer和ResponseProcessor 是相似的结构, 这里也有一个关键的ackQueue (数据包队列),所以理解了客户端的生产-消费模型, DN这里就容易理解了, 需要注意的是最后一个DN以及携带Sync参数的时候, packet入队时间有所调整, 接着看看代码:

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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
void receiveBlock(mirrOut, mirrIn, replyOut,DNInfo[] downstreams, params..) {

    syncOnClose = datanode.getDnConf().syncOnClose;
    dirSyncOnFinalize = syncOnClose;
    boolean responderClosed = false;
    mirrorOut = mirrOut;
    mirrorAddr = mirrAddr;
    initPerfMonitoring(downstreams);
    throttler = throttlerArg;

    this.replyOut = replyOut;
    this.isReplaceBlock = isReplaceBlock;

    try {
      if (isClient && !isTransfer) { 
        // 1. 先启动responder线程
        responder = new Daemon(datanode.threadGroup,
                               new PacketResponder(replyOut, mirrIn, downstreams));
        responder.start(); 
      }

      // 2. 然后读取 & 转发packet数据
      while (receivePacket() >= 0) { /* 循环读上游发送的packet, 并转发下游 */ }

      // 3. 等所有的packet都acked, 然后关闭responder线程
      if (responder != null) {
        ((PacketResponder)responder.getRunnable()).close();
        responderClosed = true;
      }

      // 副本复制相关, 暂略 (客户端写与此无关)
      if (isDatanode || isTransfer) {
        try (ReplicaHandler handler = claimReplicaHandler()) {
          close();
          block.setNumBytes(replicaInfo.getNumBytes());
          // 根据副本当前状态判断是否Finalized
          if (stage == TRANSFER_RBW) datanode.data.convertTemporaryToRbw(block);
           else datanode.data.finalizeBlock(block, dirSyncOnFinalize);
        }
      }

    } catch (IOException ioe) {
      replicaInfo.releaseAllBytesReserved();
      if (datanode.isRestarting()) {
        LOG.info("Shutting down for restart (" + block + ").");
      } else {
        LOG.info("Exception for " + block, ioe);
        throw ioe;
      }
    } finally {
      // 重启和异常判断相关, 暂略  
      Thread.interrupted();

      // 针对重启DN类的特殊情况, 给上游DN发送特殊通知 (可跳过)
      if (!responderClosed) {
        if (responder != null) {
          // 当前DN只是重启导致的短暂关闭, 给上游发送特殊ack
          if (datanode.isRestarting() && isClient && !isTransfer) {
            try (Writer out =
                 new OutputStreamWriter(replicaInfo.createRestartMetaStream())) {
              out.write(Long.toString(Time.now() + restartBudget));
              out.flush();
            } catch (IOException ioe) { //最坏情况就是Client触发写错误恢复
            } finally {
              IOUtils.closeStream(streams.getDataOut());
            }
            try { //等待原因略
              Thread.sleep(1000);
            } catch (InterruptedException ie) {}
          }
          responder.interrupt();
        }
        IOUtils.closeStream(this);
        cleanupBlock();
      }
        
      if (responder != null) {
        try { // join超时相关 (暂跳)
          responder.interrupt();
          long joinTimeout = datanode.getDnConf().getXceiverStopTimeout();
          joinTimeout = joinTimeout > 1  ? joinTimeout*8/10 : joinTimeout;
          responder.join(joinTimeout);
          if (responder.isAlive()) throw new IOException(..);
        } catch (InterruptedException e) {
          responder.interrupt();
        if (!datanode.isRestarting()) throw new IOException("Interrupted receiveBlock");
        }
        responder = null;
 }}}

可看到关键的处理工作都被交给了receivePacket() , 其他无关内容可暂略过, 后续再看

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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
// 返回当前packet的字节大小
private int receivePacket() throws IOException {
  // 1.按照packet格式从输入流中读取数据然后放入ByteBuffer (读流程详述)
  packetReceiver.receiveNextPacket(in);
  PacketHeader header = packetReceiver.getHeader();
  checkHeader(header);

  long offsetInBlock = header.getOffsetInBlock();
  long seqno = header.getSeqno();
  boolean lastPacketInBlock = header.isLastPacketInBlock();
  final int len = header.getDataLen();
  boolean syncBlock = header.getSyncBlock();

  // 避免sync和尾包标记重复作用
  if (syncBlock && lastPacketInBlock) {
    this.syncOnClose = false;
    this.dirSyncOnFinalize = true;
  }

  // 更新当前block记录的大小
  final long firstByteInBlock = offsetInBlock;
  offsetInBlock += len;
  if (replicaInfo.getNumBytes() < offsetInBlock) replicaInfo.setNumBytes(offsetInBlock);
 
  
  // 2.如果当前非最后一个DN, 也未设置sync, 直接放入ackQueue(队列)
  if (responder != null && !syncBlock && !shouldVerifyChecksum())
    ((PacketResponder) responder.getRunnable()).enqueue(new Packet(seqno, params..));

  // 3.向下游DN转发packet
  if (mirrorOut != null && !mirrorError) {
    try {
      long begin = Time.monotonicNow();
      packetReceiver.mirrorPacketTo(mirrorOut);
      mirrorOut.flush();
      long now = Time.monotonicNow();
      setLastSentTime(now);
      long duration = now - begin;
      trackSendPacketToLastNodeInPipeline(duration); // TODO, 之后看看记录原因
      // 关键的慢日志, 代表给下游DN发packet超时 (各种资源不足情况, 需要特别注意)
      if (duration > datanodeSlowLogThresholdMs) {
        LOG.warn("Slow BlockReceiver write packet to mirror took " + duration +
                 "ms (threshold=" + datanodeSlowLogThresholdMs + "ms), "+
                 "downstream DNs=" + Arrays.toString(downstreamDNs)+
                 ", blockId=" + replicaInfo.getBlockId());
      }
    } catch (IOException e) {
      handleMirrorOutError(e);
    }
  }
  
  ByteBuffer dataBuf = packetReceiver.getDataSlice();
  ByteBuffer checksumBuf = packetReceiver.getChecksumSlice();
  
  // 3. 如果当前是最后的空尾包, 或者是心跳包 (数据域为0)
  if (lastPacketInBlock || len == 0) { // 记录一条debug日志, 此略
    if (syncBlock) flushOrSync(true); // 携带了sync参数, 则立刻将数据刷到磁盘
  } else {
    final int checksumLen = diskChecksum.getChecksumSize(len);
    final int checksumReceivedLen = checksumBuf.capacity();
    if (checksumReceivedLen > 0 && checksumReceivedLen != checksumLen) 
      throw new IOException("received length is xx,  but expected length is xx");

    // 如果当前为最后一个DN, 则校验packet的数据和是否匹配
    // 不匹配会标记错误, 放入ackQueue, 等待responder线程发送给上游节点, 然后抛出异常终止
    if (checksumReceivedLen > 0 && shouldVerifyChecksum()) {
      try {
        verifyChunks(dataBuf, checksumBuf);
      } catch (IOException ioe) { // 校验异常. 则标记为checksum_error告知上游DN
        if (responder != null) {
          try {
          ((PacketResponder) responder.getRunnable()).enqueue(params.., ERROR_CHECKSUM);
          Thread.sleep(3000);
          } catch (InterruptedException e) {..}
        } // 转发之后然后自己抛出异常, 关闭自身的写数据线程
        throw new IOException("Terminating due to a checksum error." + ioe);
      }
      // 客户端和DN的校验方式不一样, 尝试转换, 暂略
      if (needsChecksumTranslation) translateChunks(dataBuf, checksumBuf);
    }

    // 校验和缺失, 需要重新计算它 (暂略)
    if (checksumReceivedLen == 0 && !streams.isTransientStorage()) {
      checksumBuf = ByteBuffer.allocate(checksumLen);
      diskChecksum.calculateChunkedSums(dataBuf, checksumBuf);
    }
    // 此时, buffer中的数据会使用磁盘的校验和
    final boolean shouldNotWriteChecksum = (checksumReceivedLen == 0 &&
                                            streams.isTransientStorage());
    
    // (可跳)写入磁盘前的一系列参数计算, 和判断.. 细节暂略 (边界判断挺复杂)
    try {
      long onDiskLen = replicaInfo.getBytesOnDisk();
      if (onDiskLen<offsetInBlock) {
        long partialChunkSizeOnDisk = onDiskLen % bytesPerChecksum;
        long lastChunkBoundary = onDiskLen - partialChunkSizeOnDisk;
        boolean alignedOnDisk = partialChunkSizeOnDisk == 0;
        boolean alignedInPacket = firstByteInBlock % bytesPerChecksum == 0;
        boolean overwriteLastCrc = !alignedOnDisk && !shouldNotWriteChecksum;
        boolean doCrcRecalc = overwriteLastCrc && (lastChunkBoundary != firstByteInBlock);
        if (!alignedInPacket && len > bytesPerChecksum)
          throw new IOException("Unexpected packet data length for block xx");

        // 如果当前block的最后一个chunk不足512字节, 则根据之前的chunk重新计算一次checksum
        // 如果client端已经提供了整个block的新checksum (多次写), 则DN不需要重新计算
        Checksum partialCrc = null;
        if (doCrcRecalc) {
          long offsetInChecksum = BlockMetadataHeader.getHeaderSize() +
                                  onDiskLen / bytesPerChecksum * checksumSize;
          partialCrc = computePartialChunkCrc(onDiskLen, offsetInChecksum);
        }

        int startByteToDisk = (int)(onDiskLen-firstByteInBlock) +
                              dataBuf.arrayOffset() + dataBuf.position();
        int numBytesToDisk = (int)(offsetInBlock-onDiskLen);

        final byte[] lastCrc;
        if (shouldNotWriteChecksum) {
          lastCrc = null;
        } else {
          int skip = 0;
          byte[] crcBytes = null;

          if (overwriteLastCrc) adjustCrcFilePosition(); // 准备覆盖最后chunk的checksum
            
          if (doCrcRecalc) { // 是否需要重新为最后chunk计算校验和
            int bytesToReadForRecalc = (int)(bytesPerChecksum - partialChunkSizeOnDisk);
            if (numBytesToDisk < bytesToReadForRecalc) bytesToReadForRecalc = numBytesToDisk;

            partialCrc.update(dataBuf.array(), startByteToDisk, bytesToReadForRecalc);
            byte[] buf = FSOutputSummer.convertToByteStream(partialCrc, checksumSize);
            crcBytes = copyLastChunkChecksum(buf, checksumSize, buf.length);
            checksumOut.write(buf);
            skip++;
          }
          long skippedDataBytes = lastChunkBoundary - firstByteInBlock;

          if (skippedDataBytes > 0) {
            skip += (int)(skippedDataBytes / bytesPerChecksum) +
                         ((skippedDataBytes % bytesPerChecksum == 0) ? 0 : 1);
          }
          skip *= checksumSize;

          // 写入剩余的checksum
          final int offset = checksumBuf.arrayOffset() + checksumBuf.position() + skip;
          final int end = offset + checksumLen - skip;
          if (offset >= end && doCrcRecalc) {
            lastCrc = crcBytes;
          } else {
            final int remainingBytes = checksumLen - skip;
            lastCrc = copyLastChunkChecksum(checksumBuf.array(), checksumSize, end);
            checksumOut.write(checksumBuf.array(), offset, remainingBytes); // 写校验和到buffer
          }
        }
          
        long begin = Time.monotonicNow();
        // 从buffer刷到OS缓存中, 然后刷盘 (写盘细节待补)
        streams.writeDataToDisk(dataBuf.array(), startByteToDisk, numBytesToDisk);
        long duration = Time.monotonicNow() - begin;
        if (duration > datanodeSlowLogThresholdMs) // 写盘的慢日志标记 (重要!!)
          LOG.warn("Slow BlockReceiver write data to disk cost: ms" + duration);
        if (duration > maxWriteToDiskMs) maxWriteToDiskMs = duration;

        // 这里真正把数据和校验和文件写入磁盘, 带sync参数则会确保一定落盘
        flushOrSync(syncBlock); // flushDataOut() 或 syncDataOut()
        
        replicaInfo.setLastChecksumAndDataLen(offsetInBlock, lastCrc);
        manageWriterOsCache(offsetInBlock); // 清除OS缓存, 待看
      }
    } catch (IOException iex) { // 直接抛出写盘异常的错误
      throw iex;
    }
  }

  // 当前DN为最后一个DN, 或客户端设置了sync标志位, 则此时(最后)才把packet放入ack队中. (在fsync完成后)
  if (responder != null && (syncBlock || shouldVerifyChecksum())) {
    ((PacketResponder) responder.getRunnable()).enqueue(params...);
  }

  // replaceBlock相关 (暂略)
  if (isReplaceBlock && (Time.monotonicNow() - lastResponseTime > responseInterval)) {
    BlockOpResponseProto.Builder response = BlockOpResponseProto.newBuilder().
                                            setStatus(Status.IN_PROGRESS);
    response.build().writeDelimitedTo(replyOut);
    replyOut.flush();
    lastResponseTime = Time.monotonicNow();
  }
  
  if (throttler != null) throttler.throttle(len); // 限流器相关, 暂略
  
  return lastPacketInBlock ? -1 : len; // 读到空尾包就返回-1 (停止)
}

0x04. 写块到磁盘

上面receivePacket()的最后, 会把数据+校验和写到DN的本地磁盘, 写完成后, 应该是如下的结构:

hdfsBlock00

写盘这里有许多OS层的相关调用, 细节比较复杂. 待补充….

0x05. PacketResponder线程

最后来看看PacketResponder线程, 它比Client的response线程做的事要多一些, 不过整体也比较直观, 如图所示:

stateDiagram
[*] --> 不是最后的空尾包

不是最后的空尾包 --> 非最后一个DN
非最后一个DN --> 读取下游DN的ack返回
读取下游DN的ack返回 --> 从ackQueue中取出packet
从ackQueue中取出packet --> 判断下游返回是否匹配
判断下游返回是否匹配 --> 若接受到空尾包_则完成当前block

不是最后的空尾包 --> 最后一个DN
最后一个DN --> 从ack队中取出packet
从ack队中取出packet --> 若接受到空尾包_则完成当前block
若接受到空尾包_则完成当前block --> 向上游节点发送响应ACK
向上游节点发送响应ACK --> 从ackQueue中移除当前packet
从ackQueue中移除当前packet --> 不是最后的空尾包: 循环判断
从ackQueue中移除当前packet --> [*]: 处理最后的空尾包

然后它有3个状态, 初始化线程时根据targetDN 数决定

  • NON_PIPELINE (下游节点为null?)
  • LAST_IN_PIPELINE (最后一个DN)
  • HAS_DOWNSTREAM_IN_PIPELINE (非最后一个DN)
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
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
   @Override
   public void run() {
     boolean lastPacketInBlock = false;
     final long startTime = ClientTraceLog.isInfoEnabled() ? System.nanoTime() : 0;
     while (isRunning() && !lastPacketInBlock) {
       long totalAckTimeNanos = 0;
       boolean isInterrupted = false;
       try {
         Packet pkt = null;
         long expected = -2;
         PipelineAck ack = new PipelineAck();
         long seqno = PipelineAck.UNKOWN_SEQNO; // 值为-2
         long ackRecvNanoTime = 0;
         try {
           if (type != PacketResponderType.LAST_IN_PIPELINE && !mirrorError) {
             // 从下游节点的返回读取ack信息
             ack.readFields(downstreamIn);
             ackRecvNanoTime = System.nanoTime();
             // 处理OOB_ACK, 暂略
             seqno = ack.getSeqno();
           }
           if (seqno != PipelineAck.UNKOWN_SEQNO || type == LAST_IN_PIPELINE) {
             pkt = waitForAckHead(seqno);
             if (!isRunning()) break;
               
             expected = pkt.seqno;
             
             if (type == HAS_DOWNSTREAM_IN_PIPELINE && seqno != expected)
               throw new IOException("seqno: expected=" + expected + ", received=" + seqno);
             
             if (type == PacketResponderType.HAS_DOWNSTREAM_IN_PIPELINE) {
               // Ack总时间包括下游DN的处理时间
               // 如果当前DN没有下游节点, 比如最后一个DN, 则此值为0
               totalAckTimeNanos = ackRecvNanoTime - pkt.ackEnqueueNanoTime;
               long ackTimeNanos = totalAckTimeNanos - ack.getDownstreamAckTimeNanos();
             }
             lastPacketInBlock = pkt.lastPacketInBlock;
           }
         } catch (InterruptedException ine) {
           isInterrupted = true;
         } catch (IOException ioe) {
           // 这里抛出异常时线程是否中断的考量比较复杂, 需要仔细看看(待补)
           if (Thread.interrupted()) {
             isInterrupted = true;
           } else if (ioe instanceof EOFException && !packetSentInTime()) {
             LOG.warn("The downstream error might be due to congestion in " +
                      "upstream including this node. Propagating the error: xx");
             throw ioe;
           } else {
             // 标记错误返回, 但是不中断当前response线程.
             mirrorError = true;
             LOG.info(myString, ioe);
           }
         }

         if (Thread.interrupted() || isInterrupted) {
           LOG.info(myString + ": Thread is interrupted.");
           running = false;
           continue;
         }

         // 如果是最后一个Packet, 则使block变为Finalized状态, 然后关闭块文件
         if (lastPacketInBlock) finalizeBlock(startTime);

         Status myStatus = pkt != null ? pkt.ackStatus : Status.SUCCESS;
         // 发送处理结果ack返回给上游节点
         sendAckUpstream(ack, expected, totalAckTimeNanos,
                         (pkt != null ? pkt.offsetInBlock : 0),
                         PipelineAck.combineHeader(datanode.getECN(), myStatus));
         // 从ackQueue中移除
         if (pkt != null) removeAckHead();
           
       } catch (IOException e) {
         LOG.warn("IOException in BlockReceiver.run(): ", e);
         if (running) {
           running = false;
           if (!Thread.interrupted()) receiverThread.interrupt();
         }
       } catch (Throwable e) {
         if (running) {
           running = false;
           receiverThread.interrupt();
         }
}}}

其中finalizedBlock() 的主要执行过程是在BPOfferService中调用增量块汇报实现的, 它由一个DN的核心守护线程负责, 在这里就不详说了.

至此, 整个写入流程在Datanode端的整体步骤就讲完了, 其中遗留了不少特殊情况的注释, 之后会一一再来细看.

未完待续…..

0x06. 遗留问题

主要是两个问题: (代码中已标记部分)

  1. 写DN中response线程出现异常, 和它是否中断的考量 (并发体系)
  2. DN写数据到磁盘的过程的问题:
    • 写失败会马上感知么?
    • 写的很慢会怎么样? 卡住, response线程还会转发ACK么, 如果转发, 那当前DN一直写卡住么?
    • 慢盘/慢DN问题是个重要的点, 之后单独来说….

参考资料:

  1. HDFS3.1.2官方源码
  2. Hadoop 2.X HDFS源码剖析Chapter4.x图
  3. HDFS1.X官方源码(对比高版用)