#19916

Committed 18 Jul 2025 04:05PM UTC coverage: 88.586% (+0.006%) from 88.58%

Build # #19916

Build Type

push

github

Committed by

ejona86

Commit Message

LBs should avoid calling LBs after lb.shutdown()

LoadBalancers shouldn't be called after shutdown(), but RingHashLb could
have enqueued work to the SynchronizationContext that executed after
shutdown(). This commit fixes problems discovered when auditing all LBs
usage of the syncContext for that type of problem.

Similarly, PickFirstLb could have requested a new connection after
shutdown(). We want to avoid that sort of thing too.

RingHashLb's test changed from CONNECTING to TRANSIENT_FAILURE to get
the latest picker. Because two subchannels have failed it will be in
TRANSIENT_FAILURE. Previously the test was using an older picker with
out-of-date subchannelView, and the verifyConnection() was too imprecise
to notice it was creating the wrong subchannel.

As discovered in b/430347751, where ClusterImplLb was seeing a new
subchannel being called after the child LB was shutdown (the shutdown
itself had been caused by RingHashConfig not implementing equals() and
was fixed by a8de9f07ab, which caused ClusterResolverLb to replace its
state):

```
java.lang.NullPointerException
	at io.grpc.xds.ClusterImplLoadBalancer$ClusterImplLbHelper.createClusterLocalityFromAttributes(ClusterImplLoadBalancer.java:322)
	at io.grpc.xds.ClusterImplLoadBalancer$ClusterImplLbHelper.createSubchannel(ClusterImplLoadBalancer.java:236)
	at io.grpc.util.ForwardingLoadBalancerHelper.createSubchannel(ForwardingLoadBalancerHelper.java:47)
	at io.grpc.util.ForwardingLoadBalancerHelper.createSubchannel(ForwardingLoadBalancerHelper.java:47)
	at io.grpc.internal.PickFirstLeafLoadBalancer.createNewSubchannel(PickFirstLeafLoadBalancer.java:527)
	at io.grpc.internal.PickFirstLeafLoadBalancer.requestConnection(PickFirstLeafLoadBalancer.java:459)
	at io.grpc.internal.PickFirstLeafLoadBalancer.acceptResolvedAddresses(PickFirstLeafLoadBalancer.java:174)
	at io.grpc.xds.LazyLoadBalancer$LazyDelegate.activate(LazyLoadBalancer.java:64)
	at io.grpc.xds.LazyLoadBalancer$LazyDelegate.requestConnec... (continued)

Run Details

34653 of 39118 relevant lines covered (88.59%)

0.89 hits per line

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

83.02

/../core/src/main/java/io/grpc/internal/Http2Ping.java

/*
 * Copyright 2015 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.internal;

import com.google.common.base.Stopwatch;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import io.grpc.Status;
import io.grpc.internal.ClientTransport.PingCallback;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * Represents an outstanding PING operation on an HTTP/2 channel. This can be used by HTTP/2-based
 * transports to implement {@link ClientTransport#ping}.
 *
 * <p>A typical transport need only support one outstanding ping at a time. So, if a ping is
 * requested while an operation is already in progress, the given callback is notified when the
 * existing operation completes.
 */
public class Http2Ping {
  private static final Logger log = Logger.getLogger(Http2Ping.class.getName());

  /**
   * The PING frame includes 8 octets of payload data, e.g. 64 bits.
   */
  private final long data;

  /**
   * Used to measure elapsed time.
   */
  private final Stopwatch stopwatch;

  /**
   * The registered callbacks and the executor used to invoke them.
   */
  @GuardedBy("this") private Map<PingCallback, Executor> callbacks
      = new LinkedHashMap<>();

  /**
   * False until the operation completes, either successfully (other side sent acknowledgement) or
   * unsuccessfully.
   */
  @GuardedBy("this") private boolean completed;

  /**
   * If non-null, indicates the ping failed.
   */
  @GuardedBy("this") private Status failureCause;

  /**
   * The round-trip time for the ping, in nanoseconds. This value is only meaningful when
   * {@link #completed} is true and {@link #failureCause} is null.
   */
  @GuardedBy("this") private long roundTripTimeNanos;

  /**
   * Creates a new ping operation. The caller is responsible for sending a ping on an HTTP/2 channel
   * using the given payload. The caller is also responsible for starting the stopwatch when the
   * PING frame is sent.
   *
   * @param data the ping payload
   * @param stopwatch a stopwatch for measuring round-trip time
   */
  public Http2Ping(long data, Stopwatch stopwatch) {
    this.data = data;
    this.stopwatch = stopwatch;
  }

  /**
   * Registers a callback that is invoked when the ping operation completes. If this ping operation
   * is already completed, the callback is invoked immediately.
   *
   * @param callback the callback to invoke
   * @param executor the executor to use
   */
  public void addCallback(final ClientTransport.PingCallback callback, Executor executor) {
    Runnable runnable;
    synchronized (this) {
      if (!completed) {
        callbacks.put(callback, executor);
        return;
      }
      // otherwise, invoke callback immediately (but not while holding lock)
      runnable = this.failureCause != null ? asRunnable(callback, failureCause)
                                           : asRunnable(callback, roundTripTimeNanos);
    }
    doExecute(executor, runnable);
  }

  /**
   * Returns the expected ping payload for this outstanding operation.
   *
   * @return the expected payload for this outstanding ping
   */
  public long payload() {
    return data;
  }

  /**
   * Completes this operation successfully. The stopwatch given during construction is used to
   * measure the elapsed time. Registered callbacks are invoked and provided the measured elapsed
   * time.
   *
   * @return true if the operation is marked as complete; false if it was already complete
   */
  public boolean complete() {
    Map<ClientTransport.PingCallback, Executor> callbacks;
    long roundTripTimeNanos;
    synchronized (this) {
      if (completed) {
        return false;
      }
      completed = true;
      roundTripTimeNanos = this.roundTripTimeNanos = stopwatch.elapsed(TimeUnit.NANOSECONDS);
      callbacks = this.callbacks;
      this.callbacks = null;
    }
    for (Map.Entry<ClientTransport.PingCallback, Executor> entry : callbacks.entrySet()) {
      doExecute(entry.getValue(), asRunnable(entry.getKey(), roundTripTimeNanos));
    }
    return true;
  }

  /**
   * Completes this operation exceptionally. Registered callbacks are invoked and provided the
   * given throwable as the cause of failure.
   *
   * @param failureCause the cause of failure
   */
  public void failed(Status failureCause) {
    Map<ClientTransport.PingCallback, Executor> callbacks;
    synchronized (this) {
      if (completed) {
        return;
      }
      completed = true;
      this.failureCause = failureCause;
      callbacks = this.callbacks;
      this.callbacks = null;
    }
    for (Map.Entry<ClientTransport.PingCallback, Executor> entry : callbacks.entrySet()) {
      notifyFailed(entry.getKey(), entry.getValue(), failureCause);
    }
  }

  /**
   * Notifies the given callback that the ping operation failed.
   *
   * @param callback the callback
   * @param executor the executor used to invoke the callback
   * @param cause the cause of failure
   */
  public static void notifyFailed(PingCallback callback, Executor executor, Status cause) {
    doExecute(executor, asRunnable(callback, cause));
  }

  /**
   * Executes the given runnable. This prevents exceptions from propagating so that an exception
   * thrown by one callback won't prevent subsequent callbacks from being executed.
   */
  private static void doExecute(Executor executor, Runnable runnable) {
    try {
      executor.execute(runnable);
    } catch (Throwable th) {
      log.log(Level.SEVERE, "Failed to execute PingCallback", th);
    }
  }

  /**
   * Returns a runnable that, when run, invokes the given callback, providing the given round-trip
   * duration.
   */
  private static Runnable asRunnable(final ClientTransport.PingCallback callback,
                                     final long roundTripTimeNanos) {
    return new Runnable() {
      @Override
      public void run() {
        callback.onSuccess(roundTripTimeNanos);
      }
    };

  }

  /**
   * Returns a runnable that, when run, invokes the given callback, providing the given cause of
   * failure.
   */
  private static Runnable asRunnable(final ClientTransport.PingCallback callback,
                                     final Status failureCause) {
    return new Runnable() {
      @Override
      public void run() {
        callback.onFailure(failureCause);
      }
    };
  }
}

1	/*
2	* Copyright 2015 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.internal;
18
19	import com.google.common.base.Stopwatch;
20	import com.google.errorprone.annotations.concurrent.GuardedBy;
21	import io.grpc.Status;
22	import io.grpc.internal.ClientTransport.PingCallback;
23	import java.util.LinkedHashMap;
24	import java.util.Map;
25	import java.util.concurrent.Executor;
26	import java.util.concurrent.TimeUnit;
27	import java.util.logging.Level;
28	import java.util.logging.Logger;
29
30	/**
31	* Represents an outstanding PING operation on an HTTP/2 channel. This can be used by HTTP/2-based
32	* transports to implement {@link ClientTransport#ping}.
33	*
34	* <p>A typical transport need only support one outstanding ping at a time. So, if a ping is
35	* requested while an operation is already in progress, the given callback is notified when the
36	* existing operation completes.
37	*/
38	public class Http2Ping {
39	private static final Logger log = Logger.getLogger(Http2Ping.class.getName());	1✔
40
41	/**
42	* The PING frame includes 8 octets of payload data, e.g. 64 bits.
43	*/
44	private final long data;
45
46	/**
47	* Used to measure elapsed time.
48	*/
49	private final Stopwatch stopwatch;
50
51	/**
52	* The registered callbacks and the executor used to invoke them.
53	*/
54	@GuardedBy("this") private Map<PingCallback, Executor> callbacks	1✔
55	= new LinkedHashMap<>();
56
57	/**
58	* False until the operation completes, either successfully (other side sent acknowledgement) or
59	* unsuccessfully.
60	*/
61	@GuardedBy("this") private boolean completed;
62
63	/**
64	* If non-null, indicates the ping failed.
65	*/
66	@GuardedBy("this") private Status failureCause;
67
68	/**
69	* The round-trip time for the ping, in nanoseconds. This value is only meaningful when
70	* {@link #completed} is true and {@link #failureCause} is null.
71	*/
72	@GuardedBy("this") private long roundTripTimeNanos;
73
74	/**
75	* Creates a new ping operation. The caller is responsible for sending a ping on an HTTP/2 channel
76	* using the given payload. The caller is also responsible for starting the stopwatch when the
77	* PING frame is sent.
78	*
79	* @param data the ping payload
80	* @param stopwatch a stopwatch for measuring round-trip time
81	*/
82	public Http2Ping(long data, Stopwatch stopwatch) {	1✔
83	this.data = data;	1✔
84	this.stopwatch = stopwatch;	1✔
85	}	1✔
86
87	/**
88	* Registers a callback that is invoked when the ping operation completes. If this ping operation
89	* is already completed, the callback is invoked immediately.
90	*
91	* @param callback the callback to invoke
92	* @param executor the executor to use
93	*/
94	public void addCallback(final ClientTransport.PingCallback callback, Executor executor) {
95	Runnable runnable;
96	synchronized (this) {	1✔
97	if (!completed) {	1✔
98	callbacks.put(callback, executor);	1✔
99	return;	1✔
100	}
101	// otherwise, invoke callback immediately (but not while holding lock)
102	runnable = this.failureCause != null ? asRunnable(callback, failureCause)	×
103	: asRunnable(callback, roundTripTimeNanos);	×
104	}	×
105	doExecute(executor, runnable);	×
106	}	×
107
108	/**
109	* Returns the expected ping payload for this outstanding operation.
110	*
111	* @return the expected payload for this outstanding ping
112	*/
113	public long payload() {
114	return data;	1✔
115	}
116
117	/**
118	* Completes this operation successfully. The stopwatch given during construction is used to
119	* measure the elapsed time. Registered callbacks are invoked and provided the measured elapsed
120	* time.
121	*
122	* @return true if the operation is marked as complete; false if it was already complete
123	*/
124	public boolean complete() {
125	Map<ClientTransport.PingCallback, Executor> callbacks;
126	long roundTripTimeNanos;
127	synchronized (this) {	1✔
128	if (completed) {	1✔
129	return false;	×
130	}
131	completed = true;	1✔
132	roundTripTimeNanos = this.roundTripTimeNanos = stopwatch.elapsed(TimeUnit.NANOSECONDS);	1✔
133	callbacks = this.callbacks;	1✔
134	this.callbacks = null;	1✔
135	}	1✔
136	for (Map.Entry<ClientTransport.PingCallback, Executor> entry : callbacks.entrySet()) {	1✔
137	doExecute(entry.getValue(), asRunnable(entry.getKey(), roundTripTimeNanos));	1✔
138	}	1✔
139	return true;	1✔
140	}
141
142	/**
143	* Completes this operation exceptionally. Registered callbacks are invoked and provided the
144	* given throwable as the cause of failure.
145	*
146	* @param failureCause the cause of failure
147	*/
148	public void failed(Status failureCause) {
149	Map<ClientTransport.PingCallback, Executor> callbacks;
150	synchronized (this) {	1✔
151	if (completed) {	1✔
152	return;	×
153	}
154	completed = true;	1✔
155	this.failureCause = failureCause;	1✔
156	callbacks = this.callbacks;	1✔
157	this.callbacks = null;	1✔
158	}	1✔
159	for (Map.Entry<ClientTransport.PingCallback, Executor> entry : callbacks.entrySet()) {	1✔
160	notifyFailed(entry.getKey(), entry.getValue(), failureCause);	1✔
161	}	1✔
162	}	1✔
163
164	/**
165	* Notifies the given callback that the ping operation failed.
166	*
167	* @param callback the callback
168	* @param executor the executor used to invoke the callback
169	* @param cause the cause of failure
170	*/
171	public static void notifyFailed(PingCallback callback, Executor executor, Status cause) {
172	doExecute(executor, asRunnable(callback, cause));	1✔
173	}	1✔
174
175	/**
176	* Executes the given runnable. This prevents exceptions from propagating so that an exception
177	* thrown by one callback won't prevent subsequent callbacks from being executed.
178	*/
179	private static void doExecute(Executor executor, Runnable runnable) {
180	try {
181	executor.execute(runnable);	1✔
182	} catch (Throwable th) {	×
183	log.log(Level.SEVERE, "Failed to execute PingCallback", th);	×
184	}	1✔
185	}	1✔
186
187	/**
188	* Returns a runnable that, when run, invokes the given callback, providing the given round-trip
189	* duration.
190	*/
191	private static Runnable asRunnable(final ClientTransport.PingCallback callback,
192	final long roundTripTimeNanos) {
193	return new Runnable() {	1✔
194	@Override
195	public void run() {
196	callback.onSuccess(roundTripTimeNanos);	1✔
197	}	1✔
198	};
199
200	}
201
202	/**
203	* Returns a runnable that, when run, invokes the given callback, providing the given cause of
204	* failure.
205	*/
206	private static Runnable asRunnable(final ClientTransport.PingCallback callback,
207	final Status failureCause) {
208	return new Runnable() {	1✔
209	@Override
210	public void run() {
211	callback.onFailure(failureCause);	1✔
212	}	1✔
213	};
214	}
215	}

grpc / grpc-java / #19916

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