#20387

Committed 03 Aug 2026 09:27AM UTC coverage: 89.149%. Remained the same

Build # #20387

Build Type

push

github

Committed by

web-flow

Commit Message

Increase initial timeout and add retry loop in gae-interop-testing (#12868)

1. Allow the newly deployed App Engine instance up to 2 minutes to
complete its JVM initialization and handle the cold start, to avoid the
SocketTimeoutException seen in GAE interop testing action ([to fix
failure logs](http://shortn/_y9PNlGvrki)).
2. Wait and Retry Loop: Added a loop that attempts the check up to 5
times with a 2-second sleep between attempts. This will gracefully
absorb any transient warmup or App Engine routing latency ([to fix
failure logs even with initial 2 min delay](http://shortn/_ONTi0mqEdB)).
3. Detailed Failure Logs: If all attempts fail, we now retrieve the
response bodies (result.body().string()) and include them along with the
status codes in the final error message, allowing for straightforward
debugging of any persistent downstream issues. ([failure logs even after
retry failures](http://shortn/_y9PNlGvrki))

The 3rd logs above show grpc-test.sandbox.googleapis.com must be
experiencing a backend outage.

Coverage Stats

38310 of 42973 relevant lines covered (89.15%)

0.89 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

73.27

/../servlet/src/main/java/io/grpc/servlet/AsyncServletOutputStreamWriter.java

/*
 * Copyright 2019 The gRPC Authors
 *
 * 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 io.grpc.servlet;

import static com.google.common.base.Preconditions.checkState;
import static io.grpc.servlet.ServletServerStream.toHexString;
import static java.util.logging.Level.FINE;
import static java.util.logging.Level.FINEST;

import com.google.common.annotations.VisibleForTesting;
import com.google.errorprone.annotations.CheckReturnValue;
import io.grpc.InternalLogId;
import io.grpc.servlet.ServletServerStream.ServletTransportState;
import java.io.IOException;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import java.util.function.BiFunction;
import java.util.function.BooleanSupplier;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;
import javax.servlet.AsyncContext;
import javax.servlet.ServletOutputStream;

/** Handles write actions from the container thread and the application thread. */
final class AsyncServletOutputStreamWriter {

  /**
   * Memory boundary for write actions.
   *
   * <pre>
   * WriteState curState = writeState.get();  // mark a boundary
   * doSomething();  // do something within the boundary
   * boolean successful = writeState.compareAndSet(curState, newState); // try to mark a boundary
   * if (successful) {
   *   // state has not changed since
   *   return;
   * } else {
   *   // state is changed by another thread while doSomething(), need recompute
   * }
   * </pre>
   *
   * <p>There are two threads, the container thread (calling {@code onWritePossible()}) and the
   * application thread (calling {@code runOrBuffer()}) that read and update the
   * writeState. Only onWritePossible() may turn {@code readyAndDrained} from false to true, and
   * only runOrBuffer() may turn it from true to false.
   */
  private final AtomicReference<WriteState> writeState = new AtomicReference<>(WriteState.DEFAULT);

  private final Log log;
  private final BiFunction<byte[], Integer, ActionItem> writeAction;
  private final ActionItem flushAction;
  private final ActionItem completeAction;
  private final BooleanSupplier isReady;

  /**
   * New write actions will be buffered into this queue if the servlet output stream is not ready or
   * the queue is not drained.
   */
  // SPSC queue would do
  private final Queue<ActionItem> writeChain = new ConcurrentLinkedQueue<>();
  // for a theoretical race condition that onWritePossible() is called immediately after isReady()
  // returns false and before writeState.compareAndSet()

  @Nullable
  private volatile Thread parkingThread;

  AsyncServletOutputStreamWriter(
      AsyncContext asyncContext,
      ServletTransportState transportState,
      InternalLogId logId) throws IOException {
    Logger logger = Logger.getLogger(AsyncServletOutputStreamWriter.class.getName());
    this.log = new Log() {
      @Override
      public boolean isLoggable(Level level) {
        return logger.isLoggable(level);
      }

      @Override
      public void fine(String str, Object... params) {
        if (logger.isLoggable(FINE)) {
          logger.log(FINE, "[" + logId + "]" + str, params);
        }
      }

      @Override
      public void finest(String str, Object... params) {
        if (logger.isLoggable(FINEST)) {
          logger.log(FINEST, "[" + logId + "] " + str, params);
        }
      }
    };

    ServletOutputStream outputStream = asyncContext.getResponse().getOutputStream();
    this.writeAction = (byte[] bytes, Integer numBytes) -> () -> {
      outputStream.write(bytes, 0, numBytes);
      transportState.runOnTransportThread(() -> transportState.onSentBytes(numBytes));
      if (log.isLoggable(Level.FINEST)) {
        log.finest("outbound data: length={0}, bytes={1}", numBytes, toHexString(bytes, numBytes));
      }
    };
    this.flushAction = () -> {
      log.finest("flushBuffer");
      asyncContext.getResponse().flushBuffer();
    };
    this.completeAction = () -> {
      log.fine("call is completing");
      transportState.runOnTransportThread(
          () -> {
            transportState.complete();
            asyncContext.complete();
            log.fine("call completed");
          });
    };
    this.isReady = outputStream::isReady;
  }

  /**
   * Constructor without java.util.logging and javax.servlet.* dependency, so that Lincheck can run.
   *
   * @param writeAction Provides an {@link ActionItem} to write given bytes with specified length.
   * @param isReady Indicates whether the writer can write bytes at the moment (asynchronously).
   */
  @VisibleForTesting
  AsyncServletOutputStreamWriter(
      BiFunction<byte[], Integer, ActionItem> writeAction,
      ActionItem flushAction,
      ActionItem completeAction,
      BooleanSupplier isReady,
      Log log) {
    this.writeAction = writeAction;
    this.flushAction = flushAction;
    this.completeAction = completeAction;
    this.isReady = isReady;
    this.log = log;
  }

  /** Called from application thread. */
  void writeBytes(byte[] bytes, int numBytes) throws IOException {
    runOrBuffer(writeAction.apply(bytes, numBytes));
  }

  /** Called from application thread. */
  void flush() throws IOException {
    runOrBuffer(flushAction);
  }

  /** Called from application thread. */
  void complete() {
    try {
      runOrBuffer(completeAction);
    } catch (IOException ignore) {
      // actually completeAction does not throw IOException
    }
  }

  /** Called from the container thread {@link javax.servlet.WriteListener#onWritePossible()}. */
  void onWritePossible() throws IOException {
    log.finest("onWritePossible: ENTRY. The servlet output stream becomes ready");
    if (writeState.get().readyAndDrained) {
      assureReadyAndDrainedTurnsFalse();
    }
    while (isReady.getAsBoolean()) {
      WriteState curState = writeState.get();

      ActionItem actionItem = writeChain.poll();
      if (actionItem != null) {
        actionItem.run();
        continue;
      }

      if (writeState.compareAndSet(curState, curState.withReadyAndDrained(true))) {
        // state has not changed since.
        log.finest(
            "onWritePossible: EXIT. All data available now is sent out and the servlet output"
                + " stream is still ready");
        return;
      }
      // else, state changed by another thread (runOrBuffer()), need to drain the writeChain
      // again
    }
    log.finest("onWritePossible: EXIT. The servlet output stream becomes not ready");
  }

  private void assureReadyAndDrainedTurnsFalse() {
    // readyAndDrained should have been set to false already.
    // Just in case due to a race condition readyAndDrained is still true at this moment and is
    // being set to false by runOrBuffer() concurrently.
    parkingThread = Thread.currentThread();
    while (writeState.get().readyAndDrained) {
      LockSupport.parkNanos(TimeUnit.MINUTES.toNanos(1)); // should return immediately
    }
    parkingThread = null;
  }

  private void markNotReadyAndUnpark(WriteState curState) {
    boolean successful = writeState.compareAndSet(curState, curState.withReadyAndDrained(false));
    checkState(successful, "Bug: curState is unexpectedly changed by another thread");
    LockSupport.unpark(parkingThread);
    log.finest("the servlet output stream becomes not ready");
  }

  /**
   * Either execute the write action directly, or buffer the action and let the container thread
   * drain it.
   *
   * <p>Called from application thread.
   */
  private void runOrBuffer(ActionItem actionItem) throws IOException {
    WriteState curState = writeState.get();

    // Tomcat Spontaneous State Change Mitigation ---
    // If our cache says true, but the container is secretly not ready,
    // intercept the stale state and sync it before proceeding.
    if (curState.readyAndDrained && !isReady.getAsBoolean()) {
      markNotReadyAndUnpark(curState);
      // Update local state so it gracefully bypasses the 
      // direct write and falls into the buffer block
      curState = writeState.get(); 
    }
    // -------------------------------------------------------    
    if (curState.readyAndDrained) {
      actionItem.run();
      if (actionItem == completeAction) {
        return;
      }
      if (!isReady.getAsBoolean()) {
        markNotReadyAndUnpark(curState);
      }
      return;
    }

    writeChain.offer(actionItem);
    if (!writeState.compareAndSet(curState, curState.withReadyAndDrained(false))) {
      checkState(
          writeState.get().readyAndDrained,
          "Bug: onWritePossible() should have changed readyAndDrained to true, but not");
      ActionItem lastItem = writeChain.poll();
      if (lastItem != null) {
        checkState(lastItem == actionItem, "Bug: lastItem != actionItem");
        runOrBuffer(lastItem);
      }
    }
  }

  /** Write actions, e.g. writeBytes, flush, complete. */
  @FunctionalInterface
  @VisibleForTesting
  interface ActionItem {
    void run() throws IOException;
  }

  @VisibleForTesting // Lincheck test can not run with java.util.logging dependency.
  interface Log {
    default boolean isLoggable(Level level) {
      return false;
    }

    default void fine(String str, Object...params) {}

    default void finest(String str, Object...params) {}
  }

  private static final class WriteState {

    static final WriteState DEFAULT = new WriteState(false);

    /**
     * The servlet output stream is ready and the writeChain is empty.
     *
     * <p>readyAndDrained turns from false to true when:
     * {@code onWritePossible()} exits while currently there is no more data to write, but the last
     * check of {@link javax.servlet.ServletOutputStream#isReady()} is true.
     *
     * <p>readyAndDrained turns from true to false when:
     * {@code runOrBuffer()} exits while either the action item is written directly to the
     * servlet output stream and the check of {@link javax.servlet.ServletOutputStream#isReady()}
     * right after that returns false, or the action item is buffered into the writeChain.
     */
    final boolean readyAndDrained;

    WriteState(boolean readyAndDrained) {
      this.readyAndDrained = readyAndDrained;
    }

    /**
     * Only {@code onWritePossible()} can set readyAndDrained to true, and only {@code
     * runOrBuffer()} can set it to false.
     */
    @CheckReturnValue
    WriteState withReadyAndDrained(boolean readyAndDrained) {
      return new WriteState(readyAndDrained);
    }
  }
}

1	/*
2	* Copyright 2019 The gRPC Authors
3	*
4	* Licensed under the Apache License, Version 2.0 (the "License");
5	* you may not use this file except in compliance with the License.
6	* You may obtain a copy of the License at
7	*
8	* http://www.apache.org/licenses/LICENSE-2.0
9	*
10	* Unless required by applicable law or agreed to in writing, software
11	* distributed under the License is distributed on an "AS IS" BASIS,
12	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	* See the License for the specific language governing permissions and
14	* limitations under the License.
15	*/
16
17	package io.grpc.servlet;
18
19	import static com.google.common.base.Preconditions.checkState;
20	import static io.grpc.servlet.ServletServerStream.toHexString;
21	import static java.util.logging.Level.FINE;
22	import static java.util.logging.Level.FINEST;
23
24	import com.google.common.annotations.VisibleForTesting;
25	import com.google.errorprone.annotations.CheckReturnValue;
26	import io.grpc.InternalLogId;
27	import io.grpc.servlet.ServletServerStream.ServletTransportState;
28	import java.io.IOException;
29	import java.util.Queue;
30	import java.util.concurrent.ConcurrentLinkedQueue;
31	import java.util.concurrent.TimeUnit;
32	import java.util.concurrent.atomic.AtomicReference;
33	import java.util.concurrent.locks.LockSupport;
34	import java.util.function.BiFunction;
35	import java.util.function.BooleanSupplier;
36	import java.util.logging.Level;
37	import java.util.logging.Logger;
38	import javax.annotation.Nullable;
39	import javax.servlet.AsyncContext;
40	import javax.servlet.ServletOutputStream;
41
42	/** Handles write actions from the container thread and the application thread. */
43	final class AsyncServletOutputStreamWriter {
44
45	/**
46	* Memory boundary for write actions.
47	*
48	* <pre>
49	* WriteState curState = writeState.get(); // mark a boundary
50	* doSomething(); // do something within the boundary
51	* boolean successful = writeState.compareAndSet(curState, newState); // try to mark a boundary
52	* if (successful) {
53	* // state has not changed since
54	* return;
55	* } else {
56	* // state is changed by another thread while doSomething(), need recompute
57	* }
58	* </pre>
59	*
60	* <p>There are two threads, the container thread (calling {@code onWritePossible()}) and the
61	* application thread (calling {@code runOrBuffer()}) that read and update the
62	* writeState. Only onWritePossible() may turn {@code readyAndDrained} from false to true, and
63	* only runOrBuffer() may turn it from true to false.
64	*/
65	private final AtomicReference<WriteState> writeState = new AtomicReference<>(WriteState.DEFAULT);	1✔
66
67	private final Log log;
68	private final BiFunction<byte[], Integer, ActionItem> writeAction;
69	private final ActionItem flushAction;
70	private final ActionItem completeAction;
71	private final BooleanSupplier isReady;
72
73	/**
74	* New write actions will be buffered into this queue if the servlet output stream is not ready or
75	* the queue is not drained.
76	*/
77	// SPSC queue would do
78	private final Queue<ActionItem> writeChain = new ConcurrentLinkedQueue<>();	1✔
79	// for a theoretical race condition that onWritePossible() is called immediately after isReady()
80	// returns false and before writeState.compareAndSet()
81
82	@Nullable
83	private volatile Thread parkingThread;
84
85	AsyncServletOutputStreamWriter(
86	AsyncContext asyncContext,
87	ServletTransportState transportState,
88	InternalLogId logId) throws IOException {	1✔
89	Logger logger = Logger.getLogger(AsyncServletOutputStreamWriter.class.getName());	1✔
90	this.log = new Log() {	1✔
91	@Override
92	public boolean isLoggable(Level level) {
93	return logger.isLoggable(level);	1✔
94	}
95
96	@Override
97	public void fine(String str, Object... params) {
98	if (logger.isLoggable(FINE)) {	1✔
99	logger.log(FINE, "[" + logId + "]" + str, params);	×
100	}
101	}	1✔
102
103	@Override
104	public void finest(String str, Object... params) {
105	if (logger.isLoggable(FINEST)) {	1✔
106	logger.log(FINEST, "[" + logId + "] " + str, params);	×
107	}
108	}	1✔
109	};
110
111	ServletOutputStream outputStream = asyncContext.getResponse().getOutputStream();	1✔
112	this.writeAction = (byte[] bytes, Integer numBytes) -> () -> {	1✔
113	outputStream.write(bytes, 0, numBytes);	1✔
114	transportState.runOnTransportThread(() -> transportState.onSentBytes(numBytes));	1✔
115	if (log.isLoggable(Level.FINEST)) {	1✔
116	log.finest("outbound data: length={0}, bytes={1}", numBytes, toHexString(bytes, numBytes));	×
117	}
118	};	1✔
119	this.flushAction = () -> {	1✔
120	log.finest("flushBuffer");	1✔
121	asyncContext.getResponse().flushBuffer();	1✔
122	};	1✔
123	this.completeAction = () -> {	1✔
124	log.fine("call is completing");	1✔
125	transportState.runOnTransportThread(	1✔
126	() -> {
127	transportState.complete();	1✔
128	asyncContext.complete();	1✔
129	log.fine("call completed");	1✔
130	});	1✔
131	};	1✔
132	this.isReady = outputStream::isReady;	1✔
133	}	1✔
134
135	/**
136	* Constructor without java.util.logging and javax.servlet.* dependency, so that Lincheck can run.
137	*
138	* @param writeAction Provides an {@link ActionItem} to write given bytes with specified length.
139	* @param isReady Indicates whether the writer can write bytes at the moment (asynchronously).
140	*/
141	@VisibleForTesting
142	AsyncServletOutputStreamWriter(
143	BiFunction<byte[], Integer, ActionItem> writeAction,
144	ActionItem flushAction,
145	ActionItem completeAction,
146	BooleanSupplier isReady,
147	Log log) {	×
148	this.writeAction = writeAction;	×
149	this.flushAction = flushAction;	×
150	this.completeAction = completeAction;	×
151	this.isReady = isReady;	×
152	this.log = log;	×
153	}	×
154
155	/** Called from application thread. */
156	void writeBytes(byte[] bytes, int numBytes) throws IOException {
157	runOrBuffer(writeAction.apply(bytes, numBytes));	1✔
158	}	1✔
159
160	/** Called from application thread. */
161	void flush() throws IOException {
162	runOrBuffer(flushAction);	1✔
163	}	1✔
164
165	/** Called from application thread. */
166	void complete() {
167	try {
168	runOrBuffer(completeAction);	1✔
169	} catch (IOException ignore) {	×
170	// actually completeAction does not throw IOException
171	}	1✔
172	}	1✔
173
174	/** Called from the container thread {@link javax.servlet.WriteListener#onWritePossible()}. */
175	void onWritePossible() throws IOException {
176	log.finest("onWritePossible: ENTRY. The servlet output stream becomes ready");	1✔
177	if (writeState.get().readyAndDrained) {	1✔
178	assureReadyAndDrainedTurnsFalse();	×
179	}
180	while (isReady.getAsBoolean()) {	1✔
181	WriteState curState = writeState.get();	1✔
182
183	ActionItem actionItem = writeChain.poll();	1✔
184	if (actionItem != null) {	1✔
185	actionItem.run();	1✔
186	continue;	1✔
187	}
188
189	if (writeState.compareAndSet(curState, curState.withReadyAndDrained(true))) {	1✔
190	// state has not changed since.
191	log.finest(	1✔
192	"onWritePossible: EXIT. All data available now is sent out and the servlet output"
193	+ " stream is still ready");
194	return;	1✔
195	}
196	// else, state changed by another thread (runOrBuffer()), need to drain the writeChain
197	// again
198	}	×
199	log.finest("onWritePossible: EXIT. The servlet output stream becomes not ready");	1✔
200	}	1✔
201
202	private void assureReadyAndDrainedTurnsFalse() {
203	// readyAndDrained should have been set to false already.
204	// Just in case due to a race condition readyAndDrained is still true at this moment and is
205	// being set to false by runOrBuffer() concurrently.
206	parkingThread = Thread.currentThread();	1✔
207	while (writeState.get().readyAndDrained) {	1✔
208	LockSupport.parkNanos(TimeUnit.MINUTES.toNanos(1)); // should return immediately	×
209	}
210	parkingThread = null;	×
211	}	×
212
213	private void markNotReadyAndUnpark(WriteState curState) {
214	boolean successful = writeState.compareAndSet(curState, curState.withReadyAndDrained(false));	1✔
215	checkState(successful, "Bug: curState is unexpectedly changed by another thread");	1✔
216	LockSupport.unpark(parkingThread);	1✔
217	log.finest("the servlet output stream becomes not ready");	1✔
218	}	1✔
219
220	/**
221	* Either execute the write action directly, or buffer the action and let the container thread
222	* drain it.
223	*
224	* <p>Called from application thread.
225	*/
226	private void runOrBuffer(ActionItem actionItem) throws IOException {
227	WriteState curState = writeState.get();	1✔
228
229	// Tomcat Spontaneous State Change Mitigation ---
230	// If our cache says true, but the container is secretly not ready,
231	// intercept the stale state and sync it before proceeding.
232	if (curState.readyAndDrained && !isReady.getAsBoolean()) {	1✔
233	markNotReadyAndUnpark(curState);	×
234	// Update local state so it gracefully bypasses the
235	// direct write and falls into the buffer block
236	curState = writeState.get();	×
237	}
238	// -------------------------------------------------------
239	if (curState.readyAndDrained) {	1✔
240	actionItem.run();	1✔
241	if (actionItem == completeAction) {	1✔
242	return;	1✔
243	}
244	if (!isReady.getAsBoolean()) {	1✔
245	markNotReadyAndUnpark(curState);	1✔
246	}
247	return;	1✔
248	}
249
250	writeChain.offer(actionItem);	1✔
251	if (!writeState.compareAndSet(curState, curState.withReadyAndDrained(false))) {	1✔
252	checkState(	×
253	writeState.get().readyAndDrained,	×
254	"Bug: onWritePossible() should have changed readyAndDrained to true, but not");
255	ActionItem lastItem = writeChain.poll();	×
256	if (lastItem != null) {	×
257	checkState(lastItem == actionItem, "Bug: lastItem != actionItem");	×
258	runOrBuffer(lastItem);	×
259	}
260	}
261	}	1✔
262
263	/** Write actions, e.g. writeBytes, flush, complete. */
264	@FunctionalInterface
265	@VisibleForTesting
266	interface ActionItem {
267	void run() throws IOException;
268	}
269
270	@VisibleForTesting // Lincheck test can not run with java.util.logging dependency.
271	interface Log {
272	default boolean isLoggable(Level level) {
273	return false;	×
274	}
275
276	default void fine(String str, Object...params) {}	×
277
278	default void finest(String str, Object...params) {}	×
279	}
280
281	private static final class WriteState {
282
283	static final WriteState DEFAULT = new WriteState(false);	1✔
284
285	/**
286	* The servlet output stream is ready and the writeChain is empty.
287	*
288	* <p>readyAndDrained turns from false to true when:
289	* {@code onWritePossible()} exits while currently there is no more data to write, but the last
290	* check of {@link javax.servlet.ServletOutputStream#isReady()} is true.
291	*
292	* <p>readyAndDrained turns from true to false when:
293	* {@code runOrBuffer()} exits while either the action item is written directly to the
294	* servlet output stream and the check of {@link javax.servlet.ServletOutputStream#isReady()}
295	* right after that returns false, or the action item is buffered into the writeChain.
296	*/
297	final boolean readyAndDrained;
298
299	WriteState(boolean readyAndDrained) {	1✔
300	this.readyAndDrained = readyAndDrained;	1✔
301	}	1✔
302
303	/**
304	* Only {@code onWritePossible()} can set readyAndDrained to true, and only {@code
305	* runOrBuffer()} can set it to false.
306	*/
307	@CheckReturnValue
308	WriteState withReadyAndDrained(boolean readyAndDrained) {
309	return new WriteState(readyAndDrained);	1✔
310	}
311	}
312	}

grpc / grpc-java / #20387

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous