This repository has been archived by the owner on Apr 24, 2023. It is now read-only.
-
Notifications
You must be signed in to change notification settings - Fork 63
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Parallel watch processing. * Bug fix. * Bugfix. * Make the shard count and max threads a parameter. Co-authored-by: Scott Crosby <scrosby@twosigma.com>
- Loading branch information
Showing
4 changed files
with
164 additions
and
13 deletions.
There are no files selected for viewing
131 changes: 131 additions & 0 deletions
131
scheduler/java/com/twosigma/cook/kubernetes/ParallelWatchQueue.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,131 @@ | ||
| package com.twosigma.cook.kubernetes; | ||
|
|
||
| import java.util.ArrayDeque; | ||
| import java.util.ArrayList; | ||
| import java.util.Deque; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Semaphore; | ||
| import java.util.concurrent.locks.Lock; | ||
| import java.util.concurrent.locks.ReentrantLock; | ||
|
|
||
| /** | ||
| * Implement a processor that can process watch events from k8s in parallel. | ||
| * Events are respresented as Runnable's (presmuably as clojures over some | ||
| * underlying event state). Events on the same shard are processed sequentially. | ||
| * Events on different shards are processed in parallel on the supplied | ||
| * ExecutorService. Backpressure is implemented. When there are too many | ||
| * outstanding events, submitting an event will block. | ||
| */ | ||
| public class ParallelWatchQueue { | ||
| /** | ||
| * Shard array | ||
| */ | ||
| final ArrayList<Shard> shards; | ||
| /** | ||
| * Implements backpressure. Semaphore is grabbed whenever queueing. | ||
| */ | ||
| final Semaphore queueSlotsLeft; | ||
| final ExecutorService executor; | ||
| final int shardCount; | ||
|
|
||
| /** | ||
| * Create a processor that can process watch events concurrently. It | ||
| * guarantees that events on the same shard are processed in a linear | ||
| * order. This code is multithread safe and can | ||
| * | ||
| * @param executor The executor that has runnables submitted. | ||
| * @param maxOutstanding How many events should we have queued up | ||
| * before we backpressure and block submitting watch events. | ||
| * @param maxShards how many shards to use. | ||
| */ | ||
| public ParallelWatchQueue(ExecutorService executor, int maxOutstanding, int maxShards) { | ||
| this.executor = executor; | ||
| // Immutable after construction. | ||
| this.shards = new ArrayList<>(maxShards); | ||
| this.shardCount = maxShards; | ||
| this.queueSlotsLeft = new Semaphore(maxOutstanding, true); | ||
| for (int ii = 0; ii < maxShards; ii++) { | ||
| shards.add(new Shard()); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Submit a given watch event, represented as a Runnable, on the given | ||
| * shard number. | ||
| * We guarantee that processing. is done asynchronously on a thread | ||
| * such that events submitted in a happensBefore order on the same | ||
| * shard number are processed in that same order. | ||
| * | ||
| * @param event The event operation. | ||
| * @param shardNum The shard number it is submitted on. | ||
| * @throws InterruptedException | ||
| */ | ||
| public void submitEvent(Runnable event, int shardNum) throws InterruptedException { | ||
| // Block if we are backpressuring. | ||
| queueSlotsLeft.acquire(); | ||
| shards.get(shardNum).submitEvent(event); | ||
| } | ||
|
|
||
| /** | ||
| * Number of shards the workload is divided between. | ||
| * @return shard count. | ||
| */ | ||
| public int getShardCount() { | ||
| return shardCount; | ||
| } | ||
|
|
||
| /** | ||
| * Implements a single queue shard. | ||
| */ | ||
| private class Shard { | ||
| /* Used to lock adding events or processing events from a shard */ | ||
| private final Lock lock = new ReentrantLock(); | ||
| /** | ||
| * Implements the actual queue of outstanding events | ||
| */ | ||
| private final Deque<Runnable> deque = new ArrayDeque<Runnable>(); | ||
|
|
||
| /** | ||
| * Process all queued events in this shard. This can be called | ||
| * from multiple threads. It internally locks, allowing only one | ||
| * thread to actually process this shard at a time. | ||
| */ | ||
| void processDeque() { | ||
| lock.lock(); | ||
| try { | ||
| while (!deque.isEmpty()) { | ||
| var event = deque.removeFirst(); | ||
| // Mark event as dequeed so we release any backpressure. | ||
| queueSlotsLeft.release(); | ||
| executor.submit(event); | ||
| } | ||
| } finally { | ||
| lock.unlock(); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Submit an event. | ||
| * <p> | ||
| * If two events are submitted with a happensBefore relationship the locks guarantee | ||
| * that the events will be queued sequentially in the the queue. | ||
| * <p> | ||
| * Because processQueue locks on the same lock. Any events currently being processed | ||
| * on this shard will be processed before the new event is submitted. | ||
| * <p> | ||
| * We have an invariant that whenever a deque has an event on it, there is at least | ||
| * one runnable in the executorservice that will process this shard. | ||
| * | ||
| * @param event The runnable to run. | ||
| */ | ||
| void submitEvent(Runnable event) { | ||
| lock.lock(); | ||
| try { | ||
| deque.addLast(event); | ||
| } finally { | ||
| lock.unlock(); | ||
| } | ||
| executor.submit(this::processDeque); | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters