Hadoop二次开发必备，Hadoop源码分析（十三）

继续对NameNode实现的接口做分析。public DatanodeCommand blockReport(DatanodeRegistration nodeReg,
                                     long[] blocks) throws IOExceptionDataNode向NameNode报告它拥有的所有数据块，其中，参数blocks包含了数组化以后数据块的信息。FSNamesystem.processReport处理这个请求。一番检查以后，调用DatanodeDescriptor的reportDiff，将上报的数据块分成三组，分别是：l          删除：其它情况；
l          加入：BlocksMap中有数据块，但目前的DatanodeDescriptor上没有对应信息；
l          使无效：BlocksMap中没有找到数据块。对于删除的数据块，调用removeStoredBlock，这个方法我们前面已经分析过啦。对应需要加入的数据块，调用addStoredBlock方法，处理流程如下：
l          从BlocksMap获取现在的信息，记为storedBlock；如果为空，返回；
l          记录block和DatanodeDescriptor的关系；
l          新旧数据块记录不是同一个（我们这个流程是肯定不是啦）：1.      如果现有数据块长度为0，更新为上报的block的值；
2.      如果现有数据块长度比新上报的长，invalidateBlock（前面分析过，很简单的一个方法）当前数据块；
3.      如果现有数据块长度比新上报的小，那么会删除所有老的数据块（还是通过invalidateBlock），并更新BlocksMap中数据块的大小信息；4.      跟新可用存储空间等信息；
l          根据情况确定数据块需要复制的数目和目前副本数；
l          如果文件处于构建状态或系统现在是安全模式，返回；
l          处理当前副本数和文件的目标副本数不一致的情况；
l          如果当前副本数大于系统设定门限，开始删除标记为无效的数据块。还是给个流程图吧：





对于标记为使无效的数据块，调用addToInvalidates方法，很简单的方法，直接加到FSNamesystem的成员变量recentInvalidateSets中。

public voidblockReceived(DatanodeRegistration registration,
                            Blockblocks[],

                            String[] delHints)

DataNode可以通过blockReceived，向NameNode报告它最近接受到的数据块，同时给出如果数据块副本数太多时，可以删除数据块的节点（参数delHints）。在DataNode中，这个信息是通过方法notifyNamenodeReceivedBlock，记录到对应的列表中。




NameNode上的处理不算复杂，对输入参数进行检查以后，调用上面分析的addStoredBlock方法。然后在PendingReplicationBlocks对象中删除相应的block。   public voiderrorReport(DatanodeRegistration registration,                          int errorCode,
                          String msg)向NameNode报告DataNode上的一个错误，如果错误是硬盘错，会删除该DataNode，其它情况只是简单地记录收到一条出错信息。  publicNamespaceInfo versionRequest() throws IOException;从NameNode上获取NamespaceInfo，该信息用于构造DataNode上的DataStorage。


UpgradeCommandprocessUpgradeCommand(UpgradeCommand comm) throws IOException;我们不讨论。
public voidreportBadBlocks(LocatedBlock[] blocks) throws IOException

报告错误的数据块。NameNode会循环调用FSNamesystem的markBlockAsCorrupt方法。处理流程不是很复杂，找对应的INodeFile，如果副本数够，那么调用invalidateBlock，使该DataNode上的Block无效；如果副本数不够，加Block到CorruptReplicasMap中，然后准备对好数据块进行复制。

目前为止，我们已经完成了NameNode上的ClientProtocol和DatanodeProtocol的分析了，NamenodeProtocol我们在理解从NameNode的时候，才会进行分析。

---------------------------------------------------------------------------------------------------------------------------------------------------

除了对外提供的接口，NameNode上还有一系列的线程，不断检查系统的状态，下面是这些线程的功能分析。
在NameNode中，定义了如下线程：

  Daemon hbthread = null;   // HeartbeatMonitor thread

  public Daemon lmthread = null;   // LeaseMonitor thread

  Daemon smmthread = null;  // SafeModeMonitor thread
  public Daemon replthread = null;  // Replication thread

  private Daemon dnthread = null;

PendingReplicationBlocks中也有一个线程：
  Daemon timerThread = null;
NameNode内嵌的HTTP服务器中自然也有线程，这块我们就不分析啦。
  HttpServer infoServer;
心跳线程用于对DataNode的心态进行检查，以间隔heartbeatRecheckInterval运行heartbeatCheck方法。如果在一定时间内没收到DataNode的心跳信息，我们就认为该节点已经死掉，调用removeDatanode（前面分析过）将DataNode标记为无效。
租约lmthread用于检查租约的硬超时，如果租约硬超时，调用前面分析过的internalReleaseLease，释放租约。
smmthread运行的SafeModeMonitor我们前面已经分析过了。
replthread运行ReplicationMonitor，这个线程会定期调用computeDatanodeWork和processPendingReplications。
computeDatanodeWork会执行computeDatanodeWork或computeInvalidateWork。computeDatanodeWork从neededReplications中扫描，取出需要复制的项，然后：


l           检查文件不存在或者处于构造状态；如果是，从队列中删除复制项，退出对复制项的处理（接着处理下一个）；l           得到当前数据块副本数并选择复制的源DataNode，如果空，退出对复制项的处理；
l           再次检查副本数（很可能有DataNode从故障中恢复），如果发现不需要复制，从队列中删除复制项，退出对复制项的处理；
l           选择复制的目标，如果目标空，退出对复制项的处理；
l           将复制的信息（数据块和目标DataNode）加入到源目标DataNode中；在目标DataNode中记录复制请求；
l           从队列中将复制项移动到pendingReplications。
可见，这个方法执行后，复制项从neededReplications挪到pendingReplications中。DataNode在某次心跳的应答中，可以拿到相应的信息，执行复制操作。computeInvalidateWork当然是用于删除无效的数据块。它的主要工作在invalidateWorkForOneNode中完成。和上面computeDatanodeWork类似，不过它的处理更简单，将recentInvalidateSets的数据通过DatanodeDescriptor.addBlocksToBeInvalidated挪到DataNode中。dnthread执行的是DecommissionedMonitor，它的run方法周期调用decommissionedDatanodeCheck，再到checkDecommissionStateInternal，定期将完成Decommission任务的DataNode状态从DECOMMISSION_INPROGRESS改为DECOMMISSIONED。PendingReplicationMonitor中的线程用于对处在等待复制状态的数据块进行检查。如果发现长时间该数据块没被复制，那么会将它挪到timedOutItems中。请参考PendingReplicationBlocks的讨论。infoServer的相关线程我们就不分析了，它们都用于处理HTTP请求。上面已经总结了NameNode上的一些为特殊任务启动的线程，除了这些线程，NameNode上还运行着RPC服务器的相关线程，具体可以看前面章节。在我们开始分析Secondary NameNode前，我们给出了以NameNode上一些状态转移图，大家可以通过这个图，更好理解NameNode。NameNode：


DataNode：


文件：


Block，比较复杂：


上面的图不是很严格，只是用于帮助大家理解NameNode对Block复杂的处理过程。
稍微说明一下，“Block in inited DataNode”表明这个数据块在一个刚初始化的DataNode上。“Block in INodeFile”是数据块属于某个文件，“Block in INodeFileUnderConstruction” 表明这数据块属于一个正在构建的文件，当然，处于这个状态的Block可能因为租约恢复而转移到“Block in Recover”。右上方描述了需要复制的数据块的状态，UnderReplicatedBlocks和PendingReplicationBlocks的区别在于Block是否被插入到某一个DatanodeDescriptor中。Corrupt和Invalidate的就好理解啦。

下一篇

Hadoop二次开发必备，Hadoop源码分析（十四）

上一篇

Hadoop二次开发必备，Hadoop源码分析（十二）