Add workqueue support

Author: cizezsy

extended/src/main/java/io/kubernetes/client/extended/workqueue/DelayingWorkQueue.java

@@ -0,0 +1,92 @@
+package io.kubernetes.client.extended.workqueue;
+
+import java.time.Duration;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.Temporal;
+import java.util.concurrent.*;
+import javax.annotation.Nonnull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * DelayingWorkQueue can Add an item at a later time. This makes it easier to requeue items after
+ * failures without ending up in a hot-loop.
+ */
+public class DelayingWorkQueue<T> extends WorkQueue<T> {
+  private static final Logger log = LoggerFactory.getLogger(DelayingWorkQueue.class);
+
+  private DelayQueue<DelayEntry> delayItems;
+
+  public DelayingWorkQueue(ExecutorService executorService) {
+    this.delayItems = new DelayQueue<>();
+    executorService.submit(this::waitingLoop);
+  }
+
+  public DelayingWorkQueue() {
+    this(Executors.newSingleThreadExecutor());
+  }
+
+  public synchronized void addAfter(T item, long amount, ChronoUnit timeUnit) {
+    addAfter(item, Duration.of(amount, timeUnit));
+  }
+
+  /**
+   * AddAfter adds an item to the workqueue after the indicated duration has passed
+   *
+   * @param item Item will be added to the queue
+   * @param duration Item will be added to the queue after the duration has passed
+   */
+  public synchronized void addAfter(T item, Duration duration) {
+    if (shuttingDown()) {
+      return;
+    }
+
+    if (duration.isZero() || duration.isNegative()) {
+      add(item);
+      return;
+    }
+
+    delayItems.add(new DelayEntry(item, duration));
+  }
+
+  private void waitingLoop() {
+    try {
+      while (true) {
+        if (shuttingDown()) {
+          break;
+        }
+
+        DelayEntry delayEntry = delayItems.take();
+        add(delayEntry.data);
+      }
+    } catch (InterruptedException e) {
+      if (log.isDebugEnabled()) {
+        log.debug("Waiting loop is stopped, reason: {}", e);
+      }
+    }
+  }
+
+  private class DelayEntry implements Delayed {
+
+    private T data;
+    private Temporal expectedEndTime;
+
+    public DelayEntry(T data, Duration delay) {
+      this.data = data;
+
+      expectedEndTime = delay.addTo(Instant.now());
+    }
+
+    @Override
+    public long getDelay(@Nonnull TimeUnit unit) {
+      Duration duration = Duration.between(Instant.now(), expectedEndTime);
+      return unit.convert(duration.toMillis(), TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(@Nonnull Delayed o) {
+      return Long.compare(getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
+    }
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/RateLimitingQueue.java

@@ -0,0 +1,31 @@
+package io.kubernetes.client.extended.workqueue;
+
+import io.kubernetes.client.extended.workqueue.ratelimiter.RateLimiter;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+public class RateLimitingQueue<T> extends DelayingWorkQueue<T> {
+  private RateLimiter<T> rateLimiter;
+
+  public RateLimitingQueue(RateLimiter<T> rateLimiter, ExecutorService executorService) {
+    super(executorService);
+
+    this.rateLimiter = rateLimiter;
+  }
+
+  public RateLimitingQueue(RateLimiter<T> rateLimiter) {
+    this(rateLimiter, Executors.newSingleThreadExecutor());
+  }
+
+  public void addRateLimited(T item) {
+    addAfter(item, rateLimiter.when(item));
+  }
+
+  public void forget(T item) {
+    rateLimiter.forget(item);
+  }
+
+  public int numRequests(T item) {
+    return rateLimiter.numRequests(item);
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/WorkQueue.java

@@ -0,0 +1,122 @@
+package io.kubernetes.client.extended.workqueue;
+
+import java.util.HashSet;
+import java.util.LinkedList;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * workqueue provides a simple queue that supports the following features:
+ *
+ * <ul>
+ *   <li>Fair: items processed in the order in which they are added.
+ *   <li>Stingy: a single item will not be processed multiple times concurrently, and if an item is
+ *       added multiple times before it can be processed, it will only be processed once.
+ *   <li>Multiple consumers and producers. In particular, it is allowed for an item to be reenqueued
+ *       while it is being processed.
+ *   <li>Shutdown notifications.
+ * </ul>
+ */
+public class WorkQueue<T> {
+  // queue defines the order in which we will work on items. Every
+  // element of queue should be in the dirty set and not in the
+  // processing set.
+  private LinkedList<T> queue;
+
+  // dirty defines all of the items that need to be processed.
+  private HashSet<T> dirty;
+
+  // Things that are currently being processed are in the processing set.
+  // These things may be simultaneously in the dirty set. When we finish
+  // processing something and remove it from this set, we'll check if
+  // it's in the dirty set, and if so, add it to the queue.
+  private HashSet<T> processing;
+
+  private AtomicBoolean shuttingDown;
+
+  public WorkQueue() {
+    queue = new LinkedList<>();
+    dirty = new HashSet<>();
+    processing = new HashSet<>();
+
+    shuttingDown = new AtomicBoolean(false);
+  }
+
+  /**
+   * Add marks item as needing processing.
+   *
+   * @param item item need to be processed.
+   */
+  public synchronized void add(T item) {
+    if (shuttingDown.get()) {
+      return;
+    }
+
+    if (dirty.contains(item)) {
+      return;
+    }
+
+    dirty.add(item);
+    if (processing.contains(item)) {
+      return;
+    }
+
+    queue.add(item);
+    this.notify();
+  }
+
+  /**
+   * Get blocks until it can return an item to be processed.
+   *
+   * @return item to be processed
+   * @throws InterruptedException if queue is shutting down
+   */
+  public synchronized T get() throws InterruptedException {
+    while (queue.size() == 0 && !shuttingDown.get()) {
+      this.wait();
+    }
+
+    if (queue.size() == 0) {
+      // We must be shutting down
+      throw new InterruptedException("WorkQueue is shutting down");
+    }
+
+    T item = queue.poll();
+
+    processing.add(item);
+    dirty.remove(item);
+
+    return item;
+  }
+
+  public synchronized int size() {
+    return queue.size();
+  }
+
+  /**
+   * Done marks item as done processing, and if it has been marked as dirty again while it was being
+   * processed, it will be re-added to the queue for re-processing.
+   *
+   * @param item Item that marked as done processing
+   */
+  public synchronized void done(T item) {
+    processing.remove(item);
+    if (dirty.contains(item)) {
+      queue.add(item);
+      this.notify();
+    }
+  }
+
+  /**
+   * shutDown will cause queue to ignore all new items added to it. All threads that waiting to get
+   * item will be interrupted.
+   */
+  public synchronized void shutDown() {
+    shuttingDown.set(true);
+    this.notifyAll();
+  }
+
+  /** @return true is this queue is shutting down */
+  public synchronized boolean shuttingDown() {
+    return shuttingDown.get();
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/ItemExponentialFailureRateLimiter.java

@@ -0,0 +1,51 @@
+package io.kubernetes.client.extended.workqueue.ratelimiter;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * ItemExponentialFailureRateLimiter does a simple baseDelay*10<sup>num-failures</sup> limit dealing
+ * with max failures and expiration are up to the caller
+ */
+public class ItemExponentialFailureRateLimiter<T> implements RateLimiter<T> {
+
+  private Duration baseDelay;
+  private Duration maxDelay;
+
+  private Map<T, Integer> failures;
+
+  public ItemExponentialFailureRateLimiter(Duration baseDelay, Duration maxDelay) {
+    this.baseDelay = baseDelay;
+    this.maxDelay = maxDelay;
+
+    failures = new HashMap<>();
+  }
+
+  @Override
+  public synchronized Duration when(T item) {
+    int exp = failures.getOrDefault(item, 0);
+    failures.put(item, exp + 1);
+
+    double backOff = baseDelay.toNanos() * Math.pow(2, exp);
+    if (backOff > Long.MAX_VALUE) {
+      return maxDelay;
+    }
+
+    if (backOff > maxDelay.toNanos()) {
+      return maxDelay;
+    }
+
+    return Duration.ofNanos((long) backOff);
+  }
+
+  @Override
+  public synchronized void forget(T item) {
+    failures.remove(item);
+  }
+
+  @Override
+  public synchronized int numRequests(T item) {
+    return failures.getOrDefault(item, 0);
+  }
+}

extended/src/main/java/io/kubernetes/client/extended/workqueue/ratelimiter/ItemFastSlowRateLimiter.java

@@ -0,0 +1,48 @@
+package io.kubernetes.client.extended.workqueue.ratelimiter;
+
+import java.time.Duration;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * ItemFastSlowRateLimiter does a quick retry for a certain number of attempts, then a slow retry
+ * after that
+ */
+public class ItemFastSlowRateLimiter<T> implements RateLimiter<T> {
+
+  private Map<T, Integer> failures;
+
+  private Duration fastDelay;
+  private Duration slowDelay;
+  private int maxFastAttempts;
+
+  public ItemFastSlowRateLimiter(Duration fastDelay, Duration slowDelay, int maxFastAttempts) {
+    this.fastDelay = fastDelay;
+    this.slowDelay = slowDelay;
+    this.maxFastAttempts = maxFastAttempts;
+
+    failures = new HashMap<>();
+  }
+
+  @Override
+  public synchronized Duration when(T item) {
+    int attempts = failures.getOrDefault(item, 0);
+    failures.put(item, attempts + 1);
+
+    if (attempts <= maxFastAttempts) {
+      return fastDelay;
+    }
+
+    return slowDelay;
+  }
+
+  @Override
+  public synchronized void forget(T item) {
+    failures.remove(item);
+  }
+
+  @Override
+  public synchronized int numRequests(T item) {
+    return failures.getOrDefault(item, 0);
+  }
+}