Added lineage file recording in lineage master

Alluxio · Sep 19, 2015 · f1cb755 · f1cb755
1 parent aa0d537
commit f1cb755
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 16 deletions.
diff --git a/servers/src/main/java/tachyon/master/lineage/meta/Lineage.java b/servers/src/main/java/tachyon/master/lineage/meta/Lineage.java
@@ -73,4 +73,21 @@ public void setState(LineageState newState) {
   public long getId() {
     return mId;
   }
+
+  public void recordOutputFile(long fileId) {
+    boolean allRecorded = true;
+    for (LineageFile outputFile : mOutputFiles) {
+      if (outputFile.getFileId() == fileId) {
+        outputFile.setState(LineageFileState.RECORDED);
+      }
+
+      if (outputFile.getState() != LineageFileState.RECORDED) {
+        allRecorded = false;
+      }
+    }
+
+    if (allRecorded) {
+      mState = LineageState.RECORDED;
+    }
+  }
 }
diff --git a/servers/src/main/java/tachyon/master/lineage/meta/LineageStore.java b/servers/src/main/java/tachyon/master/lineage/meta/LineageStore.java
@@ -16,12 +16,10 @@
 
 import java.util.List;
 import java.util.Map;
-import java.util.Set;
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
 
 import tachyon.client.file.TachyonFile;
 import tachyon.dag.DAG;
@@ -35,14 +33,12 @@ public final class LineageStore {
 
   /** Indices for lineages */
   /** Index of the output files of lineage to lineage */
-  private Map<TachyonFile, Lineage> mOutputFileIndex;
+  private Map<Long, Lineage> mOutputFileIndex;
-  private Map<LineageState, Set<Lineage>> mStateIndex;
   private Map<Long, Lineage> mIdIndex;
 
   public LineageStore() {
     mLineageDAG = new DAG<Lineage>();
     mOutputFileIndex = Maps.newHashMap();
-    mStateIndex = Maps.newHashMap();
     mIdIndex = Maps.newHashMap();
   }
 
@@ -60,14 +56,19 @@ public long addLineage(List<TachyonFile> inputFiles, List<LineageFile> outputFil
 
     // update index
     for (TachyonFile outputFile : outputFiles) {
-      mOutputFileIndex.put(outputFile, lineage);
+      mOutputFileIndex.put(outputFile.getFileId(), lineage);
     }
-    updateState(lineage, lineage.getState());
     mIdIndex.put(lineage.getId(), lineage);
 
     return lineage.getId();
   }
 
+  public void recordFileForAsyncWrite(long fileId, String underFsPath) {
+    Preconditions.checkState(mOutputFileIndex.containsKey(fileId));
+    Lineage lineage = mOutputFileIndex.get(fileId);
+    lineage.recordOutputFile(fileId);
+  }
+
   public void deleteLineage(long lineageId) {
     Preconditions.checkState(mIdIndex.containsKey(lineageId),
         "lineage id " + lineageId + " does not exist");
@@ -84,7 +85,6 @@ public void deleteLineage(long lineageId) {
     for (TachyonFile outputFile : toDelete.getOutputFiles()) {
       mOutputFileIndex.remove(outputFile);
     }
-    mStateIndex.get(toDelete.getState()).remove(toDelete);
   }
 
   public Lineage getLineage(long lineageId) {
@@ -97,12 +97,4 @@ public List<Lineage> getChildren(Lineage lineage) {
 
     return mLineageDAG.getChildren(lineage);
   }
-
-  private void updateState(Lineage lineage, LineageState newState) {
-    lineage.setState(newState);
-    if (!mStateIndex.containsKey(newState)) {
-      mStateIndex.put(newState, Sets.<Lineage>newHashSet());
-    }
-    mStateIndex.get(newState).add(lineage);
-  }
 }