第七章：小朱笔记hadoop之源码分析-hdfs分析 第九节：block Recovery过程分析

4591566

第七章：小朱笔记hadoop之源码分析-hdfs分析

第九节：block Recovery过程分析

Lease Recovery Algorithm lease recovery算法：

1) Namenode retrieves lease information
name node查找到lease的信息
2) For each file f in the lease, consider the last block b off
对于lease中的每一个文件，获取其最后一个block b进行以下处理
2.1)  Get the datanodes which contains b
获取包含block b 的全部data node
2.2)  Assign one of the datanodes as the primary datanode p
获取一个data node作为其primary data node
2.3)  p obtains a new generation stamp form the namenode
primary datanode向 nn获取一个新的generation stamp
2.4)  p get the block info from each datanode
primary datanode 向每一个dn获取block info
2.5)  p computes the minimum block length
primary datanode 计算最小块的长度
2.6) p updates the datanodes, which have a valid generation stamp,with the new generation stamp and the minimum block length
primary datanode 用最小块的长度和新生成的 genetation stamp来更新 dn
2.7)  p acknowledges the namenode the update results
primary datanode  ack nn update的结果
2.8)  Namenode updates the BlockInfo
nn 更新 block info
2.9)  Namenode removes f from the lease and removes the lease once all files have been removed
nn 删除文件 f 的lease
2.10) Namenode commit changes to edit log
nn向edit log提交 lease 这个change
      一个client在持有某个文件的Lease情况下，如果写入数据过程中发生宕机，或者其他事故，导致无法继续对文件进行写入。由于该文件的Lease是由namenode来维护的，此时namenode认为该文件正在被该client持有，所以其他client对该文件是不允许进行写入的。
      为了解决上面的问题，namenode中对某个client对应某个文件的Lease是有一个限期的，一旦过了这 个限期，该Lease没有发生任何改变（比如更新时间），没有写入任何数据，那么namenode就认为该lease对应的client发生了异常，需要在 namenode端对这个Lease进行释放，以便其他的client能够对文件进行写入操作。释放Lease的时候会处理正在写入的文件，把该文件的最后一个block和targets datanode数组加入到需要recovery的队列中，进行处理之后等待目标datanode心跳获取该数据。
     datanode获取了需要recovery的block的数据，会遍历targets datanode进行recovery操作，recovery结束会按照最小块的长度进行截取、更新块信息和元信息。
 
第一步：namenode租约检查，准备recovery block 数据
LeaseManager.Monitor.checkLeases 关于租约过程有专门一节讲解。
（1）从namenode内存中找到该filePath对应的文件INode，通常这个时候该INode是一个INodeFileUnderConstruction的实例，表示这个文件是正在被写入，还没有complete的一个文件。
（2）如果该文件的Targets为空，且该文件的block队列为空，表示这个文件是个空文件，那么直接将该文件complete，删除其对应的lease记录，然后返回
（3）如果该文件的Targets为空，且该文件的block队列非空，那么获取该文件的block队列，并找到该队列的最后一个block，将该block的最后一个block对应的 datanode设置为该文件的targets，这个操作的原因在于：由于HDFS的文件只能向最后一个block写入输入，所以lease过期肯定是出 了最后一个block有问题外，其他block应该都是完整的，所以获取最后一个block。而targets表示最后一个block应该保存在哪几个 datanode上，该targets是一个datanode队列，也就是说，namenode知道这最后一个block是在这几台datanode 上，以便向这几个datanode发送block recovery命令
（4）在targets队列中选择一个datanode作为primary datanode，把该文件的最后一个block和targets datanode数组加入到primary datanode的recoverBlocks中。重点关注recoverBlocks这个队列
（5）修改Lease Holder为'NN_Recovery'并续组
 


//源码一：
if(pendingFile.getTargets()==null||pendingFile.getTargets().length== 0){
if (pendingFile.getBlocks().length == 0) {
finalizeINodeFileUnderConstruction(src, pendingFile);
return;
}
// setup the Inode.targets for the last block from the blocksMap
//最后一个block对应的datanode设置为该文件的targets
Block[] blocks = pendingFile.getBlocks();
Block last = blocks[blocks.length-1];
DatanodeDescriptor[] targets =  new DatanodeDescriptor[blocksMap.numNodes(last)];
Iterator<DatanodeDescriptor> it = blocksMap.nodeIterator(last);
for (int i = 0; it != null && it.hasNext(); i++) {
targets = it.next();
}
pendingFile.setTargets(targets);
}
//指定primary datanode，start lease recovery of the last block for this file.
pendingFile.assignPrimaryDatanode();
//再分配Lease 注意Holder NN_Recovery
Lease reassignedLease = reassignLease(lease, src, HdfsConstants.NN_RECOVERY_LEASEHOLDER, pendingFile);
//续约
leaseManager.renewLease(reassignedLease);
//源码2：
//将该INodeFileUnderConstruction文件分配给指定的客户端进程，也就是执行租约恢复的操作为该文件初始化租约的恢复的处理（存储选择的主Datanode所激活的块列表） ，该过程会挑选一个目前还活着的DataNode，作为租约的主节点，并把<block，block目标     DataNode数组>加到该DataNode的recoverBlocks队列中；
void assignPrimaryDatanode() {
//assign the first alive datanode as the primary datanode
// 指派第一个活跃的为主Datanode结点  
if (targets.length == 0) {
NameNode.stateChangeLog.warn("BLOCK*"
+ " INodeFileUnderConstruction.initLeaseRecovery:"
+ " No blocks found, lease removed.");
}
int previous = primaryNodeIndex;  
//primaryNodeIndex初始值是-1，它用来保证每次找到的primary不在同一个位置
// 从索引previous开始查找到一个活跃的Datanode进程  
for(int i = 1; i <= targets.length; i++) {  
int j = (previous + i)%targets.length;  
if (targets[j].isAlive) { // 保证第j个Datanode处于活跃状态  
DatanodeDescriptor primary = targets[primaryNodeIndex = j];  
//把该文件的最后一个block和targets datanode数组加入到
//primary datanode 的recoverBlocks中
primary.addBlockToBeRecovered(blocks[blocks.length - 1], targets);
//存储被主Datanode激活的块，实际存储到该Datanode的块队列中  
NameNode.stateChangeLog.info("BLOCK* " + blocks[blocks.length - 1] + " recovery started, primary=" + primary);  
return;  
}  
}  
}

 
第二步：datanode心跳获取recovery block
     在 HDFS中，datanode是会每隔几秒钟向namenode定期的发送心跳，namenode会返回给datanode一个“命令集(cmds)”的，这个命令集就是namenode需要 datanode执行的某些操作，比如
    （1）将该datanode上的某个block拷贝到其他datanode上去的DNA_TRANSFER命令
    （2）将该datanode上的某个block从物理磁盘上删除的DNA_TRANSFER命令
    （3）停止该datanode的DNA_SHUTDOWN命令
    （4）对某个block进行block recovery的DNA_RECOVERBLOCK命令
         

cmd = nodeinfo.getLeaseRecoveryCommand(Integer.MAX_VALUE);
       从上可以看出，从heartbeat的返回命令集中，就包括了对某个block进行recovery的命令。所以，datanode某个block进行 recovery操作的动作，实际上是来自namenode的指令。也就是说，namenode认为这个block需要做recovery了，并且这个 block在某几个datanode上保存，那么namenode就会在这几个datanode的heartbeat发送过来后，给这几个 datanode返回指令集，指令集中就包括对这个block进行recovery的指令。于是datanode接受到这个指令后，对block进行数据 本身的recovery操作。
    注意： BlockCommand数据结构，它存储了需要执行的操作（action）以及这个操作涉及的block和这个block所对应的所有的datanode。
 

public class BlockCommand extends DatanodeCommand {
Block blocks[];
DatanodeInfo targets[][];
}
第三步：datanode执行 block recovery
datanode上最终会调用recoverBlock方法，此时closeFile=true。它是recoverLease发起的，此时要关闭文件，并使得这个文件的block在datanode上的信息一致。
 

Public Daemon recoverBlocks(final Block[] blocks,final DatanodeInfo[][]targets){
Daemon d = new Daemon(threadGroup, new Runnable() {
/** Recover a list of blocks. It is run by the primary datanode. */
public void run() {
for(int i = 0; i < blocks.length; i++) {
try {
logRecoverBlock("NameNode", blocks, targets);
recoverBlock(blocks, false, targets, true);
} catch (IOException e) {
LOG.warn("recoverBlocks FAILED, blocks[" + i + "]=" + blocks, e);
}
}
}
});
d.start();
return d;
}
 （1）由于block recovery 是由primary datanode发起，但该recovery操作需要在三个datanode上对该block进行操作（假设文件副本为3），所以primary datanode接收到命令的时候同时还收到了该block的targets datanode数组（其中就包括该datanode自身）
（2）primary datanode遍历targets datanode数组，对每一个datanode，向其发送一个start block recovery的指令。判断本地调用还是远程RPC，如果是其自身，则直接执行该指令。
（3）start block recovery指令会在datanode的磁盘中找到该block的物理块，并确认该block对应的验证信息和meta信息正确，并返回一个BlockRecord对象，表示这个block正在被recovery。
（4）对每个BlockRecord，查看keepLength标志位是否为true，如果为true，则只recovery blocksize 跟 namenode中记录的blocksize一致的block，否则全部都算。并且block的size为BlockRecord最小的。
（5）对每个物理块，一旦真正开始recovery操作，则进行如下操作：在该datanode上找到该block，同时找到这个block对应的meta文件 （每一个block都对应一个meta文件，用来记录该block的验证码等原信息），更新该block的stamp号（表示该block已经被修改过一 次）
 
（6）如果需要recovery成的block的size 小于实际的block的size，则将实际的block截断成其需要的大小，并更新meta文件和验证信息。
 
第四步：datanode处理recovery结果
primary datanode调用各个BlockRecord对应的datanode进行Block同步，然后向namenode提交块同步信息。

for(BlockRecord r : syncList) {
try {
r.datanode.updateBlock(r.info.getBlock(), newblock, closeFile);
successList.add(r.id);
} catch (IOException e) {
InterDatanodeProtocol.LOG.warn("Failed to updateBlock (newblock="
+ newblock + ", datanode=" + r.id + ")", e);
}
}
namenode.commitBlockSynchronization(block,newblock.getGenerationStamp(), newblock.getNumBytes(), closeFile, false,nlist);
 
（1）updateBlock 更新Block
  updateBlock的最外层是一个死循环，循环的结束条件，是没有任何和这个数据块相关的写线程。每次循环，updateBlock都会去调用一个叫tryUpdateBlock的内部方法。tryUpdateBlock发现已经没有线程在写这个块，就会跟新和这个数据块相关的信息，包括元文件和内存中的映射表volumeMap。如果tryUpdateBlock发现还有活跃的线程和该块关联，那么，updateBlock会试图结束该线程，并等在join上等待。

public void updateBlock(Block oldblock, Block newblock) throws IOException {
if (oldblock.getBlockId() != newblock.getBlockId()) {
throw new IOException("Cannot update oldblock (=" + oldblock
+ ") to newblock (=" + newblock + ").");
}

// Protect against a straggler updateblock call moving a block backwards
// in time.
boolean isValidUpdate =
(newblock.getGenerationStamp() > oldblock.getGenerationStamp()) ||
(newblock.getGenerationStamp() == oldblock.getGenerationStamp() &&
newblock.getNumBytes() == oldblock.getNumBytes());
if (!isValidUpdate) {
throw new IOException(
"Cannot update oldblock=" + oldblock +
" to newblock=" + newblock + " since generation stamps must " +
"increase, or else length must not change.");
}

for(;;) {
final List<Thread> threads = tryUpdateBlock(oldblock, newblock);
if (threads == null) {
return;
}
interruptAndJoinThreads(threads);
}
}
 
 

  /**
* Try to update an old block to a new block.
* If there are ongoing create threads running for the old block,
* the threads will be returned without updating the block.
* 用于将旧块截断成新块(调用truncateBlock)，并截断相应的元数据文件，以及更新ongoingCreates、volumeMap。
* @return ongoing create threads if there is any. Otherwise, return null.
*/
private synchronized List<Thread> tryUpdateBlock(Block oldblock, Block newblock) throws IOException {
//check ongoing create threads 获取与此块相关的文件及访问这个块的线程
ArrayList<Thread> activeThreads = getActiveThreads(oldblock);
if (activeThreads != null) {
return activeThreads; //如果近期有对此块进行操作，返回存活的操作线程
}
//近期无对此块操作的线程，就更新块  
////获得旧块的文件  
//No ongoing create threads is alive.  Update block.
File blockFile = findBlockFile(oldblock.getBlockId());
if (blockFile == null) {
throw new IOException("Block " + oldblock + " does not exist.");
}
File oldMetaFile = findMetaFile(blockFile);
long oldgs = parseGenerationStamp(blockFile, oldMetaFile);
// First validate the update
//update generation stamp
//旧块stamp比新块stamp大，不合法  
if (oldgs > newblock.getGenerationStamp()) {
throw new IOException("Cannot update block (id=" + newblock.getBlockId()
+ ") generation stamp from " + oldgs
+ " to " + newblock.getGenerationStamp());
}
//update length
//新块的大小大于旧块的大小  
if (newblock.getNumBytes() > oldblock.getNumBytes()) {
throw new IOException("Cannot update block file (=" + blockFile
+ ") length from " + oldblock.getNumBytes() + " to " + newblock.getNumBytes());
}
// Now perform the update
//rename meta file to a tmp file
//旧块元数据文件重命名  
File tmpMetaFile = new File(oldMetaFile.getParent(),oldMetaFile.getName()+"_tmp" + newblock.getGenerationStamp());
if (!oldMetaFile.renameTo(tmpMetaFile)){
throw new IOException("Cannot rename block meta file to " + tmpMetaFile);
}
//新块的大小小于旧块的大小 截断旧块和旧块的元数据文件
if (newblock.getNumBytes() < oldblock.getNumBytes()) {
truncateBlock(blockFile, tmpMetaFile, oldblock.getNumBytes(), newblock.getNumBytes());
}
//rename the tmp file to the new meta file (with new generation stamp)
File newMetaFile = getMetaFile(blockFile, newblock);
if (!tmpMetaFile.renameTo(newMetaFile)) {
throw new IOException("Cannot rename tmp meta file to " + newMetaFile);
}
updateBlockMap(ongoingCreates, oldblock, newblock);
updateBlockMap(volumeMap, oldblock, newblock);
// paranoia! verify that the contents of the stored block
// matches the block file on disk.
validateBlockMetadata(newblock);
return null;
}
//truncateBlock对旧块blockFile和对应的元数据文件metaFile进行截断，截断后旧块长度为newlen(newlen<oldlen)。  
static void truncateBlock(File blockFile, File metaFile,long oldlen, long newlen) throws IOException {
if (newlen == oldlen) {
return;
}
if (newlen > oldlen) {
throw new IOException("Cannot truncate block to from oldlen (=" + oldlen
+ ") to newlen (=" + newlen + ")");
}
if (newlen == 0) {
// Special case for truncating to 0 length, since there's no previous
// chunk.
RandomAccessFile blockRAF = new RandomAccessFile(blockFile, "rw");
try {
//truncate blockFile
blockRAF.setLength(newlen);   
} finally {
blockRAF.close();
}
//update metaFile
RandomAccessFile metaRAF = new RandomAccessFile(metaFile, "rw");
try {
metaRAF.setLength(BlockMetadataHeader.getHeaderSize());
} finally {
metaRAF.close();
}
return;
}
//由于只是对就块进行截断，所有新块的最后一个校验和字段可能在旧块中不一样，  
//所有setLength进行截断后，要读取最后一个校验和字段  
DataChecksum dcs = BlockMetadataHeader.readHeader(metaFile).getChecksum();
int checksumsize = dcs.getChecksumSize();
int bpc = dcs.getBytesPerChecksum();
long newChunkCount = (newlen - 1)/bpc + 1;//校验和的段数  
long newmetalen = BlockMetadataHeader.getHeaderSize() + newChunkCount*checksumsize;//新的校验和文件的长度  
long lastchunkoffset = (newChunkCount - 1)*bpc;//最后一个校验和字段的偏移位置  
int lastchunksize = (int)(newlen - lastchunkoffset); //最后一个校验和的开始位置  
byte[] b = new byte[Math.max(lastchunksize, checksumsize)];
RandomAccessFile blockRAF = new RandomAccessFile(blockFile, "rw");//对旧块进行读取  
try {
//truncate blockFile
blockRAF.setLength(newlen);
//read last chunk
blockRAF.seek(lastchunkoffset);
blockRAF.readFully(b, 0, lastchunksize);
} finally {
blockRAF.close();
}
//compute checksum
dcs.update(b, 0, lastchunksize);
dcs.writeValue(b, 0, false);
//update metaFile
RandomAccessFile metaRAF = new RandomAccessFile(metaFile, "rw");
try {
metaRAF.setLength(newmetalen);
metaRAF.seek(newmetalen - checksumsize);
metaRAF.write(b, 0, checksumsize);
} finally {
metaRAF.close();
}
}
 
 
（2） commitBlockSynchronization
  参数分别是block，数据块；newgenerationstamp，新的时间戳；newlength，新长度；closeFile，是否关闭文件，deleteblock，是否删除文件；newtargets，新的目标列表。
  处理流程：
  参数检查，获取对应的文件，记为pendingFile；从BlocksMap中删除老的信息；如果deleteblock为true，从pendingFile删除Block记录；否则，更新Block的信息；如果不关闭文件，那么写日志保存更新，返回；最后如果关闭文件的话，调用finalizeINodeFileUnderConstruction。