Improve journal handling and code comments

servers/src/main/java/tachyon/master/MasterBase.java

@@ -27,23 +27,30 @@
 import tachyon.master.journal.Journal;
 import tachyon.master.journal.JournalEntry;
 import tachyon.master.journal.JournalInputStream;
+import tachyon.master.journal.JournalOutputStream;
 import tachyon.master.journal.JournalSerializable;
 import tachyon.master.journal.JournalTailer;
 import tachyon.master.journal.JournalTailerThread;
 import tachyon.master.journal.JournalWriter;
 
+/**
+ * This is the base class for all masters, and contains common functionality. Common functionality
+ * mostly consists of journal operations, like initializing it, tailing it when in standby mode, or
+ * writing to it when the master is the leader.
+ */
 public abstract class MasterBase implements Master {
   private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
 
+  /** A handler to the journal for this master. */
   private final Journal mJournal;
+  /** The executor used for running maintenance threads for the master. */
   private final ExecutorService mExecutorService;
 
   /** true if this master is in leader mode, and not standby mode. */
   private boolean mIsLeader = false;
-
   /** The thread that tails the journal when the master is in standby mode. */
   private JournalTailerThread mStandbyJournalTailer = null;
-
+  /** The journal writer for when the master is the leader. */
   private JournalWriter mJournalWriter = null;
 
   protected MasterBase(Journal journal, ExecutorService executorService) {
@@ -62,39 +69,48 @@ public void processJournalCheckpoint(JournalInputStream inputStream) throws IOEx
 
   @Override
   public void start(boolean isLeader) throws IOException {
-    LOG.info("Starting master. isLeader: " + isLeader);
+    LOG.info(getProcessorName() + ": Starting master. isLeader: " + isLeader);
     mIsLeader = isLeader;
     if (mIsLeader) {
-      // Replay all the state of the checkpoint and the completed log files.
+      mJournalWriter = mJournal.getNewWriter();
+
+      /**
+       * The sequence for dealing with the journal before starting as the leader:
+       *
+       * Phase 1. Mark all the logs as completed. Since this master is the leader, it is allowed to
+       * write the journal, so it can mark the current log as completed. After this step, the
+       * current log file will not exist, and all logs will be complete.
+       *
+       * Phase 2. Reconstruct the state from the journal. This uses the JournalTailer to process all
+       * of the checkpoint and the complete log files. Since all logs are complete, after this step,
+       * the master will reflect the state of all of the journal entries.
+       *
+       * Phase 3. Write out the checkpoint file. Since this master is completely up-to-date, it
+       * writes out the checkpoint file. When the checkpoint file is closed, it will then delete the
+       * complete log files.
+       */
+
+      // Phase 1: Mark all logs as complete, including the current log. After this call, the current
+      // log should not exist, and all the log files will be complete.
+      mJournalWriter.completeAllLogs();
+
+      // Phase 2: Replay all the state of the checkpoint and the completed log files.
       // TODO: only do this if this is a fresh start, not if this master had already been tailing
       // the journal.
       LOG.info(getProcessorName() + ": process completed logs before becoming master.");
       JournalTailer catchupTailer = new JournalTailer(this, mJournal);
-      boolean checkpointExists = true;
-      try {
-        catchupTailer.getCheckpointLastModifiedTimeMs();
-      } catch (IOException ioe) {
-        // The checkpoint doesn't exist yet. This is probably the first execution ever, or this is a
-        // testing master.
-        checkpointExists = false;
-      }
-      if (checkpointExists) {
+      if (catchupTailer.checkpointExists()) {
         catchupTailer.processJournalCheckpoint(true);
         catchupTailer.processNextJournalLogFiles();
       }
-
-      // initialize the journal and write out the checkpoint file (the state of all completed logs).
-      mJournalWriter = mJournal.getNewWriter();
-      writeToJournal(mJournalWriter.getCheckpointOutputStream());
-      mJournalWriter.getCheckpointOutputStream().close();
-
-      // Final catchup stage. The last in-progress file (if it existed) was marked as complete when
-      // the checkpoint file was closed. That last completed file must be processed to get to the
-      // latest state. Read and process the completed file.
-      LOG.info(getProcessorName() + ": process the last completed log before becoming master.");
-      catchupTailer = new JournalTailer(this, mJournal);
-      catchupTailer.processJournalCheckpoint(false);
-      catchupTailer.processNextJournalLogFiles();
+      long latestSequenceNumber = catchupTailer.getLatestSequenceNumber();
+
+      // Phase 3: initialize the journal and write out the checkpoint file (the state of all
+      // completed logs).
+      JournalOutputStream checkpointStream =
+          mJournalWriter.getCheckpointOutputStream(latestSequenceNumber);
+      writeToJournal(checkpointStream);
+      checkpointStream.close();
     } else {
       // in standby mode. Start the journal tailer thread.
       mStandbyJournalTailer = new JournalTailerThread(this, mJournal);
@@ -104,7 +120,7 @@ public void start(boolean isLeader) throws IOException {
 
   @Override
   public void stop() throws IOException {
-    LOG.info("Stopping master. isLeader: " + isLeaderMode());
+    LOG.info(getProcessorName() + ":Stopping master. isLeader: " + isLeaderMode());
     if (isStandbyMode()) {
       if (mStandbyJournalTailer != null) {
         // stop and wait for the journal tailer thread.
@@ -160,6 +176,9 @@ protected void flushJournal() {
     }
   }
 
+  /**
+   * @return the executor service for this master.
+   */
   protected ExecutorService getExecutorService() {
     return mExecutorService;
   }

servers/src/main/java/tachyon/master/file/FileSystemMaster.java

@@ -42,7 +42,6 @@
 import tachyon.master.file.journal.CompleteFileEntry;
 import tachyon.master.file.journal.DeleteFileEntry;
 import tachyon.master.file.journal.DependencyEntry;
-import tachyon.master.file.journal.FreeEntry;
 import tachyon.master.file.journal.InodeDirectoryIdGeneratorEntry;
 import tachyon.master.file.journal.InodeEntry;
 import tachyon.master.file.journal.RenameEntry;
@@ -144,8 +143,6 @@ public void processJournalEntry(JournalEntry entry) throws IOException {
       completeFileFromEntry((CompleteFileEntry) entry);
     } else if (entry instanceof AddCheckpointEntry) {
       completeFileCheckpointFromEntry((AddCheckpointEntry) entry);
-    } else if (entry instanceof FreeEntry) {
-      freeFromEntry((FreeEntry) entry);
     } else if (entry instanceof SetPinnedEntry) {
       setPinnedFromEntry((SetPinnedEntry) entry);
     } else if (entry instanceof DeleteFileEntry) {
@@ -799,41 +796,26 @@ public boolean free(long fileId, boolean recursive)
         // true
         return false;
       }
-      freeInternal(inode);
-      writeJournalEntry(new FreeEntry(fileId));
-      flushJournal();
-    }
-
-    return true;
-  }
 
-  private void freeInternal(Inode inode) {
-    List<Inode> freeInodes = new ArrayList<Inode>();
-    freeInodes.add(inode);
-    if (inode.isDirectory()) {
-      freeInodes.addAll(mInodeTree.getInodeChildrenRecursive((InodeDirectory) inode));
-    }
+      List<Inode> freeInodes = new ArrayList<Inode>();
+      freeInodes.add(inode);
+      if (inode.isDirectory()) {
+        freeInodes.addAll(mInodeTree.getInodeChildrenRecursive((InodeDirectory) inode));
+      }
 
-    // We go through each inode.
-    for (int i = freeInodes.size() - 1; i >= 0; i--) {
-      Inode freeInode = freeInodes.get(i);
+      // We go through each inode.
+      for (int i = freeInodes.size() - 1; i >= 0; i --) {
+        Inode freeInode = freeInodes.get(i);
 
-      if (freeInode.isFile()) {
-        // Remove corresponding blocks from workers.
-        mBlockMaster.removeBlocks(((InodeFile) freeInode).getBlockIds());
+        if (freeInode.isFile()) {
+          // Remove corresponding blocks from workers.
+          mBlockMaster.removeBlocks(((InodeFile) freeInode).getBlockIds());
+        }
       }
     }
+    return true;
   }
 
-  private void freeFromEntry(FreeEntry entry) {
-    try {
-      freeInternal(mInodeTree.getInodeById(entry.getId()));
-    } catch (FileDoesNotExistException fdnee) {
-      throw new RuntimeException(fdnee);
-    }
-  }
-
-
   /**
    * Gets the path of a file with the given id
    *

servers/src/main/java/tachyon/master/file/journal/InodeDirectoryEntry.java

@@ -34,6 +34,7 @@ public InodeDirectoryEntry(long creationTimeMs, long id, String name, long paren
   public InodeDirectory toInodeDirectory() {
     InodeDirectory inode = new InodeDirectory(mName, mId, mParentId, mCreationTimeMs);
     inode.setPinned(mIsPinned);
+    inode.setLastModificationTimeMs(mLastModificationTimeMs);
     return inode;
   }
 

servers/src/main/java/tachyon/master/file/journal/InodeFileEntry.java

@@ -50,8 +50,13 @@ public InodeFile toInodeFile() {
     if (mIsComplete) {
       inode.setComplete(mLength);
     }
+    if (mBlocks != null) {
+      inode.setBlockIds(mBlocks);
+    }
     inode.setPinned(mIsPinned);
     inode.setCache(mIsCache);
+    inode.setLastModificationTimeMs(mLastModificationTimeMs);
+    inode.setUfsPath(mUfsPath);
 
     return inode;
   }

servers/src/main/java/tachyon/master/journal/Journal.java

@@ -28,45 +28,63 @@
  * completed entry files are in the "completed/" sub-directory.
  */
 public class Journal {
+  /** The log number for the first completed log file. */
   public static final int FIRST_COMPLETED_LOG_NUMBER = 1;
+  /** The directory for completed log files, relative to the base journal directory. */
   private static final String COMPLETED_DIRECTORY = "completed/";
+  /** The file extension for the current log file. */
   private static final String CURRENT_LOG_EXTENSION = ".out";
-
-  // TODO: should this be a config parameter?
   /** The filename of the checkpoint file. */
-  private final String mCheckpointFilename = "checkpoint.data";
-  // TODO: should this be a config parameter?
+  private static final String CHECKPOINT_FILENAME = "checkpoint.data";
   /** The base of the entry log filenames, without the file extension. */
-  private final String mEntryLogFilenameBase = "log";
-  private final String mDirectory;
+  private static final String ENTRY_LOG_FILENAME_BASE = "log";
+
   private final TachyonConf mTachyonConf;
+  /** The directory where this journal is stored. */
+  private final String mDirectory;
+  /** The formatter for this journal. */
   private final JournalFormatter mJournalFormatter;
 
+  /**
+   * @param directory the base directory for this journal
+   * @param tachyonConf the tachyon conf
+   */
   public Journal(String directory, TachyonConf tachyonConf) {
     if (!directory.endsWith(TachyonURI.SEPARATOR)) {
       // Ensure directory format.
       directory += TachyonURI.SEPARATOR;
     }
     mDirectory = directory;
     mTachyonConf = tachyonConf;
-    // TODO: maybe this can be constructed, specified by a parameter in tachyonConf.
     mJournalFormatter = new JsonJournalFormatter();
   }
 
+  /**
+   * @return the base directory for this journal
+   */
   public String getDirectory() {
     return mDirectory;
   }
 
+  /**
+   * @return the directory for where the completed log files are stored
+   */
   public String getCompletedDirectory() {
     return mDirectory + COMPLETED_DIRECTORY;
   }
 
+  /**
+   * @return the absolute path for the journal checkpoint file
+   */
   public String getCheckpointFilePath() {
-    return mDirectory + mCheckpointFilename;
+    return mDirectory + CHECKPOINT_FILENAME;
   }
 
+  /**
+   * @return the absolute path for the current log file.
+   */
   public String getCurrentLogFilePath() {
-    return mDirectory + mEntryLogFilenameBase + CURRENT_LOG_EXTENSION;
+    return mDirectory + ENTRY_LOG_FILENAME_BASE + CURRENT_LOG_EXTENSION;
   }
 
   /**
@@ -76,21 +94,33 @@ public String getCurrentLogFilePath() {
    * @return The absolute path of the completed log for a given log number.
    */
   public String getCompletedLogFilePath(int logNumber) {
-    return getCompletedDirectory() + String.format("%s.%07d", mEntryLogFilenameBase, logNumber);
+    return getCompletedDirectory() + String.format("%s.%07d", ENTRY_LOG_FILENAME_BASE, logNumber);
   }
 
+  /**
+   * @return the formatter for this journal
+   */
   public JournalFormatter getJournalFormatter() {
     return mJournalFormatter;
   }
 
+  /**
+   * @return a readonly version of this journal
+   */
   public ReadOnlyJournal getReadOnlyJournal() {
     return new ReadOnlyJournal(mDirectory, mTachyonConf);
   }
 
+  /**
+   * @return the writer for this journal
+   */
   public JournalWriter getNewWriter() {
     return new JournalWriter(this, mTachyonConf);
   }
 
+  /**
+   * @return the reader for this journal
+   */
   public JournalReader getNewReader() {
     return new JournalReader(this, mTachyonConf);
   }

servers/src/main/java/tachyon/master/journal/JournalEntry.java

@@ -20,10 +20,14 @@
 // TODO: In the future, implementations of this interface can be represented as ProtoBuf
 public interface JournalEntry {
 
+  /**
+   * @return the {@link JournalEntryType} of this entry.
+   */
   JournalEntryType getType();
 
   /**
-   * Gets parameters of this entry which is a map from parameter name to parameter value.
+   *
+   * @return parameters of this entry which is a map from parameter name to parameter value.
    */
   Map<String, Object> getParameters();
 }