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

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

第四节：namenode分析

4.4 namenode文件租约分析LeaseManagerMonitor

      文件租约就是将操作的文件和操作它的客户端进行绑定，若文件不存在一个租约，则说明该文件当前没有被任何客户端写，否则，就表示它正在被该文件租约中 的客户端holder写。这中间可能会发生一些意想不到的异常情况，比如正在对某个文件进行写操作的客户端突然宕机了，那么与这个文件相关的租约会迟迟得 不到客户端的续租而过期，那么NameNode会释放这些过期的租约，好让其它的客户端能及时的操作该租约对应文件。
      Lease内部类：一个用来表示租约的实体类org.apache.hadoop.hdfs.server.namenode.LeaseManager.Lease
      因为文件系统客户端需要向Namenode请求写数据（向Namenode结点写数据块），因此一个文件系统客户端必须持有一个Lease实例。该 Lease实例包含文件系统客户端的持有身份（客户端名称）、最后更新Lease时间（用于判断是否超时）、所要写数据的路径（应该与指定的 Datanode上文件系统的Path相关）。对于每一个文件系统客户端，都应该持有一个Lease，Lease管理一个客户端的全部锁（写锁），其中 Lease中包含的最后更新时间需要文件系统客户端周期地检查来实现更新，这样写数据才不会因为超时而放弃一个Lease。当然，如果一个文件系统客户端 发生故障，或者它不需要持有该Lease，也就是不需要执行文件的写操作，那么它会释放掉由它所持有的Lease管理的全部的锁，以便满足其它客户端的请求。
    Monitor内部线程类：一个内部线程类 org.apache.hadoop.hdfs.server.namenode.LeaseManager.Monitor，它是用来周期性地（每2s 检查一次），检查LeaseManager管理器所维护的Lease列表中是否有Lease过期的文件系统客户端，如果过期则从Lease列表中删除掉。
  基本可以清楚LeaseManager的是如何管理Lease，其中LeaseManager还提供了向它维护的列表中添加Lease、删除Lease、更新Lease等等操作。
  重要数据结构：

    //保存了LeaseHolder到Lease的映射。  
    private SortedMap<String, Lease> leases = new TreeMap<String, Lease>();  
    // Set of: Lease保存了所有lease   
    private SortedSet<Lease> sortedLeases = new TreeSet<Lease>();  
    //保存了path到Lease的映射。   
    private SortedMap<String, Lease> sortedLeasesByPath = new TreeMap<String, Lease>();  

（1）租约 Lease

     一个租约由一个holder（客户端名），lastUpdate（上次更新时间）和paths（该客户端操作的文件集合）构成。

    /************************************************************ 
      * A Lease governs all the locks held by a single client. 
      * For each client there's a corresponding lease, whose 
      * timestamp is updated when the client periodically 
      * checks in.  If the client dies and allows its lease to 
      * expire, all the corresponding locks can be released. 
      *  
      * 一个租约由一个holder（客户端名），lastUpdate（上次更新时间）和paths（该客户端操作的文件集合）构成 
      *  
      *************************************************************/  
     class Lease implements Comparable<Lease> {  
       private final String holder;  
       private long lastUpdate;  
       private final Collection<String> paths = new TreeSet<String>();  
     。。。  

 

（2）过期检查

NameNode通过后台工作线程LeaseManager$Monitor来定期检查LeaseManager中的文件租约是否过期，如果过期就释放该文件租约。

    /****************************************************** 
       * Monitor checks for leases that have expired, 
       * and disposes of them. 
       ******************************************************/  
      class Monitor implements Runnable {  
        final String name = getClass().getSimpleName();  
      
        /** Check leases periodically. */  
        public void run() {  
          for(; fsnamesystem.isRunning(); ) {  
            synchronized(fsnamesystem) {  
              //检查租约是否过期，过期则internalReleaseLeaseOne和removeLease  
              checkLeases();  
            }  
      
            try {  
              Thread.sleep(2000);  
            } catch(InterruptedException ie) {  
              if (LOG.isDebugEnabled()) {  
                LOG.debug(name + " is interrupted", ie);  
              }  
            }  
          }  
        }  
      }  

 

 

（3）释放租约

 

      租约有两个超时时间，一个被称为软超时（1分钟），另一个是硬超时（1小时）。如果租 约软超时，那么就会触发internalReleaseLease方法。检查src对应的INodeFile，如果不存在，不处于构造状态，返回；文件处 于构造状态，而文件目标DataNode为空，而且没有数据块，则finalize该文件（该过程在completeFileInternal中已经讨论 过，租约在过程中被释放），并返回；文件处于构造状态，而文件目标DataNode为空，数据块非空，则将最后一个数据块存放的DataNode目标取出 （在BlocksMap中），然后设置为文件现在的目标DataNode；调用 INodeFileUnderConstruction.assignPrimaryDatanode，该过程会挑选一个目前还活着的DataNode， 作为租约的主节点，
并把<block，block目标DataNode数组>加到该DataNode的recoverBlocks队 列中；更新租约。上面分析了租约软超时的情况下NameNode发生租约恢复的过程。DataNode上收到这个命令后，将会启动一个新的线程，该线程为 每个Block调用recoverBlock方法：recoverBlock(blocks[i], false, targets[i], true)。

 

下面我们从客户端写文件开始分析租约创建到移除的过程
第一步：客户端创建了DFSOutputStream文件流之后，然后把文件路径名和对应的文件流存入LeaseChecker中。

public FSDataOutputStream create(Path f, FsPermission permission,  
    boolean overwrite,  
    int bufferSize, short replication, long blockSize,  
    Progressable progress) throws IOException {  
  
    statistics.incrementWriteOps(1);  
    return new FSDataOutputStream  
       (dfs.create(getPathName(f), permission,  
                   overwrite, true, replication, blockSize, progress, bufferSize),  
        statistics);  
  }  
  
public OutputStream create(String src,   
                             FsPermission permission,  
                             boolean overwrite,   
                             boolean createParent,  
                             short replication,  
                             long blockSize,  
                             Progressable progress,  
                             int buffersize  
                             ) throws IOException {  
    checkOpen();  
    if (permission == null) {  
      permission = FsPermission.getDefault();  
    }  
    FsPermission masked = permission.applyUMask(FsPermission.getUMask(conf));  
    LOG.debug(src + ": masked=" + masked);  
    OutputStream result = new DFSOutputStream(src, masked,  
        overwrite, createParent, replication, blockSize, progress, buffersize,  
        conf.getInt("io.bytes.per.checksum", 512));  
    leasechecker.put(src, result);  
    return result;  
  }  

  关于LeaseChecker：如果是第一次leasechecker.put(src, result)，将会启动LeaseChecker这个守护线程）。这个过程是一个同步调用的过程。文件create成功之后，守护线程 LeaseChecker会每30s, 通过rpc调用renew一下 该DFSClient所拥有的lease。

 

 

第二步：新增租约
      Client与NameNode通信，调用NameNode中的create方 法，NameNode的create方法调用FSNameSystem的startFile方法，startFile调用 startFileInternal方法，startFileInternal检查src对应的INodeFIle是否存在，如果存在检查该 INodeFIle的租约情况，如果有client持有该INodeFile租约不同于操作发起的client，检查租约是否过期 (expiredSoftLimit)，如果过期调用internalReleaseLease方法。如果不存在该文件则新增文件并添加租约。 LeaseManager的线程Monitor会循环检测该租约是否过期。创建一个新的文件，状态为under construction，没有任何data block与之对应。
INodeFileUnderConstruction newNode = dir.addFile(src, permissions,replication, blockSize, holder, clientMachine, clientNode, genstamp);
新增租约
leaseManager.addLease(newNode.clientName, src);

 

 

第三步：完成文件写入，移除租约

      DFSClient调用ClientProtocol的远程方法complete，向NameNode节点解除文件src的租约。
释放租约；

    //将对应的INodeFileUnderConstruction对象转换为INodeFile对象，并在FSDirectory进行替换；  
    //调用FSDirectory.closeFile关闭文件，其中会写日志logCloseFile(path, file)。  
    //检查副本数，如果副本数小于INodeFile中的目标数，那么添加数据块复制任务。  
    private void finalizeINodeFileUnderConstruction(String src,  
    INodeFileUnderConstruction pendingFile) throws IOException {  
        NameNode.stateChangeLog.info("Removing lease on  file " + src +   
                                     " from client " + pendingFile.clientName);  
        leaseManager.removeLease(pendingFile.clientName, src);  
      
        // The file is no longer pending.  
        // Create permanent INode, update blockmap  
        INodeFile newFile = pendingFile.convertToInodeFile();  
        dir.replaceNode(src, pendingFile, newFile);  
      
        // close file and persist block allocations for this file  
        dir.closeFile(src, newFile);  
      
        checkReplicationFactor(newFile);  
      }