/*
 * Licensed to the University of California, Berkeley under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package tachyon.worker.block.evictor;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import tachyon.Pair;
import tachyon.worker.block.BlockMetadataManager;
import tachyon.worker.block.BlockStoreLocation;
import tachyon.worker.block.meta.BlockMeta;
import tachyon.worker.block.meta.StorageDir;

public class EvictorUtils {
  /**
   * Checks whether the plan is valid for an evictor with cascading eviction feature like
   * {@link LRUEvictor}. The plan is legal when the requested space can be satisfied, and there is
   * enough space in lower tier to move in blocks from upper tier.
   *
   * @param bytesToBeAvailable requested bytes to be available after eviction
   * @param plan the eviction plan, should not be empty
   * @param metaManager the meta data manager
   * @return true if the above requirements are satisfied, otherwise false.
   */
  public static boolean validCascadingPlan(long bytesToBeAvailable, EvictionPlan plan,
      BlockMetadataManager metaManager) throws IOException {
    // reassure the plan is feasible: enough free space to satisfy bytesToBeAvailable, and enough
    // space in lower tier to move blocks in upper tier there

    // Map from dir to a pair of bytes to be available in this dir and bytes to move into this dir
    // after the plan taking action
    Map<StorageDir, Pair<Long, Long>> spaceInfoInDir = new HashMap<StorageDir, Pair<Long, Long>>();

    for (long blockId : plan.toEvict()) {
      BlockMeta block = metaManager.getBlockMeta(blockId);
      StorageDir dir = block.getParentDir();
      if (spaceInfoInDir.containsKey(dir)) {
        Pair<Long, Long> spaceInfo = spaceInfoInDir.get(dir);
        spaceInfo.setFirst(spaceInfo.getFirst() + block.getBlockSize());
      } else {
        spaceInfoInDir.put(dir, new Pair<Long, Long>(
            dir.getAvailableBytes() + block.getBlockSize(), 0L));
      }
    }

    for (Pair<Long, BlockStoreLocation> move : plan.toMove()) {
      long blockId = move.getFirst();
      BlockMeta block = metaManager.getBlockMeta(blockId);
      long blockSize = block.getBlockSize();
      StorageDir srcDir = block.getParentDir();
      StorageDir destDir = metaManager.getDir(move.getSecond());

      if (spaceInfoInDir.containsKey(srcDir)) {
        Pair<Long, Long> spaceInfo = spaceInfoInDir.get(srcDir);
        spaceInfo.setFirst(spaceInfo.getFirst() + blockSize);
      } else {
        spaceInfoInDir
            .put(srcDir, new Pair<Long, Long>(srcDir.getAvailableBytes() + blockSize, 0L));
      }

      if (spaceInfoInDir.containsKey(destDir)) {
        Pair<Long, Long> spaceInfo = spaceInfoInDir.get(destDir);
        spaceInfo.setSecond(spaceInfo.getSecond() + blockSize);
      } else {
        spaceInfoInDir.put(destDir, new Pair<Long, Long>(destDir.getAvailableBytes(), blockSize));
      }
    }

    int firstTierAlias = Integer.MAX_VALUE;
    for (StorageDir dir : spaceInfoInDir.keySet()) {
      firstTierAlias = Math.min(firstTierAlias, dir.getParentTier().getTierAlias());
    }
    long maxSpace = Long.MIN_VALUE; // maximum bytes to be available in a dir in the first tier
    for (StorageDir dir : spaceInfoInDir.keySet()) {
      if (dir.getParentTier().getTierAlias() == firstTierAlias) {
        Pair<Long, Long> space = spaceInfoInDir.get(dir);
        maxSpace = Math.max(maxSpace, space.getFirst() - space.getSecond());
      }
    }
    if (maxSpace < bytesToBeAvailable) {
      return false;
    }

    for (StorageDir dir : spaceInfoInDir.keySet()) {
      Pair<Long, Long> spaceInfo = spaceInfoInDir.get(dir);
      if (spaceInfo.getFirst() < spaceInfo.getSecond()) {
        return false;
      }
    }

    return true;
  }
}

@@ -1178,93 +1178,194 @@
 lan 
-is valid for an evictor with cascading eviction feature like%0A   * %7B@link LRUEvictor%7D.
+of a cascading evictor is valid.%0A   *%0A   * A cascading evictor will try to free space by recursively moving blocks to next 1 tier and%0A   * evict blocks only in the bottom tier.%0A   *%0A   *
  The
@@ -1377,13 +1377,15 @@
  is 
-legal
+invalid
  whe
@@ -1410,16 +1410,20 @@
 ace can 
+not 
 be satis
@@ -1430,88 +1430,110 @@
 fied
-, and there is%0A   * enough space in lower tier to move in blocks from upper
+ or lower level of tiers do%0A   * not have enough space to hold blocks moved from higher level of
  tier
+s
 .%0A  
@@ -1795,14 +1795,47 @@
 se.%0A
-
    */%0A
+  // TODO: unit test this method%0A
   pu
@@ -3801,17 +3801,86 @@
     
-int first
+// the top tier among all tiers where blocks in the plan reside in%0A    int top
 Tier
@@ -3965,21 +3965,19 @@
 %7B%0A      
-first
+top
 TierAlia
@@ -3989,21 +3989,19 @@
 ath.min(
-first
+top
 TierAlia
@@ -4131,21 +4131,19 @@
  in the 
-first
+top
  tier%0A  
@@ -4245,13 +4245,11 @@
  == 
-first
+top
 Tier
@@ -4443,24 +4443,110 @@
 vailable) %7B%0A
+      // plan is invalid because requested space can not be satisfied in the top tier%0A
       return
@@ -4727,24 +4727,161 @@
 Second()) %7B%0A
+        // plan is invalid because there is not enough space in this dir to hold the blocks waiting%0A        // to be moved into this dir%0A
         retu

resolve bin's comment