fix block not deleted issue during promotion and data folder fix

conf/tachyon-env.sh.template

@@ -67,7 +67,7 @@ export TACHYON_JAVA_OPTS+="
   -Dtachyon.data.folder=$TACHYON_UNDERFS_ADDRESS/tmp/tachyon/data
   -Dtachyon.workers.folder=$TACHYON_UNDERFS_ADDRESS/tmp/tachyon/workers
   -Dtachyon.worker.memory.size=$TACHYON_WORKER_MEMORY_SIZE
-  -Dtachyon.worker.data.folder=$TACHYON_RAM_FOLDER/tachyonworker/
+  -Dtachyon.worker.data.folder=/tachyonworker/
   -Dtachyon.master.worker.timeout.ms=60000
   -Dtachyon.master.hostname=$TACHYON_MASTER_ADDRESS
   -Dtachyon.master.journal.folder=$TACHYON_HOME/journal/

core/src/main/java/tachyon/worker/WorkerStorage.java

@@ -741,13 +741,8 @@ public long lockBlock(long blockId, long userId) {
    * @param userId the id of the user
    * @param blockId the id of the block
    * @return true if success, false otherwise
-   * @throws FileDoesNotExistException
-   * @throws SuspectedFileSizeException
-   * @throws BlockInfoException
    */
-  public boolean promoteBlock(long userId, long blockId)
-      throws FileDoesNotExistException, SuspectedFileSizeException, BlockInfoException {
-
+  public boolean promoteBlock(long userId, long blockId) {
     long storageDirIdLocked = lockBlock(blockId, userId);
     if (StorageDirId.isUnknown(storageDirIdLocked)) {
       return false;
@@ -758,21 +753,24 @@ public boolean promoteBlock(long userId, long blockId)
       StorageDir dstStorageDir = requestSpace(userId, blockSize);
       if (dstStorageDir == null) {
         LOG.error("Failed to promote block! blockId:" + blockId);
-        unlockBlock(blockId, userId);
+        srcStorageDir.unlockBlock(blockId, userId);
         return false;
       }
-      boolean result;
+      boolean result = false;
       try {
-        result = srcStorageDir.moveBlock(blockId, dstStorageDir);
-        mMasterClient.worker_cacheBlock(mWorkerId, mCapacityBytes, dstStorageDir.getStorageDirId(),
-            blockId, blockSize);
+        try {
+          result = srcStorageDir.copyBlock(blockId, dstStorageDir);
+        } finally {
+          srcStorageDir.unlockBlock(blockId, userId);
+        }
+        if (result) {
+          srcStorageDir.deleteBlock(blockId);
+        }
+        return result;
       } catch (IOException e) {
         LOG.error("Failed to promote block! blockId:" + blockId);
         return false;
-      } finally {
-        unlockBlock(blockId, userId);
       }
-      return result;
     } else {
       unlockBlock(blockId, userId);
       return true;

core/src/main/java/tachyon/worker/hierarchy/StorageDir.java

@@ -19,6 +19,8 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -56,6 +58,8 @@ public final class StorageDir {
   /** List of added block Ids to be reported */
   private final BlockingQueue<Long> mAddedBlockIdList = new ArrayBlockingQueue<Long>(
       Constants.WORKER_BLOCKS_QUEUE_SIZE);
+  /** List of to be removed block Ids */
+  private final Set<Long> mToRemoveBlockIdSet = Collections.synchronizedSet(new HashSet<Long>());
   /** Space counter of current StorageDir */
   private final SpaceCounter mSpaceCounter;
   /** Id of StorageDir */
@@ -244,6 +248,8 @@ public boolean deleteBlock(long blockId) throws IOException {
         // Should check lock status here 
         if (!isBlockLocked(blockId)) {
           result = mFs.delete(blockfile, true);
+        } else {
+          mToRemoveBlockIdSet.add(blockId);
         }
       } finally {
         if (result) {
@@ -649,6 +655,15 @@ public boolean unlockBlock(long blockId, long userId) {
     }
     mUserPerLockedBlock.remove(blockId, userId);
     mLockedBlocksPerUser.remove(userId, blockId);
+    if (!mUserPerLockedBlock.containsKey(blockId) && mToRemoveBlockIdSet.contains(blockId)) {
+      try {
+        if (deleteBlock(blockId)) {
+          mToRemoveBlockIdSet.remove(blockId);
+        }
+      } catch (IOException e) {
+        LOG.error(e.getMessage());
+      }
+    }
     return true;
   }
 }

core/src/test/java/tachyon/worker/hierarchy/HierarchyStoreTest.java

@@ -109,5 +109,7 @@ public void promoteBlock() throws IOException, InterruptedException {
     Assert.assertEquals(true, file1.isInMemory());
     Assert.assertEquals(false, file2.isInMemory());
     Assert.assertEquals(true, file3.isInMemory());
+    Assert.assertEquals(mMemCapacityBytes / 6 + mMemCapacityBytes,
+        mLocalTachyonCluster.getMasterInfo().getUsedBytes());
   }
 }