#20385

Committed 31 Jul 2026 10:26AM UTC coverage: 89.151% (-0.03%) from 89.182%

Build # #20385

Build Type

push

github

Committed by

web-flow

Commit Message

Update README etc to reference 1.83.1 (#12960)

Coverage Stats

38311 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

75.25

/../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);	1✔
234	// Update local state so it gracefully bypasses the
235	// direct write and falls into the buffer block
236	curState = writeState.get();	1✔
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 / #20385

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