手机版
你好,游客 登录 注册 搜索
背景:
阅读新闻

HDFS上块恢复机制

[日期:2013-05-13] 来源:Linux社区  作者:ckl_soft [字体: ]

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);
      }
    }
  }

linux
相关资讯       HDFS 
本文评论   查看全部评论 (0)
表情: 表情 姓名: 字数

       

评论声明
  • 尊重网上道德,遵守中华人民共和国的各项有关法律法规
  • 承担一切因您的行为而直接或间接导致的民事或刑事法律责任
  • 本站管理人员有权保留或删除其管辖留言中的任意内容
  • 本站有权在网站内转载或引用您的评论
  • 参与本评论即表明您已经阅读并接受上述条款