#20349

Committed 07 Jul 2026 10:24AM UTC coverage: 89.111% (-0.01%) from 89.122%

Build # #20349

Build Type

push

github

Committed by

web-flow

Commit Message

enable child channel plugins (#12578)

Implements https://github.com/grpc/proposal/pull/529

Coverage Stats

38030 of 42677 relevant lines covered (89.11%)

0.89 hits per line

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

90.63

/../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));	×
223	return;	×
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;	1✔
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 / #20349

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