HDFS源码分析(二)-----元数据备份机制

前言

在Hadoop中，所有的元数据的保存都是在namenode节点之中，每次重新启动整个集群，Hadoop都需要从这些持久化了的文件中恢复数据到内存中，然后通过镜像和编辑日志文件进行定期的扫描与合并，ok，这些稍微了解Hadoop的人应该都知道，这不就是SecondNameNode干的事情嘛，但是很多人只是了解此机制的表象，内部的一些实现机理估计不是每个人都又去深究过，你能想象在写入编辑日志的过程中，用到了双缓冲区来加大并发量的写吗，你能想象为了避免操作的一致性性，作者在写入的时候做过多重的验证操作，还有你有想过作者是如何做到操作中断的处理情况吗，如果你不能很好的回答上述几个问题，那么没有关系，下面来分享一下我的学习成果。

相关涉及类

建议读者在阅读本文的过程中，结合代码一起阅读， Hadoop的源码可自行下载，这样效果可能会更好。与本文有涉及的类包括下面一些，比上次的分析略多一点，个别主类的代码量已经在上千行了。

1.FSImage--命名空间镜像类，所有关于命名空间镜像的操作方法都被包括在了这个方法里面。

2.FSEditLog--编辑日志类，编辑日志类涉及到了每个操作记录的写入。

3.EditLogInputStream和EditLogOutputStream--编辑日志输入类和编辑日志输出类，编辑日志的许多文件写入读取操作都是在这2个类的基础上实现的，这2个类是普通输入流类的继承类，对外开放了几个针对于日志读写的特有方法。

4.Storage和StorageInfo--存储信息相关类。其中后者是父类，Storage继承了后者，这2个类是存储目录直接相关的类，元数据的备份相关的很多目录操作都与此相关。

5.SecondaryNameNode--本来不想把这个类拉进去来的，但是为了使整个备份机制更加完整的呈现出来，同样也是需要去了解这一部分的代码。

ok，介绍完了这些类，下面正式进入元数据备份机制的讲解，不过在此之前，必须要了解各个对象的具体操作实现，里面有很多巧妙的设计，同时也为了防止被后面的方法绕晕，方法的确很多，但我会挑出几个代表性的来讲。

命名空间镜像

命名空间镜像在这里简称为FsImage，镜像这个词我最早听的时候是在虚拟机镜像恢复的时候听过的，很强大，不过在这里的镜像好像比较小规模一些，只是用于目录树的恢复。镜像的保存路径是由配置文件中的下面这个属性所控制

[java] view plaincopyprint?

1. ${dfs.name.dir}  

当然你可以不配，是有默认值的。在命名空间镜像中，FSImage起着主导的作用，他管理着存储空间的生存期。下面是这个类的基本变量定义

[java] view plaincopyprint?

1. /** 
2.  * FSImage handles checkpointing and logging of the namespace edits. 
3.  * fsImage镜像类 
4.  */  
5. public class FSImage extends Storage {  
6.   //标准时间格式  
7.   private static final SimpleDateFormat DATE_FORM =  
8.     new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");  
9.   
10.   //  
11.   // The filenames used for storing the images  
12.   // 在命名空间镜像中可能用的几种名称  
13.   //  
14.   enum NameNodeFile {  
15.     IMAGE     ("fsimage"),  
16.     TIME      ("fstime"),  
17.     EDITS     ("edits"),  
18.     IMAGE_NEW ("fsimage.ckpt"),  
19.     EDITS_NEW ("edits.new");  
20.       
21.     private String fileName = null;  
22.     private NameNodeFile(String name) {this.fileName = name;}  
23.     String getName() {return fileName;}  
24.   }  
25.   
26.   // checkpoint states  
27.   // 检查点击几种状态  
28.   enum CheckpointStates{START, ROLLED_EDITS, UPLOAD_START, UPLOAD_DONE; }  
29.   /** 
30.    * Implementation of StorageDirType specific to namenode storage 
31.    * A Storage directory could be of type IMAGE which stores only fsimage, 
32.    * or of type EDITS which stores edits or of type IMAGE_AND_EDITS which  
33.    * stores both fsimage and edits. 
34.    * 名字节点目录存储类型 
35.    */  
36.   static enum NameNodeDirType implements StorageDirType {  
37.     //名字节点存储类型定义主要有以下4种定义     
38.     UNDEFINED,  
39.     IMAGE,  
40.     EDITS,  
41.     IMAGE_AND_EDITS;  
42.       
43.     public StorageDirType getStorageDirType() {  
44.       return this;  
45.     }  
46.       
47.     //做存储类型的验证  
48.     public boolean isOfType(StorageDirType type) {  
49.       if ((this == IMAGE_AND_EDITS) && (type == IMAGE || type == EDITS))  
50.         return true;  
51.       return this == type;  
52.     }  
53.   }  
54.     
55.   protected long checkpointTime = -1L;  
56.   //内部维护了编辑日志类，与镜像类配合操作  
57.   protected FSEditLog editLog = null;  
58.   private boolean isUpgradeFinalized = false;  

马上看到的是几个文件状态的名称，什么edit，edit.new，fsimage.ckpt，后面这些都会在元数据机制的中进行细致讲解，可以理解为就是临时存放的目录名称。而对于这些目录的遍历，查询操作，都是下面这个类实现的

[java] view plaincopyprint?

1. //目录迭代器  
2.   private class DirIterator implements Iterator<StorageDirectory> {  
3.     //目录存储类型  
4.     StorageDirType dirType;  
5.     //向前的指标,用于移除操作  
6.     int prevIndex; // for remove()  
7.     //向后指标  
8.     int nextIndex; // for next()  
9.       
10.     DirIterator(StorageDirType dirType) {  
11.       this.dirType = dirType;  
12.       this.nextIndex = 0;  
13.       this.prevIndex = 0;  
14.     }  
15.       
16.     public boolean hasNext() {  
17.       ....  
18.     }  
19.       
20.     public StorageDirectory next() {  
21.       StorageDirectory sd = getStorageDir(nextIndex);  
22.       prevIndex = nextIndex;  
23.       nextIndex++;  
24.       if (dirType != null) {  
25.         while (nextIndex < storageDirs.size()) {  
26.           if (getStorageDir(nextIndex).getStorageDirType().isOfType(dirType))  
27.             break;  
28.           nextIndex++;  
29.         }  
30.       }  
31.       return sd;  
32.     }  
33.       
34.     public void remove() {  
35.       ...  
36.     }  
37.   }  

根据传入的目录类型，获取不同的目录，这些存储目录指的就是editlog，fsimage这些目录文件，有一些共有的信息，如下

[java] view plaincopyprint?

1. /** 
2.  * Common class for storage information. 
3.  * 存储信息公告类 
4.  * TODO namespaceID should be long and computed as hash(address + port) 
5.  * 命名空间ID必须足够长，ip地址+端口号做哈希计算而得 
6.  */  
7. public class StorageInfo {  
8.   //存储信息版本号  
9.   public int   layoutVersion;  // Version read from the stored file.  
10.   //命名空间ID  
11.   public int   namespaceID;    // namespace id of the storage  
12.   //存储信息创建时间  
13.   public long  cTime;          // creation timestamp  
14.     
15.   public StorageInfo () {  
16.     //默认构造函数，全为0  
17.     this(0, 0, 0L);  
18.   }  

下面从1个保存镜像的方法作为切入口

[java] view plaincopyprint?

1. /** 
2.    * Save the contents of the FS image to the file. 
3.    * 保存镜像文件 
4.    */  
5.   void saveFSImage(File newFile) throws IOException {  
6.     FSNamesystem fsNamesys = FSNamesystem.getFSNamesystem();  
7.     FSDirectory fsDir = fsNamesys.dir;  
8.     long startTime = FSNamesystem.now();  
9.     //  
10.     // Write out data  
11.     //  
12.     DataOutputStream out = new DataOutputStream(  
13.                                                 new BufferedOutputStream(  
14.                                                                          new FileOutputStream(newFile)));  
15.     try {  
16.       //写入版本号  
17.       out.writeInt(FSConstants.LAYOUT_VERSION);  
18.       //写入命名空间ID  
19.       out.writeInt(namespaceID);  
20.       //写入目录下的孩子总数  
21.       out.writeLong(fsDir.rootDir.numItemsInTree());  
22.       //写入时间  
23.       out.writeLong(fsNamesys.getGenerationStamp());  
24.       byte[] byteStore = new byte[4*FSConstants.MAX_PATH_LENGTH];  
25.       ByteBuffer strbuf = ByteBuffer.wrap(byteStore);  
26.       // save the root  
27.       saveINode2Image(strbuf, fsDir.rootDir, out);  
28.       // save the rest of the nodes  
29.       saveImage(strbuf, 0, fsDir.rootDir, out);  
30.       fsNamesys.saveFilesUnderConstruction(out);  
31.       fsNamesys.saveSecretManagerState(out);  
32.       strbuf = null;  
33.     } finally {  
34.       out.close();  
35.     }  
36.   
37.     LOG.info("Image file of size " + newFile.length() + " saved in "   
38.         + (FSNamesystem.now() - startTime)/1000 + " seconds.");  
39.   }  

从上面的几行可以看到，一个完整的镜像文件首部应该包括版本号，命名空间iD，文件数，数据块版本块，然后后面是具体的文件信息。在这里人家还保存了构建节点文件信息以及安全信息。在保存文件目录的信息时，采用的saveINode2Image()先保留目录信息，然后再调用saveImage()保留孩子文件信息，因为在saveImage()中会调用 saveINode2Image()方法。

[java] view plaincopyprint?

1. /* 
2.    * Save one inode's attributes to the image. 
3.    * 保留一个节点的属性到镜像中 
4.    */  
5.   private static void saveINode2Image(ByteBuffer name,  
6.                                       INode node,  
7.                                       DataOutputStream out) throws IOException {  
8.     int nameLen = name.position();  
9.     out.writeShort(nameLen);  
10.     out.write(name.array(), name.arrayOffset(), nameLen);  
11.     if (!node.isDirectory()) {  // write file inode  
12.       INodeFile fileINode = (INodeFile)node;  
13.       //写入的属性包括,副本数,最近修改数据,最近访问时间  
14.       out.writeShort(fileINode.getReplication());  
15.       out.writeLong(fileINode.getModificationTime());  
16.       out.writeLong(fileINode.getAccessTime());  
17.       out.writeLong(fileINode.getPreferredBlockSize());  
18.       Block[] blocks = fileINode.getBlocks();  
19.       out.writeInt(blocks.length);  
20.       for (Block blk : blocks)  
21.         //将数据块信息也写入  
22.         blk.write(out);  
23.       FILE_PERM.fromShort(fileINode.getFsPermissionShort());  
24.       PermissionStatus.write(out, fileINode.getUserName(),  
25.                              fileINode.getGroupName(),  
26.                              FILE_PERM);  
27.     } else {   // write directory inode  
28.       //如果是目录,则还要写入节点的配额限制值  
29.       out.writeShort(0);  // replication  
30.       out.writeLong(node.getModificationTime());  
31.       out.writeLong(0);   // access time  
32.       out.writeLong(0);   // preferred block size  
33.       out.writeInt(-1);    // # of blocks  
34.       out.writeLong(node.getNsQuota());  
35.       out.writeLong(node.getDsQuota());  
36.       FILE_PERM.fromShort(node.getFsPermissionShort());  
37.       PermissionStatus.write(out, node.getUserName(),  
38.                              node.getGroupName(),  
39.                              FILE_PERM);  
40.     }  
41.   }  

在这里面会写入更多的关于文件目录的信息，此方法也会被saveImage()递归调用

[java] view plaincopyprint?

1. /** 
2.    * Save file tree image starting from the given root. 
3.    * This is a recursive procedure, which first saves all children of 
4.    * a current directory and then moves inside the sub-directories. 
5.    * 按照给定节点进行镜像的保存,每个节点目录会采取递归的方式进行遍历 
6.    */  
7.   private static void saveImage(ByteBuffer parentPrefix,  
8.                                 int prefixLength,  
9.                                 INodeDirectory current,  
10.                                 DataOutputStream out) throws IOException {  
11.     int newPrefixLength = prefixLength;  
12.     if (current.getChildrenRaw() == null)  
13.       return;  
14.     for(INode child : current.getChildren()) {  
15.       // print all children first  
16.       parentPrefix.position(prefixLength);  
17.       parentPrefix.put(PATH_SEPARATOR).put(child.getLocalNameBytes());  
18.       saveINode2Image(parentPrefix, child, out);  
19.       ....  
20.     }  

写入正在处理的文件的方法是一个static静态方法，是被外部方法所引用的

[java] view plaincopyprint?

1. // Helper function that writes an INodeUnderConstruction  
2.   // into the input stream  
3.   // 写入正在操作的文件的信息  
4.   //  
5.   static void writeINodeUnderConstruction(DataOutputStream out,  
6.                                            INodeFileUnderConstruction cons,  
7.                                            String path)   
8.                                            throws IOException {  
9.     writeString(path, out);  
10.     out.writeShort(cons.getReplication());  
11.     out.writeLong(cons.getModificationTime());  
12.     out.writeLong(cons.getPreferredBlockSize());  
13.     int nrBlocks = cons.getBlocks().length;  
14.     out.writeInt(nrBlocks);  
15.     for (int i = 0; i < nrBlocks; i++) {  
16.       cons.getBlocks()[i].write(out);  
17.     }  
18.     cons.getPermissionStatus().write(out);  
19.     writeString(cons.getClientName(), out);  
20.     writeString(cons.getClientMachine(), out);  
21.   
22.     out.writeInt(0); //  do not store locations of last block  
23.   }  

在这里顺便看下格式化相关的方法，格式化的操作是在每次开始使用HDFS前进行的，在这个过程中会生出新的版本号和命名空间ID，在代码中是如何实现的呢

[java] view plaincopyprint?

1. public void format() throws IOException {  
2.     this.layoutVersion = FSConstants.LAYOUT_VERSION;  
3.       //对每个目录进行格式化操作  
4.       format(sd);  
5.     }  
6.   }  
7.     
8.   /** Create new dfs name directory.  Caution: this destroys all files 
9.    * 格式化操作,会创建一个dfs/name的目录 
10.    * in this filesystem. */  
11.   void format(StorageDirectory sd) throws IOException {  
12.     sd.clearDirectory(); // create currrent dir  
13.     sd.lock();  
14.     try {  
15.       saveCurrent(sd);  
16.     } finally {  
17.       sd.unlock();  
18.     }  
19.     LOG.info("Storage directory " + sd.getRoot()  
20.              + " has been successfully formatted.");  
21.   }  

操作很简单，就是清空原有目录并创建新的目录。

编辑日志

下面开始另外一个大类的分析，就是编辑日志，英文名就是EditLog，这个里面将会有许多出彩的设计。打开这个类，你马上就看到的是几十种的操作码

[java] view plaincopyprint?

1. /** 
2.  * FSEditLog maintains a log of the namespace modifications. 
3.  * 编辑日志类包含了命名空间各种修改操作的日志记录 
4.  */  
5. public class FSEditLog {  
6.   //操作参数种类  
7.   private static final byte OP_INVALID = -1;  
8.   // 文件操作相关  
9.   private static final byte OP_ADD = 0;  
10.   private static final byte OP_RENAME = 1;  // rename  
11.   private static final byte OP_DELETE = 2;  // delete  
12.   private static final byte OP_MKDIR = 3;   // create directory  
13.   private static final byte OP_SET_REPLICATION = 4; // set replication  
14.   //the following two are used only for backward compatibility :  
15.   @Deprecated private static final byte OP_DATANODE_ADD = 5;  
16.   @Deprecated private static final byte OP_DATANODE_REMOVE = 6;  
17.   //下面2个权限设置相关  
18.   private static final byte OP_SET_PERMISSIONS = 7;  
19.   private static final byte OP_SET_OWNER = 8;  
20.   private static final byte OP_CLOSE = 9;    // close after write  
21.   private static final byte OP_SET_GENSTAMP = 10;    // store genstamp  
22.   /* The following two are not used any more. Should be removed once 
23.    * LAST_UPGRADABLE_LAYOUT_VERSION is -17 or newer. */  
24.   //配额设置相关  
25.   private static final byte OP_SET_NS_QUOTA = 11; // set namespace quota  
26.   private static final byte OP_CLEAR_NS_QUOTA = 12; // clear namespace quota  
27.   private static final byte OP_TIMES = 13; // sets mod & access time on a file  
28.   private static final byte OP_SET_QUOTA = 14; // sets name and disk quotas.  
29.   //Token认证相关  
30.   private static final byte OP_GET_DELEGATION_TOKEN = 18; //new delegation token  
31.   private static final byte OP_RENEW_DELEGATION_TOKEN = 19; //renew delegation token  
32.   private static final byte OP_CANCEL_DELEGATION_TOKEN = 20; //cancel delegation token  
33.   private static final byte OP_UPDATE_MASTER_KEY = 21; //update master key  

可见操作之多啊，然后才是基本变量的定义

[java] view plaincopyprint?

1. //日志刷入的缓冲大小值512k  
2.   private static int sizeFlushBuffer = 512*1024;  
3.     
4.   //编辑日志同时有多个输出流对象  
5.   private ArrayList<EditLogOutputStream> editStreams = null;  
6.   //内部维护了1个镜像类，与镜像进行交互  
7.   private FSImage fsimage = null;  
8.   
9.   // a monotonically increasing counter that represents transactionIds.  
10.   //每次进行同步刷新的事物ID  
11.   private long txid = 0;  
12.   
13.   // stores the last synced transactionId.  
14.   //最近一次已经同步的事物Id  
15.   private long synctxid = 0;  
16.   
17.   // the time of printing the statistics to the log file.  
18.   private long lastPrintTime;  
19.   
20.   // is a sync currently running?  
21.   //是否有日志同步操作正在进行  
22.   private boolean isSyncRunning;  
23.   
24.   // these are statistics counters.  
25.   //事务相关的统计变量  
26.   //事务的总数  
27.   private long numTransactions;        // number of transactions  
28.   //未能即使被同步的事物次数统计  
29.   private long numTransactionsBatchedInSync;  
30.   //事务的总耗时  
31.   private long totalTimeTransactions;  // total time for all transactions  
32.   private NameNodeInstrumentation metrics;  

这里txtid,synctxid等变量会在后面的同步操作时频繁出现。作者为了避免多线程事务id之间的相互干扰，采用了ThreadLocal的方式来维护自己的事务id

[java] view plaincopyprint?

1. //事物ID对象类,内部包含long类型txid值  
2.   private static class TransactionId {  
3.     //操作事物Id  
4.     public long txid;  
5.   
6.     TransactionId(long value) {  
7.       this.txid = value;  
8.     }  
9.   }  
10.   
11.   // stores the most current transactionId of this thread.  
12.   //通过ThreadLocal类保存线程私有的状态信息  
13.   private static final ThreadLocal<TransactionId> myTransactionId = new ThreadLocal<TransactionId>() {  
14.     protected synchronized TransactionId initialValue() {  
15.       return new TransactionId(Long.MAX_VALUE);  
16.     }  
17.   };  

在EditLog编辑日志中，所有的文件操作都是通过特有的EditLog输入输出流实现的，他是一个父类，这里以EditLogOutput为例

，代码被我简化了一些

[java] view plaincopyprint?

1. /** 
2.  * A generic abstract class to support journaling of edits logs into  
3.  * a persistent storage. 
4.  */  
5. abstract class EditLogOutputStream extends OutputStream {  
6.   // these are statistics counters  
7.   //下面是2个统计量  
8.   //文件同步的次数，可以理解为就是缓冲写入的次数  
9.   private long numSync;        // number of sync(s) to disk  
10.   //同步写入的总时间计数  
11.   private long totalTimeSync;  // total time to sync  
12.   
13.   EditLogOutputStream() throws IOException {  
14.     numSync = totalTimeSync = 0;  
15.   }  
16.     
17.   abstract String getName();  
18.   abstract public void write(int b) throws IOException;  
19.   abstract void write(byte op, Writable ... writables) throws IOException;  
20.   abstract void create() throws IOException;  
21.   abstract public void close() throws IOException;  
22.   abstract void setReadyToFlush() throws IOException;  
23.   abstract protected void flushAndSync() throws IOException;  
24.   
25.   /** 
26.    * Flush data to persistent store. 
27.    * Collect sync metrics. 
28.    * 刷出时间方法 
29.    */  
30.   public void flush() throws IOException {  
31.     //同步次数加1  
32.     numSync++;  
33.     long start = FSNamesystem.now();  
34.     //刷出同步方法为抽象方法，由继承的子类具体  
35.     flushAndSync();  
36.     long end = FSNamesystem.now();  
37.     //同时进行耗时的累加  
38.     totalTimeSync += (end - start);  
39.   }  
40.   
41.   abstract long length() throws IOException;  
42.     
43.   long getTotalSyncTime() {  
44.     return totalTimeSync;  
45.   }  
46.     
47.   long getNumSync() {  
48.     return numSync;  
49.   }  
50. }  

人家在这里对同步相关的操作做了一些设计，包括一些计数的统计。输入流与此类似，就不展开讨论了，但是EditLog并没有直接用了此类，而是在这个类中继承了一个内容更加丰富的EditLogFileOutputStream

[java] view plaincopyprint?

1. /** 
2.  * An implementation of the abstract class {@link EditLogOutputStream}, 
3.  * which stores edits in a local file. 
4.  * 所有的写日志文件的操作,都会通过这个输出流对象实现 
5.  */  
6. static private class EditLogFileOutputStream extends EditLogOutputStream {  
7.   private File file;  
8.   //内部维护了一个文件输出流对象  
9.   private FileOutputStream fp;    // file stream for storing edit logs   
10.   private FileChannel fc;         // channel of the file stream for sync  
11.   //这里设计了一个双缓冲区的设计,大大加强并发度,bufCurrent负责写入写入缓冲区  
12.   private DataOutputBuffer bufCurrent;  // current buffer for writing  
13.   //bufReady负载刷入数据到文件中  
14.   private DataOutputBuffer bufReady;    // buffer ready for flushing  
15.   static ByteBuffer fill = ByteBuffer.allocateDirect(512); // preallocation  

注意这里有双缓冲的设计，双缓冲的设计在许多的别的优秀的系统中都有用到。现在从编辑日志写文件开始看起

[java] view plaincopyprint?

1. /** 
2.    * Create empty edit log files. 
3.    * Initialize the output stream for logging. 
4.    *  
5.    * @throws IOException 
6.    */  
7.   public synchronized void open() throws IOException {  
8.     //在文件打开的时候,计数值都初始化0  
9.     numTransactions = totalTimeTransactions = numTransactionsBatchedInSync = 0;  
10.     if (editStreams == null) {  
11.       editStreams = new ArrayList<EditLogOutputStream>();  
12.     }  
13.     //传入目录类型获取迭代器  
14.     Iterator<StorageDirectory> it = fsimage.dirIterator(NameNodeDirType.EDITS);   
15.     while (it.hasNext()) {  
16.       StorageDirectory sd = it.next();  
17.       File eFile = getEditFile(sd);  
18.       try {  
19.         //打开存储目录下的文件获取输出流  
20.         EditLogOutputStream eStream = new EditLogFileOutputStream(eFile);  
21.         editStreams.add(eStream);  
22.       } catch (IOException ioe) {  
23.         fsimage.updateRemovedDirs(sd, ioe);  
24.         it.remove();  
25.       }  
26.     }  
27.     exitIfNoStreams();  
28.   }  

这里将会把一个新的输出流加入到editStreams全局变量中。那么对于一次标准的写入过程是怎么样的呢，我们以文件关闭的方法为例，因为文件关闭会触发一次最后剩余数据的写入操作

[java] view plaincopyprint?

1. /** 
2.  * Shutdown the file store. 
3.  * 关闭操作 
4.  */  
5. public synchronized void close() throws IOException {  
6.   while (isSyncRunning) {  
7.     //如果同正在进行，则等待1s  
8.     try {  
9.       wait(1000);  
10.     } catch (InterruptedException ie) {   
11.     }  
12.   }  
13.   if (editStreams == null) {  
14.     return;  
15.   }  
16.   printStatistics(true);  
17.   //当文件关闭的时候重置计数  
18.   numTransactions = totalTimeTransactions = numTransactionsBatchedInSync = 0;  
19.   
20.   for (int idx = 0; idx < editStreams.size(); idx++) {  
21.     EditLogOutputStream eStream = editStreams.get(idx);  
22.     try {  
23.       //关闭将最后的数据刷出缓冲  
24.       eStream.setReadyToFlush();  
25.       eStream.flush();  
26.       eStream.close();  
27.     } catch (IOException ioe) {  
28.       removeEditsAndStorageDir(idx);  
29.       idx--;  
30.     }  
31.   }  
32.   editStreams.clear();  
33. }  

主要是中的2行代码，setReadyToFlush()交换缓冲区

[java] view plaincopyprint?

1. /** 
2.      * All data that has been written to the stream so far will be flushed. 
3.      * New data can be still written to the stream while flushing is performed. 
4.      */  
5.     @Override  
6.     void setReadyToFlush() throws IOException {  
7.       assert bufReady.size() == 0 : "previous data is not flushed yet";  
8.       write(OP_INVALID);           // insert end-of-file marker  
9.       //交换2个缓冲区  
10.       DataOutputBuffer tmp = bufReady;  
11.       bufReady = bufCurrent;  
12.       bufCurrent = tmp;  
13.     }  

bufCurrent的缓冲用于外部写进行的数据缓冲，而bufReady则是将要写入文件的数据缓冲。而真正起作用的是flush()方法，他是父类中的方法

[java] view plaincopyprint?

1. /** 
2.  * Flush data to persistent store. 
3.  * Collect sync metrics. 
4.  * 刷出时间方法 
5.  */  
6. public void flush() throws IOException {  
7.   //同步次数加1  
8.   numSync++;  
9.   long start = FSNamesystem.now();  
10.   //刷出同步方法为抽象方法，由继承的子类具体  
11.   flushAndSync();  
12.   long end = FSNamesystem.now();  
13.   //同时进行耗时的累加  
14.   totalTimeSync += (end - start);  
15. }  

会调用到同步方法

[java] view plaincopyprint?

1. /** 
2.      * Flush ready buffer to persistent store. 
3.      * currentBuffer is not flushed as it accumulates new log records 
4.      * while readyBuffer will be flushed and synced. 
5.      */  
6.     @Override  
7.     protected void flushAndSync() throws IOException {  
8.       preallocate();            // preallocate file if necessary  
9.       //将ready缓冲区中的数据写入文件中  
10.       bufReady.writeTo(fp);     // write data to file  
11.       bufReady.reset();         // erase all data in the buffer  
12.       fc.force(false);          // metadata updates not needed because of preallocation  
13.       //跳过无效标志位，因为无效标志位每次都会写入  
14.       fc.position(fc.position()-1); // skip back the end-of-file marker  
15.     }  

你也许会想，简简单单的文件写入过程，的确设计的有点精巧。再回顾之前文件最顶上的几十种操作码类型，代表了各式各样的操作，他们是如何被调用的呢，第一反应当然是外界传入参数值，然后我调用相应语句做操作匹配，EditLog沿用的也是这个思路。

[java] view plaincopyprint?

1. /**  
2.    * Add set replication record to edit log 
3.    */  
4.   void logSetReplication(String src, short replication) {  
5.     logEdit(OP_SET_REPLICATION,   
6.             new UTF8(src),   
7.             FSEditLog.toLogReplication(replication));  
8.   }  
9.     
10.   /** Add set namespace quota record to edit log 
11.    *  
12.    * @param src the string representation of the path to a directory 
13.    * @param quota the directory size limit 
14.    */  
15.   void logSetQuota(String src, long nsQuota, long dsQuota) {  
16.     logEdit(OP_SET_QUOTA, new UTF8(src),   
17.             new LongWritable(nsQuota), new LongWritable(dsQuota));  
18.   }  
19.   
20.   /**  Add set permissions record to edit log */  
21.   void logSetPermissions(String src, FsPermission permissions) {  
22.     logEdit(OP_SET_PERMISSIONS, new UTF8(src), permissions);  
23.   }  

其实还有很多的logSet*系列的方法，形式都是传入操作码，操作对象以及附加参数，就会调用到更加基层的logEdit方法，这个方法才是最终写入操作记录的方法。

[java] view plaincopyprint?

1. /** 
2.    * Write an operation to the edit log. Do not sync to persistent 
3.    * store yet. 
4.    * 写入一个操作到编辑日志中 
5.    */  
6.   synchronized void logEdit(byte op, Writable ... writables) {  
7.     if (getNumEditStreams() < 1) {  
8.       throw new AssertionError("No edit streams to log to");  
9.     }  
10.     long start = FSNamesystem.now();  
11.     for (int idx = 0; idx < editStreams.size(); idx++) {  
12.       EditLogOutputStream eStream = editStreams.get(idx);  
13.       try {  
14.         // 写入操作到每个输出流中  
15.         eStream.write(op, writables);  
16.       } catch (IOException ioe) {  
17.         removeEditsAndStorageDir(idx);  
18.         idx--;   
19.       }  
20.     }  
21.     exitIfNoStreams();  
22.     // get a new transactionId  
23.     //获取一个新的事物Id  
24.     txid++;  
25.   
26.     //  
27.     // record the transactionId when new data was written to the edits log  
28.     //  
29.     TransactionId id = myTransactionId.get();  
30.     id.txid = txid;  
31.   
32.     // update statistics  
33.     long end = FSNamesystem.now();  
34.     //在每次进行logEdit写入记录操作的时候,都会累加事物次数和耗时  
35.     numTransactions++;  
36.     totalTimeTransactions += (end-start);  
37.     if (metrics != null) // Metrics is non-null only when used inside name node  
38.       metrics.addTransaction(end-start);  
39.   }  

每次新的操作，在这里都生成一个新的事务id，并且会统计事务执行写入缓冲时间等，但是此时只是写入的输出流中，还没有写到文件。原因是你要考虑到多线程操作的情况。

[java] view plaincopyprint?

1. //  
2.   // Sync all modifications done by this thread.  
3.   //  
4.   public void logSync() throws IOException {  
5.     ArrayList<EditLogOutputStream> errorStreams = null;  
6.     long syncStart = 0;  
7.   
8.     // Fetch the transactionId of this thread.   
9.     long mytxid = myTransactionId.get().txid;  
10.   
11.     ArrayList<EditLogOutputStream> streams = new ArrayList<EditLogOutputStream>();  
12.     boolean sync = false;  
13.     try {  
14.       synchronized (this) {  
15.         printStatistics(false);  
16.   
17.         // if somebody is already syncing, then wait  
18.         while (mytxid > synctxid && isSyncRunning) {  
19.           try {  
20.             wait(1000);  
21.           } catch (InterruptedException ie) {   
22.           }  
23.         }  
24.   
25.         //  
26.         // If this transaction was already flushed, then nothing to do  
27.         //  
28.         if (mytxid <= synctxid) {  
29.           //当执行的事物id小于已同步的Id,也进行计数累加  
30.           numTransactionsBatchedInSync++;  
31.           if (metrics != null) // Metrics is non-null only when used inside name node  
32.             metrics.incrTransactionsBatchedInSync();  
33.           return;  
34.         }  
35.   
36.         // now, this thread will do the sync  
37.         syncStart = txid;  
38.         isSyncRunning = true;  
39.         sync = true;  
40.   
41.         // swap buffers  
42.         exitIfNoStreams();  
43.         for(EditLogOutputStream eStream : editStreams) {  
44.           try {  
45.             //交换缓冲  
46.             eStream.setReadyToFlush();  
47.             streams.add(eStream);  
48.           } catch (IOException ie) {  
49.             FSNamesystem.LOG.error("Unable to get ready to flush.", ie);  
50.             //  
51.             // remember the streams that encountered an error.  
52.             //  
53.             if (errorStreams == null) {  
54.               errorStreams = new ArrayList<EditLogOutputStream>(1);  
55.             }  
56.             errorStreams.add(eStream);  
57.           }  
58.         }  
59.       }  
60.   
61.       // do the sync  
62.       long start = FSNamesystem.now();  
63.       for (EditLogOutputStream eStream : streams) {  
64.         try {  
65.           //同步完成之后，做输入数据操作  
66.           eStream.flush();  
67.        ....  
68.   }  

ok，整个操作过程总算理清了。写入的过程完成之后，编辑日志类是如何读入编辑日志文件，并完成内存元数据的恢复 的呢，整个过程其实就是一个解码的过程

[java] view plaincopyprint?

1. /** 
2.    * Load an edit log, and apply the changes to the in-memory structure 
3.    * This is where we apply edits that we've been writing to disk all 
4.    * along. 
5.    * 导入编辑日志文件,并在内存中构建此时状态 
6.    */  
7.   static int loadFSEdits(EditLogInputStream edits) throws IOException {  
8.     FSNamesystem fsNamesys = FSNamesystem.getFSNamesystem();  
9.     //FSDirectory是一个门面模式的体现,所有的操作都是在这个类中分给里面的子系数实现  
10.     FSDirectory fsDir = fsNamesys.dir;  
11.     int numEdits = 0;  
12.     int logVersion = 0;  
13.     String clientName = null;  
14.     String clientMachine = null;  
15.     String path = null;  
16.     int numOpAdd = 0, numOpClose = 0, numOpDelete = 0,  
17.         numOpRename = 0, numOpSetRepl = 0, numOpMkDir = 0,  
18.         numOpSetPerm = 0, numOpSetOwner = 0, numOpSetGenStamp = 0,  
19.         numOpTimes = 0, numOpGetDelegationToken = 0,  
20.         numOpRenewDelegationToken = 0, numOpCancelDelegationToken = 0,  
21.         numOpUpdateMasterKey = 0, numOpOther = 0;  
22.   
23.     long startTime = FSNamesystem.now();  
24.   
25.     DataInputStream in = new DataInputStream(new BufferedInputStream(edits));  
26.     try {  
27.       // Read log file version. Could be missing.   
28.       in.mark(4);  
29.       // If edits log is greater than 2G, available method will return negative  
30.       // numbers, so we avoid having to call available  
31.       boolean available = true;  
32.       try {  
33.         // 首先读入日志版本号  
34.         logVersion = in.readByte();  
35.       } catch (EOFException e) {  
36.         available = false;  
37.       }  
38.       if (available) {  
39.         in.reset();  
40.         logVersion = in.readInt();  
41.         if (logVersion < FSConstants.LAYOUT_VERSION) // future version  
42.           throw new IOException(  
43.                           "Unexpected version of the file system log file: "  
44.                           + logVersion + ". Current version = "   
45.                           + FSConstants.LAYOUT_VERSION + ".");  
46.       }  
47.       assert logVersion <= Storage.LAST_UPGRADABLE_LAYOUT_VERSION :  
48.                             "Unsupported version " + logVersion;  
49.   
50.       while (true) {  
51.         ....  
52.           
53.         //下面根据操作类型进行值的设置  
54.         switch (opcode) {  
55.         case OP_ADD:  
56.         case OP_CLOSE: {  
57.           ...  
58.           break;  
59.         }   
60.         case OP_SET_REPLICATION: {  
61.           numOpSetRepl++;  
62.           path = FSImage.readString(in);  
63.           short replication = adjustReplication(readShort(in));  
64.           fsDir.unprotectedSetReplication(path, replication, null);  
65.           break;  
66.         }   
67.         case OP_RENAME: {  
68.           numOpRename++;  
69.           int length = in.readInt();  
70.           if (length != 3) {  
71.             throw new IOException("Incorrect data format. "   
72.                                   + "Mkdir operation.");  
73.           }  
74.           String s = FSImage.readString(in);  
75.           String d = FSImage.readString(in);  
76.           timestamp = readLong(in);  
77.           HdfsFileStatus dinfo = fsDir.getFileInfo(d);  
78.           fsDir.unprotectedRenameTo(s, d, timestamp);  
79.           fsNamesys.changeLease(s, d, dinfo);  
80.           break;  
81.         }  
82.         ...  

整个函数代码非常长，大家理解思路即可。里面的很多操作都是在FSDirectory中实现的，你可以理解整个类为一个门面模式，各个相关的子系统都包括在这个类中。

NameNode元数据备份机制

有了以上的方法做铺垫，元数据的备份机制才得以灵活的实现，无非就是调用上述的基础方法进行各个文件的拷贝，重命名等操作。整体上需要发生文件状态变化的操作如下：

1.原current镜像目录-->lastcheckpoint.tmp

2.第二名字节点上传新的镜像文件后fsimage.ckpt-->fsimage，并创建新的current目录

3.lastcheckpoint.tmp变为previous.checkpoint

4.日志文件edit.new-->edit文件

大体上是以上4条思路。

首先镜像文件的备份都是从第二名字节点的周期性检查点检测开始的

[java] view plaincopyprint?

1. //  
2.   // The main work loop  
3.   //  
4.   public void doWork() {  
5.     long period = 5 * 60;              // 5 minutes  
6.     long lastCheckpointTime = 0;  
7.     if (checkpointPeriod < period) {  
8.       period = checkpointPeriod;  
9.     }  
10.       
11.     //主循环程序  
12.     while (shouldRun) {  
13.       try {  
14.         Thread.sleep(1000 * period);  
15.       } catch (InterruptedException ie) {  
16.         // do nothing  
17.       }  
18.       if (!shouldRun) {  
19.         break;  
20.       }  
21.       try {  
22.         // We may have lost our ticket since last checkpoint, log in again, just in case  
23.         if(UserGroupInformation.isSecurityEnabled())  
24.           UserGroupInformation.getCurrentUser().reloginFromKeytab();  
25.           
26.         long now = System.currentTimeMillis();  
27.   
28.         long size = namenode.getEditLogSize();  
29.         if (size >= checkpointSize ||   
30.             now >= lastCheckpointTime + 1000 * checkpointPeriod) {  
31.           //周期性调用检查点方法  
32.           doCheckpoint();  
33.           ...  
34.     }  
35.   }  

然后我们就找doCheckpoint()检查点检查方法

[java] view plaincopyprint?

1. /** 
2.    * Create a new checkpoint 
3.    */  
4.   void doCheckpoint() throws IOException {  
5.   
6.     // Do the required initialization of the merge work area.  
7.     //做初始化的镜像操作  
8.     startCheckpoint();  
9.   
10.     // Tell the namenode to start logging transactions in a new edit file  
11.     // Retuns a token that would be used to upload the merged image.  
12.     CheckpointSignature sig = (CheckpointSignature)namenode.rollEditLog();  
13.   
14.     // error simulation code for junit test  
15.     if (ErrorSimulator.getErrorSimulation(0)) {  
16.       throw new IOException("Simulating error0 " +  
17.                             "after creating edits.new");  
18.     }  
19.   
20.     //从名字节点获取当前镜像或编辑日志  
21.     downloadCheckpointFiles(sig);   // Fetch fsimage and edits  
22.     //进行镜像合并操作  
23.     doMerge(sig);                   // Do the merge  
24.     
25.     //  
26.     // Upload the new image into the NameNode. Then tell the Namenode  
27.     // to make this new uploaded image as the most current image.  
28.     //把合并好后的镜像重新上传到名字节点  
29.     putFSImage(sig);  
30.   
31.     // error simulation code for junit test  
32.     if (ErrorSimulator.getErrorSimulation(1)) {  
33.       throw new IOException("Simulating error1 " +  
34.                             "after uploading new image to NameNode");  
35.     }  
36.       
37.     //通知名字节点进行镜像的替换操作，包括将edit.new的名称重新改为edit，镜像名称fsimage.ckpt改为fsImage  
38.     namenode.rollFsImage();  
39.     checkpointImage.endCheckpoint();  
40.   
41.     LOG.info("Checkpoint done. New Image Size: "  
42.               + checkpointImage.getFsImageName().length());  
43.   }  

这个方法中描述了非常清晰的备份机制。我们主要再来看下文件的替换方法，也就是namenode.rollFsImage方法，这个方法最后还是会调到FSImage的同名方法。

[java] view plaincopyprint?

1. /** 
2.    * Moves fsimage.ckpt to fsImage and edits.new to edits 
3.    * Reopens the new edits file. 
4.    * 完成2个文件的名称替换 
5.    */  
6.   void rollFSImage() throws IOException {  
7.     if (ckptState != CheckpointStates.UPLOAD_DONE) {  
8.       throw new IOException("Cannot roll fsImage before rolling edits log.");  
9.     }  
10.     //  
11.     // First, verify that edits.new and fsimage.ckpt exists in all  
12.     // checkpoint directories.  
13.     //  
14.     if (!editLog.existsNew()) {  
15.       throw new IOException("New Edits file does not exist");  
16.     }  
17.     Iterator<StorageDirectory> it = dirIterator(NameNodeDirType.IMAGE);  
18.     while (it.hasNext()) {  
19.       StorageDirectory sd = it.next();  
20.       File ckpt = getImageFile(sd, NameNodeFile.IMAGE_NEW);  
21.       if (!ckpt.exists()) {  
22.         throw new IOException("Checkpoint file " + ckpt +  
23.                               " does not exist");  
24.       }  
25.     }  
26.     editLog.purgeEditLog(); // renamed edits.new to edits  

方法前半部分交待的很明确，做2类文件的替换，

[java] view plaincopyprint?

1. //  
2.     // Renames new image  
3.     // 重命名新镜像名称  
4.     //  
5.     it = dirIterator(NameNodeDirType.IMAGE);  
6.     while (it.hasNext()) {  
7.       StorageDirectory sd = it.next();  
8.       File ckpt = getImageFile(sd, NameNodeFile.IMAGE_NEW);  
9.       File curFile = getImageFile(sd, NameNodeFile.IMAGE);  
10.       // renameTo fails on Windows if the destination file   
11.       // already exists.  
12.       if (!ckpt.renameTo(curFile)) {  
13.         curFile.delete();  
14.         if (!ckpt.renameTo(curFile)) {  
15.           editLog.removeEditsForStorageDir(sd);  
16.           updateRemovedDirs(sd);  
17.           it.remove();  
18.         }  
19.       }  
20.     }  
21.     editLog.exitIfNoStreams();  

中间代码部分完成fsimage.ckpt的新镜像重命名为当前名称fsimage，最后要对旧的目录文件进行删除操作

[java] view plaincopyprint?

1. //  
2.     // Updates the fstime file on all directories (fsimage and edits)  
3.     // and write version file  
4.     //  
5.     this.layoutVersion = FSConstants.LAYOUT_VERSION;  
6.     this.checkpointTime = FSNamesystem.now();  
7.     it = dirIterator();  
8.     while (it.hasNext()) {  
9.       StorageDirectory sd = it.next();  
10.       // delete old edits if sd is the image only the directory  
11.       if (!sd.getStorageDirType().isOfType(NameNodeDirType.EDITS)) {  
12.         File editsFile = getImageFile(sd, NameNodeFile.EDITS);  
13.         editsFile.delete();  
14.       }  
15.       // delete old fsimage if sd is the edits only the directory  
16.       if (!sd.getStorageDirType().isOfType(NameNodeDirType.IMAGE)) {  
17.         File imageFile = getImageFile(sd, NameNodeFile.IMAGE);  
18.         imageFile.delete();  
19.       }  

这个过程本身比较复杂，还是用书中的一张图来表示好了，图可能有点大。

总结

至此，本篇内容阐述完毕，其实我还是忽略了许多细节部分，还是主要从主要的操作一步步的理清整个线索。建议可以在Hadoop运行的时候，跑到name和edit目录下观察各个目录的情况以此验证这整套机制了，或直接做upgrade测试升级工作也是可以的。全部代码的分析请点击链接

https://github.com/linyiqun/hadoop-hdfs,后续将会继续更新HDFS其他方面的代码分析。

参考文献

《Hadoop技术内部–HDFS结构设计与实现原理》.蔡斌等