#20347

Committed 03 Jul 2026 10:23AM UTC coverage: 89.132% (+0.01%) from 89.118%

Build # #20347

Build Type

push

github

Committed by

web-flow

Commit Message

xds: Move backend service label plumbing to CDS (#12882)

Addresses #12431.

This moves backend service metric label plumbing from `xds_cluster_impl`
to `cds`, matching the A75/A89 aggregate cluster behavior.

Previously, `xds_cluster_impl` added the `grpc.lb.backend_service`
pick-details label and propagated `NameResolver.ATTR_BACKEND_SERVICE` to
its child policy. That works for simple clusters, but it is the wrong
layer after A75 because aggregate clusters need the leaf CDS cluster
name, not the aggregate root cluster name.

This change makes CDS responsible for backend service plumbing:

- non-aggregate CDS adds the `grpc.lb.backend_service` pick-details
label using the CDS cluster name
- non-aggregate CDS propagates `NameResolver.ATTR_BACKEND_SERVICE` to
its child policy, so WRR can consume the backend service context
- aggregate root CDS does not add the backend service attribute or
pick-details label
- leaf CDS instances under an aggregate cluster add their own leaf
cluster name
- `xds_cluster_impl` no longer owns backend service pick-details label
or child attribute propagation

The existing `InternalEquivalentAddressGroup.ATTR_BACKEND_SERVICE` path
is left unchanged for endpoint/subchannel metrics.

Tests cover:

- non-aggregate CDS propagating `NameResolver.ATTR_BACKEND_SERVICE`
- non-aggregate CDS adding `grpc.lb.backend_service`
- aggregate root CDS not adding the root cluster name
- aggregate leaf CDS instances propagating leaf cluster names
- `xds_cluster_impl` no longer adding backend service labels or
attributes
- picker tests using a non-null `PickDetailsConsumer`, matching the real
`PickSubchannelArgs` contract

Coverage Stats

38003 of 42637 relevant lines covered (89.13%)

0.89 hits per line

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

90.97

/../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 string describing what this call is waiting on. */
  private final String bufferContext;
  /**
   * 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(
      String bufferContext,
      Executor callExecutor,
      ScheduledExecutorService scheduler,
      @Nullable Deadline deadline) {
    this.bufferContext = checkNotNull(bufferContext, "bufferContext");
    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 waiting for ");
        buf.append(bufferContext);
        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<>(this, 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 DelayedClientCall<?, RespT> call;
    private final Listener<RespT> realListener;
    private volatile boolean passThrough;
    private volatile Status exceptionStatus;
    @GuardedBy("this")
    private List<Runnable> pendingCallbacks = new ArrayList<>();

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

    /**
     * Cancels call and schedules onClose() notification. May only be called from within a
     * DelayedListener callback dispatch (either queued drain or passThrough). Visibility of the
     * write to {@code exceptionStatus} does not rely on a single callback executor; it is a
     * {@code volatile} field, and callback queuing/pass-through transitions are coordinated by
     * this listener's synchronization so subsequent callbacks observe the updated status.
     */
    private void exceptionThrown(Throwable t, String description) {
      // onClose() must be delivered exactly once and last. Other callbacks may already be queued
      // ahead of realCall's eventual onClose, so we can't call onClose() here. We set the status
      // and overwrite the onClose() details when it arrives.
      exceptionStatus = Status.CANCELLED.withCause(t).withDescription(description);
      call.cancel(description, t);
    }

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

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

    private void deliverHeaders(Metadata headers) {
      if (exceptionStatus != null) {
        return;
      }
      try {
        realListener.onHeaders(headers);
      } catch (Throwable t) {
        exceptionThrown(t, "Failed to read headers");
      }
    }

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

    private void deliverMessage(RespT message) {
      if (exceptionStatus != null) {
        return;
      }
      try {
        realListener.onMessage(message);
      } catch (Throwable t) {
        exceptionThrown(t, "Failed to read message.");
      }
    }

    @Override
    public void onClose(final Status status, final Metadata trailers) {
      delayOrExecute(new Runnable() {
        @Override
        public void run() {
          Status effectiveStatus = status;
          Metadata effectiveTrailers = trailers;
          if (exceptionStatus != null) {
            // Ideally status matches exceptionStatus, since exceptionStatus was used to cancel
            // the call. However, cancel() may reconstruct a new Status instance, and the cancel
            // is racy so this onClose may have already been queued when the cancellation
            // occurred. Since other callbacks throw away data if exceptionStatus != null, it is
            // semantically essential that we _not_ use a status provided by the server.
            effectiveStatus = exceptionStatus;
            // Replace trailers to prevent mixing sources of status and trailers.
            effectiveTrailers = new Metadata();
          }
          try {
            realListener.onClose(effectiveStatus, effectiveTrailers);
          } catch (RuntimeException ex) {
            logger.log(Level.WARNING, "Exception thrown by onClose() in ClientCall", ex);
          }
        }
      });
    }

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

    private void deliverOnReady() {
      if (exceptionStatus != null) {
        return;
      }
      try {
        realListener.onReady();
      } catch (Throwable t) {
        exceptionThrown(t, "Failed to call 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.
          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 string describing what this call is waiting on. */
54	private final String bufferContext;
55	/**
56	* A timer to monitor the initial deadline. The timer must be cancelled on transition to the real
57	* call.
58	*/
59	@Nullable
60	private final ScheduledFuture<?> initialDeadlineMonitor;
61	private final Executor callExecutor;
62	private final Context context;
63	/** {@code true} once realCall is valid and all pending calls have been drained. */
64	private volatile boolean passThrough;
65	/**
66	* Non-{@code null} iff start has been called. Used to assert methods are called in appropriate
67	* order, but also used if an error occurs before {@code realCall} is set.
68	*/
69	private Listener<RespT> listener;
70	// No need to synchronize; start() synchronization provides a happens-before
71	private Metadata startHeaders;
72	// Must hold {@code this} lock when setting.
73	private ClientCall<ReqT, RespT> realCall;
74	@GuardedBy("this")
75	private Status error;
76	@GuardedBy("this")	1✔
77	private List<Runnable> pendingRunnables = new ArrayList<>();
78	@GuardedBy("this")
79	private DelayedListener<RespT> delayedListener;
80
81	protected DelayedClientCall(
82	String bufferContext,
83	Executor callExecutor,
84	ScheduledExecutorService scheduler,
85	@Nullable Deadline deadline) {	1✔
86	this.bufferContext = checkNotNull(bufferContext, "bufferContext");	1✔
87	this.callExecutor = checkNotNull(callExecutor, "callExecutor");	1✔
88	checkNotNull(scheduler, "scheduler");	1✔
89	context = Context.current();	1✔
90	initialDeadlineMonitor = scheduleDeadlineIfNeeded(scheduler, deadline);	1✔
91	}	1✔
92
93	// If one argument is null, consider the other the "Before"
94	private boolean isAbeforeB(@Nullable Deadline a, @Nullable Deadline b) {
95	if (b == null) {	1✔
96	return true;	1✔
97	} else if (a == null) {	×
98	return false;	×
99	}
100
101	return a.isBefore(b);	×
102	}
103
104	@Nullable
105	private ScheduledFuture<?> scheduleDeadlineIfNeeded(
106	ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
107	Deadline contextDeadline = context.getDeadline();	1✔
108	String deadlineName;
109	long remainingNanos;
110	if (deadline != null && isAbeforeB(deadline, contextDeadline)) {	1✔
111	deadlineName = "CallOptions";	1✔
112	remainingNanos = deadline.timeRemaining(NANOSECONDS);	1✔
113	} else if (contextDeadline != null) {	1✔
114	deadlineName = "Context";	×
115	remainingNanos = contextDeadline.timeRemaining(NANOSECONDS);	×
116	if (logger.isLoggable(Level.FINE)) {	×
117	StringBuilder builder =	×
118	new StringBuilder(
119	String.format(	×
120	Locale.US,
121	"Call timeout set to '%d' ns, due to context deadline.", remainingNanos));	×
122	if (deadline == null) {	×
123	builder.append(" Explicit call timeout was not set.");	×
124	} else {
125	long callTimeout = deadline.timeRemaining(TimeUnit.NANOSECONDS);	×
126	builder.append(String.format(	×
127	Locale.US, " Explicit call timeout was '%d' ns.", callTimeout));	×
128	}
129	logger.fine(builder.toString());	×
130	}	×
131	} else {
132	return null;	1✔
133	}
134
135	/* Cancels the call if deadline exceeded prior to the real call being set. */
136	class DeadlineExceededRunnable implements Runnable {	1✔
137	@Override
138	public void run() {
139	long seconds = Math.abs(remainingNanos) / TimeUnit.SECONDS.toNanos(1);	1✔
140	long nanos = Math.abs(remainingNanos) % TimeUnit.SECONDS.toNanos(1);	1✔
141	StringBuilder buf = new StringBuilder();	1✔
142	if (remainingNanos < 0) {	1✔
143	buf.append("ClientCall started after ");	1✔
144	buf.append(deadlineName);	1✔
145	buf.append(" deadline was exceeded. Deadline has been exceeded for ");	1✔
146	} else {
147	buf.append("Deadline ");	1✔
148	buf.append(deadlineName);	1✔
149	buf.append(" was exceeded after ");	1✔
150	}
151	buf.append(seconds);	1✔
152	buf.append(String.format(Locale.US, ".%09d", nanos));	1✔
153	buf.append("s waiting for ");	1✔
154	buf.append(bufferContext);	1✔
155	cancel(	1✔
156	Status.DEADLINE_EXCEEDED.withDescription(buf.toString()),	1✔
157	// We should not cancel the call if the realCall is set because there could be a
158	// race between cancel() and realCall.start(). The realCall will handle deadline by
159	// itself.
160	/* onlyCancelPendingCall= */ true);
161	}	1✔
162	}
163
164	return scheduler.schedule(new DeadlineExceededRunnable(), remainingNanos, NANOSECONDS);	1✔
165	}
166
167	/**
168	* Transfers all pending and future requests and mutations to the given call.
169	*
170	* <p>No-op if either this method or {@link #cancel} have already been called.
171	*/
172	// When this method returns, passThrough is guaranteed to be true
173	public final Runnable setCall(ClientCall<ReqT, RespT> call) {
174	Listener<RespT> savedDelayedListener;
175	synchronized (this) {	1✔
176	// If realCall != null, then either setCall() or cancel() has been called.
177	if (realCall != null) {	1✔
178	return null;	1✔
179	}
180	setRealCall(checkNotNull(call, "call"));	1✔
181	// start() not yet called
182	if (delayedListener == null) {	1✔
183	assert pendingRunnables.isEmpty();	1✔
184	pendingRunnables = null;	1✔
185	passThrough = true;	1✔
186	return null;	1✔
187	}
188	savedDelayedListener = this.delayedListener;	1✔
189	}	1✔
190	internalStart(savedDelayedListener);	1✔
191	return new ContextRunnable(context) {	1✔
192	@Override
193	public void runInContext() {
194	drainPendingCalls();	1✔
195	}	1✔
196	};
197	}
198
199	private void internalStart(Listener<RespT> listener) {
200	Metadata savedStartHeaders = this.startHeaders;	1✔
201	this.startHeaders = null;	1✔
202	context.run(() -> realCall.start(listener, savedStartHeaders));	1✔
203	}	1✔
204
205	@Override
206	public final void start(Listener<RespT> listener, final Metadata headers) {
207	checkNotNull(headers, "headers");	1✔
208	checkState(this.listener == null, "already started");	1✔
209	Status savedError;
210	boolean savedPassThrough;
211	synchronized (this) {	1✔
212	this.listener = checkNotNull(listener, "listener");	1✔
213	// If error != null, then cancel() has been called and was unable to close the listener
214	savedError = error;	1✔
215	savedPassThrough = passThrough;	1✔
216	if (!savedPassThrough) {	1✔
217	listener = delayedListener = new DelayedListener<>(this, listener);	1✔
218	startHeaders = headers;	1✔
219	}
220	}	1✔
221	if (savedError != null) {	1✔
222	callExecutor.execute(new CloseListenerRunnable(listener, savedError));	1✔
223	return;	1✔
224	}
225	if (savedPassThrough) {	1✔
226	realCall.start(listener, headers);	1✔
227	} // else realCall.start() will be called by setCall
228	}	1✔
229
230	// When this method returns, passThrough is guaranteed to be true
231	@Override
232	public final void cancel(@Nullable final String message, @Nullable final Throwable cause) {
233	Status status = Status.CANCELLED;	1✔
234	if (message != null) {	1✔
235	status = status.withDescription(message);	1✔
236	} else {
237	status = status.withDescription("Call cancelled without message");	1✔
238	}
239	if (cause != null) {	1✔
240	status = status.withCause(cause);	1✔
241	}
242	cancel(status, false);	1✔
243	}	1✔
244
245	/**
246	* Cancels the call unless {@code realCall} is set and {@code onlyCancelPendingCall} is true.
247	*/
248	private void cancel(final Status status, boolean onlyCancelPendingCall) {
249	boolean delegateToRealCall = true;	1✔
250	Listener<RespT> listenerToClose = null;	1✔
251	synchronized (this) {	1✔
252	// If realCall != null, then either setCall() or cancel() has been called
253	if (realCall == null) {	1✔
254	@SuppressWarnings("unchecked")
255	ClientCall<ReqT, RespT> noopCall = (ClientCall<ReqT, RespT>) NOOP_CALL;	1✔
256	setRealCall(noopCall);	1✔
257	delegateToRealCall = false;	1✔
258	// If listener == null, then start() will later call listener with 'error'
259	listenerToClose = listener;	1✔
260	error = status;	1✔
261	} else if (onlyCancelPendingCall) {	1✔
262	return;	×
263	}
264	}	1✔
265	if (delegateToRealCall) {	1✔
266	delayOrExecute(new Runnable() {	1✔
267	@Override
268	public void run() {
269	realCall.cancel(status.getDescription(), status.getCause());	1✔
270	}	1✔
271	});
272	} else {
273	if (listenerToClose != null) {	1✔
274	callExecutor.execute(new CloseListenerRunnable(listenerToClose, status));	1✔
275	}
276	internalStart(listenerToClose); // listener instance doesn't matter	1✔
277	drainPendingCalls();	1✔
278	}
279	callCancelled();	1✔
280	}	1✔
281
282	protected void callCancelled() {
283	}	1✔
284
285	private void delayOrExecute(Runnable runnable) {
286	synchronized (this) {	1✔
287	if (!passThrough) {	1✔
288	pendingRunnables.add(runnable);	1✔
289	return;	1✔
290	}
291	}	1✔
292	runnable.run();	1✔
293	}	1✔
294
295	/**
296	* Called to transition {@code passThrough} to {@code true}. This method is not safe to be called
297	* multiple times; the caller must ensure it will only be called once, ever. {@code this} lock
298	* should not be held when calling this method.
299	*/
300	private void drainPendingCalls() {
301	assert realCall != null;	1✔
302	assert !passThrough;	1✔
303	List<Runnable> toRun = new ArrayList<>();	1✔
304	DelayedListener<RespT> delayedListener ;
305	while (true) {
306	synchronized (this) {	1✔
307	if (pendingRunnables.isEmpty()) {	1✔
308	pendingRunnables = null;	1✔
309	passThrough = true;	1✔
310	delayedListener = this.delayedListener;	1✔
311	break;	1✔
312	}
313	// Since there were pendingCalls, we need to process them. To maintain ordering we can't set
314	// passThrough=true until we run all pendingCalls, but new Runnables may be added after we
315	// drop the lock. So we will have to re-check pendingCalls.
316	List<Runnable> tmp = toRun;	1✔
317	toRun = pendingRunnables;	1✔
318	pendingRunnables = tmp;	1✔
319	}	1✔
320	for (Runnable runnable : toRun) {	1✔
321	// Must not call transport while lock is held to prevent deadlocks.
322	// TODO(ejona): exception handling
323	runnable.run();	1✔
324	}	1✔
325	toRun.clear();	1✔
326	}
327	if (delayedListener != null) {	1✔
328	final DelayedListener<RespT> listener = delayedListener;	1✔
329	class DrainListenerRunnable extends ContextRunnable {
330	DrainListenerRunnable() {	1✔
331	super(context);	1✔
332	}	1✔
333
334	@Override
335	public void runInContext() {
336	listener.drainPendingCallbacks();	1✔
337	}	1✔
338	}
339
340	callExecutor.execute(new DrainListenerRunnable());	1✔
341	}
342	}	1✔
343
344	@GuardedBy("this")
345	private void setRealCall(ClientCall<ReqT, RespT> realCall) {
346	checkState(this.realCall == null, "realCall already set to %s", this.realCall);	1✔
347	if (initialDeadlineMonitor != null) {	1✔
348	initialDeadlineMonitor.cancel(false);	1✔
349	}
350	this.realCall = realCall;	1✔
351	}	1✔
352
353	@VisibleForTesting
354	final ClientCall<ReqT, RespT> getRealCall() {
355	return realCall;	×
356	}
357
358	@Override
359	public final void sendMessage(final ReqT message) {
360	if (passThrough) {	1✔
361	realCall.sendMessage(message);	1✔
362	} else {
363	delayOrExecute(new Runnable() {	1✔
364	@Override
365	public void run() {
366	realCall.sendMessage(message);	1✔
367	}	1✔
368	});
369	}
370	}	1✔
371
372	@Override
373	public final void setMessageCompression(final boolean enable) {
374	if (passThrough) {	1✔
375	realCall.setMessageCompression(enable);	1✔
376	} else {
377	delayOrExecute(new Runnable() {	1✔
378	@Override
379	public void run() {
380	realCall.setMessageCompression(enable);	1✔
381	}	1✔
382	});
383	}
384	}	1✔
385
386	@Override
387	public final void request(final int numMessages) {
388	if (passThrough) {	1✔
389	realCall.request(numMessages);	1✔
390	} else {
391	delayOrExecute(new Runnable() {	1✔
392	@Override
393	public void run() {
394	realCall.request(numMessages);	1✔
395	}	1✔
396	});
397	}
398	}	1✔
399
400	@Override
401	public final void halfClose() {
402	delayOrExecute(new Runnable() {	1✔
403	@Override
404	public void run() {
405	realCall.halfClose();	1✔
406	}	1✔
407	});
408	}	1✔
409
410	@Override
411	public final boolean isReady() {
412	if (passThrough) {	1✔
413	return realCall.isReady();	1✔
414	} else {
415	return false;	×
416	}
417	}
418
419	@Override
420	public final Attributes getAttributes() {
421	ClientCall<ReqT, RespT> savedRealCall;
422	synchronized (this) {	1✔
423	savedRealCall = realCall;	1✔
424	}	1✔
425	if (savedRealCall != null) {	1✔
426	return savedRealCall.getAttributes();	1✔
427	} else {
428	return Attributes.EMPTY;	×
429	}
430	}
431
432	@Override
433	public String toString() {
434	return MoreObjects.toStringHelper(this)	×
435	.add("realCall", realCall)	×
436	.toString();	×
437	}
438
439	private final class CloseListenerRunnable extends ContextRunnable {
440	final Listener<RespT> listener;
441	final Status status;
442
443	CloseListenerRunnable(Listener<RespT> listener, Status status) {	1✔
444	super(context);	1✔
445	this.listener = listener;	1✔
446	this.status = status;	1✔
447	}	1✔
448
449	@Override
450	public void runInContext() {
451	listener.onClose(status, new Metadata());	1✔
452	}	1✔
453	}
454
455	private static final class DelayedListener<RespT> extends Listener<RespT> {	1✔
456	private final DelayedClientCall<?, RespT> call;
457	private final Listener<RespT> realListener;
458	private volatile boolean passThrough;
459	private volatile Status exceptionStatus;
460	@GuardedBy("this")	1✔
461	private List<Runnable> pendingCallbacks = new ArrayList<>();
462
463	public DelayedListener(DelayedClientCall<?, RespT> call, Listener<RespT> listener) {	1✔
464	this.call = call;	1✔
465	this.realListener = listener;	1✔
466	}	1✔
467
468	/**
469	* Cancels call and schedules onClose() notification. May only be called from within a
470	* DelayedListener callback dispatch (either queued drain or passThrough). Visibility of the
471	* write to {@code exceptionStatus} does not rely on a single callback executor; it is a
472	* {@code volatile} field, and callback queuing/pass-through transitions are coordinated by
473	* this listener's synchronization so subsequent callbacks observe the updated status.
474	*/
475	private void exceptionThrown(Throwable t, String description) {
476	// onClose() must be delivered exactly once and last. Other callbacks may already be queued
477	// ahead of realCall's eventual onClose, so we can't call onClose() here. We set the status
478	// and overwrite the onClose() details when it arrives.
479	exceptionStatus = Status.CANCELLED.withCause(t).withDescription(description);	1✔
480	call.cancel(description, t);	1✔
481	}	1✔
482
483	private void delayOrExecute(Runnable runnable) {
484	synchronized (this) {	1✔
485	if (!passThrough) {	1✔
486	pendingCallbacks.add(runnable);	1✔
487	return;	1✔
488	}
489	}	1✔
490	runnable.run();	1✔
491	}	1✔
492
493	@Override
494	public void onHeaders(final Metadata headers) {
495	if (passThrough) {	1✔
496	deliverHeaders(headers);	1✔
497	} else {
498	delayOrExecute(new Runnable() {	1✔
499	@Override
500	public void run() {
501	deliverHeaders(headers);	1✔
502	}	1✔
503	});
504	}
505	}	1✔
506
507	private void deliverHeaders(Metadata headers) {
508	if (exceptionStatus != null) {	1✔
509	return;	×
510	}
511	try {
512	realListener.onHeaders(headers);	1✔
513	} catch (Throwable t) {	1✔
514	exceptionThrown(t, "Failed to read headers");	1✔
515	}	1✔
516	}	1✔
517
518	@Override
519	public void onMessage(final RespT message) {
520	if (passThrough) {	1✔
521	deliverMessage(message);	1✔
522	} else {
523	delayOrExecute(new Runnable() {	1✔
524	@Override
525	public void run() {
526	deliverMessage(message);	1✔
527	}	1✔
528	});
529	}
530	}	1✔
531
532	private void deliverMessage(RespT message) {
533	if (exceptionStatus != null) {	1✔
534	return;	1✔
535	}
536	try {
537	realListener.onMessage(message);	1✔
538	} catch (Throwable t) {	1✔
539	exceptionThrown(t, "Failed to read message.");	1✔
540	}	1✔
541	}	1✔
542
543	@Override
544	public void onClose(final Status status, final Metadata trailers) {
545	delayOrExecute(new Runnable() {	1✔
546	@Override
547	public void run() {
548	Status effectiveStatus = status;	1✔
549	Metadata effectiveTrailers = trailers;	1✔
550	if (exceptionStatus != null) {	1✔
551	// Ideally status matches exceptionStatus, since exceptionStatus was used to cancel
552	// the call. However, cancel() may reconstruct a new Status instance, and the cancel
553	// is racy so this onClose may have already been queued when the cancellation
554	// occurred. Since other callbacks throw away data if exceptionStatus != null, it is
555	// semantically essential that we _not_ use a status provided by the server.
556	effectiveStatus = exceptionStatus;	1✔
557	// Replace trailers to prevent mixing sources of status and trailers.
558	effectiveTrailers = new Metadata();	1✔
559	}
560	try {
561	realListener.onClose(effectiveStatus, effectiveTrailers);	1✔
562	} catch (RuntimeException ex) {	1✔
563	logger.log(Level.WARNING, "Exception thrown by onClose() in ClientCall", ex);	1✔
564	}	1✔
565	}	1✔
566	});
567	}	1✔
568
569	@Override
570	public void onReady() {
571	if (passThrough) {	1✔
572	deliverOnReady();	1✔
573	} else {
574	delayOrExecute(new Runnable() {	1✔
575	@Override
576	public void run() {
577	deliverOnReady();	1✔
578	}	1✔
579	});
580	}
581	}	1✔
582
583	private void deliverOnReady() {
584	if (exceptionStatus != null) {	1✔
585	return;	×
586	}
587	try {
588	realListener.onReady();	1✔
589	} catch (Throwable t) {	1✔
590	exceptionThrown(t, "Failed to call onReady.");	1✔
591	}	1✔
592	}	1✔
593
594	void drainPendingCallbacks() {
595	assert !passThrough;	1✔
596	List<Runnable> toRun = new ArrayList<>();	1✔
597	while (true) {
598	synchronized (this) {	1✔
599	if (pendingCallbacks.isEmpty()) {	1✔
600	pendingCallbacks = null;	1✔
601	passThrough = true;	1✔
602	break;	1✔
603	}
604	// Since there were pendingCallbacks, we need to process them. To maintain ordering we
605	// can't set passThrough=true until we run all pendingCallbacks, but new Runnables may be
606	// added after we drop the lock. So we will have to re-check pendingCallbacks.
607	List<Runnable> tmp = toRun;	1✔
608	toRun = pendingCallbacks;	1✔
609	pendingCallbacks = tmp;	1✔
610	}	1✔
611	for (Runnable runnable : toRun) {	1✔
612	// Avoid calling listener while lock is held to prevent deadlocks.
613	runnable.run();	1✔
614	}	1✔
615	toRun.clear();	1✔
616	}
617	}	1✔
618	}
619
620	private static final ClientCall<Object, Object> NOOP_CALL = new ClientCall<Object, Object>() {	1✔
621	@Override
622	public void start(Listener<Object> responseListener, Metadata headers) {}	1✔
623
624	@Override
625	public void request(int numMessages) {}	1✔
626
627	@Override
628	public void cancel(String message, Throwable cause) {}	1✔
629
630	@Override
631	public void halfClose() {}	1✔
632
633	@Override
634	public void sendMessage(Object message) {}	1✔
635
636	// Always returns {@code false}, since this is only used when the startup of the call fails.
637	@Override
638	public boolean isReady() {
639	return false;	×
640	}
641	};
642	}

grpc / grpc-java / #20347

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