alluxio-master-2.x/core/server/worker/src/main/java/alluxio/worker/block/BlockMetadataManager.java
/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.worker.block;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.collect.ImmutableList.toImmutableList;
import static com.google.common.collect.ImmutableMap.toImmutableMap;
import static java.util.function.Function.identity;

import alluxio.DefaultStorageTierAssoc;
import alluxio.StorageTierAssoc;
import alluxio.conf.Configuration;
import alluxio.conf.PropertyKey;
import alluxio.exception.ExceptionMessage;
import alluxio.worker.block.allocator.Allocator;
import alluxio.worker.block.annotator.BlockAnnotator;
import alluxio.worker.block.annotator.BlockIterator;
import alluxio.worker.block.annotator.DefaultBlockIterator;
import alluxio.worker.block.annotator.EmulatingBlockIterator;
import alluxio.worker.block.evictor.Evictor;
import alluxio.worker.block.meta.BlockMeta;
import alluxio.worker.block.meta.DefaultBlockMeta;
import alluxio.worker.block.meta.DefaultStorageTier;
import alluxio.worker.block.meta.StorageDir;
import alluxio.worker.block.meta.StorageTier;
import alluxio.worker.block.meta.TempBlockMeta;

import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.IntStream;
import javax.annotation.concurrent.NotThreadSafe;

/**
 * Manages the metadata of all blocks in managed space. This information is used by the
 * {@link TieredBlockStore}, {@link Allocator} and {@link Evictor}.
 * <p>
 * All operations on block metadata such as {@link StorageTier}, {@link StorageDir} should go
 * through this class.
 */
@NotThreadSafe
// TODO(bin): consider how to better expose information to Evictor and Allocator.
public final class BlockMetadataManager {
  private static final Logger LOG = LoggerFactory.getLogger(BlockMetadataManager.class);
  public static final StorageTierAssoc WORKER_STORAGE_TIER_ASSOC =
      new DefaultStorageTierAssoc(
          PropertyKey.WORKER_TIERED_STORE_LEVELS,
          PropertyKey.Template.WORKER_TIERED_STORE_LEVEL_ALIAS);

  /** A list of managed {@link StorageTier}, in order from lowest tier ordinal to greatest. */
  private final List<StorageTier> mTiers;

  /** A map from tier alias to {@link StorageTier}. */
  private final Map<String, StorageTier> mAliasToTiers;

  /** Used to get iterators per locations. */
  private final BlockIterator mBlockIterator;

  /** Deprecated evictors. */
  private static final String DEPRECATED_LRU_EVICTOR = "alluxio.worker.block.evictor.LRUEvictor";
  private static final String DEPRECATED_PARTIAL_LRUEVICTOR =
      "alluxio.worker.block.evictor.PartialLRUEvictor";
  private static final String DEPRECATED_LRFU_EVICTOR = "alluxio.worker.block.evictor.LRFUEvictor";
  private static final String DEPRECATED_GREEDY_EVICTOR =
      "alluxio.worker.block.evictor.GreedyEvictor";

  private BlockMetadataManager() {
    mTiers = IntStream.range(0, WORKER_STORAGE_TIER_ASSOC.size()).mapToObj(
        tierOrdinal -> DefaultStorageTier.newStorageTier(
            WORKER_STORAGE_TIER_ASSOC.getAlias(tierOrdinal),
            tierOrdinal,
            WORKER_STORAGE_TIER_ASSOC.size() > 1))
        .parallel()
        .collect(toImmutableList());
    mAliasToTiers = mTiers.stream().collect(toImmutableMap(StorageTier::getTierAlias, identity()));
    // Create the block iterator.
    if (Configuration.isSet(PropertyKey.WORKER_EVICTOR_CLASS)) {
      LOG.warn(String.format("Evictor is being emulated. Please use %s instead.",
          PropertyKey.Name.WORKER_BLOCK_ANNOTATOR_CLASS));
      String evictorType = Configuration.getString(PropertyKey.WORKER_EVICTOR_CLASS);
      switch (evictorType) {
        case DEPRECATED_LRU_EVICTOR:
        case DEPRECATED_PARTIAL_LRUEVICTOR:
        case DEPRECATED_GREEDY_EVICTOR:
          LOG.warn("Evictor is deprecated, switching to LRUAnnotator");
          Configuration.set(PropertyKey.WORKER_BLOCK_ANNOTATOR_CLASS,
              "alluxio.worker.block.annotator.LRUAnnotator");
          mBlockIterator = new DefaultBlockIterator(this, BlockAnnotator.Factory.create());
          break;
        case DEPRECATED_LRFU_EVICTOR:
          LOG.warn("Evictor is deprecated, switching to LRFUAnnotator");
          Configuration.set(PropertyKey.WORKER_BLOCK_ANNOTATOR_CLASS,
              "alluxio.worker.block.annotator.LRFUAnnotator");
          mBlockIterator = new DefaultBlockIterator(this, BlockAnnotator.Factory.create());
          break;
        default:
          //For user defined evictor
          BlockMetadataEvictorView initManagerView = new BlockMetadataEvictorView(this,
              Collections.emptySet(), Collections.emptySet());
          mBlockIterator = new EmulatingBlockIterator(this,
              Evictor.Factory.create(initManagerView, Allocator.Factory.create(initManagerView)));
      }
    } else {
      // Create default block iterator
      mBlockIterator = new DefaultBlockIterator(this, BlockAnnotator.Factory.create());
    }
  }

  /**
   * @return the configured iteration provider
   */
  public BlockIterator getBlockIterator() {
    return mBlockIterator;
  }

  /**
   * Creates a new instance of {@link BlockMetadataManager}.
   *
   * @return a {@link BlockMetadataManager} instance
   */
  public static BlockMetadataManager createBlockMetadataManager() {
    return new BlockMetadataManager();
  }

  /**
   * Aborts a temp block.
   *
   * @param tempBlockMeta the metadata of the temp block to add
   */
  public void abortTempBlockMeta(TempBlockMeta tempBlockMeta) {
    tempBlockMeta.getParentDir().removeTempBlockMeta(tempBlockMeta);
  }

  /**
   * Adds a temp block.
   *
   * @param tempBlockMeta the metadata of the temp block to add
   */
  public void addTempBlockMeta(TempBlockMeta tempBlockMeta) {
    StorageDir dir = tempBlockMeta.getParentDir();
    dir.addTempBlockMeta(tempBlockMeta);
  }

  /**
   * Commits a temp block.
   *
   * @param tempBlockMeta the metadata of the temp block to commit
   */
  public void commitTempBlockMeta(TempBlockMeta tempBlockMeta) {
    long blockId = tempBlockMeta.getBlockId();
    if (hasBlockMeta(blockId)) {
      throw new IllegalStateException(ExceptionMessage.ADD_EXISTING_BLOCK.getMessage(blockId,
          getBlockMeta(blockId).get().getBlockLocation().tierAlias()));
    }
    BlockMeta block = new DefaultBlockMeta(Preconditions.checkNotNull(tempBlockMeta));
    StorageDir dir = tempBlockMeta.getParentDir();
    dir.removeTempBlockMeta(tempBlockMeta);
    dir.addBlockMeta(block);
  }

  /**
   * Cleans up the metadata of the given temp block ids.
   *
   * @param sessionId the id of the client associated with the temp blocks
   * @param tempBlockIds the list of temporary block ids to be cleaned up, non temporary block ids
   *        will be ignored.
   * @deprecated As of version 0.8.
   */
  @Deprecated
  public void cleanupSessionTempBlocks(long sessionId, List<Long> tempBlockIds) {
    for (StorageTier tier : mTiers) {
      for (StorageDir dir : tier.getStorageDirs()) {
        dir.cleanupSessionTempBlocks(sessionId, tempBlockIds);
      }
    }
  }

  /**
   * Gets the amount of available space of given location in bytes. Master queries the total number
   * of bytes available on each tier of the worker, and Evictor/Allocator often cares about the
   * bytes at a {@link StorageDir}. Throws an {@link IllegalArgumentException} when the location
   * does not belong to the tiered storage.
   *
   * @param location location the check available bytes
   * @return available bytes
   */
  public long getAvailableBytes(BlockStoreLocation location) {
    long spaceAvailable = 0;

    if (location.hasNoRestriction()) {
      for (StorageTier tier : mTiers) {
        spaceAvailable += tier.getAvailableBytes();
      }
      return spaceAvailable;
    }

    if (!location.isAnyMedium() && location.isAnyDir() && location.isAnyTier()) {
      for (StorageTier tier : mTiers) {
        for (StorageDir dir : tier.getStorageDirs()) {
          if (dir.getDirMedium().equals(location.mediumType())) {
            spaceAvailable += dir.getAvailableBytes();
          }
        }
      }
      return spaceAvailable;
    }

    String tierAlias = location.tierAlias();
    StorageTier tier = getTier(tierAlias);
    // TODO(calvin): This should probably be max of the capacity bytes in the dirs?
    if (location.isAnyDir()) {
      return tier.getAvailableBytes();
    }

    int dirIndex = location.dir();
    StorageDir dir = tier.getDir(dirIndex);
    return dir == null ? 0 : dir.getAvailableBytes();
  }

  /**
   * Gets the metadata of a block given its block id.
   *
   * @param blockId the block id
   * @return metadata of the block
   */
  public Optional<BlockMeta> getBlockMeta(long blockId) {
    for (StorageTier tier : mTiers) {
      for (StorageDir dir : tier.getStorageDirs()) {
        if (dir.hasBlockMeta(blockId)) {
          return dir.getBlockMeta(blockId);
        }
      }
    }
    return Optional.empty();
  }

  /**
   * Gets a summary of the metadata.
   *
   * @return the metadata of this block store
   */
  public BlockStoreMeta getBlockStoreMeta() {
    return new DefaultBlockStoreMeta(this, false);
  }

  /**
   * Gets a full summary of block store metadata. This is an expensive operation.
   *
   * @return the full metadata of this block store
   */
  public BlockStoreMeta getBlockStoreMetaFull() {
    return new DefaultBlockStoreMeta(this, true);
  }

  /**
   * Gets the {@link StorageDir} given its location in the store. Throws an
   * {@link IllegalArgumentException} if the location is not a specific dir or the location is
   * invalid.
   *
   * @param location Location of the dir
   * @return the {@link StorageDir} object
   */
  public StorageDir getDir(BlockStoreLocation location) {
    checkArgument(!(location.isAnyTier() || location.isAnyDir()),
        MessageFormat.format("Cannot get path from non-specific dir {0}", location));
    return getTier(location.tierAlias()).getDir(location.dir());
  }

  /**
   * Gets the metadata of a temp block.
   *
   * @param blockId the id of the temp block
   * @return metadata of the block
   */
  public Optional<TempBlockMeta> getTempBlockMeta(long blockId) {
    for (StorageTier tier : mTiers) {
      for (StorageDir dir : tier.getStorageDirs()) {
        if (dir.hasTempBlockMeta(blockId)) {
          return dir.getTempBlockMeta(blockId);
        }
      }
    }
    return Optional.empty();
  }

  /**
   * Gets the {@link StorageTier} given its tierAlias. Throws an {@link IllegalArgumentException} if
   * the tierAlias is not found.
   *
   * @param tierAlias the alias of this tier
   * @return the {@link StorageTier} object associated with the alias
   */
  public StorageTier getTier(String tierAlias) {
    StorageTier tier = mAliasToTiers.get(tierAlias);
    if (tier == null) {
      throw new IllegalArgumentException(
          ExceptionMessage.TIER_ALIAS_NOT_FOUND.getMessage(tierAlias));
    }
    return tier;
  }

  /**
   * Gets the list of {@link StorageTier} managed.
   *
   * @return the list of {@link StorageTier}s
   */
  public List<StorageTier> getTiers() {
    return mTiers;
  }

  /**
   * Gets the list of {@link StorageTier} below the tier with the given tierAlias. Throws an
   * {@link IllegalArgumentException} if the tierAlias is not found.
   *
   * @param tierAlias the alias of a tier
   * @return the list of {@link StorageTier}
   */
  public List<StorageTier> getTiersBelow(String tierAlias) {
    int ordinal = getTier(tierAlias).getTierOrdinal();
    return mTiers.subList(ordinal + 1, mTiers.size());
  }

  /**
   * Gets all the temporary blocks associated with a session, empty list is returned if the session
   * has no temporary blocks.
   *
   * @param sessionId the id of the session
   * @return A list of temp blocks associated with the session
   */
  public List<TempBlockMeta> getSessionTempBlocks(long sessionId) {
    List<TempBlockMeta> sessionTempBlocks = new ArrayList<>();
    for (StorageTier tier : mTiers) {
      for (StorageDir dir : tier.getStorageDirs()) {
        sessionTempBlocks.addAll(dir.getSessionTempBlocks(sessionId));
      }
    }
    return sessionTempBlocks;
  }

  /**
   * Checks if the storage has a given block.
   *
   * @param blockId the block id
   * @return true if the block is contained, false otherwise
   */
  public boolean hasBlockMeta(long blockId) {
    for (StorageTier tier : mTiers) {
      for (StorageDir dir : tier.getStorageDirs()) {
        if (dir.hasBlockMeta(blockId)) {
          return true;
        }
      }
    }
    return false;
  }

  /**
   * Checks if the storage has a given temp block.
   *
   * @param blockId the temp block id
   * @return true if the block is contained, false otherwise
   */
  public boolean hasTempBlockMeta(long blockId) {
    for (StorageTier tier : mTiers) {
      for (StorageDir dir : tier.getStorageDirs()) {
        if (dir.hasTempBlockMeta(blockId)) {
          return true;
        }
      }
    }
    return false;
  }

  /**
   * Moves an existing block to another location currently hold by a temp block.
   *
   * @param blockMeta the metadata of the block to move
   * @param tempBlockMeta a placeholder in the destination directory
   * @return the new block metadata if success, absent otherwise
   */
  public BlockMeta moveBlockMeta(BlockMeta blockMeta, TempBlockMeta tempBlockMeta) {
    StorageDir srcDir = blockMeta.getParentDir();
    StorageDir dstDir = tempBlockMeta.getParentDir();
    srcDir.removeBlockMeta(blockMeta);
    BlockMeta newBlockMeta =
        new DefaultBlockMeta(blockMeta.getBlockId(), blockMeta.getBlockSize(), dstDir);
    dstDir.removeTempBlockMeta(tempBlockMeta);
    dstDir.addBlockMeta(newBlockMeta);
    return newBlockMeta;
  }

  /**
   * Removes the metadata of a specific block.
   *
   * @param block the metadata of the block to remove
   */
  public void removeBlockMeta(BlockMeta block) {
    StorageDir dir = block.getParentDir();
    dir.removeBlockMeta(block);
  }

  /**
   * Modifies the size of a temp block.
   *
   * @param tempBlockMeta the temp block to modify
   * @param newSize new size in bytes
   */
  public void resizeTempBlockMeta(TempBlockMeta tempBlockMeta, long newSize) {
    StorageDir dir = tempBlockMeta.getParentDir();
    dir.resizeTempBlockMeta(tempBlockMeta, newSize);
  }

  /**
   * @return the storage tier mapping
   */
  public StorageTierAssoc getStorageTierAssoc() {
    return WORKER_STORAGE_TIER_ASSOC;
  }
}
