mirror of
https://github.com/VolmitSoftware/Iris.git
synced 2026-10-26 05:50:39 +00:00
Reduced admission lock contention
- Stage admission avoids FIFO metadata lock handoffs - Cutover priority and runtime retirement retain their existing checks
This commit is contained in:
commit
0d2e8a45ff
2 files changed
+50
-1
No files matched your search
@@ -104,7 +104,7 @@ public final class GenerationAdmission {
|
|||||||
|
|
||||||
private State(Path root) {
|
private State(Path root) {
|
||||||
this.root = root;
|
this.root = root;
|
||||||
this.lock = new ReentrantLock(true);
|
this.lock = new ReentrantLock();
|
||||||
this.changed = lock.newCondition();
|
this.changed = lock.newCondition();
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,10 +5,16 @@ import org.junit.Test;
|
|||||||
import org.junit.rules.TemporaryFolder;
|
import org.junit.rules.TemporaryFolder;
|
||||||
|
|
||||||
import java.nio.file.Path;
|
import java.nio.file.Path;
|
||||||
|
import java.util.ArrayList;
|
||||||
|
import java.util.List;
|
||||||
import java.util.concurrent.CompletableFuture;
|
import java.util.concurrent.CompletableFuture;
|
||||||
import java.util.concurrent.CountDownLatch;
|
import java.util.concurrent.CountDownLatch;
|
||||||
|
import java.util.concurrent.ExecutorService;
|
||||||
|
import java.util.concurrent.Executors;
|
||||||
|
import java.util.concurrent.Future;
|
||||||
import java.util.concurrent.FutureTask;
|
import java.util.concurrent.FutureTask;
|
||||||
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeUnit;
|
||||||
|
import java.util.concurrent.atomic.AtomicBoolean;
|
||||||
|
|
||||||
import static org.junit.Assert.assertEquals;
|
import static org.junit.Assert.assertEquals;
|
||||||
import static org.junit.Assert.assertFalse;
|
import static org.junit.Assert.assertFalse;
|
||||||
@@ -77,6 +83,37 @@ public class GenerationAdmissionTest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void cutoverDrainsContinuouslyArrivingStages() throws Exception {
|
||||||
|
GenerationAdmission admission = new GenerationAdmission(temporaryFolder.getRoot().toPath());
|
||||||
|
GenerationAdmission.RuntimeLease runtime = admission.retainRuntime();
|
||||||
|
ExecutorService executor = Executors.newFixedThreadPool(9, Thread.ofPlatform().daemon().factory());
|
||||||
|
AtomicBoolean running = new AtomicBoolean(true);
|
||||||
|
CountDownLatch ready = new CountDownLatch(8);
|
||||||
|
List<Future<?>> stages = new ArrayList<>(8);
|
||||||
|
try {
|
||||||
|
for (int i = 0; i < 8; i++) {
|
||||||
|
stages.add(executor.submit(() -> enterStagesUntilCutover(admission, running, ready)));
|
||||||
|
}
|
||||||
|
assertTrue(ready.await(5, TimeUnit.SECONDS));
|
||||||
|
Future<?> cutover = executor.submit(() -> {
|
||||||
|
try (GenerationAdmission.CutoverLease ignored = admission.beginCutover()) {
|
||||||
|
running.set(false);
|
||||||
|
}
|
||||||
|
});
|
||||||
|
cutover.get(5, TimeUnit.SECONDS);
|
||||||
|
for (Future<?> stage : stages) {
|
||||||
|
stage.get(5, TimeUnit.SECONDS);
|
||||||
|
}
|
||||||
|
} finally {
|
||||||
|
running.set(false);
|
||||||
|
executor.shutdownNow();
|
||||||
|
assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS));
|
||||||
|
runtime.close();
|
||||||
|
}
|
||||||
|
assertThrows(IllegalStateException.class, admission::enterStage);
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void aStageMayCompleteOnAThreadOtherThanItsCaller() throws Exception {
|
public void aStageMayCompleteOnAThreadOtherThanItsCaller() throws Exception {
|
||||||
GenerationAdmission admission = new GenerationAdmission(temporaryFolder.getRoot().toPath());
|
GenerationAdmission admission = new GenerationAdmission(temporaryFolder.getRoot().toPath());
|
||||||
@@ -197,6 +234,18 @@ public class GenerationAdmissionTest {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private static void enterStagesUntilCutover(GenerationAdmission admission, AtomicBoolean running, CountDownLatch ready) {
|
||||||
|
boolean started = false;
|
||||||
|
while (running.get() && !Thread.currentThread().isInterrupted()) {
|
||||||
|
try (GenerationAdmission.StageLease ignored = admission.enterStage()) {
|
||||||
|
if (!started) {
|
||||||
|
started = true;
|
||||||
|
ready.countDown();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private static void awaitWaiting(Thread thread) throws InterruptedException {
|
private static void awaitWaiting(Thread thread) throws InterruptedException {
|
||||||
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5L);
|
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5L);
|
||||||
while (thread.getState() != Thread.State.WAITING && System.nanoTime() < deadline) {
|
while (thread.getState() != Thread.State.WAITING && System.nanoTime() < deadline) {
|
||||||
|
|||||||
Reference in new issue
Block a user