[ALLUXIO-2226] Add a LocalFirstPolicy that without evict action

core/client/src/main/java/alluxio/client/file/policy/LocalFirstAvoidEvictionPolicy.java

@@ -0,0 +1,104 @@
+/*
+ * 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.client.file.policy;
+
+import alluxio.Configuration;
+import alluxio.PropertyKey;
+import alluxio.client.block.BlockWorkerInfo;
+import alluxio.util.network.NetworkAddressUtils;
+import alluxio.wire.WorkerNetAddress;
+
+import com.google.common.base.Objects;
+import com.google.common.collect.Lists;
+
+import java.util.Collections;
+import java.util.List;
+
+import javax.annotation.concurrent.ThreadSafe;
+/**
+ * A policy that returns local host first, and if the local worker doesn't have enough availability,
+ * it randomly picks a worker from the active workers list for each block write.
+ * If No worker meets the demands, return local host.
+ * USER_FILE_WRITE_AVOID_EVICTION_POLICY_RESERVED_BYTES is used to reserve some space of the worker
+ * to store the block, for the values mCapacityBytes minus mUsedBytes is not the available bytes.
+ */
+@ThreadSafe
+public final class LocalFirstAvoidEvictionPolicy implements FileWriteLocationPolicy {
+  private String mLocalHostName;
+  /**
+   * Constructs a {@link LocalFirstAvoidEvictionPolicy}.
+   */
+  public LocalFirstAvoidEvictionPolicy() {
+    mLocalHostName = NetworkAddressUtils.getLocalHostName();
+  }
+
+  @Override
+  public WorkerNetAddress getWorkerForNextBlock(Iterable<BlockWorkerInfo> workerInfoList,
+      long blockSizeBytes) {
+    // try the local host first
+    WorkerNetAddress localWorkerNetAddress = null;
+    for (BlockWorkerInfo workerInfo : workerInfoList) {
+      if (workerInfo.getNetAddress().getHost().equals(mLocalHostName)) {
+        localWorkerNetAddress = workerInfo.getNetAddress();
+        if (getAvailableBytes(workerInfo) >= blockSizeBytes) {
+          return localWorkerNetAddress;
+        }
+      }
+    }
+
+    // otherwise randomly pick a worker that has enough availability
+    List<BlockWorkerInfo> shuffledWorkers = Lists.newArrayList(workerInfoList);
+    Collections.shuffle(shuffledWorkers);
+    for (BlockWorkerInfo workerInfo : shuffledWorkers) {
+      if (getAvailableBytes(workerInfo) >= blockSizeBytes) {
+        return workerInfo.getNetAddress();
+      }
+    }
+    return localWorkerNetAddress;
+  }
+
+  /**
+   * @param workerInfo BlockWorkerInfo of the worker
+   * @return the available bytes of the worker
+   */
+  private long getAvailableBytes(BlockWorkerInfo workerInfo) {
+    long mUserFileWriteCapacityReserved = Configuration
+            .getBytes(PropertyKey.USER_FILE_WRITE_AVOID_EVICTION_POLICY_RESERVED_BYTES);
+    long mCapacityBytes = workerInfo.getCapacityBytes();
+    long mUsedBytes = workerInfo.getUsedBytes();
+    return mCapacityBytes - mUsedBytes - mUserFileWriteCapacityReserved;
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof LocalFirstAvoidEvictionPolicy)) {
+      return false;
+    }
+    LocalFirstAvoidEvictionPolicy that = (LocalFirstAvoidEvictionPolicy) o;
+    return Objects.equal(mLocalHostName, that.mLocalHostName);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(mLocalHostName);
+  }
+
+  @Override
+  public String toString() {
+    return Objects.toStringHelper(this)
+        .add("localHostName", mLocalHostName)
+        .toString();
+  }
+}

core/client/src/test/java/alluxio/client/file/policy/LocalFirstAvoidEvictionPolicyTest.java

@@ -0,0 +1,85 @@
+/*
+ * 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.client.file.policy;
+
+import alluxio.CommonTestUtils;
+import alluxio.Constants;
+import alluxio.client.block.BlockWorkerInfo;
+import alluxio.util.network.NetworkAddressUtils;
+import alluxio.wire.WorkerNetAddress;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Tests {@link LocalFirstAvoidEvictionPolicy}.
+ */
+public class LocalFirstAvoidEvictionPolicyTest {
+  private static final int PORT = 1;
+
+  /**
+   * Tests that the local host is returned first.
+   */
+  @Test
+  public void getLocalFirst() {
+    String localhostName = NetworkAddressUtils.getLocalHostName();
+    LocalFirstAvoidEvictionPolicy policy = new LocalFirstAvoidEvictionPolicy();
+    List<BlockWorkerInfo> workerInfoList = new ArrayList<>();
+    workerInfoList.add(new BlockWorkerInfo(new WorkerNetAddress().setHost("worker1")
+        .setRpcPort(PORT).setDataPort(PORT).setWebPort(PORT), Constants.GB, 0));
+    workerInfoList.add(new BlockWorkerInfo(new WorkerNetAddress().setHost(localhostName)
+        .setRpcPort(PORT).setDataPort(PORT).setWebPort(PORT), Constants.GB, 0));
+    Assert.assertEquals(localhostName,
+        policy.getWorkerForNextBlock(workerInfoList, Constants.MB).getHost());
+  }
+
+  /**
+   * Tests that another worker is picked in case the local host does not have enough space.
+   */
+  @Test
+  public void getOthersWhenNotEnoughSpaceOnLocal() {
+    String localhostName = NetworkAddressUtils.getLocalHostName();
+    LocalFirstAvoidEvictionPolicy policy = new LocalFirstAvoidEvictionPolicy();
+    List<BlockWorkerInfo> workerInfoList = new ArrayList<>();
+    workerInfoList.add(new BlockWorkerInfo(new WorkerNetAddress().setHost("worker1")
+        .setRpcPort(PORT).setDataPort(PORT).setWebPort(PORT), Constants.GB, 0));
+    workerInfoList.add(new BlockWorkerInfo(new WorkerNetAddress().setHost(localhostName)
+        .setRpcPort(PORT).setDataPort(PORT).setWebPort(PORT), Constants.MB, Constants.MB));
+    Assert.assertEquals("worker1",
+        policy.getWorkerForNextBlock(workerInfoList, Constants.MB).getHost());
+  }
+
+  /**
+   * Tests that local host is picked if none of the workers has enough availability.
+   */
+  @Test
+  public void getLocalWhenNoneHasSpace() {
+    String localhostName = NetworkAddressUtils.getLocalHostName();
+    LocalFirstAvoidEvictionPolicy policy = new LocalFirstAvoidEvictionPolicy();
+    List<BlockWorkerInfo> workerInfoList = new ArrayList<>();
+    workerInfoList.add(new BlockWorkerInfo(new WorkerNetAddress().setHost("worker1")
+        .setRpcPort(PORT).setDataPort(PORT).setWebPort(PORT), Constants.GB, Constants.MB));
+    workerInfoList.add(new BlockWorkerInfo(new WorkerNetAddress().setHost(localhostName)
+        .setRpcPort(PORT).setDataPort(PORT).setWebPort(PORT), Constants.GB, Constants.MB));
+    Assert.assertEquals(localhostName,
+        policy.getWorkerForNextBlock(workerInfoList, Constants.GB).getHost());
+  }
+
+  @Test
+  public void equalsTest() throws Exception {
+    CommonTestUtils.testEquals(LocalFirstAvoidEvictionPolicy.class);
+  }
+
+}

core/common/src/main/java/alluxio/PropertyKey.java

@@ -252,6 +252,8 @@ public enum PropertyKey {
       Name.USER_FILE_WORKER_CLIENT_POOL_GC_THRESHOLD_MS, 300 * Constants.SECOND_MS),
   USER_FILE_WRITE_LOCATION_POLICY(Name.USER_FILE_WRITE_LOCATION_POLICY,
       "alluxio.client.file.policy.LocalFirstPolicy"),
+  USER_FILE_WRITE_AVOID_EVICTION_POLICY_RESERVED_BYTES(
+      Name.USER_FILE_WRITE_AVOID_EVICTION_POLICY_RESERVED_BYTES, "0MB"),
   USER_FILE_WRITE_TYPE_DEFAULT(Name.USER_FILE_WRITE_TYPE_DEFAULT, "MUST_CACHE"),
   USER_HEARTBEAT_INTERVAL_MS(Name.USER_HEARTBEAT_INTERVAL_MS, 1000),
   USER_LINEAGE_ENABLED(Name.USER_LINEAGE_ENABLED, false),
@@ -661,6 +663,8 @@ public static final class Name {
         "alluxio.user.file.worker.client.pool.gc.threshold.ms";
     public static final String USER_FILE_WRITE_LOCATION_POLICY =
         "alluxio.user.file.write.location.policy.class";
+    public static final String USER_FILE_WRITE_AVOID_EVICTION_POLICY_RESERVED_BYTES =
+        "alluxio.user.file.write.avoid.eviction.policy.reserved.size.bytes";
     public static final String USER_FILE_WRITE_TYPE_DEFAULT = "alluxio.user.file.writetype.default";
     public static final String USER_HEARTBEAT_INTERVAL_MS = "alluxio.user.heartbeat.interval.ms";
     public static final String USER_LINEAGE_ENABLED = "alluxio.user.lineage.enabled";

docs/_data/table/cn/user-configuration.yml

@@ -28,6 +28,8 @@ alluxio.user.file.worker.client.threads:
   文件worker client从worker读取数据时使用的线程数目。
 alluxio.user.file.write.location.policy.class:
   选择worker进行写文件数据块时的默认定位机制。
+alluxio.user.file.write.avoid.eviction.policy.reserved.size.bytes:
+  当用户选择LocalFirstAvoidEvictionPolicy作为写文件数据块的定位机制时,用户需要配置worker预留一些数据量来保证数据的存储，默认是0MB。
 alluxio.user.file.readtype.default:
   创建Alluxio文件时的默认读类型。可选值为`CACHE_PROMOTE` (如果数据已经在Alluxio存储内，将其移动到最高存储层，如果数据需要从底层存储进行读取，将其写到本地Alluxio的最高存储层)、`CACHE` (如果数据需要从底层存储进行读取，将其写到本地Alluxio的最高存储层), `NO_CACHE` (数据不与Alluxio交互，如果是从Alluxio中进行读取，将不会发生数据块迁移或者剔除)。
 alluxio.user.file.writetype.default:

docs/_data/table/en/user-configuration.yml

@@ -30,6 +30,9 @@ alluxio.user.file.worker.client.threads:
   How many threads to use for file worker clients to read from workers.
 alluxio.user.file.write.location.policy.class:
   The default location policy for choosing workers for writing a file's blocks
+alluxio.user.file.write.avoid.eviction.policy.reserved.size.bytes:
+  The portion of space reserved in worker when user use the LocalFirstAvoidEvictionPolicy class
+  as file write location policy, default 0 MB.
 alluxio.user.file.readtype.default:
   Default read type when creating Alluxio files.
   Valid options are `CACHE_PROMOTE` (move data to highest tier if already in Alluxio storage,

docs/_data/table/user-configuration.csv

@@ -12,6 +12,7 @@ alluxio.user.file.master.client.threads,10
 alluxio.user.file.waitcompleted.poll.ms,1000
 alluxio.user.file.worker.client.threads,10
 alluxio.user.file.write.location.policy.class,alluxio.client.file.policy.LocalFirstPolicy
+alluxio.user.file.write.avoid.eviction.policy.reserved.size.bytes,0MB
 alluxio.user.file.readtype.default,CACHE_PROMOTE
 alluxio.user.file.writetype.default,MUST_CACHE
 alluxio.user.heartbeat.interval.ms,1000