HDFS上块恢复机制

Hadoop的一致性模型：一次写，多次读。当一个文件被创建-写入-关闭之后，只能执行append操作。任意时刻只运行一个client对一个文件进行write，却可以允许多个client对一个文件进行append操作(类似于线程的并发)。lease是为了处理wirte操作的同步问题而使用的。
 hdfs支持write-once-read-many，也就是说不支持并行写，那么对读写的互斥同步就是靠Lease实现的。Lease是一个有时间约束的锁。客户端写文件时需要先申请一个Lease，对应到namenode中的LeaseManager，客户端的client name就作为一个lease的holder，即租约持有者。LeaseManager维护了文件的path与lease的对应关系，还有clientname->lease的对应关系。LeaseManager中有两个时间限制：softLimit and hardLimit。

 软限制就是写文件时规定的租约超时时间，硬限制则是考虑到文件close时未来得及释放lease的情况强制回收租约。
 软租约的超时检测则在DFSClient的LeaseChecker中进行。
 

当客户端（DFSClient）create一个文件的时候，会通过RPC 调用 namenode 的create方法来创建文件(在DFSOutputStream构造函数)。进而又调用FSNameSystem的startFile方法，又调用 LeaseManager 的addLease方法为新创建的文件添加一个lease。如果lease已存在，则更新该lease的lastUpdate （最近更新时间）值，并将该文件的path对应到该lease上。之后DFSClient 将该文件的path 添加 LeaseChecker中。文件创建成功后，守护线程LeaseChecker会每隔一定时间间隔调用namenode.renewLease(clientName)刷新NN上此DFSClient的租约。
 软限制超时之后，会跟硬限制超时一样，调用internalReleaseLease进行处理。

LeaseManager中还有一个Monitor线程来检测Lease是否超过hardLimit。租约硬限制(1小时)的情况下NameNode发生租约恢复的过程：
 LeaseManager的监视线程Monitor定时调用checkLeases检查当前租约列表中最早的文件有没租约过期，如果过期了就调用FSNamesystem.internalReleaseLease对此文件进行租约释放：
 首先找此文件的InodeFileUnderConstruction，然后找到此文件的最后一个块和此块所在的所有DN，然后调用assignPrimaryDatanode在这些DN中选出一个DN进行块恢复，将此块加入到选出的DN的recoverBlocks中。
 NN在接收到DN发过来的心跳时，会检查该DN的recoverBlocks上是否有需要进行恢复的块。有的话就给DN回送一个类型为DatanodeProtocol.DNA_RECOVERBLOCK的DatanodeCommand，并附上要<恢复的块,块所在的Datanode集>
 2、DN通过RPC机制调用DatanodeCommand[] cmds = namenode.sendHeartbeat(...)得到DatanodeCommand后，在processCommand里调用recoverBlocks(bcmd.getBlocks(), bcmd.getTargets());对这些指令进行处理。
 如果是DNA_RECOVERBLOCK消息，就启动一个recoverBlocks线程处理块进行恢复。
 线程是通过调用recoverBlock方法对块进行恢复：
 首先查看此块是否在ongoingRecovery内，如果在表示已经有在对此块进行恢复，直接返回。否则将这个块插入到ongoingRecovery内。然后找出块所在的所有DN中，时间戳比此块新的DN，将这些DN放到syncList中。查找的过程中也记录最小的块的长度。如果不保存原块的长度(keepLength=false)就将要同步的块的长度设为这个最小的长度block.setNumBytes(minlength)，然后调用syncBlock对syncList上的每个DN更新此块。

 private LocatedBlock recoverBlock(Block block, boolean keepLength,
      DatanodeID[] datanodeids, boolean closeFile) throws IOException {
    // If the block is already being recovered, then skip recovering it.
    // This can happen if the namenode and client start recovering the same
    // file at the same time.
    synchronized (ongoingRecovery) {
      Block tmp = new Block();
      tmp.set(block.getBlockId(), block.getNumBytes(), GenerationStamp.WILDCARD_STAMP);
      if (ongoingRecovery.get(tmp) != null) {//此要恢复的块已经在ongoingRecovery上，即已经处于恢复过程
        String msg = "Block " + block + " is already being recovered, " +
                    " ignoring this request to recover it.";
        LOG.info(msg);
        throw new IOException(msg);
      }
      ongoingRecovery.put(block, block);//没在ongoingRecovery中，就put进去
    }
    try {
      List<BlockRecord> syncList = new ArrayList<BlockRecord>();//建立需要sync块的节点列表
      long minlength = Long.MAX_VALUE;
      int errorCount = 0;

      //check generation stamps
      for(DatanodeID id : datanodeids) {//遍历该块所在的DNs，找出
        try {
         //获取到DatanodeId的协议，这是一个D-D间的协议
          InterDatanodeProtocol datanode = dnRegistration.equals(id)/*如果此id就是本Datanode*/?
              this: DataNode.createInterDataNodeProtocolProxy(id, getConf());
          BlockMetaDataInfo info = datanode.getBlockMetaDataInfo(block);//获得该block在DN上的元数据信息
          if (info != null && info.getGenerationStamp() >= block.getGenerationStamp()) {//这个块更新
            if (keepLength) {//块长度保持不变
              if (info.getNumBytes() == block.getNumBytes()) {
                syncList.add(new BlockRecord(id, datanode, new Block(info)));
              }
            }
            else {//块长度可变
              syncList.add(new BlockRecord(id, datanode, new Block(info)));//将这个块加到要同步的列表syncList中去
              if (info.getNumBytes() < minlength) {
                minlength = info.getNumBytes();//更新这些副本的最短长度
              }
            }
          }
        } catch (IOException e) {
          ++errorCount;
          InterDatanodeProtocol.LOG.warn(
              "Failed to getBlockMetaDataInfo for block (=" + block
              + ") from datanode (=" + id + ")", e);
        }
      }

      if (syncList.isEmpty() && errorCount > 0) {//这些DN上都没有需要更新的副本
        throw new IOException("All datanodes failed: block=" + block
            + ", datanodeids=" + Arrays.asList(datanodeids));
      }
      if (!keepLength) {
        block.setNumBytes(minlength);//块大小设为所有副本中最小的
      }
      return syncBlock(block, syncList, closeFile);//进行同步更新syncList上的DN
    } finally {
      synchronized (ongoingRecovery) {
        ongoingRecovery.remove(block);
      }
    }
  }