HDFS集群的启动(8)——NameNode任务线程之FSNamesystem$SafeModeMonitor

HDFS集群启动之后并不能马上进入工作模式为用户提供服务，而是先要进入安全模式状态，因为主节点要确定当前集群中的绝大多数数据块Block处于安全可用的状态以确保为用户提供更可靠的读服务。这个工作主要是交由主节点上的FSNamesystem来完成的。FSNamesystem在初始化的时侯会首先进入安全模式状态并在之后开启一个后台工作线程——SafeModeMonitor来监控当前集群是否可以离开安全模式状态。这里要说的是，处于安全模式的HDFS集群会毫不犹豫地拒绝客户端的读写请求。

FSNamesystem是通过一个SafeModeInfo对象来保存当前集群的状态信息的。FSNamesystem在初始化的时侯会创建它对应的一个实例，并传入当前集群数据块的总数量。

class SafeModeInfo {
/**
* 文件系统应该进入安全模式状态的阈值
*/
private double threshold;

/**
* 当前文件系统中处于安全状态的数据块数量达到设定的阈值之后仍保持安全模式状态的延续时间
*/
private int extension;

/**
* 一个数据块处于安全状态时其有效副本的最少数量
*/
private int safeReplication;

// internal fields
/** Time when threshold was reached.
*
* <br>-1 safe mode is off
* <br> 0 safe mode is on, but threshold is not reached yet
*/
private long reached = -1;

/**
* 当前文件系统中数据块的总数量
*/
int blockTotal;

/**
* 当前文件系统中处于安全状态的数据块数量
*/
private int blockSafe;
}

当给SafeModeInfo对象设置当前集群中数据块总量的时候，它会判断当前集群是否应该进入安全模式状态，如果应该则它会让当前集群立即进入安全模式。至于如何判断集群是否应该进入安全模式状态，主要是依据当前主节点收到数据节点报告的数据块Block处于安全状态的数量占集群总数据块的比重是否达到系统设置的阈值threshold来确定的。相关源代码如下：

/**
* 判断当前文件系统是否应该进入安全模式状态
* 如果当前收到的有效数据块占文件系统数据块总量的比重没有达到设定的阈值则文件系统应该进入安全模式状态
*/
boolean needEnter() {
return getSafeBlockRatio() < threshold;
}

/**
* 计算当前收到的有效数据块占文件系统数据块总量的比重
*/
private float getSafeBlockRatio() {
return (blockTotal == 0 ? 1 : (float)blockSafe/blockTotal);
}

/**
* 检查当前文件系统是否应该进入安全模式状态
*/
private void checkMode() {
if(needEnter()) {
enter();
reportStatus("STATE* Safe mode ON.", false);
return;
}

//当前文件系统是否可以离开安全模式状态
if(!isOn() || extension <= 0 || threshold <= 0) {  // don't need to wait
this.leave(true); // leave safe mode
return;
}

if(reached > 0) {  // threshold has already been reached before
reportStatus("STATE* Safe mode ON.", false);
return;
}

//开启一个监控线程来判断当前文件系统是否可以真正离开
reached = now();
smmthread = new Daemon(new SafeModeMonitor());
smmthread.start();
reportStatus("STATE* Safe mode extension entered.", true);
}

/**
* 让当前文件系统进入安全模式状态.
*/
void enter() {
this.reached = 0;
}

说一个数据块Block处于安全状态指的是该数据块的副本数量(根据数据节点的数据块报告)达到了safeReplication。这里，threshold和safeReplication的值是根据集群的配置文件来确定的，它们的默认值分别为0.95和1，对应的配置项为:

<property>
<name>dfs.safemode.threshold.pct</name>
<value></value>
</property>

<property>
<name>dfs.replication.min</name>
<value></value>
</property>

集群在进入安全模式之后，主节点在处理数据节点关于数据块Block报告的过程中会不断的调用SafeModeInfo的checkMode()来判断当前集群中处于安全状态的数据块占集群总数据块的比重是否达到系统设置的阈值threshold，即使达到了这个阈值，集群也需要一个过渡期extension来确保当前安全数据块的数量是否已经稳定了。所以它会立即开启一个SafeModeMonitor后台线程来通过轮询的方式判断当前集群是否度过了这个稳定期，如果在稳定期之后，当前就能的安全数据块数量任然超过了设定的阈值则此时集群可以真正离开安全模式了。这里extension的默认值为0，但也可以通过配置文件来设置，对应的配置项为dfs.safemode.extention。该过程如下：

/**
*判断当前集群是否可以离开安全模式
*/
synchronized boolean canLeave() {
if (reached == 0) return false;
if(now() - reached < extension) {
reportStatus("STATE* Safe mode ON.", false);
return false;
}

return !needEnter();
}

class SafeModeMonitor implements Runnable {

private static final long recheckInterval = 1000;		//检测当前文件系统是否可以离开安全模式状态的间隔时间

/**
*/
public void run() {

//轮询当前文件系统是否可以离开安全模式状态
while (fsRunning && (safeMode != null && !safeMode.canLeave())) {
try {
Thread.sleep(recheckInterval);
} catch (InterruptedException ie) {
}
}

// leave safe mode and stop the monitor
try {
leaveSafeMode(true);
} catch(SafeModeException es) { // should never happen
String msg = "SafeModeMonitor may not run during distributed upgrade.";
assert false : msg;
throw new RuntimeException(msg, es);
}
smmthread = null;
}
}

集群在离开安全模式的时候会对集群中的所有数据块Blocks进行处理，其处理过程主要是：

/**
* 处理当前文件系统的所有数据块:
* 	1).无效数据块
* 	2).数据块副本不够
* 	3).数据块副本有多于
*/
private synchronized void processMisReplicatedBlocks() {
long nrInvalid = 0, nrOverReplicated = 0, nrUnderReplicated = 0;
neededReplications.clear();

for(BlocksMap.BlockInfo block : blocksMap.getBlocks()) {
INodeFile fileINode = block.getINode();

//数据块已不属于任何文件，所以应该从所有存储该数据块副本的数据节点上清除它
if(fileINode == null) {
nrInvalid++;
addToInvalidates(block);
continue;
}

// calculate current replication
short expectedReplication = fileINode.getReplication();
NumberReplicas num = countNodes(block);
int numCurrentReplica = num.liveReplicas();
//判断当前数据块是否还需要复制副本，如果需要则将其添加到待复制副本队列中
if(neededReplications.add(block, numCurrentReplica, num.decommissionedReplicas(), expectedReplication)) {
nrUnderReplicated++;
}

//处理数据块的多于副本
if(numCurrentReplica > expectedReplication) {
nrOverReplicated++;
processOverReplicatedBlock(block, expectedReplication, null, null);
}
}

LOG.info("Total number of blocks = " + blocksMap.size());
LOG.info("Number of invalid blocks = " + nrInvalid);
LOG.info("Number of under-replicated blocks = " + nrUnderReplicated);
LOG.info("Number of  over-replicated blocks = " + nrOverReplicated);
}

值得注意的是， HDFS为用户提供了一个接口来强制当前集群进入或离开安全状态模式，如果用户成功地让当前集群进入安全模式状态，则集群自己是永远无法离开安全模式的，而要想集群离开安全模式就只能通过用户手动调用API强制集群离开安全模式。此外，从上面的代码中可以看出，集群的数据块总数量是通过int型来保存的，这就是说集群的数据块总量不能超过2^32-1。