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

qazxsw1

第七章：小朱笔记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, false, targets, 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);  
}