-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Use separate ForkJoin pools for LocalServer, MTLS Server/Client
It's using UnsafeExecutors, so we'll see how long this can last. But at least for the LocalServer case for in process simulation, this seems to be a win
- Loading branch information
1 parent
24e8036
commit 0dcd865
Showing
8 changed files
with
230 additions
and
19 deletions.
There are no files selected for viewing
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
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
125 changes: 125 additions & 0 deletions
125
memberships/src/main/java/com/salesforce/apollo/archipelago/UnsafeExecutors.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,125 @@ | ||
| package com.salesforce.apollo.archipelago; | ||
|
|
||
| import java.lang.Thread.UncaughtExceptionHandler; | ||
| import java.lang.invoke.MethodHandle; | ||
| import java.lang.invoke.MethodHandles; | ||
| import java.lang.reflect.Field; | ||
| import java.lang.reflect.InvocationTargetException; | ||
| import java.util.List; | ||
| import java.util.Objects; | ||
| import java.util.concurrent.*; | ||
| import java.util.concurrent.atomic.AtomicBoolean; | ||
|
|
||
| import static java.lang.invoke.MethodHandles.insertArguments; | ||
| import static java.lang.invoke.MethodType.methodType; | ||
|
|
||
| @SuppressWarnings("unused") | ||
| public class UnsafeExecutors { | ||
| private static final MethodHandle SET_EXECUTOR; | ||
|
|
||
| static { | ||
| try { | ||
| var unsafeClass = Class.forName("sun.misc.Unsafe"); | ||
| var unsafeField = unsafeClass.getDeclaredField("theUnsafe"); | ||
| unsafeField.setAccessible(true); | ||
| var unsafe = unsafeField.get(null); | ||
| var objectFieldOffset = unsafeClass.getMethod("objectFieldOffset", Field.class); | ||
| var executorField = VTB.class.getDeclaredField("executor"); | ||
| executorField.setAccessible(true); | ||
| var executorOffset = (long) objectFieldOffset.invoke(unsafe, executorField); | ||
| var putObject = MethodHandles.lookup() | ||
| .findVirtual(unsafeClass, "putObject", | ||
| methodType(void.class, Object.class, long.class, Object.class)); | ||
| var setExecutor = insertArguments(insertArguments(putObject, 2, executorOffset), 0, unsafe); | ||
| SET_EXECUTOR = setExecutor; | ||
| } catch (ClassNotFoundException | NoSuchFieldException | NoSuchMethodException | IllegalAccessException | | ||
| InvocationTargetException e) { | ||
| throw new AssertionError(e); | ||
| } | ||
| } | ||
|
|
||
| public static ExecutorService newVirtualThreadPerTaskExecutor() { | ||
| return virtualThreadExecutor(new ForkJoinPool()); | ||
| } | ||
|
|
||
| public static <B extends Thread.Builder> B configureBuilderExecutor(B builder, Executor executor) { | ||
| if (executor != null) { | ||
| setExecutor(builder, executor); | ||
| } | ||
| return builder; | ||
| } | ||
|
|
||
| public static ExecutorService virtualThreadExecutor(ExecutorService executor) { | ||
| Objects.requireNonNull(executor); | ||
| return new VirtualThreadExecutor(executor); | ||
| } | ||
|
|
||
| private static void setExecutor(Object builder, Object executor) { | ||
| try { | ||
| SET_EXECUTOR.invokeExact(builder, executor); | ||
| } catch (Throwable e) { | ||
| throw new AssertionError(e); | ||
| } | ||
| } | ||
|
|
||
| private static class BTB { | ||
| private int characteristics; | ||
| private long counter; | ||
| private String name; | ||
| private UncaughtExceptionHandler uhe; | ||
| } | ||
|
|
||
| private static class VirtualThreadExecutor extends AbstractExecutorService { | ||
| private final ExecutorService executor; | ||
| private final AtomicBoolean started = new AtomicBoolean(true); | ||
|
|
||
| public VirtualThreadExecutor(ExecutorService executor) { | ||
| this.executor = executor; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public void execute(Runnable command) { | ||
| if (!started.get()) { | ||
| throw new RejectedExecutionException("Executor shutdown"); | ||
| } | ||
| var builder = Thread.ofVirtual(); | ||
| setExecutor(builder, executor); | ||
| builder.start(command); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isShutdown() { | ||
| return executor.isShutdown(); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isTerminated() { | ||
| return !executor.isTerminated(); | ||
| } | ||
|
|
||
| @Override | ||
| public void shutdown() { | ||
| if (!started.compareAndSet(true, false)) { | ||
| return; | ||
| } | ||
| executor.shutdown(); | ||
| } | ||
|
|
||
| @Override | ||
| public List<Runnable> shutdownNow() { | ||
| if (!started.compareAndSet(true, false)) { | ||
| return List.of(); | ||
| } | ||
| return executor.shutdownNow(); | ||
| } | ||
| } | ||
|
|
||
| private static class VTB extends BTB { | ||
| private Executor executor; | ||
| } | ||
| } |
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
90 changes: 90 additions & 0 deletions
90
memberships/src/test/java/com/salesforce/apollo/archipelago/UnsafeExecutorsTest.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,90 @@ | ||
| package com.salesforce.apollo.archipelago; | ||
|
|
||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| import java.util.ArrayDeque; | ||
| import java.util.List; | ||
| import java.util.Queue; | ||
| import java.util.concurrent.*; | ||
| import java.util.concurrent.locks.Lock; | ||
| import java.util.concurrent.locks.ReentrantLock; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertFalse; | ||
|
|
||
| public class UnsafeExecutorsTest { | ||
| private static String carrierThreadName() { | ||
| var name = Thread.currentThread().toString(); | ||
| var index = name.lastIndexOf('@'); | ||
| if (index == -1) { | ||
| throw new AssertionError(); | ||
| } | ||
| return name.substring(index + 1); | ||
| } | ||
|
|
||
| @Test | ||
| public void virtualThreadExecutorSingleThreadExecutor() throws InterruptedException { | ||
| var executor = Executors.newSingleThreadExecutor(); | ||
| var virtualExecutor = UnsafeExecutors.virtualThreadExecutor(executor); | ||
| var carrierThreadNames = new CopyOnWriteArraySet<String>(); | ||
| for (var i = 0; i < 10; i++) { | ||
| virtualExecutor.execute(() -> carrierThreadNames.add(carrierThreadName())); | ||
| } | ||
| executor.shutdown(); | ||
| executor.awaitTermination(1, TimeUnit.DAYS); | ||
| assertEquals(1, carrierThreadNames.size()); | ||
| } | ||
|
|
||
| @Test | ||
| void testVirtualThread() { | ||
| Queue<Runnable> executor = new ArrayDeque<>(); | ||
| var virtualExecutor = UnsafeExecutors.virtualThreadExecutor(wrap(executor::add)); | ||
|
|
||
| Lock lock = new ReentrantLock(); | ||
| lock.lock(); | ||
| virtualExecutor.execute(lock::lock); | ||
| assertEquals(1, executor.size(), "runnable for vthread has not been submitted"); | ||
| executor.poll().run(); | ||
| assertEquals(0, executor.size(), "vthread has not blocked"); | ||
| lock.unlock(); | ||
| assertEquals(1, executor.size(), "vthread is not schedulable"); | ||
| executor.poll().run(); | ||
| assertFalse(lock.tryLock(), "the virtual thread does not hold the lock"); | ||
| } | ||
|
|
||
| private ExecutorService wrap(Executor ex) { | ||
| return new AbstractExecutorService() { | ||
|
|
||
| @Override | ||
| public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public void execute(Runnable command) { | ||
| System.out.println("Yes!"); | ||
| ex.execute(command); | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isShutdown() { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isTerminated() { | ||
| return false; | ||
| } | ||
|
|
||
| @Override | ||
| public void shutdown() { | ||
|
|
||
| } | ||
|
|
||
| @Override | ||
| public List<Runnable> shutdownNow() { | ||
| return List.of(); | ||
| } | ||
| }; | ||
| } | ||
| } |