apache#3. datanode的启动流程

Author: Yifan122

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeDescriptor.java

@@ -453,9 +453,12 @@ public void updateHeartbeatState(StorageReport[] reports, long cacheCapacity,
           storageMap.values());
     }
 
+    // TODO 更改存储信息
     setCacheCapacity(cacheCapacity);
     setCacheUsed(cacheUsed);
     setXceiverCount(xceiverCount);
+
+    // TODO 修改上一次心跳时间
     setLastUpdate(Time.now());
     setLastUpdateMonotonic(Time.monotonicNow());
     this.volumeFailures = volFailures;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/DatanodeManager.java

@@ -582,6 +582,8 @@ void removeDeadDatanode(final DatanodeID nodeID) {
 
   /** Is the datanode dead? */
   boolean isDatanodeDead(DatanodeDescriptor node) {
+    // 如果超过10分钟30秒没有心跳，就认为是dead
+    // 超时后会复制副本到新的机器
     return (node.getLastUpdateMonotonic() <
             (monotonicNow() - heartbeatExpireInterval));
   }
@@ -1349,6 +1351,7 @@ public DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
       synchronized (datanodeMap) {
         DatanodeDescriptor nodeinfo = null;
         try {
+          // 获取node的信息
           nodeinfo = getDatanode(nodeReg);
         } catch(UnregisteredNodeException e) {
           return new DatanodeCommand[]{RegisterCommand.REGISTER};
@@ -1360,10 +1363,12 @@ public DatanodeCommand[] handleHeartbeat(DatanodeRegistration nodeReg,
           throw new DisallowedDatanodeException(nodeinfo);
         }
 
+        // 如果没有则注册node的信息
         if (nodeinfo == null || !nodeinfo.isAlive) {
           return new DatanodeCommand[]{RegisterCommand.REGISTER};
         }
 
+        // TODO 更新心跳的重要的信息
         heartbeatManager.updateHeartbeat(nodeinfo, reports,
                                          cacheCapacity, cacheUsed,
                                          xceiverCount, failedVolumes,

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/blockmanagement/HeartbeatManager.java

@@ -221,6 +221,7 @@ synchronized void updateHeartbeat(final DatanodeDescriptor node,
       int xceiverCount, int failedVolumes,
       VolumeFailureSummary volumeFailureSummary) {
     stats.subtract(node);
+    // TODO 重要代码
     node.updateHeartbeat(reports, cacheCapacity, cacheUsed,
       xceiverCount, failedVolumes, volumeFailureSummary);
     stats.add(node);
@@ -283,6 +284,7 @@ void heartbeatCheck() {
       int numOfStaleNodes = 0;
       int numOfStaleStorages = 0;
       synchronized(this) {
+        // TODO 遍历每个datanode
         for (DatanodeDescriptor d : datanodes) {
           if (dead == null && dm.isDatanodeDead(d)) {
             stats.incrExpiredHeartbeats();
@@ -355,6 +357,7 @@ public void run() {
         try {
           final long now = Time.monotonicNow();
           if (lastHeartbeatCheck + heartbeatRecheckInterval < now) {
+            // TODO 心跳检查
             heartbeatCheck();
             lastHeartbeatCheck = now;
           }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/BPServiceActor.java

@@ -692,6 +692,7 @@ private void offerService() throws Exception {
         //
         // Every so often, send heartbeat or block-report
         //
+        // TODO 心跳是每三秒发送一次
         if (startTime - lastHeartbeat >= dnConf.heartBeatInterval) {
           //
           // All heartbeat messages include following info:
@@ -702,6 +703,11 @@ private void offerService() throws Exception {
           //
           lastHeartbeat = startTime;
           if (!dn.areHeartbeatsDisabledForTests()) {
+            // Namenode 是不直接跟Datanode进行链接的
+            // DataNode 发送心跳给Namenode
+            // NameNode接收到心跳后，会返回一些命令
+            // Datanode接收到这些命令后，根据这些命令做对应的操作
+            // TODO 发送心跳，返回的是NameNode的响应命令
             HeartbeatResponse resp = sendHeartBeat();
             assert resp != null;
             dn.getMetrics().addHeartbeat(monotonicNow() - startTime);
@@ -721,6 +727,8 @@ private void offerService() throws Exception {
             }
 
             long startProcessCommands = monotonicNow();
+            // TODO 处理namenode返回的指令
+            // 比如删除block，复制block
             if (!processCommand(resp.getCommands()))
               continue;
             long endProcessCommands = monotonicNow();
@@ -877,6 +885,7 @@ public void run() {
 
       while (shouldRun()) {
         try {
+          // TODO 发送心跳
           offerService();
         } catch (Exception ex) {
           LOG.error("Exception in BPOfferService for " + this, ex);

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -4529,6 +4529,7 @@ HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
       //get datanode commands
       final int maxTransfer = blockManager.getMaxReplicationStreams()
           - xmitsInProgress;
+      // TODO NameNode处理Datanode发送过来的心跳
       DatanodeCommand[] cmds = blockManager.getDatanodeManager().handleHeartbeat(
           nodeReg, reports, blockPoolId, cacheCapacity, cacheUsed,
           xceiverCount, maxTransfer, failedVolumes, volumeFailureSummary);
@@ -4538,6 +4539,7 @@ HeartbeatResponse handleHeartbeat(DatanodeRegistration nodeReg,
           haContext.getState().getServiceState(),
           getFSImage().getLastAppliedOrWrittenTxId());
 
+      // TODO 给DataNode返回响应
       return new HeartbeatResponse(cmds, haState, rollingUpgradeInfo);
     } finally {
       readUnlock();