Hadoop3.2.1 【 HDFS 】源码分析 : DataXceiver: 读取数据块解析 [二]

野性酷女 2022-11-20 01:56 533阅读 0赞

# 一. 前言 #

Receiver.processOp()方法用于处理流式接口的请求， 它首先从数据流中读取序列化后的参数， 对参数反序列化， 然后根据操作码调用DataTransferProtocol中定义的方法， 这些方法都是在DataXceiver中具体实现的。  
流式接口中最重要的一个部分就是客户端从数据节点上读取数据块， DataTransferProtocol.readBlock()给出了读取操作的接口定义， 操作码是81。 DataXceiver.readBlock()则实现了DataTransferProtocol.readBlock()方法。

客户端通过调用Sender.readBlock()方法从指定数据节点上读取数据块， 请求通过IO流到达数据节点后， 数据节点的DataXceiverServer会创建一个DataXceiver对象响应流式接口请求。 DataXceiver.processOp()方法解析操作码为81（读请求） ， 则调用DataXceiver.readBlock()响应这个读请求。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]  
DataXceiver.readBlock()首先向客户端回复一个BlockOpResponseProto响应， 指明请求已经成功接收， 并通过BlockOpResponseProto响应给出Datanode当前使用的校验方式。 接下来DataXceiver.readBlock()方法会将数据节点上的数据块（block） 切分成若干个数据包（packet） ， 然后依次将数据包发送给客户端。 客户端会在收到每个数据包时进行校验，如果校验和错误， 客户端会切断与当前数据节点的连接， 选择新的数据节点读取数据； 如果数据块内的所有数据包都校验成功， 客户端会给数据节点发送一个Status.CHECKSUM\_OK响应， 表明读取成功。

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

# 二. DataXceiver.readBlock() #

先看一下DataTransferProtocol.readBlock()方法的定义。

/** * Read a block. * * @param blk the block being read. * @param blockToken security token for accessing the block. * @param clientName client's name. * @param blockOffset offset of the block. * @param length maximum number of bytes for this read. * @param sendChecksum if false, the DN should skip reading and sending * checksums * @param cachingStrategy The caching strategy to use. * * 从当前Datanode读取指定的数据块。 */
      void readBlock(final ExtendedBlock blk,
          final Token<BlockTokenIdentifier> blockToken,
          final String clientName,
          final long blockOffset,
          final long length,
          final boolean sendChecksum,
          final CachingStrategy cachingStrategy) throws IOException;

readBlock()需要传入的参数主要有如下几个。  
■ ExtendedBlock blk： 要读取的数据块。  
■ TokenblockToken： 数据块的访问令牌。  
■ String clientName： 客户端的名称。  
■ long blockOffset： 要读取数据在数据块中的位置。  
■ long length： 读取数据的长度。  
■ sendChecksum： Datanode是否发送校验数据， 如果为false， 则Datanode不发送校验数据。 这里要特别注意， **数据块的读取校验工作是在客户端完成的**， 客户端会将校验结果返回给Datanode。  
■ CachingStrategy cachingStrategy： 缓存策略， 这里主要包括两个重要的字段——readahead， 预读取操作， Datanode会在读取数据块文件时预读取部分数据至操作系统缓存中， 以提高读取文件效率 ； dropBehind， 如果缓存中存放的文件比较多， 那么在读取完数据之后， 就马上从缓存中将数据删除。

# 三.readBlock()方法的执行流程 #

■ 创建BlockSender对象： 首先调用getOutputStream()方法获取Datanode连接到客户端的IO流， 然后构造BlockSender对象。  
■ 成功创建BlockSender对象后， 调用writeSuccessWithChecksumInfo()方法发送BlockOpResponseProto响应给客户端， 通知客户端读请求已经成功接收， 并且告知客户端当前数据节点的校验信息。  
■ 调用BlockSender.sendBlock()方法将数据块发送给客户端。  
■ 当BlockSender完成发送数据块的所有内容后， 客户端会响应一个状态码，Datanode需要解析这个状态码。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]  
readBlock()的异常处理逻辑也比较简单， 当客户端关闭了当前Socket（可能是出现了校验错误） ， 或者无法从IO流中成功获取客户端发回的响应时， 则直接关闭Datanode到客户端的底层输出流。 在readBlock()方法的最后， 会关闭BlockSender类以执行清理操作。

@Override
      public void readBlock(final ExtendedBlock block, // BP-451827885-192.168.8.156-1584099133244:blk_1073746920_6096
          final Token<BlockTokenIdentifier> blockToken, // Kind: , Service: , Ident:
          final String clientName, //DFSClient_NONMAPREDUCE_368352401_1
          final long blockOffset, // 0
          final long length,  // 1361
          final boolean sendChecksum, // true
          final CachingStrategy cachingStrategy) throws IOException {   // CachingStrategy(dropBehind=null, readahead=null)
        // 客户端名称 DFSClient_NONMAPREDUCE_368352401_1
        previousOpClientName = clientName;
    
        long read = 0;
    
        updateCurrentThreadName("Sending block " + block);
    
        OutputStream baseStream = getOutputStream();
    
        DataOutputStream out = getBufferedOutputStream();
    
        checkAccess(out, true, block, blockToken, Op.READ_BLOCK, BlockTokenIdentifier.AccessMode.READ);
    
        // send the block
        BlockSender blockSender = null;
    
        DatanodeRegistration dnR =   datanode.getDNRegistrationForBP(block.getBlockPoolId());
        // src: /127.0.0.1:9866, dest: /127.0.0.1:51764, bytes: %d, op: HDFS_READ, cliID: DFSClient_NONMAPREDUCE_368352401_1, offset: %d, srvID: 9efa402a-df6b-48cf-9273-5468f68cc42f, blockid: BP-451827885-192.168.8.156-1584099133244:blk_1073746920_6096, duration(ns): %d
        final String clientTraceFmt =
          clientName.length() > 0 && ClientTraceLog.isInfoEnabled()
            ? String.format(DN_CLIENTTRACE_FORMAT, localAddress, remoteAddress,
                "%d", "HDFS_READ", clientName, "%d",
                dnR.getDatanodeUuid(), block, "%d")
            : dnR + " Served block " + block + " to " +
                remoteAddress;
    
        try { 
          try { 
    
            blockSender = new BlockSender(block, blockOffset, length,
                true, false, sendChecksum, datanode, clientTraceFmt,
                cachingStrategy);
    
    
          } catch(IOException e) { 
            String msg = "opReadBlock " + block + " received exception " + e; 
            LOG.info(msg);
            sendResponse(ERROR, msg);
            throw e;
          }
          
          // send op status
          writeSuccessWithChecksumInfo(blockSender, new DataOutputStream(getOutputStream()));
    
          long beginRead = Time.monotonicNow();
    
          // 发送数据
          read = blockSender.sendBlock(out, baseStream, null); // send data
    
    
          long duration = Time.monotonicNow() - beginRead;
    
          if (blockSender.didSendEntireByteRange()) { 
            // If we sent the entire range, then we should expect the client
            // to respond with a Status enum.
            try { 
              ClientReadStatusProto stat = ClientReadStatusProto.parseFrom(
                  PBHelperClient.vintPrefixed(in));
              if (!stat.hasStatus()) { 
                LOG.warn("Client {} did not send a valid status code " +
                    "after reading. Will close connection.",
                    peer.getRemoteAddressString());
                IOUtils.closeStream(out);
              }
            } catch (IOException ioe) { 
              LOG.debug("Error reading client status response. Will close connection.", ioe);
              IOUtils.closeStream(out);
              incrDatanodeNetworkErrors();
            }
          } else { 
            IOUtils.closeStream(out);
          }
          datanode.metrics.incrBytesRead((int) read);
          datanode.metrics.incrBlocksRead();
          datanode.metrics.incrTotalReadTime(duration);
        } catch ( SocketException ignored ) { 
          LOG.trace("{}:Ignoring exception while serving {} to {}",
              dnR, block, remoteAddress, ignored);
          // Its ok for remote side to close the connection anytime.
          datanode.metrics.incrBlocksRead();
          IOUtils.closeStream(out);
        } catch ( IOException ioe ) { 
          /* What exactly should we do here? * Earlier version shutdown() datanode if there is disk error. */
          if (!(ioe instanceof SocketTimeoutException)) { 
            LOG.warn("{}:Got exception while serving {} to {}",
                dnR, block, remoteAddress, ioe);
            incrDatanodeNetworkErrors();
          }
          // Normally the client reports a bad block to the NN. However if the
          // meta file is corrupt or an disk error occurs (EIO), then the client
          // never gets a chance to do validation, and hence will never report
          // the block as bad. For some classes of IO exception, the DN should
          // report the block as bad, via the handleBadBlock() method
          datanode.handleBadBlock(block, ioe, false);
          throw ioe;
        } finally { 
          IOUtils.closeStream(blockSender);
        }
    
        //update metrics
        datanode.metrics.addReadBlockOp(elapsed());
        datanode.metrics.incrReadsFromClient(peer.isLocal(), read);
      }

# 四. 数据块的传输格式 #

BlockSender类主要负责从数据节点的磁盘读取数据块， 然后发送数据块到接收方。需要注意的是， BlockSender发送的数据是以一定结构组织的。

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 3]  
PacketLength大小为： 4 + CHECKSUMS(校验数据的大小）+ DATA（真实数据的大小）

BlockSender发送数据的格式包括两个部分： **校验信息头（ChecksumHeader）** **和数据包序列（packets）** 。

> |校验信息头（ChecksumHeader） |数据包序列（packets） |

## 4.1.校验信息头（ChecksumHeader) ##

ChecksumHeader是一个校验信息头， 用于描述当前Datanode使用的校验方式等信息。

> | 1 byte校验类型（CHECKSUM—TYPE) | 4 byte校验块大小（BYTES\_PER\_CHECKSUM)|

■ 数据校验类型： 数据校验类型定义在org.apache.hadoop.util.DataChecksum中， 目前包括三种方式——空校验（不进行校验） 、 CRC32以及CRC32C。 这里使用1byte描述数据校验类型， 空校验、 CRC32、 CRC32C、分别对应于值0、 1、 2、3、4。

另外两种类型 : CHECKSUM\_DEFAULT、CHECKSUM\_MIXED 对应于值3、4 不能用户创建 DataChecksum .

■ 校验块大小： 校验信息头中的第二个部分是校验块的大小， 也就是多少字节的数据产生一个校验值。 这里以CRC32为例， 一般情况下是512字节的数据产生一个4字节的校验和， 我们把这512字节的数据称为一个校验块（chunk） 。 这个校验块的概念非常重要， 它是HDFS中读取和写入数据块操作的最小单元.

## 4.2.数据包序列 ##

BlockSender会将数据块切分成若干数据包（packet） 对外发送， 当数据发送完成后，会以一个空的数据包作为结束。

每个数据包都包括一个变长的包头、 校验数据以及若干字节的实际数据。

■ 数据包头——数据包头用于描述当前数据包的信息， 是通过ProtoBuf序列化的，包括4字节的全包长度， 以及2字节的包头长度， 之后紧跟如下数据包信息。

*  当前数据包在整个数据块中的位置。
 *  数据包在管道中的序列号。
 *  当前数据包是不是数据块中的最后一个数据包。
 *  当前数据包中数据部分的长度。
 *  是否需要DN同步。

■ 校验数据——校验数据是对实际数据做校验操作产生的， 它将实际数据以校验块为单位， 每个校验块产生一个检验和， 校验数据中包含了所有校验块的校验和。校验数据的大小为： （实际数据长度+校验块大小 - 1） ／ 校验块大小×校验和长度。  
■ 实际数据——数据包中的实际数据就是数据块文件中保存的数据， 实际数据的传输是以校验块为单位的， 一个校验块对应产生一个校验和的实际数据。 在数据包中会将校验块与校验数据分开发送， 首先将所有校验块的校验数据发送出去， 然后再发送所有的校验块。

# 五. BlockSender实现 #

数据块的发送主要是由BlockSender类执行  
BlockSender中数据块的发送过程包括： 发送准备、 发送数据块以及清理工作。

## 5.1.发送准备——构造方法 ##

BlockSender中发送数据的准备工作主要是在BlockSender的构造方法中执行的，BlockSender的构造方法执行了以下操作。

■ readahead & dropBehind的处理： 如果用户通过cachingStrategy设置了这两个字段， 则按照这两个字段初始化读取操作。 如果cachingStrategy为Null， 则按照配置文件设置dropCacheBehindLargeReads为dfs.datanode.drop.cache.behind.reads，设置readaheadLength为dfs.datanode.readahead.bytes， 默认为4MB。  
■ 赋值与校验： 检查当前Datanode上被读取数据块的时间戳、 数据块文件的长度等状态是否正常。  
■ 是否开启transferTo模式： 默认为true， transferTo机制请参考零拷贝数据传输小节内容。  
■ 获取checksum信息： 从Meta文件中获取当前数据块的校验算法、 校验和长度， 以及多少字节产生一个校验值， 也就是校验块的大小。  
■ 计算offset以及endOffset: offset变量用于标识要读取的数据在数据块的起始位置，endOffset则用于标识结束的位置。 由于读取位置往往不会落在某个校验块的起始位置， 所以在准备工作中需要确保offset在校验块的起始位置， endOffset在校验块的结束位置。 这样读取时就可以以校验块为单位读取， 方便校验和的操作。  
■ 将数据块文件与校验和文件的offset都移动到指定位置。

/** * Constructor * * @param block Block that is being read * @param startOffset starting offset to read from * @param length length of data to read * @param corruptChecksumOk if true, corrupt checksum is okay * @param verifyChecksum verify checksum while reading the data * @param sendChecksum send checksum to client. * @param datanode datanode from which the block is being read * @param clientTraceFmt format string used to print client trace logs * @throws IOException */
      BlockSender(ExtendedBlock block, long startOffset, long length,
                  boolean corruptChecksumOk, boolean verifyChecksum,
                  boolean sendChecksum, DataNode datanode, String clientTraceFmt,
                  CachingStrategy cachingStrategy)
          throws IOException { 
    
        InputStream blockIn = null;
    
        DataInputStream checksumIn = null;
    
        FsVolumeReference volumeRef = null;
        // FileIoprovider@5795 DataNode{data=FSDataset{dirpath='[/opt/tools/hadoop-3.2.1/data/hdfs/data, /opt/tools/hadoop-3.2.1/data/hdfs/data01]'}, localName='192.168.8.188:9866', datanodeUuid='9efa402a-df6b-48cf-9273-5468f68cc42f', xmitsInProgress=0}
        this.fileIoProvider = datanode.getFileIoProvider();
        try { 
    
          this.block = block;
    
          this.corruptChecksumOk = corruptChecksumOk;
    
          this.verifyChecksum = verifyChecksum;
    
          this.clientTraceFmt = clientTraceFmt;
    
          /* * If the client asked for the cache to be dropped behind all reads, * we honor that. Otherwise, we use the DataNode defaults. * When using DataNode defaults, we use a heuristic where we only * drop the cache for large reads. */
          if (cachingStrategy.getDropBehind() == null) { 
    
            this.dropCacheBehindAllReads = false;
    
            this.dropCacheBehindLargeReads =
                datanode.getDnConf().dropCacheBehindReads;
    
          } else { 
            this.dropCacheBehindAllReads =
                this.dropCacheBehindLargeReads =
                     cachingStrategy.getDropBehind().booleanValue();
          }
          /* 默认开启预读取, 除非 请求头 指定 "不开启预读取 * Similarly, if readahead was explicitly requested, we always do it. * Otherwise, we read ahead based on the DataNode settings, and only * when the reads are large. */
          if (cachingStrategy.getReadahead() == null) { 
            this.alwaysReadahead = false;
            /// readaheadLength : 4194304 = 4M 默认 预读取大小: 4M
            this.readaheadLength = datanode.getDnConf().readaheadLength;
    
          } else { 
    
            this.alwaysReadahead = true;
    
            this.readaheadLength = cachingStrategy.getReadahead().longValue();
          }
          this.datanode = datanode;
          
          if (verifyChecksum) { 
    
            // To simplify implementation, callers may not specify verification without sending.
            Preconditions.checkArgument(sendChecksum,
                "If verifying checksum, currently must also send it.");
          }
          // 如果在构造BlockSender之后有一个追加写操作，那么最后一个部分校验和可能被append覆盖，
          // BlockSender需要在append write之前使用部分校验和。
          // if there is a append write happening right after the BlockSender
          // is constructed, the last partial checksum maybe overwritten by the
          // append, the BlockSender need to use the partial checksum before
          // the append write.
          ChunkChecksum chunkChecksum = null;
    
          final long replicaVisibleLength;
          try(AutoCloseableLock lock = datanode.data.acquireDatasetLock()) { 
            // 获取datanode上的副本信息
            // FinalizedReplica, blk_1073746921_6097, FINALIZED
            // getNumBytes() = 42648690
            // getBytesOnDisk() = 42648690
            // getVisibleLength()= 42648690
            // getVolume() = /opt/tools/hadoop-3.2.1/data/hdfs/data
            // getBlockURI() = file:/opt/tools/hadoop-3.2.1/data/hdfs/data/current/BP-451827885-192.168.8.156-1584099133244/current/finalized/subdir0/subdir19/blk_1073746921
            replica = getReplica(block, datanode);
            // 42648690
            replicaVisibleLength = replica.getVisibleLength();
          }
          if (replica.getState() == ReplicaState.RBW) { 
            // 副本正在被写入 , 等待写入足够的内容
            final ReplicaInPipeline rbw = (ReplicaInPipeline) replica;
    
            waitForMinLength(rbw, startOffset + length);
    
            chunkChecksum = rbw.getLastChecksumAndDataLen();
          }
          if (replica instanceof FinalizedReplica) { 
            // 副本已经被写入完成 , 获取该副本的 ChunkChecksum :dataLength = 42648690 checksum = {byte[4]@5845} 四位 : [-11,-56,-5,28]
            chunkChecksum = getPartialChunkChecksumForFinalized(  (FinalizedReplica)replica);
          }
    
          if (replica.getGenerationStamp() < block.getGenerationStamp()) { 
            throw new IOException("Replica gen stamp < block genstamp, block="
                + block + ", replica=" + replica);
          } else if (replica.getGenerationStamp() > block.getGenerationStamp()) { 
            if (DataNode.LOG.isDebugEnabled()) { 
              DataNode.LOG.debug("Bumping up the client provided"
                  + " block's genstamp to latest " + replica.getGenerationStamp()
                  + " for block " + block);
            }
    
            block.setGenerationStamp(replica.getGenerationStamp());
          }
          /// 副本可见长度小于0 , 不可见
          if (replicaVisibleLength < 0) { 
            throw new IOException("Replica is not readable, block="+ block + ", replica=" + replica);
          }
          if (DataNode.LOG.isDebugEnabled()) { 
            DataNode.LOG.debug("block=" + block + ", replica=" + replica);
          }
          // 是否开启零拷贝 dfs.datanode.transferTo.allowed : true
          // transferToFully() fails on 32 bit platforms for block sizes >= 2GB,
          // use normal transfer in those cases
          this.transferToAllowed = datanode.getDnConf().transferToAllowed &&
            (!is32Bit || length <= Integer.MAX_VALUE);
          // 在读取数据之前获取引用 FsVolumeImple$FsVolumeReferenceImpl@5928
          // Obtain a reference before reading data
          volumeRef = datanode.data.getVolume(block).obtainReference();
    
          /* 判断是否要验证 DataChecksum * (corruptChecksumOK, meta_file_exist): operation * True, True: will verify checksum * True, False: No verify, e.g., need to read data from a corrupted file * False, True: will verify checksum * False, False: throws IOException file not found */
          DataChecksum csum = null;
          if (verifyChecksum || sendChecksum) { 
            LengthInputStream metaIn = null;
            boolean keepMetaInOpen = false;
            try { 
    
    
              DataNodeFaultInjector.get().throwTooManyOpenFiles();
    
    
              metaIn = datanode.data.getMetaDataInputStream(block);
              if (!corruptChecksumOk || metaIn != null) { 
                if (metaIn == null) { 
                  //need checksum but meta-data not found
                  throw new FileNotFoundException("Meta-data not found for " +
                      block);
                }
    
                // The meta file will contain only the header if the NULL checksum
                // type was used, or if the replica was written to transient storage.
                // Also, when only header portion of a data packet was transferred
                // and then pipeline breaks, the meta file can contain only the
                // header and 0 byte in the block data file.
                // Checksum verification is not performed for replicas on transient
                // storage. The header is important for determining the checksum
                // type later when lazy persistence copies the block to non-transient
                // storage and computes the checksum.
                int expectedHeaderSize = BlockMetadataHeader.getHeaderSize();  // 7
                if (!replica.isOnTransientStorage() &&
                    metaIn.getLength() >= expectedHeaderSize) { 
                  checksumIn = new DataInputStream(new BufferedInputStream(
                      metaIn, IO_FILE_BUFFER_SIZE));
                  // DataChecksum(type=CRC32C, chunkSize=512)
                  csum = BlockMetadataHeader.readDataChecksum(checksumIn, block);
                  keepMetaInOpen = true;
                } else if (!replica.isOnTransientStorage() &&
                    metaIn.getLength() < expectedHeaderSize) { 
                  LOG.warn("The meta file length {} is less than the expected " +
                      "header length {}, indicating the meta file is corrupt",
                      metaIn.getLength(), expectedHeaderSize);
                  throw new CorruptMetaHeaderException("The meta file length "+
                      metaIn.getLength()+" is less than the expected length "+
                      expectedHeaderSize);
                }
              } else { 
                LOG.warn("Could not find metadata file for " + block);
              }
            } catch (FileNotFoundException e) { 
              if ((e.getMessage() != null) && !(e.getMessage()
                  .contains("Too many open files"))) { 
                // The replica is on its volume map but not on disk
                datanode
                    .notifyNamenodeDeletedBlock(block, replica.getStorageUuid());
                datanode.data.invalidate(block.getBlockPoolId(),
                    new Block[] { block.getLocalBlock()});
              }
              throw e;
            } finally { 
              if (!keepMetaInOpen) {   // keepMetaInOpen : true
                IOUtils.closeStream(metaIn);
              }
            }
          }
          if (csum == null) { 
            csum = DataChecksum.newDataChecksum(DataChecksum.Type.NULL,
                (int)CHUNK_SIZE);
          }
    
          /* * If chunkSize is very large, then the metadata file is mostly * corrupted. For now just truncate bytesPerchecksum to blockLength. */       
          int size = csum.getBytesPerChecksum();
          // 如果chunkSize非常大，则元数据文件大部分已损坏。现在只需将bytesPerchecksum截断为blockLength。
          if (size > 10*1024*1024 && size > replicaVisibleLength) { 
            //元数据文件损坏了, 重新构建 DataChecksum
            csum = DataChecksum.newDataChecksum(csum.getChecksumType(),
                Math.max((int)replicaVisibleLength, 10*1024*1024));
            size = csum.getBytesPerChecksum();
    
          }
    
          //校验块大小 512
          chunkSize = size;
          //校验算法 DataChecksum(type=CRC32C, chunkSize=512)
          checksum = csum;
          //校验和长度 4
          checksumSize = checksum.getChecksumSize();
    
          // 文件大小 42648690
          length = length < 0 ? replicaVisibleLength : length;
    
          // end is either last byte on disk or the length for which we have a checksum
          // end要么是磁盘上的最后一个字节，要么是校验和的长度 : 这里是文件长度.
          long end = chunkChecksum != null ? chunkChecksum.getDataLength()  : replica.getBytesOnDisk();
    
          if (startOffset < 0 || startOffset > end
              || (length + startOffset) > end) { 
            String msg = " Offset " + startOffset + " and length " + length
            + " don't match block " + block + " ( blockLen " + end + " )";
            LOG.warn(datanode.getDNRegistrationForBP(block.getBlockPoolId()) +
                ":sendBlock() : " + msg);
            throw new IOException(msg);
          }
    
          // 将offset位置设置在校验块的边界上， 也就是校验块的起始位置
          // Ensure read offset is position at the beginning of chunk
          offset = startOffset - (startOffset % chunkSize);
          if (length >= 0) { 
    
            //计算endOffset的位置， 确保endOffset在校验块的结束位置
            // Ensure endOffset points to end of chunk.
            long tmpLen = startOffset + length;
            if (tmpLen % chunkSize != 0) { 
              tmpLen += (chunkSize - tmpLen % chunkSize); //补齐数据, 使数据正好是512的整倍数
            }
            if (tmpLen < end) { 
              //结束位置还在数据块内， 则可以使用磁盘上的校验值 , 理论上应该不走这里.
              // will use on-disk checksum here since the end is a stable chunk
              end = tmpLen;
            } else if (chunkChecksum != null) { 
              //目前有写线程[当前线程]正在处理这个校验块， 则使用内存中的校验值
              // last chunk is changing. flag that we need to use in-memory checksum 
              this.lastChunkChecksum = chunkChecksum;
            }
    
          }
          endOffset = end; // 设置最后的偏移量为文件的偏移量
          // 将校验文件的坐标移动到offset对应的位置
          // seek to the right offsets
          if (offset > 0 && checksumIn != null) { 
    
            long checksumSkip = (offset / chunkSize) * checksumSize;
            // note blockInStream is seeked when created below
    
            if (checksumSkip > 0) { 
              // Should we use seek() for checksum file as well?
              IOUtils.skipFully(checksumIn, checksumSkip);
            }
    
          }
    
          //packet序列号设置为0
          seqno = 0;
    
          if (DataNode.LOG.isDebugEnabled()) { 
    
            DataNode.LOG.debug("replica=" + replica);
    
          }
          //将数据块文件的坐标移动到offset位置 准备开始读写
          blockIn = datanode.data.getBlockInputStream(block, offset); // seek to offset
          // 构建block数据的读取流 ReplicaInputStreams
          ris = new ReplicaInputStreams(  blockIn, checksumIn, volumeRef, fileIoProvider);
    
        } catch (IOException ioe) { 
          IOUtils.closeStream(this);
          org.apache.commons.io.IOUtils.closeQuietly(blockIn);
          org.apache.commons.io.IOUtils.closeQuietly(checksumIn);
          throw ioe;
        }
      }

## 5.2.预读取&丢弃——manageOsCache() ##

BlockSender在读取数据块之前， 会先调用manageOsCache()方法执行预读取（readahead） 操作以提高读取效率。 预读取操作就是将数据块文件提前读取到操作系统的缓存中， 这样当BlockSender到文件系统中读取数据块文件时， 可以直接从操作系统的缓存中读取数据， 比直接从磁盘上读取快很多。 但是操作系统的缓存空间是有限的， 所以需要调用manageOsCache()方法将不再使用的数据从缓存中丢弃（drop-behind） ， 为新的数据挪出空间。 BlockSender在读取数据时， 使用了预读取以及丢弃这两个特性.

manageOsCache()方法在HDFS管理员设置了预读取的长度（默认是4MB） 并且设置了所有的操作都使用预读取时， 或者当前读取是一个长读取（超过256KB的读取） 时， 会调用ReadaheadPool.readaheadStream()方法触发一个预读取操作， 这个预读取操作会从磁盘  
文件上预读取部分数据块文件的数据到操作系统的缓存中。  
同时managerOsCache()还会处理丢弃操作， 如果dropCacheBehindAllReads（所有读操作后都丢弃） 为true或者当前读取是一个大读取时， 则触发丢弃操作。 manageOsCache()方法会判断如果下一次读取数据的坐标offset大于下一次丢弃操作的开始坐标， 则将lastCacheDropOffset（上一次丢弃操作的结束位置） 和offset之间的数据全部从缓存中丢弃， 因为这些数据Datanode已经读取了，不需要放在缓存中了.

manageOsCache()的开启需要readaheadPool对象实例, readaheadPoold 创建是DataNode\#startDataNode的方法进行初始化的.  
但是需要本地库的支持**ReadaheadPool.getInstance();** 如果不支持是开启不了的,需要hadoop的定制依赖 : native hadoop library .

**由BlockSender\#doSendBlock()方法调用**

/** * * Manage the OS buffer cache by performing read-ahead and drop-behind. */
      private void manageOsCache() throws IOException { 
        // We can't manage the cache for this block if we don't have a file
        // descriptor to work with.
        if (ris.getDataInFd() == null) { 
          return;
        }
    
        //按条件触发预读取操作
        // Perform readahead if necessary
        if ((readaheadLength > 0) && (datanode.readaheadPool != null) &&
              (alwaysReadahead || isLongRead())) { 
    
          //满足预读取条件， 则调用ReadaheadPool.readaheadStream()方法触发预读取
          curReadahead = datanode.readaheadPool.readaheadStream(
              clientTraceFmt, ris.getDataInFd(), offset, readaheadLength,
              Long.MAX_VALUE, curReadahead);
        }
    
        //丢弃刚才从缓存中读取的数据， 因为不再需要使用这些数据了
        // Drop what we've just read from cache, since we aren't likely to need it again
        if (dropCacheBehindAllReads ||
            (dropCacheBehindLargeReads && isLongRead())) { 
          //丢弃数据的位置
          long nextCacheDropOffset = lastCacheDropOffset + CACHE_DROP_INTERVAL_BYTES;
          if (offset >= nextCacheDropOffset) { 
            //如果下一次读取数据的位置大于丢弃数据的位置， 则将读取数据位置前的数据全部丢弃
            long dropLength = offset - lastCacheDropOffset;
            ris.dropCacheBehindReads(block.getBlockName(), lastCacheDropOffset,
                dropLength, POSIX_FADV_DONTNEED);
            lastCacheDropOffset = offset;
          }
        }
      }

ReadaheadPool.readaheadStream()方法执行了一个预读取操作， 只有在上一次预读取的数据已经使用了一半时， 才会触发一次新的预读取。 新的预读取操作是通过在Datanode.readaheadPool线程池中创建一个ReadaheadRequestImpl任务来执行的。  
ReadaheadRequestImpl.run()方法的代码如下， 它通过调用fadvise()系统调用， 完成OS层面的预读取， 将数据放入操作系统的缓存中。

@Override
        public void run() { 
          if (canceled) return;
          // There's a very narrow race here that the file will close right at
          // this instant. But if that happens, we'll likely receive an EBADF
          // error below, and see that it's canceled, ignoring the error.
          // It's also possible that we'll end up requesting readahead on some
          // other FD, which may be wasted work, but won't cause a problem.
          try { 
            if (fd.valid()) { 
    
              //调用fadvise()系统调用完成预读取
              NativeIO.POSIX.getCacheManipulator().posixFadviseIfPossible(
                  identifier, fd, off, len, POSIX_FADV_WILLNEED);
            }
          } catch (IOException ioe) { 
            if (canceled) { 
              // no big deal - the reader canceled the request and closed
              // the file.
              return;
            }
            LOG.warn("Failed readahead on " + identifier,
                ioe);
          }
        }

## 5.3.发送数据块——sendBlock() ##

sendBlock()方法， 这个方法用于读取数据以及校验和， 并将它们发送到接收方。

整个发送的流程可以分为如下几步:  
■ 在刚开始读取文件时， 触发一次预读取， 预读取部分数据到操作系统的缓冲区中。  
■ 构造pktBuf缓冲区， 也就是能容纳一个数据包的缓冲区。 这里首先要确定的就是pktBuf缓冲区的大小， 最好是一个数据包的大小。  
对于两种不同的发送数据包的模式**transferTo**和**ioStream**， 缓冲区的大小是不同的。 在transfertTo模式中 ， 数据块文件是通过零拷贝方式直接传输给客户端的，并不需要将数据块文件写入缓冲区中， 所以pktBuf缓冲区只需要缓冲校验数据即可； 而ioStream模式则需要将实际数据以及校验数据都缓冲下来， 所以pktBuf大小是完全不同的。  
■ 接下来就是循环调用sendPacket()方法发送数据包序列， 直到offset>=endOffset，也就是整个数据块都发送完成了。 这里首先调用manageOsCache()进行预读取，然后循环调用sendPacket()依次将所有数据包发送到客户端。 最后更新offset——也就是数据游标， 更新seqno——记录已经发送了几个数据包。  
■ 发送一个空的数据包用以标识数据块的结束。  
■ 完成数据块发送操作之后， 调用close()方法关闭数据块以及校验文件， 并从操作系统的缓存中删除已读取的数据。

private long doSendBlock(DataOutputStream out, OutputStream baseStream,
            DataTransferThrottler throttler) throws IOException { 
        if (out == null) { 
          throw new IOException( "out stream is null" );
        }
        initialOffset = offset;
    
        long totalRead = 0;
    
        OutputStream streamForSendChunks = out;
        
        lastCacheDropOffset = initialOffset;
    
        if (isLongRead() && ris.getDataInFd() != null) { 
          // Advise that this file descriptor will be accessed sequentially.
          ris.dropCacheBehindReads(block.getBlockName(), 0, 0,
              POSIX_FADV_SEQUENTIAL);
        }
        //1． 将数据预读取至操作系统的缓存中
        // Trigger readahead of beginning of file if configured.
        manageOsCache();
    
        final long startTime = ClientTraceLog.isDebugEnabled() ? System.nanoTime() : 0;
    
        //2． 构造存放数据包（packet） 的缓冲区
        try { 
          int maxChunksPerPacket;
    
          // pktBufSize : 33
          int pktBufSize = PacketHeader.PKT_MAX_HEADER_LEN;
          boolean transferTo = transferToAllowed && !verifyChecksum
              && baseStream instanceof SocketOutputStream
              && ris.getDataIn() instanceof FileInputStream;
    
    
          if (transferTo) { 
            FileChannel fileChannel =  ((FileInputStream)ris.getDataIn()).getChannel();
            blockInPosition = fileChannel.position();
            streamForSendChunks = baseStream;
    
            // 这里的TRANSFERTO_BUFFER_SIZE大小默认是64KB
            // maxChunksPerPacket变量表明一个数据包中最多包含多少个校验块 : 128 个
            maxChunksPerPacket = numberOfChunks(TRANSFERTO_BUFFER_SIZE);
            //缓沖区中只存放校验数据 pktBufSize : 545
            // Smaller packet size to only hold checksum when doing transferTo
            pktBufSize += checksumSize * maxChunksPerPacket;
          } else { 
    
    
            //这里的IO—FILE_BUFFER_SIZE大小默认是4KB
            maxChunksPerPacket = Math.max(1, numberOfChunks(IO_FILE_BUFFER_SIZE));
            // Packet size includes both checksum and data
            //缓冲区存放校验数据以及实际数据
            pktBufSize += (chunkSize + checksumSize) * maxChunksPerPacket;
          }
          //构造缓沖区pktBuf : 545
          ByteBuffer pktBuf = ByteBuffer.allocate(pktBufSize);
    
          //循环调用sendPacket()发送packet
          while (endOffset > offset && !Thread.currentThread().isInterrupted()) { 
    
            manageOsCache();
            long len = sendPacket(pktBuf, maxChunksPerPacket, streamForSendChunks,
                transferTo, throttler);
            offset += len;
            totalRead += len + (numberOfChunks(len) * checksumSize);
            seqno++;
          }
          //如果当前线程被中断， 则不再发送完整的数据块
          // If this thread was interrupted, then it did not send the full block.
          if (!Thread.currentThread().isInterrupted()) { 
            try { 
              // 发送一个空的数据包用以标识数据块的结束
              // send an empty packet to mark the end of the block
              sendPacket(pktBuf, maxChunksPerPacket, streamForSendChunks, transferTo,
                  throttler);
              out.flush();
            } catch (IOException e) {  //socket error
              throw ioeToSocketException(e);
            }
    
            sentEntireByteRange = true;
          }
        } finally { 
          if ((clientTraceFmt != null) && ClientTraceLog.isDebugEnabled()) { 
            final long endTime = System.nanoTime();
            ClientTraceLog.debug(String.format(clientTraceFmt, totalRead,
                initialOffset, endTime - startTime));
          }
          // 调用close()文件关闭数据块文件、 校验文件以及回收操作系统缓冲区
          close();
        }
        return totalRead;
      }

## 5.4.发送数据包——sendPacket() ##

**sendPacket()也可以分为三个部分**

■ 首先计算数据包头域在pkt缓存中的位置headerOff， 再计算checksum在pkt中的位置checksumOff， 以及实际数据在pkt中的位置dataOff。 然后将数据包头域、 校验数据以及实际数据写入pkt缓存中。 如果verifyChecksum属性被设置为true， 则调用verifyChecksum()方法确认校验和数据正确。  
■ 接下来就是发送数据块了， 将pkt缓存中的数据写入IO流中。 这里要注意， 如果是transferT()方式， pkt中只有数据包头域以及校验数据， 实际数据则直接通过transferTo方式从文件通道（FileChannel） 直接写入IO流中。  
■ 使用节流器控制写入的速度

/** * Sends a packet with up to maxChunks chunks of data. * * @param pkt buffer used for writing packet data * @param maxChunks maximum number of chunks to send * @param out stream to send data to * @param transferTo use transferTo to send data * @param throttler used for throttling data transfer bandwidth */
      private int sendPacket(ByteBuffer pkt, int maxChunks, OutputStream out,
          boolean transferTo, DataTransferThrottler throttler) throws IOException { 
    
    
        int dataLen = (int) Math.min(endOffset - offset,
                                 (chunkSize * (long) maxChunks));
    
    
        //数据包中包含多少个校验块
        int numChunks = numberOfChunks(dataLen); // Number of chunks be sent in the packet
    
        //校验数据长度
        int checksumDataLen = numChunks * checksumSize;
    
        //数据包长度
        int packetLen = dataLen + checksumDataLen + 4;
    
        boolean lastDataPacket = offset + dataLen == endOffset && dataLen > 0;
    
        // The packet buffer is organized as follows:
        // _______HHHHCCCCD?D?D?D?
        // ^ ^
        // | \ checksumOff
        // \ headerOff
        // _ padding, since the header is variable-length
        // H = header and length prefixes
        // C = checksums
        // D? = data, if transferTo is false.
    
        //将数据包头域写入缓存中
        int headerLen = writePacketHeader(pkt, dataLen, packetLen);
    
        //数据包头域在缓存中的位置
        // Per above, the header doesn't start at the beginning of the buffer
        int headerOff = pkt.position() - headerLen;
    
        //校验数据在缓存中的位置
        int checksumOff = pkt.position();
    
        byte[] buf = pkt.array();
        
        if (checksumSize > 0 && ris.getChecksumIn() != null) { 
    
          //将校验数据写入缓存中
          readChecksum(buf, checksumOff, checksumDataLen);
    
          // write in progress that we need to use to get last checksum
          if (lastDataPacket && lastChunkChecksum != null) { 
    
    
            int start = checksumOff + checksumDataLen - checksumSize;
    
    
            byte[] updatedChecksum = lastChunkChecksum.getChecksum();
    
    
            if (updatedChecksum != null) { 
              System.arraycopy(updatedChecksum, 0, buf, start, checksumSize);
            }
          }
        }
        
        int dataOff = checksumOff + checksumDataLen;
    
    
        if (!transferTo) {  // normal transfer
          try { 
    
            //在普通模式下， 将实际数据写入缓存中
            ris.readDataFully(buf, dataOff, dataLen);
          } catch (IOException ioe) { 
            if (ioe.getMessage().startsWith(EIO_ERROR)) { 
              throw new DiskFileCorruptException("A disk IO error occurred", ioe);
            }
            throw ioe;
          }
    
          if (verifyChecksum) { 
    
            //确认校验和正确
            verifyChecksum(buf, dataOff, dataLen, numChunks, checksumOff);
          }
        }
        
        try { 
          //transferTo模式
          if (transferTo) { 
    
            //将头域和校验和写入输出流中
            SocketOutputStream sockOut = (SocketOutputStream)out;
    
            //使用transfer方式， 将数据从数据块文件直接零拷贝到IO流中
            // First write header and checksums
            sockOut.write(buf, headerOff, dataOff - headerOff);
    
            // no need to flush since we know out is not a buffered stream
            FileChannel fileCh = ((FileInputStream)ris.getDataIn()).getChannel();
    
    
            LongWritable waitTime = new LongWritable();
    
            LongWritable transferTime = new LongWritable();
    
    
            fileIoProvider.transferToSocketFully(
                ris.getVolumeRef().getVolume(), sockOut, fileCh, blockInPosition,
                dataLen, waitTime, transferTime);
    
    
            datanode.metrics.addSendDataPacketBlockedOnNetworkNanos(waitTime.get());
            datanode.metrics.addSendDataPacketTransferNanos(transferTime.get());
    
    
            blockInPosition += dataLen;
          } else { 
    
            //在正常模式下
            //将缓存中的所有数据（包括头域、 校验和以及实际数据） 写入输出流中
            // normal transfer
            out.write(buf, headerOff, dataOff + dataLen - headerOff);
          }
        } catch (IOException e) { 
          if (e instanceof SocketTimeoutException) { 
            /* * writing to client timed out. This happens if the client reads * part of a block and then decides not to read the rest (but leaves * the socket open). * * Reporting of this case is done in DataXceiver#run */
          } else { 
            /* Exception while writing to the client. Connection closure from * the other end is mostly the case and we do not care much about * it. But other things can go wrong, especially in transferTo(), * which we do not want to ignore. * * The message parsing below should not be considered as a good * coding example. NEVER do it to drive a program logic. NEVER. * It was done here because the NIO throws an IOException for EPIPE. */
            String ioem = e.getMessage();
            /* * If we got an EIO when reading files or transferTo the client socket, * it's very likely caused by bad disk track or other file corruptions. */
            if (ioem.startsWith(EIO_ERROR)) { 
              throw new DiskFileCorruptException("A disk IO error occurred", e);
            }
            if (!ioem.startsWith("Broken pipe") && !ioem.startsWith("Connection reset")) { 
              LOG.error("BlockSender.sendChunks() exception: ", e);
              datanode.getBlockScanner().markSuspectBlock(
                  ris.getVolumeRef().getVolume().getStorageID(),
                  block);
            }
          }
          throw ioeToSocketException(e);
        }
    
        if (throttler != null) { 
    
          // rebalancing so throttle
          //调整节流器
          throttler.throttle(packetLen);
        }
    
        return dataLen;
      }

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

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

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/34ac124081f94c4588018432250201bf.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/95f33368bd5b4b57a8601ffea71d49ea.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/01bf870270ad4cf3aa3293bee26c4c45.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3poYW5nbG9uZ180NDQ0_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/657650d37abe42ba8eb331d98307448e.png