From 991e7b63aacd10e8cd747aec8b7596f95241a70e Mon Sep 17 00:00:00 2001 From: Atharva Moroney Date: Tue, 29 Jul 2025 14:56:11 -0700 Subject: [PATCH 1/2] perf: Add support for virtual threads --- .../google/cloud/spanner/adapter/Adapter.java | 13 ++- .../cloud/spanner/adapter/Launcher.java | 6 ++ .../adapter/util/ThreadFactoryUtil.java | 86 +++++++++++++++++++ .../adapter/ThreadFactoryUtilTest.java | 57 ++++++++++++ 4 files changed, 158 insertions(+), 4 deletions(-) create mode 100644 google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/util/ThreadFactoryUtil.java create mode 100644 google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/ThreadFactoryUtilTest.java diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java index d6813486..8cff5126 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java @@ -15,6 +15,8 @@ */ package com.google.cloud.spanner.adapter; +import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor; + import com.google.api.gax.core.CredentialsProvider; import com.google.api.gax.core.FixedCredentialsProvider; import com.google.api.gax.core.GaxProperties; @@ -91,13 +93,18 @@ void start() { } final CredentialsProvider credentialsProvider = setUpCredentialsProvider(credentials); + executor = tryCreateVirtualThreadPerTaskExecutor("virtual-thread"); + if (executor == null) { + executor = Executors.newCachedThreadPool(); + } + InstantiatingGrpcChannelProvider.Builder channelProviderBuilder = AdapterSettings.defaultGrpcTransportProviderBuilder(); channelProviderBuilder .setAllowNonDefaultServiceAccount(true) - .setChannelPoolSettings( - ChannelPoolSettings.staticallySized(options.getNumGrpcChannels())); + .setChannelPoolSettings(ChannelPoolSettings.staticallySized(options.getNumGrpcChannels())) + .setExecutor(executor); if (isEnableDirectPathXdsEnv()) { channelProviderBuilder.setAttemptDirectPath(true); @@ -140,8 +147,6 @@ void start() { options.getTcpPort(), DEFAULT_CONNECTION_BACKLOG, options.getInetAddress()); LOG.info("Local TCP server started on {}:{}", options.getInetAddress(), options.getTcpPort()); - executor = Executors.newCachedThreadPool(); - // Start accepting client connections. executor.execute(this::acceptClientConnections); diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java index b7c0efb4..edcf299e 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Launcher.java @@ -102,5 +102,11 @@ public static void main(String[] args) throws Exception { maxCommitDelayProperty); adapter.start(); + + try { + Thread.currentThread().join(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } } } diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/util/ThreadFactoryUtil.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/util/ThreadFactoryUtil.java new file mode 100644 index 00000000..b4bfed0b --- /dev/null +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/util/ThreadFactoryUtil.java @@ -0,0 +1,86 @@ +/* +Copyright 2025 Google LLC + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package com.google.cloud.spanner.adapter.util; + +import com.google.api.core.InternalApi; +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Method; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import javax.annotation.Nullable; + +/** Utility class for creating a thread factory for daemon or virtual threads. */ +@InternalApi +public final class ThreadFactoryUtil { + + /** + * Tries to create a {@link ThreadFactory} that creates virtual threads. Returns null if virtual + * threads are not supported on this JVM. + */ + @InternalApi + @Nullable + public static ThreadFactory tryCreateVirtualThreadFactory(String baseNameFormat) { + try { + Class threadBuilderClass = Class.forName("java.lang.Thread$Builder"); + Method ofVirtualMethod = Thread.class.getDeclaredMethod("ofVirtual"); + Object virtualBuilder = ofVirtualMethod.invoke(null); + Method nameMethod = threadBuilderClass.getDeclaredMethod("name", String.class, long.class); + virtualBuilder = nameMethod.invoke(virtualBuilder, baseNameFormat + "-", 0); + Method factoryMethod = threadBuilderClass.getDeclaredMethod("factory"); + return (ThreadFactory) factoryMethod.invoke(virtualBuilder); + } catch (ClassNotFoundException | NoSuchMethodException ignore) { + return null; + } catch (InvocationTargetException | IllegalAccessException e) { + // Java 20 supports virtual threads as an experimental feature. It will throw an + // UnsupportedOperationException if experimental features have not been enabled. + if (e.getCause() instanceof UnsupportedOperationException) { + return null; + } + throw new RuntimeException(e); + } + } + + /** + * Tries to create an {@link ExecutorService} that creates a new virtual thread for each task that + * it runs. Creating a new virtual thread is the recommended way to create executors using virtual + * threads, instead of creating a pool of virtual threads. Returns null if virtual threads are not + * supported on this JVM. + */ + @InternalApi + @Nullable + public static ExecutorService tryCreateVirtualThreadPerTaskExecutor(String baseNameFormat) { + ThreadFactory factory = tryCreateVirtualThreadFactory(baseNameFormat); + if (factory != null) { + try { + Method newThreadPerTaskExecutorMethod = + Executors.class.getDeclaredMethod("newThreadPerTaskExecutor", ThreadFactory.class); + return (ExecutorService) newThreadPerTaskExecutorMethod.invoke(null, factory); + } catch (NoSuchMethodException ignore) { + return null; + } catch (InvocationTargetException | IllegalAccessException e) { + // Java 20 supports virtual threads as an experimental feature. It will throw an + // UnsupportedOperationException if experimental features have not been enabled. + if (e.getCause() instanceof UnsupportedOperationException) { + return null; + } + throw new RuntimeException(e); + } + } + return null; + } +} diff --git a/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/ThreadFactoryUtilTest.java b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/ThreadFactoryUtilTest.java new file mode 100644 index 00000000..d891747d --- /dev/null +++ b/google-cloud-spanner-cassandra/src/test/java/com/google/cloud/spanner/adapter/ThreadFactoryUtilTest.java @@ -0,0 +1,57 @@ +/* +Copyright 2025 Google LLC + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package com.google.cloud.spanner.adapter; + +import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadFactory; +import static com.google.cloud.spanner.adapter.util.ThreadFactoryUtil.tryCreateVirtualThreadPerTaskExecutor; +import static com.google.common.truth.Truth.assertThat; + +import java.util.concurrent.ThreadFactory; +import org.junit.Test; + +public class ThreadFactoryUtilTest { + + @Test + public void testCreateThreadFactory() throws Exception { + if (isJava21OrHigher()) { + ThreadFactory virtualFactory = tryCreateVirtualThreadFactory("test-thread"); + assertThat(virtualFactory).isNotNull(); + } else { + assertThat(tryCreateVirtualThreadFactory("test-thread")).isNull(); + } + } + + @Test + public void testTryCreateVirtualThreadPerTaskExecutor() { + if (isJava21OrHigher()) { + assertThat(tryCreateVirtualThreadPerTaskExecutor("test-virtual-thread")).isNotNull(); + } else { + assertThat(tryCreateVirtualThreadPerTaskExecutor("test-virtual-thread")).isNull(); + } + } + + private static boolean isJava21OrHigher() { + String[] versionElements = System.getProperty("java.version").split("\\."); + int majorVersion = Integer.parseInt(versionElements[0]); + // Java 1.8 (Java 8) and lower used the format 1.8 etc. + // Java 9 and higher use the format 9.x + if (majorVersion == 1) { + majorVersion = Integer.parseInt(versionElements[1]); + } + return majorVersion >= 21; + } +} From a598e701904b5499e3fc02fde2b5b39b488d5e38 Mon Sep 17 00:00:00 2001 From: Atharva Moroney Date: Wed, 30 Jul 2025 16:29:32 -0700 Subject: [PATCH 2/2] chore: Make the use of virtual threads configurable --- .../google/cloud/spanner/adapter/Adapter.java | 20 +++++++++++-------- .../cloud/spanner/adapter/AdapterOptions.java | 13 ++++++++++++ .../adapter/SpannerCqlSessionBuilder.java | 12 +++++++++++ 3 files changed, 37 insertions(+), 8 deletions(-) diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java index b0e1fd6b..c74dcefe 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/Adapter.java @@ -67,7 +67,7 @@ final class Adapter extends AbstractApiService { private AdapterClientWrapper adapterClientWrapper; private ServerSocket serverSocket; - private ExecutorService executor; + private ExecutorService executor = null; private AdapterOptions options; /** @@ -101,18 +101,18 @@ protected void doStart() { } final CredentialsProvider credentialsProvider = setUpCredentialsProvider(credentials); - executor = tryCreateVirtualThreadPerTaskExecutor("virtual-thread"); - if (executor == null) { - executor = Executors.newCachedThreadPool(); - } - InstantiatingGrpcChannelProvider.Builder channelProviderBuilder = AdapterSettings.defaultGrpcTransportProviderBuilder(); + if (options.getUseVirtualThreads()) { + executor = tryCreateVirtualThreadPerTaskExecutor("virtual-thread"); + channelProviderBuilder.setExecutor(executor); + } + channelProviderBuilder .setAllowNonDefaultServiceAccount(true) - .setChannelPoolSettings(ChannelPoolSettings.staticallySized(options.getNumGrpcChannels())) - .setExecutor(executor); + .setChannelPoolSettings( + ChannelPoolSettings.staticallySized(options.getNumGrpcChannels())); if (isEnableDirectPathXdsEnv()) { channelProviderBuilder.setAttemptDirectPath(true); @@ -155,6 +155,10 @@ protected void doStart() { options.getTcpPort(), DEFAULT_CONNECTION_BACKLOG, options.getInetAddress()); LOG.info("Local TCP server started on {}:{}", options.getInetAddress(), options.getTcpPort()); + if (executor == null) { + executor = Executors.newCachedThreadPool(); + } + // Start accepting client connections. executor.execute(this::acceptClientConnections); diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java index 65857b11..0bd2431b 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/AdapterOptions.java @@ -37,6 +37,7 @@ static class Builder { Optional maxCommitDelay = Optional.empty(); private TransportChannelProvider channelProvider = null; private Credentials credentials; + private boolean useVirtualThreads = false; /** The Cloud Spanner endpoint. */ Builder spannerEndpoint(String spannerEndpoint) { @@ -92,6 +93,12 @@ Builder credentials(Credentials credentials) { return this; } + /** (Optional) Whether to use virtual threads (Java 21+ only) */ + Builder useVirtualThreads(boolean useVirtualThreads) { + this.useVirtualThreads = useVirtualThreads; + return this; + } + AdapterOptions build() { return new AdapterOptions(this); } @@ -105,6 +112,7 @@ AdapterOptions build() { private final Optional maxCommitDelay; private TransportChannelProvider channelProvider; private Credentials credentials; + private boolean useVirtualThreads; private AdapterOptions(Builder builder) { this.spannerEndpoint = builder.spannerEndpoint; @@ -115,6 +123,7 @@ private AdapterOptions(Builder builder) { this.maxCommitDelay = builder.maxCommitDelay; this.channelProvider = builder.channelProvider; this.credentials = builder.credentials; + this.useVirtualThreads = builder.useVirtualThreads; } static Builder newBuilder() { @@ -149,6 +158,10 @@ Credentials getCredentials() { return credentials; } + boolean getUseVirtualThreads() { + return useVirtualThreads; + } + Optional getMaxCommitDelay() { return maxCommitDelay; } diff --git a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java index c5f96e04..083d19bc 100644 --- a/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java +++ b/google-cloud-spanner-cassandra/src/main/java/com/google/cloud/spanner/adapter/SpannerCqlSessionBuilder.java @@ -59,6 +59,7 @@ public final class SpannerCqlSessionBuilder private Optional maxCommitDelay = Optional.empty(); private TransportChannelProvider channelProvider = null; private Credentials credentials; + private boolean useVirtualThreads; /** * Wraps the default CQL session with a SpannerCqlSession instance. @@ -121,6 +122,16 @@ public SpannerCqlSessionBuilder setGoogleCloudCredentials(Credentials credential return this; } + /** + * (Optional, default `false`) Enables/disables the use of virtual threads for the gRPC executor. + * Setting this option only has any effect on Java 21 and higher. In all other cases, the option + * will be ignored. + */ + protected SpannerCqlSessionBuilder setUseVirtualThreads(boolean useVirtualThreads) { + this.useVirtualThreads = useVirtualThreads; + return this; + } + /** * Creates the session with the options set by this builder. * @@ -232,6 +243,7 @@ private void createAndStartAdapter() { .numGrpcChannels(numGrpcChannels) .channelProvider(channelProvider) .credentials(credentials) + .useVirtualThreads(useVirtualThreads) .build(); adapter = new Adapter(adapterOptions);