Hadoop学习十三：Hadoop-Hdfs DataStorage源码

一.DataStorage

• DataStorage继承Storage。
• DataStorage对应Hdfs的dfs.data.dir目录。
• DataNode启动时，检查恢复状态转换dfs.data.dir下所有的StorageDirectory。

二.DataStorage类图

三.详细描述

1. org.apache.hadoop.hdfs.server.protocol.NamespaceInfo：NamespaceInfo is returned by the name-node in reply  to a data-node handshake.当我们配置完整个Hadoop系统时，做的第一件事就是

   $bin/hadoop namenode -format

    format后的NameNode产生一个namespaceId。当DataNode启动时，会向NameNode handshake，NameNode返回这个namespaceId。

   NamespaceInfo nsInfo = handshake();

2.  DataStorage：
   1. 参考代码中的步骤阅读。状态分析-----恢复操作-----状态转换
   2. 关于“一致性”的说明：参考http://zy19982004.iteye.com/blog/1876706第三张图，/update /rollback /finalize指令后，StorageDirectory下会存在不同的文件夹，当整个文件夹是某个中间状态时，就是“不一致”。举例：/update开始后，“一致”状态指的是current和previous都存在，也就是/update的123步都完成了；如果123中的中间某一步发生了问题或者还没有进行，造成了previous.tmp存在current不存在的情况，就是“不一致”，造成了previous.tmp存在current存在也是不一致，都需要doRecover。

四.DataStorage源码时序图

五.DataStorage源码

package org.apache.hadoop.hdfs.server.datanode;

public class DataStorage extends Storage {

	//DataStorag独特属性
	private String storageID;

	DataStorage() {
		super(NodeType.DATA_NODE);
		storageID = "";
	}

	// DataNode.start时触发此方法
	void recoverTransitionRead(NamespaceInfo nsInfo, Collection<File> dataDirs,
			StartupOption startOpt) throws IOException {
		// 1.在doTransition之前首先对每个SorageDirectory进行分析，检查其是否处于“一致状态”
		// 如果处于“不一致状态“，format or doRecover
		this.storageID = "";
		this.storageDirs = new ArrayList<StorageDirectory>(dataDirs.size());
		ArrayList<StorageState> dataDirStates = new ArrayList<StorageState>(
				dataDirs.size());
		for (Iterator<File> it = dataDirs.iterator(); it.hasNext();) {
			File dataDir = it.next();
			StorageDirectory sd = new StorageDirectory(dataDir);
			StorageState curState;
			try {
				// 1.1 调用Storage.analyzeStorage进行状态分析
				curState = sd.analyzeStorage(startOpt);
				// 1.2 调用format或者Storage.doRecover进行恢复操作
				switch (curState) {
				case NORMAL:
					break;
				case NON_EXISTENT:
					// ignore this storage
					it.remove();
					continue;
				case NOT_FORMATTED: // format
					format(sd, nsInfo);
					break;
				default: // recovery part is common
					sd.doRecover(curState);
				}
			} catch (IOException ioe) {
				sd.unlock();
				throw ioe;
			}
			// 1.3 "一致状态"的StorageDirectory才会加入到Storage
			addStorageDir(sd);
			dataDirStates.add(curState);
		}

		// 2. doTransition 状态转换
		// 根据DataNode启动参数做upgrade or rollback，其它正常启动
		for (int idx = 0; idx < getNumStorageDirs(); idx++) {
			doTransition(getStorageDir(idx), nsInfo, startOpt);
		}

		// 3. Update 每个 StorageDirectory的VERSION
		this.writeAll();
	}

	void format(StorageDirectory sd, NamespaceInfo nsInfo) throws IOException {
		sd.clearDirectory(); // create directory
		this.layoutVersion = FSConstants.LAYOUT_VERSION;
		this.namespaceID = nsInfo.getNamespaceID();
		this.cTime = 0;
		// store storageID as it currently is
		sd.write();
	}

	//重写,DataNode增加storageID到VERSION
	protected void setFields(Properties props, StorageDirectory sd)
			throws IOException {
		super.setFields(props, sd);
		props.setProperty("storageID", storageID);
	}

	//重写,DataNode增加storageID到VERSION
	protected void getFields(Properties props, StorageDirectory sd)
			throws IOException {
		super.getFields(props, sd);
		String ssid = props.getProperty("storageID");
		storageID = ssid;
	}

	/**
	 * Analize which and whether a transition of the fs state is required and
	 * perform it if necessary. Rollback if previousLV >= LAYOUT_VERSION &&
	 * prevCTime <= namenode.cTime Upgrade if this.LV > LAYOUT_VERSION ||
	 * this.cTime < namenode.cTime Regular startup if this.LV = LAYOUT_VERSION
	 * && this.cTime = namenode.cTime
	 */
	// 2 doTransition 状态转换
	// 根据DataNode启动参数做upgrade or rollback，其它正常启动
	private void doTransition( StorageDirectory sd,  NamespaceInfo nsInfo,  StartupOption startOpt ) throws IOException {
    if (startOpt == StartupOption.ROLLBACK)
      //2.1 doRollback
      doRollback(sd, nsInfo); // rollback if applicable
    sd.read();
    ...
    if (this.layoutVersion > FSConstants.LAYOUT_VERSION
        || this.cTime < nsInfo.getCTime()) {
      //2.2 doUpgrade
      doUpgrade(sd, nsInfo);  // upgrade
      return;
    }
  }

	// 2.1 doRollback
	// 1.current-----> removed.tmp
	// 2.previous---->current
	// 3.删除removed.tmp
	void doRollback(StorageDirectory sd, NamespaceInfo nsInfo)
			throws IOException {
		rename(curDir, tmpDir);
		rename(prevDir, curDir);
		deleteDir(tmpDir);
	}

	// 2.2 doUpgrade
	// 1.current-----> previous.tmp
	// 2.重建current：创建previous.tmp到current的硬链接，写入VERSION
	// 3.previous.tmp----->previous
	void doUpgrade(StorageDirectory sd, NamespaceInfo nsInfo)
			throws IOException {
		HardLink hardLink = new HardLink();
		rename(curDir, tmpDir);
		linkBlocks(tmpDir, curDir, this.getLayoutVersion(), hardLink);
		sd.write();
		rename(tmpDir, prevDir);
	}

	void finalizeUpgrade() throws IOException {
		for (Iterator<StorageDirectory> it = storageDirs.iterator(); it
				.hasNext();) {
			doFinalize(it.next());
		}
	}

	// 1.previous---->finalized.tmp
	// 2.删除finalized.tmp
	void doFinalize(StorageDirectory sd) throws IOException {
		rename(prevDir, tmpDir);
		new Daemon(new Runnable() {
			public void run() {
				deleteDir(tmpDir);
			}
		}).start();
	}

}