diff --git a/core/src/main/java/io/grpc/internal/RetriableStream.java b/core/src/main/java/io/grpc/internal/RetriableStream.java index ba9424ea25c..ba31a8420b3 100644 --- a/core/src/main/java/io/grpc/internal/RetriableStream.java +++ b/core/src/main/java/io/grpc/internal/RetriableStream.java @@ -86,7 +86,7 @@ public void uncaughtException(Thread t, Throwable e) { private final boolean isHedging; /** Must be held when updating state, accessing state.buffer, or certain substream attributes. */ - private final Object lock = new Object(); + static final Object lock = new Object(); private final ChannelBufferMeter channelBufferUsed; private final long perRpcBufferLimit; @@ -149,11 +149,11 @@ public void uncaughtException(Thread t, Throwable e) { this.throttle = throttle; } - @SuppressWarnings("GuardedBy") // TODO(b/145386688) this.lock==ScheduledCancellor.lock so ok @Nullable // null if already committed @CheckReturnValue + @GuardedBy("lock") private Runnable commit(final Substream winningSubstream) { - synchronized (lock) { + synchronized (RetriableStream.lock) { if (state.winningSubstream != null) { return null; } @@ -164,21 +164,26 @@ private Runnable commit(final Substream winningSubstream) { // subtract the share of this RPC from channelBufferUsed. channelBufferUsed.addAndGet(-perRpcBufferUsed); - final boolean wasCancelled = (scheduledRetry != null) ? scheduledRetry.isCancelled() : false; + final boolean wasCancelled; final Future retryFuture; - if (scheduledRetry != null) { - retryFuture = scheduledRetry.markCancelled(); - scheduledRetry = null; - } else { - retryFuture = null; + synchronized (scheduledRetry.lock) { + wasCancelled = (scheduledRetry != null) ? scheduledRetry.isCancelled() : false; + if (scheduledRetry != null) { + retryFuture = scheduledRetry.markCancelled(); + scheduledRetry = null; + } else { + retryFuture = null; + } } // cancel the scheduled hedging if it is scheduled prior to the commitment final Future hedgingFuture; - if (scheduledHedging != null) { - hedgingFuture = scheduledHedging.markCancelled(); - scheduledHedging = null; - } else { - hedgingFuture = null; + synchronized (scheduledHedging.lock) { + if (scheduledHedging != null) { + hedgingFuture = scheduledHedging.markCancelled(); + scheduledHedging = null; + } else { + hedgingFuture = null; + } } class CommitTask implements Runnable { @@ -234,6 +239,7 @@ public void run() { * For a failed/closed winning stream, the last closed stream closes the master listener, and * callExecutor scheduling happens-before that. */ + @GuardedBy("lock") private void commitAndRun(Substream winningSubstream) { Runnable postCommitTask = commit(winningSubstream); @@ -385,47 +391,47 @@ public void runWith(Substream substream) { /** Starts the first PRC attempt. */ @Override public final void start(ClientStreamListener listener) { - masterListener = listener; - - Status shutdownStatus = prestart(); + synchronized (RetriableStream.lock) { + masterListener = listener; + Status shutdownStatus = prestart(); - if (shutdownStatus != null) { - cancel(shutdownStatus); - return; - } - - synchronized (lock) { - state.buffer.add(new StartEntry()); - } - - Substream substream = createSubstream(0, false); - if (substream == null) { - return; - } - if (isHedging) { - FutureCanceller scheduledHedgingRef = null; + if (shutdownStatus != null) { + cancel(shutdownStatus); + return; + } synchronized (lock) { - state = state.addActiveHedge(substream); - if (hasPotentialHedging(state) - && (throttle == null || throttle.isAboveThreshold())) { - scheduledHedging = scheduledHedgingRef = new FutureCanceller(lock); - } + state.buffer.add(new StartEntry()); } - if (scheduledHedgingRef != null) { - scheduledHedgingRef.setFuture( - scheduledExecutorService.schedule( - new HedgingRunnable(scheduledHedgingRef), - hedgingPolicy.hedgingDelayNanos, - TimeUnit.NANOSECONDS)); + Substream substream = createSubstream(0, false); + if (substream == null) { + return; } - } + if (isHedging) { + FutureCanceller scheduledHedgingRef = null; + + synchronized (lock) { + state = state.addActiveHedge(substream); + if (hasPotentialHedging(state) + && (throttle == null || throttle.isAboveThreshold())) { + scheduledHedging = scheduledHedgingRef = new FutureCanceller(lock); + } + } - drain(substream); + if (scheduledHedgingRef != null) { + scheduledHedgingRef.setFuture( + scheduledExecutorService.schedule( + new HedgingRunnable(scheduledHedgingRef), + hedgingPolicy.hedgingDelayNanos, + TimeUnit.NANOSECONDS)); + } + } + drain(substream); + } } - @SuppressWarnings("GuardedBy") // TODO(b/145386688) this.lock==ScheduledCancellor.lock so ok + @GuardedBy("lock") private void pushbackHedging(@Nullable Integer delayMillis) { if (delayMillis == null) { return; @@ -439,7 +445,7 @@ private void pushbackHedging(@Nullable Integer delayMillis) { FutureCanceller future; Future futureToBeCancelled; - synchronized (lock) { + synchronized (scheduledHedging.lock) { if (scheduledHedging == null) { return; } @@ -477,23 +483,24 @@ public void run() { } callExecutor.execute( new Runnable() { - @SuppressWarnings("GuardedBy") //TODO(b/145386688) lock==ScheduledCancellor.lock so ok @Override public void run() { boolean cancelled = false; FutureCanceller future = null; - synchronized (lock) { + synchronized (scheduledHedgingRef.lock) { if (scheduledHedgingRef.isCancelled()) { cancelled = true; } else { state = state.addActiveHedge(newSubstream); - if (hasPotentialHedging(state) - && (throttle == null || throttle.isAboveThreshold())) { - scheduledHedging = future = new FutureCanceller(lock); - } else { - state = state.freezeHedging(); - scheduledHedging = null; + synchronized (RetriableStream.lock) { + if (hasPotentialHedging(state) + && (throttle == null || throttle.isAboveThreshold())) { + scheduledHedging = future = new FutureCanceller(lock); + } else { + state = state.freezeHedging(); + scheduledHedging = null; + } } } } @@ -517,33 +524,36 @@ public void run() { } } + @GuardedBy("lock") @Override public final void cancel(final Status reason) { - Substream noopSubstream = new Substream(0 /* previousAttempts doesn't matter here */); - noopSubstream.stream = new NoopClientStream(); - Runnable runnable = commit(noopSubstream); + synchronized (RetriableStream.lock) { + Substream noopSubstream = new Substream(0 /* previousAttempts doesn't matter here */); + noopSubstream.stream = new NoopClientStream(); + Runnable runnable = commit(noopSubstream); + + if (runnable != null) { + synchronized (lock) { + state = state.substreamDrained(noopSubstream); + } + runnable.run(); + safeCloseMasterListener(reason, RpcProgress.PROCESSED, new Metadata()); + return; + } - if (runnable != null) { + Substream winningSubstreamToCancel = null; synchronized (lock) { - state = state.substreamDrained(noopSubstream); + if (state.drainedSubstreams.contains(state.winningSubstream)) { + winningSubstreamToCancel = state.winningSubstream; + } else { // the winningSubstream will be cancelled while draining + cancellationStatus = reason; + } + state = state.cancelled(); } - runnable.run(); - safeCloseMasterListener(reason, RpcProgress.PROCESSED, new Metadata()); - return; - } - - Substream winningSubstreamToCancel = null; - synchronized (lock) { - if (state.drainedSubstreams.contains(state.winningSubstream)) { - winningSubstreamToCancel = state.winningSubstream; - } else { // the winningSubstream will be cancelled while draining - cancellationStatus = reason; + if (winningSubstreamToCancel != null) { + winningSubstreamToCancel.stream.cancel(reason); } - state = state.cancelled(); - } - if (winningSubstreamToCancel != null) { - winningSubstreamToCancel.stream.cancel(reason); - } + } } private void delayOrExecute(BufferEntry bufferEntry) { @@ -815,10 +825,10 @@ private boolean hasPotentialHedging(State state) { && !state.hedgingFrozen; } - @SuppressWarnings("GuardedBy") // TODO(b/145386688) this.lock==ScheduledCancellor.lock so ok + @GuardedBy("lock") private void freezeHedging() { Future futureToBeCancelled = null; - synchronized (lock) { + synchronized (scheduledHedging.lock) { if (scheduledHedging != null) { futureToBeCancelled = scheduledHedging.markCancelled(); scheduledHedging = null; @@ -870,6 +880,7 @@ private final class Sublistener implements ClientStreamListener { this.substream = substream; } + @GuardedBy("lock") @Override public void headersRead(final Metadata headers) { if (substream.previousAttemptCount > 0) { @@ -891,6 +902,7 @@ public void run() { } } + @GuardedBy("lock") @Override public void closed( final Status status, final RpcProgress rpcProgress, final Metadata trailers) { diff --git a/cronet/src/main/java/io/grpc/cronet/CronetClientTransport.java b/cronet/src/main/java/io/grpc/cronet/CronetClientTransport.java index b0b18620d0c..1681139ccaa 100644 --- a/cronet/src/main/java/io/grpc/cronet/CronetClientTransport.java +++ b/cronet/src/main/java/io/grpc/cronet/CronetClientTransport.java @@ -151,13 +151,14 @@ public void run() { return new StartCallback().clientStream; } - @SuppressWarnings("GuardedBy") @GuardedBy("lock") private void startStream(CronetClientStream stream) { - streams.add(stream); - // TODO(b/145386688): This access should be guarded by 'stream.transportState().lock'; instead - // found: 'this.lock' - stream.transportState().start(streamFactory); + synchronized (lock) { + streams.add(stream); + synchronized (stream.transportState().lock) { + stream.transportState().start(streamFactory); + } + } } @Override diff --git a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java index 9d9fe160715..654e995bc5c 100644 --- a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java +++ b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientStream.java @@ -147,7 +147,7 @@ public void writeHeaders(Metadata metadata, byte[] payload) { useGet = true; defaultPath += "?" + BaseEncoding.base64().encode(payload); } - synchronized (state.lock) { + synchronized (OkHttpClientStream.this.state.lock) { state.streamReady(metadata, defaultPath); } } @@ -178,7 +178,7 @@ public void writeFrame( @Override public void cancel(Status reason) { try (TaskCloseable ignore = PerfMark.traceTask("OkHttpClientStream$Sink.cancel")) { - synchronized (state.lock) { + synchronized (OkHttpClientStream.this.state.lock) { state.cancel(reason, true, null); } } @@ -188,7 +188,7 @@ public void cancel(Status reason) { class TransportState extends Http2ClientStreamTransportState implements OutboundFlowController.Stream { private final int initialWindowSize; - private final Object lock; + final Object lock; @GuardedBy("lock") private List
requestHeaders; @GuardedBy("lock") @@ -236,16 +236,14 @@ public TransportState( tag = PerfMark.createTag(methodName); } - @SuppressWarnings("GuardedBy") @GuardedBy("lock") public void start(int streamId) { - checkState(id == ABSENT_ID, "the stream has been started with id %s", streamId); - id = streamId; - outboundFlowState = outboundFlow.createState(this, streamId); - // TODO(b/145386688): This access should be guarded by 'OkHttpClientStream.this.state.lock'; - // instead found: 'this.lock' - state.onStreamAllocated(); - + synchronized (OkHttpClientStream.this.state.lock) { + checkState(id == ABSENT_ID, "the stream has been started with id %s", streamId); + id = streamId; + outboundFlowState = outboundFlow.createState(this, streamId); + state.onStreamAllocated(); + } if (canStart) { // Only happens when the stream has neither been started nor cancelled. frameWriter.synStream(useGet, false, id, 0, requestHeaders); @@ -352,28 +350,27 @@ private void onEndOfStream() { } } - @SuppressWarnings("GuardedBy") @GuardedBy("lock") private void cancel(Status reason, boolean stopDelivery, Metadata trailers) { - if (cancelSent) { - return; - } - cancelSent = true; - if (canStart) { - // stream is pending. - // TODO(b/145386688): This access should be guarded by 'this.transport.lock'; instead found: - // 'this.lock' - transport.removePendingStream(OkHttpClientStream.this); - // release holding data, so they can be GCed or returned to pool earlier. - requestHeaders = null; - pendingData.clear(); - canStart = false; - transportReportStatus(reason, true, trailers != null ? trailers : new Metadata()); - } else { - // If pendingData is null, start must have already been called, which means synStream has - // been called as well. - transport.finishStream( - id(), reason, PROCESSED, stopDelivery, ErrorCode.CANCEL, trailers); + synchronized (OkHttpClientTransport.lock) { + if (cancelSent) { + return; + } + cancelSent = true; + if (canStart) { + // stream is pending. + transport.removePendingStream(OkHttpClientStream.this); + // release holding data, so they can be GCed or returned to pool earlier. + requestHeaders = null; + pendingData.clear(); + canStart = false; + transportReportStatus(reason, true, trailers != null ? trailers : new Metadata()); + } else { + // If pendingData is null, start must have already been called, which means synStream has + // been called as well. + transport.finishStream( + id(), reason, PROCESSED, stopDelivery, ErrorCode.CANCEL, trailers); + } } } @@ -396,20 +393,19 @@ private void sendBuffer(Buffer buffer, boolean endOfStream, boolean flush) { } } - @SuppressWarnings("GuardedBy") @GuardedBy("lock") private void streamReady(Metadata metadata, String path) { - requestHeaders = - Headers.createRequestHeaders( - metadata, - path, - authority, - userAgent, - useGet, - transport.isUsingPlaintext()); - // TODO(b/145386688): This access should be guarded by 'this.transport.lock'; instead found: - // 'this.lock' - transport.streamReadyToStart(OkHttpClientStream.this); + synchronized (OkHttpClientTransport.lock) { + requestHeaders = + Headers.createRequestHeaders( + metadata, + path, + authority, + userAgent, + useGet, + transport.isUsingPlaintext()); + transport.streamReadyToStart(OkHttpClientStream.this); + } } Tag tag() { diff --git a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java index 2f6b836dc3a..fe0a0d26337 100644 --- a/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java +++ b/okhttp/src/main/java/io/grpc/okhttp/OkHttpClientTransport.java @@ -156,7 +156,7 @@ private static Map buildErrorCodeToStatusMap() { @GuardedBy("lock") private ExceptionHandlingFrameWriter frameWriter; private OutboundFlowController outboundFlow; - private final Object lock = new Object(); + static Object lock = new Object(); private final InternalLogId logId; @GuardedBy("lock") private int nextStreamId; @@ -429,16 +429,15 @@ void streamReadyToStart(OkHttpClientStream clientStream) { } } - @SuppressWarnings("GuardedBy") @GuardedBy("lock") private void startStream(OkHttpClientStream stream) { Preconditions.checkState( stream.transportState().id() == OkHttpClientStream.ABSENT_ID, "StreamId already assigned"); streams.put(nextStreamId, stream); setInUse(stream); - // TODO(b/145386688): This access should be guarded by 'stream.transportState().lock'; instead - // found: 'this.lock' - stream.transportState().start(nextStreamId); + synchronized (stream.transportState().lock) { + stream.transportState().start(nextStreamId); + } // For unary and server streaming, there will be a data frame soon, no need to flush the header. if ((stream.getType() != MethodType.UNARY && stream.getType() != MethodType.SERVER_STREAMING) || stream.useGet()) { @@ -1182,7 +1181,6 @@ public void run() { /** * Handle an HTTP2 DATA frame. */ - @SuppressWarnings("GuardedBy") @Override public void data(boolean inFinished, int streamId, BufferedSource in, int length, int paddedLength) @@ -1208,9 +1206,7 @@ public void data(boolean inFinished, int streamId, BufferedSource in, int length buf.write(in.getBuffer(), length); PerfMark.event("OkHttpClientTransport$ClientFrameHandler.data", stream.transportState().tag()); - synchronized (lock) { - // TODO(b/145386688): This access should be guarded by 'stream.transportState().lock'; - // instead found: 'OkHttpClientTransport.this.lock' + synchronized (stream.transportState().lock) { stream.transportState().transportDataReceived(buf, inFinished, paddedLength - length); } } @@ -1228,7 +1224,6 @@ public void data(boolean inFinished, int streamId, BufferedSource in, int length /** * Handle HTTP2 HEADER and CONTINUATION frames. */ - @SuppressWarnings("GuardedBy") @Override public void headers(boolean outFinished, boolean inFinished, @@ -1251,24 +1246,28 @@ public void headers(boolean outFinished, metadataSize)); } } - synchronized (lock) { - OkHttpClientStream stream = streams.get(streamId); + OkHttpClientStream stream; + synchronized (OkHttpClientTransport.lock) { + stream = streams.get(streamId); if (stream == null) { if (mayHaveCreatedStream(streamId)) { frameWriter.rstStream(streamId, ErrorCode.STREAM_CLOSED); } else { unknownStream = true; } - } else { + } + } + if (stream != null) { + synchronized (stream.transportState().lock) { if (failedStatus == null) { PerfMark.event("OkHttpClientTransport$ClientFrameHandler.headers", stream.transportState().tag()); - // TODO(b/145386688): This access should be guarded by 'stream.transportState().lock'; - // instead found: 'OkHttpClientTransport.this.lock' stream.transportState().transportHeadersReceived(headerBlock, inFinished); } else { - if (!inFinished) { - frameWriter.rstStream(streamId, ErrorCode.CANCEL); + synchronized (OkHttpClientTransport.lock) { + if (!inFinished) { + frameWriter.rstStream(streamId, ErrorCode.CANCEL); + } } stream.transportState().transportReportStatus(failedStatus, false, new Metadata()); }