AdaptivePool.java

package org.robotninjas.util.pool;

import com.google.common.base.Ticker;
import org.apache.commons.pool2.BaseObjectPool;
import org.apache.commons.pool2.KeyedObjectPool;
import org.apache.commons.pool2.KeyedPooledObjectFactory;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

public class AdaptivePool<K, V> extends BaseObjectPool<V> {

  private final Map<K, Decaying> errorRates = new HashMap<>();
  private final Map<V, K> outstanding = new ConcurrentHashMap<>();
  private final KeyedObjectPool<K, V> pool;

  private final Comparator<Map.Entry<K, Decaying>> entryComparator =
      (o1, o2) -> Double.compare(o1.getValue().value(), o2.getValue().value());

  public AdaptivePool(List<K> keys, KeyedPooledObjectFactory<K, V> objectFactory) {

    for (K address : keys) {
      errorRates.put(address, new Decaying());
    }

    GenericKeyedObjectPool objectPool =
        new GenericKeyedObjectPool<>(objectFactory);
    objectPool.setBlockWhenExhausted(false);

    pool = objectPool;

  }

  @Override
  public V borrowObject() throws Exception {

    Iterator<K> sorted;
    sorted = errorRates
        .entrySet()
        .stream()
        .sorted(entryComparator)
        .map((x) -> x.getKey())
        .iterator();

    while (sorted.hasNext()) {
      K address = sorted.next();
      V borrowedObject;
      if ((borrowedObject = pool.borrowObject(address)) != null) {
        outstanding.put(borrowedObject, address);
        return borrowedObject;
      }
    }

    return null;

  }

  @Override
  public void returnObject(V obj) throws Exception {

    K key = outstanding.remove(obj);
    if (key != null) {
      pool.returnObject(key, obj);
    }

    throw new Exception("Object was not borrowed from this pool");

  }

  @Override
  public void invalidateObject(V obj) throws Exception {

    K key = outstanding.remove(obj);
    if (key != null) {
      pool.invalidateObject(key, obj);
      errorRates.get(key).inc();
    }

    throw new Exception("Object was not borrowed from this pool");

  }

  private static class Decaying {

    private final Ticker clock;
    private final double e;
    private final double r;
    private long t0;
    private double p;

    private Decaying(Ticker clock, double p, double e, double r) {
      this.clock = clock;
      this.p = p;
      this.e = e;
      this.r = r;
      this.t0 = clock.read();
    }

    private Decaying(double p, double e, double r) {
      this(Ticker.systemTicker(), p, e, r);
    }

    private Decaying() {
      this(0.0, Math.E, Math.log(0.5) / 10);
    }

    synchronized void inc(double d) {
      p = value() + d;
    }

    synchronized void inc() {
      inc(1.0);
    }

    synchronized double value() {
      long now = clock.read();
      long dt = now - t0;
      t0 = now;
      p = p * Math.pow(e, r * dt);
      return p;
    }

  }

}