#20262

Committed 05 May 2026 04:08PM UTC coverage: 88.807% (-0.002%) from 88.809%

Build # #20262

Build Type

push

github

Committed by

ejona86

Commit Message

Upgrading Build Environment to Ubuntu 24.04
1. Transition to Ubuntu 24.04
Ubuntu 18.04 has reached End-of-Life (EOL), making standard package repositories unreliable. More critically, legacy versions like 18.04 and 20.04 lack modern root SSL certificates required to securely communicate with internal build infrastructure, leading to handshake failures. Moving to Ubuntu 24.04 provides a modern baseline with updated certificate stores that trust required mirrors out-of-the-box.

2. Mandatory Use of Internal Mirrors
The Kokoro grpc-ubuntu22 worker pool operates under strict egress restrictions that block connections to public Ubuntu archives (e.g., archive.ubuntu.com), resulting in Connection failed errors. To bypass this firewall, we must redirect apt traffic to the internal Google mirror (mirror.bazel.build) using a sed command in the Dockerfile.

3. Bumping JDK to Version 11
Ubuntu 24.04 no longer includes the ancient openjdk-8-jdk in its default repositories. To ensure compatibility with the new OS and maintain a supportable toolchain, we have bumped the version to openjdk-11-jdk.

4. Transition from pkg-config to pkgconf
Following modern Debian/Ubuntu packaging standards, we have replaced pkg-config with pkgconf which is the preferred, actively maintained implementation for package discovery in newer Ubuntu releases.

Coverage Stats

36140 of 40695 relevant lines covered (88.81%)

0.89 hits per line

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

90.91

/../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<>(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 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<>(this, listener);	1✔
210	startHeaders = headers;	1✔
211	}
212	}	1✔
213	if (savedError != null) {	1✔
214	callExecutor.execute(new CloseListenerRunnable(listener, savedError));	1✔
215	return;	1✔
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 DelayedClientCall<?, RespT> call;
449	private final Listener<RespT> realListener;
450	private volatile boolean passThrough;
451	private volatile Status exceptionStatus;
452	@GuardedBy("this")	1✔
453	private List<Runnable> pendingCallbacks = new ArrayList<>();
454
455	public DelayedListener(DelayedClientCall<?, RespT> call, Listener<RespT> listener) {	1✔
456	this.call = call;	1✔
457	this.realListener = listener;	1✔
458	}	1✔
459
460	/**
461	* Cancels call and schedules onClose() notification. May only be called from within a
462	* DelayedListener callback dispatch (either queued drain or passThrough). Visibility of the
463	* write to {@code exceptionStatus} does not rely on a single callback executor; it is a
464	* {@code volatile} field, and callback queuing/pass-through transitions are coordinated by
465	* this listener's synchronization so subsequent callbacks observe the updated status.
466	*/
467	private void exceptionThrown(Throwable t, String description) {
468	// onClose() must be delivered exactly once and last. Other callbacks may already be queued
469	// ahead of realCall's eventual onClose, so we can't call onClose() here. We set the status
470	// and overwrite the onClose() details when it arrives.
471	exceptionStatus = Status.CANCELLED.withCause(t).withDescription(description);	1✔
472	call.cancel(description, t);	1✔
473	}	1✔
474
475	private void delayOrExecute(Runnable runnable) {
476	synchronized (this) {	1✔
477	if (!passThrough) {	1✔
478	pendingCallbacks.add(runnable);	1✔
479	return;	1✔
480	}
481	}	1✔
482	runnable.run();	1✔
483	}	1✔
484
485	@Override
486	public void onHeaders(final Metadata headers) {
487	if (passThrough) {	1✔
488	deliverHeaders(headers);	1✔
489	} else {
490	delayOrExecute(new Runnable() {	1✔
491	@Override
492	public void run() {
493	deliverHeaders(headers);	1✔
494	}	1✔
495	});
496	}
497	}	1✔
498
499	private void deliverHeaders(Metadata headers) {
500	if (exceptionStatus != null) {	1✔
501	return;	×
502	}
503	try {
504	realListener.onHeaders(headers);	1✔
505	} catch (Throwable t) {	1✔
506	exceptionThrown(t, "Failed to read headers");	1✔
507	}	1✔
508	}	1✔
509
510	@Override
511	public void onMessage(final RespT message) {
512	if (passThrough) {	1✔
513	deliverMessage(message);	1✔
514	} else {
515	delayOrExecute(new Runnable() {	1✔
516	@Override
517	public void run() {
518	deliverMessage(message);	1✔
519	}	1✔
520	});
521	}
522	}	1✔
523
524	private void deliverMessage(RespT message) {
525	if (exceptionStatus != null) {	1✔
526	return;	1✔
527	}
528	try {
529	realListener.onMessage(message);	1✔
530	} catch (Throwable t) {	1✔
531	exceptionThrown(t, "Failed to read message.");	1✔
532	}	1✔
533	}	1✔
534
535	@Override
536	public void onClose(final Status status, final Metadata trailers) {
537	delayOrExecute(new Runnable() {	1✔
538	@Override
539	public void run() {
540	Status effectiveStatus = status;	1✔
541	Metadata effectiveTrailers = trailers;	1✔
542	if (exceptionStatus != null) {	1✔
543	// Ideally status matches exceptionStatus, since exceptionStatus was used to cancel
544	// the call. However, cancel() may reconstruct a new Status instance, and the cancel
545	// is racy so this onClose may have already been queued when the cancellation
546	// occurred. Since other callbacks throw away data if exceptionStatus != null, it is
547	// semantically essential that we _not_ use a status provided by the server.
548	effectiveStatus = exceptionStatus;	1✔
549	// Replace trailers to prevent mixing sources of status and trailers.
550	effectiveTrailers = new Metadata();	1✔
551	}
552	try {
553	realListener.onClose(effectiveStatus, effectiveTrailers);	1✔
554	} catch (RuntimeException ex) {	1✔
555	logger.log(Level.WARNING, "Exception thrown by onClose() in ClientCall", ex);	1✔
556	}	1✔
557	}	1✔
558	});
559	}	1✔
560
561	@Override
562	public void onReady() {
563	if (passThrough) {	1✔
564	deliverOnReady();	1✔
565	} else {
566	delayOrExecute(new Runnable() {	1✔
567	@Override
568	public void run() {
569	deliverOnReady();	1✔
570	}	1✔
571	});
572	}
573	}	1✔
574
575	private void deliverOnReady() {
576	if (exceptionStatus != null) {	1✔
577	return;	×
578	}
579	try {
580	realListener.onReady();	1✔
581	} catch (Throwable t) {	1✔
582	exceptionThrown(t, "Failed to call onReady.");	1✔
583	}	1✔
584	}	1✔
585
586	void drainPendingCallbacks() {
587	assert !passThrough;	1✔
588	List<Runnable> toRun = new ArrayList<>();	1✔
589	while (true) {
590	synchronized (this) {	1✔
591	if (pendingCallbacks.isEmpty()) {	1✔
592	pendingCallbacks = null;	1✔
593	passThrough = true;	1✔
594	break;	1✔
595	}
596	// Since there were pendingCallbacks, we need to process them. To maintain ordering we
597	// can't set passThrough=true until we run all pendingCallbacks, but new Runnables may be
598	// added after we drop the lock. So we will have to re-check pendingCallbacks.
599	List<Runnable> tmp = toRun;	1✔
600	toRun = pendingCallbacks;	1✔
601	pendingCallbacks = tmp;	1✔
602	}	1✔
603	for (Runnable runnable : toRun) {	1✔
604	// Avoid calling listener while lock is held to prevent deadlocks.
605	runnable.run();	1✔
606	}	1✔
607	toRun.clear();	1✔
608	}
609	}	1✔
610	}
611
612	private static final ClientCall<Object, Object> NOOP_CALL = new ClientCall<Object, Object>() {	1✔
613	@Override
614	public void start(Listener<Object> responseListener, Metadata headers) {}	1✔
615
616	@Override
617	public void request(int numMessages) {}	1✔
618
619	@Override
620	public void cancel(String message, Throwable cause) {}	1✔
621
622	@Override
623	public void halfClose() {}	1✔
624
625	@Override
626	public void sendMessage(Object message) {}	1✔
627
628	// Always returns {@code false}, since this is only used when the startup of the call fails.
629	@Override
630	public boolean isReady() {
631	return false;	×
632	}
633	};
634	}

grpc / grpc-java / #20262

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