Hadoop3.2.1 【 HDFS 】源码分析 : 写数据

川长思鸟来 2022-12-23 06:52 182阅读 0赞

# 一. 前言 #

DataTransferProtocol.write()方法给出了写操作的接口定义， 操作码是80, DataXceiver.writeBlock()则真正实现了DataTransferProtocol.writeBlock()方法。

HDFS使用数据流管道方式来写数据。 DFSClient通过调用Sender.writeBlock()方法触发一个写数据块请求， 这个请求会传送到数据流管道中的每一个数据节点， 数据流管道中的最后一个数据节点会回复请求确认， 这个确认消息逆向地通过数据流管道送回DFSClient。 DFSClient收到请求确认后， 将要写入的数据块切分成若干个数据包（packet） ， 然后依次向数据流管道中发送这些数据包。 数据包会首先从DFSClient发送到数据流管道中的第一个数据节点（这里是Datanode1） ， Datanode1成功接收数据包后， 会将数据包写入磁盘， 然后将数据包发送到数据流管道中的第二个节点（Datanode2） 。 依此类推， 当数据包到达数据流管道中的最后一个节点（Datanode3）时， Datanode3会对收到的数据包进行校验， 如果校验成功， Datanode3会发送数据包确认消息， 这个确认消息会逆向地通过数据流管道送回DFSClient。 当一个数据块中的所有数据包都成功发送完毕， 并且收到确认消息后， DFSClient会发送一个空数据包标识当前数据块发送完毕。 至此， 整个数据块发送流程结束。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]

这个写数据块请求通过流式接口到达Datanode之后， Datanode上监听流式接口请求的DataXceiverServer会接收这个请求， 并构造一个DataXceiver对象， 然后在DataXceiver对象上调用DataXceiver.writeBlock()方法响应这个请求。 当前Datanode的DataXceiver.writeBlock()方法会级联向数据流管道中的下一个Datanode发送写数据块请求， 这个流式请求会一直在数据流管道中传递下去， 直到写数据块请求到达数据流管道中的最后一个Datanode。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]

# 二. DataXceiver.writeBlock() #

DataXceiver.writeBlock()方法实现了向DataNode写数据的所有逻辑

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]

writeBlock()方法的开始部分定义了三个boolean类型的变量， 用于控制处理流程。

//isDatanode变量指示当前写操作是否是DFSClient发起的 : false
        final boolean isDatanode = clientname.length() == 0;
    
        //isClient变量与isDatanode相反， 表示是Datanode触发的写操作 true
        final boolean isClient = !isDatanode;
    
        //isTransfer变量指示当前的写操作是否为数据块复制操作， 利用数据流管道状态来判断 false
        final boolean isTransfer = stage == BlockConstructionStage.TRANSFER_RBW
            || stage == BlockConstructionStage.TRANSFER_FINALIZED;
    
        // 是否开启持久化 false
        allowLazyPersist = allowLazyPersist &&
            (dnConf.getAllowNonLocalLazyPersist() || peer.isLocal());

在writeBlock()方法中还使用到了两组输入／ 输出流

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 3]  
Datanode与数据流管道中的上游节点通信用到了输入流in以及输出流replyOut，  
与数据流管道中的下游节点通信则用到了输入流mirrorIn以及输出流mirrorOut。

// 创建replyOut输出流
        // reply to upstream datanode or client 
        final DataOutputStream replyOut = getBufferedOutputStream();
    
    	
    	    // 到下游数据节点的输出流
        // stream to next target
        DataOutputStream mirrorOut = null;
    
        //下游数据节点的输入流
        // reply from next target
        DataInputStream mirrorIn = null;
    
        //到下游节点的Socket
        // socket to next target
        Socket mirrorSock = null;
    
        //下游节点的名称： 端口
        // the name:port of next target
        String mirrorNode = null;
    
        // 数据流管道中第一个失败的Datanode
        // first datanode that failed in connection setup
        String firstBadLink = "";
    
        Status mirrorInStatus = SUCCESS;
    
        //保存这个数据块的Datanode存储的id
        final String storageUuid;
    
        final boolean isOnTransientStorage;
    
        try { 
    
          final Replica replica;
    
          if (isDatanode ||  stage != BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) { 
    
            //打开一个BlockReceiver， 用于从上游节点接收数据块
            // open a block receiver
            setCurrentBlockReceiver(getBlockReceiver(block, storageType, in,
                peer.getRemoteAddressString(),
                peer.getLocalAddressString(),
                stage, latestGenerationStamp, minBytesRcvd, maxBytesRcvd,
                clientname, srcDataNode, datanode, requestedChecksum,
                cachingStrategy, allowLazyPersist, pinning, storageId));
    
            replica = blockReceiver.getReplica();
    
          } else { 
    
            replica = datanode.data.recoverClose(  block, latestGenerationStamp, minBytesRcvd);
    
          }
    
          storageUuid = replica.getStorageUuid();
    
          isOnTransientStorage = replica.isOnTransientStorage();
    
          // 连接到下游节点
          // Connect to downstream machine, if appropriate
          //
          if (targets.length > 0) { 
    
            InetSocketAddress mirrorTarget = null;
    
            // Connect to backup machine
            mirrorNode = targets[0].getXferAddr(connectToDnViaHostname);
    
            LOG.debug("Connecting to datanode {}", mirrorNode);
    
            mirrorTarget = NetUtils.createSocketAddr(mirrorNode);
    
            mirrorSock = datanode.newSocket();
            try { 
    
              DataNodeFaultInjector.get().failMirrorConnection();
    
              int timeoutValue = dnConf.socketTimeout + (HdfsConstants.READ_TIMEOUT_EXTENSION * targets.length);
    
              int writeTimeout = dnConf.socketWriteTimeout +  (HdfsConstants.WRITE_TIMEOUT_EXTENSION * targets.length);
    
              // 建立到下游节点的Socket连接
              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);
    
    
              SecretKey secretKey = null;
    
              if (dnConf.overwriteDownstreamDerivedQOP) { 
    
                String bpid = block.getBlockPoolId();
    
                BlockKey blockKey = datanode.blockPoolTokenSecretManager.get(bpid).getCurrentKey();
    
                secretKey = blockKey.getKey();
              }
    
              IOStreamPair saslStreams = datanode.saslClient.socketSend( mirrorSock, unbufMirrorOut, unbufMirrorIn, keyFactory, blockToken, targets[0], secretKey);
    
              unbufMirrorOut = saslStreams.out;
    
              unbufMirrorIn = saslStreams.in;
    
              //创建mirrorOut和mirrorIn建立到下游节点的输出流以及输入流
              mirrorOut = new DataOutputStream(new BufferedOutputStream(unbufMirrorOut,  smallBufferSize));
    
              mirrorIn = new DataInputStream(unbufMirrorIn);
    
              String targetStorageId = null;
    
    
              if (targetStorageIds.length > 0) { 
    
                // Older clients may not have provided any targetStorageIds
                targetStorageId = targetStorageIds[0];
              }
              if (targetPinnings != null && targetPinnings.length > 0) { 
    
                //向下游节点发送数据块写入请求
                new Sender(mirrorOut).writeBlock(originalBlock, targetStorageTypes[0],
                    blockToken, clientname, targets, targetStorageTypes,
                    srcDataNode, stage, pipelineSize, minBytesRcvd, maxBytesRcvd,
                    latestGenerationStamp, requestedChecksum, cachingStrategy,
                    allowLazyPersist, targetPinnings[0], targetPinnings,
                    targetStorageId, targetStorageIds);
    
    
              } else { 
    
                //接收来自下游节点的请求确认， 并记录请求确认状态
                new Sender(mirrorOut).writeBlock(originalBlock, targetStorageTypes[0],
                    blockToken, clientname, targets, targetStorageTypes,
                    srcDataNode, stage, pipelineSize, minBytesRcvd, maxBytesRcvd,
                    latestGenerationStamp, requestedChecksum, cachingStrategy,
                    allowLazyPersist, false, targetPinnings,
                    targetStorageId, targetStorageIds);
    
              }
    
              mirrorOut.flush();
    
              DataNodeFaultInjector.get().writeBlockAfterFlush();
    
              // read connect ack (only for clients, not for replication req)
              if (isClient) { 
                //接收来自下游节点的请求确认， 并记录请求确认状态
                BlockOpResponseProto connectAck =  BlockOpResponseProto.parseFrom(PBHelperClient.vintPrefixed(mirrorIn));
    
    
                mirrorInStatus = connectAck.getStatus();
    
                firstBadLink = connectAck.getFirstBadLink();
    
    
                if (mirrorInStatus != SUCCESS) { 
                  LOG.debug("Datanode {} got response for connect" +
                      "ack from downstream datanode with firstbadlink as {}",
                      targets.length, firstBadLink);
                }
              }
    
            } catch (IOException e) { 
              //出现异常， 向上游节点发送异常响应
              if (isClient) { 
                BlockOpResponseProto.newBuilder()
                  .setStatus(ERROR)
                   // NB: Unconditionally using the xfer addr w/o hostname
                  .setFirstBadLink(targets[0].getXferAddr())
                  .build()
                  .writeDelimitedTo(replyOut);
                replyOut.flush();
              }
    
              // 关闭到下游节点的Socket、 输入流以及输出流
              IOUtils.closeStream(mirrorOut);
              mirrorOut = null;
              IOUtils.closeStream(mirrorIn);
              mirrorIn = null;
              IOUtils.closeSocket(mirrorSock);
              mirrorSock = null;
              if (isClient) { 
                LOG.error("{}:Exception transfering block {} to mirror {}",
                    datanode, block, mirrorNode, e);
                throw e;
              } else { 
                LOG.info("{}:Exception transfering {} to mirror {}- continuing " +
                    "without the mirror", datanode, block, mirrorNode, e);
                incrDatanodeNetworkErrors();
              }
            }
          }
    
          // send connect-ack to source for clients and not transfer-RBW/Finalized
          if (isClient && !isTransfer) { 
    
            if (mirrorInStatus != SUCCESS) { 
              LOG.debug("Datanode {} forwarding connect ack to upstream " +
                  "firstbadlink is {}", targets.length, firstBadLink);
            }
    
            // 向上游节点返回请求确认
            BlockOpResponseProto.newBuilder()
              .setStatus(mirrorInStatus)
              .setFirstBadLink(firstBadLink)
              .build()
              .writeDelimitedTo(replyOut);
    
    
            replyOut.flush();
          }

成功建立了与下游节点的输入／ 输出流后， writeBlock()方法就会调用blockReceiver.receiveBlock()方法从上游节点接收数据块， 然后将数据块发送到下游节点。同时blockReceiver对象还会从下游节点接收数据块中数据包的确认消息， 并且将这个确认消息转发到上游节点。

// receive the block and mirror to the next target
          if (blockReceiver != null) { 
    
            String mirrorAddr = (mirrorSock == null) ? null : mirrorNode;
    
            // 调用BlockReceiver.receiveBlock()从上游节点接收数据块， 然后将数据块发送到下游节点
            blockReceiver.receiveBlock(mirrorOut, mirrorIn, replyOut,  mirrorAddr, null, targets, false);
    
    
            // 对于复制操作， 不需要向下游节点转发数据块， 也不需要接收下游节点的确认
            // 所以成功接收完数据块之后， 在当前节点直接返回确认消息
            // send close-ack for transfer-RBW/Finalized 
            if (isTransfer) { 
              LOG.trace("TRANSFER: send close-ack");
              writeResponse(SUCCESS, null, replyOut);
            }
    
          }

成功执行了BlockReceiver.receiveBlock()之后， writeBlock()方法就会更新当前数据节点上新写入数据块副本的时间戳、 副本文件长度等信息。 如果是数据流管道关闭的恢复操作或者是数据块的复制操作， 则调用Datanode.closeBlock()方法向Namenode汇报Datanode  
接收了新的数据块， Datanode.closeBlock()调用BPOfferService.notifyNamenodeReceivedBlock()通知Namenode。 对于客户端发起的写数据请求， 在PacketResponder线程中已经通过调用Datanode.closeBlock()方法关闭了数据块

// update its generation stamp
          if (isClient &&   stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) { 
    
            block.setGenerationStamp(latestGenerationStamp);
    
            block.setNumBytes(minBytesRcvd);
          }
          
          // if this write is for a replication request or recovering
          // a failed close for client, then confirm block. For other client-writes,
          // the block is finalized in the PacketResponder.
          if (isDatanode || stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) { 
    
            datanode.closeBlock(block, null, storageUuid, isOnTransientStorage);
    
            LOG.info("Received {} src: {} dest: {} of size {}", block, remoteAddress, localAddress, block.getNumBytes());
          }
    
          if(isClient) { 
    
            size = block.getNumBytes();
          }

DataXceiver.writeBlock()完成上述操作后， 会在finally中关闭上下游节点的输入／ 输出流， 同时关闭blockReceiver对象。

finally { 
          // 关闭上下游节点的输入／ 输 出流， 同时关闭blockReceiver对象
          // close all opened streams
          IOUtils.closeStream(mirrorOut);
          IOUtils.closeStream(mirrorIn);
          IOUtils.closeStream(replyOut);
          IOUtils.closeSocket(mirrorSock);
          IOUtils.closeStream(blockReceiver);
          setCurrentBlockReceiver(null);
        }

完整代码如下:

@Override
      public void writeBlock(final ExtendedBlock block,
          final StorageType storageType, 
          final Token<BlockTokenIdentifier> blockToken,
          final String clientname,
          final DatanodeInfo[] targets,
          final StorageType[] targetStorageTypes,
          final DatanodeInfo srcDataNode,
          final BlockConstructionStage stage,
          final int pipelineSize,
          final long minBytesRcvd,
          final long maxBytesRcvd,
          final long latestGenerationStamp,
          DataChecksum requestedChecksum,
          CachingStrategy cachingStrategy,
          boolean allowLazyPersist,
          final boolean pinning,
          final boolean[] targetPinnings,
          final String storageId,
          final String[] targetStorageIds) throws IOException { 
        previousOpClientName = clientname;
        updateCurrentThreadName("Receiving block " + block);
    
        //isDatanode变量指示当前写操作是否是DFSClient发起的 : false
        final boolean isDatanode = clientname.length() == 0;
    
        //isClient变量与isDatanode相反， 表示是Datanode触发的写操作 true
        final boolean isClient = !isDatanode;
    
        //isTransfer变量指示当前的写操作是否为数据块复制操作， 利用数据流管道状态来判断 false
        final boolean isTransfer = stage == BlockConstructionStage.TRANSFER_RBW
            || stage == BlockConstructionStage.TRANSFER_FINALIZED;
    
        // 是否开启持久化 false
        allowLazyPersist = allowLazyPersist &&
            (dnConf.getAllowNonLocalLazyPersist() || peer.isLocal());
    
        long size = 0;
    
        // 创建replyOut输出流
        // reply to upstream datanode or client 
        final DataOutputStream replyOut = getBufferedOutputStream();
    
        int nst = targetStorageTypes.length;
    
        StorageType[] storageTypes = new StorageType[nst + 1];
    
        storageTypes[0] = storageType;
    
        if (targetStorageTypes.length > 0) { 
    
          System.arraycopy(targetStorageTypes, 0, storageTypes, 1, nst);
        }
    
        // To support older clients, we don't pass in empty 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];
        }
    
        checkAccess(replyOut, isClient, block, blockToken, Op.WRITE_BLOCK,
            BlockTokenIdentifier.AccessMode.WRITE,
            storageTypes, storageIds);
    
    
        // check single target for transfer-RBW/Finalized
        if (isTransfer && targets.length > 0) { 
          throw new IOException(stage + " does not support multiple targets "
              + Arrays.asList(targets));
        }
    
        if (LOG.isDebugEnabled()) { 
    
          // opWriteBlock: stage=PIPELINE_SETUP_CREATE, clientname=DFSClient_NONMAPREDUCE_1654943232_1
          // block =BP-451827885-192.168.8.156-1584099133244:blk_1073746922_6098, newGs=0, bytesRcvd=[0, 0]
          // targets=[]; pipelineSize=1, srcDataNode=:0, pinning=false
    
    
    
          LOG.debug("opWriteBlock: stage={}, clientname={}\n " +
                  "block ={}, newGs={}, bytesRcvd=[{}, {}]\n " +
                  "targets={}; pipelineSize={}, srcDataNode={}, pinning={}",
              stage, clientname, block, latestGenerationStamp, minBytesRcvd,
              maxBytesRcvd, Arrays.asList(targets), pipelineSize, srcDataNode,
              pinning);
    
          // isDatanode=false, isClient=true, isTransfer=false
          LOG.debug("isDatanode={}, isClient={}, isTransfer={}",
              isDatanode, isClient, isTransfer);
    
          // writeBlock receive buf size 408118 tcp no delay true
          LOG.debug("writeBlock receive buf size {} tcp no delay {}",
              peer.getReceiveBufferSize(), peer.getTcpNoDelay());
        }
    
    
    
        // We later mutate block's generation stamp and length, but we need to
        // forward the original version of the block to downstream mirrors, so
        // make a copy here.
        final ExtendedBlock originalBlock = new ExtendedBlock(block);
    
        if (block.getNumBytes() == 0) { 
    
          block.setNumBytes(dataXceiverServer.estimateBlockSize);
    
        }
        LOG.info("Receiving {} src: {} dest: {}",
            block, remoteAddress, localAddress);
    
        // 到下游数据节点的输出流
        // stream to next target
        DataOutputStream mirrorOut = null;
    
        //下游数据节点的输入流
        // reply from next target
        DataInputStream mirrorIn = null;
    
        //到下游节点的Socket
        // socket to next target
        Socket mirrorSock = null;
    
        //下游节点的名称： 端口
        // the name:port of next target
        String mirrorNode = null;
    
        // 数据流管道中第一个失败的Datanode
        // first datanode that failed in connection setup
        String firstBadLink = "";
    
        Status mirrorInStatus = SUCCESS;
    
        //保存这个数据块的Datanode存储的id
        final String storageUuid;
    
        final boolean isOnTransientStorage;
    
        try { 
    
          final Replica replica;
    
          if (isDatanode ||  stage != BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) { 
    
            //打开一个BlockReceiver， 用于从上游节点接收数据块
            // open a block receiver
            setCurrentBlockReceiver(getBlockReceiver(block, storageType, in,
                peer.getRemoteAddressString(),
                peer.getLocalAddressString(),
                stage, latestGenerationStamp, minBytesRcvd, maxBytesRcvd,
                clientname, srcDataNode, datanode, requestedChecksum,
                cachingStrategy, allowLazyPersist, pinning, storageId));
    
            replica = blockReceiver.getReplica();
    
          } else { 
    
            replica = datanode.data.recoverClose(  block, latestGenerationStamp, minBytesRcvd);
    
          }
    
          storageUuid = replica.getStorageUuid();
    
          isOnTransientStorage = replica.isOnTransientStorage();
    
          // 连接到下游节点
          // Connect to downstream machine, if appropriate
          //
          if (targets.length > 0) { 
    
            InetSocketAddress mirrorTarget = null;
    
            // Connect to backup machine
            mirrorNode = targets[0].getXferAddr(connectToDnViaHostname);
    
            LOG.debug("Connecting to datanode {}", mirrorNode);
    
            mirrorTarget = NetUtils.createSocketAddr(mirrorNode);
    
            mirrorSock = datanode.newSocket();
            try { 
    
              DataNodeFaultInjector.get().failMirrorConnection();
    
              int timeoutValue = dnConf.socketTimeout + (HdfsConstants.READ_TIMEOUT_EXTENSION * targets.length);
    
              int writeTimeout = dnConf.socketWriteTimeout +  (HdfsConstants.WRITE_TIMEOUT_EXTENSION * targets.length);
    
              // 建立到下游节点的Socket连接
              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);
    
    
              SecretKey secretKey = null;
    
              if (dnConf.overwriteDownstreamDerivedQOP) { 
    
                String bpid = block.getBlockPoolId();
    
                BlockKey blockKey = datanode.blockPoolTokenSecretManager.get(bpid).getCurrentKey();
    
                secretKey = blockKey.getKey();
              }
    
              IOStreamPair saslStreams = datanode.saslClient.socketSend( mirrorSock, unbufMirrorOut, unbufMirrorIn, keyFactory, blockToken, targets[0], secretKey);
    
              unbufMirrorOut = saslStreams.out;
    
              unbufMirrorIn = saslStreams.in;
    
              //创建mirrorOut和mirrorIn建立到下游节点的输出流以及输入流
              mirrorOut = new DataOutputStream(new BufferedOutputStream(unbufMirrorOut,  smallBufferSize));
    
              mirrorIn = new DataInputStream(unbufMirrorIn);
    
              String targetStorageId = null;
    
    
              if (targetStorageIds.length > 0) { 
    
                // Older clients may not have provided any targetStorageIds
                targetStorageId = targetStorageIds[0];
              }
              if (targetPinnings != null && targetPinnings.length > 0) { 
    
                //向下游节点发送数据块写入请求
                new Sender(mirrorOut).writeBlock(originalBlock, targetStorageTypes[0],
                    blockToken, clientname, targets, targetStorageTypes,
                    srcDataNode, stage, pipelineSize, minBytesRcvd, maxBytesRcvd,
                    latestGenerationStamp, requestedChecksum, cachingStrategy,
                    allowLazyPersist, targetPinnings[0], targetPinnings,
                    targetStorageId, targetStorageIds);
    
    
              } else { 
    
                //接收来自下游节点的请求确认， 并记录请求确认状态
                new Sender(mirrorOut).writeBlock(originalBlock, targetStorageTypes[0],
                    blockToken, clientname, targets, targetStorageTypes,
                    srcDataNode, stage, pipelineSize, minBytesRcvd, maxBytesRcvd,
                    latestGenerationStamp, requestedChecksum, cachingStrategy,
                    allowLazyPersist, false, targetPinnings,
                    targetStorageId, targetStorageIds);
    
              }
    
              mirrorOut.flush();
    
              DataNodeFaultInjector.get().writeBlockAfterFlush();
    
              // read connect ack (only for clients, not for replication req)
              if (isClient) { 
                //接收来自下游节点的请求确认， 并记录请求确认状态
                BlockOpResponseProto connectAck =  BlockOpResponseProto.parseFrom(PBHelperClient.vintPrefixed(mirrorIn));
    
    
                mirrorInStatus = connectAck.getStatus();
    
                firstBadLink = connectAck.getFirstBadLink();
    
    
                if (mirrorInStatus != SUCCESS) { 
                  LOG.debug("Datanode {} got response for connect" +
                      "ack from downstream datanode with firstbadlink as {}",
                      targets.length, firstBadLink);
                }
              }
    
            } catch (IOException e) { 
              //出现异常， 向上游节点发送异常响应
              if (isClient) { 
                BlockOpResponseProto.newBuilder()
                  .setStatus(ERROR)
                   // NB: Unconditionally using the xfer addr w/o hostname
                  .setFirstBadLink(targets[0].getXferAddr())
                  .build()
                  .writeDelimitedTo(replyOut);
                replyOut.flush();
              }
    
              // 关闭到下游节点的Socket、 输入流以及输出流
              IOUtils.closeStream(mirrorOut);
              mirrorOut = null;
              IOUtils.closeStream(mirrorIn);
              mirrorIn = null;
              IOUtils.closeSocket(mirrorSock);
              mirrorSock = null;
              if (isClient) { 
                LOG.error("{}:Exception transfering block {} to mirror {}",
                    datanode, block, mirrorNode, e);
                throw e;
              } else { 
                LOG.info("{}:Exception transfering {} to mirror {}- continuing " +
                    "without the mirror", datanode, block, mirrorNode, e);
                incrDatanodeNetworkErrors();
              }
            }
          }
    
          // send connect-ack to source for clients and not transfer-RBW/Finalized
          if (isClient && !isTransfer) { 
    
            if (mirrorInStatus != SUCCESS) { 
              LOG.debug("Datanode {} forwarding connect ack to upstream " +
                  "firstbadlink is {}", targets.length, firstBadLink);
            }
    
            // 向上游节点返回请求确认
            BlockOpResponseProto.newBuilder()
              .setStatus(mirrorInStatus)
              .setFirstBadLink(firstBadLink)
              .build()
              .writeDelimitedTo(replyOut);
    
    
            replyOut.flush();
          }
    
          // receive the block and mirror to the next target
          if (blockReceiver != null) { 
    
            String mirrorAddr = (mirrorSock == null) ? null : mirrorNode;
    
            // 调用BlockReceiver.receiveBlock()从上游节点接收数据块， 然后将数据块发送到下游节点
            blockReceiver.receiveBlock(mirrorOut, mirrorIn, replyOut,  mirrorAddr, null, targets, false);
    
    
            // 对于复制操作， 不需要向下游节点转发数据块， 也不需要接收下游节点的确认
            // 所以成功接收完数据块之后， 在当前节点直接返回确认消息
            // send close-ack for transfer-RBW/Finalized 
            if (isTransfer) { 
              LOG.trace("TRANSFER: send close-ack");
              writeResponse(SUCCESS, null, replyOut);
            }
    
          }
    
          // update its generation stamp
          if (isClient &&   stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) { 
    
            block.setGenerationStamp(latestGenerationStamp);
    
            block.setNumBytes(minBytesRcvd);
          }
          
          // if this write is for a replication request or recovering
          // a failed close for client, then confirm block. For other client-writes,
          // the block is finalized in the PacketResponder.
          if (isDatanode || stage == BlockConstructionStage.PIPELINE_CLOSE_RECOVERY) { 
    
            datanode.closeBlock(block, null, storageUuid, isOnTransientStorage);
    
            LOG.info("Received {} src: {} dest: {} of size {}", block, remoteAddress, localAddress, block.getNumBytes());
          }
    
          if(isClient) { 
    
            size = block.getNumBytes();
          }
        } catch (IOException ioe) { 
          LOG.info("opWriteBlock {} received exception {}",
              block, ioe.toString());
          incrDatanodeNetworkErrors();
          throw ioe;
        } finally { 
          // 关闭上下游节点的输入／ 输 出流， 同时关闭blockReceiver对象
          // close all opened streams
          IOUtils.closeStream(mirrorOut);
          IOUtils.closeStream(mirrorIn);
          IOUtils.closeStream(replyOut);
          IOUtils.closeSocket(mirrorSock);
          IOUtils.closeStream(blockReceiver);
          setCurrentBlockReceiver(null);
        }
    
        //update metrics
        datanode.getMetrics().addWriteBlockOp(elapsed());
    
        datanode.getMetrics().incrWritesFromClient(peer.isLocal(), size);
      }

# 三. BlockReceiver #

BlockReceiver类负责从数据流管道中的上游节点接收数据块， 然后保存数据块到当前数据节点的存储中， 再将数据块转发到数据流管道中的下游节点。 同时BlockReceiver还会接收来自下游节点的响应， 并把这个响应发送给数据流管道中的上游节点。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 4]

BlockReceiver.receiveBlock()方法的逻辑比较简单， 它先启动PacketResponder线程负责接收并转发下游节点发送的确认数据包的ACK消息。 之后receiveBlock()方法循环调用receivePacket()方法接收上游写入的数据包并转发这个数据包到下游节点。 成功完成整个数据块的写入操作后， receiveBlock()方法关闭PacketResponder线程

void receiveBlock(
          DataOutputStream mirrOut, // output to next datanode  : null
          DataInputStream mirrIn,   // input from next datanode  : null
          DataOutputStream replyOut,  // output to previous datanode : DataOutputStream
          String mirrAddr, DataTransferThrottler throttlerArg,  // null
          DatanodeInfo[] downstreams, // DatanodeInfo : null
          boolean isReplaceBlock) throws IOException {   // isReplaceBlock false
    
        // false
        syncOnClose = datanode.getDnConf().syncOnClose;
        // false
        dirSyncOnFinalize = syncOnClose;
    
        boolean responderClosed = false;
        // null
        mirrorOut = mirrOut;
        // null
        mirrorAddr = mirrAddr;
    
        initPerfMonitoring(downstreams);
        // null
        throttler = throttlerArg;
        // DataOutputStream
        this.replyOut = replyOut;
        // false
        this.isReplaceBlock = isReplaceBlock;
    
        try { 
    
          if (isClient && !isTransfer) { 
            // 构建 PacketResponder
            // 启动PacketResponder线程处理确认包的接收和转发
            responder = new Daemon(datanode.threadGroup,  new PacketResponder(replyOut, mirrIn, downstreams));
    
    
            responder.start(); // start thread to processes responses
          }
    
          //  循环调用receivePacket()接收并转发数据块中的所有数据包
          while (receivePacket() >= 0) {  /* Receive until the last packet */ }
    
          // wait for all outstanding packet responses. And then
          // indicate responder to gracefully shutdown.
          // Mark that responder has been closed for future processing
          if (responder != null) { 
    
            //完成数据块的写入操作后， 结束PacketResponder线程
            ((PacketResponder)responder.getRunnable()).close();
    
            responderClosed = true;
          }
    
          // If this write is for a replication or transfer-RBW/Finalized,
          // then finalize block or convert temporary to RBW.
          // For client-writes, the block is finalized in the PacketResponder.
          if (isDatanode || isTransfer) { 
    
            // Hold a volume reference to finalize block.
            try (ReplicaHandler handler = claimReplicaHandler()) { 
    
    
              // close the block/crc files
              close();
              block.setNumBytes(replicaInfo.getNumBytes());
    
              if (stage == BlockConstructionStage.TRANSFER_RBW) { 
    
    
                // for TRANSFER_RBW, convert temporary to RBW
                datanode.data.convertTemporaryToRbw(block);
              } else { 
                // for isDatnode or TRANSFER_FINALIZED Finalize the block.
    
    
                datanode.data.finalizeBlock(block, dirSyncOnFinalize);
              }
            }
    
            datanode.metrics.incrBlocksWritten();
          }
    
        } catch (IOException ioe) { 
          replicaInfo.releaseAllBytesReserved();
          if (datanode.isRestarting()) { 
            // Do not throw if shutting down for restart. Otherwise, it will cause
            // premature termination of responder.
            LOG.info("Shutting down for restart (" + block + ").");
          } else { 
            LOG.info("Exception for " + block, ioe);
            throw ioe;
          }
        } finally { 
          // Clear the previous interrupt state of this thread.
          Thread.interrupted();
    
          // If a shutdown for restart was initiated, upstream needs to be notified.
          // There is no need to do anything special if the responder was closed
          // normally.
          if (!responderClosed) {  // Data transfer was not complete.
            if (responder != null) { 
              // In case this datanode is shutting down for quick restart,
              // send a special ack upstream.
              if (datanode.isRestarting() && isClient && !isTransfer) { 
                try (Writer out = new OutputStreamWriter(
                    replicaInfo.createRestartMetaStream(), "UTF-8")) { 
                  // write out the current time.
                  out.write(Long.toString(Time.now() + restartBudget));
                  out.flush();
                } catch (IOException ioe) { 
                  // The worst case is not recovering this RBW replica. 
                  // Client will fall back to regular pipeline recovery.
                } finally { 
                  IOUtils.closeStream(streams.getDataOut());
                }
                try {               
                  // Even if the connection is closed after the ack packet is
                  // flushed, the client can react to the connection closure 
                  // first. Insert a delay to lower the chance of client 
                  // missing the OOB ack.
                  Thread.sleep(1000);
                } catch (InterruptedException ie) { 
                  // It is already going down. Ignore this.
                }
              }
              responder.interrupt();
            }
            IOUtils.closeStream(this);
            cleanupBlock();
          }
          if (responder != null) { 
            try { 
              responder.interrupt();
              // join() on the responder should timeout a bit earlier than the
              // configured deadline. Otherwise, the join() on this thread will
              // likely timeout as well.
              long joinTimeout = datanode.getDnConf().getXceiverStopTimeout();
              joinTimeout = joinTimeout > 1  ? joinTimeout*8/10 : joinTimeout;
              responder.join(joinTimeout);
              if (responder.isAlive()) { 
                String msg = "Join on responder thread " + responder
                    + " timed out";
                LOG.warn(msg + "\n" + StringUtils.getStackTrace(responder));
                throw new IOException(msg);
              }
            } catch (InterruptedException e) { 
              responder.interrupt();
              // do not throw if shutting down for restart.
              if (!datanode.isRestarting()) { 
                throw new IOException("Interrupted receiveBlock");
              }
            }
            responder = null;
          }
        }
      }

receiveBlock()方法将大部分处理工作都交给了receivePacket()方法

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 5]

receivePacket()方法首先调用packetReceiver.receiveNextPacket()方法从输入流中读入一个数据包（packet） ， 并将这个数据包放入ByteBuffer缓冲区curPacketBuf中。

readNextPacket()方法就是按照数据包格式从输入流中读取数据并放入指定的ByteBuffer缓冲区中。

receivePacket()成功接收数据包后， 会判断当前节点是否是数据流管道中的最后一个节点， 或者是输入流启动了sync标识（syncBlock） 要求Datanode立即将数据包同步到磁盘。 在这两种情况下， Datanode会先将数据写入磁盘， 然后再通知PacketResponder处理确认（ACK） 消息； 否则， receivePacket()方法接收完数据包后会立即通知PacketResponder处理确认消息。

接下来receivePacket()会将数据包发送给数据流管道中的下游节点， 然后就可以将数据块文件和校验文件写入数据节点的磁盘了。 写入磁盘之后， receivePacket()方法需要调用flushOrSync()方法将输出流缓存中的数据全部同步到磁盘， 最后还需要调用manageWriterOsCache（清理） 操作系统缓存中的数据。 需要注意的是， 如果当前节点是数据流管道中的最后一个节点， 则在写入磁盘前， 需要先对数据块中的所有数据包进行校验.  
由于在Datanode是数据流管道中最后一个节点， 以及携带了sync标识两种情况下，receivePacket()方法并没有通知PacketResponder处理响应消息， 所以在receiverPacket()方法的最后通知PacketResponder为这两种情况处理响应。

/** 
       * Receives and processes a packet. It can contain many chunks.
       * returns the number of data bytes that the packet has.
       */
      private int receivePacket() throws IOException { 
        // read the next packet
        // 调用packetReceiver.receiveNextPacket()方法从输入流中读入一个数据包
        packetReceiver.receiveNextPacket(in);
    
        PacketHeader header = packetReceiver.getHeader();
        if (LOG.isDebugEnabled()){ 
          LOG.debug("Receiving one packet for block " + block +
                    ": " + header);
        }
    
        // Sanity check the header
        if (header.getOffsetInBlock() > replicaInfo.getNumBytes()) { 
          throw new IOException("Received an out-of-sequence packet for " + block + 
              "from " + inAddr + " at offset " + header.getOffsetInBlock() +
              ". Expecting packet starting at " + replicaInfo.getNumBytes());
        }
        if (header.getDataLen() < 0) { 
          throw new IOException("Got wrong length during writeBlock(" + block + 
                                ") from " + inAddr + " at offset " + 
                                header.getOffsetInBlock() + ": " +
                                header.getDataLen()); 
        }
    
        long offsetInBlock = header.getOffsetInBlock();
        long seqno = header.getSeqno();
        boolean lastPacketInBlock = header.isLastPacketInBlock();
        final int len = header.getDataLen();
        boolean syncBlock = header.getSyncBlock();
    
        // avoid double sync'ing on close
        if (syncBlock && lastPacketInBlock) { 
          this.syncOnClose = false;
          // sync directory for finalize irrespective of syncOnClose config since
          // sync is requested.
          this.dirSyncOnFinalize = true;
        }
    
        // update received bytes
        final long firstByteInBlock = offsetInBlock;
        offsetInBlock += len;
        if (replicaInfo.getNumBytes() < offsetInBlock) { 
          replicaInfo.setNumBytes(offsetInBlock);
        }
        
        // put in queue for pending acks, unless sync was requested
        // 如果不是数据流管道中的最后一个数据节点， 则立即处理响应消息
        if (responder != null && !syncBlock && !shouldVerifyChecksum()) { 
          ((PacketResponder) responder.getRunnable()).enqueue(seqno,
              lastPacketInBlock, offsetInBlock, Status.SUCCESS);
        }
    
        // Drop heartbeat for testing.
        if (seqno < 0 && len == 0 &&
            DataNodeFaultInjector.get().dropHeartbeatPacket()) { 
          return 0;
        }
        // 向下游节点发送数据包
        //First write the packet to the mirror:
        if (mirrorOut != null && !mirrorError) { 
          try { 
            long begin = Time.monotonicNow();
            // For testing. Normally no-op.
            DataNodeFaultInjector.get().stopSendingPacketDownstream(mirrorAddr);
            packetReceiver.mirrorPacketTo(mirrorOut);
            mirrorOut.flush();
            long now = Time.monotonicNow();
            setLastSentTime(now);
            long duration = now - begin;
            DataNodeFaultInjector.get().logDelaySendingPacketDownstream(
                mirrorAddr,
                duration);
            trackSendPacketToLastNodeInPipeline(duration);
            if (duration > datanodeSlowLogThresholdMs && LOG.isWarnEnabled()) { 
              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();
        
        if (lastPacketInBlock || len == 0) { 
          if(LOG.isDebugEnabled()) { 
            LOG.debug("Receiving an empty packet or the end of the block " + block);
          }
          // sync block if requested
          // 如果接收了完整的数据块， 并且启动了sync标识， 则立即将数据同步到磁盘
          if (syncBlock) { 
            flushOrSync(true);
          }
        } else { 
          //如果当前节点是数据流管道中的最后一个节点， 则验证数据包的校验和
          final int checksumLen = diskChecksum.getChecksumSize(len);
          final int checksumReceivedLen = checksumBuf.capacity();
    
          if (checksumReceivedLen > 0 && checksumReceivedLen != checksumLen) { 
            throw new IOException("Invalid checksum length: received length is "
                + checksumReceivedLen + " but expected length is " + checksumLen);
          }
    
          if (checksumReceivedLen > 0 && shouldVerifyChecksum()) { 
            try { 
              //调用verifyChunks()验证数据包校验和
              verifyChunks(dataBuf, checksumBuf);
            } catch (IOException ioe) { 
              // checksum error detected locally. there is no reason to continue.
              // 验证出现异常， 则向上游客户端发送校验异常消息
              if (responder != null) { 
                try { 
                  ((PacketResponder) responder.getRunnable()).enqueue(seqno,
                      lastPacketInBlock, offsetInBlock,
                      Status.ERROR_CHECKSUM);
                  // Wait until the responder sends back the response
                  // and interrupt this thread.
                  Thread.sleep(3000);
                } catch (InterruptedException e) {  }
              }
              throw new IOException("Terminating due to a checksum error." + ioe);
            }
     
            if (needsChecksumTranslation) { 
              // overwrite the checksums in the packet buffer with the
              // appropriate polynomial for the disk storage.
              // 如果客户端发送的数据校验方式和当前数据节点的不一致， 则转换校验和
              translateChunks(dataBuf, checksumBuf);
            }
          }
    
          if (checksumReceivedLen == 0 && !streams.isTransientStorage()) { 
            // checksum is missing, need to calculate it
            checksumBuf = ByteBuffer.allocate(checksumLen);
            diskChecksum.calculateChunkedSums(dataBuf, checksumBuf);
          }
          
          // by this point, the data in the buffer uses the disk checksum
    
          final boolean shouldNotWriteChecksum = checksumReceivedLen == 0
              && streams.isTransientStorage();
          try { 
            long onDiskLen = replicaInfo.getBytesOnDisk();
            if (onDiskLen<offsetInBlock) { 
              // Normally the beginning of an incoming packet is aligned with the
              // existing data on disk. If the beginning packet data offset is not
              // checksum chunk aligned, the end of packet will not go beyond the
              // next chunk boundary.
              // When a failure-recovery is involved, the client state and the
              // the datanode state may not exactly agree. I.e. the client may
              // resend part of data that is already on disk. Correct number of
              // bytes should be skipped when writing the data and checksum
              // buffers out to disk.
              long partialChunkSizeOnDisk = onDiskLen % bytesPerChecksum;
              long lastChunkBoundary = onDiskLen - partialChunkSizeOnDisk;
              boolean alignedOnDisk = partialChunkSizeOnDisk == 0;
              boolean alignedInPacket = firstByteInBlock % bytesPerChecksum == 0;
    
              // 如果磁盘上数据的结尾未对齐块，则需要覆盖最后一个校验和。
              // If the end of the on-disk data is not chunk-aligned, the last
              // checksum needs to be overwritten.
              boolean overwriteLastCrc = !alignedOnDisk && !shouldNotWriteChecksum;
              // If the starting offset of the packat data is at the last chunk
              // boundary of the data on disk, the partial checksum recalculation
              // can be skipped and the checksum supplied by the client can be used
              // instead. This reduces disk reads and cpu load.
              boolean doCrcRecalc = overwriteLastCrc &&
                  (lastChunkBoundary != firstByteInBlock);
    
              // If this is a partial chunk, then verify that this is the only
              // chunk in the packet. If the starting offset is not chunk
              // aligned, the packet should terminate at or before the next
              // chunk boundary.
              if (!alignedInPacket && len > bytesPerChecksum) { 
                throw new IOException("Unexpected packet data length for "
                    +  block + " from " + inAddr + ": a partial chunk must be "
                    + " sent in an individual packet (data length = " + len
                    +  " > bytesPerChecksum = " + bytesPerChecksum + ")");
              }
    
              // If the last portion of the block file is not a full chunk,
              // then read in pre-existing partial data chunk and recalculate
              // the checksum so that the checksum calculation can continue
              // from the right state. If the client provided the checksum for
              // the whole chunk, this is not necessary.
              Checksum partialCrc = null;
              if (doCrcRecalc) { 
                if (LOG.isDebugEnabled()) { 
                  LOG.debug("receivePacket for " + block 
                      + ": previous write did not end at the chunk boundary."
                      + " onDiskLen=" + onDiskLen);
                }
                long offsetInChecksum = BlockMetadataHeader.getHeaderSize() +
                    onDiskLen / bytesPerChecksum * checksumSize;
                partialCrc = computePartialChunkCrc(onDiskLen, offsetInChecksum);
              }
    
              // The data buffer position where write will begin. If the packet
              // data and on-disk data have no overlap, this will not be at the
              // beginning of the buffer.
              int startByteToDisk = (int)(onDiskLen-firstByteInBlock) 
                  + dataBuf.arrayOffset() + dataBuf.position();
    
              // Actual number of data bytes to write.
              int numBytesToDisk = (int)(offsetInBlock-onDiskLen);
              
              // Write data to disk.
              long begin = Time.monotonicNow();
    
              streams.writeDataToDisk(dataBuf.array(),  startByteToDisk, numBytesToDisk);
              long duration = Time.monotonicNow() - begin;
              if (duration > datanodeSlowLogThresholdMs && LOG.isWarnEnabled()) { 
                LOG.warn("Slow BlockReceiver write data to disk cost:" + duration
                    + "ms (threshold=" + datanodeSlowLogThresholdMs + "ms), "
                    + "volume=" + getVolumeBaseUri()
                    + ", blockId=" + replicaInfo.getBlockId());
              }
    
              if (duration > maxWriteToDiskMs) { 
                maxWriteToDiskMs = duration;
              }
    
              final byte[] lastCrc;
              if (shouldNotWriteChecksum) { 
                lastCrc = null;
              } else { 
                int skip = 0;
                byte[] crcBytes = null;
    
                // First, prepare to overwrite the partial crc at the end.
                if (overwriteLastCrc) {  // not chunk-aligned on disk
                  // prepare to overwrite last checksum
                  adjustCrcFilePosition();
                }
    
                // The CRC was recalculated for the last partial chunk. Update the
                // CRC by reading the rest of the chunk, then write it out.
                if (doCrcRecalc) { 
                  // Calculate new crc for this 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);
                  if(LOG.isDebugEnabled()) { 
                    LOG.debug("Writing out partial crc for data len " + len +
                        ", skip=" + skip);
                  }
                  skip++; //  For the partial chunk that was just read.
                }
    
                // Determine how many checksums need to be skipped up to the last
                // boundary. The checksum after the boundary was already counted
                // above. Only count the number of checksums skipped up to the
                // boundary here.
                long skippedDataBytes = lastChunkBoundary - firstByteInBlock;
    
                if (skippedDataBytes > 0) { 
                  skip += (int)(skippedDataBytes / bytesPerChecksum) +
                      ((skippedDataBytes % bytesPerChecksum == 0) ? 0 : 1); } skip *= checksumSize; // Convert to number of bytes // write the rest of checksum final int offset = checksumBuf.arrayOffset() + checksumBuf.position() + skip; final int end = offset + checksumLen - skip; // If offset >= end, there is no more checksum to write. // I.e. a partial chunk checksum rewrite happened and there is no // more to write after that. if (offset >= end && doCrcRecalc) { lastCrc = crcBytes; } else { final int remainingBytes = checksumLen - skip; lastCrc = copyLastChunkChecksum(checksumBuf.array(), checksumSize, end); checksumOut.write(checksumBuf.array(), offset, remainingBytes); } } /// flush entire packet, sync if requested flushOrSync(syncBlock); replicaInfo.setLastChecksumAndDataLen(offsetInBlock, lastCrc); datanode.metrics.incrBytesWritten(len); datanode.metrics.incrTotalWriteTime(duration); manageWriterOsCache(offsetInBlock); } } catch (IOException iex) { // Volume error check moved to FileIoProvider throw iex; } } // if sync was requested, put in queue for pending acks here // (after the fsync finished) // 如果是最后一个节点， 或者是syncBlock方式， 则在数据写完磁盘以后再对ack进行处理 if (responder != null && (syncBlock || shouldVerifyChecksum())) { 
          ((PacketResponder) responder.getRunnable()).enqueue(seqno,
              lastPacketInBlock, offsetInBlock, Status.SUCCESS);
        }
    
        /*
         * Send in-progress responses for the replaceBlock() calls back to caller to
         * avoid timeouts due to balancer throttling. HDFS-6247
         */
        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) {  // throttle I/O
          throttler.throttle(len);
        }
        
        return lastPacketInBlock?-1:len;
      }

# 四.BlockReceiver.PacketResponder #

BlockReceiver负责从上游数据节点接收数据包并转发到下游数据节点。 同时， 如果当前节点是数据流管道中的最后一个节点， BlockReceiver还需要验证数据包的校验和是否正确。 除此之外， 数据节点还需要从下游节点接收数据包的确认消息， 然后转发给上游节点， BlockReceiver将这部分功能委托给了内部类PacketResponder。

PacketResponder是一个独立的线程类， 它和BlockReceiver所在的线程共同完成数据块的写操作流程。 这里之所以将数据块处理和数据块响应消息处理放在两个线程中， 是因为如果使用一个线程处理， 需要同时监听上游节点的输入流和下游节点的输入流， 任意一个输入流阻塞都会造成另一个输入流上消息处理的延迟。

BlockReceiver完成对指定数据包的处理之后， 会触发PacketResponder类处理当前数据包的响应消息。 PacketResponder类监听下游的输入流， 接收到这个数据包的确认消息后， 在确认消息中添加当前数据节点的消息， 然后将这个消息发送给上游数据节点.

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 6]

BlockReceiver完成对指定数据包的处理之后， 会调用enqueue()方法通知PacketResponder类处理这个数据包的响应。

ackQueue是一个典型的生产者-消费者队列 , 通过enqueue()方法添加任务， 它将等待PacketResponder类处理的数据包加入ackQueue队列中，ackQueue队列中保存的所有数据包都会由PacketResponder的run()方法处理。 成功将数据包添加到ackQueue队列后， enqueue()方法会调用notify()方法通知run()方法处理数据包。

/** queue for packets waiting for ack - synchronization using monitor lock */
        private final LinkedList<Packet> ackQueue = new LinkedList<Packet>(); 
    
    
    
     
        /**
         * enqueue the seqno that is still be to acked by the downstream datanode.
         * @param seqno sequence number of the packet
         * @param lastPacketInBlock if true, this is the last packet in block
         * @param offsetInBlock offset of this packet in block
         */
        void enqueue(final long seqno, final boolean lastPacketInBlock,
            final long offsetInBlock, final Status ackStatus) { 
          final Packet p = new Packet(seqno, lastPacketInBlock, offsetInBlock,
              System.nanoTime(), ackStatus);
          if(LOG.isDebugEnabled()) { 
            LOG.debug(myString + ": enqueue " + p);
          }
          synchronized(ackQueue) { 
            if (running) { 
              // 加入ackQueue队列 
              ackQueue.addLast(p);
              ackQueue.notifyAll();
            }
          }
        }

PacketResponder.run()方法会循环对数据块中的所有数据包执行确认响应的逻辑， 如果抛出异常， 则将PacketResponder.running设置为false， 使得while循环的isRunning()判断为false， PacketResponder线程也就会结束。 如果PacketResponder阻塞了， 则通过interrupt()方法中断PacketResponder线程。

run()方法首先会从下游节点的输入流中读取一个响应， 并判断这个响应中是否有OOB消息（Datanode在写操作时被触发重启的情况下， 会通过数据流管道逆向发送一个OOB响应消息给客户端， 由客户端处理数据流管道中Datanode节点重启的情况） ， 如果有OOB消息， 则立即将这个消息转发给上游数据节点。 接下来run()方法会在ackQueue队列上等待需要处理的数据包， 然后判断从下游节点接收的数据包响应与从ackQueue队列中取出的待处理数据包是否匹配， 如果不匹配则抛出异常。 如果PacketResponder在从下游节点读入ack的过程中出现异常， 则将mirrorError字段设置为true， run()方法会在后续向上游节点发送的响应中携带错误信息。

PacketResponder会判断当前接收的数据包响应是否为数据块中最后一个数据包的响应， 如果是， 则调用finalizeBlock()方法向Namenode提交这个数据块。 接下来run()方法会调用sendAckUpstream()方法复制下游节点的数据包响应， 并在该响应中加入当前节点的状态， 然后构造新的响应发送给上游节点。 完成数据包响应的处理后， 从ackQueue队列中移除这个数据包。

/**
         * Thread to process incoming acks.
         * @see java.lang.Runnable#run()
         */
        @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;
              long ackRecvNanoTime = 0;
              try { 
                if (type != PacketResponderType.LAST_IN_PIPELINE && !mirrorError) { 
                  // 从下游节点的输出流中读入一个响应
                  DataNodeFaultInjector.get().failPipeline(replicaInfo, mirrorAddr);
                  // read an ack from downstream datanode
                  ack.readFields(downstreamIn);
                  ackRecvNanoTime = System.nanoTime();
                  if (LOG.isDebugEnabled()) { 
                    LOG.debug(myString + " got " + ack);
                  }
                  // Process an OOB ACK.
                  // 判断响应消息是否是OOB消息， 例如下游节点重启
                  Status oobStatus = ack.getOOBStatus();
                  if (oobStatus != null) { 
                    LOG.info("Relaying an out of band ack of type " + oobStatus);
                    //将OOB消息转发给上游节点处理
                    sendAckUpstream(ack, PipelineAck.UNKOWN_SEQNO, 0L, 0L,
                        PipelineAck.combineHeader(datanode.getECN(),
                          Status.SUCCESS));
                    continue;
                  }
                  seqno = ack.getSeqno();
                }
                if (seqno != PipelineAck.UNKOWN_SEQNO
                    || type == PacketResponderType.LAST_IN_PIPELINE) { 
    
                  //从ackQueue队列中取出待处理数据包
                  pkt = waitForAckHead(seqno);
                  if (!isRunning()) { 
                    break;
                  }
                  expected = pkt.seqno;
                  //判断当前接收的数据包响应是否匹配待处理数据包
                  if (type == PacketResponderType.HAS_DOWNSTREAM_IN_PIPELINE
                      && seqno != expected) { 
                    //如果不匹配则抛出异常
                    throw new IOException(myString + "seqno: expected=" + expected
                        + ", received=" + seqno);
                  }
                  if (type == PacketResponderType.HAS_DOWNSTREAM_IN_PIPELINE) { 
                    // The total ack time includes the ack times of downstream
                    // nodes.
                    // The value is 0 if this responder doesn't have a downstream
                    // DN in the pipeline.
                    totalAckTimeNanos = ackRecvNanoTime - pkt.ackEnqueueNanoTime;
                    // Report the elapsed time from ack send to ack receive minus
                    // the downstream ack time.
                    long ackTimeNanos = totalAckTimeNanos
                        - ack.getDownstreamAckTimeNanos();
                    if (ackTimeNanos < 0) { 
                      if (LOG.isDebugEnabled()) { 
                        LOG.debug("Calculated invalid ack time: " + ackTimeNanos
                            + "ns.");
                      }
                    } else { 
                      datanode.metrics.addPacketAckRoundTripTimeNanos(ackTimeNanos);
                    }
                  }
                  lastPacketInBlock = pkt.lastPacketInBlock;
                }
              } catch (InterruptedException ine) { 
                isInterrupted = true;
              } catch (IOException ioe) { 
                if (Thread.interrupted()) { 
                  isInterrupted = true;
                } else if (ioe instanceof EOFException && !packetSentInTime()) { 
                  // The downstream error was caused by upstream including this
                  // node not sending packet in time. Let the upstream determine
                  // who is at fault.  If the immediate upstream node thinks it
                  // has sent a packet in time, this node will be reported as bad.
                  // Otherwise, the upstream node will propagate the error up by
                  // closing the connection.
                  LOG.warn("The downstream error might be due to congestion in " +
                      "upstream including this node. Propagating the error: ",
                      ioe);
                  throw ioe;
                } else { 
                  // continue to run even if can not read from mirror
                  // notify client of the error
                  // and wait for the client to shut down the pipeline
                  //如果从下游节点读取数据时抛出异常， 则将mirrorError设置为true
                  mirrorError = true;
                  LOG.info(myString, ioe);
                }
              }
    
              if (Thread.interrupted() || isInterrupted) { 
                /*
                 * The receiver thread cancelled this thread. We could also check
                 * any other status updates from the receiver thread (e.g. if it is
                 * ok to write to replyOut). It is prudent to not send any more
                 * status back to the client because this datanode has a problem.
                 * The upstream datanode will detect that this datanode is bad, and
                 * rightly so.
                 *
                 * The receiver thread can also interrupt this thread for sending
                 * an out-of-band response upstream.
                 */
                LOG.info(myString + ": Thread is interrupted.");
                //对于线程中断， 则将running设置为false， 停止当前线程
                running = false;
                continue;
              }
    
              if (lastPacketInBlock) { 
                //如果数据包响应是数据块的最后一个数据包响应， 则提交这个数据块
                // Finalize the block and close the block file
                finalizeBlock(startTime);
              }
    
              Status myStatus = pkt != null ? pkt.ackStatus : Status.SUCCESS;
    
              //复制下游节点的数据包响应， 加入当前数据节点的状态， 并发送给上游节点
              sendAckUpstream(ack, expected, totalAckTimeNanos,
                (pkt != null ? pkt.offsetInBlock : 0),
                PipelineAck.combineHeader(datanode.getECN(), myStatus));
              if (pkt != null) { 
                // remove the packet from the ack queue
                //已经完成响应处理的数据包， 从ackQueue队列中移除
                removeAckHead();
              }
            } catch (IOException e) { 
              LOG.warn("IOException in BlockReceiver.run(): ", e);
              if (running) { 
                // Volume error check moved to FileIoProvider
                LOG.info(myString, e);
                // 设置running标志位为false,停止PacketResponder线程的执行
                running = false;
                if (!Thread.interrupted()) {  // failure not caused by interruption
                  // 中断PacketResponder线程
                  receiverThread.interrupt();
                }
              }
            } catch (Throwable e) { 
              if (running) { 
                LOG.info(myString, e);
                //设置running标志位为false
                running = false;
                //中断PacketResponder线程
                receiverThread.interrupt();
              }
            }
          }
          LOG.info(myString + " terminating");
        }

sendAckUpstream()方法会调用sendAckUpstreamUnprotected()方法发送响应给上游节点。

向上游节点发送的响应分为如下几种情况 :

■ 下游节点发送的是OOB消息， 则将下游节点的数据包响应原封不动地保存。  
■ 如果从下游节点读取响应异常， 也就是mirrorError为true， 则将异常Status.MIRROR\_ERROR\_STATUS记录在数据包响应中。  
■ 对于其他流程， 则将当前Datanode的状态放入数据包响应中。

/**
         * @param ack Ack received from downstream
         * @param seqno sequence number of ack to be sent upstream
         * @param totalAckTimeNanos total ack time including all the downstream
         *          nodes
         * @param offsetInBlock offset in block for the data in packet
         * @param myHeader the local ack header
         */
        private void sendAckUpstreamUnprotected(PipelineAck ack, long seqno,
            long totalAckTimeNanos, long offsetInBlock, int myHeader)
            throws IOException { 
          final int[] replies;
    
          //如果下游节点发送的是OOB消息， 则保留下游节点消息内容
          if (ack == null) { 
            // A new OOB response is being sent from this node. Regardless of
            // downstream nodes, reply should contain one reply.
            replies = new int[] {  myHeader };
          } else if (mirrorError) {  // ack read error
    
            //如果是当前节点从下游节点读取响应消息异常
            //则将当前Datanode的错误记录在响应消息中
    
            int h = PipelineAck.combineHeader(datanode.getECN(), Status.SUCCESS);
            int h1 = PipelineAck.combineHeader(datanode.getECN(), Status.ERROR);
            replies = new int[] { h, h1};
    
    
          } else { 
    
            //其他流程
    
            short ackLen = type == PacketResponderType.LAST_IN_PIPELINE ? 0 : ack
                .getNumOfReplies();
    
            replies = new int[ackLen + 1];
            //将当前数据的状态放入响应中
            replies[0] = myHeader;
    
            for (int i = 0; i < ackLen; ++i) { 
              //将下游数据节点的状态放入响应中
              replies[i + 1] = ack.getHeaderFlag(i);
            }
            // If the mirror has reported that it received a corrupt packet,
            // do self-destruct to mark myself bad, instead of making the
            // mirror node bad. The mirror is guaranteed to be good without
            // corrupt data on disk.
            if (ackLen > 0 && PipelineAck.getStatusFromHeader(replies[1]) ==
              Status.ERROR_CHECKSUM) { 
              throw new IOException("Shutting down writer and responder "
                  + "since the down streams reported the data sent by this "
                  + "thread is corrupt");
            }
          }
    
          //构造新的数据包响应消息
    
          PipelineAck replyAck = new PipelineAck(seqno, replies,
              totalAckTimeNanos);
          if (replyAck.isSuccess()
              && offsetInBlock > replicaInfo.getBytesAcked()) { 
            replicaInfo.setBytesAcked(offsetInBlock);
          }
          // 将数据包响应消息发送给上游节点
          // send my ack back to upstream datanode
          long begin = Time.monotonicNow();
          /* for test only, no-op in production system */
          DataNodeFaultInjector.get().delaySendingAckToUpstream(inAddr);
          replyAck.write(upstreamOut);
          upstreamOut.flush();
          long duration = Time.monotonicNow() - begin;
          DataNodeFaultInjector.get().logDelaySendingAckToUpstream(
              inAddr,
              duration);
          if (duration > datanodeSlowLogThresholdMs) { 
            LOG.warn("Slow PacketResponder send ack to upstream took " + duration
                + "ms (threshold=" + datanodeSlowLogThresholdMs + "ms), " + myString
                + ", replyAck=" + replyAck
                + ", downstream DNs=" + Arrays.toString(downstreamDNs)
                + ", blockId=" + replicaInfo.getBlockId());
          } else if (LOG.isDebugEnabled()) { 
            LOG.debug(myString + ", replyAck=" + replyAck);
          }
    
          // If a corruption was detected in the received data, terminate after
          // sending ERROR_CHECKSUM back.
          //如果当前节检测到校验和错误， 则停止BlockReceiver以及PacketResponder线程
          Status myStatus = PipelineAck.getStatusFromHeader(myHeader);
          if (myStatus == Status.ERROR_CHECKSUM) { 
            throw new IOException("Shutting down writer and responder "
                + "due to a checksum error in received data. The error "
                + "response has been sent upstream.");
          }
        }

如果发现下游节点汇报了数据包的校验和错误， 则抛出异常停止当前节点的BlockReceiver以及PacketResponder线程； 如果是当前节点出现了数据包的校验和错误， 则将错误消息发送给上游节点， 然后抛出异常停止当前节点的BlockReceiver以及PacketResponder线程。

# 五.处理结果上报Namenode #

Datanode完成数据块的写入操作后需要向Namenode汇报这个新的数据块， 以方便Namenode更新命名空间。 PacketResponder在确认数据块中所有数据包的响应都正确处理之后， 会调用BlockReceiver.finalizeBlock()方法通知名字节点当前Datanode成功接收了这个数据块。

finalizeBlock()方法又会调用DataNode.closeBlock()方法 通过 DataNode.closeBlock()在底层调用了BPOfferService.notifyNamenodeReceivedBlock() 方法通知Namenode。

/**
         * Finalize block 关闭block文件
         * Finalize the block and close the block file
         * @param startTime time when BlockReceiver started receiving the block
         */
        private void finalizeBlock(long startTime) throws IOException { 
          long endTime = 0;
          // Hold a volume reference to finalize block.
          try (ReplicaHandler handler = BlockReceiver.this.claimReplicaHandler()) { 
            BlockReceiver.this.close();
            endTime = ClientTraceLog.isInfoEnabled() ? System.nanoTime() : 0;
            block.setNumBytes(replicaInfo.getNumBytes());
            datanode.data.finalizeBlock(block, dirSyncOnFinalize);
          }
    
          if (pinning) { 
            datanode.data.setPinning(block);
          }
          /// datanode 关闭block
          datanode.closeBlock(block, null, replicaInfo.getStorageUuid(), replicaInfo.isOnTransientStorage());
    
          if (ClientTraceLog.isInfoEnabled() && isClient) { 
            long offset = 0;
            DatanodeRegistration dnR = datanode.getDNRegistrationForBP(block
                .getBlockPoolId());
            ClientTraceLog.info(String.format(DN_CLIENTTRACE_FORMAT, inAddr,
                myAddr, block.getNumBytes(), "HDFS_WRITE", clientname, offset,
                dnR.getDatanodeUuid(), block, endTime - startTime));
          } else { 
            LOG.info("Received " + block + " size " + block.getNumBytes()
                + " from " + inAddr);
          }
        }

在写数据流程中， 另一处调用DataNode.closeBlock()方法是在DataXceiver.writeBlock()中， 当这个写操作是由数据块复制操作触发， 并且Datanode已经调用blockReceiver.receiveBlock()成功保存了数据块时， 会调用这个方法通Namenode节点。

--------------------

参考:  
Hadoop 2.X HDFS源码剖析 – 徐鹏

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/5430deb585f24208926772e5ddc28cc0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/a326eaf48bcf40f9b33529b2a1cf8edb.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]: https://img-blog.csdnimg.cn/20201123155735928.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 3]: https://img-blog.csdnimg.cn/20201123215028993.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 4]: https://img-blog.csdnimg.cn/20201124000346370.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20201124104410959.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 6]: https://img-blog.csdnimg.cn/20201124142710685.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0,size_16,color_FFFFFF,t_70#pic_center