alluxio-master-2.x/core/common/src/test/java/alluxio/resource/DynamicResourcePoolTest.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.resource;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

import alluxio.Constants;
import alluxio.clock.ManualClock;
import alluxio.metrics.MetricsSystem;
import alluxio.util.ThreadFactoryUtils;

import com.codahale.metrics.Counter;
import org.junit.Test;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

public final class DynamicResourcePoolTest {

  /**
   * Resource class used to test {@link DynamicResourcePool}.
   */
  private static final class Resource {
    private Integer mInteger = 0;
    // Threshold for invalid resource.
    private static final int INVALID_RESOURCE = 10;

    /**
     * Constructor of Resource class.
     *
     * @param i the number represents the current capacity of Resource
     */
    public Resource(Integer i) {
      mInteger = i;
    }

    /**
     * Sets the number representing current capacity of Resource and returns Resource Object.
     *
     * @param i the value of member variable represents the current capacity of Resource
     * @return the Resource object
     */
    public Resource setInteger(Integer i) {
      mInteger = i;
      return this;
    }
  }

  /**
   * The subclass of DynamicResourcePool to be tested.
   */
  private static final class TestPool extends DynamicResourcePool<Resource> {
    private int mGcThresholdInSecs = 120;
    private int mCounter = 0;

    private static final ScheduledExecutorService GC_EXECUTOR =
        new ScheduledThreadPoolExecutor(5, ThreadFactoryUtils.build("TestPool-%d", true));

    /**
     * Constructor of TestPool class.
     *
     * @param options the Options object to set ScheduledExecutorService object
     * @param clock the object of Clock class
     */
    public TestPool(Options options, ManualClock clock) {
      super(options.setGcExecutor(GC_EXECUTOR));
      mClock = clock;
    }

    /**
     * Constructor of TestPool class.
     *
     * @param options the Options object to set ScheduledExecutorService object
     */
    public TestPool(Options options) {
      super(options.setGcExecutor(GC_EXECUTOR));
    }

    @Override
    protected Counter getMetricCounter() {
      return MetricsSystem.counter("Test.DynamicResourcePoolResourceCount");
    }

    @Override
    protected boolean shouldGc(ResourceInternal<Resource> resourceInternal) {
      return mClock.millis() - resourceInternal.getLastAccessTimeMs()
          >= (long) mGcThresholdInSecs * (long) Constants.SECOND_MS;
    }

    @Override
    protected boolean isHealthy(Resource resource) {
      return resource.mInteger < Resource.INVALID_RESOURCE;
    }

    @Override
    protected void closeResource(Resource resource) {
      resource.setInteger(Resource.INVALID_RESOURCE);
    }

    @Override
    protected Resource createNewResource() {
      return new Resource(mCounter++);
    }

    /**
     * Set the value representing the max value of Interval, managing when should Gc.
     *
     * @param gcThresholdInSecs the value of Gc Threshold Interval
     */
    public void setGcThresholdInSecs(int gcThresholdInSecs) {
      mGcThresholdInSecs = gcThresholdInSecs;
    }
  }

  /**
   * Tests the logic to acquire a resource when the pool is not full.
   */
  @Test
  public void acquireWithCapacity() throws Exception {
    TestPool pool = new TestPool(DynamicResourcePool.Options.defaultOptions());
    List<Resource> resourceList = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      Resource resource = pool.acquire();
      resourceList.add(resource);
      assertEquals(i, resource.mInteger.intValue());
    }

    for (Resource resource : resourceList) {
      pool.release(resource);
    }

    Set<Integer> resources = new HashSet<>();
    for (int i = 0; i < 3; i++) {
      Resource resource = pool.acquire();
      resources.add(resource.mInteger);
    }
    // Make sure we are not creating new resources.
    for (int i = 0; i < 3; i++) {
      assertTrue(resources.contains(i));
    }
  }

  /**
   * Tests if resources are acquired in a fifo manner when fifo is set to true.
   */
  @Test
  public void acquireFIFO() throws Exception {
    TestPool pool = new TestPool(DynamicResourcePool.Options.defaultOptions().setSelectionPolicy(
        DynamicResourcePool.SelectionPolicy.FIFO));
    List<Resource> resourceList = new ArrayList<>();
    for (int i = 0; i < 3; i++) {
      Resource resource = pool.acquire();
      resourceList.add(resource);
    }
    for (int i = 0; i < 3; i++) {
      pool.release(resourceList.get(i));
    }

    for (int iteration = 0; iteration < 10; ++iteration) {
      for (int i = 0; i < 3; ++i) {
        Resource resource = pool.acquire();
        assertEquals(resourceList.get(i), resource);
        pool.release(resource);
      }
    }
  }

  /**
   * Acquire without capacity.
   */
  @Test
  public void acquireWithoutCapacity() throws Exception {
    TestPool pool = new TestPool(DynamicResourcePool.Options.defaultOptions().setMaxCapacity(1));
    List<Resource> resourceList = new ArrayList<>();
    boolean timeout = false;
    try {
      Resource resource = pool.acquire();
      resourceList.add(resource);
      assertEquals(0, resource.mInteger.intValue());

      resource = pool.acquire(1, TimeUnit.SECONDS);
      resourceList.add(resource);
    } catch (TimeoutException e) {
      timeout = true;
    }
    assertEquals(1, resourceList.size());
    assertTrue(timeout);
  }

  /**
   * Tests the logic that invalid resource won't be acquired.
   */
  @Test
  public void UnhealthyResource() throws Exception {
    TestPool pool = new TestPool(DynamicResourcePool.Options.defaultOptions());
    Resource resource = pool.acquire();
    assertEquals(0, resource.mInteger.intValue());
    resource.setInteger(Resource.INVALID_RESOURCE);

    pool.release(resource);
    resource = pool.acquire();

    // The 0-th resource is not acquired because it is unhealthy.
    assertEquals(1, resource.mInteger.intValue());
  }

  /**
   * Tests the logic that the recently used resource is preferred.
   */
  @Test
  public void acquireRentlyUsed() throws Exception {
    ManualClock manualClock = new ManualClock();
    TestPool pool = new TestPool(DynamicResourcePool.Options.defaultOptions(), manualClock);
    List<Resource> resourceList = new ArrayList<>();
    resourceList.add(pool.acquire());
    resourceList.add(pool.acquire());
    resourceList.add(pool.acquire());

    pool.release(resourceList.get(2));
    pool.release(resourceList.get(0));

    manualClock.addTimeMs(1500);
    pool.release(resourceList.get(1));

    for (int i = 0; i < 10; i++) {
      Resource resource = pool.acquire();
      assertEquals(1, resource.mInteger.intValue());
      pool.release(resource);
    }
  }

  @Test
  public void gc() throws Exception {
    ManualClock manualClock = new ManualClock();
    TestPool pool = new TestPool(
        DynamicResourcePool.Options.defaultOptions().setGcIntervalMs(10).setInitialDelayMs(1),
        manualClock);
    pool.setGcThresholdInSecs(1);

    List<Resource> resourceList = new ArrayList<>();
    resourceList.add(pool.acquire());
    resourceList.add(pool.acquire());

    pool.release(resourceList.get(0));
    manualClock.addTimeMs(1001);

    // Sleep 1 second to make sure the GC has run.
    Thread.sleep(1000);

    // Resource 0 is gc-ed.
    assertEquals(2, pool.acquire().mInteger.intValue());
  }

  @Test
  public void multiClients() throws Exception {
    TestPool pool = new TestPool(DynamicResourcePool.Options.defaultOptions().setMaxCapacity(1));
    final Resource resource1 = pool.acquire();
    assertEquals(0, resource1.mInteger.intValue());

    class ReleaseThread extends Thread {
      private TestPool mPool;
      private Resource mResource;

      ReleaseThread(TestPool pool, Resource resource) {
        mPool = pool;
        mResource = resource;
      }

      @Override
      public void run() {
        try {
          // Sleep sometime to test wait logic.
          Thread.sleep(1000);
        } catch (InterruptedException e) {
          return;
        }
        mPool.release(mResource);
      }
    }

    ReleaseThread releaseThread = new ReleaseThread(pool, resource1);
    releaseThread.start();
    Resource resource2 = pool.acquire(2, TimeUnit.SECONDS);
    assertEquals(0, resource2.mInteger.intValue());
  }

  /**
   * Tests that an exception is thrown if the timestamps overflow and the method
   * terminate before 5 seconds.
   */
  @Test
  public void TimestampOverflow() {
    Callable<Resource> task = () -> {
      TestPool pool = new TestPool(DynamicResourcePool.Options.defaultOptions().setMaxCapacity(1));
      pool.acquire(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
      return pool.acquire(Long.MAX_VALUE, TimeUnit.MILLISECONDS);
    };
    ExecutorService executor = Executors.newFixedThreadPool(1);
    Future<Resource> future = executor.submit(task);
    boolean timeout = false;
    try {
      future.get(5000, TimeUnit.MILLISECONDS);
    } catch (Exception ex) {
      timeout = true;
    }
    assertTrue(timeout);
    assertFalse(future.isDone());
    future.cancel(true);
  }
}
