HDFS源码分析EditLog之读取操作符

2023-01-31 19:42:02

在《HDFS源码分析EditLog之获取编辑日志输入流》一文中，我们详细了解了如何获取编辑日志输入流EditLogInputStream。在我们得到编辑日志输入流后，是不是就该从输入流中获取数据来处理呢？答案是显而易见的！在《HDFS源码分析之EditLogTailer》一文中，我们在讲编辑日志追踪同步时，也讲到了如下两个连续的处理流程：

4、从编辑日志editLog中获取编辑日志输入流集合streams，获取的输入流为最新事务ID加1之后的数据
        5、调用文件系统镜像FSImage实例image的loadEdits()，利用编辑日志输入流集合streams，加载编辑日志至目标namesystem中的文件系统镜像FSImage，并获得编辑日志加载的大小editsLoaded；

可见，我们在获得编辑日志输入流EditLogInputStream的集合streams后，就需要调用FSImage的loadEdits()方法，利用编辑日志输入流集合streams，加载编辑日志至目标namesystem中的文件系统镜像FSImage。而HDFS是如何从编辑日志输入流中读取数据的呢？本文，我们将进行详细的探究！

首先，在加载编辑日志的主要类FSEditLogLoader中，其核心方法loadEditRecords()中有如下一段代码：

  1. while (true) {
  2. try {
  3. FSEditLogOp op;
  4. try {
  5. // 从编辑日志输入流in中读取操作符op
  6. op = in.readOp();
  7. // 如果操作符op为空，直接跳出循环，并返回
  8. if (op == null) {
  9. break;
  10. }
  11. } catch (Throwable e) {
  12. // ...省略部分代码
  13. }
  14. // ...省略部分代码
  15. try {
  16. // ...省略部分代码
  17. long inodeId = applyEditLogOp(op, fsDir, startOpt,
  18. in.getVersion(true), lastInodeId);
  19. if (lastInodeId < inodeId) {
  20. lastInodeId = inodeId;
  21. }
  22. } catch (RollingUpgradeOp.RollbackException e) {
  23. // ...省略部分代码
  24. } catch (Throwable e) {
  25. // ...省略部分代码
  26. }
  27. // ...省略部分代码
  28. } catch (RollingUpgradeOp.RollbackException e) {
  29. // ...省略部分代码
  30. } catch (MetaRecoveryContext.RequestStopException e) {
  31. // ...省略部分代码
  32. }
  33. }

它会从编辑日志输入流in中读取一个操作符op，然后调用applyEditLogOp()方法，将操作符作用于内存元数据FSNamesystem。那么问题来了，这个操作符如何从数据流中被读取并解析的呢？

接下来，我们就看下如何从编辑日志输出流EditLogInputStream中读取一个操作符，我们先看其readOp()方法，代码如下：

  1. /**
  2. * Read an operation from the stream
  3. * @return an operation from the stream or null if at end of stream
  4. * @throws IOException if there is an error reading from the stream
  5. */
  6. public FSEditLogOp readOp() throws IOException {
  7. FSEditLogOp ret;
  8. // 如果缓存的cachedOp不为null，返回缓存的cachedOp，并将其清空
  9. if (cachedOp != null) {
  10. ret = cachedOp;
  11. cachedOp = null;
  12. return ret;
  13. }
  14. // 如果缓存的cachedOp为null，调用nextOp()进行处理
  15. return nextOp();
  16. }

很简单，如果缓存的cachedOp不为null，返回缓存的cachedOp，并将其清空，如果缓存的cachedOp为null，则调用nextOp()进行处理。而EditLogInputStream中nextOp()是一个抽象方法，我们需要看其子类的实现方法，下面就以EditLogFileInputStream为例，看下其nextOp()方法：

  1. @Override
  2. protected FSEditLogOp nextOp() throws IOException {
  3. return nextOpImpl(false);
  4. }

继续追踪nextOpImpl()方法，代码如下：

  1. private FSEditLogOp nextOpImpl(boolean skipBrokenEdits) throws IOException {
  2. FSEditLogOp op = null;
  3. // 根据编辑日志文件输入流的状态判断：
  4. switch (state) {
  5. case UNINIT:// 如果为未初始化状态UNINIT
  6. try {
  7. // 调用init()方法进行初始化
  8. init(true);
  9. } catch (Throwable e) {
  10. LOG.error("caught exception initializing " + this, e);
  11. if (skipBrokenEdits) {
  12. return null;
  13. }
  14. Throwables.propagateIfPossible(e, IOException.class);
  15. }
  16. // 检测编辑日志文件输入流状态，此时不应为UNINIT
  17. Preconditions.checkState(state != State.UNINIT);
  18. // 再次调用nextOpImpl()方法
  19. return nextOpImpl(skipBrokenEdits);
  20. case OPEN:// 如果为打开OPEN状态
  21. // 调用FSEditLogOp.Reader的readOp()方法，读取操作符
  22. op = reader.readOp(skipBrokenEdits);
  23. if ((op != null) && (op.hasTransactionId())) {
  24. long txId = op.getTransactionId();
  25. if ((txId >= lastTxId) &&
  26. (lastTxId != HdfsConstants.INVALID_TXID)) {
  27. //
  28. // Sometimes, the NameNode crashes while it's writing to the
  29. // edit log.  In that case, you can end up with an unfinalized edit log
  30. // which has some garbage at the end.
  31. // JournalManager#recoverUnfinalizedSegments will finalize these
  32. // unfinished edit logs, giving them a defined final transaction
  33. // ID.  Then they will be renamed, so that any subsequent
  34. // readers will have this information.
  35. //
  36. // Since there may be garbage at the end of these "cleaned up"
  37. // logs, we want to be sure to skip it here if we've read everything
  38. // we were supposed to read out of the stream.
  39. // So we force an EOF on all subsequent reads.
  40. //
  41. long skipAmt = log.length() - tracker.getPos();
  42. if (skipAmt > 0) {
  43. if (LOG.isDebugEnabled()) {
  44. LOG.debug("skipping " + skipAmt + " bytes at the end " +
  45. "of edit log  '" + getName() + "': reached txid " + txId +
  46. " out of " + lastTxId);
  47. }
  48. tracker.clearLimit();
  49. IOUtils.skipFully(tracker, skipAmt);
  50. }
  51. }
  52. }
  53. break;
  54. case CLOSED: // 如果为关闭CLOSED状态，直接返回null
  55. break; // return null
  56. }
  57. return op;
  58. }

nextOpImpl()方法的大体处理逻辑如下：

根据编辑日志文件输入流的状态判断：

1、如果为未初始化状态UNINIT，调用init()方法进行初始化，然后检测编辑日志文件输入流状态，此时不应为UNINIT，最后再次调用nextOpImpl()方法；

2、如果为打开OPEN状态，调用FSEditLogOp.Reader的readOp()方法，读取操作符op；

3、如果为关闭CLOSED状态，直接返回null。

我们重点关注下FSEditLogOp.Reader的readOp()方法，代码如下：

  1. /**
  2. * Read an operation from the input stream.
  3. *
  4. * Note that the objects returned from this method may be re-used by future
  5. * calls to the same method.
  6. *
  7. * @param skipBrokenEdits    If true, attempt to skip over damaged parts of
  8. * the input stream, rather than throwing an IOException
  9. * @return the operation read from the stream, or null at the end of the
  10. *         file
  11. * @throws IOException on error.  This function should only throw an
  12. *         exception when skipBrokenEdits is false.
  13. */
  14. public FSEditLogOp readOp(boolean skipBrokenEdits) throws IOException {
  15. while (true) {
  16. try {
  17. // 调用decodeOp()方法
  18. return decodeOp();
  19. } catch (IOException e) {
  20. in.reset();
  21. if (!skipBrokenEdits) {
  22. throw e;
  23. }
  24. } catch (RuntimeException e) {
  25. // FSEditLogOp#decodeOp is not supposed to throw RuntimeException.
  26. // However, we handle it here for recovery mode, just to be more
  27. // robust.
  28. in.reset();
  29. if (!skipBrokenEdits) {
  30. throw e;
  31. }
  32. } catch (Throwable e) {
  33. in.reset();
  34. if (!skipBrokenEdits) {
  35. throw new IOException("got unexpected exception " +
  36. e.getMessage(), e);
  37. }
  38. }
  39. // Move ahead one byte and re-try the decode process.
  40. if (in.skip(1) < 1) {
  41. return null;
  42. }
  43. }
  44. }

继续追踪decodeOp()方法，代码如下：

  1. /**
  2. * Read an opcode from the input stream.
  3. * 从输入流中读取一个操作符code
  4. *
  5. * @return   the opcode, or null on EOF.
  6. *
  7. * If an exception is thrown, the stream's mark will be set to the first
  8. * problematic byte.  This usually means the beginning of the opcode.
  9. */
  10. private FSEditLogOp decodeOp() throws IOException {
  11. limiter.setLimit(maxOpSize);
  12. in.mark(maxOpSize);
  13. if (checksum != null) {
  14. checksum.reset();
  15. }
  16. byte opCodeByte;
  17. try {
  18. // 从输入流in中读取一个byte，即opCodeByte
  19. opCodeByte = in.readByte();
  20. } catch (EOFException eof) {
  21. // EOF at an opcode boundary is expected.
  22. return null;
  23. }
  24. // 将byte类型的opCodeByte转换成FSEditLogOpCodes对象opCode
  25. FSEditLogOpCodes opCode = FSEditLogOpCodes.fromByte(opCodeByte);
  26. if (opCode == OP_INVALID) {
  27. verifyTerminator();
  28. return null;
  29. }
  30. // 根据FSEditLogOpCodes对象opCode从cache中获取FSEditLogOp对象op
  31. FSEditLogOp op = cache.get(opCode);
  32. if (op == null) {
  33. throw new IOException("Read invalid opcode " + opCode);
  34. }
  35. // 如果支持编辑日志长度，从输入流读入一个int，
  36. if (supportEditLogLength) {
  37. in.readInt();
  38. }
  39. if (NameNodeLayoutVersion.supports(
  40. LayoutVersion.Feature.STORED_TXIDS, logVersion)) {
  41. // Read the txid
  42. // 如果支持事务ID，读入一个long，作为事务ID，并在FSEditLogOp实例op中设置事务ID
  43. op.setTransactionId(in.readLong());
  44. } else {
  45. // 如果不支持事务ID，在FSEditLogOp实例op中设置事务ID为-12345
  46. op.setTransactionId(HdfsConstants.INVALID_TXID);
  47. }
  48. // 从输入流in中读入其他域，并设置入FSEditLogOp实例op
  49. op.readFields(in, logVersion);
  50. validateChecksum(in, checksum, op.txid);
  51. return op;
  52. }

decodeOp()方法的逻辑很简单：

1、从输入流in中读取一个byte，即opCodeByte，确定操作类型；

2、将byte类型的opCodeByte转换成FSEditLogOpCodes对象opCode；

3、根据FSEditLogOpCodes对象opCode从cache中获取FSEditLogOp对象op，这样我们就得到了操作符对象；

4、如果支持编辑日志长度，从输入流读入一个int；

5、如果支持事务ID，读入一个long，作为事务ID，并在FSEditLogOp实例op中设置事务ID，否则在FSEditLogOp实例op中设置事务ID为-12345；

6、调用操作符对象op的readFields()方法，从输入流in中读入其他域，并设置入FSEditLogOp实例op。

接下来，我们再看下操作符对象的readFields()方法，因为不同种类的操作符肯定包含不同的属性，所以它们的readFields()方法肯定也各不相同。下面，我们就以操作符AddCloseOp为例来分析，其readFields()方法如下：

  1. @Override
  2. void readFields(DataInputStream in, int logVersion)
  3. throws IOException {
  4. // 读取长度：如果支持读入长度，从输入流in读取一个int，赋值给length
  5. if (!NameNodeLayoutVersion.supports(
  6. LayoutVersion.Feature.EDITLOG_OP_OPTIMIZATION, logVersion)) {
  7. this.length = in.readInt();
  8. }
  9. // 读取节点ID：如果支持读入节点ID，从输入流in读取一个long，赋值给inodeId，否则inodeId默认为0
  10. if (NameNodeLayoutVersion.supports(
  11. LayoutVersion.Feature.ADD_INODE_ID, logVersion)) {
  12. this.inodeId = in.readLong();
  13. } else {
  14. // The inodeId should be updated when this editLogOp is applied
  15. this.inodeId = INodeId.GRANDFATHER_INODE_ID;
  16. }
  17. // 版本兼容性校验
  18. if ((-17 < logVersion && length != 4) ||
  19. (logVersion <= -17 && length != 5 && !NameNodeLayoutVersion.supports(
  20. LayoutVersion.Feature.EDITLOG_OP_OPTIMIZATION, logVersion))) {
  21. throw new IOException("Incorrect data format."  +
  22. " logVersion is " + logVersion +
  23. " but writables.length is " +
  24. length + ". ");
  25. }
  26. // 读取路径：从输入流in读取一个String，赋值给path
  27. this.path = FSImageSerialization.readString(in);
  28. // 读取副本数、修改时间：如果支持读取副本数、修改时间，分别从输入流读取一个short、long，
  29. // 赋值给replication、mtime
  30. if (NameNodeLayoutVersion.supports(
  31. LayoutVersion.Feature.EDITLOG_OP_OPTIMIZATION, logVersion)) {
  32. this.replication = FSImageSerialization.readShort(in);
  33. this.mtime = FSImageSerialization.readLong(in);
  34. } else {
  35. this.replication = readShort(in);
  36. this.mtime = readLong(in);
  37. }
  38. // 读取访问时间：如果支持读取访问时间，从输入流读取一个long，赋值给atime，否则atime默认为0
  39. if (NameNodeLayoutVersion.supports(
  40. LayoutVersion.Feature.FILE_ACCESS_TIME, logVersion)) {
  41. if (NameNodeLayoutVersion.supports(
  42. LayoutVersion.Feature.EDITLOG_OP_OPTIMIZATION, logVersion)) {
  43. this.atime = FSImageSerialization.readLong(in);
  44. } else {
  45. this.atime = readLong(in);
  46. }
  47. } else {
  48. this.atime = 0;
  49. }
  50. // 读取数据块大小：如果支持读取数据块大小，从输入流读取一个long，赋值给blockSize
  51. if (NameNodeLayoutVersion.supports(
  52. LayoutVersion.Feature.EDITLOG_OP_OPTIMIZATION, logVersion)) {
  53. this.blockSize = FSImageSerialization.readLong(in);
  54. } else {
  55. this.blockSize = readLong(in);
  56. }
  57. // 调用readBlocks()方法读取数据块，赋值给数据块数组blocks
  58. this.blocks = readBlocks(in, logVersion);
  59. // 从输入流读入权限，赋值给permissions
  60. this.permissions = PermissionStatus.read(in);
  61. // 如果是ADD操作，需要额外处理客户端名称clientName、客户端机器clientMachine、覆盖写标志overwrite等属性
  62. if (this.opCode == OP_ADD) {
  63. aclEntries = AclEditLogUtil.read(in, logVersion);
  64. this.xAttrs = readXAttrsFromEditLog(in, logVersion);
  65. this.clientName = FSImageSerialization.readString(in);
  66. this.clientMachine = FSImageSerialization.readString(in);
  67. if (NameNodeLayoutVersion.supports(
  68. NameNodeLayoutVersion.Feature.CREATE_OVERWRITE, logVersion)) {
  69. this.overwrite = FSImageSerialization.readBoolean(in);
  70. } else {
  71. this.overwrite = false;
  72. }
  73. if (NameNodeLayoutVersion.supports(
  74. NameNodeLayoutVersion.Feature.BLOCK_STORAGE_POLICY, logVersion)) {
  75. this.storagePolicyId = FSImageSerialization.readByte(in);
  76. } else {
  77. this.storagePolicyId = BlockStoragePolicySuite.ID_UNSPECIFIED;
  78. }
  79. // read clientId and callId
  80. readRpcIds(in, logVersion);
  81. } else {
  82. this.clientName = "";
  83. this.clientMachine = "";
  84. }
  85. }

这个没有什么特别好讲的，依次读入操作符需要的，在输入流中依次存在的属性即可。
        不过，我们仍然需要重点讲解下读入数据块的readBlocks()方法，代码如下：

  1. private static Block[] readBlocks(
  2. DataInputStream in,
  3. int logVersion) throws IOException {
  4. // 读取block数目numBlocks，占一个int
  5. int numBlocks = in.readInt();
  6. // 校验block数目numBlocks，应大于等于0，小于等于1024 * 1024 * 64
  7. if (numBlocks < 0) {
  8. throw new IOException("invalid negative number of blocks");
  9. } else if (numBlocks > MAX_BLOCKS) {
  10. throw new IOException("invalid number of blocks: " + numBlocks +
  11. ".  The maximum number of blocks per file is " + MAX_BLOCKS);
  12. }
  13. // 构造block数组blocks，大小即为numBlocks
  14. Block[] blocks = new Block[numBlocks];
  15. // 从输入流中读取numBlocks个数据块
  16. for (int i = 0; i < numBlocks; i++) {
  17. // 构造数据块Block实例blk
  18. Block blk = new Block();
  19. // 调用Block的readFields()方法，从输入流读入数据块
  20. blk.readFields(in);
  21. // 将数据块blk放入数据块数组blocks
  22. blocks[i] = blk;
  23. }
  24. // 返回数据块数组blocks
  25. return blocks;
  26. }

很简单，先从输入流读取block数目numBlocks，确定一共需要读取多少个数据块，然后构造block数组blocks，大小即为numBlocks，最后从输入流中读取numBlocks个数据块，每次都是先构造数据块Block实例blk，调用Block的readFields()方法，从输入流读入数据块，然后将数据块blk放入数据块数组blocks。全部数据块读取完毕后，返回数据块数组blocks。

我们再看下数据块Block的readFields()方法，如下：

  1. @Override // Writable
  2. public void readFields(DataInput in) throws IOException {
  3. readHelper(in);
  4. }

继续看readHelper()方法，如下：

  1. final void readHelper(DataInput in) throws IOException {
  2. // 从输入流读取一个long，作为数据块艾迪blockId
  3. this.blockId = in.readLong();
  4. // 从输入流读取一个long，作为数据块大小numBytes
  5. this.numBytes = in.readLong();
  6. // 从输入流读取一个long，作为数据块产生的时间戳generationStamp
  7. this.generationStamp = in.readLong();
  8. // 校验：数据块大小numBytes应大于等于0
  9. if (numBytes < 0) {
  10. throw new IOException("Unexpected block size: " + numBytes);
  11. }
  12. }

从输入流依次读入数据块艾迪blockId、数据块大小numBytes、数据块产生的时间戳generationStamp即可，三者均为long类型。

上一篇：HDFS源码分析EditLog之获取编辑日志输入流


下一篇：[HeadFirst-HTMLCSS学习笔记][第三章创建网页]