/*
 * 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.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

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

public class LRUEvictor extends BlockStoreEventListenerBase implements Evictor {
  private static final Logger LOG = LoggerFactory.getLogger(Constants.LOGGER_TYPE);
  private final BlockMetadataManager mMeta;

  /**
   * access-ordered {@link java.util.LinkedHashMap} from blockId to {@code true}(just to occupy the
   * value), acts as a LRU double linked list where most recently accessed element is put at the
   * tail while least recently accessed element is put at the head.
   */
  private Map<Long, Boolean> mLRUCache = Collections
      .synchronizedMap(new LinkedHashMap<Long, Boolean>(200, 0.75f, true));

  public LRUEvictor(BlockMetadataManager meta) {
    mMeta = meta;
  }

  @Override
  public EvictionPlan freeSpace(long availableBytes, BlockStoreLocation location)
      throws IOException {
    List<Pair<Long, BlockStoreLocation>> toMove = new ArrayList<Pair<Long, BlockStoreLocation>>();
    List<Long> toEvict = new ArrayList<Long>();
    EvictionPlan plan = null;

    long alreadyAvailableBytes = mMeta.getAvailableBytes(location);
    if (alreadyAvailableBytes >= availableBytes) {
      plan = new EvictionPlan(toMove, toEvict);
      return plan;
    }
    long toEvictBytes = availableBytes - alreadyAvailableBytes;

    EvictionDirCandidates dirCandidates = new EvictionDirCandidates();

    Iterator<Map.Entry<Long, Boolean>> it = mLRUCache.entrySet().iterator();
    while (it.hasNext() && dirCandidates.maxAvailableBytes() < toEvictBytes) {
      long blockId = it.next().getKey();

      try {
        BlockMeta meta = mMeta.getBlockMeta(blockId);

        BlockStoreLocation dir = meta.getBlockLocation();
        if (dir.belongTo(location)) {
          dirCandidates.add(dir, meta.getBlockId(), meta.getBlockSize());
        }
      } catch (IOException ioe) {
        LOG.warn("Remove block %d from LRU Cache because %s", blockId, ioe);
        it.remove();
      }
    }

    // enough free space
    if (dirCandidates.maxAvailableBytes() >= toEvictBytes) {
      // candidate blockIds for eviction from current tier, sorted from less recently used to more
      // recently used
      toEvict = dirCandidates.toEvict();
      // reverse list so that the more recently used blocks are moved to next tier first
      Collections.reverse(toEvict);
      // move as many blocks to next tier as possible
      int nextTierLevel = mMeta.getBlockMeta(toEvict.get(0)).getBlockLocation().tierLevel() + 1;
      List<StorageTier> tiers = mMeta.getTiers();
      int nTotalTiers = tiers.size();
      while (nextTierLevel < nTotalTiers && toEvict.size() > 0) {
        // find space in next tier
        List<StorageDir> dirs = tiers.get(nextTierLevel).getStorageDirs();
        for (StorageDir dir : dirs) {
          BlockStoreLocation dest = dir.toBlockStoreLocation();
          Iterator<Long> blocks = toEvict.iterator();
          long remainBytes = dir.getAvailableBytes();
          while (blocks.hasNext() && remainBytes >= 0) {
            long blockId = blocks.next();
            long blockSize = mMeta.getBlockMeta(blockId).getBlockSize();
            if (blockSize <= remainBytes) {
              // the block can be moved to the dir
              toMove.add(new Pair<Long, BlockStoreLocation>(blockId, dest));
              blocks.remove();
              remainBytes -= blockSize;
            }
          }
        }
        nextTierLevel ++;
      }
      for (Long blockId : toEvict) {
        mLRUCache.remove(blockId);
      }
      plan = new EvictionPlan(toMove, toEvict);
    }

    return plan;
  }

  /**
   * Thread safe
   */
  @Override
  public void onAccessBlock(long userId, long blockId) {
    mLRUCache.put(blockId, true);
  }

  @Override
  public void onCommitBlock(long userId, long blockId, BlockStoreLocation location) {
    // Since the temp block has been committed, update Evictor about the new added blocks
    mLRUCache.put(blockId, true);
  }

  @Override
  public void onPreloadBlock(long blockId) {
    mLRUCache.put(blockId, true);
  }

  @Override
  public void onRemoveBlockByClient(long userId, long blockId) {
    mLRUCache.remove(blockId);
  }

  @Override
  public void onRemoveBlockByWorker(long userId, long blockId) {
    mLRUCache.remove(blockId);
  }
}

@@ -4801,24 +4801,25 @@
 ++;%0A      %7D%0A
+%0A
       for (L
@@ -4882,24 +4882,614 @@
 d);%0A      %7D%0A
+%0A      // assure all blocks are in the store, if not, remove from plan and lru cache%0A      Iterator%3CPair%3CLong, BlockStoreLocation%3E%3E moveIt = toMove.iterator();%0A      while (moveIt.hasNext()) %7B%0A        long id = moveIt.next().getFirst();%0A        if (!mMeta.hasBlockMeta(id)) %7B%0A          moveIt.remove();%0A          mLRUCache.remove(id);%0A        %7D%0A      %7D%0A      Iterator%3CLong%3E evictIt = toEvict.iterator();%0A      while (evictIt.hasNext()) %7B%0A        long id = evictIt.next();%0A        if (!mMeta.hasBlockMeta(id)) %7B%0A          evictIt.remove();%0A          mLRUCache.remove(id);%0A        %7D%0A      %7D%0A%0A
       plan =

check whether blocks are in store before returning the plan