content: /*
   * 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.checkState;
  import static java.lang.String.format;
  
  import alluxio.conf.Configuration;
  import alluxio.conf.PropertyKey;
  import alluxio.exception.ExceptionMessage;
  import alluxio.exception.runtime.AlluxioRuntimeException;
  import alluxio.exception.runtime.BlockDoesNotExistRuntimeException;
  import alluxio.exception.runtime.ResourceExhaustedRuntimeException;
  import alluxio.exception.status.DeadlineExceededException;
  import alluxio.master.block.BlockId;
  import alluxio.resource.LockResource;
  import alluxio.util.io.FileUtils;
  import alluxio.worker.block.allocator.Allocator;
  import alluxio.worker.block.annotator.BlockIterator;
  import alluxio.worker.block.annotator.BlockOrder;
  import alluxio.worker.block.io.BlockReader;
  import alluxio.worker.block.io.BlockWriter;
  import alluxio.worker.block.io.DelegatingBlockReader;
  import alluxio.worker.block.io.LocalFileBlockReader;
  import alluxio.worker.block.io.MetricAccountingBlockReader;
  import alluxio.worker.block.io.StoreBlockReader;
  import alluxio.worker.block.io.StoreBlockWriter;
  import alluxio.worker.block.management.DefaultStoreLoadTracker;
  import alluxio.worker.block.management.ManagementTaskCoordinator;
  import alluxio.worker.block.meta.BlockMeta;
  import alluxio.worker.block.meta.StorageDir;
  import alluxio.worker.block.meta.StorageDirView;
  import alluxio.worker.block.meta.StorageTier;
  import alluxio.worker.block.meta.TempBlockMeta;
  
  import com.google.common.annotations.VisibleForTesting;
  import com.google.common.base.Preconditions;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.Closeable;
  import java.io.IOException;
  import java.nio.channels.FileChannel;
  import java.nio.file.Files;
  import java.nio.file.NoSuchFileException;
  import java.nio.file.Path;
  import java.nio.file.Paths;
  import java.nio.file.attribute.BasicFileAttributes;
  import java.text.MessageFormat;
  import java.util.Collections;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.CopyOnWriteArrayList;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.locks.Lock;
  import java.util.concurrent.locks.ReentrantReadWriteLock;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * This class represents an object store that manages all the blocks in the local tiered storage.
   * This store exposes simple public APIs to operate blocks. Inside this store, it creates an
   * Allocator to decide where to put a new block, an Evictor to decide where to evict a stale block,
   * a BlockMetadataManager to maintain the status of the tiered storage, and a LockManager to
   * coordinate read/write on the same block.
   * <p>
   * This class is thread-safe, using the following lock hierarchy to ensure thread-safety:
   * <ul>
   * <li>Any block-level operation (e.g., read, move or remove) on an existing block must acquire a
   * block lock for this block via {@link TieredBlockStore#mLockManager}. This block lock is a
   * read/write lock, guarding both the metadata operations and the following I/O on this block. It
   * coordinates different threads (clients) when accessing the same block concurrently.</li>
   * <li>Any metadata operation (read or write) must go through {@link TieredBlockStore#mMetaManager}
   * and guarded by {@link TieredBlockStore#mMetadataLock}. This is also a read/write lock and
   * coordinates different threads (clients) when accessing the shared data structure for metadata.
   * </li>
   * <li>Method {@link #createBlock} does not acquire the block lock, because it only creates a
   * temp block which is only visible to its writer before committed (thus no concurrent access).</li>
   * <li>Method {@link #abortBlock(long, long)} does not acquire the block lock, because only
   * temporary blocks can be aborted, and they are only visible to their writers (thus no concurrent
   * access).
   * <li>Eviction is done in {@link #freeSpace} and it is on the basis of best effort. For
   * operations that may trigger this eviction (e.g., move, create, requestSpace), retry is used</li>
   * </ul>
   */
  @ThreadSafe
  public class TieredBlockStore implements LocalBlockStore {
    private static final Logger LOG = LoggerFactory.getLogger(TieredBlockStore.class);
    private static final Long REMOVE_BLOCK_TIMEOUT_MS = 60_000L;
    private static final long FREE_AHEAD_BYTETS =
        Configuration.getBytes(PropertyKey.WORKER_TIERED_STORE_FREE_AHEAD_BYTES);
    private final BlockMetadataManager mMetaManager;
    private final BlockLockManager mLockManager;
    private final Allocator mAllocator;
  
    private final List<BlockStoreEventListener> mBlockStoreEventListeners =
        new CopyOnWriteArrayList<>();
  
    /** A set of pinned inodes fetched from the master. */
    private final Set<Long> mPinnedInodes = new HashSet<>();
  
    /** Lock to guard metadata operations. */
    private final ReentrantReadWriteLock mMetadataLock = new ReentrantReadWriteLock();
  
    /** ReadLock provided by {@link #mMetadataLock} to guard metadata read operations. */
    private final Lock mMetadataReadLock = mMetadataLock.readLock();
  
    /** WriteLock provided by {@link #mMetadataLock} to guard metadata write operations. */
    private final Lock mMetadataWriteLock = mMetadataLock.writeLock();
  
    /** Management task coordinator. */
    private final ManagementTaskCoordinator mTaskCoordinator;
  
    /**
     * Creates a new instance of {@link TieredBlockStore}.
     */
    public TieredBlockStore() {
      this(BlockMetadataManager.createBlockMetadataManager(), new BlockLockManager());
    }
  
    /**
     * Creates a new instance of {@link TieredBlockStore}.
     *
     * @param metaManager the block metadata manager
     * @param lockManager the lock manager
     */
    @VisibleForTesting
    public TieredBlockStore(BlockMetadataManager metaManager,
        BlockLockManager lockManager) {
      mMetaManager = metaManager;
      mLockManager = lockManager;
  
      BlockIterator blockIterator = mMetaManager.getBlockIterator();
      // Register listeners required by the block iterator.
      for (BlockStoreEventListener listener : blockIterator.getListeners()) {
        registerBlockStoreEventListener(listener);
      }
  
      BlockMetadataEvictorView initManagerView = new BlockMetadataEvictorView(mMetaManager,
          Collections.emptySet(), Collections.emptySet());
      mAllocator = Allocator.Factory.create(initManagerView);
      if (mAllocator instanceof BlockStoreEventListener) {
        registerBlockStoreEventListener((BlockStoreEventListener) mAllocator);
      }
  
      // Initialize and start coordinator.
      mTaskCoordinator = new ManagementTaskCoordinator(this, mMetaManager,
          new DefaultStoreLoadTracker(), this::getUpdatedView);
      mTaskCoordinator.start();
    }
  
    @Override
    public Optional<BlockLock> pinBlock(long sessionId, long blockId) {
      LOG.debug("pinBlock: sessionId={}, blockId={}", sessionId, blockId);
      BlockLock lock = mLockManager.acquireBlockLock(sessionId, blockId, BlockLockType.READ);
      if (hasBlockMeta(blockId)) {
        return Optional.of(lock);
      }
      lock.close();
      return Optional.empty();
    }
  
    @Override
    public BlockWriter createBlockWriter(long sessionId, long blockId) {
      LOG.debug("getBlockWriter: sessionId={}, blockId={}", sessionId, blockId);
      // NOTE: a temp block is supposed to only be visible by its own writer, unnecessary to acquire
      // block lock here since no sharing
      // TODO(bin): Handle the case where multiple writers compete for the same block.
      checkBlockDoesNotExist(blockId);
      return new StoreBlockWriter(checkAndGetTempBlockMeta(sessionId, blockId));
    }
  
    @Override
    public BlockReader createBlockReader(long sessionId, long blockId, long offset)
        throws IOException {
      LOG.debug("createBlockReader: sessionId={}, blockId={}, offset={}",
          sessionId, blockId, offset);
      Closeable blockLock = mLockManager.acquireBlockLock(sessionId, blockId, BlockLockType.READ);
      Optional<BlockMeta> blockMeta;
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        blockMeta = mMetaManager.getBlockMeta(blockId);
      }
      if (!blockMeta.isPresent()) {
        blockLock.close();
        throw new BlockDoesNotExistRuntimeException(blockId);
      }
      BlockMeta block = blockMeta.get();
      try {
        validateBlockIntegrityForRead(block);
      } catch (IllegalStateException validationError) {
        LOG.warn("Block {} is corrupted, removing it: {}",
            blockId, validationError.getMessage());
        // in case of a corrupted block, remove it and propagate the exception
        // release the read lock because removeBlockInternal needs a write lock on the same block
        blockLock.close();
        // at this point we are not holding any lock, so two threads may attempt to remove the same
        // block concurrently. This is fine as long as removeBlockInternal is no-op for a
        // non-existing block.
        try {
          removeBlockInternal(sessionId, blockId, REMOVE_BLOCK_TIMEOUT_MS);
          for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
            synchronized (listener) {
              listener.onRemoveBlockByWorker(blockId);
              listener.onRemoveBlock(blockId, block.getBlockLocation());
            }
          }
        } catch (Exception removeBlockError) {
          LOG.warn("Failed to remove a corrupted block {}", blockId, removeBlockError);
          validationError.addSuppressed(removeBlockError);
        }
        throw new BlockDoesNotExistRuntimeException(blockId, validationError);
      }
  
      try {
        LocalFileBlockReader reader = new StoreBlockReader(sessionId, block);
        ((FileChannel) reader.getChannel()).position(offset);
        accessBlock(sessionId, blockId);
        BlockReader mareader = new MetricAccountingBlockReader(reader);
        return new DelegatingBlockReader(mareader, blockLock);
      } catch (Exception e) {
        blockLock.close();
        throw new IOException(format("Failed to get local block reader, sessionId=%d, "
            + "blockId=%d, offset=%d", sessionId, blockId, offset), e);
      }
    }
  
    /**
     * Validates the integrity of the block for reading:
     * 1. the block file should exist
     * 2. the length of the block file should match its BlockMeta
     * If any of the above does not hold, this can be a result of corrupted block files
     * due to faulty storage hardware, manual manipulation of the block files by admin,
     * or a bug where the block was pre-maturely committed when it was not done writing.
     *
     * @param blockMeta the block meta acquired from meta data manager
     * @throws IllegalStateException if the block is deemed corrupted
     */
    public static void validateBlockIntegrityForRead(BlockMeta blockMeta)
        throws IllegalStateException {
      final long blockId = blockMeta.getBlockId();
      final Path blockPath = Paths.get(blockMeta.getPath());
      final BasicFileAttributes blockFileAttrs;
      try {
        blockFileAttrs = Files.readAttributes(blockPath, BasicFileAttributes.class);
      } catch (NoSuchFileException e) {
        throw new IllegalStateException(String.format(
            "Block %s exists in block meta but actual physical block file %s does not exist",
            blockId, blockPath));
      } catch (IOException e) {
        // cannot read file attributes, possibly due to bad permission or bad file type
        LOG.debug("Cannot read file attributes for block {}", blockId, e);
        throw new IllegalStateException(String.format(
            "Cannot read attributes of file %s for block %s during validation", blockId, blockPath));
      }
      // need to check if file is a regular file, as for directories and device files the file length
      // is unspecified
      if (!blockFileAttrs.isRegularFile()) {
        throw new IllegalStateException(String.format(
            "Block file %s for block %s is not a regular file", blockPath, blockId));
      }
      final long actualLength = blockFileAttrs.size();
      final long expectedLength = blockMeta.getBlockSize();
      // check if the actual file length matches the expected length from block meta
      if (actualLength != expectedLength) {
        LOG.debug("Block {} is expected to be {} bytes, "
            + "but the actual block file length is {}", blockId, expectedLength, actualLength);
        // Note: we only errors out on 0-sized blocks which are definitely not correct
        // but if the size is not 0, we treat it as valid
        if (actualLength == 0) {
          throw new IllegalStateException(String.format(
              "Block %s exists in block meta but the size from block meta does not match that of "
                  + "the block file %s, expected block size = %d, actual block file length = %d",
              blockId, blockPath, expectedLength, actualLength));
        }
      }
    }
  
    @Override
    public TempBlockMeta createBlock(long sessionId, long blockId, AllocateOptions options) {
      LOG.debug("createBlock: sessionId={}, blockId={}, options={}", sessionId, blockId, options);
      TempBlockMeta tempBlockMeta = createBlockMetaInternal(sessionId, blockId, true, options);
      createBlockFile(tempBlockMeta.getPath());
      return tempBlockMeta;
    }
  
    // TODO(bin): Make this method to return a snapshot.
    @Override
    public Optional<BlockMeta> getVolatileBlockMeta(long blockId) {
      LOG.debug("getVolatileBlockMeta: blockId={}", blockId);
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        return mMetaManager.getBlockMeta(blockId);
      }
    }
  
    @Override
    public Optional<TempBlockMeta> getTempBlockMeta(long blockId) {
      LOG.debug("getTempBlockMeta: blockId={}", blockId);
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        return mMetaManager.getTempBlockMeta(blockId);
      }
    }
  
    @Override
    public void commitBlock(long sessionId, long blockId, boolean pinOnCreate) {
      LOG.debug("commitBlock: sessionId={}, blockId={}, pinOnCreate={}",
          sessionId, blockId, pinOnCreate);
      try (BlockLock lock = mLockManager.acquireBlockLock(sessionId, blockId, BlockLockType.WRITE)) {
        BlockStoreLocation loc = commitBlockInternal(sessionId, blockId, pinOnCreate);
        for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
          synchronized (listener) {
            listener.onCommitBlockToLocal(blockId, loc);
          }
        }
      }
    }
  
    @Override
    public BlockLock commitBlockLocked(long sessionId, long blockId, boolean pinOnCreate) {
      LOG.debug("commitBlock: sessionId={}, blockId={}, pinOnCreate={}",
          sessionId, blockId, pinOnCreate);
      BlockLock lock = mLockManager.acquireBlockLock(sessionId, blockId, BlockLockType.WRITE);
      BlockStoreLocation loc;
      try {
        loc = commitBlockInternal(sessionId, blockId, pinOnCreate);
      } catch (RuntimeException e) {
        lock.close();
        throw e;
      }
      for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
        synchronized (listener) {
          listener.onCommitBlockToLocal(blockId, loc);
        }
      }
      return lock;
    }
  
    @Override
    public void abortBlock(long sessionId, long blockId) {
      LOG.debug("abortBlock: sessionId={}, blockId={}", sessionId, blockId);
      abortBlockInternal(sessionId, blockId);
      for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
        synchronized (listener) {
          listener.onAbortBlock(blockId);
        }
      }
    }
  
    @Override
    public void requestSpace(long sessionId, long blockId, long additionalBytes) {
      LOG.debug("requestSpace: sessionId={}, blockId={}, additionalBytes={}", sessionId, blockId,
          additionalBytes);
      if (additionalBytes <= 0) {
        return;
      }
      // NOTE: a temp block is only visible to its own writer, unnecessary to acquire
      // block lock here since no sharing
      try (LockResource r = new LockResource(mMetadataWriteLock)) {
        TempBlockMeta tempBlockMeta = checkAndGetTempBlockMeta(sessionId, blockId);
        BlockStoreLocation location = tempBlockMeta.getBlockLocation();
        StorageDirView allocationDir = allocateSpace(sessionId,
            AllocateOptions.forRequestSpace(additionalBytes, location));
        checkState(allocationDir.toBlockStoreLocation().equals(location),
            format("Allocation error: location enforcement failed for location: %s",
                allocationDir.toBlockStoreLocation()));
        // Increase the size of this temp block
        mMetaManager.resizeTempBlockMeta(tempBlockMeta,
            tempBlockMeta.getBlockSize() + additionalBytes);
      }
    }
  
    @Override
    public void moveBlock(long sessionId, long blockId, AllocateOptions moveOptions)
        throws IOException {
      LOG.debug("moveBlock: sessionId={}, blockId={}, options={}", sessionId,
          blockId, moveOptions);
      BlockMeta meta = getVolatileBlockMeta(blockId).orElseThrow(
          () -> new IllegalStateException(ExceptionMessage.BLOCK_META_NOT_FOUND.getMessage(blockId)));
      if (meta.getBlockLocation().belongsTo(moveOptions.getLocation())) {
        return;
      }
      // Execute the block move if necessary
      MoveBlockResult result = moveBlockInternal(sessionId, blockId, moveOptions);
      if (result.getSuccess()) {
        for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
          synchronized (listener) {
            listener.onMoveBlockByClient(blockId, result.getSrcLocation(),
                result.getDstLocation());
          }
        }
        return;
      }
      throw new ResourceExhaustedRuntimeException(
          ExceptionMessage.NO_SPACE_FOR_BLOCK_MOVE.getMessage(moveOptions.getLocation(), blockId),
          false);
    }
  
    @Override
    public void removeBlock(long sessionId, long blockId) throws IOException {
      LOG.debug("removeBlock: sessionId={}, blockId={}", sessionId, blockId);
      Optional<BlockMeta> blockMeta = removeBlockInternal(
          sessionId, blockId, REMOVE_BLOCK_TIMEOUT_MS);
      for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
        synchronized (listener) {
          listener.onRemoveBlockByClient(blockId);
          blockMeta.ifPresent(meta -> listener.onRemoveBlock(
              blockId, meta.getBlockLocation()));
        }
      }
    }
  
    @VisibleForTesting
    Optional<BlockMeta> removeBlockInternal(long sessionId, long blockId, long timeoutMs)
        throws IOException {
      Optional<BlockLock> optionalLock =
          mLockManager.tryAcquireBlockLock(sessionId, blockId, BlockLockType.WRITE,
              timeoutMs, TimeUnit.MILLISECONDS);
      if (!optionalLock.isPresent()) {
        throw new DeadlineExceededException(
            format("Can not acquire lock to remove block %d for session %d after %d ms",
                blockId, sessionId, REMOVE_BLOCK_TIMEOUT_MS));
      }
  
      try (BlockLock lock = optionalLock.get();
          LockResource r = new LockResource(mMetadataWriteLock)) {
        if (mMetaManager.hasTempBlockMeta(blockId)) {
          throw new IllegalStateException(
              ExceptionMessage.REMOVE_UNCOMMITTED_BLOCK.getMessage(blockId));
        }
        Optional<BlockMeta> blockMeta = mMetaManager.getBlockMeta(blockId);
        if (blockMeta.isPresent()) {
          removeBlockFileAndMeta(blockMeta.get());
        }
        return blockMeta;
      }
    }
  
    @Override
    public void accessBlock(long sessionId, long blockId) {
      LOG.debug("accessBlock: sessionId={}, blockId={}", sessionId, blockId);
      Optional<BlockMeta> blockMeta;
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        blockMeta = mMetaManager.getBlockMeta(blockId);
      }
      if (blockMeta.isPresent()) {
        for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
          synchronized (listener) {
            listener.onAccessBlock(blockId);
            listener.onAccessBlock(blockId, blockMeta.get().getBlockLocation());
          }
        }
      }
    }
  
    @Override
    public void cleanupSession(long sessionId) {
      LOG.debug("cleanupSession: sessionId={}", sessionId);
      // Release all locks the session is holding.
      mLockManager.cleanupSession(sessionId);
  
      // Collect a list of temp blocks the given session owns and abort all of them with best effort
      List<TempBlockMeta> tempBlocksToRemove;
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        tempBlocksToRemove = mMetaManager.getSessionTempBlocks(sessionId);
      }
      for (TempBlockMeta tempBlockMeta : tempBlocksToRemove) {
        try {
          LOG.warn("Clean up expired temporary block {} from session {}.", tempBlockMeta.getBlockId(),
              sessionId);
          abortBlockInternal(sessionId, tempBlockMeta.getBlockId());
        } catch (Exception e) {
          LOG.error("Failed to cleanup tempBlock {}", tempBlockMeta.getBlockId(), e);
        }
      }
    }
  
    @Override
    public boolean hasBlockMeta(long blockId) {
      LOG.debug("hasBlockMeta: blockId={}", blockId);
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        return mMetaManager.hasBlockMeta(blockId);
      }
    }
  
    @Override
    public boolean hasTempBlockMeta(long blockId) {
      LOG.debug("hasBlockMeta: blockId={}", blockId);
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        return mMetaManager.hasTempBlockMeta(blockId);
      }
    }
  
    @Override
    public BlockStoreMeta getBlockStoreMeta() {
      // Removed DEBUG logging because this is very noisy
      // LOG.debug("getBlockStoreMeta:");
      BlockStoreMeta storeMeta;
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        storeMeta = mMetaManager.getBlockStoreMeta();
      }
      return storeMeta;
    }
  
    @Override
    public BlockStoreMeta getBlockStoreMetaFull() {
      // Removed DEBUG logging because this is very noisy
      // LOG.debug("getBlockStoreMetaFull:");
      BlockStoreMeta storeMeta;
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        storeMeta = mMetaManager.getBlockStoreMetaFull();
      }
      return storeMeta;
    }
  
    @Override
    public void registerBlockStoreEventListener(BlockStoreEventListener listener) {
      LOG.debug("registerBlockStoreEventListener: listener={}", listener);
      mBlockStoreEventListeners.add(listener);
    }
  
    /**
     * Checks if block id is a temporary block and owned by session id. This method must be enclosed
     * by {@link #mMetadataLock}.
     *
     * @param sessionId the id of session
     * @param blockId the id of block
     */
    private TempBlockMeta checkAndGetTempBlockMeta(long sessionId, long blockId) {
      Optional<TempBlockMeta> tempBlockMeta;
      try (LockResource r = new LockResource(mMetadataReadLock)) {
        tempBlockMeta = mMetaManager.getTempBlockMeta(blockId);
      }
      checkState(tempBlockMeta.isPresent(),
          ExceptionMessage.TEMP_BLOCK_META_NOT_FOUND.getMessage(blockId));
      checkState(tempBlockMeta.get().getSessionId() == sessionId,
          ExceptionMessage.BLOCK_ID_FOR_DIFFERENT_SESSION.getMessage(blockId,
              tempBlockMeta.get().getSessionId(), sessionId));
      return tempBlockMeta.get();
    }
  
    private void checkBlockDoesNotExist(long blockId) {
      checkState(!hasBlockMeta(blockId),
          ExceptionMessage.TEMP_BLOCK_ID_COMMITTED.getMessage(blockId));
    }
  
    private void checkTempBlockDoesNotExist(long blockId) {
      checkState(!hasTempBlockMeta(blockId), MessageFormat
          .format("Temp blockId {0,number,#} is not available, because it already exists", blockId));
    }
  
    /**
     * Aborts a temp block.
     *
     * @param sessionId the id of session
     * @param blockId the id of block
     */
    private void abortBlockInternal(long sessionId, long blockId) {
      checkBlockDoesNotExist(blockId);
      TempBlockMeta tempBlockMeta = checkAndGetTempBlockMeta(sessionId, blockId);
  
      // The metadata lock is released during heavy IO. The temp block is private to one session, so
      // we do not lock it.
      FileUtils.delete(tempBlockMeta.getPath());
      try (LockResource r = new LockResource(mMetadataWriteLock)) {
        mMetaManager.abortTempBlockMeta(tempBlockMeta);
      }
    }
  
    /**
     * Commits a temp block.
     *
     * @param sessionId the id of session
     * @param blockId the id of block
     * @param pinOnCreate is block pinned on create
     * @return destination location to move the block
     */
    @VisibleForTesting
    BlockStoreLocation commitBlockInternal(long sessionId, long blockId,
        boolean pinOnCreate) {
      if (mMetaManager.hasBlockMeta(blockId)) {
        LOG.debug("Block {} has been in block store, this could be a retry due to master-side RPC "
            + "failure", blockId);
        return mMetaManager.getBlockMeta(blockId).get().getBlockLocation();
      }
      // When committing TempBlockMeta, the final BlockMeta calculates the block size according to
      // the actual file size of this TempBlockMeta. Therefore, commitTempBlockMeta must happen
      // after moving actual block file to its committed path.
      TempBlockMeta tempBlockMeta = checkAndGetTempBlockMeta(sessionId, blockId);
      String srcPath = tempBlockMeta.getPath();
      String dstPath = tempBlockMeta.getCommitPath();
      BlockStoreLocation loc = tempBlockMeta.getBlockLocation();
  
      // Heavy IO is guarded by block lock but not metadata lock. This may throw IOException.
      try {
        FileUtils.move(srcPath, dstPath);
      } catch (IOException e) {
        // TODO(jianjian) move IOException handling once we figure out how MoveResult works
        throw AlluxioRuntimeException.from(e);
      }
  
      try (LockResource r = new LockResource(mMetadataWriteLock)) {
        mMetaManager.commitTempBlockMeta(tempBlockMeta);
      }
  
      // Check if block is pinned on commit
      if (pinOnCreate) {
        addToPinnedInodes(BlockId.getFileId(blockId));
      }
  
      return loc;
    }
  
    private StorageDirView allocateSpace(long sessionId, AllocateOptions options) {
      StorageDirView dirView;
      BlockMetadataView allocatorView =
          new BlockMetadataAllocatorView(mMetaManager, options.canUseReservedSpace());
      // Convenient way to break on failure cases, no intention to loop
      while (true) {
        if (options.isForceLocation()) {
          // Try allocating from given location. Skip the review because the location is forced.
          dirView = mAllocator.allocateBlockWithView(options.getSize(),
              options.getLocation(), allocatorView, true);
          if (dirView != null) {
            return dirView;
          }
          if (options.isEvictionAllowed()) {
            LOG.debug("Free space for block expansion: freeing {} bytes on {}. ",
                    options.getSize(), options.getLocation());
            freeSpace(sessionId, options.getSize(), options.getSize(), options.getLocation());
            // Block expansion are forcing the location. We do not want the review's opinion.
            dirView = mAllocator.allocateBlockWithView(options.getSize(),
                options.getLocation(), allocatorView.refreshView(), true);
            if (LOG.isDebugEnabled()) {
              LOG.debug("Allocation after freeing space for block expansion, {}", dirView == null
                  ? "no available dir." : "available bytes in dir: " + dirView.getAvailableBytes());
            }
            if (dirView == null) {
              LOG.error("Target tier: {} has no evictable space to store {} bytes for session: {}",
                  options.getLocation(), options.getSize(), sessionId);
              break;
            }
          } else {
            // We are not evicting in the target tier so having no available space just
            // means the tier is currently full.
            LOG.warn("Target tier: {} has no available space to store {} bytes for session: {}",
                options.getLocation(), options.getSize(), sessionId);
            break;
          }
        } else {
          // Try allocating from given location. This may be rejected by the review logic.
          dirView = mAllocator.allocateBlockWithView(options.getSize(),
              options.getLocation(), allocatorView, false);
          if (dirView != null) {
            return dirView;
          }
          LOG.debug("Allocate to anyTier for {} bytes on {}", options.getSize(),
                  options.getLocation());
          dirView = mAllocator.allocateBlockWithView(options.getSize(),
              BlockStoreLocation.anyTier(), allocatorView, false);
          if (dirView != null) {
            return dirView;
          }
  
          if (options.isEvictionAllowed()) {
            // There is no space left on worker.
            // Free more than requested by configured free-ahead size.
            long toFreeBytes = options.getSize() + FREE_AHEAD_BYTETS;
            LOG.debug("Allocation on anyTier failed. Free space for {} bytes on anyTier",
                    toFreeBytes);
            freeSpace(sessionId, options.getSize(), toFreeBytes,
                BlockStoreLocation.anyTier());
            // Skip the review as we want the allocation to be in the place we just freed
            dirView = mAllocator.allocateBlockWithView(options.getSize(),
                BlockStoreLocation.anyTier(), allocatorView.refreshView(), true);
            if (LOG.isDebugEnabled()) {
              LOG.debug("Allocation after freeing space for block creation, {} ", dirView == null
                  ? "no available dir." : "available bytes in dir: " + dirView.getAvailableBytes());
            }
          }
        }
        if (dirView == null) {
          break;
        }
        return dirView;
      }
      throw new ResourceExhaustedRuntimeException(
          format("Allocation failure. Options: %s. Error:", options), false);
    }
  
    /**
     * Creates a temp block meta only if allocator finds available space. This method will not trigger
     * any eviction.
     *
     * @param sessionId session id
     * @param blockId block id
     * @param newBlock true if this temp block is created for a new block
     * @param options block allocation options
     * @return a temp block created if successful
     */
    private TempBlockMeta createBlockMetaInternal(long sessionId, long blockId, boolean newBlock,
        AllocateOptions options) {
      if (newBlock) {
        checkBlockDoesNotExist(blockId);
        checkTempBlockDoesNotExist(blockId);
      }
      try (LockResource r = new LockResource(mMetadataWriteLock)) {
        // NOTE: a temp block is supposed to be visible for its own writer,
        // unnecessary to acquire block lock here since no sharing.
        // Allocate space.
        StorageDirView dirView = allocateSpace(sessionId, options);
  
        // TODO(carson): Add tempBlock to corresponding storageDir and remove the use of
        // StorageDirView.createTempBlockMeta.
        TempBlockMeta tempBlock = dirView.createTempBlockMeta(sessionId, blockId, options.getSize());
        // Add allocated temp block to metadata manager. This should never fail if allocator
        // correctly assigns a StorageDir.
        mMetaManager.addTempBlockMeta(tempBlock);
        return tempBlock;
      }
    }
  
    /**
     * Free space is the entry for immediate block deletion in order to open up space for
     * new or ongoing blocks.
     *
     * - New blocks creations will not try to free space until all tiers are out of space.
     * - Ongoing blocks could end up freeing space oftenly, when the file's origin location is
     * low on space.
     *
     * This method is synchronized in order to prevent race in its only client, allocations.
     * If not synchronized, new allocations could steal space reserved by ongoing ones.
     * Removing synchronized requires implementing retries to this call along with an optimal
     * locking strategy for fairness.
     *
     * TODO(ggezer): Remove synchronized.
     *
     * @param sessionId the session id
     * @param minContiguousBytes the minimum amount of contiguous free space in bytes
     * @param minAvailableBytes the minimum amount of free space in bytes
     * @param location the location to free space
     */
    @VisibleForTesting
    public synchronized void freeSpace(long sessionId, long minContiguousBytes,
        long minAvailableBytes, BlockStoreLocation location) {
      LOG.debug("freeSpace: sessionId={}, minContiguousBytes={}, minAvailableBytes={}, location={}",
          sessionId, minAvailableBytes, minAvailableBytes, location);
      // TODO(ggezer): Too much memory pressure when pinned-inodes list is large.
      BlockMetadataEvictorView evictorView = getUpdatedView();
      boolean contiguousSpaceFound = false;
      boolean availableBytesFound = false;
  
      int blocksIterated = 0;
      int blocksRemoved = 0;
      int spaceFreed = 0;
  
      // List of all dirs that belong to the given location.
      List<StorageDirView> dirViews = evictorView.getDirs(location);
  
      Iterator<Long> evictionCandidates = mMetaManager.getBlockIterator()
          .getIterator(location, BlockOrder.NATURAL);
      while (true) {
        // Check if minContiguousBytes is satisfied.
        if (!contiguousSpaceFound) {
          for (StorageDirView dirView : dirViews) {
            if (dirView.getAvailableBytes() >= minContiguousBytes) {
              contiguousSpaceFound = true;
              break;
            }
          }
        }
  
        // Check minAvailableBytes is satisfied.
        if (!availableBytesFound) {
          if (evictorView.getAvailableBytes(location) >= minAvailableBytes) {
            availableBytesFound = true;
          }
        }
  
        if (contiguousSpaceFound && availableBytesFound) {
          break;
        }
  
        if (!evictionCandidates.hasNext()) {
          break;
        }
  
        long blockToDelete = evictionCandidates.next();
        blocksIterated++;
        if (evictorView.isBlockEvictable(blockToDelete)) {
          Optional<BlockMeta> optionalBlockMeta = mMetaManager.getBlockMeta(blockToDelete);
          if (!optionalBlockMeta.isPresent()) {
            LOG.warn("Failed to evict blockId {}, it could be already deleted", blockToDelete);
            continue;
          }
          BlockMeta blockMeta = optionalBlockMeta.get();
          removeBlockFileAndMeta(blockMeta);
          blocksRemoved++;
          for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
            synchronized (listener) {
              listener.onRemoveBlockByWorker(blockMeta.getBlockId());
              listener.onRemoveBlock(blockMeta.getBlockId(),
                  blockMeta.getBlockLocation());
            }
          }
          spaceFreed += blockMeta.getBlockSize();
        }
      }
  
      if (!contiguousSpaceFound || !availableBytesFound) {
        throw new ResourceExhaustedRuntimeException(
            format("Failed to free %d bytes space at location %s. "
                    + "Min contiguous requested: %d, Min available requested: %d, "
                    + "Blocks iterated: %d, Blocks removed: %d, Space freed: %d",
                minAvailableBytes, location.tierAlias(), minContiguousBytes, minAvailableBytes,
                blocksIterated, blocksRemoved, spaceFreed), false);
      }
    }
  
    /**
     * Gets the most updated view with most recent information on pinned inodes, and currently locked
     * blocks.
     *
     * @return {@link BlockMetadataEvictorView}, an updated view with most recent information
     */
    private BlockMetadataEvictorView getUpdatedView() {
      // TODO(calvin): Update the view object instead of creating new one every time.
      synchronized (mPinnedInodes) {
        return new BlockMetadataEvictorView(mMetaManager, mPinnedInodes,
            mLockManager.getLockedBlocks());
      }
    }
  
    /**
     * Moves a block to new location only if allocator finds available space in newLocation. This
     * method will not trigger any eviction. Returns {@link MoveBlockResult}.
     *
     * @param sessionId session id
     * @param blockId block id
     * @param moveOptions the allocate options for the move
     * @return the resulting information about the move operation
     */
    private MoveBlockResult moveBlockInternal(long sessionId, long blockId,
        AllocateOptions moveOptions) throws IOException {
      try (BlockLock lock =  mLockManager.acquireBlockLock(sessionId, blockId, BlockLockType.WRITE)) {
        checkTempBlockDoesNotExist(blockId);
        BlockMeta srcBlockMeta;
        try (LockResource r = new LockResource(mMetadataReadLock)) {
          srcBlockMeta = mMetaManager.getBlockMeta(blockId).orElseThrow(() ->
            new IllegalStateException(ExceptionMessage.BLOCK_META_NOT_FOUND.getMessage(blockId)));
        }
  
        BlockStoreLocation srcLocation = srcBlockMeta.getBlockLocation();
        String srcFilePath = srcBlockMeta.getPath();
        long blockSize = srcBlockMeta.getBlockSize();
        // Update moveOptions with the block size.
        moveOptions.setSize(blockSize);
        if (srcLocation.belongsTo(moveOptions.getLocation())) {
          return new MoveBlockResult(true, blockSize, srcLocation, srcLocation);
        }
  
        TempBlockMeta dstTempBlock;
        try {
          dstTempBlock = createBlockMetaInternal(sessionId, blockId, false, moveOptions);
        } catch (Exception e) {
          return new MoveBlockResult(false, blockSize, null, null);
        }
  
        // When `newLocation` is some specific location, the `newLocation` and the `dstLocation` are
        // just the same; while for `newLocation` with a wildcard significance, the `dstLocation`
        // is a specific one with specific tier and dir which belongs to newLocation.
        BlockStoreLocation dstLocation = dstTempBlock.getBlockLocation();
  
        // When the dstLocation belongs to srcLocation, simply abort the tempBlockMeta just created
        // internally from the newLocation and return success with specific block location.
        if (dstLocation.belongsTo(srcLocation)) {
          mMetaManager.abortTempBlockMeta(dstTempBlock);
          return new MoveBlockResult(true, blockSize, srcLocation, dstLocation);
        }
        String dstFilePath = dstTempBlock.getCommitPath();
  
        // Heavy IO is guarded by block lock but not metadata lock. This may throw IOException.
        FileUtils.move(srcFilePath, dstFilePath);
  
        try (LockResource r = new LockResource(mMetadataWriteLock)) {
          // If this metadata update fails, we panic for now.
          // TODO(bin): Implement rollback scheme to recover from IO failures.
          mMetaManager.moveBlockMeta(srcBlockMeta, dstTempBlock);
        }
        return new MoveBlockResult(true, blockSize, srcLocation, dstLocation);
      }
    }
  
    /**
     * Removes a block physically and from metadata.
     *
     * @param blockMeta block metadata
     */
    private void removeBlockFileAndMeta(BlockMeta blockMeta) {
      FileUtils.deleteIfExists(blockMeta.getPath());
      mMetaManager.removeBlockMeta(blockMeta);
    }
  
    /**
     * Creates a file to represent a block denoted by the given block path. This file will be owned
     * by the Alluxio worker but have 777 permissions so processes under users different from the
     * user that launched the Alluxio worker can read and write to the file. The tiered storage
     * directory has the sticky bit so only the worker user can delete or rename files it creates.
     *
     * @param blockPath the block path to create
     */
    // TODO(peis): Consider using domain socket to avoid setting the permission to 777.
    private static void createBlockFile(String blockPath) {
      FileUtils.createBlockPath(blockPath,
          Configuration.getString(PropertyKey.WORKER_DATA_FOLDER_PERMISSIONS));
      FileUtils.createFile(blockPath);
      FileUtils.changeLocalFileToFullPermission(blockPath);
      LOG.debug("Created new file block, block path: {}", blockPath);
    }
  
    /**
     * Updates the pinned blocks.
     *
     * @param inodes a set of ids inodes that are pinned
     */
    @Override
    public void updatePinnedInodes(Set<Long> inodes) {
      LOG.debug("updatePinnedInodes: inodes={}", inodes);
      synchronized (mPinnedInodes) {
        mPinnedInodes.clear();
        mPinnedInodes.addAll(Preconditions.checkNotNull(inodes));
      }
    }
  
    /**
     * Add a single inode to set of pinned ids.
     *
     * @param inode an inode that is pinned
     */
    private void addToPinnedInodes(Long inode) {
      LOG.debug("addToPinnedInodes: inode={}", inode);
      synchronized (mPinnedInodes) {
        mPinnedInodes.add(Preconditions.checkNotNull(inode));
      }
    }
  
    @Override
    public void removeInaccessibleStorage() {
      try (LockResource r = new LockResource(mMetadataWriteLock)) {
        for (StorageTier tier : mMetaManager.getTiers()) {
          for (StorageDir dir : tier.getStorageDirs()) {
            String path = dir.getDirPath();
            if (!FileUtils.isStorageDirAccessible(path)) {
              LOG.error("Storage check failed for path {}. The directory will be excluded.", path);
              removeDir(dir);
            }
          }
        }
      }
    }
  
    /**
     * Removes a storage directory.
     *
     * @param dir storage directory to be removed
     */
    public void removeDir(StorageDir dir) {
      // TODO(feng): Add a command for manually removing directory
      try (LockResource r = new LockResource(mMetadataWriteLock)) {
        String tierAlias = dir.getParentTier().getTierAlias();
        dir.getParentTier().removeStorageDir(dir);
        for (BlockStoreEventListener listener : mBlockStoreEventListeners) {
          synchronized (listener) {
            dir.getBlockIds().forEach(listener::onBlockLost);
            listener.onStorageLost(tierAlias, dir.getDirPath());
            listener.onStorageLost(dir.toBlockStoreLocation());
          }
        }
      }
    }
  
    @Override
    public void close() throws IOException {
      mTaskCoordinator.close();
    }
  
    /**
     * A wrapper on necessary info after a move block operation.
     */
    private static class MoveBlockResult {
      /** Whether this move operation succeeds. */
      private final boolean mSuccess;
      /** Size of this block in bytes. */
      private final long mBlockSize;
      /** Source location of this block to move. */
      private final BlockStoreLocation mSrcLocation;
      /** Destination location of this block to move. */
      private final BlockStoreLocation mDstLocation;
  
      /**
       * Creates a new instance of {@link MoveBlockResult}.
       *
       * @param success success indication
       * @param blockSize block size
       * @param srcLocation source location
       * @param dstLocation destination location
       */
      MoveBlockResult(boolean success, long blockSize, BlockStoreLocation srcLocation,
          BlockStoreLocation dstLocation) {
        mSuccess = success;
        mBlockSize = blockSize;
        mSrcLocation = srcLocation;
        mDstLocation = dstLocation;
      }
  
      /**
       * @return the success indicator
       */
      boolean getSuccess() {
        return mSuccess;
      }
  
      /**
       * @return the block size
       */
      long getBlockSize() {
        return mBlockSize;
      }
  
      /**
       * @return the source location
       */
      BlockStoreLocation getSrcLocation() {
        return mSrcLocation;
      }
  
      /**
       * @return the destination location
       */
      BlockStoreLocation getDstLocation() {
        return mDstLocation;
      }
    }
  }
  