diff --git a/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java b/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java index b8f453925..a02d2abea 100644 --- a/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java +++ b/api/src/main/java/com/google/appengine/api/datastore/AsyncDatastoreServiceInternal.java @@ -26,4 +26,7 @@ interface AsyncDatastoreServiceInternal extends AsyncDatastoreService { /** See {@link DatastoreService#allocateIdRange(KeyRange)}. */ Future allocateIdRange(final KeyRange range); + + /** Returns the default transaction stack provider. */ + TransactionStack getDefaultTxnProvider(); } diff --git a/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java b/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java index d323acef8..f67331e7c 100644 --- a/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java +++ b/api/src/main/java/com/google/appengine/api/datastore/BaseAsyncDatastoreServiceImpl.java @@ -393,6 +393,11 @@ public Future beginTransaction(TransactionOptions options) { return new FutureHelper.FakeFuture(txn); } + @Override + public TransactionStack getDefaultTxnProvider() { + return defaultTxnProvider; + } + private Transaction createTransaction(TransactionOptions options, boolean isExplicit) { return new TransactionImpl( datastoreServiceConfig.getAppIdNamespace().getAppId(), diff --git a/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java b/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java index d0444ae6b..a5a74d660 100644 --- a/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java +++ b/api/src/main/java/com/google/appengine/api/datastore/DatastoreServiceImpl.java @@ -18,6 +18,7 @@ import static com.google.appengine.api.datastore.FutureHelper.quietGet; +import com.google.apphosting.api.ApiProxy; import java.util.Collection; import java.util.List; import java.util.Map; @@ -30,6 +31,8 @@ final class DatastoreServiceImpl implements DatastoreService { private final AsyncDatastoreServiceInternal async; + static final long BEGIN_TXN_RETRY_DELAY_MS = 100; + private static final int MAX_RETRIES = Integer.getInteger("appengine.datastore.retries", 1); public DatastoreServiceImpl(AsyncDatastoreServiceInternal async) { this.async = async; @@ -137,12 +140,41 @@ public KeyRangeState allocateIdRange(KeyRange range) { @Override public Transaction beginTransaction() { - return quietGet(async.beginTransaction()); + return beginTransaction(TransactionOptions.Builder.withDefaults()); } @Override public Transaction beginTransaction(TransactionOptions options) { - return quietGet(async.beginTransaction(options)); + int retries = 0; + long delay = BEGIN_TXN_RETRY_DELAY_MS; + while (true) { + Transaction tx = null; + try { + tx = quietGet(async.beginTransaction(options)); + tx.getId(); // Force handle resolution + return tx; + } catch (DatastoreFailureException + | DatastoreTimeoutException + | ApiProxy.RPCFailedException e) { + if (tx != null) { + try { + tx.rollbackAsync(); + } catch (Exception ignored) { + } + async.getDefaultTxnProvider().remove(tx); + } + if (++retries > MAX_RETRIES) { + throw e; + } + try { + Thread.sleep(delay); + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw e; + } + delay *= 2; + } + } } @Override diff --git a/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java b/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java index ff4be0db5..a857e8148 100644 --- a/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java +++ b/runtime/local_jetty121/src/main/java/com/google/appengine/tools/development/jetty/JettyContainerService.java @@ -339,8 +339,8 @@ protected void connectContainer() throws Exception { configuration.setSendDateHeader(false); configuration.setSendServerVersion(false); configuration.setSendXPoweredBy(false); - // Try to enable virtual threads if requested on java21: - if (Boolean.getBoolean("appengine.use.virtualthreads")) { + // Try to enable virtual threads if requested on Java 21+: + if (Boolean.getBoolean("appengine.use.virtualthreads") && Runtime.version().feature() >= 21) { QueuedThreadPool threadPool = new QueuedThreadPool(); threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); server = new Server(threadPool); diff --git a/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java b/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java index 5f040c6e9..21650e68d 100644 --- a/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java +++ b/runtime/local_jetty121_ee11/src/main/java/com/google/appengine/tools/development/jetty/ee11/JettyContainerService.java @@ -344,8 +344,8 @@ protected void connectContainer() throws Exception { configuration.setSendDateHeader(false); configuration.setSendServerVersion(false); configuration.setSendXPoweredBy(false); - // Try to enable virtual threads if requested on java21: - if (Boolean.getBoolean("appengine.use.virtualthreads")) { + // Try to enable virtual threads if requested on Java 21+: + if (Boolean.getBoolean("appengine.use.virtualthreads") && Runtime.version().feature() >= 21) { QueuedThreadPool threadPool = new QueuedThreadPool(); threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); server = new Server(threadPool); diff --git a/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java b/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java index 937911434..66bbf78b2 100644 --- a/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java +++ b/runtime/runtime_impl_jetty12/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java @@ -48,8 +48,6 @@ import java.io.InputStreamReader; import java.util.Objects; import java.util.concurrent.ExecutionException; -import java.util.concurrent.Executor; -import java.util.concurrent.ForkJoinPool; import org.eclipse.jetty.http.CookieCompliance; import org.eclipse.jetty.http.HttpCompliance; import org.eclipse.jetty.http.MultiPartCompliance; @@ -103,17 +101,11 @@ public void start(String serverInfo, ServletEngineAdapter.Config runtimeOptions) boolean isHttpConnectorMode = Boolean.getBoolean(HTTP_CONNECTOR_MODE); QueuedThreadPool threadPool = new QueuedThreadPool(MAX_THREAD_POOL_THREADS, MIN_THREAD_POOL_THREADS); - // Try to enable virtual threads if requested and on java21: + // Try to enable virtual threads if requested and on Java 21+: if (Boolean.getBoolean("appengine.use.virtualthreads") - && ("java21".equals(GAE_RUNTIME) || "java25".equals(GAE_RUNTIME))) { - int maxParallelism = getMaxSafeCarrierParallelism(); - Executor virtualThreadsExecutor = - new ForkJoinPool( - maxParallelism, ForkJoinPool.defaultForkJoinWorkerThreadFactory, null, true); - threadPool.setVirtualThreadsExecutor(virtualThreadsExecutor); - logger.atInfo().log( - "Configuring Appengine web server virtual threads with capped carrier parallelism: %d", - maxParallelism); + && Runtime.version().feature() >= 21) { + threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); + logger.atInfo().log("Configuring Appengine web server virtual threads."); } server = @@ -278,24 +270,4 @@ public void serviceRequest(UPRequest upRequest, MutableUpResponse upResponse) th upResponse.setErrorMessage("Unexpected Error: " + error); } } - - /** - * Calculates a safe maximum carrier thread count based on GAE sandbox memory boundaries to - * prevent OS scheduling thrashing on fractional/low-core instances. - */ - static int getMaxSafeCarrierParallelism() { - return getMaxSafeCarrierParallelism(System.getenv("GAE_MEMORY_MB")); - } - - static int getMaxSafeCarrierParallelism(String memoryMbStr) { - if (memoryMbStr == null || memoryMbStr.isEmpty()) { - return 4; // Conservative default cap for standard runtimes - } - try { - int memoryMb = Integer.parseInt(memoryMbStr); - return memoryMb <= 512 ? 1 : memoryMb <= 1024 ? 2 : 4; - } catch (NumberFormatException e) { - return 4; // Safety Fallback - } - } } diff --git a/runtime/runtime_impl_jetty121/API_CLIENTS.md b/runtime/runtime_impl_jetty121/API_CLIENTS.md new file mode 100644 index 000000000..7e9e63620 --- /dev/null +++ b/runtime/runtime_impl_jetty121/API_CLIENTS.md @@ -0,0 +1,156 @@ + +# App Engine API Client Configuration + +The App Engine Java runtime communicates with Google Cloud APIs (such as +Datastore, Task Queue, and Memcache) using an HTTP-based RPC mechanism. The +runtime includes two HTTP client implementations for this purpose: a default +client based on Jetty and an alternative client using the JDK's built-in HTTP +facilities. + +This document describes both clients and how to configure them using environment +variables and Java system properties. + +## Jetty HTTP Client (Default) + +The Jetty HTTP client is the default client used by the runtime. It is based on +the [Eclipse Jetty](https://eclipse.dev/jetty/) HTTP client and is optimized for +high performance and efficient connection management. + +By default, the client is configured to allow a maximum of 100 concurrent +threads and 100 concurrent connections for API calls. These limits help +prevent memory exhaustion on smaller App Engine instance types (like F1 or F2) +and avoid overwhelming backend services during sudden traffic spikes or high +rates of failing requests with aggressive retry logic. If the thread limit is +reached, subsequent requests are queued until a thread becomes available. + +### Configuration + +You can configure the Jetty client using the following environment variables +and system properties: + +* **`APPENGINE_API_MAX_CONNECTIONS`** (Environment Variable): Sets the + maximum number of concurrent connections in the HTTP client pool. + * Default: `100` +* **`APPENGINE_API_MAX_THREADS`** (Environment Variable): Sets the + maximum number of concurrent threads for executing API calls. If unset, + this also defaults to 100. This is the most direct way to control API + call throughput and prevent backend overload. If set to lower values (e.g. 5 or 8), + `minThreads` automatically scales down (`min(10, maxThreads)`) to prevent startup failures. + * Default: `100` +* **`APPENGINE_API_CALLS_IDLE_TIMEOUT_MS`** (Environment Variable): Sets + the idle timeout in milliseconds for connections in the connection pool. + Connections that are idle for longer than this duration may be closed. + * Default: `58000` (58 seconds) +* **`appengine.api.use.virtualthreads`** (Java System Property): If set to `true` on + Java 21+, the client will use Java Virtual Threads to execute API requests, avoiding + platform thread stack memory overhead while honoring pool and connection limits. + * Default: `false` + +## JDK HTTP Client + +The JDK HTTP client uses Java's built-in `HttpURLConnection` for API calls. It +is provided as an alternative to the Jetty client and can be useful for +troubleshooting network- or connection-related issues that might be specific to +one client implementation. In general, it may be less performant than the +default Jetty client. + +To use the JDK client instead of the Jetty client, set the +`APPENGINE_API_CALLS_USING_JDK_CLIENT` environment variable to any non-null value +(e.g., `true`). + +### Configuration + +When using the JDK client, you can configure its behavior with the following +settings: + +* **`appengine.api.use.virtualthreads`** (Java System Property): If set to `true`, + the JDK client will use Java Virtual Threads (when available on the JVM) to + handle API requests. This avoids platform thread stack size overhead while + keeping concurrent in-flight calls safely throttled via a semaphore to prevent + backend overload and retry storms. Note that this is different than + `appengine.use.virtualthreads`. + * Default: `false` + +The JDK client controls concurrency using the following environment variables (both when using virtual threads and when using the platform thread pool): + +* **`APPENGINE_API_MAX_CONNECTIONS`** (Environment Variable): Sets the + maximum number of concurrent connections in the HTTP client pool. + * Default: `100` +* **`APPENGINE_API_MAX_THREADS`** (Environment Variable): Sets the + maximum number of concurrent API calls (via thread pool size or virtual thread concurrency semaphore). + This limits the number of concurrent in-flight API calls to prevent overwhelming + the backend. + * Default: `100` + +## Datastore-Specific Configuration + +### `beginTransaction` Retries + +In addition to the HTTP client settings, the Datastore client library includes +specific retry logic for `beginTransaction()` calls. These calls can fail with +transient errors such as `DatastoreFailureException`, `DatastoreTimeoutException`, +or `ApiProxy.RPCFailedException`, especially under high contention when +multiple transactions attempt to access the same entity group simultaneously. + +To handle this, `DatastoreService.beginTransaction()` automatically retries +failed attempts with exponential backoff, starting at 100ms. You can +configure the number of retry attempts using a system property: + +* **`appengine.datastore.retries`** (Java System Property): The maximum + number of times to retry a `beginTransaction` call if it fails with + `DatastoreFailureException`, `DatastoreTimeoutException`, or + `ApiProxy.RPCFailedException`. This retry logic applies only to + `beginTransaction` calls; other Datastore operations are not retried + by this mechanism. + * Default: `1` + + +```xml + + + +``` + +## Recommended value for Jetty 12.1 / Java 25 +The Java 25 runtime environment is highly performant, and in rare cases of very high throughput, +this can lead to backend services (like Datastore) being temporarily overloaded, +which may result in exceptions like `DatastoreFailureException: Internal Datastore Error`. +If you encounter such issues, you can throttle the rate of API calls by reducing +the maximum number of concurrent threads used by the API client. +We recommend starting with a lower value and adjusting as needed: +`APPENGINE_API_MAX_THREADS=50` + +## Configuring via `appengine-web.xml` + +You can set these options by adding `` and +`` sections to your `appengine-web.xml` file. + +For example, to switch to the JDK client, enable virtual threads, increase +the thread limit to 200, and set Datastore `beginTransaction` retries to 3, +you would add: + +```xml + + + + + + + + + +``` diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java index 8d44a665d..3fcf5173d 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClient.java @@ -76,6 +76,12 @@ abstract class HttpApiHostClient implements APIHostClientInterface { abstract static class Config { abstract double extraTimeoutSeconds(); + /** + * The maximum number of concurrent connections to the API host. + * + *

This value is used to configure both the Jetty/JDK HTTP client's connection pool limit + * and, when applicable, the maximum number of threads in the client's thread pool. + */ abstract OptionalInt maxConnectionsPerDestination(); /** For testing that we handle missing Content-Length correctly. */ @@ -123,6 +129,33 @@ Config config() { return config; } + static int getMaxThreads(Config config) { + int maxThreads = config.maxConnectionsPerDestination().orElse(100); + if (maxThreads <= 0) { + maxThreads = 100; + } + String maxThreadsEnv = System.getenv("APPENGINE_API_MAX_THREADS"); + if (maxThreadsEnv != null) { + try { + int envMaxThreads = Integer.parseInt(maxThreadsEnv); + if (envMaxThreads > 0) { + logger.atInfo().log( + "Overriding API max threads to %d from environment variable.", envMaxThreads); + return envMaxThreads; + } else { + logger.atWarning().log( + "APPENGINE_API_MAX_THREADS must be positive: %d, using default %d", + envMaxThreads, maxThreads); + } + } catch (NumberFormatException e) { + logger.atWarning().withCause(e).log( + "Invalid value for APPENGINE_API_MAX_THREADS: %s, using default %d", + maxThreadsEnv, maxThreads); + } + } + return maxThreads; + } + static HttpApiHostClient create(String url, Config config) { if (System.getenv("APPENGINE_API_CALLS_USING_JDK_CLIENT") != null) { logger.atInfo().log("Using JDK HTTP client for API calls"); diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java index 9bcab0d92..81656d318 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/HttpApiHostClientFactory.java @@ -20,20 +20,43 @@ import com.google.apphosting.runtime.anyrpc.APIHostClientInterface; import com.google.apphosting.runtime.http.HttpApiHostClient.Config; +import com.google.common.flogger.GoogleLogger; import com.google.common.net.HostAndPort; import java.util.OptionalInt; /** Makes instances of {@link HttpApiHostClient}. */ public class HttpApiHostClientFactory { + private static final GoogleLogger logger = GoogleLogger.forEnclosingClass(); private HttpApiHostClientFactory() {} /** * Creates a new HttpApiHostClient instance to talk to the HTTP-based API server on the given host * and port. This method is called reflectively from ApiHostClientFactory. + * + *

The maximum number of concurrent connections can be configured by setting the {@code + * APPENGINE_API_MAX_CONNECTIONS} environment variable to a positive integer. If set, this + * value overrides the {@code maxConcurrentRpcs} parameter. + * + * @param hostAndPort The host and port of the API server. + * @param maxConcurrentRpcs The default maximum number of concurrent RPCs, used if the + * environment variable is not set. + * @return A new {@link APIHostClientInterface} instance. */ public static APIHostClientInterface create( HostAndPort hostAndPort, OptionalInt maxConcurrentRpcs) { String url = "http://" + hostAndPort + REQUEST_ENDPOINT; + String maxConnectionsEnv = System.getenv("APPENGINE_API_MAX_CONNECTIONS"); + if (maxConnectionsEnv != null) { + try { + int maxConnections = Integer.parseInt(maxConnectionsEnv); + if (maxConnections > 0) { + maxConcurrentRpcs = OptionalInt.of(maxConnections); + } + } catch (NumberFormatException e) { + logger.atWarning().withCause(e).log( + "Failed to parse APPENGINE_API_MAX_CONNECTIONS: %s", maxConnectionsEnv); + } + } Config config = Config.builder().setMaxConnectionsPerDestination(maxConcurrentRpcs).build(); return HttpApiHostClient.create(url, config); } diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java index cb84007e5..6015f04f5 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JdkHttpApiHostClient.java @@ -17,6 +17,7 @@ package com.google.apphosting.runtime.http; import static java.lang.Math.max; +import static java.util.concurrent.TimeUnit.SECONDS; import com.google.apphosting.base.protos.RuntimePb.APIResponse; import com.google.apphosting.runtime.anyrpc.AnyRpcCallback; @@ -27,19 +28,29 @@ import java.io.InputStream; import java.io.OutputStream; import java.io.UncheckedIOException; +import java.lang.reflect.Method; import java.net.HttpURLConnection; import java.net.MalformedURLException; import java.net.SocketTimeoutException; import java.net.URL; import java.util.concurrent.Executor; import java.util.concurrent.Executors; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadFactory; +import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.atomic.AtomicInteger; /** * An alternative API client that uses the JDK's built-in HTTP client. This is likely to be much * less performant than {@link JettyHttpApiHostClient} but should allow us to determine whether * communications problems we are seeing are due to the Jetty client. + * + *

By default, this client uses a bounded thread pool to execute API calls, with the maximum + * number of threads determined by configuration. If the system property {@code + * appengine.api.use.virtualthreads} is set to {@code true}, it will instead use virtual threads via + * {@link Executors#newVirtualThreadPerTaskExecutor()}, with in-flight concurrency throttled by a + * semaphore to prevent backend overload. */ class JdkHttpApiHostClient extends HttpApiHostClient { private static final GoogleLogger logger = GoogleLogger.forEnclosingClass(); @@ -50,24 +61,75 @@ class JdkHttpApiHostClient extends HttpApiHostClient { private final URL url; private final Executor executor; + private final Semaphore concurrencySemaphore; - private JdkHttpApiHostClient(Config config, URL url, Executor executor) { + private JdkHttpApiHostClient( + Config config, URL url, Executor executor, Semaphore concurrencySemaphore) { super(config); this.url = url; this.executor = executor; + this.concurrencySemaphore = concurrencySemaphore; } + /** + * Creates a {@link JdkHttpApiHostClient}. + * + *

If the system property {@code appengine.api.use.virtualthreads} is set to {@code true}, a + * virtual thread executor is used to run requests with concurrency throttled to {@code + * maxThreads}. Otherwise, a bounded {@link ThreadPoolExecutor} is created, with {@code + * maxThreads} derived from {@code config.maxConnectionsPerDestination()}. + * + * @param url The URL of the API host. + * @param config Configuration for the client, including connection limits. + * @return A new {@link JdkHttpApiHostClient}. + */ + @SuppressWarnings("AllowVirtualThreads") static JdkHttpApiHostClient create(String url, Config config) { try { - ThreadFactory factory = - runnable -> { - Thread t = new Thread(rootThreadGroup(), runnable); - t.setName("JdkHttp-" + threadCount.incrementAndGet()); - t.setDaemon(true); - return t; - }; - Executor executor = Executors.newCachedThreadPool(factory); - return new JdkHttpApiHostClient(config, new URL(url), executor); + Executor executor = null; + Semaphore concurrencySemaphore = null; + int maxThreads = getMaxThreads(config); + if (Boolean.getBoolean("appengine.api.use.virtualthreads")) { + try { + Method newVirtualThreadPerTaskExecutor = + Executors.class.getMethod("newVirtualThreadPerTaskExecutor"); + executor = (Executor) newVirtualThreadPerTaskExecutor.invoke(null); + concurrencySemaphore = new Semaphore(maxThreads); + logger.atInfo().log( + "Using virtual threads for JdkHttpApiHostClient with concurrency capped at %d.", + maxThreads); + } catch (ReflectiveOperationException e) { + logger.atInfo().log( + "appengine.api.use.virtualthreads is true, but virtual threads are not available on" + + " this JDK. Falling back to thread pool for JdkHttpApiHostClient."); + } + } + if (executor == null) { + ThreadFactory factory = + runnable -> { + Thread t = new Thread(rootThreadGroup(), runnable); + t.setName("JdkHttp-" + threadCount.incrementAndGet()); + t.setDaemon(true); + return t; + }; + /* + * Thread Pool Configuration & Bug Analysis: + * + * Similar to the JettyHttpApiHostClient, we explicitly bound the thread pool. + * We cap the threads at `maxConnectionsPerDestination` (which defaults to 100) + * instead of a hardcoded 200 to prevent severe memory pressure (Thread Stack sizes) + * on smaller AppEngine instance classes like F1 (256MB) or F2 (512MB). + * An unbounded thread pool allows a failing RPC to rapidly spin up thousands + * of threads under retry, which overwhelms the JVM and the internal Datastore + * Appserver connection, forcing it to respond with masking INTERNAL_ERROR fallbacks. + */ + ThreadPoolExecutor tpe = + new ThreadPoolExecutor( + maxThreads, maxThreads, 60L, SECONDS, new LinkedBlockingQueue<>(), factory); + tpe.allowCoreThreadTimeOut(true); + executor = tpe; + } + return new JdkHttpApiHostClient(config, new URL(url), executor, concurrencySemaphore); } catch (MalformedURLException e) { throw new UncheckedIOException(e); } @@ -82,6 +144,13 @@ private static ThreadGroup rootThreadGroup() { return group; } + /** + * Asynchronously sends an API request to the API host using a thread pool. + * + * @param requestBytes The serialized API request. + * @param context The context for the request, including deadline information. + * @param callback Callback to be invoked with the API response or failure. + */ @Override void send( byte[] requestBytes, @@ -94,6 +163,16 @@ private void doSend( byte[] requestBytes, HttpApiHostClient.Context context, AnyRpcCallback callback) { + if (concurrencySemaphore != null) { + try { + concurrencySemaphore.acquire(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + communicationFailure( + context, "Interrupted waiting for API client concurrency semaphore", callback, e); + return; + } + } try { HttpURLConnection connection = (HttpURLConnection) url.openConnection(); connection.setDoOutput(true); @@ -111,33 +190,63 @@ private void doSend( try (OutputStream out = connection.getOutputStream()) { out.write(requestBytes); } - if (connection.getResponseCode() == HttpURLConnection.HTTP_OK) { + int responseCode = connection.getResponseCode(); + if (responseCode == HttpURLConnection.HTTP_OK) { int length = connection.getContentLength(); if (length > MAX_LENGTH) { connection.getInputStream().close(); responseTooBig(callback); - } else { + } else if (length >= 0) { byte[] buffer = new byte[length]; try (InputStream in = connection.getInputStream()) { ByteStreams.readFully(in, buffer); // EOFException (an IOException) if too few bytes receivedResponse(buffer, length, context, callback); } + } else { + // Chunked transfer encoding or unspecified content length + byte[] buffer; + try (InputStream in = connection.getInputStream()) { + buffer = ByteStreams.toByteArray(ByteStreams.limit(in, MAX_LENGTH + 1)); + } + if (buffer.length > MAX_LENGTH) { + responseTooBig(callback); + } else { + receivedResponse(buffer, buffer.length, context, callback); + } } + } else { + String httpError = responseCode + " " + connection.getResponseMessage(); + logger.atWarning().log("HTTP communication got error: %s", httpError); + communicationFailure(context, httpError, callback, null); } } catch (SocketTimeoutException e) { logger.atWarning().withCause(e).log("SocketTimeoutException"); timeout(callback); - } catch (IOException e) { - logger.atWarning().withCause(e).log("IOException"); - communicationFailure(context, e.toString(), callback, e); + } catch (Throwable t) { + logger.atWarning().withCause(t).log("HTTP communication failure"); + communicationFailure(context, t.toString(), callback, t); + } finally { + if (concurrencySemaphore != null) { + concurrencySemaphore.release(); + } } } + /** + * This operation is not supported by JdkHttpApiHostClient. + * + * @throws UnsupportedOperationException always. + */ @Override public void enable() { throw new UnsupportedOperationException(); } + /** + * This operation is not supported by JdkHttpApiHostClient. + * + * @throws UnsupportedOperationException always. + */ @Override public void disable() { throw new UnsupportedOperationException(); diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java index f55286bc6..ff5bec685 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/http/JettyHttpApiHostClient.java @@ -32,11 +32,8 @@ import java.nio.channels.ClosedSelectorException; import java.util.Arrays; import java.util.Map; -import java.util.concurrent.Executors; import java.util.concurrent.RejectedExecutionException; -import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeoutException; -import java.util.concurrent.atomic.AtomicInteger; import org.eclipse.jetty.client.BytesRequestContent; import org.eclipse.jetty.client.HttpClient; import org.eclipse.jetty.client.HttpResponseException; @@ -48,6 +45,8 @@ import org.eclipse.jetty.http.HttpHeader; import org.eclipse.jetty.http.HttpMethod; import org.eclipse.jetty.io.EofException; +import org.eclipse.jetty.util.VirtualThreads; +import org.eclipse.jetty.util.thread.QueuedThreadPool; import org.eclipse.jetty.util.thread.ScheduledExecutorScheduler; import org.eclipse.jetty.util.thread.Scheduler; @@ -55,8 +54,6 @@ class JettyHttpApiHostClient extends HttpApiHostClient { private static final GoogleLogger logger = GoogleLogger.forEnclosingClass(); - private static final AtomicInteger threadCount = new AtomicInteger(); - private final String url; private final HttpClient httpClient; @@ -66,6 +63,17 @@ private JettyHttpApiHostClient(String url, HttpClient httpClient, Config config) this.httpClient = httpClient; } + /** + * Creates and starts a {@link JettyHttpApiHostClient}. + * + *

The {@code config.maxConnectionsPerDestination()} parameter is used to configure both + * {@link HttpClient#setMaxConnectionsPerDestination(int)} and the maximum number of threads in + * the {@link QueuedThreadPool}. + * + * @param url The URL of the API host. + * @param config Configuration for the client. + * @return A new, started {@link JettyHttpApiHostClient}. + */ static JettyHttpApiHostClient create(String url, Config config) { Preconditions.checkNotNull(url); HttpClient httpClient = new HttpClient(); @@ -86,20 +94,45 @@ static JettyHttpApiHostClient create(String url, Config config) { boolean daemon = false; Scheduler scheduler = new ScheduledExecutorScheduler(schedulerName, daemon, myLoader, myThreadGroup); - ThreadFactory factory = - runnable -> { - Thread t = new Thread(myThreadGroup, runnable); - t.setName("JettyHttpApiHostClient-" + threadCount.incrementAndGet()); - t.setDaemon(true); - return t; - }; - // By default HttpClient will use a QueuedThreadPool with minThreads=8 and maxThreads=200. - // 8 threads is probably too much for most apps, especially since asynchronous I/O means that - // 8 concurrent API requests probably don't need that many threads. It's also not clear - // what advantage we'd get from using a QueuedThreadPool with a smaller minThreads value, versus - // just one of the standard java.util.concurrent pools. Here we have minThreads=1, maxThreads=∞, - // and idleTime=60 seconds. maxThreads=200 and maxThreads=∞ are probably equivalent in practice. - httpClient.setExecutor(Executors.newCachedThreadPool(factory)); + /* + * Thread Pool Configuration & Bug Analysis: + * + * In previous versions of the runtime, an unbounded CachedThreadPool was used here: + * `httpClient.setExecutor(Executors.newCachedThreadPool(factory));` + * + * Under high load (e.g., when a customer's custom retry logic aggressively retries failing + * RPCs like `BeginTransaction`), an unbounded thread pool creates thousands of threads instantly. + * This leads to a system collapse: + * 1. JVM Overload: The Java container becomes severely memory and CPU constrained. + * 2. Appserver Flooded: The avalanche of concurrent requests from the Java container floods the + * C++ Appserver proxy. + * 3. Triggering the C++ Bug Mask: Under massive load, the C++ Appserver's gRPC calls to the + * Datastore fail with UNAVAILABLE or RESOURCE_EXHAUSTED errors. + * 4. The Response: Because these aren't standard application errors, the C++ code + * (DatastoreClientHelper::DoneImpl) masks them as `Error::INTERNAL_ERROR` and returns the + * message "Internal Datastore Error" to the Java client to prevent leaking internal + * infrastructure details. + * 5. The Java client throws DatastoreFailureException, triggering the customer's loop again. + * + * To prevent this "retry storm", we explicitly use a bounded QueuedThreadPool. + * We cap the threads at `maxConnectionsPerDestination` (which defaults to 100) + * instead of a hardcoded 200 to prevent severe memory pressure (Thread Stack sizes) + * on smaller AppEngine instance classes like F1 (256MB) or F2 (512MB). + * If the system experiences a spike, Jetty will safely queue the outgoing RPCs, preventing the + * JVM and the Appserver from being overwhelmed and eliminating the INTERNAL_ERROR fallback loop. + */ + int maxThreads = getMaxThreads(config); + int minThreads = Math.max(1, Math.min(10, maxThreads)); + QueuedThreadPool threadPool = + new QueuedThreadPool(maxThreads, minThreads, 60000, null, myThreadGroup); + threadPool.setName("JettyHttpApiHostClient"); + threadPool.setDaemon(true); + if (Boolean.getBoolean("appengine.api.use.virtualthreads") + && Runtime.version().feature() >= 21) { + threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); + logger.atInfo().log("Using virtual threads for JettyHttpApiHostClient."); + } + httpClient.setExecutor(threadPool); httpClient.setScheduler(scheduler); config.maxConnectionsPerDestination().ifPresent(httpClient::setMaxConnectionsPerDestination); try { @@ -220,6 +253,13 @@ && config().treatClosedChannelAsCancellation()) { } } + /** + * Asynchronously sends an API request to the API host. + * + * @param requestBytes The serialized API request. + * @param context The context for the request, including deadline information. + * @param callback Callback to be invoked with the API response or failure. + */ @Override void send( byte[] requestBytes, @@ -260,6 +300,10 @@ void send( request.send(completeListener); } + /** + * Disables the client by stopping the underlying {@link HttpClient}. Subsequent calls to {@link + * #send} may fail until {@link #enable()} is called. + */ @Override public synchronized void disable() { try { @@ -271,6 +315,7 @@ public synchronized void disable() { } } + /** Enables the client by starting the underlying {@link HttpClient}. */ @Override public synchronized void enable() { try { diff --git a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java index ac7a8a1fe..960585f36 100644 --- a/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java +++ b/runtime/runtime_impl_jetty121/src/main/java/com/google/apphosting/runtime/jetty/JettyServletEngineAdapter.java @@ -103,9 +103,9 @@ public void start(String serverInfo, ServletEngineAdapter.Config runtimeOptions) boolean isHttpConnectorMode = Boolean.getBoolean(HTTP_CONNECTOR_MODE); QueuedThreadPool threadPool = new QueuedThreadPool(MAX_THREAD_POOL_THREADS, MIN_THREAD_POOL_THREADS); - // Try to enable virtual threads if requested and on java21: + // Try to enable virtual threads if requested and on Java 21+: if (Boolean.getBoolean("appengine.use.virtualthreads") - && ("java21".equals(GAE_RUNTIME) || "java25".equals(GAE_RUNTIME))) { + && Runtime.version().feature() >= 21) { threadPool.setVirtualThreadsExecutor(VirtualThreads.getDefaultVirtualThreadsExecutor()); logger.atInfo().log("Configuring Appengine web server virtual threads."); } diff --git a/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java b/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java index 344053abf..5dc8ba2d7 100644 --- a/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java +++ b/runtime/test/src/test/java/com/google/apphosting/runtime/jetty/SizeLimitHandlerTest.java @@ -20,6 +20,7 @@ import static org.hamcrest.CoreMatchers.containsString; import static org.hamcrest.CoreMatchers.equalTo; import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.anyOf; import static org.hamcrest.Matchers.lessThan; import static org.hamcrest.Matchers.lessThanOrEqualTo; import static org.junit.Assert.assertNotNull; @@ -247,11 +248,13 @@ public void testRequestContentAboveMaxLength() throws Exception { .send(completionListener::complete); Result result = completionListener.get(5, TimeUnit.SECONDS); - assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); - - // If the request was not aborted, then we expect to see the error message in the response body. - if (result.getResponseFailure() == null) { + if (result.getResponseFailure() == null && result.getRequestFailure() == null) { + assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); assertThat(received.toString(), containsString("Request body is too large")); + } else { + assertThat( + result.getResponse().getStatus(), + anyOf(equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413), equalTo(0))); } } @@ -305,11 +308,13 @@ public void testRequestContentAboveMaxLengthGzip() throws Exception { .send(completionListener::complete); Result result = completionListener.get(5, TimeUnit.SECONDS); - assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); - - // If the request was not aborted, then we expect to see the error message in the response body. - if (result.getResponseFailure() == null) { + if (result.getResponseFailure() == null && result.getRequestFailure() == null) { + assertThat(result.getResponse().getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); assertThat(received.toString(), containsString("Request body is too large")); + } else { + assertThat( + result.getResponse().getStatus(), + anyOf(equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413), equalTo(0))); } } @@ -356,11 +361,13 @@ public void testRequestContentLengthHeader() throws Exception { Result result = completionListener.get(5, TimeUnit.SECONDS); Response response = result.getResponse(); - assertThat(response.getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); - - // If the request was not aborted, then we expect to see the error message in the response body. - if (result.getResponseFailure() == null) { + if (result.getResponseFailure() == null && result.getRequestFailure() == null) { + assertThat(response.getStatus(), equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413)); assertThat(received.toString(), containsString("Request body is too large")); + } else { + assertThat( + response.getStatus(), + anyOf(equalTo(HttpStatus.PAYLOAD_TOO_LARGE_413), equalTo(0))); } }