#20345

Committed 01 Jul 2026 05:33PM UTC coverage: 88.778% (-0.02%) from 88.802%

Build # #20345

Build Type

push

github

Committed by

ejona86

Commit Message

Revert "xds: reuse GrpcXdsTransport and underlying gRPC channel to the same xDS server by ref-counting"

This reverts commit e39c38bf5.

This causes hangs for xds when MAX_CONCURRENT_STREAMS limit is too low
for the number of targets being used. Each XdsClient will often use 2
streams (ADS, LRS), so a concurrency limit of 100 will likely have
issues after 50 targets.

Hitting MAX_CONCURRENT_STREAMS prevents the resource timer from being
started as the ADS RPC does not become ready, so users see a hang
instead of an xds resource timeout after 15 seconds.

b/529399722

Coverage Stats

35996 of 40546 relevant lines covered (88.78%)

0.89 hits per line

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

87.35

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

/*
 * Copyright 2020 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 static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.base.Preconditions.checkState;
import static java.util.concurrent.TimeUnit.NANOSECONDS;

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.MoreObjects;
import com.google.errorprone.annotations.concurrent.GuardedBy;
import io.grpc.Attributes;
import io.grpc.ClientCall;
import io.grpc.Context;
import io.grpc.Deadline;
import io.grpc.Metadata;
import io.grpc.Status;
import java.util.ArrayList;
import java.util.List;
import java.util.Locale;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.annotation.Nullable;

/**
 * A call that queues requests before a real call is ready to be delegated to.
 *
 * <p>{@code ClientCall} itself doesn't require thread-safety. However, the state of {@code
 * DelayedCall} may be internally altered by different threads, thus internal synchronization is
 * necessary.
 */
public class DelayedClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
  private static final Logger logger = Logger.getLogger(DelayedClientCall.class.getName());
  /**
   * A timer to monitor the initial deadline. The timer must be cancelled on transition to the real
   * call.
   */
  @Nullable
  private final ScheduledFuture<?> initialDeadlineMonitor;
  private final Executor callExecutor;
  private final Context context;
  /** {@code true} once realCall is valid and all pending calls have been drained. */
  private volatile boolean passThrough;
  /**
   * Non-{@code null} iff start has been called. Used to assert methods are called in appropriate
   * order, but also used if an error occurs before {@code realCall} is set.
   */
  private Listener<RespT> listener;
  // No need to synchronize; start() synchronization provides a happens-before
  private Metadata startHeaders;
  // Must hold {@code this} lock when setting.
  private ClientCall<ReqT, RespT> realCall;
  @GuardedBy("this")
  private Status error;
  @GuardedBy("this")
  private List<Runnable> pendingRunnables = new ArrayList<>();
  @GuardedBy("this")
  private DelayedListener<RespT> delayedListener;

  protected DelayedClientCall(
      Executor callExecutor, ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
    this.callExecutor = checkNotNull(callExecutor, "callExecutor");
    checkNotNull(scheduler, "scheduler");
    context = Context.current();
    initialDeadlineMonitor = scheduleDeadlineIfNeeded(scheduler, deadline);
  }

  // If one argument is null, consider the other the "Before"
  private boolean isAbeforeB(@Nullable Deadline a, @Nullable Deadline b) {
    if (b == null) {
      return true;
    } else if (a == null) {
      return false;
    }

    return a.isBefore(b);
  }

  @Nullable
  private ScheduledFuture<?> scheduleDeadlineIfNeeded(
      ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
    Deadline contextDeadline = context.getDeadline();
    String deadlineName;
    long remainingNanos;
    if (deadline != null && isAbeforeB(deadline, contextDeadline)) {
      deadlineName = "CallOptions";
      remainingNanos = deadline.timeRemaining(NANOSECONDS);
    } else if (contextDeadline != null) {
      deadlineName = "Context";
      remainingNanos = contextDeadline.timeRemaining(NANOSECONDS);
      if (logger.isLoggable(Level.FINE)) {
        StringBuilder builder =
            new StringBuilder(
                String.format(
                    Locale.US,
                    "Call timeout set to '%d' ns, due to context deadline.", remainingNanos));
        if (deadline == null) {
          builder.append(" Explicit call timeout was not set.");
        } else {
          long callTimeout = deadline.timeRemaining(TimeUnit.NANOSECONDS);
          builder.append(String.format(
              Locale.US, " Explicit call timeout was '%d' ns.", callTimeout));
        }
        logger.fine(builder.toString());
      }
    } else {
      return null;
    }

    /* Cancels the call if deadline exceeded prior to the real call being set. */
    class DeadlineExceededRunnable implements Runnable {
      @Override
      public void run() {
        long seconds = Math.abs(remainingNanos) / TimeUnit.SECONDS.toNanos(1);
        long nanos = Math.abs(remainingNanos) % TimeUnit.SECONDS.toNanos(1);
        StringBuilder buf = new StringBuilder();
        if (remainingNanos < 0) {
          buf.append("ClientCall started after ");
          buf.append(deadlineName);
          buf.append(" deadline was exceeded. Deadline has been exceeded for ");
        } else {
          buf.append("Deadline ");
          buf.append(deadlineName);
          buf.append(" was exceeded after ");
        }
        buf.append(seconds);
        buf.append(String.format(Locale.US, ".%09d", nanos));
        buf.append("s");
        cancel(
            Status.DEADLINE_EXCEEDED.withDescription(buf.toString()),
            // We should not cancel the call if the realCall is set because there could be a
            // race between cancel() and realCall.start(). The realCall will handle deadline by
            // itself.
            /* onlyCancelPendingCall= */ true);
      }
    }

    return scheduler.schedule(new DeadlineExceededRunnable(), remainingNanos, NANOSECONDS);
  }

  /**
   * Transfers all pending and future requests and mutations to the given call.
   *
   * <p>No-op if either this method or {@link #cancel} have already been called.
   */
  // When this method returns, passThrough is guaranteed to be true
  public final Runnable setCall(ClientCall<ReqT, RespT> call) {
    Listener<RespT> savedDelayedListener;
    synchronized (this) {
      // If realCall != null, then either setCall() or cancel() has been called.
      if (realCall != null) {
        return null;
      }
      setRealCall(checkNotNull(call, "call"));
      // start() not yet called
      if (delayedListener == null) {
        assert pendingRunnables.isEmpty();
        pendingRunnables = null;
        passThrough = true;
        return null;
      }
      savedDelayedListener = this.delayedListener;
    }
    internalStart(savedDelayedListener);
    return new ContextRunnable(context) {
      @Override
      public void runInContext() {
        drainPendingCalls();
      }
    };
  }

  private void internalStart(Listener<RespT> listener) {
    Metadata savedStartHeaders = this.startHeaders;
    this.startHeaders = null;
    context.run(() -> realCall.start(listener, savedStartHeaders));
  }

  @Override
  public final void start(Listener<RespT> listener, final Metadata headers) {
    checkNotNull(headers, "headers");
    checkState(this.listener == null, "already started");
    Status savedError;
    boolean savedPassThrough;
    synchronized (this) {
      this.listener = checkNotNull(listener, "listener");
      // If error != null, then cancel() has been called and was unable to close the listener
      savedError = error;
      savedPassThrough = passThrough;
      if (!savedPassThrough) {
        listener = delayedListener = new DelayedListener<>(listener);
        startHeaders = headers;
      }
    }
    if (savedError != null) {
      callExecutor.execute(new CloseListenerRunnable(listener, savedError));
      return;
    }
    if (savedPassThrough) {
      realCall.start(listener, headers);
    } // else realCall.start() will be called by setCall
  }

  // When this method returns, passThrough is guaranteed to be true
  @Override
  public final void cancel(@Nullable final String message, @Nullable final Throwable cause) {
    Status status = Status.CANCELLED;
    if (message != null) {
      status = status.withDescription(message);
    } else {
      status = status.withDescription("Call cancelled without message");
    }
    if (cause != null) {
      status = status.withCause(cause);
    }
    cancel(status, false);
  }

  /**
   * Cancels the call unless {@code realCall} is set and {@code onlyCancelPendingCall} is true.
   */
  private void cancel(final Status status, boolean onlyCancelPendingCall) {
    boolean delegateToRealCall = true;
    Listener<RespT> listenerToClose = null;
    synchronized (this) {
      // If realCall != null, then either setCall() or cancel() has been called
      if (realCall == null) {
        @SuppressWarnings("unchecked")
        ClientCall<ReqT, RespT> noopCall = (ClientCall<ReqT, RespT>) NOOP_CALL;
        setRealCall(noopCall);
        delegateToRealCall = false;
        // If listener == null, then start() will later call listener with 'error'
        listenerToClose = listener;
        error = status;
      } else if (onlyCancelPendingCall) {
        return;
      }
    }
    if (delegateToRealCall) {
      delayOrExecute(new Runnable() {
        @Override
        public void run() {
          realCall.cancel(status.getDescription(), status.getCause());
        }
      });
    } else {
      if (listenerToClose != null) {
        callExecutor.execute(new CloseListenerRunnable(listenerToClose, status));
      }
      internalStart(listenerToClose); // listener instance doesn't matter
      drainPendingCalls();
    }
    callCancelled();
  }

  protected void callCancelled() {
  }

  private void delayOrExecute(Runnable runnable) {
    synchronized (this) {
      if (!passThrough) {
        pendingRunnables.add(runnable);
        return;
      }
    }
    runnable.run();
  }

  /**
   * Called to transition {@code passThrough} to {@code true}. This method is not safe to be called
   * multiple times; the caller must ensure it will only be called once, ever. {@code this} lock
   * should not be held when calling this method.
   */
  private void drainPendingCalls() {
    assert realCall != null;
    assert !passThrough;
    List<Runnable> toRun = new ArrayList<>();
    DelayedListener<RespT> delayedListener ;
    while (true) {
      synchronized (this) {
        if (pendingRunnables.isEmpty()) {
          pendingRunnables = null;
          passThrough = true;
          delayedListener = this.delayedListener;
          break;
        }
        // Since there were pendingCalls, we need to process them. To maintain ordering we can't set
        // passThrough=true until we run all pendingCalls, but new Runnables may be added after we
        // drop the lock. So we will have to re-check pendingCalls.
        List<Runnable> tmp = toRun;
        toRun = pendingRunnables;
        pendingRunnables = tmp;
      }
      for (Runnable runnable : toRun) {
        // Must not call transport while lock is held to prevent deadlocks.
        // TODO(ejona): exception handling
        runnable.run();
      }
      toRun.clear();
    }
    if (delayedListener != null) {
      final DelayedListener<RespT> listener = delayedListener;
      class DrainListenerRunnable extends ContextRunnable {
        DrainListenerRunnable() {
          super(context);
        }

        @Override
        public void runInContext() {
          listener.drainPendingCallbacks();
        }
      }

      callExecutor.execute(new DrainListenerRunnable());
    }
  }

  @GuardedBy("this")
  private void setRealCall(ClientCall<ReqT, RespT> realCall) {
    checkState(this.realCall == null, "realCall already set to %s", this.realCall);
    if (initialDeadlineMonitor != null) {
      initialDeadlineMonitor.cancel(false);
    }
    this.realCall = realCall;
  }

  @VisibleForTesting
  final ClientCall<ReqT, RespT> getRealCall() {
    return realCall;
  }

  @Override
  public final void sendMessage(final ReqT message) {
    if (passThrough) {
      realCall.sendMessage(message);
    } else {
      delayOrExecute(new Runnable() {
        @Override
        public void run() {
          realCall.sendMessage(message);
        }
      });
    }
  }

  @Override
  public final void setMessageCompression(final boolean enable) {
    if (passThrough) {
      realCall.setMessageCompression(enable);
    } else {
      delayOrExecute(new Runnable() {
        @Override
        public void run() {
          realCall.setMessageCompression(enable);
        }
      });
    }
  }

  @Override
  public final void request(final int numMessages) {
    if (passThrough) {
      realCall.request(numMessages);
    } else {
      delayOrExecute(new Runnable() {
        @Override
        public void run() {
          realCall.request(numMessages);
        }
      });
    }
  }

  @Override
  public final void halfClose() {
    delayOrExecute(new Runnable() {
      @Override
      public void run() {
        realCall.halfClose();
      }
    });
  }

  @Override
  public final boolean isReady() {
    if (passThrough) {
      return realCall.isReady();
    } else {
      return false;
    }
  }

  @Override
  public final Attributes getAttributes() {
    ClientCall<ReqT, RespT> savedRealCall;
    synchronized (this) {
      savedRealCall = realCall;
    }
    if (savedRealCall != null) {
      return savedRealCall.getAttributes();
    } else {
      return Attributes.EMPTY;
    }
  }

  @Override
  public String toString() {
    return MoreObjects.toStringHelper(this)
        .add("realCall", realCall)
        .toString();
  }

  private final class CloseListenerRunnable extends ContextRunnable {
    final Listener<RespT> listener;
    final Status status;

    CloseListenerRunnable(Listener<RespT> listener, Status status) {
      super(context);
      this.listener = listener;
      this.status = status;
    }

    @Override
    public void runInContext() {
      listener.onClose(status, new Metadata());
    }
  }

  private static final class DelayedListener<RespT> extends Listener<RespT> {
    private final Listener<RespT> realListener;
    private volatile boolean passThrough;
    @GuardedBy("this")
    private List<Runnable> pendingCallbacks = new ArrayList<>();

    public DelayedListener(Listener<RespT> listener) {
      this.realListener = listener;
    }

    private void delayOrExecute(Runnable runnable) {
      synchronized (this) {
        if (!passThrough) {
          pendingCallbacks.add(runnable);
          return;
        }
      }
      runnable.run();
    }

    @Override
    public void onHeaders(final Metadata headers) {
      if (passThrough) {
        realListener.onHeaders(headers);
      } else {
        delayOrExecute(new Runnable() {
          @Override
          public void run() {
            realListener.onHeaders(headers);
          }
        });
      }
    }

    @Override
    public void onMessage(final RespT message) {
      if (passThrough) {
        realListener.onMessage(message);
      } else {
        delayOrExecute(new Runnable() {
          @Override
          public void run() {
            realListener.onMessage(message);
          }
        });
      }
    }

    @Override
    public void onClose(final Status status, final Metadata trailers) {
      delayOrExecute(new Runnable() {
        @Override
        public void run() {
          realListener.onClose(status, trailers);
        }
      });
    }

    @Override
    public void onReady() {
      if (passThrough) {
        realListener.onReady();
      } else {
        delayOrExecute(new Runnable() {
          @Override
          public void run() {
            realListener.onReady();
          }
        });
      }
    }

    void drainPendingCallbacks() {
      assert !passThrough;
      List<Runnable> toRun = new ArrayList<>();
      while (true) {
        synchronized (this) {
          if (pendingCallbacks.isEmpty()) {
            pendingCallbacks = null;
            passThrough = true;
            break;
          }
          // Since there were pendingCallbacks, we need to process them. To maintain ordering we
          // can't set passThrough=true until we run all pendingCallbacks, but new Runnables may be
          // added after we drop the lock. So we will have to re-check pendingCallbacks.
          List<Runnable> tmp = toRun;
          toRun = pendingCallbacks;
          pendingCallbacks = tmp;
        }
        for (Runnable runnable : toRun) {
          // Avoid calling listener while lock is held to prevent deadlocks.
          // TODO(ejona): exception handling
          runnable.run();
        }
        toRun.clear();
      }
    }
  }

  private static final ClientCall<Object, Object> NOOP_CALL = new ClientCall<Object, Object>() {
    @Override
    public void start(Listener<Object> responseListener, Metadata headers) {}

    @Override
    public void request(int numMessages) {}

    @Override
    public void cancel(String message, Throwable cause) {}

    @Override
    public void halfClose() {}

    @Override
    public void sendMessage(Object message) {}

    // Always returns {@code false}, since this is only used when the startup of the call fails.
    @Override
    public boolean isReady() {
      return false;
    }
  };
}

1	/*
2	* Copyright 2020 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 static com.google.common.base.Preconditions.checkNotNull;
20	import static com.google.common.base.Preconditions.checkState;
21	import static java.util.concurrent.TimeUnit.NANOSECONDS;
22
23	import com.google.common.annotations.VisibleForTesting;
24	import com.google.common.base.MoreObjects;
25	import com.google.errorprone.annotations.concurrent.GuardedBy;
26	import io.grpc.Attributes;
27	import io.grpc.ClientCall;
28	import io.grpc.Context;
29	import io.grpc.Deadline;
30	import io.grpc.Metadata;
31	import io.grpc.Status;
32	import java.util.ArrayList;
33	import java.util.List;
34	import java.util.Locale;
35	import java.util.concurrent.Executor;
36	import java.util.concurrent.ScheduledExecutorService;
37	import java.util.concurrent.ScheduledFuture;
38	import java.util.concurrent.TimeUnit;
39	import java.util.logging.Level;
40	import java.util.logging.Logger;
41	import javax.annotation.Nullable;
42
43	/**
44	* A call that queues requests before a real call is ready to be delegated to.
45	*
46	* <p>{@code ClientCall} itself doesn't require thread-safety. However, the state of {@code
47	* DelayedCall} may be internally altered by different threads, thus internal synchronization is
48	* necessary.
49	*/
50	public class DelayedClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
51	private static final Logger logger = Logger.getLogger(DelayedClientCall.class.getName());	1 ✔
52	/**
53	* A timer to monitor the initial deadline. The timer must be cancelled on transition to the real
54	* call.
55	*/
56	@Nullable
57	private final ScheduledFuture<?> initialDeadlineMonitor;
58	private final Executor callExecutor;
59	private final Context context;
60	/** {@code true} once realCall is valid and all pending calls have been drained. */
61	private volatile boolean passThrough;
62	/**
63	* Non-{@code null} iff start has been called. Used to assert methods are called in appropriate
64	* order, but also used if an error occurs before {@code realCall} is set.
65	*/
66	private Listener<RespT> listener;
67	// No need to synchronize; start() synchronization provides a happens-before
68	private Metadata startHeaders;
69	// Must hold {@code this} lock when setting.
70	private ClientCall<ReqT, RespT> realCall;
71	@GuardedBy("this")
72	private Status error;
73	@GuardedBy("this")	1 ✔
74	private List<Runnable> pendingRunnables = new ArrayList<>();
75	@GuardedBy("this")
76	private DelayedListener<RespT> delayedListener;
77
78	protected DelayedClientCall(
79	Executor callExecutor, ScheduledExecutorService scheduler, @Nullable Deadline deadline) {	1 ✔
80	this.callExecutor = checkNotNull(callExecutor, "callExecutor");	1 ✔
81	checkNotNull(scheduler, "scheduler");	1 ✔
82	context = Context.current();	1 ✔
83	initialDeadlineMonitor = scheduleDeadlineIfNeeded(scheduler, deadline);	1 ✔
84	}	1 ✔
85
86	// If one argument is null, consider the other the "Before"
87	private boolean isAbeforeB(@Nullable Deadline a, @Nullable Deadline b) {
88	if (b == null) {	1 ✔
89	return true;	1 ✔
90	} else if (a == null) {	×
91	return false;	×
92	}
93
94	return a.isBefore(b);	×
95	}
96
97	@Nullable
98	private ScheduledFuture<?> scheduleDeadlineIfNeeded(
99	ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
100	Deadline contextDeadline = context.getDeadline();	1 ✔
101	String deadlineName;
102	long remainingNanos;
103	if (deadline != null && isAbeforeB(deadline, contextDeadline)) {	1 ✔
104	deadlineName = "CallOptions";	1 ✔
105	remainingNanos = deadline.timeRemaining(NANOSECONDS);	1 ✔
106	} else if (contextDeadline != null) {	1 ✔
107	deadlineName = "Context";	×
108	remainingNanos = contextDeadline.timeRemaining(NANOSECONDS);	×
109	if (logger.isLoggable(Level.FINE)) {	×
110	StringBuilder builder =	×
111	new StringBuilder(
112	String.format(	×
113	Locale.US,
114	"Call timeout set to '%d' ns, due to context deadline.", remainingNanos));	×
115	if (deadline == null) {	×
116	builder.append(" Explicit call timeout was not set.");	×
117	} else {
118	long callTimeout = deadline.timeRemaining(TimeUnit.NANOSECONDS);	×
119	builder.append(String.format(	×
120	Locale.US, " Explicit call timeout was '%d' ns.", callTimeout));	×
121	}
122	logger.fine(builder.toString());	×
123	}	×
124	} else {
125	return null;	1 ✔
126	}
127
128	/* Cancels the call if deadline exceeded prior to the real call being set. */
129	class DeadlineExceededRunnable implements Runnable {	1 ✔
130	@Override
131	public void run() {
132	long seconds = Math.abs(remainingNanos) / TimeUnit.SECONDS.toNanos(1);	1 ✔
133	long nanos = Math.abs(remainingNanos) % TimeUnit.SECONDS.toNanos(1);	1 ✔
134	StringBuilder buf = new StringBuilder();	1 ✔
135	if (remainingNanos < 0) {	1 ✔
136	buf.append("ClientCall started after ");	1 ✔
137	buf.append(deadlineName);	1 ✔
138	buf.append(" deadline was exceeded. Deadline has been exceeded for ");	1 ✔
139	} else {
140	buf.append("Deadline ");	1 ✔
141	buf.append(deadlineName);	1 ✔
142	buf.append(" was exceeded after ");	1 ✔
143	}
144	buf.append(seconds);	1 ✔
145	buf.append(String.format(Locale.US, ".%09d", nanos));	1 ✔
146	buf.append("s");	1 ✔
147	cancel(	1 ✔
148	Status.DEADLINE_EXCEEDED.withDescription(buf.toString()),	1 ✔
149	// We should not cancel the call if the realCall is set because there could be a
150	// race between cancel() and realCall.start(). The realCall will handle deadline by
151	// itself.
152	/* onlyCancelPendingCall= */ true);
153	}	1 ✔
154	}
155
156	return scheduler.schedule(new DeadlineExceededRunnable(), remainingNanos, NANOSECONDS);	1 ✔
157	}
158
159	/**
160	* Transfers all pending and future requests and mutations to the given call.
161	*
162	* <p>No-op if either this method or {@link #cancel} have already been called.
163	*/
164	// When this method returns, passThrough is guaranteed to be true
165	public final Runnable setCall(ClientCall<ReqT, RespT> call) {
166	Listener<RespT> savedDelayedListener;
167	synchronized (this) {	1 ✔
168	// If realCall != null, then either setCall() or cancel() has been called.
169	if (realCall != null) {	1 ✔
170	return null;	1 ✔
171	}
172	setRealCall(checkNotNull(call, "call"));	1 ✔
173	// start() not yet called
174	if (delayedListener == null) {	1 ✔
175	assert pendingRunnables.isEmpty();	1 ✔
176	pendingRunnables = null;	1 ✔
177	passThrough = true;	1 ✔
178	return null;	1 ✔
179	}
180	savedDelayedListener = this.delayedListener;	1 ✔
181	}	1 ✔
182	internalStart(savedDelayedListener);	1 ✔
183	return new ContextRunnable(context) {	1 ✔
184	@Override
185	public void runInContext() {
186	drainPendingCalls();	1 ✔
187	}	1 ✔
188	};
189	}
190
191	private void internalStart(Listener<RespT> listener) {
192	Metadata savedStartHeaders = this.startHeaders;	1 ✔
193	this.startHeaders = null;	1 ✔
194	context.run(() -> realCall.start(listener, savedStartHeaders));	1 ✔
195	}	1 ✔
196
197	@Override
198	public final void start(Listener<RespT> listener, final Metadata headers) {
199	checkNotNull(headers, "headers");	1 ✔
200	checkState(this.listener == null, "already started");	1 ✔
201	Status savedError;
202	boolean savedPassThrough;
203	synchronized (this) {	1 ✔
204	this.listener = checkNotNull(listener, "listener");	1 ✔
205	// If error != null, then cancel() has been called and was unable to close the listener
206	savedError = error;	1 ✔
207	savedPassThrough = passThrough;	1 ✔
208	if (!savedPassThrough) {	1 ✔
209	listener = delayedListener = new DelayedListener<>(listener);	1 ✔
210	startHeaders = headers;	1 ✔
211	}
212	}	1 ✔
213	if (savedError != null) {	1 ✔
214	callExecutor.execute(new CloseListenerRunnable(listener, savedError));	×
215	return;	×
216	}
217	if (savedPassThrough) {	1 ✔
218	realCall.start(listener, headers);	1 ✔
219	} // else realCall.start() will be called by setCall
220	}	1 ✔
221
222	// When this method returns, passThrough is guaranteed to be true
223	@Override
224	public final void cancel(@Nullable final String message, @Nullable final Throwable cause) {
225	Status status = Status.CANCELLED;	1 ✔
226	if (message != null) {	1 ✔
227	status = status.withDescription(message);	1 ✔
228	} else {
229	status = status.withDescription("Call cancelled without message");	1 ✔
230	}
231	if (cause != null) {	1 ✔
232	status = status.withCause(cause);	1 ✔
233	}
234	cancel(status, false);	1 ✔
235	}	1 ✔
236
237	/**
238	* Cancels the call unless {@code realCall} is set and {@code onlyCancelPendingCall} is true.
239	*/
240	private void cancel(final Status status, boolean onlyCancelPendingCall) {
241	boolean delegateToRealCall = true;	1 ✔
242	Listener<RespT> listenerToClose = null;	1 ✔
243	synchronized (this) {	1 ✔
244	// If realCall != null, then either setCall() or cancel() has been called
245	if (realCall == null) {	1 ✔
246	@SuppressWarnings("unchecked")
247	ClientCall<ReqT, RespT> noopCall = (ClientCall<ReqT, RespT>) NOOP_CALL;	1 ✔
248	setRealCall(noopCall);	1 ✔
249	delegateToRealCall = false;	1 ✔
250	// If listener == null, then start() will later call listener with 'error'
251	listenerToClose = listener;	1 ✔
252	error = status;	1 ✔
253	} else if (onlyCancelPendingCall) {	1 ✔
254	return;	×
255	}
256	}	1 ✔
257	if (delegateToRealCall) {	1 ✔
258	delayOrExecute(new Runnable() {	1 ✔
259	@Override
260	public void run() {
261	realCall.cancel(status.getDescription(), status.getCause());	1 ✔
262	}	1 ✔
263	});
264	} else {
265	if (listenerToClose != null) {	1 ✔
266	callExecutor.execute(new CloseListenerRunnable(listenerToClose, status));	1 ✔
267	}
268	internalStart(listenerToClose); // listener instance doesn't matter	1 ✔
269	drainPendingCalls();	1 ✔
270	}
271	callCancelled();	1 ✔
272	}	1 ✔
273
274	protected void callCancelled() {
275	}	1 ✔
276
277	private void delayOrExecute(Runnable runnable) {
278	synchronized (this) {	1 ✔
279	if (!passThrough) {	1 ✔
280	pendingRunnables.add(runnable);	1 ✔
281	return;	1 ✔
282	}
283	}	1 ✔
284	runnable.run();	1 ✔
285	}	1 ✔
286
287	/**
288	* Called to transition {@code passThrough} to {@code true}. This method is not safe to be called
289	* multiple times; the caller must ensure it will only be called once, ever. {@code this} lock
290	* should not be held when calling this method.
291	*/
292	private void drainPendingCalls() {
293	assert realCall != null;	1 ✔
294	assert !passThrough;	1 ✔
295	List<Runnable> toRun = new ArrayList<>();	1 ✔
296	DelayedListener<RespT> delayedListener ;
297	while (true) {
298	synchronized (this) {	1 ✔
299	if (pendingRunnables.isEmpty()) {	1 ✔
300	pendingRunnables = null;	1 ✔
301	passThrough = true;	1 ✔
302	delayedListener = this.delayedListener;	1 ✔
303	break;	1 ✔
304	}
305	// Since there were pendingCalls, we need to process them. To maintain ordering we can't set
306	// passThrough=true until we run all pendingCalls, but new Runnables may be added after we
307	// drop the lock. So we will have to re-check pendingCalls.
308	List<Runnable> tmp = toRun;	1 ✔
309	toRun = pendingRunnables;	1 ✔
310	pendingRunnables = tmp;	1 ✔
311	}	1 ✔
312	for (Runnable runnable : toRun) {	1 ✔
313	// Must not call transport while lock is held to prevent deadlocks.
314	// TODO(ejona): exception handling
315	runnable.run();	1 ✔
316	}	1 ✔
317	toRun.clear();	1 ✔
318	}
319	if (delayedListener != null) {	1 ✔
320	final DelayedListener<RespT> listener = delayedListener;	1 ✔
321	class DrainListenerRunnable extends ContextRunnable {
322	DrainListenerRunnable() {	1 ✔
323	super(context);	1 ✔
324	}	1 ✔
325
326	@Override
327	public void runInContext() {
328	listener.drainPendingCallbacks();	1 ✔
329	}	1 ✔
330	}
331
332	callExecutor.execute(new DrainListenerRunnable());	1 ✔
333	}
334	}	1 ✔
335
336	@GuardedBy("this")
337	private void setRealCall(ClientCall<ReqT, RespT> realCall) {
338	checkState(this.realCall == null, "realCall already set to %s", this.realCall);	1 ✔
339	if (initialDeadlineMonitor != null) {	1 ✔
340	initialDeadlineMonitor.cancel(false);	1 ✔
341	}
342	this.realCall = realCall;	1 ✔
343	}	1 ✔
344
345	@VisibleForTesting
346	final ClientCall<ReqT, RespT> getRealCall() {
347	return realCall;	×
348	}
349
350	@Override
351	public final void sendMessage(final ReqT message) {
352	if (passThrough) {	1 ✔
353	realCall.sendMessage(message);	1 ✔
354	} else {
355	delayOrExecute(new Runnable() {	1 ✔
356	@Override
357	public void run() {
358	realCall.sendMessage(message);	1 ✔
359	}	1 ✔
360	});
361	}
362	}	1 ✔
363
364	@Override
365	public final void setMessageCompression(final boolean enable) {
366	if (passThrough) {	1 ✔
367	realCall.setMessageCompression(enable);	1 ✔
368	} else {
369	delayOrExecute(new Runnable() {	1 ✔
370	@Override
371	public void run() {
372	realCall.setMessageCompression(enable);	1 ✔
373	}	1 ✔
374	});
375	}
376	}	1 ✔
377
378	@Override
379	public final void request(final int numMessages) {
380	if (passThrough) {	1 ✔
381	realCall.request(numMessages);	1 ✔
382	} else {
383	delayOrExecute(new Runnable() {	1 ✔
384	@Override
385	public void run() {
386	realCall.request(numMessages);	1 ✔
387	}	1 ✔
388	});
389	}
390	}	1 ✔
391
392	@Override
393	public final void halfClose() {
394	delayOrExecute(new Runnable() {	1 ✔
395	@Override
396	public void run() {
397	realCall.halfClose();	1 ✔
398	}	1 ✔
399	});
400	}	1 ✔
401
402	@Override
403	public final boolean isReady() {
404	if (passThrough) {	1 ✔
405	return realCall.isReady();	1 ✔
406	} else {
407	return false;	×
408	}
409	}
410
411	@Override
412	public final Attributes getAttributes() {
413	ClientCall<ReqT, RespT> savedRealCall;
414	synchronized (this) {	1 ✔
415	savedRealCall = realCall;	1 ✔
416	}	1 ✔
417	if (savedRealCall != null) {	1 ✔
418	return savedRealCall.getAttributes();	1 ✔
419	} else {
420	return Attributes.EMPTY;	×
421	}
422	}
423
424	@Override
425	public String toString() {
426	return MoreObjects.toStringHelper(this)	×
427	.add("realCall", realCall)	×
428	.toString();	×
429	}
430
431	private final class CloseListenerRunnable extends ContextRunnable {
432	final Listener<RespT> listener;
433	final Status status;
434
435	CloseListenerRunnable(Listener<RespT> listener, Status status) {	1 ✔
436	super(context);	1 ✔
437	this.listener = listener;	1 ✔
438	this.status = status;	1 ✔
439	}	1 ✔
440
441	@Override
442	public void runInContext() {
443	listener.onClose(status, new Metadata());	1 ✔
444	}	1 ✔
445	}
446
447	private static final class DelayedListener<RespT> extends Listener<RespT> {	1 ✔
448	private final Listener<RespT> realListener;
449	private volatile boolean passThrough;
450	@GuardedBy("this")	1 ✔
451	private List<Runnable> pendingCallbacks = new ArrayList<>();
452
453	public DelayedListener(Listener<RespT> listener) {	1 ✔
454	this.realListener = listener;	1 ✔
455	}	1 ✔
456
457	private void delayOrExecute(Runnable runnable) {
458	synchronized (this) {	1 ✔
459	if (!passThrough) {	1 ✔
460	pendingCallbacks.add(runnable);	1 ✔
461	return;	1 ✔
462	}
463	}	1 ✔
464	runnable.run();	1 ✔
465	}	1 ✔
466
467	@Override
468	public void onHeaders(final Metadata headers) {
469	if (passThrough) {	1 ✔
470	realListener.onHeaders(headers);	1 ✔
471	} else {
472	delayOrExecute(new Runnable() {	×
473	@Override
474	public void run() {
475	realListener.onHeaders(headers);	×
476	}	×
477	});
478	}
479	}	1 ✔
480
481	@Override
482	public void onMessage(final RespT message) {
483	if (passThrough) {	1 ✔
484	realListener.onMessage(message);	1 ✔
485	} else {
486	delayOrExecute(new Runnable() {	×
487	@Override
488	public void run() {
489	realListener.onMessage(message);	×
490	}	×
491	});
492	}
493	}	1 ✔
494
495	@Override
496	public void onClose(final Status status, final Metadata trailers) {
497	delayOrExecute(new Runnable() {	1 ✔
498	@Override
499	public void run() {
500	realListener.onClose(status, trailers);	1 ✔
501	}	1 ✔
502	});
503	}	1 ✔
504
505	@Override
506	public void onReady() {
507	if (passThrough) {	1 ✔
508	realListener.onReady();	1 ✔
509	} else {
510	delayOrExecute(new Runnable() {	1 ✔
511	@Override
512	public void run() {
513	realListener.onReady();	1 ✔
514	}	1 ✔
515	});
516	}
517	}	1 ✔
518
519	void drainPendingCallbacks() {
520	assert !passThrough;	1 ✔
521	List<Runnable> toRun = new ArrayList<>();	1 ✔
522	while (true) {
523	synchronized (this) {	1 ✔
524	if (pendingCallbacks.isEmpty()) {	1 ✔
525	pendingCallbacks = null;	1 ✔
526	passThrough = true;	1 ✔
527	break;	1 ✔
528	}
529	// Since there were pendingCallbacks, we need to process them. To maintain ordering we
530	// can't set passThrough=true until we run all pendingCallbacks, but new Runnables may be
531	// added after we drop the lock. So we will have to re-check pendingCallbacks.
532	List<Runnable> tmp = toRun;	1 ✔
533	toRun = pendingCallbacks;	1 ✔
534	pendingCallbacks = tmp;	1 ✔
535	}	1 ✔
536	for (Runnable runnable : toRun) {	1 ✔
537	// Avoid calling listener while lock is held to prevent deadlocks.
538	// TODO(ejona): exception handling
539	runnable.run();	1 ✔
540	}	1 ✔
541	toRun.clear();	1 ✔
542	}
543	}	1 ✔
544	}
545
546	private static final ClientCall<Object, Object> NOOP_CALL = new ClientCall<Object, Object>() {	1 ✔
547	@Override
548	public void start(Listener<Object> responseListener, Metadata headers) {}	1 ✔
549
550	@Override
551	public void request(int numMessages) {}	1 ✔
552
553	@Override
554	public void cancel(String message, Throwable cause) {}	1 ✔
555
556	@Override
557	public void halfClose() {}	1 ✔
558
559	@Override
560	public void sendMessage(Object message) {}	1 ✔
561
562	// Always returns {@code false}, since this is only used when the startup of the call fails.
563	@Override
564	public boolean isReady() {
565	return false;	×
566	}
567	};
568	}

grpc / grpc-java / #20345

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