/*
 * 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.meta;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

import com.google.common.base.Optional;

import tachyon.Constants;
import tachyon.conf.TachyonConf;
import tachyon.util.CommonUtils;

/**
 * Represents a tier of storage, for example memory or SSD. It serves as a container of
 * {@link StorageDir} which actually contains metadata information about blocks stored and space
 * used/available.
 * <p>
 * This class does not guarantee thread safety.
 */
public class StorageTier {
  private List<StorageDir> mDirs;
  private final int mTierAlias;

  public StorageTier(TachyonConf tachyonConf, int tierAlias) {
    mTierAlias = tierAlias;
    int level = tierAlias - 1;
    String tierDirPathConf =
        String.format(Constants.WORKER_TIERED_STORAGE_LEVEL_DIRS_PATH_FORMAT, level);
    String[] dirPaths = tachyonConf.get(tierDirPathConf, "/mnt/ramdisk").split(",");

    String tierDirCapacityConf =
        String.format(Constants.WORKER_TIERED_STORAGE_LEVEL_DIRS_QUOTA_FORMAT, level);
    String[] dirQuotas = tachyonConf.get(tierDirCapacityConf, "0").split(",");

    mDirs = new ArrayList<StorageDir>(dirPaths.length);

    for (int i = 0; i < dirPaths.length; i ++) {
      int index = i >= dirQuotas.length ? dirQuotas.length - 1 : i;
      long capacity = CommonUtils.parseSpaceSize(dirQuotas[index]);
      mDirs.add(new StorageDir(this, i, capacity, dirPaths[i]));
    }
  }

  public int getTierAlias() {
    return mTierAlias;
  }

  public long getCapacityBytes() {
    long capacityBytes = 0;
    for (StorageDir dir : mDirs) {
      capacityBytes += dir.getCapacityBytes();
    }
    return capacityBytes;
  }

  public long getAvailableBytes() {
    long availableBytes = 0;
    for (StorageDir dir : mDirs) {
      availableBytes += dir.getAvailableBytes();
    }
    return availableBytes;
  }

  public StorageDir getDir(int dirIndex) {
    return mDirs.get(dirIndex);
  }

  public Set<StorageDir> getStorageDirs() {
    return new HashSet<StorageDir>(mDirs);
  }

  public Optional<BlockMeta> getBlockMeta(long blockId) {
    for (StorageDir dir : mDirs) {
      Optional<BlockMeta> optionalBlock = dir.getBlockMeta(blockId);
      if (optionalBlock.isPresent()) {
        return optionalBlock;
      }
    }
    return Optional.absent();
  }
}

@@ -413,16 +413,17 @@
 se at%0A *
+ 
 %0A * http
@@ -463,16 +463,17 @@
 E-2.0%0A *
+ 
 %0A * Unle
@@ -824,16 +824,44 @@
 .meta;%0A%0A
+import java.io.IOException;%0A
 import j
@@ -956,49 +956,8 @@
 t;%0A%0A
-import com.google.common.base.Optional;%0A%0A
 impo
@@ -1345,69 +1345,252 @@
 %7B%0A  
-private List%3CStorageDir%3E mDirs;%0A  private final int mTierAlia
+/** Alias of this tier, e.g., memory tier is 1, SSD tier is 2 and HDD tier is 3 */%0A  private final int mTierAlias;%0A  /** Level of this tier in tiered storage, highest level is 0 */%0A  private final int mTierLevel;%0A  private List%3CStorageDir%3E mDir
 s;%0A%0A
@@ -1634,16 +1634,31 @@
 yonConf,
+ int tierLevel,
  int tie
@@ -1703,13 +1703,14 @@
     
-int l
+mTierL
 evel
@@ -1720,17 +1720,13 @@
 tier
-Alias - 1
+Level
 ;%0A  
@@ -1822,33 +1822,38 @@
 RS_PATH_FORMAT, 
-l
+mTierL
 evel);%0A    Strin
@@ -2041,17 +2041,22 @@
 FORMAT, 
-l
+mTierL
 evel);%0A 
@@ -2502,32 +2502,90 @@
 TierAlias;%0A  %7D%0A%0A
+  public int getTierLevel() %7B%0A    return mTierLevel;%0A  %7D%0A%0A
   public long ge
@@ -3115,24 +3115,229 @@
 %0A%0A  
-public Optional%3C
+/**%0A   * Returns the BlockMeta for a given blockId.%0A   *%0A   * @param blockId the ID of the block%0A   * @return BlockMeta%0A   * @throws IOException if no BlockMeta for this blockId is found in this StorageDir%0A   */%0A  public 
 Bloc
@@ -3341,17 +3341,16 @@
 lockMeta
-%3E
  getBloc
@@ -3368,16 +3368,35 @@
 blockId)
+ throws IOException
  %7B%0A    f
@@ -3435,43 +3435,55 @@
     
-Optional%3CBlockMeta%3E optionalBlock =
+if (dir.hasBlockMeta(blockId)) %7B%0A        return
  dir
@@ -3516,112 +3516,94 @@
     
-if (optionalBlock.isPresent()) %7B
+%7D
 %0A    
+%7D%0A
     
-return optionalBlock;%0A      %7D%0A    %7D%0A    return Optional.absent(
+throw new IOException(%22Cannot get blockId %22 + blockId + %22 in tier %22 + this
 );%0A 

Refactor StorageTier to use exceptions