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

import java.io.IOException;

import com.google.common.base.Preconditions;

import tachyon.worker.block.BlockStoreLocation;
import tachyon.worker.block.BlockMetadataManager;
import tachyon.worker.block.meta.StorageDir;
import tachyon.worker.block.meta.StorageTier;
import tachyon.worker.block.meta.TempBlockMeta;

/**
 * An allocator that allocates a block in the storage dir with most free space.
 */
public class MaxFreeAllocator implements Allocator {
  private final BlockMetadataManager mMetaManager;

  public MaxFreeAllocator(BlockMetadataManager metadata) {
    mMetaManager = Preconditions.checkNotNull(metadata);
  }

  @Override
  public TempBlockMeta allocateBlock(long userId, long blockId, long blockSize,
      BlockStoreLocation location) throws IOException {

    StorageDir candidateDir = null;
    long maxFreeBytes = blockSize;

    if (location.equals(BlockStoreLocation.anyTier())) {
      for (StorageTier tier : mMetaManager.getTiers()) {
        for (StorageDir dir : tier.getStorageDirs()) {
          if (dir.getAvailableBytes() >= maxFreeBytes) {
            maxFreeBytes = dir.getAvailableBytes();
            candidateDir = dir;
          }
        }
      }
    } else if (location.equals(BlockStoreLocation.anyDirInTier(location.tierAlias()))) {
      StorageTier tier = mMetaManager.getTier(location.tierAlias());
      for (StorageDir dir : tier.getStorageDirs()) {
        if (dir.getAvailableBytes() >= maxFreeBytes) {
          maxFreeBytes = dir.getAvailableBytes();
          candidateDir = dir;
        }
      }
    } else {
      StorageTier tier = mMetaManager.getTier(location.tierAlias());
      StorageDir dir = tier.getDir(location.dir());
      if (dir.getAvailableBytes() >= blockSize) {
        candidateDir = dir;
      }
    }

    return candidateDir != null ? new TempBlockMeta(userId, blockId, blockSize, candidateDir)
        : null;
  }
}

@@ -1223,16 +1223,110 @@
 e space.
+ It always allocates%0A * to the highest tier if the requested block store location is any tier.
 %0A */%0Apub
@@ -1696,17 +1696,16 @@
 ption %7B%0A
-%0A
     Stor
@@ -1735,43 +1735,8 @@
 ull;
-%0A    long maxFreeBytes = blockSize;
 %0A%0A  
@@ -1859,108 +1859,95 @@
     
-for (StorageDir dir : tier.getStorageDirs()) %7B%0A          if (dir.getAvailableBytes() %3E= maxFreeBytes
+candidateDir = getCandidateDirInTier(tier, blockSize);%0A        if (candidateDir != null
 ) %7B%0A
@@ -1960,93 +1960,75 @@
     
-  
-maxFreeBytes = dir.getAvailableBytes();%0A            candidateDir = dir;%0A          %7D
+return new TempBlockMeta(userId, blockId, blockSize, candidateDir);
 %0A   
@@ -2210,207 +2210,62 @@
     
-for (StorageDir dir : tier.getStorageDirs()) %7B%0A        if (dir.getAvailableBytes() %3E= maxFreeBytes) %7B%0A          maxFreeBytes = dir.getAvailableBytes();%0A          candidateDir = dir;%0A        %7D%0A      %7D
+candidateDir = getCandidateDirInTier(tier, blockSize);
 %0A   
@@ -2601,11 +2601,636 @@
 ull;%0A  %7D
+%0A%0A  /**%0A   * Find a directory in a tier that has max free space and is able to store the block.%0A   *%0A   * @param tier the storage tier%0A   * @param blockSize the size of block in bytes%0A   * @return the storage directory if found, null otherwise%0A   */%0A  private StorageDir getCandidateDirInTier(StorageTier tier, long blockSize) %7B%0A    StorageDir candidateDir = null;%0A    long maxFreeBytes = blockSize - 1;%0A    for (StorageDir dir : tier.getStorageDirs()) %7B%0A      if (dir.getAvailableBytes() %3E maxFreeBytes) %7B%0A        maxFreeBytes = dir.getAvailableBytes();%0A        candidateDir = dir;%0A      %7D%0A    %7D%0A    return candidateDir;%0A  %7D
 %0A%7D%0A

Optimize max free allocator policy