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.resource;
  
  import alluxio.Constants;
  import alluxio.clock.SystemClock;
  
  import com.codahale.metrics.Counter;
  import com.google.common.base.Preconditions;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.IOException;
  import java.time.Clock;
  import java.util.ArrayDeque;
  import java.util.ArrayList;
  import java.util.Deque;
  import java.util.Iterator;
  import java.util.List;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.ScheduledExecutorService;
  import java.util.concurrent.ScheduledFuture;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.locks.Condition;
  import java.util.concurrent.locks.ReentrantLock;
  import javax.annotation.concurrent.GuardedBy;
  import javax.annotation.concurrent.ThreadSafe;
  
  /**
   * A dynamic pool that manages the resources. It clears old resources.
   * It accepts a min and max capacity.
   *
   * When acquiring resources, the most recently used resource is returned.
   *
   * @param <T> the type of the resource
   */
  @ThreadSafe
  public abstract class DynamicResourcePool<T> implements Pool<T> {
    /**
     * A policy specifying in what order to pick a resource item from a pool.
     */
    public enum SelectionPolicy {
      // first-in-first-out, use the hottest resource
      FIFO,
      // last-in-first-out, use the coldest resource
      LIFO,
    }
  
    private static final Logger LOG = LoggerFactory.getLogger(DynamicResourcePool.class);
  
    /**
     * A wrapper on the resource to include the last time at which it was used.
     *
     * @param <R> the resource type
     */
    protected class ResourceInternal<R> {
      /** The resource. */
      private R mResource;
  
      /** The last access time in ms. */
      private long mLastAccessTimeMs;
  
      /**
       * @param lastAccessTimeMs the last access time in ms
       */
      public void setLastAccessTimeMs(long lastAccessTimeMs) {
        mLastAccessTimeMs = lastAccessTimeMs;
      }
  
      /**
       * @return the last access time in ms
       */
      public long getLastAccessTimeMs() {
        return mLastAccessTimeMs;
      }
  
      /**
       * Creates a {@link ResourceInternal} instance.
       *
       * @param resource the resource
       */
      public ResourceInternal(R resource) {
        mResource = resource;
        mLastAccessTimeMs = mClock.millis();
      }
    }
  
    /**
     * Options to initialize a Dynamic resource pool.
     */
    public static final class Options {
      /** The max capacity. */
      private int mMaxCapacity = 1024;
  
      /** The min capacity. */
      private int mMinCapacity = 1;
  
      /** The initial delay. */
      private long mInitialDelayMs = 100;
  
      /** The gc interval. */
      private long mGcIntervalMs = 120L * Constants.SECOND_MS;
  
      /** The gc executor. */
      private ScheduledExecutorService mGcExecutor;
  
      /**
       * If set to true, when a resource needs to be taken from the pool, the last returned resource
       * will take priority. {@link #acquire()} tends to return a different object every time.
       * If set to false, the first returned resource will take priority.
       * {@link #acquire()} tends to reuse the most fresh resource if possible.
       */
      private SelectionPolicy mSelectionPolicy = SelectionPolicy.LIFO;
  
      /**
       * @return the max capacity
       */
      public int getMaxCapacity() {
        return mMaxCapacity;
      }
  
      /**
       * @return the min capacity
       */
      public int getMinCapacity() {
        return mMinCapacity;
      }
  
      /**
       * @return the initial delay
       */
      public long getInitialDelayMs() {
        return mInitialDelayMs;
      }
  
      /**
       * @return the gc interval
       */
      public long getGcIntervalMs() {
        return mGcIntervalMs;
      }
  
      /**
       * @return the gc executor
       */
      public ScheduledExecutorService getGcExecutor() {
        return mGcExecutor;
      }
  
      /**
       * @return the selection policy
       */
      public SelectionPolicy getSelectionPolicy() {
        return mSelectionPolicy;
      }
  
      /**
       * @param policy how to select a client from the pool
       * @return the updated object
       */
      public Options setSelectionPolicy(SelectionPolicy policy) {
        mSelectionPolicy = policy;
        return this;
      }
  
      /**
       * @param maxCapacity the max capacity
       * @return the updated object
       */
      public Options setMaxCapacity(int maxCapacity) {
        Preconditions.checkArgument(maxCapacity >= 1);
        mMaxCapacity = maxCapacity;
        return this;
      }
  
      /**
       * @param minCapacity the min capacity
       * @return the updated object
       */
      public Options setMinCapacity(int minCapacity) {
        Preconditions.checkArgument(minCapacity >= 0);
        mMinCapacity = minCapacity;
        return this;
      }
  
      /**
       * @param initialDelayMs the initial delay
       * @return the updated object
       */
      public Options setInitialDelayMs(long initialDelayMs) {
        Preconditions.checkArgument(initialDelayMs >= 0);
        mInitialDelayMs = initialDelayMs;
        return this;
      }
  
      /**
       * @param gcIntervalMs the gc interval
       * @return the updated object
       */
      public Options setGcIntervalMs(long gcIntervalMs) {
        Preconditions.checkArgument(gcIntervalMs > 0);
        mGcIntervalMs = gcIntervalMs;
        return this;
      }
  
      /**
       * @param gcExecutor the gc executor
       * @return updated object
       */
      public Options setGcExecutor(ScheduledExecutorService gcExecutor) {
        mGcExecutor = gcExecutor;
        return this;
      }
  
      private Options() {
      }  // prevents instantiation
  
      /**
       * @return the default option
       */
      public static Options defaultOptions() {
        return new Options();
      }
    }
  
    private final ReentrantLock mLock = new ReentrantLock();
    private final Condition mNotEmpty = mLock.newCondition();
  
    /** The max capacity. */
    private final int mMaxCapacity;
  
    /** The min capacity. */
    private final int mMinCapacity;
  
    /**
     * the selection policy of the resource pool. see {@link SelectionPolicy} for details
     */
    protected final SelectionPolicy mSelectionPolicy;
  
    // Tracks the resources that are available ordered by lastAccessTime (the head is
    // the most recently used resource).
    // These are the resources that acquire() will take.
    // This is always a subset of the other data structure mResources.
    @GuardedBy("mLock")
    private final Deque<ResourceInternal<T>> mAvailableResources;
  
    // Tracks all the resources that are not closed.
    // put/delete operations are guarded by "mLock" so that we can control its size to be within
    // a [min, max] range. mLock is reused for simplicity. A separate lock can be used if we see
    // any performance overhead.
    protected final ConcurrentHashMap<T, ResourceInternal<T>> mResources =
        new ConcurrentHashMap<>(32);
    private final Counter mCounter;
  
    // Thread to scan mAvailableResources to close those resources that are old.
    private ScheduledExecutorService mExecutor;
    private ScheduledFuture<?> mGcFuture;
  
    protected Clock mClock = new SystemClock();
  
    /**
     * Creates a dynamic pool instance.
     *
     * @param options the options
     */
    public DynamicResourcePool(Options options) {
      mExecutor = Preconditions.checkNotNull(options.getGcExecutor(), "executor");
      mCounter = Preconditions.checkNotNull(getMetricCounter(),
          "cannot find resource count metric for %s", getClass().getName());
      mMaxCapacity = options.getMaxCapacity();
      mMinCapacity = options.getMinCapacity();
      mSelectionPolicy = options.getSelectionPolicy();
      mAvailableResources = new ArrayDeque<>(Math.min(mMaxCapacity, 32));
      mGcFuture = mExecutor.scheduleAtFixedRate(() -> {
        List<T> resourcesToGc = new ArrayList<>();
  
        try {
          mLock.lock();
          if (mResources.size() <= mMinCapacity) {
            return;
          }
          int currentSize = mResources.size();
          Iterator<ResourceInternal<T>> iterator = mAvailableResources.iterator();
          while (iterator.hasNext()) {
            ResourceInternal<T> next = iterator.next();
            if (shouldGc(next)) {
              resourcesToGc.add(next.mResource);
              iterator.remove();
              mResources.remove(next.mResource);
              mCounter.dec();
              currentSize--;
              if (currentSize <= mMinCapacity) {
                break;
              }
            }
          }
        } finally {
          mLock.unlock();
        }
  
        for (T resource : resourcesToGc) {
          LOG.debug("Resource {} is garbage collected.", resource);
          try {
            closeResource(resource);
          } catch (IOException e) {
            LOG.warn("Failed to close resource {}.", resource, e);
          }
        }
      }, options.getInitialDelayMs(), options.getGcIntervalMs(), TimeUnit.MILLISECONDS);
    }
  
    protected abstract Counter getMetricCounter();
  
    /**
     * Acquires a resource of type {code T} from the pool.
     *
     * @return the acquired resource
     */
    @Override
    public T acquire() throws IOException {
      try {
        return acquire(100  /* no timeout */, TimeUnit.DAYS);
      } catch (TimeoutException e) {
        // Never should timeout in acquire().
        throw new RuntimeException(e);
      }
    }
  
    /**
     * Acquires a resource of type {code T} from the pool.
     *
     * This method is like {@link #acquire()}, but it will time out if an object cannot be
     * acquired before the specified amount of time.
     *
     * @param time an amount of time to wait
     * @param unit the unit to use for time
     * @return a resource taken from the pool
     * @throws TimeoutException if it fails to acquire because of time out
     * @throws IOException if the thread is interrupted while acquiring the resource
     */
    @Override
    public T acquire(long time, TimeUnit unit) throws TimeoutException, IOException {
      long endTimeMs = mClock.millis() + unit.toMillis(time);
  
      // Try to take a resource without blocking
      ResourceInternal<T> resource = poll();
      if (resource != null) {
        return checkHealthyAndRetry(resource.mResource, endTimeMs);
      }
  
      if (!isFull()) {
        // If the resource pool is empty but capacity is not yet full, create a new resource.
        T newResource = createNewResource();
        ResourceInternal<T> resourceInternal = new ResourceInternal<>(newResource);
        if (add(resourceInternal)) {
          return newResource;
        } else {
          closeResource(newResource);
        }
      }
  
      // Otherwise, try to take a resource from the pool, blocking if none are available.
      try {
        mLock.lock();
        while (true) {
          resource = poll();
          if (resource != null) {
            break;
          }
          long currTimeMs = mClock.millis();
          try {
            // one should use t1-t0<0, not t1<t0, because of the possibility of numerical overflow.
            // For further detail see: https://docs.oracle.com/javase/8/docs/api/java/lang/System.html
            if (endTimeMs - currTimeMs <= 0 || !mNotEmpty
                .await(endTimeMs - currTimeMs, TimeUnit.MILLISECONDS)) {
              throw new TimeoutException("Acquire resource times out.");
            }
          } catch (InterruptedException e) {
            // TODO(calvin): Propagate the interrupted exception instead of converting to IOException
            Thread.currentThread().interrupt();
            throw new IOException("Thread interrupted while acquiring client from pool: " + this);
          }
        }
      } finally {
        mLock.unlock();
      }
  
      return checkHealthyAndRetry(resource.mResource, endTimeMs);
    }
  
    /**
     * Releases the resource to the pool. It expects the resource to be released was acquired from
     * this pool.
     * {@link DynamicResourcePool#release(Object)} and {@link DynamicResourcePool#acquire()} must be
     * paired. Do not release the resource acquired multiple times. The behavior is undefined if
     * that happens.
     *
     * @param resource the resource to release
     */
    @Override
    public void release(T resource) {
      // We don't need to acquire mLock here because the resource is guaranteed not to be removed
      // if it is not available (i.e. not in mAvailableResources list).
      if (!mResources.containsKey(resource)) {
        throw new IllegalArgumentException(
            "Resource " + resource.toString() + " was not acquired from this resource pool.");
      }
      ResourceInternal<T> resourceInternal = mResources.get(resource);
      resourceInternal.setLastAccessTimeMs(mClock.millis());
      try {
        mLock.lock();
        mAvailableResources.addFirst(resourceInternal);
        mNotEmpty.signal();
      } finally {
        mLock.unlock();
      }
    }
  
    /**
     * Closes the pool and clears all the resources. The resource pool should not be used after this.
     */
    @Override
    public void close() throws IOException {
      try {
        mLock.lock();
        if (mAvailableResources.size() != mResources.size()) {
          LOG.warn("{} resources are not released when closing the resource pool.",
              mResources.size() - mAvailableResources.size());
        }
        for (ResourceInternal<T> resourceInternal : mAvailableResources) {
          closeResource(resourceInternal.mResource);
        }
        mAvailableResources.clear();
      } finally {
        mLock.unlock();
      }
      mGcFuture.cancel(true);
    }
  
    @Override
    public int size() {
      return mResources.size();
    }
  
    /**
     * @return true if the pool is full
     */
    private boolean isFull() {
      return mResources.size() >= mMaxCapacity;
    }
  
    /**
     * Adds a newly created resource to the pool. The resource is not available when it is added.
     *
     * @param resource
     * @return true if the resource is successfully added
     */
    private boolean add(ResourceInternal<T> resource) {
      try {
        mLock.lock();
        if (mResources.size() >= mMaxCapacity) {
          return false;
        } else {
          mResources.put(resource.mResource, resource);
          mCounter.inc();
          return true;
        }
      } finally {
        mLock.unlock();
      }
    }
  
    /**
     * Removes an existing resource from the pool.
     *
     * @param resource
     */
    private void remove(T resource) {
      try {
        mLock.lock();
        mResources.remove(resource);
        mCounter.dec();
      } finally {
        mLock.unlock();
      }
    }
  
    /**
     * @return the most recently used resource and null if there are no free resources
     */
    private ResourceInternal<T> poll() {
      try {
        mLock.lock();
        switch (mSelectionPolicy) {
          case FIFO:
            return mAvailableResources.pollLast();
          case LIFO:
            return mAvailableResources.pollFirst();
          default:
            throw new UnsupportedOperationException(
                "Policy " + mSelectionPolicy + " is not supported!");
        }
      } finally {
        mLock.unlock();
      }
    }
  
    /**
     * Checks whether the resource is healthy. If not retry. When this called, the resource
     * is not in mAvailableResources.
     *
     * @param resource the resource to check
     * @param endTimeMs the end time to wait till
     * @return the resource
     * @throws TimeoutException if it times out to wait for a resource
     */
    private T checkHealthyAndRetry(T resource, long endTimeMs) throws TimeoutException, IOException {
      if (isHealthy(resource)) {
        return resource;
      } else {
        LOG.debug("Clearing unhealthy resource {}.", resource);
        remove(resource);
        closeResource(resource);
        return acquire(endTimeMs - mClock.millis(), TimeUnit.MILLISECONDS);
      }
    }
  
    // The following functions should be overridden by implementations.
  
    /**
     * @param resourceInternal the resource to check
     * @return true if the resource should be garbage collected
     */
    protected abstract boolean shouldGc(ResourceInternal<T> resourceInternal);
  
    /**
     * Checks whether a resource is healthy or not.
     *
     * @param resource the resource to check
     * @return true if the resource is healthy
     */
    protected abstract boolean isHealthy(T resource);
  
    /**
     * Closes the resource. After this, the resource should not be used. It is not guaranteed that
     * the resource is closed after the function returns.
     *
     * @param resource the resource to close
     */
    protected abstract void closeResource(T resource) throws IOException;
  
    /**
     * Creates a new resource.
     *
     * @return the newly created resource
     */
    protected abstract T createNewResource() throws IOException;
  }
  