#19095

Committed 11 Mar 2024 08:40PM CUT coverage: 88.305%. Remained the same

Build # #19095

Build Type

push

github

Committed by

ejona86

Commit Message

rls: Fix a local and remote race

The local race passes `rlsPicker` to the channel before
CachingRlsLbClient is finished constructing. `RlsPicker` can use
multiple of the fields not yet initialized. This seems not to be
happening in practice, because it appears like it would break things
very loudly (e.g., NPE).

The remote race seems incredibly hard to hit, because it requires an RPC
to complete before the pending data tracking the RPC is added to a map.
But with if a system is at 100% CPU utilization, maybe it can be hit. If
it is hit, all RPCs needing the impacted cache entry will forever be
buffered.

Run Details

31064 of 35178 relevant lines covered (88.31%)

0.88 hits per line

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

83.16

/../okhttp/src/main/java/io/grpc/okhttp/ExceptionHandlingFrameWriter.java

/*
 * Copyright 2018 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.okhttp;

import static com.google.common.base.Preconditions.checkNotNull;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import io.grpc.okhttp.internal.framed.ErrorCode;
import io.grpc.okhttp.internal.framed.FrameWriter;
import io.grpc.okhttp.internal.framed.Header;
import io.grpc.okhttp.internal.framed.Settings;
import java.io.IOException;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import okio.Buffer;
import okio.ByteString;

/**
 * FrameWriter that propagates IOExceptions via callback instead of throwing. This allows
 * centralized handling of errors. Exceptions only impact the single call that throws them; callers
 * should be sure to kill the connection after an exception (potentially after sending a GOAWAY) as
 * otherwise additional frames after the failed/omitted one could cause HTTP/2 confusion.
 */
final class ExceptionHandlingFrameWriter implements FrameWriter {

  private static final Logger log = Logger.getLogger(OkHttpClientTransport.class.getName());

  private final TransportExceptionHandler transportExceptionHandler;

  private final FrameWriter frameWriter;

  private final OkHttpFrameLogger frameLogger =
      new OkHttpFrameLogger(Level.FINE, OkHttpClientTransport.class);

  ExceptionHandlingFrameWriter(
      TransportExceptionHandler transportExceptionHandler, FrameWriter frameWriter) {
    this.transportExceptionHandler =
        checkNotNull(transportExceptionHandler, "transportExceptionHandler");
    this.frameWriter = Preconditions.checkNotNull(frameWriter, "frameWriter");
  }

  @Override
  public void connectionPreface() {
    try {
      frameWriter.connectionPreface();
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void ackSettings(Settings peerSettings) {
    frameLogger.logSettingsAck(OkHttpFrameLogger.Direction.OUTBOUND);
    try {
      frameWriter.ackSettings(peerSettings);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void pushPromise(int streamId, int promisedStreamId, List<Header> requestHeaders) {
    frameLogger.logPushPromise(OkHttpFrameLogger.Direction.OUTBOUND,
        streamId, promisedStreamId, requestHeaders);
    try {
      frameWriter.pushPromise(streamId, promisedStreamId, requestHeaders);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void flush() {
    try {
      frameWriter.flush();
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void synStream(
      boolean outFinished,
      boolean inFinished,
      int streamId,
      int associatedStreamId,
      List<Header> headerBlock) {
    try {
      frameWriter.synStream(outFinished, inFinished, streamId, associatedStreamId, headerBlock);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void synReply(boolean outFinished, int streamId,
      List<Header> headerBlock) {
    try {
      frameWriter.synReply(outFinished, streamId, headerBlock);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void headers(int streamId, List<Header> headerBlock) {
    frameLogger.logHeaders(OkHttpFrameLogger.Direction.OUTBOUND, streamId, headerBlock, false);
    try {
      frameWriter.headers(streamId, headerBlock);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void rstStream(int streamId, ErrorCode errorCode) {
    frameLogger.logRstStream(OkHttpFrameLogger.Direction.OUTBOUND, streamId, errorCode);
    try {
      frameWriter.rstStream(streamId, errorCode);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public int maxDataLength() {
    return frameWriter.maxDataLength();
  }

  @Override
  public void data(boolean outFinished, int streamId, Buffer source, int byteCount) {
    frameLogger.logData(OkHttpFrameLogger.Direction.OUTBOUND,
        streamId, source.buffer(), byteCount, outFinished);
    try {
      frameWriter.data(outFinished, streamId, source, byteCount);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void settings(Settings okHttpSettings) {
    frameLogger.logSettings(OkHttpFrameLogger.Direction.OUTBOUND, okHttpSettings);
    try {
      frameWriter.settings(okHttpSettings);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void ping(boolean ack, int payload1, int payload2) {
    if (ack) {
      frameLogger.logPingAck(OkHttpFrameLogger.Direction.OUTBOUND,
          ((long) payload1 << 32) | (payload2 & 0xFFFFFFFFL));
    } else {
      frameLogger.logPing(OkHttpFrameLogger.Direction.OUTBOUND,
          ((long) payload1 << 32) | (payload2 & 0xFFFFFFFFL));
    }
    try {
      frameWriter.ping(ack, payload1, payload2);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void goAway(int lastGoodStreamId, ErrorCode errorCode,
      byte[] debugData) {
    frameLogger.logGoAway(OkHttpFrameLogger.Direction.OUTBOUND,
        lastGoodStreamId, errorCode, ByteString.of(debugData));
    try {
      frameWriter.goAway(lastGoodStreamId, errorCode, debugData);
      // Flush it since after goAway, we are likely to close this writer.
      frameWriter.flush();
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void windowUpdate(int streamId, long windowSizeIncrement) {
    frameLogger.logWindowsUpdate(OkHttpFrameLogger.Direction.OUTBOUND,
        streamId, windowSizeIncrement);
    try {
      frameWriter.windowUpdate(streamId, windowSizeIncrement);
    } catch (IOException e) {
      transportExceptionHandler.onException(e);
    }
  }

  @Override
  public void close() {
    try {
      frameWriter.close();
    } catch (IOException e) {
      log.log(getLogLevel(e), "Failed closing connection", e);
    }
  }

  /**
   * Accepts a throwable and returns the appropriate logging level. Uninteresting exceptions
   * should not clutter the log.
   */
  @VisibleForTesting
  static Level getLogLevel(Throwable t) {
    if (t.getClass().equals(IOException.class)) {
      return Level.FINE;
    }
    return Level.INFO;
  }

  /** A class that handles transport exception. */
  interface TransportExceptionHandler {
    /** Handles exception. */
    void onException(Throwable throwable);
  }
}

1	/*
2	* Copyright 2018 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.okhttp;
18
19	import static com.google.common.base.Preconditions.checkNotNull;
20
21	import com.google.common.annotations.VisibleForTesting;
22	import com.google.common.base.Preconditions;
23	import io.grpc.okhttp.internal.framed.ErrorCode;
24	import io.grpc.okhttp.internal.framed.FrameWriter;
25	import io.grpc.okhttp.internal.framed.Header;
26	import io.grpc.okhttp.internal.framed.Settings;
27	import java.io.IOException;
28	import java.util.List;
29	import java.util.logging.Level;
30	import java.util.logging.Logger;
31	import okio.Buffer;
32	import okio.ByteString;
33
34	/**
35	* FrameWriter that propagates IOExceptions via callback instead of throwing. This allows
36	* centralized handling of errors. Exceptions only impact the single call that throws them; callers
37	* should be sure to kill the connection after an exception (potentially after sending a GOAWAY) as
38	* otherwise additional frames after the failed/omitted one could cause HTTP/2 confusion.
39	*/
40	final class ExceptionHandlingFrameWriter implements FrameWriter {
41
42	private static final Logger log = Logger.getLogger(OkHttpClientTransport.class.getName());	1✔
43
44	private final TransportExceptionHandler transportExceptionHandler;
45
46	private final FrameWriter frameWriter;
47
48	private final OkHttpFrameLogger frameLogger =	1✔
49	new OkHttpFrameLogger(Level.FINE, OkHttpClientTransport.class);
50
51	ExceptionHandlingFrameWriter(
52	TransportExceptionHandler transportExceptionHandler, FrameWriter frameWriter) {	1✔
53	this.transportExceptionHandler =	1✔
54	checkNotNull(transportExceptionHandler, "transportExceptionHandler");	1✔
55	this.frameWriter = Preconditions.checkNotNull(frameWriter, "frameWriter");	1✔
56	}	1✔
57
58	@Override
59	public void connectionPreface() {
60	try {
61	frameWriter.connectionPreface();	1✔
62	} catch (IOException e) {	×
63	transportExceptionHandler.onException(e);	×
64	}	1✔
65	}	1✔
66
67	@Override
68	public void ackSettings(Settings peerSettings) {
69	frameLogger.logSettingsAck(OkHttpFrameLogger.Direction.OUTBOUND);	1✔
70	try {
71	frameWriter.ackSettings(peerSettings);	1✔
72	} catch (IOException e) {	×
73	transportExceptionHandler.onException(e);	×
74	}	1✔
75	}	1✔
76
77	@Override
78	public void pushPromise(int streamId, int promisedStreamId, List<Header> requestHeaders) {
79	frameLogger.logPushPromise(OkHttpFrameLogger.Direction.OUTBOUND,	1✔
80	streamId, promisedStreamId, requestHeaders);
81	try {
82	frameWriter.pushPromise(streamId, promisedStreamId, requestHeaders);	1✔
83	} catch (IOException e) {	×
84	transportExceptionHandler.onException(e);	×
85	}	1✔
86	}	1✔
87
88	@Override
89	public void flush() {
90	try {
91	frameWriter.flush();	1✔
92	} catch (IOException e) {	1✔
93	transportExceptionHandler.onException(e);	1✔
94	}	1✔
95	}	1✔
96
97	@Override
98	public void synStream(
99	boolean outFinished,
100	boolean inFinished,
101	int streamId,
102	int associatedStreamId,
103	List<Header> headerBlock) {
104	try {
105	frameWriter.synStream(outFinished, inFinished, streamId, associatedStreamId, headerBlock);	1✔
106	} catch (IOException e) {	×
107	transportExceptionHandler.onException(e);	×
108	}	1✔
109	}	1✔
110
111	@Override
112	public void synReply(boolean outFinished, int streamId,
113	List<Header> headerBlock) {
114	try {
115	frameWriter.synReply(outFinished, streamId, headerBlock);	1✔
116	} catch (IOException e) {	1✔
117	transportExceptionHandler.onException(e);	1✔
118	}	1✔
119	}	1✔
120
121	@Override
122	public void headers(int streamId, List<Header> headerBlock) {
123	frameLogger.logHeaders(OkHttpFrameLogger.Direction.OUTBOUND, streamId, headerBlock, false);	1✔
124	try {
125	frameWriter.headers(streamId, headerBlock);	1✔
126	} catch (IOException e) {	×
127	transportExceptionHandler.onException(e);	×
128	}	1✔
129	}	1✔
130
131	@Override
132	public void rstStream(int streamId, ErrorCode errorCode) {
133	frameLogger.logRstStream(OkHttpFrameLogger.Direction.OUTBOUND, streamId, errorCode);	1✔
134	try {
135	frameWriter.rstStream(streamId, errorCode);	1✔
136	} catch (IOException e) {	1✔
137	transportExceptionHandler.onException(e);	1✔
138	}	1✔
139	}	1✔
140
141	@Override
142	public int maxDataLength() {
143	return frameWriter.maxDataLength();	1✔
144	}
145
146	@Override
147	public void data(boolean outFinished, int streamId, Buffer source, int byteCount) {
148	frameLogger.logData(OkHttpFrameLogger.Direction.OUTBOUND,	1✔
149	streamId, source.buffer(), byteCount, outFinished);	1✔
150	try {
151	frameWriter.data(outFinished, streamId, source, byteCount);	1✔
152	} catch (IOException e) {	×
153	transportExceptionHandler.onException(e);	×
154	}	1✔
155	}	1✔
156
157	@Override
158	public void settings(Settings okHttpSettings) {
159	frameLogger.logSettings(OkHttpFrameLogger.Direction.OUTBOUND, okHttpSettings);	1✔
160	try {
161	frameWriter.settings(okHttpSettings);	1✔
162	} catch (IOException e) {	×
163	transportExceptionHandler.onException(e);	×
164	}	1✔
165	}	1✔
166
167	@Override
168	public void ping(boolean ack, int payload1, int payload2) {
169	if (ack) {	1✔
170	frameLogger.logPingAck(OkHttpFrameLogger.Direction.OUTBOUND,	1✔
171	((long) payload1 << 32) \| (payload2 & 0xFFFFFFFFL));
172	} else {
173	frameLogger.logPing(OkHttpFrameLogger.Direction.OUTBOUND,	1✔
174	((long) payload1 << 32) \| (payload2 & 0xFFFFFFFFL));
175	}
176	try {
177	frameWriter.ping(ack, payload1, payload2);	1✔
178	} catch (IOException e) {	1✔
179	transportExceptionHandler.onException(e);	1✔
180	}	1✔
181	}	1✔
182
183	@Override
184	public void goAway(int lastGoodStreamId, ErrorCode errorCode,
185	byte[] debugData) {
186	frameLogger.logGoAway(OkHttpFrameLogger.Direction.OUTBOUND,	1✔
187	lastGoodStreamId, errorCode, ByteString.of(debugData));	1✔
188	try {
189	frameWriter.goAway(lastGoodStreamId, errorCode, debugData);	1✔
190	// Flush it since after goAway, we are likely to close this writer.
191	frameWriter.flush();	1✔
192	} catch (IOException e) {	1✔
193	transportExceptionHandler.onException(e);	1✔
194	}	1✔
195	}	1✔
196
197	@Override
198	public void windowUpdate(int streamId, long windowSizeIncrement) {
199	frameLogger.logWindowsUpdate(OkHttpFrameLogger.Direction.OUTBOUND,	1✔
200	streamId, windowSizeIncrement);
201	try {
202	frameWriter.windowUpdate(streamId, windowSizeIncrement);	1✔
203	} catch (IOException e) {	1✔
204	transportExceptionHandler.onException(e);	1✔
205	}	1✔
206	}	1✔
207
208	@Override
209	public void close() {
210	try {
211	frameWriter.close();	1✔
212	} catch (IOException e) {	×
213	log.log(getLogLevel(e), "Failed closing connection", e);	×
214	}	1✔
215	}	1✔
216
217	/**
218	* Accepts a throwable and returns the appropriate logging level. Uninteresting exceptions
219	* should not clutter the log.
220	*/
221	@VisibleForTesting
222	static Level getLogLevel(Throwable t) {
223	if (t.getClass().equals(IOException.class)) {	1✔
224	return Level.FINE;	1✔
225	}
226	return Level.INFO;	1✔
227	}
228
229	/** A class that handles transport exception. */
230	interface TransportExceptionHandler {
231	/** Handles exception. */
232	void onException(Throwable throwable);
233	}
234	}

grpc / grpc-java / #19095

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