FSEditLog解析

在 Namenode 中，命名空间，也就是文件系统中的目录树、 文件元数据等信息，是被全部缓存在内存中的， 一旦 Namenode 重启或者宕机，内存中的所有数据将会全部丢失，所以必须要有一种机制能够将整个命名空间持久化保存， 并且能在 Namenode 重启时重建命名空间

目前 Namenode 的实现是将命名空间信息记录在一个叫作 fsimage（命名空间镜像）的二进制文件中， fsimage将文件系统目录树中的每个文件或者目录的信息保存为一条记录， 每条记录中包括了该文件（ 或目录）的名称、大小、用户、用户组、修改时间、创建时间等信息

Namenode 重启时，会读取这个 fsimage 文件来重构命名空间。但是 fsimage 始终是磁盘上的一个文件，不可能时时刻刻都跟 Namenode 内存中的数据结构保持同步，并且 fsimage 文件一般都很大，GB级别的很常见， 如果所有的更新操作都实时地写 fsimage 文件，则会导致Namenode运行得十分缓慢， 所以HDFS每过一段时间才更新一次fsimage文件

HDFS 将这些操作记录在editlog（编辑日志）文件中，editlog 是一个日志文件，HDFS 客户端执行的所有写操作首先会被记录到editlog文件中。HDFS 会定期地将 editlog 文件与 fsimage 文件进行合并，以保持 fsimage 跟Namenode 内存中记录的命名空间完全同步

在 HDFS 源码中，使用 FSEditLog 类来管理 editlog 文件。和 fsimage 文件不同，editlog 文件会随着 Namenode的运行实时更新，所以 FSEditLog 类的实现依赖于底层的输入流和输出流，同时 FSEditLog 类还需要对外提供大量的 log 方法用于记录命名空间的修改操作

TransactionId机制

TransactionId 与客户端每次发起的 RPC 操作相关，当客户端发起一次 RPC 请求对 Namenode 的命名空间修改后，Namenode 就会在 editlog 中发起一个新的 transaction 用于记录这次操作，每个 transaction 会用一个唯一的 transactionId 标识

• edits_start transaction id——end transaction id：edits 文件就是 editlog 文件，edits 文件中存放的是客户端执行的所有更新命名空间的操作。每个 edits 文件都包含了文件名中start trancsaction id - end transaction id之间的所有事务

  • 比如 edits_0000000000000000001-0000000000000000006，这个文件记录了 transaction id 在1和6之间的所有事务

• edits_inprogress_start transaction id：正在进行处理的 editlog。所有从 start transaction id 开始的新的修改操作都会记录在这个文件中，直到HDFS重置这个日志文件。重置操作会将 inprogress 文件关闭，并将inprogress 文件改名为正常的 editlog 文件（如上一项所示），同时还会打开一个新的 inprogress 文件， 记录正在进行的事务

  • 例如 edits_inprogress_0000000000000000478 文件，这个文件记录了所有 transaction id 大于478的新开始的事务，我们将这个事务区间称为一个日志段落（segment）
  • Namenode 元数据文件夹中存在这个文件有两种可能：要么是 Active Namenode 正在写入数据， 要么是前一个 Namenode 没有正确地关闭

• fsimage_end transaction id：fsimage 文件是 Hadoop 文件系统元数据的一个永久性的检查点，包含Hadoop文件系统中 end transaction id 前的完整的 HDFS 命名空间元数据镜像，也就是 HDFS 所有目录和文件对应的INode 的序列化信息

  • fsimage_0000000000000000473就是fsimage_0000000000000000472与edits_0000000000000000473-0000000000000000473合并后的镜像文件，保存了 transaction id 小于473的 HDFS 命名空间的元数据。 每个 fsimage 文件还有一个对应的md5文件，用来确保 fsimage 文件的正确性，以防止磁盘异常发生

• seen_txid：这个文件中保存了上一个检查点（checkpoint），即合并 edits 和 fsimage 文件，以及编辑日志重置（editlog roll）（持久化当前的 inprogress 文件并且创建一个新的 inprogress 文件）时最新的事务id （transaction id）。要特别注意的是，这个事务id并不是 Namenode 内存中最新的事务id， 因为 seen_txid只在检查点操作以及编辑日志重置操作时更新。这个文件的作用在于 Namenode 启动时， 可以利用这个文件判断是否有edits文件丢失

  • Namenode 使用不同的目录保存 fsimage 以及 edits 文件，如果保存 edits 的目录内容丢失， Namenode 将会使用上一个检查点保存的 fsimage 启动，那么上一个检查点之后的所有事务都会丢失。 为了防止发生这种状况，Namenode 启动时会检查 seen_txid 并确保内存中加载的事务 id 至少超过seen_txid；否则Namenode将终止启动操作 ```java /**
  • TransactionId与客户端每次发起的RPC操作相关，
  • 当客户端发起一次RPC请求对Namenode的命名空间修改后，
  • Namenode就会在editlog中发起一个新的transaction用于记录这次操作，

  • 每个transaction会用一个唯一的transactionId标识。 / private static class TransactionId { public long txid;

    TransactionId(long value) { this.txid = value; } } ```

构造方法

FSEditLog 是通过 newInstance 方法进行构造的，可以根据配置 dfs.namenode.edits.asynclogging 生成不同的FSEditLog 实例，默认是 FSEditLogAsync

static FSEditLog newInstance(Configuration conf, NNStorage storage,
      List<URI> editsDirs) {
    boolean asyncEditLogging = conf.getBoolean(
        DFSConfigKeys.DFS_NAMENODE_EDITS_ASYNC_LOGGING,
        DFSConfigKeys.DFS_NAMENODE_EDITS_ASYNC_LOGGING_DEFAULT);
    LOG.info("Edit logging is async:" + asyncEditLogging);
    return asyncEditLogging
        ? new FSEditLogAsync(conf, storage, editsDirs)
        : new FSEditLog(conf, storage, editsDirs);
}

FSEditLog状态机

FSEditLog 类被设计成一个状态机，用内部类 FSEditLog.State 描述

FSEditLog有以下5个状态：

• UNINITIALIZED：editlog的初始状态
• BETWEEN_LOG_SEGMENTS：editlog的前一个segment已经关闭，新的还没开始
• IN_SEGMENT：editlog处于可写状态
• OPEN_FOR_READING：editlog处于可读状态
• CLOSED：editlog处于关闭状态

private enum State {
    // editlog的初始状态。
    UNINITIALIZED,
    // editlog的前一个segment已经关闭，新的还没开始。
    BETWEEN_LOG_SEGMENTS,
    // editlog处于可写状态。
    IN_SEGMENT,
    // editlog处于可读状态。
    OPEN_FOR_READING,
    // editlog处于关闭状态。
    CLOSED;
}

对于非 HA 机制的情况：

• FSEditLog 应该开始于 UNINITIALIZED 或者 CLOSED 状态（因为在构造 FSEditLog 对象时，FSEditLog 的成员变量 state 默认为 State.UNINITIALIZED）
• FSEditLog 初始化完成之后进入BETWEEN_LOG_SEGMENTS 状态，表示前一个 segment 已经关闭，新的还没开始，日志已经做好准备了
• 当打开日志服务时，改变 FSEditLog 状态为 IN_SEGMENT 状态，表示可以写 editlog 文件了

对于 HA 机制的情况：

• FSEditLog同样应该开始于 UNINITIALIZED 或者 CLOSED 状态，但是在完成初始化后 FSEditLog 并不进入BETWEEN_LOG_SEGMENTS状态， 而是进入 OPEN_FOR_READING 状态，因为目前 Namenode 启动时都是以 Standby 模式启动的，然后通过 DFSHAAdmin 发送命令把其中一个 Standby NameNode 转换成 Active Namenode

InitJournalsForWrite()

IniJournalsForWrite() 方法是 FSEditLog 的 public 方法，调用这个方法会将 FSEditLog 从 UNINITIALIZED 状态转换为 BETWEEN_LOG_SEGMENTS 状态

public synchronized void initJournalsForWrite() {
    Preconditions.checkState(state == State.UNINITIALIZED ||
        state == State.CLOSED, "Unexpected state: %s", state);
    // 调用initJournals()方法
    // initJournals()方法会根据传入的 dirs 变量
    // (保存的是 editlog 文件的存储位置，都是URI)
    // 初始化journalSet字段 (JournalManager对象的集合)。
    // 初始化之后，FSEditLog就可以调用journalSet对象的方法向多个日志存储位置写editlog文件了。
    initJournals(this.editsDirs);
    //状态转换为BETWEEN_LOG_SEGMENTS
    state = State.BETWEEN_LOG_SEGMENTS;
}

JournalManager 类是负责在特定存储目录上持久化 editlog 文件的类，它的 format() 方法负责格式化底层存储，startLogSegment() 方法负责从指定事务 id 开始记录一个操作的段落，finalizeLogSegment() 方法负责完成指定事务id区间的写操作

这里之所以抽象这个接口，是因为 Namenode 可能将 editlog 文件持久化到不同类型的存储上，也就需要不同类型的 JournalManager 来管理，所以需要定义一个抽象的接口。JoumalManager 有多个子类，普通的文件系统由 FileJournalManager 类管理，共享 NFS 由 BackupJournalManager 类管理、Bookkeeper由 BookkeeperJournalManager 类管理、Quorum集群则由 QuorumJournalManager 类管理

/**
   * dirs  editsDirs
   * @param dirs
   */
private synchronized void initJournals(List<URI> dirs) {
    // dfs.namenode.edits.dir.minimum 默认值: 1
    int minimumRedundantJournals = conf.getInt(
        DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_KEY,
        DFSConfigKeys.DFS_NAMENODE_EDITS_DIR_MINIMUM_DEFAULT);
    synchronized(journalSetLock) {
        // 初始化journalSet集合，存放存储路径对应的所有JournalManager对象
        journalSet = new JournalSet(minimumRedundantJournals);
        // 根据传入的URI获取对应的JournalManager对象
        for (URI u : dirs) {
            boolean required = FSNamesystem.getRequiredNamespaceEditsDirs(conf)
              .contains(u);
            if (u.getScheme().equals(NNStorage.LOCAL_URI_SCHEME)) {
                StorageDirectory sd = storage.getStorageDirectory(u);
                if (sd != null) {
                // 本地URI，则加入FileJournalManager即可
                journalSet.add(new FileJournalManager(conf, sd, storage),
                    required, sharedEditsDirs.contains(u)); 
                }
            } else {
                // 否则根椐URI创建对应的JournalManager对象，并放入journalSet中保存
                journalSet.add(createJournal(u), required,
                               sharedEditsDirs.contains(u));
            }
        }
    }
    if (journalSet.isEmpty()) {
        LOG.error("No edits directories configured!");
    } 
}

InitSharedJournalsForRead()

InitSharedJournalsForRead() 方法是 FSEditLog 的 public 方法，用在 HA 情况下。调用这个方法会将 FSEditLog 从 UNINITIALIZED 状态转换为 OPEN_FOR_READING 状态

与 initJournalsForWrite() 方法相同，initSharedJournalsForRead() 方法也调用了 initJournals() 方法执行初始化操作，只不过 editlog 文件的存储位置不同，在 HA 的情况下，editlog 文件的存储目录为共享存储目录，这个共享存储目录由 Active Namenode 和 StandbyNamenode 共享读取

public synchronized void initSharedJournalsForRead() {
    if (state == State.OPEN_FOR_READING) {
        LOG.warn("Initializing shared journals for READ, already open for READ",
            new Exception());
        return;
    }
    Preconditions.checkState(state == State.UNINITIALIZED ||
        state == State.CLOSED);
    // 对于HA的情况，editlog的日志存储目录为共享的目录sharedEditsDirs
    initJournals(this.sharedEditsDirs);
    state = State.OPEN_FOR_READING;
}

OpenForWrite()

OpenForWrite() 方法用于初始化 editlog 文件的输出流，并且打开第一个日志段落（log segment）。 在非 HA 机制下，调用这个方法会完成 BETWEEN_LOG_SEGMENTS 状态到 IN_SEGMENT 状态的转换

synchronized void openForWrite(int layoutVersion) throws IOException {
    Preconditions.checkState(state == State.BETWEEN_LOG_SEGMENTS,
        "Bad state: %s", state);
    // 返回最后一个写入log的transactionId+1，作为本次操作的transactionId , 
    // 假设当前的transactionId为31
    long segmentTxId = getLastWrittenTxId() + 1;
    // Safety check: we should never start a segment if there are
    // newer txids readable.
    List<EditLogInputStream> streams = new ArrayList<EditLogInputStream>();
    // 传入了参数segmentTxId，
    // 这个参数会作为这次操作的transactionId，
    // 值为editlog已经记录的最新的transactionId加1(这里是 31+1=32)。
    //
    // selectInputStreams()方法会判断有没有一个以segmentTxId(32)开始的日志，
    // 如果没有则表示当前transactionId的值选择正确，可以打开新的editlog文件记录
    // 以segmentTxId开始的日志段落。 
    // 如果方法找到了包含这个transactionId的editlog文件，
    // 则表示出现了两个日志 transactionId交叉的情况，抛出异常。
    journalSet.selectInputStreams(streams, segmentTxId, true, false);
    // 这里判断，有没有包含这个新的segmentTxId的editlog文件，如果有则抛出异常
    if (!streams.isEmpty()) {
        String error = String.format("Cannot start writing at txid %s " +
        "when there is a stream available for read: %s",
          segmentTxId, streams.get(0));
        IOUtils.cleanupWithLogger(LOG,
            streams.toArray(new EditLogInputStream[0]));
        throw new IllegalStateException(error);
    }
    //写入日志
    startLogSegmentAndWriteHeaderTxn(segmentTxId, layoutVersion);
    assert state == State.IN_SEGMENT : "Bad state: " + state;
}

在所有 editlog 文件的存储路径上构造输出流，并将这些输出流保存在 FSEditLog 的字段 journalSet.journals 中

/**
 *
 * Start writing to the log segment with the given txid.
 * Transitions from BETWEEN_LOG_SEGMENTS state to IN_LOG_SEGMENT state. 
*/
private void startLogSegment(final long segmentTxId, int layoutVersion)
    throws IOException {
    assert Thread.holdsLock(this);
    LOG.info("Starting log segment at " + segmentTxId);
    Preconditions.checkArgument(segmentTxId > 0,
        "Bad txid: %s", segmentTxId);
    Preconditions.checkState(state == State.BETWEEN_LOG_SEGMENTS,
        "Bad state: %s", state);
    Preconditions.checkState(segmentTxId > curSegmentTxId,
        "Cannot start writing to log segment " + segmentTxId +
        " when previous log segment started at " + curSegmentTxId);
    Preconditions.checkArgument(segmentTxId == txid + 1,
        "Cannot start log segment at txid %s when next expected " +
        "txid is %s", segmentTxId, txid + 1);
    numTransactions = 0;
    totalTimeTransactions = 0;
    numTransactionsBatchedInSync.set(0L);
    // TODO no need to link this back to storage anymore!
    // See HDFS-2174.
    storage.attemptRestoreRemovedStorage();
    try {
      //初始化editLogStream
      editLogStream = journalSet.startLogSegment(segmentTxId, layoutVersion);
    } catch (IOException ex) {
      throw new IOException("Unable to start log segment " +
          segmentTxId + ": too few journals successfully started.", ex);
    }
    //当前正在写入txid设置为segmentTxId
    curSegmentTxId = segmentTxId;
    state = State.IN_SEGMENT;
}

EndCurrentLogSegment()

EndCurrentLogSegment() 会将当前正在写入的日志段落关闭，它调用了 journalSet.finalizeLogSegment() 方法将curSegmentTxid -> lastTxId 之间的操作持久化到磁盘上

持久化是将程序数据在持久状态和瞬时状态间转换的机制。通俗的讲，就是瞬时数据（比如内存中的数据，是不能永久保存的）持久化为持久数据（比如持久化至数据库中，能够长久保存）

这个方法会将 FSEditLog 状态机更改为 BETWEEN_LOG_SEGMENTS 状态

/**
  *
  * Finalize the current log segment.
  * Transitions from IN_SEGMENT state to BETWEEN_LOG_SEGMENTS state.
  */
public synchronized void endCurrentLogSegment(boolean writeEndTxn) {
    LOG.info("Ending log segment " + curSegmentTxId +
        ", " + getLastWrittenTxId());
    Preconditions.checkState(isSegmentOpen(),
        "Bad state: %s", state);
    if (writeEndTxn) {
      logEdit(LogSegmentOp.getInstance(cache.get(), 
          FSEditLogOpCodes.OP_END_LOG_SEGMENT));
    }
    // always sync to ensure all edits are flushed.
    logSyncAll();
    printStatistics(true);
    final long lastTxId = getLastWrittenTxId();
    // 获取当前写入的最后一个id
    final long lastSyncedTxId = getSyncTxId();
    Preconditions.checkArgument(lastTxId == lastSyncedTxId,
        "LastWrittenTxId %s is expected to be the same as lastSyncedTxId %s",
        lastTxId, lastSyncedTxId);
    try {
      // 调用journalSet.finalizeLogSegment将curSegmentTxid -> lastTxId之间的操作
      // 写入磁盘(例如editlog文件edits_0032-0034)
      journalSet.finalizeLogSegment(curSegmentTxId, lastTxId);
      editLogStream = null;
    } catch (IOException e) {
      // All journals have failed, it will be handled in logSync.
    }
    // 更改状态机的状态
    state = State.BETWEEN_LOG_SEGMENTS;
}

journalSet.finalizeLogSegment() 方法也会调用 mapJournalsAndReportErrors() 方法将 finalizeLogSegment() 调用前转到 journals 集合中保存的所有的 JournalManager 对象上。比如 FileJournalManager, FileJoumalManager.finalizeLogSegment() 方法会将 edit_inprogress 文件改名为 edit 文件，新生成的 edit 文件覆盖了 curSegmentTxid -> lastTxId 之间的所有事务

@Override
synchronized public void finalizeLogSegment(long firstTxId, long lastTxId)  throws IOException {
    // 原有的inprogress文件
    File inprogressFile = NNStorage.getInProgressEditsFile(sd, firstTxId);
    // 构造新的edit文件
    File dstFile = NNStorage.getFinalizedEditsFile( sd, firstTxId, lastTxId);
    LOG.info("Finalizing edits file " + inprogressFile + " -> " + dstFile);
    Preconditions.checkState(!dstFile.exists(),
        "Can't finalize edits file " + inprogressFile + " since finalized file " +
        "already exists");
    try {
        //执行重命名操作
        NativeIO.renameTo(inprogressFile, dstFile);
    } catch (IOException e) {
        errorReporter.reportErrorOnFile(dstFile);
        throw new IllegalStateException("Unable to finalize edits file " + inprogressFile, e);
    }
    if (inprogressFile.equals(currentInProgress)) {
        currentInProgress = null;
    }
}

close()

close() 方法用于关闭 editlog 文件的存储，完成了 IN_SEGMENT 到 CLOSED 状态的改变。close()会首先等待 sync操作完成，然后调用 endCurrentLogSegment() 方法，将当前正在进行写操作的日志段落结束。之后 close() 方法会关闭 journalSet 对象，并将 FSEditLog 状态机转变为 CLOSED 状态

synchronized void close() {
    if (state == State.CLOSED) {
        LOG.debug("Closing log when already closed");
        return;
    }
    try {
        if (state == State.IN_SEGMENT) {
          assert editLogStream != null;
          //如果有sync操作， 则等待sync操作完成
          waitForSyncToFinish();
          //结束当前logSegment
          endCurrentLogSegment(true);
        }
    } finally {
        //关闭journalSet
        if (journalSet != null && !journalSet.isEmpty()) {
            try {
                synchronized(journalSetLock) {
                journalSet.close();
            }
        } catch (IOException ioe) {
            LOG.warn("Error closing journalSet", ioe);
            }
        }
      //将状态机更改为CLOSED状态
      state = State.CLOSED;
    }
}

EditLogOutputStream

FSEditLog 类会调用 FSEditLog.editLogStream 字段的 write() 方法在 editlog 文件中记录一个操作，数据会先被写入到 editlog 文件输出流的缓存中，然后 FSEditLog 类会调用 editLogStream.flush() 方法将缓存中的数据同步到磁盘上

FSEditLog 的 editLogStream 字段是 EditLogOutputStream 类型的，EditLogOutputStream类是一个抽象类，它定义了向持久化存储上写 editlog 文件的相关接口

EditLogOutputStream定义了多个子类来向不同存储系统上的 editlog 文件中写入数据

JournalSetOutputStream

JournalSetOutputStream 类是 EditLogOutputStream 的子类，在 JournalSetOutputStream 对象上调用的所有EditLogOutputStream 接口方法都会被前转到 FSEditLog.journalSet 字段中保存的 editlog 文件在所有存储位置上的输出流对象（通过调用 mapJournalsAndReportErrors() 方法实现）

FSEditLog 的 editLogStream 字段就是 JournalSetOutputStream 类型的（是在 startLogSegment() 方法中赋值的），通过调用 JournalSetOutputStream 对象提供的方法，FSEditLog 可以将 Namenode 多个存储位置上的editlog 文件输出流对外封装成一个输出流，大大方便了调用

JournalSetOutputStream 类是通过 mapJournalsAndReportErrors() 方法，将 EditLogOutputStream 接口上的write() 调用前转到了 FSEditLog 中保存的所有存储路径上 editlog 文件对应的 EditLogOutputStream 输出流对象上的。这个方法会遍历 FSEditLog.journalSet.journals 集合，然后将 write() 请求前转到 journals 集合中保存的所有 JournalAndStream 对象上

journalSet 的 journals 字段是一个 JournalAndStream 对象的集合，JournalAndStream 对象封装了一个JournalManager 对象，以及在这个 JournalManager 上打开的 editlog 文件的 EditLogOutputStream 对象

journalSet.journals 字段是在 FSEditLog.startLogSegment() 方法中赋值的 ，这个方法调用了journalSet.startLogSegment() 方法在所有 editlog 文件的存储路径上构造输出流，并将这些输出流保存在FSEditLog 的 journalSet.journals 字段中

/**
   * Apply the given operation across all of the journal managers, disabling
   * any for which the closure throws an IOException.
   * @param closure {@link JournalClosure} object encapsulating the operation.
   * @param status message used for logging errors (e.g. "opening journal")
   * @throws IOException If the operation fails on all the journals.
   */
private void mapJournalsAndReportErrors(
      JournalClosure closure, String status) throws IOException{
    List<JournalAndStream> badJAS = Lists.newLinkedList();
    //遍历journals字段中保存的所有JournalAndStream对象
    for (JournalAndStream jas : journals) {
        try {
            //在闭包对象上调用apply()方法前转请求
            closure.apply(jas);
        } catch (Throwable t) {
            if (jas.isRequired()) {
                final String msg = "Error: " + status + " failed for required journal ("
            + jas + ")";
                LOG.error(msg, t);
                abortAllJournals();
                terminate(1, msg);
            } else {
                LOG.error("Error: " + status + " failed for (journal " + jas + ")", t);
                badJAS.add(jas);          
            }
        }
    }
    disableAndReportErrorOnJournals(badJAS);
    if (!NameNodeResourcePolicy.areResourcesAvailable(journals,
        minimumRedundantJournals)) {
        String message = status + " failed for too many journals";
        LOG.error("Error: " + message);
        throw new IOException(message);
    }
}

mapJournalsAndReportErrors() 方法在调用时传入了一个闭包对象 closure，这个对象是在JournalSetOutputStream 实现的 EditLogOutputStream 接口方法上定义的。以 JournalSetOutputStream.write() 方法为例，write() 方法定义了写操作的闭包对象，这个闭包对象会提取出 JournalAndStream 对象中封装的EditLogOutputStream 对象，然后调用这个对象上的 write() 方法来完成写数据的功能。通过这种闭包机制， JournalSetOutputStream 完成了将 EditLogOutputStream 接口上的 write() 调用前转到 JournalAndStream 保存的EditLogOutputStream 对象上的操作

写入方法

@Override
public void write(final FSEditLogOp op)
    throws IOException {
    mapJournalsAndReportErrors(new JournalClosure() {
      public void apply(JournalAndStream jas) throws IOException {
          if (jas.isActive()) {
          // 提取出JournalAndStream对象中封装的EditLogOutputStream对象，
          // 并在EditLogOutputStream对象上调用write()方法
          jas.getCurrentStream().write(op);
          }
      }
    }, "write op");
}

EditLogFileOutputStream

EditLogFileOutputStream 是向本地文件系统中保存的 editlog 文件写数据的输出流，向 EditLogFileOutputStream写数据时，数据首先被写入到输出流的缓冲区中，当显式地调用 flush() 操作后，数据才会从缓冲区同步到editlog文件中

构造方法

  /**
   * Creates output buffers and file object.
   * 
   * @param conf
   *          Configuration object
   * @param name
   *          File name to store edit log
   * @param size
   *          Size of flush buffer
   * @throws IOException
   */
public EditLogFileOutputStream(Configuration conf, File name, int size)
      throws IOException {
    super();
    shouldSyncWritesAndSkipFsync = conf.getBoolean(
            DFSConfigKeys.DFS_NAMENODE_EDITS_NOEDITLOGCHANNELFLUSH,
            DFSConfigKeys.DFS_NAMENODE_EDITS_NOEDITLOGCHANNELFLUSH_DEFAULT);
    file = name;
    doubleBuf = new EditsDoubleBuffer(size);
    RandomAccessFile rp;
    if (shouldSyncWritesAndSkipFsync) {
        rp = new RandomAccessFile(name, "rws");
    } else {
        rp = new RandomAccessFile(name, "rw");
    }
    fp = new FileOutputStream(rp.getFD()); // open for append
    fc = rp.getChannel();
    fc.position(fc.size());
}

常量

  public static final int MIN_PREALLOCATION_LENGTH = 1024 * 1024;
  // 输出流对应的editlog文件
  private File file;
  // editlog文件对应的输出流
  private FileOutputStream fp; // file stream for storing edit logs
  // editlog文件对应的输出流通道。
  private FileChannel fc; // channel of the file stream for sync
  // 一个具有两块缓存的缓冲区， 数据必须先写入缓存，然后再由缓存同步到磁盘上
  private EditsDoubleBuffer doubleBuf;
  // 用来扩充editlog文件大小的数据块
  // 当要进行同步操作时，如果editlog文件不够大，则使用fill来扩充editlog
  // 文件最小1M
  static final ByteBuffer fill = ByteBuffer.allocateDirect(MIN_PREALLOCATION_LENGTH);
  private boolean shouldSyncWritesAndSkipFsync = false;
  private static boolean shouldSkipFsyncForTests = false;
  // EditLogFileOutputStream有一个static的代码段
  // 将fill字段用FSEditLogOpCodes.OP_INVALID 字节填满
  static {
      fill.position(0);
      for (int i = 0; i < fill.capacity(); i++) {
          fill.put(FSEditLogOpCodes.OP_INVALID.getOpCode());
      }
  }

在创建 edits_inprogress_0000000000000000485 文件的时候，首先会用 -1 填充 1M 大小的文件空间，然后将写入的指针归0。当有数据的时候进行写入，写入的时候，会覆盖之前预制填充的数据。但不管怎么样，如果数据大小不满1M的话，那么edits文件的大小最小为1M

每次重启 namenode 的时候都会将之前的 editsinprogress 文件关闭，并重命名为 edits** 文件, 创建一个新的 edits_inprogress_0000000000000000485 文件

write()、 setReadyToFlush()

// 直接调用doubleBuf中的对应方法，向输出流写入一个操作
@Override
public void write(FSEditLogOp op) throws IOException {
    // 向doubleBuf写入FSEditLogOp对象
    doubleBuf.writeOp(op, getCurrentLogVersion());
}
/**
*
* 为同步数据做准备
* 调用doubleBuf.setReadyToFlush()交换两个缓冲区
*
* All data that has been written to the stream so far will be flushed. New
* data can be still written to the stream while flushing is performed.
*/
@Override
public void setReadyToFlush() throws IOException {
    doubleBuf.setReadyToFlush();
}

flushAndSync() 方法则用于将输出流中缓存的数据同步到磁盘上的 editlog 文件中

/**
*
* 将准备好的缓冲区刷新到持久性存储
* 由于会刷新和同步readyBuffer，因此currentBuffer不会累积新的日志记录，因此不会刷新
*
* Flush ready buffer to persistent store. currentBuffer is not flushed as it
* accumulates new log records while readyBuffer will be flushed and synced.
*/
@Override
public void flushAndSync(boolean durable) throws IOException {
    // fp: editlog文件对应的输出流
    if (fp == null) {
        throw new IOException("Trying to use aborted output stream");
    }
    if (doubleBuf.isFlushed()) {
        LOG.info("Nothing to flush");
        return;
    }
    // preallocate()方法用于在 editLog 文件大小不够时，填充editlog文件
    preallocate(); // preallocate file if necessary
    //将缓存中的数据同步到editlog文件中
    doubleBuf.flushTo(fp);
    if (durable && !shouldSkipFsyncForTests && !shouldSyncWritesAndSkipFsync) {
        fc.force(false); // metadata updates not needed
    }
}

EditsDoubleBuffer类

EditsDoubleBuffer 中包括两块缓存，数据会先被写入到 EditsDoubleBuffer 的一块缓存中，而 EditsDoubleBuffer的另一块缓存可能正在进行磁盘的同步操作（就是将缓存中的文件写入磁盘的操作）

EditsDoubleBuffer 这样的设计会保证输出流进行磁盘同步操作的同时，并不影响数据写入的功能

// 正在写入的缓冲区
private TxnBuffer bufCurrent; // current buffer for writing
// 准备好同步的缓冲区
private TxnBuffer bufReady; // buffer ready for flushing
// 缓冲区的大小  默认 512K
private final int initBufferSize;

输出流要进行同步操作时，首先要调用 EditsDoubleBuffer.setReadyToFlush() 方法交换两个缓冲区，将正在写入的缓存改变为同步缓存，然后才可以进行同步操作

  // 将正在写入的缓存改变为同步缓存， 然后才可以进行同步操作。
public void setReadyToFlush() {
    assert isFlushed() : "previous data not flushed yet";
    //交换两个缓冲区
    TxnBuffer tmp = bufReady;
    bufReady = bufCurrent;
    bufCurrent = tmp;
}

完成了 setReadyToFlush() 调用之后，输出流就可以调用 flushTo() 方法将同步缓存中的数据写入到文件中

  /**
   * Writes the content of the "ready" buffer to the given output stream,
   * and resets it. Does not swap any buffers.
   *
   */
public void flushTo(OutputStream out) throws IOException {
    // 将同步缓存中的数据写入文件
    bufReady.writeTo(out); // write data to file
    // 将同步缓存中保存的数据清空
    bufReady.reset(); // erase all data in the buffer
}

EditLogFileInputStream

EditLogFileInputStream 类抽象了从持久化存储上读 editlog 文件的相关接口

EditLogFileInputStream 定义了本地文件系统的 editlog 文件的输入流。它定义的方法都很简单，都是返回了EditLogFileInputStream 初始化以后的相应字段，或者调用了 FSEditLogOp.Reader 对象的 readOp() 方法从editlog 文件中解析出一个 FSEditLogOp 对象

构造方法

private EditLogFileInputStream(LogSource log,
    long firstTxId, long lastTxId,
    boolean isInProgress) {
    this.log = log;
    this.firstTxId = firstTxId;
    this.lastTxId = lastTxId;
    this.isInProgress = isInProgress;
    // 最大值 50 * 1024 * 1024  ==> 50M  ???????
    this.maxOpSize = DFSConfigKeys.DFS_NAMENODE_MAX_OP_SIZE_DEFAULT;
}

属性

private final LogSource log;
private final long firstTxId;
private final long lastTxId;
private final boolean isInProgress;
private int maxOpSize;
static private enum State {
    UNINIT,
    OPEN,
    CLOSED
}
private State state = State.UNINIT;
private InputStream fStream = null;
private int logVersion = 0;
private FSEditLogOp.Reader reader = null;
private FSEditLogLoader.PositionTrackingInputStream tracker = null;
private DataInputStream dataIn = null;
static final Logger LOG = LoggerFactory.getLogger(EditLogInputStream.class);

init

private void init(boolean verifyLayoutVersion)
      throws LogHeaderCorruptException, IOException {
    Preconditions.checkState(state == State.UNINIT);
    BufferedInputStream bin = null;
    try {
        fStream = log.getInputStream();
        bin = new BufferedInputStream(fStream);
        tracker = new FSEditLogLoader.PositionTrackingInputStream(bin);
        dataIn = new DataInputStream(tracker);
      try {
          logVersion = readLogVersion(dataIn, verifyLayoutVersion);
      } catch (EOFException eofe) {
          throw new LogHeaderCorruptException("No header found in log");
      }
      if (logVersion == -1) {
        // The edits in progress file is pre-allocated with 1MB of "-1" bytes
        // when it is created, then the header is written. If the header is
        // -1, it indicates the an exception occurred pre-allocating the file
        // and the header was never written. Therefore this is effectively a
        // corrupt and empty log.
        throw new LogHeaderCorruptException("No header present in log (value " +
            "is -1), probably due to disk space issues when it was created. " +
            "The log has no transactions and will be sidelined.");
      }
      // We assume future layout will also support ADD_LAYOUT_FLAGS
      if (NameNodeLayoutVersion.supports(
          LayoutVersion.Feature.ADD_LAYOUT_FLAGS, logVersion) ||
          logVersion < NameNodeLayoutVersion.CURRENT_LAYOUT_VERSION) {
        try {
          LayoutFlags.read(dataIn);
        } catch (EOFException eofe) {
          throw new LogHeaderCorruptException("EOF while reading layout " +
              "flags from log");
        }
      }
      reader = FSEditLogOp.Reader.create(dataIn, tracker, logVersion);
      reader.setMaxOpSize(maxOpSize);
      state = State.OPEN;
    } finally {
      if (reader == null) {
          IOUtils.cleanupWithLogger(LOG, dataIn, tracker, bin, fStream);
          state = State.CLOSED;
      }
    }
}

ReadOp

 /** 
   * Read an operation from the stream
   * @return an operation from the stream or null if at end of stream
   * @throws IOException if there is an error reading from the stream
   */
public FSEditLogOp readOp() throws IOException {
    FSEditLogOp ret;
    if (cachedOp != null) {
      ret = cachedOp;
      cachedOp = null;
      return ret;
    }
    return nextOp();
}

NextOpImpl

// skipBrokenEdits 是否跳过阻塞的edits . 
private FSEditLogOp nextOpImpl(boolean skipBrokenEdits) throws IOException {
    FSEditLogOp op = null;
    switch (state) {
    case UNINIT:
     try {
        // 执行初始化操作
        init(true);
     } catch (Throwable e) {
          LOG.error("caught exception initializing " + this, e);
          if (skipBrokenEdits) {
            return null;
          }
          Throwables.propagateIfPossible(e, IOException.class);
      }
      Preconditions.checkState(state != State.UNINIT);
      return nextOpImpl(skipBrokenEdits);
    case OPEN:
      // 读取 FSEditLogOp 类型操作 
      op = reader.readOp(skipBrokenEdits);
      if ((op != null) && (op.hasTransactionId())) {
        long txId = op.getTransactionId();
        if ((txId >= lastTxId) &&
            (lastTxId != HdfsServerConstants.INVALID_TXID)) {
          //
          // Sometimes, the NameNode crashes while it's writing to the
          // edit log.  In that case, you can end up with an unfinalized edit log
          // which has some garbage at the end.
          // JournalManager#recoverUnfinalizedSegments will finalize these
          // unfinished edit logs, giving them a defined final transaction 
          // ID.  Then they will be renamed, so that any subsequent
          // readers will have this information.
          //
          // Since there may be garbage at the end of these "cleaned up"
          // logs, we want to be sure to skip it here if we've read everything
          // we were supposed to read out of the stream.
          // So we force an EOF on all subsequent reads.
          //
          long skipAmt = log.length() - tracker.getPos();
          if (skipAmt > 0) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("skipping " + skipAmt + " bytes at the end " +
                  "of edit log  '" + getName() + "': reached txid " + txId +
                  " out of " + lastTxId);
            }
            tracker.clearLimit();
            IOUtils.skipFully(tracker, skipAmt);
          }
        }
      }
      break;
      case CLOSED:
        break; // return null
    }
    return op;
}

FSEditLog

FSEditLog 类最重要的作用就是在 editlog 文件中记录 Namenode 命名空间的更改，FSEditLog 类对外提供了若干方法用于执行这个操作

LogDelete()

logDelete() 方法用于在 editlog 文件中记录删除 HDFS 文件的操作

logDelete() 方法首先会构造一个 DeleteOp 对象，这个 DeleteOp 类是 FSEditLogOp 类的子类，用于记录删除操作的相关信息，包括了 ClientProtocol.delete() 调用中所有参数携带的信息。构造 DeleteOp 对象后， logDelete()方法会调用 logRpcIds() 方法在 DeleteOp 对象中添加 RPC 调用相关信息，之后 logDelete() 方法会调用 logEdit()方法在 editlog 文件中记录这次删除操作

/** 
* Add delete file record to edit log
*/
void logDelete(String src, long timestamp, boolean toLogRpcIds) {
    // 构造DeleteOp对象
    DeleteOp op = DeleteOp.getInstance(cache.get())
      .setPath(src)
      .setTimestamp(timestamp);
    // 记录RPC调用相关信息
    logRpcIds(op, toLogRpcIds);
    // 调用logEdit()方法记录删除操作
    logEdit(op);
}

logEdit()

基本上所有的 log*( )方法在底层都调用了 logEdit() 方法来执行记录操作，这里会传入一个 FSEditLogOp 对象来标识当前需要被记录的操作类型以及操作的信息

/**
* Write an operation to the edit log.
* <p/>
* Additionally, this will sync the edit log if required by the underlying
* edit stream's automatic sync policy (e.g. when the buffer is full, or
* if a time interval has elapsed).
*/
void logEdit(final FSEditLogOp op) {
    boolean needsSync = false;
    synchronized (this) {
      assert isOpenForWrite() :
        "bad state: " + state;
      // 如果自动同步开启，则等待同、 步完成
      waitIfAutoSyncScheduled();
      // check if it is time to schedule an automatic sync
      needsSync = doEditTransaction(op);
      if (needsSync) {
        isAutoSyncScheduled = true;
      }
    }
    // Sync the log if an automatic sync is required.
    if (needsSync) {
      logSync();
    }
}

// 同步操作, 即使是多个线程, 依旧会进行同步操作. txid 不会错乱
// 保证了多个线程调用FSEditLog.log*()方法向editlog文件中写数据时，editlog文件记录的内容不会相互影响。
// 也保证了这几个并发线程保存操作对应的transactionId；通过调用beginTransaction()方法获得
synchronized boolean doEditTransaction(final FSEditLogOp op) {
    //开启一个新的transaction , 更新 txid
    long start = beginTransaction();
    op.setTransactionId(txid);
    try {
      // 使用editLogStream写入Op操作
      editLogStream.write(op);
    } catch (IOException ex) {
      // All journals failed, it is handled in logSync.
    } finally {
      op.reset();
    }
    //结束当前的transaction
    endTransaction(start);
    //检查是否需要强制同步
    return shouldForceSync();
  }

logEdit() 方法会调用 beginTransaction() 方法在 editlog 文件中开启一个新的 transaction，然后使用 editlog 输入流写入要被记录的操作，接下来调用 endTransaction() 方法关闭这个transaction，最后调用 logSync() 方法将写入的信息同步到磁盘上

logEdit() 方法调用 beginTransaction()、editLogStream.write() 以及 endTransaction() 三个方法时使用了synchronized 关键字进行同步操作，这样就保证了多个线程调用 FSEditLog.log*() 方法向 editlog 文件中写数据时，editlog 文件记录的内容不会相互影响。 同时， 也保证了这几个并发线程保存操作对应的 transactionId（通过调用 beginTransaction() 方法获得）是唯一并递增的

logEdit() 方法中调用 logSync() 方法执行刷新操作的语句并不在 synchronized 代码段中。这是因为调用 logSync()方法必然会触发写editlog文件的磁盘操作，这是一个非常耗时的操作，如果放入同步模块中会造成其他调用FSEditLog.log() 线程的等待时间过长。所以，HDFS 设计者将需要进行同步操作的 synchronized 代码段放入logSync() 方法中，也就让输出日志记录和刷新缓冲区数据到磁盘这两个操作分离了。同时，利用EditLogOutputStream 的两个缓冲区，使得日志记录和刷新缓冲区数据这两个操作可以并发执行，*大大地提高了Namenode的吞吐量

BeginTransaction()

logEdit() 方法会调用 beginTransaction() 方法开启一个新的 transaction，也就是将 FSEditLog.txid 字段增加 1 并作为当前操作的 transactionId

FSEditLog.txid 字段维护了一个全局递增的 transactionId，这样也就保证了 FSEditLog 为所有操作分配的transactionId 是唯一且递增的。调用 beginTransaction() 方法之后会将新申请的 transactionId 放入 ThreadLocal的变量 my TransactionId 中， myTransactionId 保存了当前线程记录操作对应的 transactionId，方便了以后线程做sync同步操作

对于 FSEditLog 类，可能同时有多个线程并发地调用 log() 方法执行日志记录操作，所以 FSEditLog 类使用了一个 ThreadLocal 变量 myTransactionId 为每个调用log()操作的线程保存独立的 txid，这个 txid 为当前线程记录操作对应的transactionId

private long beginTransaction() {
    assert Thread.holdsLock(this);
    // get a new transactionId
    // 全局的transactionId ++
    txid++;
    //
    // 使用ThreadLocal变量保存当前线程持有的transactionId
    // record the transactionId when new data was written to the edits log
    //
    TransactionId id = myTransactionId.get();
    id.txid = txid;
    return monotonicNow();
}

EndTransaction()

logEdit() 方法会调用 endTransaction() 方法结束一个 transaction，这个方法就是更改一些统计数据

private void endTransaction(long start) {
    assert Thread.holdsLock(this);
    // update statistics
    long end = monotonicNow();
    numTransactions++;
    totalTimeTransactions += (end-start);
    if (metrics != null) // Metrics is non-null only when used inside name node
      metrics.addTransaction(end-start);
}

LogSync()

logEdit() 方法通过调用 beginTransaction() 方法成功地获取一个 transactionId 之后，就会通过输出流向 editlog 文件写数据以记录当前的操作，但是写入的这些数据并没有直接保存在 editlog 文件中， 而是暂存在输出流的缓冲区中。所以当 logEdit() 方法将一个完整的操作写入输出流后，需要调用 logSync() 方法同步当前线程对editlog文件所做的修改

editlog同步策略:

• 所有的操作项同步的写入缓存时，每个操作会被赋予一个唯一的 transactionId
• 当一个线程要将它的操作同步到 editlog 文件中时，logSync() 方法会使用 ThreadLocal 变量 myTransactionId获取该线程需要同步的 transactionId，然后对比这个 transactionId 和已经同步到 editlog 文件中的transactionId。如果当前线程的 transactionId 大于 editlog 文件中的 transactionId，则表明 editlog 文件中记录的数据不是最新的，同时如果当前没有别的线程执行同步操作，则开始同步操作将输出流缓存中的数据写入 editlog 文件中。需要注意的是，由于 editlog 输出流使用了双 buffer 的结构，所以在进行 sync 操作的同时，并不影响 editlog 输出流的使用
• 在 logSync() 方法中使用 isSyncRunning 变量标识当前是否有线程正在进行同步操作， 这里注意isSyncRunning 是一个 volatile 的 boolean 类型变量

logSync()方法分为以下三个部分，并分开进行加锁操作，这样的设计提高了并发的程度：

• 判断当前操作是否已经同步到了 editlog 文件中，如果还没有同步，则将 editlog 的双 buffer 调换位置，为同步操作做准备，同时将 isSyncRunning 标志位设置为 true，这部分代码需要进行 synchronized 加锁操作
• 调用 logStream.flush() 方法将缓存的数据持久化到存储上，这部分代码不需要进行加锁操作，因为在上一段同步代码中已经将双buffer调换了位置，不会有线程向用于刷新数据的缓冲区中写入数据，所以调用 flush() 操作并不需要加锁

• 重置 isSyncRunning 标志位，并且通知等待的线程，这部分代码需要进行 synchronized 加锁操作

  protected void logSync(long mytxid) {
    long syncStart = 0;
    boolean sync = false;
    long editsBatchedInSync = 0;
    try {
      EditLogOutputStream logStream = null;
      synchronized (this) {
        try {
          // 第一部分，打印统计信息
          printStatistics(false);
          // 当前txid大于editlog中已经同步的txid，
          // 并且有线程正在同步， 则等待.
          // if somebody is already syncing, then wait
          while (mytxid > synctxid && isSyncRunning) {
            try {
              wait(1000);
            } catch (InterruptedException ie) {
            }
          }
          //
          // 如果txid小于editlog中已经同步的txid，则表明当前操作已经被同步到存储上， 
          // 不需要再次同步
          //
          if (mytxid <= synctxid) {
            return;
          }
          // 开始同步操作，将isSyncRunning标志位设置为true
          // now, this thread will do the sync.  track if other edits were
          // included in the sync - ie. batched.  if this is the only edit
          // synced then the batched count is 0
          editsBatchedInSync = txid - synctxid - 1;
          syncStart = txid;
          isSyncRunning = true;
          sync = true;
          // swap buffers
          try {
            if (journalSet.isEmpty()) {
              throw new IOException("No journals available to flush");
            }
            // 通过调用 setReadyToFlush() 方法将两个缓冲区互换，为同步做准备
            editLogStream.setReadyToFlush();
          } catch (IOException e) {
            final String msg =
                "Could not sync enough journals to persistent storage " +
                "due to " + e.getMessage() + ". " +
                "Unsynced transactions: " + (txid - synctxid);
            LOG.error(msg, new Exception());
            synchronized(journalSetLock) {
              IOUtils.cleanupWithLogger(LOG, journalSet);
            }
            terminate(1, msg);
          }
        } finally {
          // 防止其他log edit 写入阻塞, 引起的RuntimeException
          // Prevent RuntimeException from blocking other log edit write 
          doneWithAutoSyncScheduling();
        }
        //editLogStream may become null,
        //so store a local variable for flush.
        logStream = editLogStream;
      }
      // 第二部分，调用flush()方法，将缓存中的数据同步到editlog文件中
      // do the sync
      long start = monotonicNow();
      try {
        if (logStream != null) {
          logStream.flush();
        }
      } catch (IOException ex) {
        synchronized (this) {
          final String msg =
              "Could not sync enough journals to persistent storage. "
              + "Unsynced transactions: " + (txid - synctxid);
          LOG.error(msg, new Exception());
          synchronized(journalSetLock) {
            IOUtils.cleanupWithLogger(LOG, journalSet);
          }
          terminate(1, msg);
        }
      }
      long elapsed = monotonicNow() - start;
      if (metrics != null) { // Metrics non-null only when used inside name node
        metrics.addSync(elapsed);
        metrics.incrTransactionsBatchedInSync(editsBatchedInSync);
        numTransactionsBatchedInSync.addAndGet(editsBatchedInSync);
      }
    } finally {
      // Prevent RuntimeException from blocking other log edit sync
      //第三部分， 恢复标志位
      synchronized (this) {
        if (sync) {
          // 已同步txid赋值为开始sync操作的txid
          synctxid = syncStart;
          for (JournalManager jm : journalSet.getJournalManagers()) {
            /**
             * {@link FileJournalManager#lastReadableTxId} is only meaningful
             * for file-based journals. Therefore the interface is not added to
             * other types of {@link JournalManager}.
             */
            if (jm instanceof FileJournalManager) {
              ((FileJournalManager)jm).setLastReadableTxId(syncStart);
            }
          }
          isSyncRunning = false;
        }
        this.notifyAll();
     }
    }
  }

  由于 logEdit() 方法中输出日志记录和调用 logSync() 刷新缓冲区数据到磁盘这两个操作是独立加锁的，同时EditLogOutputStream 提供了两个缓冲区可以同时进行日志记录和刷新缓冲区操作，它们都使用 FSEditLog 对象作为锁对象，所以 logEdit() 方法中使用 synchronized 关键字同步的日志记录操作和 logSync() 方法中使用synchronized 关键字同步的刷新缓冲区数据到磁盘的操作是可以并发同步进行的。这种设计大大地提高了多个线程记录 editlog 操作的并发性，且通过 transactionId 机制保证了 editlog 日志记录的正确性