Hadoop源码分析笔记(十一)：数据节点--数据节点整体运行

数据节点整体运行

数据节点通过数据节点存储和文件系统数据集，管理着保存在Linux文件系统上的数据块，通过流式接口提供数据块的读、写、替换、复制和校验信息等功能。建立在上述基础上的数据节点，还需要维护和名字节点的关系，周期性地检查数据块，并作为一个整体保证文件系统的正常工作。

数据节点与名字节点的交互

数据节点提供了对HDFS文件数据的读写支持，但是，客户端访问HDFS文件，在读数据时，必须了解文件对应的数据块的存储位置;在写数据时，需要名字节点提供目标数据节点列表，这些操作的实现，都离不开数据节点和名字节点的配合。

数据节点和名字节点间通过IPC接口DatanodeProtocol进行通信，数据节点是客户端，名字节点是服务器。这个接口一共有11个方法，在对写数据块的流程进行分析时，已经涉及了这个接口中的远程方法blockReceived()、reportBadBlocks()、commitBlockSynchronization()和nextGenerationStamp()。其中，前两个方法用于告知名字节点写数据的处理结果，后两个方法直接应用与数据块恢复流程中。

1、握手、注册、数据块上报和心跳

无论数据节点是以“-regular”参数启动或者“-rollback”启动，DataNode.startNode()方法都会通过handshake()间接调用DatanodeProtocol.versionRequest()，获取名字节点的NamespaceInfo。NamespaceInfo包含了一些存储管理相关的信息，数据节点的后续处理，如版本检查、注册，都需要使用NamespaceInfo中的信息。例如，在handshake()中，数据节点会保证它的构建信息和存储系统结构版本号和名字节点一致，代码如下：

private NamespaceInfo handshake() throws IOException {
NamespaceInfo nsInfo = new NamespaceInfo();
while (shouldRun) {
try {
nsInfo = namenode.versionRequest();
break;
} catch(SocketTimeoutException e) {  // namenode is busy
LOG.info("Problem connecting to server: " + getNameNodeAddr());
try {
Thread.sleep(1000);
} catch (InterruptedException ie) {}
}
}
String errorMsg = null;
// verify build version
if( ! nsInfo.getBuildVersion().equals( Storage.getBuildVersion() )) {
errorMsg = "Incompatible build versions: namenode BV = "
+ nsInfo.getBuildVersion() + "; datanode BV = "
+ Storage.getBuildVersion();
LOG.fatal( errorMsg );
notifyNamenode(DatanodeProtocol.NOTIFY, errorMsg);
throw new IOException( errorMsg );
}
assert FSConstants.LAYOUT_VERSION == nsInfo.getLayoutVersion() :
"Data-node and name-node layout versions must be the same."
+ "Expected: "+ FSConstants.LAYOUT_VERSION + " actual "+ nsInfo.getLayoutVersion();
return nsInfo;
}

正常启动数据节点，在handshake()调用后，数据节点必须通过远程方法register()方法后名字节点注册，注册的输入和输出都是DatanodeRegistration对象，即数据节点的成员变量dnRegistration。DataNode.startNode()方法会根据数据节点的执行环境，构造对象。代码如下：

  /**
* This method starts the data node with the specified conf.
*
* @param conf - the configuration
*  if conf's CONFIG_PROPERTY_SIMULATED property is set
*  then a simulated storage based data node is created.
*
* @param dataDirs - only for a non-simulated storage data node
* @throws IOException
* @throws MalformedObjectNameException
* @throws MBeanRegistrationException
* @throws InstanceAlreadyExistsException
*/
void startDataNode(Configuration conf,
AbstractList<File> dataDirs, SecureResources resources
) throws IOException {
.....
storage = new DataStorage();
// construct registration
this.dnRegistration = new DatanodeRegistration(machineName + ":" + tmpPort);
.....
if (simulatedFSDataset) {
setNewStorageID(dnRegistration);
dnRegistration.storageInfo.layoutVersion = FSConstants.LAYOUT_VERSION;
dnRegistration.storageInfo.namespaceID = nsInfo.namespaceID;
// it would have been better to pass storage as a parameter to
// constructor below - need to augment ReflectionUtils used below.
conf.set("StorageId", dnRegistration.getStorageID());
try {
//Equivalent of following (can't do because Simulated is in test dir)
//  this.data = new SimulatedFSDataset(conf);
this.data = (FSDatasetInterface) ReflectionUtils.newInstance(
Class.forName("org.apache.hadoop.hdfs.server.datanode.SimulatedFSDataset"), conf);
} catch (ClassNotFoundException e) {
throw new IOException(StringUtils.stringifyException(e));
}
} else { // real storage
// read storage info, lock data dirs and transition fs state if necessary
storage.recoverTransitionRead(nsInfo, dataDirs, startOpt);
// adjust
this.dnRegistration.setStorageInfo(storage);
// initialize data node internal structure
this.data = new FSDataset(storage, conf);
}
this.dnRegistration.setName(machineName + ":" + tmpPort);
// adjust info port
this.dnRegistration.setInfoPort(this.infoServer.getPort());
myMetrics = DataNodeInstrumentation.create(conf,
dnRegistration.getStorageID());
dnRegistration.setIpcPort(ipcServer.getListenerAddress().getPort());
......
}

DataNode的同名方法register()调用名字节点提供的注册远程方法，并处理应答。

数据节点顺利注册后，在register()方法中，还需要根据目前数据节点的配置情况执行一些后续处理，主要包括：可能的节点存储DataStorage初始化，数据节点注册完成后，名字节点会返回系统统一存储标识等创建“VERSION”文件的必须信息，帮助数据节点完成节点存储的初始化工作。至此，我们清楚数据节点“VERSION”文件中各个属性的来源，它们大部分来自数据节点，是整个HDFS集群的统一属性，如namaspaceID和layoutVersion等；有一些是数据节点自己产生的，包括storageID和storageType。

register方法如下：

private void register() throws IOException {
if (dnRegistration.getStorageID().equals("")) {
setNewStorageID(dnRegistration);
}
while(shouldRun) {
try {
// reset name to machineName. Mainly for web interface.
dnRegistration.name = machineName + ":" + dnRegistration.getPort();
dnRegistration = namenode.register(dnRegistration);
break;
} catch(SocketTimeoutException e) {  // namenode is busy
LOG.info("Problem connecting to server: " + getNameNodeAddr());
try {
Thread.sleep(1000);
} catch (InterruptedException ie) {}
}
}
assert ("".equals(storage.getStorageID())
&& !"".equals(dnRegistration.getStorageID()))
|| storage.getStorageID().equals(dnRegistration.getStorageID()) :
"New storageID can be assigned only if data-node is not formatted";
if (storage.getStorageID().equals("")) {
storage.setStorageID(dnRegistration.getStorageID());
storage.writeAll();
LOG.info("New storage id " + dnRegistration.getStorageID()
+ " is assigned to data-node " + dnRegistration.getName());
}
if(! storage.getStorageID().equals(dnRegistration.getStorageID())) {
throw new IOException("Inconsistent storage IDs. Name-node returned "
+ dnRegistration.getStorageID()
+ ". Expecting " + storage.getStorageID());
}
if (!isBlockTokenInitialized) {
/* first time registering with NN */
ExportedBlockKeys keys = dnRegistration.exportedKeys;
this.isBlockTokenEnabled = keys.isBlockTokenEnabled();
if (isBlockTokenEnabled) {
long blockKeyUpdateInterval = keys.getKeyUpdateInterval();
long blockTokenLifetime = keys.getTokenLifetime();
LOG.info("Block token params received from NN: keyUpdateInterval="
+ blockKeyUpdateInterval / (60 * 1000) + " min(s), tokenLifetime="
+ blockTokenLifetime / (60 * 1000) + " min(s)");
blockTokenSecretManager.setTokenLifetime(blockTokenLifetime);
}
isBlockTokenInitialized = true;
}
if (isBlockTokenEnabled) {
blockTokenSecretManager.setKeys(dnRegistration.exportedKeys);
dnRegistration.exportedKeys = ExportedBlockKeys.DUMMY_KEYS;
}
if (supportAppends) {
Block[] bbwReport = data.getBlocksBeingWrittenReport();
long[] blocksBeingWritten = BlockListAsLongs
.convertToArrayLongs(bbwReport);
namenode.blocksBeingWrittenReport(dnRegistration, blocksBeingWritten);
}
// random short delay - helps scatter the BR from all DNs
scheduleBlockReport(initialBlockReportDelay);
}

名字节点保存并持久化整个文件系统的文件目录树以及文件的数据快索引，但名字节点不持久化数据块的保存位置。HDFS启动时，数据节点需要报告它上面保存的数据块信息，帮组名字节点建立数据块和保存数据块的数据节点的对应关系。这个操作会定时执行。通过FSDataset的getBlockReport()方法，DataNode.offerService()获得当前数据节点上所有数据块的列表，然后将这些数据块信息序列化成一个长整形数组，发送数组到名字节点。远程方法调用结束后，名字节点返回一个名字节点指令，数据节点随后执行该指令。需要强调的是：数据块上报blockReport()定期执行，为数据节点和名字节点之间数据的一致性提供了重要的保证。代码如下：

 if (startTime - lastHeartbeat > heartBeatInterval) {
//
// All heartbeat messages include following info:
// -- Datanode name
// -- data transfer port
// -- Total capacity
// -- Bytes remaining
//
lastHeartbeat = startTime;
DatanodeCommand[] cmds = namenode.sendHeartbeat(dnRegistration,
data.getCapacity(),
data.getDfsUsed(),
data.getRemaining(),
xmitsInProgress.get(),
getXceiverCount());
myMetrics.addHeartBeat(now() - startTime);
//LOG.info("Just sent heartbeat, with name " + localName);
if (!processCommand(cmds))
continue;
}
......
// Send latest blockinfo report if timer has expired.
if (startTime - lastBlockReport > blockReportInterval) {
// Create block report
long brCreateStartTime = now();
Block[] bReport = data.getBlockReport();
// Send block report
long brSendStartTime = now();
DatanodeCommand cmd = namenode.blockReport(dnRegistration,
BlockListAsLongs.convertToArrayLongs(bReport));
// Log the block report processing stats from Datanode perspective
long brSendCost = now() - brSendStartTime;
long brCreateCost = brSendStartTime - brCreateStartTime;
myMetrics.addBlockReport(brSendCost);
LOG.info("BlockReport of " + bReport.length
+ " blocks took " + brCreateCost + " msec to generate and "
+ brSendCost + " msecs for RPC and NN processing");
//
// If we have sent the first block report, then wait a random
// time before we start the periodic block reports.
//
if (resetBlockReportTime) {
lastBlockReport = startTime - R.nextInt((int)(blockReportInterval));
resetBlockReportTime = false;
} else {
/* say the last block report was at 8:20:14. The current report
* should have started around 9:20:14 (default 1 hour interval).
* If current time is :
*   1) normal like 9:20:18, next report should be at 10:20:14
*   2) unexpected like 11:35:43, next report should be at 12:20:14
*/
lastBlockReport += (now() - lastBlockReport) /
blockReportInterval * blockReportInterval;
}
processCommand(cmd);
}

DataNode.offerService()循环中另一个和名字节点的重要交互式心跳，名字节点根据数据节点的定期心跳，判断数据节点是否正常工作。心跳上报过程中，数据节点会发送能够描述当前节点负载的一些信息，如数据节点存储容器、目前已使用容量等，名字节点根据这些信息估计数据节点的工作状态，均衡各个节点的负载。远程方法sendHeartbeat()执行结束，会携带名字节点到数据节点的指令，数据节点执行这些指令，保证HDFS系统的健康、稳定运行。

名字节点指令的执行

名字节点通过IPC(主要是心跳)调用返回值，通知数据节点执行名字节点指令，这些指令最后都由DataNode.processCommand()方法处理。方法的主体是一个case语句，根据命令编号执行不同的方法。代码如下：

private boolean processCommand(DatanodeCommand cmd) throws IOException {
if (cmd == null)
return true;
final BlockCommand bcmd = cmd instanceof BlockCommand? (BlockCommand)cmd: null;
switch(cmd.getAction()) {
case DatanodeProtocol.DNA_TRANSFER:
// Send a copy of a block to another datanode
transferBlocks(bcmd.getBlocks(), bcmd.getTargets());
myMetrics.incrBlocksReplicated(bcmd.getBlocks().length);
break;
case DatanodeProtocol.DNA_INVALIDATE:
//
// Some local block(s) are obsolete and can be
// safely garbage-collected.
//
Block toDelete[] = bcmd.getBlocks();
try {
if (blockScanner != null) {
blockScanner.deleteBlocks(toDelete);
}
data.invalidate(toDelete);
} catch(IOException e) {
checkDiskError();
throw e;
}
myMetrics.incrBlocksRemoved(toDelete.length);
break;
case DatanodeProtocol.DNA_SHUTDOWN:
// shut down the data node
this.shutdown();
return false;
case DatanodeProtocol.DNA_REGISTER:
// namenode requested a registration - at start or if NN lost contact
LOG.info("DatanodeCommand action: DNA_REGISTER");
if (shouldRun) {
register();
}
break;
case DatanodeProtocol.DNA_FINALIZE:
storage.finalizeUpgrade();
break;
case UpgradeCommand.UC_ACTION_START_UPGRADE:
// start distributed upgrade here
processDistributedUpgradeCommand((UpgradeCommand)cmd);
break;
case DatanodeProtocol.DNA_RECOVERBLOCK:
recoverBlocks(bcmd.getBlocks(), bcmd.getTargets());
break;
case DatanodeProtocol.DNA_ACCESSKEYUPDATE:
LOG.info("DatanodeCommand action: DNA_ACCESSKEYUPDATE");
if (isBlockTokenEnabled) {
blockTokenSecretManager.setKeys(((KeyUpdateCommand) cmd).getExportedKeys());
}
break;
case DatanodeProtocol.DNA_BALANCERBANDWIDTHUPDATE:
LOG.info("DatanodeCommand action: DNA_BALANCERBANDWIDTHUPDATE");
int vsn = ((BalancerBandwidthCommand) cmd).getBalancerBandwidthVersion();
if (vsn >= 1) {
long bandwidth =
((BalancerBandwidthCommand) cmd).getBalancerBandwidthValue();
if (bandwidth > 0) {
DataXceiverServer dxcs =
(DataXceiverServer) this.dataXceiverServer.getRunnable();
dxcs.balanceThrottler.setBandwidth(bandwidth);
}
}
break;
default:
LOG.warn("Unknown DatanodeCommand action: " + cmd.getAction());
}
return true;
}

命令DNA_TRANSFER、DNA_INVALIDATE和DNA_RECOVERBLOCK与数据块相关，其中，DNA_INVALIDATE指令的实现较为简单，顺序在数据块扫描器和文件系统数据集对象中删除数据块，FSDataset.invalidate()通过异步磁盘操作服务FSDatasetAsyncDiskService删除Linux文件系统上的数据块文件和校验信息文件，降低了processCommand()执行时间。

名字节点指令DNA_RECOVERBLOCK用于恢复数据块，这是由名字节点发起的数据块恢复，作为整个系统故障恢复的重要措施。指令DNA_RECOVERBLOCK可以恢复客户端永久崩溃形成的，还处于写状态的数据块。由名字节点触发的数据块恢复，恢复策略永久是截断，即将数据块会参与到过程的各数据节点上的数据块副本长度的最小值；同时，恢复过程结束后，主数据节点通知名字节点，关闭它上面还处于打开状态的文件。

当名字节点发现某数据块当前副本数小于目标值时，名字节点利用DNA_TRANSFER命令，通知某一个拥有该数据块的数据节点将数据块复制到其他数据节点。

数据块扫描器

每个数据节点都会执行一个数据块扫描器DataBlockScanner，它周期性地验证节点所存储的数据块，通过DataBlockScanner数据节点可以尽早发现有问题的数据块，并汇报给数据节点。

数据块扫描器是数据节点中比较独立的一个模块，包括扫描器的主要实现DataBlockScanner和辅助类BlockScanInfo、LogEntry和LogFileHandler等。

BlockScanInfo类保存的数据块扫描相关信息包括：

block：数据块

lastScanTime：最后一次扫描的时间

lastLogTime：最后写日志的时间

lastScanType：扫描的类型，定义在枚举DataBlockScanner.ScanType中

lastScanOk：扫描结果

其中，扫描的类型包括：

NONE：还没有执行过扫描

REMOTE_READ：最后一次扫描结果是由客户端读产生

VERIFICATION_SCAN：扫描结果由数据块的扫描器产生

前面介绍客户端读数据的过程时分析过，客户端成功读取了一个完整数据块(包括校验)后，会发送一个附加的响应码，通知数据节点的校验成功，这个信息会更新在DataBlockScanner的记录中，对应的类型就是REMOTE_READ

BlockScanInfo对象是可比较的，比较方法compareTo()方法lastScanTime的大小，按时间先后排序，这样保存着所有待扫描数据块信息的成员变量DataBlockScanner.blockInfoSet中的元素，也就根据扫描时间排序。

辅助类LogEntry和LogFileHandler和扫描器日志文件相关。由于数据块每三周才被扫描一次，将扫描信息保存在文件里，以防止数据节点重启后丢失，就很有必要了。

数据节点的启停

DataNode.main()是数据节点的入口方法，但他只完成一件事件，即调用secureMain()方法。secureMain()方法通过createDataNode()创建并启动数据节点，然后通过调用DataNode对象的join()，等待数据节点停止运行。

DataNode.createDatNode()方法又是通过另外一个方法创建数据节点的实例，然后，方法调用runDatanodeDaemon()执行数据节点的主线程。

数据节点的构造函数很简单，主要的初始化工作由startDataNode()方法。这个方法依次完成如下工作：

1)获取节点的名字和名字节点的地址，读入一些运行时需要的配置项

2)构造注册需要的DatanodeRegistration对象

3)建立到数据节点的IPC连接并握手，即调用名字节点上的远程方法handshake()

4)执行可能的存储空间状态恢复，构造数据节点使用的FSDataset对象

5)创建流接口服务器DataXceiverServer

6)创建数据块扫描器DataBlockScanner

7)创建数据节点上的HTTP服务器，该服务器提供了数据节点的运行状态信息

8)创建数据节点IPC服务器，对外提供ClientDatanodeProtocol和InterDatanodeProtocol远程服务

可见，在startNode()方法中，几经对数据节点上的主要模块进行了初始化，但模块提供的服务都还没有投入正式使用。接下来，DataNode.createNode()会调用runDatanodeDaemon()方法，首先通过前面介绍的DataNode.register()向名字节点注册当前数据节点，成功注册后，执行数据节点的服务主线程。

DataNode类实现了Runable接口，runDatanodeDaemon()在数据节点服务主线程对象上调用start()方法，线程执行DataNode.run()方法，run()方法入口处就启动了流式接口和IPC接口的服务，这个时候，数据节点处于正常服务状态，可以接受外部请求。

DataNode.run()循环调用offerService()方法，而offerService()其实也就是一个循环，在这个循环里会执行如下操作：

1)发送到名字节点的心跳，并执行名字节点指令

2)通过blockReceiverd()上报数据节点上接收到的数据块

3)根据需求调用远程接口DatanodeProtocol.blockReport()，报告数据节点目前保存的数据块信息

4)根据需要启动数据块扫描器DataBlockScanner

数据节点的停止

相对于启动，数据节点停止不需要做太多工作。DataNode.shutdown()用于停止数据节点，它的调用时机如下：

1)DataNode构造失败，startDataNode()方法抛出IOException异常

2)数据节点服务主线程捕获如下异常：UnregisteredDatanodeException(节点未注册)、DisallowedDatanodeException(节点被撤销)和IncorrectVersionException(节点版本不正确)

3)收到名字节点指令DatanodeProtocol.DNA_SHUTDOWN

4)数据节点服务主线程退出前

shutdown()代码如下：

 public void shutdown() {
this.unRegisterMXBean();
if (infoServer != null) {
try {
infoServer.stop();
} catch (Exception e) {
LOG.warn("Exception shutting down DataNode", e);
}
}
if (ipcServer != null) {
ipcServer.stop();
}
this.shouldRun = false;
if (dataXceiverServer != null) {
((DataXceiverServer) this.dataXceiverServer.getRunnable()).kill();
this.dataXceiverServer.interrupt();
// wait for all data receiver threads to exit
if (this.threadGroup != null) {
while (true) {
this.threadGroup.interrupt();
LOG.info("Waiting for threadgroup to exit, active threads is " +
this.threadGroup.activeCount());
if (this.threadGroup.activeCount() == 0) {
break;
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {}
}
}
// wait for dataXceiveServer to terminate
try {
this.dataXceiverServer.join();
} catch (InterruptedException ie) {
}
}
RPC.stopProxy(namenode); // stop the RPC threads
if(upgradeManager != null)
upgradeManager.shutdownUpgrade();
if (blockScannerThread != null) {
blockScannerThread.interrupt();
try {
blockScannerThread.join(3600000L); // wait for at most 1 hour
} catch (InterruptedException ie) {
}
}
if (storage != null) {
try {
this.storage.unlockAll();
} catch (IOException ie) {
}
}
if (dataNodeThread != null) {
dataNodeThread.interrupt();
try {
dataNodeThread.join();
} catch (InterruptedException ie) {
}
}
if (data != null) {
data.shutdown();
}
if (myMetrics != null) {
myMetrics.shutdown();
}
}

通过这个方法回顾数据节点中的关键组件。首先停止的是数据节点上的HTTP服务器和提供ClientDatanodeProtocol和InterDatanodeProtocol远程接口的IPC服务器。由于DataNode构造过程失败也会调用shutdown()方法，所以所有的资源释放都需要判断相应的对象是否存储。上述两个服务器停止后，数据节点的成员变量shouldRun设置为false。

这里的shouldRun是一种典型的volatitle变量用法：检查某个状态标记以判断是否退出循环。在DataNode.offerService()、DataNode.run()、DataXceiverServer.run()、PacketResponder.run()、DataBlockScanner.run()等大量方法中，实现逻辑里的循环都会判断这个标记，以尽快在数据节点停止时，退出循环。

接下来要停止的是流式接口的相关服务。由于流式接口的监听线程和服务线程都可能涉及Socket上阻塞操作中，根据shouldRun标记结束循环并退出线程。

其他的清理工作还有：

1)关闭数据节点到名字节点的IPC客户端，关闭数据块扫描器

2)对文件"in_use.lock"解锁并删除文件

3)中断数据节点服务线程，关闭数据节点拥有的FSDataset对象

shutdown()方法执行结束后，数据节点的服务线程退出，secureMain()方法的join()调用返回，接着，DataNode.secureMain()执行它的finally语句，在日志系统里打印“Exiting Datanode”信息后，数据节点结束运行并退出。

版权申明：本文部分摘自【蔡斌、陈湘萍】所著【Hadoop技术内幕深入解析Hadoop Common和HDFS架构设计与实现原理】一书，仅作为学习笔记，用于技术交流，其商业版权由原作者保留，推荐大家购买图书研究，转载请保留原作者，谢谢！