grpc
/
grpc-java
mirror of https://github.com/grpc/grpc-java.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

core: promote LBv2 classes.

This commit is contained in:
Kun Zhang 2017-02-21 17:11:03 -08:00 committed by Kun Zhang
parent a6b1d8981a
commit a9bd94781c
25 changed files with 623 additions and 594 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/main/java/io/grpc/Internal.java
View File

@ -41,7 +41,7 @@ import java.lang.annotation.Target;
* Annotates a program element (class, method, package etc) which is internal to gRPC, not part of * Annotates a program element (class, method, package etc) which is internal to gRPC, not part of
* the public API, and should not be used by users of gRPC. * the public API, and should not be used by users of gRPC.
* *
* <p>However, if you want to implement a custom {@link LoadBalancer2}, an alternative transport, or * <p>However, if you want to implement a custom {@link LoadBalancer}, an alternative transport, or
* anything else that will be wired into gRPC library, you may use the internal parts. Please * anything else that will be wired into gRPC library, you may use the internal parts. Please
* consult the gRPC team first, because internal APIs don't have the same API stability guarantee as * consult the gRPC team first, because internal APIs don't have the same API stability guarantee as
* the public APIs do. * the public APIs do.

414
core/src/main/java/io/grpc/LoadBalancer.java Normal file
View File

@ -0,0 +1,414 @@
/*
* Copyright 2016, Google Inc. All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are
* met:
*
* * Redistributions of source code must retain the above copyright
* notice, this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above
* copyright notice, this list of conditions and the following disclaimer
* in the documentation and/or other materials provided with the
* distribution.
*
* * Neither the name of Google Inc. nor the names of its
* contributors may be used to endorse or promote products derived from
* this software without specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
*/
package io.grpc;
import com.google.common.base.Preconditions;
import java.util.List;
import javax.annotation.Nullable;
import javax.annotation.concurrent.Immutable;
import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;
/**
* A pluggable component that receives resolved addresses from {@link NameResolver} and provides the
* channel a usable subchannel when asked.
*
* <h3>Overview</h3>
*
* <p>A LoadBalancer typically implements three interfaces:
* <ol>
* <li>{@link LoadBalancer} is the main interface. All methods on it are invoked sequentially
* from the Channel Executor. It receives the results from the {@link NameResolver}, updates
* of subchannels' connectivity states, and the channel's request for the LoadBalancer to
* shutdown.</li>
* <li>{@link SubchannelPicker SubchannelPicker} does the actual load-balancing work. It selects
* a {@link Subchannel Subchannel} for each new RPC.</li>
* <li>{@link Factory Factory} creates a new {@link LoadBalancer} instance.
* </ol>
*
* <p>{@link Helper Helper} is implemented by gRPC library and provided to {@link Factory
* Factory}. It provides functionalities that a {@code LoadBalancer} implementation would typically
* need.
*
* <h3>Channel Executor</h3>
*
* <p>Channel Executor is an internal executor of the channel, which is used to serialize all the
* callback methods on the {@link LoadBalancer} interface, thus the balancer implementation doesn't
* need to worry about synchronization among them. However, the actual thread of the Channel
* Executor is typically the network thread, thus the following rules must be followed to prevent
* blocking or even dead-locking in a network
*
* <ol>
*
* <li><strong>Never block in Channel Executor</strong>. The callback methods must return
* quickly. Examples or work that must be avoided: CPU-intensive calculation, waiting on
* synchronization primitives, blocking I/O, blocking RPCs, etc.</li>
*
* <li><strong>Avoid calling into other components with lock held</strong>. Channel Executor may
* run callbacks under a lock, e.g., the transport lock of OkHttp. If your LoadBalancer has a
* lock, holds the lock in a callback method (e.g., {@link #handleSubchannelState
* handleSubchannelState()}) while calling into another class that may involve locks, be cautious
* of deadlock. Generally you wouldn't need any locking in the LoadBalancer.</li>
*
* </ol>
*
* <p>{@link Helper#runSerialized Helper.runSerialized()} allows you to schedule a task to be run in
* the Channel Executor.
*
* <h3>The canonical implementation pattern</h3>
*
* <p>A {@link LoadBalancer} keeps states like the latest addresses from NameResolver, the
* Subchannel(s) and their latest connectivity states. These states are mutated within the Channel
* Executor.
*
* <p>A typical {@link SubchannelPicker SubchannelPicker} holds a snapshot of these states. It may
* have its own states, e.g., a picker from a round-robin load-balancer may keep a pointer to the
* next Subchannel, which are typically mutated by multiple threads. The picker should only mutate
* its own state, and should not mutate or re-acquire the states of the LoadBalancer. This way the
* picker only needs to synchronize its own states, which is typically trivial to implement.
*
* <p>When the LoadBalancer states changes, e.g., Subchannels has become or stopped being READY, and
* we want subsequent RPCs to use the latest list of READY Subchannels, LoadBalancer would create
* a new picker, which holds a snapshot of the latest Subchannel list. Refer to the javadoc of
* {@link #handleSubchannelState handleSubchannelState()} how to do this properly.
*
* <p>No synchronization should be necessary between LoadBalancer and its pickers if you follow
* the pattern above. It may be possible to implement in a different way, but that would usually
* result in more complicated threading.
*/
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771")
@NotThreadSafe
public abstract class LoadBalancer {
/**
* Handles newly resolved server groups and metadata attributes from name resolution system.
* {@code servers} contained in {@link ResolvedServerInfoGroup} should be considered equivalent
* but may be flattened into a single list if needed.
*
* <p>Implementations should not modify the given {@code servers}.
*
* @param servers the resolved server addresses, never empty.
* @param attributes extra metadata from naming system.
*/
public abstract void handleResolvedAddresses(
List<ResolvedServerInfoGroup> servers, Attributes attributes);
/**
* Handles an error from the name resolution system.
*
* @param error a non-OK status
*/
public abstract void handleNameResolutionError(Status error);
/**
* Handles a state change on a Subchannel.
*
* <p>The initial state of a Subchannel is IDLE. You won't get a notification for the initial IDLE
* state.
*
* <p>If the new state is not SHUTDOWN, this method should create a new picker and call {@link
* Helper#updatePicker Helper.updatePicker()}. Failing to do so may result in unnecessary delays
* of RPCs. Please refer to {@link PickResult#withSubchannel PickResult.withSubchannel()}'s
* javadoc for more information.
*
* <p>SHUTDOWN can only happen in two cases. One is that LoadBalancer called {@link
* Subchannel#shutdown} earlier, thus it should have already discarded this Subchannel. The other
* is that Channel is doing a {@link ManagedChannel#shutdownNow forced shutdown} or has already
* terminated, thus there won't be further requests to LoadBalancer. Therefore, SHUTDOWN can be
* safely ignored.
*
* @param subchannel the involved Subchannel
* @param stateInfo the new state
*/
public abstract void handleSubchannelState(
Subchannel subchannel, ConnectivityStateInfo stateInfo);
/**
* The channel asks the load-balancer to shutdown. No more callbacks will be called after this
* method. The implementation should shutdown all Subchannels and OOB channels, and do any other
* cleanup as necessary.
*/
public abstract void shutdown();
/**
* The main balancing logic. It <strong>must be thread-safe</strong>. Typically it should only
* synchronize on its own state, and avoid synchronizing with the LoadBalancer's state.
*/
@ThreadSafe
public abstract static class SubchannelPicker {
/**
* Make a balancing decision for a new RPC.
*
* @param affinity the affinity attributes provided via {@link CallOptions#withAffinity}
* @param headers the headers container of the RPC. It can be mutated within this method.
*/
public abstract PickResult pickSubchannel(Attributes affinity, Metadata headers);
}
/**
* A balancing decision made by {@link SubchannelPicker SubchannelPicker} for an RPC.
*
* <p>The outcome of the decision will be one of the following:
* <ul>
* <li>Proceed: if a Subchannel is provided via {@link #withSubchannel withSubchannel()}, and is
* in READY state when the RPC tries to start on it, the RPC will proceed on that
* Subchannel.</li>
* <li>Error: if an error is provided via {@link #withError withError()}, and the RPC is not
* wait-for-ready (i.e., {@link CallOptions#withWaitForReady} was not called), the RPC will
* fail immediately with the given error.</li>
* <li>Buffer: in all other cases, the RPC will be buffered in the Channel, until the next
* picker is provided via {@link Helper#updatePicker Helper.updatePicker()}, when the RPC
* will go through the same picking process again.</li>
* </ul>
*/
@Immutable
public static final class PickResult {
private static final PickResult NO_RESULT = new PickResult(null, Status.OK);
@Nullable private final Subchannel subchannel;
// An error to be propagated to the application if subchannel == null
// Or OK if there is no error.
// subchannel being null and error being OK means RPC needs to wait
private final Status status;
private PickResult(Subchannel subchannel, Status status) {
this.subchannel = subchannel;
this.status = Preconditions.checkNotNull(status, "status");
}
/**
* A decision to proceed the RPC on a Subchannel.
*
* <p>Only Subchannels returned by {@link Helper#createSubchannel Helper.createSubchannel()}
* will work. DO NOT try to use your own implementations of Subchannels, as they won't work.
*
* <p>When the RPC tries to use the return Subchannel, which is briefly after this method
* returns, the state of the Subchannel will decide where the RPC would go:
*
* <ul>
* <li>READY: the RPC will proceed on this Subchannel.</li>
* <li>IDLE: the RPC will be buffered. Subchannel will attempt to create connection.</li>
* <li>All other states: the RPC will be buffered.</li>
* </ul>
*
* <p><strong>All buffered RPCs will stay buffered</strong> until the next call of {@link
* Helper#updatePicker Helper.updatePicker()}, which will trigger a new picking process.
*
* <p>Note that Subchannel's state may change at the same time the picker is making the
* decision, which means the decision may be made with (to-be) outdated information. For
* example, a picker may return a Subchannel known to be READY, but it has become IDLE when is
* about to be used by the RPC, which makes the RPC to be buffered. The LoadBalancer will soon
* learn about the Subchannels' transition from READY to IDLE, create a new picker and allow the
* RPC to use another READY transport if there is any.
*
* <p>You will want to avoid running into a situation where there are READY Subchannels out
* there but some RPCs are still buffered for longer than a brief time.
* <ul>
* <li>This can happen if you return Subchannels with states other than READY and IDLE. For
* example, suppose you round-robin on 2 Subchannels, in READY and CONNECTING states
* respectively. If the picker ignores the state and pick them equally, 50% of RPCs will
* be stuck in buffered state until both Subchannels are READY.</li>
* <li>This can also happen if you don't create a new picker at key state changes of
* Subchannels. Take the above round-robin example again. Suppose you do pick only READY
* and IDLE Subchannels, and initially both Subchannels are READY. Now one becomes IDLE,
* then CONNECTING and stays CONNECTING for a long time. If you don't create a new picker
* in response to the CONNECTING state to exclude that Subchannel, 50% of RPCs will hit it
* and be buffered even though the other Subchannel is READY.</li>
* </ul>
*
* <p>In order to prevent unnecessary delay of RPCs, the rules of thumb are:
* <ol>
* <li>The picker should only pick Subchannels that are known as READY or IDLE. Whether to
* pick IDLE Subchannels depends on whether you want Subchannels to connect on-demand or
* actively:
* <ul>
* <li>If you want connect-on-demand, include IDLE Subchannels in your pick results,
* because when an RPC tries to use an IDLE Subchannel, the Subchannel will try to
* connect.</li>
* <li>If you want Subchannels to be always connected even when there is no RPC, you
* would call {@link Subchannel#requestConnection Subchannel.requestConnection()}
* whenever the Subchannel has transitioned to IDLE, then you don't need to include
* IDLE Subchannels in your pick results.</li>
* </ul></li>
* <li>Always create a new picker and call {@link Helper#updatePicker Helper.updatePicker()}
* whenever {@link #handleSubchannelState handleSubchannelState()} is called, unless the
* new state is SHUTDOWN. See {@code handleSubchannelState}'s javadoc for more
* details.</li>
* </ol>
*/
public static PickResult withSubchannel(Subchannel subchannel) {
return new PickResult(Preconditions.checkNotNull(subchannel, "subchannel"), Status.OK);
}
/**
* A decision to report a connectivity error to the RPC. If the RPC is {@link
* CallOptions#withWaitForReady wait-for-ready}, it will stay buffered. Otherwise, it will fail
* with the given error.
*
* @param error the error status. Must not be OK.
*/
public static PickResult withError(Status error) {
Preconditions.checkArgument(!error.isOk(), "error status shouldn't be OK");
return new PickResult(null, error);
}
/**
* No decision could be made. The RPC will stay buffered.
*/
public static PickResult withNoResult() {
return NO_RESULT;
}
/**
* The Subchannel if this result was created by {@link #withSubchannel withSubchannel()}, or
* null otherwise.
*/
@Nullable
public Subchannel getSubchannel() {
return subchannel;
}
/**
* The status associated with this result. Non-{@code OK} if created with {@link #withError
* withError}, or {@code OK} otherwise.
*/
public Status getStatus() {
return status;
}
@Override
public String toString() {
return "[subchannel=" + subchannel + " status=" + status + "]";
}
}
/**
* Provides essentials for LoadBalancer implementations.
*/
@ThreadSafe
public abstract static class Helper {
/**
* Creates a Subchannel, which is a logical connection to the given group of addresses which are
* considered equivalent. The {@code attrs} are custom attributes associated with this
* Subchannel, and can be accessed later through {@link Subchannel#getAttributes
* Subchannel.getAttributes()}.
*
* <p>The LoadBalancer is responsible for closing unused Subchannels, and closing all
* Subchannels within {@link #shutdown}.
*/
public abstract Subchannel createSubchannel(EquivalentAddressGroup addrs, Attributes attrs);
/**
* Out-of-band channel for LoadBalancer’s own RPC needs, e.g., talking to an external
* load-balancer service.
*
* <p>The LoadBalancer is responsible for closing unused OOB channels, and closing all OOB
* channels within {@link #shutdown}.
*/
public abstract ManagedChannel createOobChannel(
EquivalentAddressGroup eag, String authority);
/**
* Set a new picker to the channel.
*
* <p>When a new picker is provided via {@code updatePicker()}, the channel will apply the
* picker on all buffered RPCs, by calling {@link SubchannelPicker#pickSubchannel
* SubchannelPicker.pickSubchannel()}.
*
* <p>The channel will hold the picker and use it for all RPCs, until {@code updatePicker()} is
* called again and a new picker replaces the old one. If {@code updatePicker()} has never been
* called, the channel will buffer all RPCs until a picker is provided.
*/
public abstract void updatePicker(SubchannelPicker picker);
/**
* Schedule a task to be run in the Channel Executor, which serializes the task with the
* callback methods on the {@link LoadBalancer} interface.
*/
public abstract void runSerialized(Runnable task);
/**
* Returns the NameResolver of the channel.
*/
public abstract NameResolver.Factory getNameResolverFactory();
/**
* Returns the authority string of the channel, which is derived from the DNS-style target name.
*/
public abstract String getAuthority();
}
/**
* A logical connection to a server, or a group of equivalent servers represented by an {@link
* EquivalentAddressGroup}.
*
* <p>It maintains at most one physical connection (aka transport) for sending new RPCs, while
* also keeps track of previous transports that has been shut down but not terminated yet.
*
* <p>If there isn't an active transport yet, and an RPC is assigned to the Subchannel, it will
* create a new transport. It won't actively create transports otherwise. {@link
* #requestConnection requestConnection()} can be used to ask Subchannel to create a transport if
* there isn't any.
*/
@ThreadSafe
public abstract static class Subchannel {
/**
* Shuts down the Subchannel. After this method is called, this Subchannel should no longer
* be returned by the latest {@link SubchannelPicker picker}, and can be safely discarded.
*/
public abstract void shutdown();
/**
* Asks the Subchannel to create a connection (aka transport), if there isn't an active one.
*/
public abstract void requestConnection();
/**
* Returns the addresses that this Subchannel is bound to.
*/
public abstract EquivalentAddressGroup getAddresses();
/**
* The same attributes passed to {@link Helper#createSubchannel Helper.createSubchannel()}.
* LoadBalancer can use it to attach additional information here, e.g., the shard this
* Subchannel belongs to.
*/
public abstract Attributes getAttributes();
}
@ThreadSafe
public abstract static class Factory {
/**
* Creates a {@link LoadBalancer} that will be used inside a channel.
*/
public abstract LoadBalancer newLoadBalancer(Helper helper);
}
}

387
core/src/main/java/io/grpc/LoadBalancer2.java
View File

@ -31,390 +31,13 @@
package io.grpc; package io.grpc;
import com.google.common.base.Preconditions;
import java.util.List;
import javax.annotation.Nullable;
import javax.annotation.concurrent.Immutable;
import javax.annotation.concurrent.NotThreadSafe;
import javax.annotation.concurrent.ThreadSafe;
/** /**
* A pluggable component that receives resolved addresses from {@link NameResolver} and provides the * An alias for {@link LoadBalancer}.
* channel a usable subchannel when asked. This is the new interface that will replace {@link
* LoadBalancer2}.
* *
* <p><strong>IMPORTANT NOTICE FOR IMPLEMENTORS: </strong>The name of this class is temporary. It * @deprecated this is going to be deleted in the next minor release. Use {@link LoadBalancer}
* will be renamed to {@code LoadBalancer} eventually. Make sure you have read through <a * instead.
* href="https://github.com/grpc/grpc-java/issues/2656" target="_blank">#2656</a> to understand the
* transition path.
*
* <h3>Overview</h3>
*
* <p>A LoadBalancer typically implements three interfaces:
* <ol>
* <li>{@link LoadBalancer2} is the main interface. All methods on it are invoked sequentially
* from the Channel Executor. It receives the results from the {@link NameResolver}, updates
* of subchannels' connectivity states, and the channel's request for the LoadBalancer to
* shutdown.</li>
* <li>{@link SubchannelPicker SubchannelPicker} does the actual load-balancing work. It selects
* a {@link Subchannel Subchannel} for each new RPC.</li>
* <li>{@link Factory Factory} creates a new {@link LoadBalancer2} instance.
* </ol>
*
* <p>{@link Helper Helper} is implemented by gRPC library and provided to {@link Factory
* Factory}. It provides functionalities that a {@code LoadBalancer2} implementation would typically
* need.
*
* <h3>Channel Executor</h3>
*
* <p>Channel Executor is an internal executor of the channel, which is used to serialize all the
* callback methods on the {@link LoadBalancer2} interface, thus the balancer implementation doesn't
* need to worry about synchronization among them. However, the actual thread of the Channel
* Executor is typically the network thread, thus the following rules must be followed to prevent
* blocking or even dead-locking in a network
*
* <ol>
*
* <li><strong>Never block in Channel Executor</strong>. The callback methods must return
* quickly. Examples or work that must be avoided: CPU-intensive calculation, waiting on
* synchronization primitives, blocking I/O, blocking RPCs, etc.</li>
*
* <li><strong>Avoid calling into other components with lock held</strong>. Channel Executor may
* run callbacks under a lock, e.g., the transport lock of OkHttp. If your LoadBalancer has a
* lock, holds the lock in a callback method (e.g., {@link #handleSubchannelState
* handleSubchannelState()}) while calling into another class that may involve locks, be cautious
* of deadlock. Generally you wouldn't need any locking in the LoadBalancer.</li>
*
* </ol>
*
* <p>{@link Helper#runSerialized Helper.runSerialized()} allows you to schedule a task to be run in
* the Channel Executor.
*
* <h3>The canonical implementation pattern</h3>
*
* <p>A {@link LoadBalancer2} keeps states like the latest addresses from NameResolver, the
* Subchannel(s) and their latest connectivity states. These states are mutated within the Channel
* Executor.
*
* <p>A typical {@link SubchannelPicker SubchannelPicker} holds a snapshot of these states. It may
* have its own states, e.g., a picker from a round-robin load-balancer may keep a pointer to the
* next Subchannel, which are typically mutated by multiple threads. The picker should only mutate
* its own state, and should not mutate or re-acquire the states of the LoadBalancer. This way the
* picker only needs to synchronize its own states, which is typically trivial to implement.
*
* <p>When the LoadBalancer states changes, e.g., Subchannels has become or stopped being READY, and
* we want subsequent RPCs to use the latest list of READY Subchannels, LoadBalancer would create
* a new picker, which holds a snapshot of the latest Subchannel list. Refer to the javadoc of
* {@link #handleSubchannelState handleSubchannelState()} how to do this properly.
*
* <p>No synchronization should be necessary between LoadBalancer and its pickers if you follow
* the pattern above. It may be possible to implement in a different way, but that would usually
* result in more complicated threading.
*/ */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771") @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771")
@NotThreadSafe @Deprecated
public abstract class LoadBalancer2 { public abstract class LoadBalancer2 extends LoadBalancer {
/**
* Handles newly resolved server groups and metadata attributes from name resolution system.
* {@code servers} contained in {@link ResolvedServerInfoGroup} should be considered equivalent
* but may be flattened into a single list if needed.
*
* <p>Implementations should not modify the given {@code servers}.
*
* @param servers the resolved server addresses, never empty.
* @param attributes extra metadata from naming system.
*/
public abstract void handleResolvedAddresses(
List<ResolvedServerInfoGroup> servers, Attributes attributes);
/**
* Handles an error from the name resolution system.
*
* @param error a non-OK status
*/
public abstract void handleNameResolutionError(Status error);
/**
* Handles a state change on a Subchannel.
*
* <p>The initial state of a Subchannel is IDLE. You won't get a notification for the initial IDLE
* state.
*
* <p>If the new state is not SHUTDOWN, this method should create a new picker and call {@link
* Helper#updatePicker Helper.updatePicker()}. Failing to do so may result in unnecessary delays
* of RPCs. Please refer to {@link PickResult#withSubchannel PickResult.withSubchannel()}'s
* javadoc for more information.
*
* <p>SHUTDOWN can only happen in two cases. One is that LoadBalancer called {@link
* Subchannel#shutdown} earlier, thus it should have already discarded this Subchannel. The other
* is that Channel is doing a {@link ManagedChannel#shutdownNow forced shutdown} or has already
* terminated, thus there won't be further requests to LoadBalancer. Therefore, SHUTDOWN can be
* safely ignored.
*
* @param subchannel the involved Subchannel
* @param stateInfo the new state
*/
public abstract void handleSubchannelState(
Subchannel subchannel, ConnectivityStateInfo stateInfo);
/**
* The channel asks the load-balancer to shutdown. No more callbacks will be called after this
* method. The implementation should shutdown all Subchannels and OOB channels, and do any other
* cleanup as necessary.
*/
public abstract void shutdown();
/**
* The main balancing logic. It <strong>must be thread-safe</strong>. Typically it should only
* synchronize on its own state, and avoid synchronizing with the LoadBalancer's state.
*/
@ThreadSafe
public abstract static class SubchannelPicker {
/**
* Make a balancing decision for a new RPC.
*
* @param affinity the affinity attributes provided via {@link CallOptions#withAffinity}
* @param headers the headers container of the RPC. It can be mutated within this method.
*/
public abstract PickResult pickSubchannel(Attributes affinity, Metadata headers);
}
/**
* A balancing decision made by {@link SubchannelPicker SubchannelPicker} for an RPC.
*
* <p>The outcome of the decision will be one of the following:
* <ul>
* <li>Proceed: if a Subchannel is provided via {@link #withSubchannel withSubchannel()}, and is
* in READY state when the RPC tries to start on it, the RPC will proceed on that
* Subchannel.</li>
* <li>Error: if an error is provided via {@link #withError withError()}, and the RPC is not
* wait-for-ready (i.e., {@link CallOptions#withWaitForReady} was not called), the RPC will
* fail immediately with the given error.</li>
* <li>Buffer: in all other cases, the RPC will be buffered in the Channel, until the next
* picker is provided via {@link Helper#updatePicker Helper.updatePicker()}, when the RPC
* will go through the same picking process again.</li>
* </ul>
*/
@Immutable
public static final class PickResult {
private static final PickResult NO_RESULT = new PickResult(null, Status.OK);
@Nullable private final Subchannel subchannel;
// An error to be propagated to the application if subchannel == null
// Or OK if there is no error.
// subchannel being null and error being OK means RPC needs to wait
private final Status status;
private PickResult(Subchannel subchannel, Status status) {
this.subchannel = subchannel;
this.status = Preconditions.checkNotNull(status, "status");
}
/**
* A decision to proceed the RPC on a Subchannel.
*
* <p>Only Subchannels returned by {@link Helper#createSubchannel Helper.createSubchannel()}
* will work. DO NOT try to use your own implementations of Subchannels, as they won't work.
*
* <p>When the RPC tries to use the return Subchannel, which is briefly after this method
* returns, the state of the Subchannel will decide where the RPC would go:
*
* <ul>
* <li>READY: the RPC will proceed on this Subchannel.</li>
* <li>IDLE: the RPC will be buffered. Subchannel will attempt to create connection.</li>
* <li>All other states: the RPC will be buffered.</li>
* </ul>
*
* <p><strong>All buffered RPCs will stay buffered</strong> until the next call of {@link
* Helper#updatePicker Helper.updatePicker()}, which will trigger a new picking process.
*
* <p>Note that Subchannel's state may change at the same time the picker is making the
* decision, which means the decision may be made with (to-be) outdated information. For
* example, a picker may return a Subchannel known to be READY, but it has become IDLE when is
* about to be used by the RPC, which makes the RPC to be buffered. The LoadBalancer will soon
* learn about the Subchannels' transition from READY to IDLE, create a new picker and allow the
* RPC to use another READY transport if there is any.
*
* <p>You will want to avoid running into a situation where there are READY Subchannels out
* there but some RPCs are still buffered for longer than a brief time.
* <ul>
* <li>This can happen if you return Subchannels with states other than READY and IDLE. For
* example, suppose you round-robin on 2 Subchannels, in READY and CONNECTING states
* respectively. If the picker ignores the state and pick them equally, 50% of RPCs will
* be stuck in buffered state until both Subchannels are READY.</li>
* <li>This can also happen if you don't create a new picker at key state changes of
* Subchannels. Take the above round-robin example again. Suppose you do pick only READY
* and IDLE Subchannels, and initially both Subchannels are READY. Now one becomes IDLE,
* then CONNECTING and stays CONNECTING for a long time. If you don't create a new picker
* in response to the CONNECTING state to exclude that Subchannel, 50% of RPCs will hit it
* and be buffered even though the other Subchannel is READY.</li>
* </ul>
*
* <p>In order to prevent unnecessary delay of RPCs, the rules of thumb are:
* <ol>
* <li>The picker should only pick Subchannels that are known as READY or IDLE. Whether to
* pick IDLE Subchannels depends on whether you want Subchannels to connect on-demand or
* actively:
* <ul>
* <li>If you want connect-on-demand, include IDLE Subchannels in your pick results,
* because when an RPC tries to use an IDLE Subchannel, the Subchannel will try to
* connect.</li>
* <li>If you want Subchannels to be always connected even when there is no RPC, you
* would call {@link Subchannel#requestConnection Subchannel.requestConnection()}
* whenever the Subchannel has transitioned to IDLE, then you don't need to include
* IDLE Subchannels in your pick results.</li>
* </ul></li>
* <li>Always create a new picker and call {@link Helper#updatePicker Helper.updatePicker()}
* whenever {@link #handleSubchannelState handleSubchannelState()} is called, unless the
* new state is SHUTDOWN. See {@code handleSubchannelState}'s javadoc for more
* details.</li>
* </ol>
*/
public static PickResult withSubchannel(Subchannel subchannel) {
return new PickResult(Preconditions.checkNotNull(subchannel, "subchannel"), Status.OK);
}
/**
* A decision to report a connectivity error to the RPC. If the RPC is {@link
* CallOptions#withWaitForReady wait-for-ready}, it will stay buffered. Otherwise, it will fail
* with the given error.
*
* @param error the error status. Must not be OK.
*/
public static PickResult withError(Status error) {
Preconditions.checkArgument(!error.isOk(), "error status shouldn't be OK");
return new PickResult(null, error);
}
/**
* No decision could be made. The RPC will stay buffered.
*/
public static PickResult withNoResult() {
return NO_RESULT;
}
/**
* The Subchannel if this result was created by {@link #withSubchannel withSubchannel()}, or
* null otherwise.
*/
@Nullable
public Subchannel getSubchannel() {
return subchannel;
}
/**
* The status associated with this result. Non-{@code OK} if created with {@link #withError
* withError}, or {@code OK} otherwise.
*/
public Status getStatus() {
return status;
}
@Override
public String toString() {
return "[subchannel=" + subchannel + " status=" + status + "]";
}
}
/**
* Provides essentials for LoadBalancer implementations.
*/
@ThreadSafe
public abstract static class Helper {
/**
* Creates a Subchannel, which is a logical connection to the given group of addresses which are
* considered equivalent. The {@code attrs} are custom attributes associated with this
* Subchannel, and can be accessed later through {@link Subchannel#getAttributes
* Subchannel.getAttributes()}.
*
* <p>The LoadBalancer is responsible for closing unused Subchannels, and closing all
* Subchannels within {@link #shutdown}.
*/
public abstract Subchannel createSubchannel(EquivalentAddressGroup addrs, Attributes attrs);
/**
* Out-of-band channel for LoadBalancer’s own RPC needs, e.g., talking to an external
* load-balancer service.
*
* <p>The LoadBalancer is responsible for closing unused OOB channels, and closing all OOB
* channels within {@link #shutdown}.
*/
public abstract ManagedChannel createOobChannel(
EquivalentAddressGroup eag, String authority);
/**
* Set a new picker to the channel.
*
* <p>When a new picker is provided via {@code updatePicker()}, the channel will apply the
* picker on all buffered RPCs, by calling {@link SubchannelPicker#pickSubchannel
* SubchannelPicker.pickSubchannel()}.
*
* <p>The channel will hold the picker and use it for all RPCs, until {@code updatePicker()} is
* called again and a new picker replaces the old one. If {@code updatePicker()} has never been
* called, the channel will buffer all RPCs until a picker is provided.
*/
public abstract void updatePicker(SubchannelPicker picker);
/**
* Schedule a task to be run in the Channel Executor, which serializes the task with the
* callback methods on the {@link LoadBalancer2} interface.
*/
public abstract void runSerialized(Runnable task);
/**
* Returns the NameResolver of the channel.
*/
public abstract NameResolver.Factory getNameResolverFactory();
/**
* Returns the authority string of the channel, which is derived from the DNS-style target name.
*/
public abstract String getAuthority();
}
/**
* A logical connection to a server, or a group of equivalent servers represented by an {@link
* EquivalentAddressGroup}.
*
* <p>It maintains at most one physical connection (aka transport) for sending new RPCs, while
* also keeps track of previous transports that has been shut down but not terminated yet.
*
* <p>If there isn't an active transport yet, and an RPC is assigned to the Subchannel, it will
* create a new transport. It won't actively create transports otherwise. {@link
* #requestConnection requestConnection()} can be used to ask Subchannel to create a transport if
* there isn't any.
*/
@ThreadSafe
public abstract static class Subchannel {
/**
* Shuts down the Subchannel. After this method is called, this Subchannel should no longer
* be returned by the latest {@link SubchannelPicker picker}, and can be safely discarded.
*/
public abstract void shutdown();
/**
* Asks the Subchannel to create a connection (aka transport), if there isn't an active one.
*/
public abstract void requestConnection();
/**
* Returns the addresses that this Subchannel is bound to.
*/
public abstract EquivalentAddressGroup getAddresses();
/**
* The same attributes passed to {@link Helper#createSubchannel Helper.createSubchannel()}.
* LoadBalancer can use it to attach additional information here, e.g., the shard this
* Subchannel belongs to.
*/
public abstract Attributes getAttributes();
}
@ThreadSafe
public abstract static class Factory {
/**
* Creates a {@link LoadBalancer2} that will be used inside a channel.
*/
public abstract LoadBalancer2 newLoadBalancer(Helper helper);
}
} }

8
core/src/main/java/io/grpc/ManagedChannelBuilder.java
View File

@ -155,9 +155,9 @@ public abstract class ManagedChannelBuilder<T extends ManagedChannelBuilder<T>>
public abstract T nameResolverFactory(NameResolver.Factory resolverFactory); public abstract T nameResolverFactory(NameResolver.Factory resolverFactory);
/** /**
* Provides a custom {@link LoadBalancer2.Factory} for the channel. * Provides a custom {@link LoadBalancer.Factory} for the channel.
* *
* <p>If this method is not called, the builder will use {@link PickFirstBalancerFactory2} * <p>If this method is not called, the builder will use {@link PickFirstBalancerFactory}
* for the channel. * for the channel.
* *
* <p>Calling this will make the channel to run the LBv2 code path. See <a * <p>Calling this will make the channel to run the LBv2 code path. See <a
@ -169,9 +169,7 @@ public abstract class ManagedChannelBuilder<T extends ManagedChannelBuilder<T>>
* this method will throw. * this method will throw.
*/ */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771") @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771")
public T loadBalancerFactory(LoadBalancer2.Factory loadBalancerFactory) { public abstract T loadBalancerFactory(LoadBalancer.Factory loadBalancerFactory);
throw new UnsupportedOperationException("Not implemented by " + this.getClass().getName());
}
/** /**
* Set the decompression registry for use in the channel. This is an advanced API call and * Set the decompression registry for use in the channel. This is an advanced API call and

42
core/src/main/java/io/grpc/PickFirstBalancerFactory2.java → core/src/main/java/io/grpc/PickFirstBalancerFactory.java
View File

@ -1,5 +1,5 @@
/* /*
* Copyright 2016, Google Inc. All rights reserved. * Copyright 2015, Google Inc. All rights reserved.
* *
* Redistribution and use in source and binary forms, with or without * Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are * modification, are permitted provided that the following conditions are
@ -31,6 +31,7 @@
package io.grpc; package io.grpc;
import static com.google.common.base.Preconditions.checkNotNull;
import static io.grpc.ConnectivityState.SHUTDOWN; import static io.grpc.ConnectivityState.SHUTDOWN;
import com.google.common.annotations.VisibleForTesting; import com.google.common.annotations.VisibleForTesting;
@ -39,44 +40,39 @@ import java.util.ArrayList;
import java.util.List; import java.util.List;
/** /**
* A {@link LoadBalancer2} that provides no load balancing mechanism over the * A {@link LoadBalancer} that provides no load balancing mechanism over the
* addresses from the {@link NameResolver}. The channel's default behavior * addresses from the {@link NameResolver}. The channel's default behavior
* (currently pick-first) is used for all addresses found. * (currently pick-first) is used for all addresses found.
*
* <p><strong>TECHNICAL PREVIEW: </strong>The name of this class is temporary. It will be renamed to
* {@code PickFirstBalancerFactory} during <a href="https://github.com/grpc/grpc-java/issues/2656"
* target="_blank">the transition to LBv2</a>. You should use it only if you want to experiment the
* LBv2 code path.
*/ */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771") @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771")
public final class PickFirstBalancerFactory2 extends LoadBalancer2.Factory { public final class PickFirstBalancerFactory extends LoadBalancer.Factory {
private static final PickFirstBalancerFactory2 INSTANCE = new PickFirstBalancerFactory2(); private static final PickFirstBalancerFactory INSTANCE = new PickFirstBalancerFactory();
private PickFirstBalancerFactory2() { private PickFirstBalancerFactory() {
} }
public static PickFirstBalancerFactory2 getInstance() { public static PickFirstBalancerFactory getInstance() {
return INSTANCE; return INSTANCE;
} }
@Override @Override
public LoadBalancer2 newLoadBalancer(LoadBalancer2.Helper helper) { public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
return new PickFirstBalancer(helper); return new PickFirstBalancer(helper);
} }
@VisibleForTesting @VisibleForTesting
static class PickFirstBalancer extends LoadBalancer2 { static final class PickFirstBalancer extends LoadBalancer {
private final Helper helper; private final Helper helper;
private Subchannel subchannel; private Subchannel subchannel;
public PickFirstBalancer(Helper helper) { PickFirstBalancer(Helper helper) {
this.helper = helper; this.helper = checkNotNull(helper, "helper");
} }
@Override @Override
public void handleResolvedAddresses(List<ResolvedServerInfoGroup> servers, public void handleResolvedAddresses(
Attributes attributes) { List<ResolvedServerInfoGroup> servers, Attributes attributes) {
// Flatten servers list received from name resolver into single address group. This means that // Flatten servers list received from name resolver into single address group. This means that
// as far as load balancer is concerned, there's virtually one single server with multiple // as far as load balancer is concerned, there's virtually one single server with multiple
// addresses so the connection will be created only for the first address (pick first). // addresses so the connection will be created only for the first address (pick first).
@ -123,7 +119,7 @@ public final class PickFirstBalancerFactory2 extends LoadBalancer2.Factory {
pickResult = PickResult.withError(stateInfo.getStatus()); pickResult = PickResult.withError(stateInfo.getStatus());
break; break;
default: default:
throw new IllegalStateException(); throw new IllegalArgumentException("Unsupported state:" + currentState);
} }
helper.updatePicker(new Picker(pickResult)); helper.updatePicker(new Picker(pickResult));
@ -157,15 +153,15 @@ public final class PickFirstBalancerFactory2 extends LoadBalancer2.Factory {
* received in constructor. * received in constructor.
*/ */
@VisibleForTesting @VisibleForTesting
static class Picker extends LoadBalancer2.SubchannelPicker { static final class Picker extends LoadBalancer.SubchannelPicker {
private final LoadBalancer2.PickResult result; private final LoadBalancer.PickResult result;
Picker(LoadBalancer2.PickResult result) { Picker(LoadBalancer.PickResult result) {
this.result = result; this.result = checkNotNull(result, "result");
} }
@Override @Override
public LoadBalancer2.PickResult pickSubchannel(Attributes affinity, Metadata headers) { public LoadBalancer.PickResult pickSubchannel(Attributes affinity, Metadata headers) {
return result; return result;
} }
} }

24
core/src/main/java/io/grpc/internal/AbstractManagedChannelImplBuilder.java
View File

@ -43,12 +43,12 @@ import io.grpc.Attributes;
import io.grpc.ClientInterceptor; import io.grpc.ClientInterceptor;
import io.grpc.CompressorRegistry; import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry; import io.grpc.DecompressorRegistry;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.ManagedChannel; import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder; import io.grpc.ManagedChannelBuilder;
import io.grpc.NameResolver; import io.grpc.NameResolver;
import io.grpc.NameResolverProvider; import io.grpc.NameResolverProvider;
import io.grpc.PickFirstBalancerFactory2; import io.grpc.PickFirstBalancerFactory;
import io.grpc.ResolvedServerInfo; import io.grpc.ResolvedServerInfo;
import io.grpc.ResolvedServerInfoGroup; import io.grpc.ResolvedServerInfoGroup;
import java.net.SocketAddress; import java.net.SocketAddress;
@ -108,7 +108,7 @@ public abstract class AbstractManagedChannelImplBuilder
@Nullable @Nullable
private NameResolver.Factory nameResolverFactory; private NameResolver.Factory nameResolverFactory;
private LoadBalancer2.Factory loadBalancerFactory; private LoadBalancer.Factory loadBalancerFactory;
@Nullable @Nullable
private DecompressorRegistry decompressorRegistry; private DecompressorRegistry decompressorRegistry;
@ -193,9 +193,9 @@ public abstract class AbstractManagedChannelImplBuilder
} }
@Override @Override
public final T loadBalancerFactory(LoadBalancer2.Factory loadBalancerFactory) { public final T loadBalancerFactory(LoadBalancer.Factory loadBalancerFactory) {
Preconditions.checkState(directServerAddress == null, Preconditions.checkState(directServerAddress == null,
"directServerAddress is set (%s), which forbids the use of LoadBalancerFactory", "directServerAddress is set (%s), which forbids the use of LoadBalancer.Factory",
directServerAddress); directServerAddress);
this.loadBalancerFactory = loadBalancerFactory; this.loadBalancerFactory = loadBalancerFactory;
return thisT(); return thisT();
@ -231,7 +231,7 @@ public abstract class AbstractManagedChannelImplBuilder
// We convert to the largest unit to avoid overflow // We convert to the largest unit to avoid overflow
if (unit.toDays(value) >= IDLE_MODE_MAX_TIMEOUT_DAYS) { if (unit.toDays(value) >= IDLE_MODE_MAX_TIMEOUT_DAYS) {
// This disables idle mode // This disables idle mode
this.idleTimeoutMillis = ManagedChannelImpl2.IDLE_TIMEOUT_MILLIS_DISABLE; this.idleTimeoutMillis = ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE;
} else { } else {
this.idleTimeoutMillis = Math.max(unit.toMillis(value), IDLE_MODE_MIN_TIMEOUT_MILLIS); this.idleTimeoutMillis = Math.max(unit.toMillis(value), IDLE_MODE_MIN_TIMEOUT_MILLIS);
} }
@ -275,21 +275,25 @@ public abstract class AbstractManagedChannelImplBuilder
// getResource(), then this shouldn't be a problem unless called on the UI thread. // getResource(), then this shouldn't be a problem unless called on the UI thread.
nameResolverFactory = NameResolverProvider.asFactory(); nameResolverFactory = NameResolverProvider.asFactory();
} }
return new ManagedChannelImpl2( return new ManagedChannelImpl(
target, target,
// TODO(carl-mastrangelo): Allow clients to pass this in // TODO(carl-mastrangelo): Allow clients to pass this in
new ExponentialBackoffPolicy.Provider(), new ExponentialBackoffPolicy.Provider(),
nameResolverFactory, nameResolverFactory,
getNameResolverParams(), getNameResolverParams(),
firstNonNull(loadBalancerFactory, PickFirstBalancerFactory2.getInstance()), firstNonNull(loadBalancerFactory, PickFirstBalancerFactory.getInstance()),
transportFactory, transportFactory,
firstNonNull(decompressorRegistry, DecompressorRegistry.getDefaultInstance()), firstNonNull(decompressorRegistry, DecompressorRegistry.getDefaultInstance()),
firstNonNull(compressorRegistry, CompressorRegistry.getDefaultInstance()), firstNonNull(compressorRegistry, CompressorRegistry.getDefaultInstance()),
SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE), SharedResourcePool.forResource(GrpcUtil.TIMER_SERVICE),
getExecutorPool(executor), getExecutorPool(executor),
SharedResourcePool.forResource(GrpcUtil.SHARED_CHANNEL_EXECUTOR), SharedResourcePool.forResource(GrpcUtil.SHARED_CHANNEL_EXECUTOR),
GrpcUtil.STOPWATCH_SUPPLIER, idleTimeoutMillis, GrpcUtil.STOPWATCH_SUPPLIER,
userAgent, interceptors, firstNonNull(statsFactory, idleTimeoutMillis,
userAgent,
interceptors,
firstNonNull(
statsFactory,
firstNonNull(Stats.getStatsContextFactory(), NoopStatsContextFactory.INSTANCE))); firstNonNull(Stats.getStatsContextFactory(), NoopStatsContextFactory.INSTANCE)));
} }

2
core/src/main/java/io/grpc/internal/ChannelExecutor.java
View File

@ -42,7 +42,7 @@ import javax.annotation.concurrent.ThreadSafe;
/** /**
* The thread-less Channel Executor used to run the state mutation logic in {@link * The thread-less Channel Executor used to run the state mutation logic in {@link
* ManagedChannelImpl}, {@link InternalSubchannel} and {@link io.grpc.LoadBalancer2}s. * ManagedChannelImpl}, {@link InternalSubchannel} and {@link io.grpc.LoadBalancer}s.
* *
* <p>Tasks are queued until {@link #drain} is called. Tasks are guaranteed to be run in the same * <p>Tasks are queued until {@link #drain} is called. Tasks are guaranteed to be run in the same
* order as they are submitted. * order as they are submitted.

23
core/src/main/java/io/grpc/internal/DelayedClientTransport2.java → core/src/main/java/io/grpc/internal/DelayedClientTransport.java
View File

@ -34,8 +34,8 @@ package io.grpc.internal;
import com.google.common.annotations.VisibleForTesting; import com.google.common.annotations.VisibleForTesting;
import io.grpc.CallOptions; import io.grpc.CallOptions;
import io.grpc.Context; import io.grpc.Context;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor;
import io.grpc.Status; import io.grpc.Status;
@ -55,7 +55,7 @@ import javax.annotation.concurrent.GuardedBy;
* for that stream, at which point the ownership of the stream is transferred to the real transport, * for that stream, at which point the ownership of the stream is transferred to the real transport,
* thus the delayed transport stops owning the stream. * thus the delayed transport stops owning the stream.
*/ */
final class DelayedClientTransport2 implements ManagedClientTransport { final class DelayedClientTransport implements ManagedClientTransport {
private final LogId lodId = LogId.allocate(getClass().getName()); private final LogId lodId = LogId.allocate(getClass().getName());
private final Object lock = new Object(); private final Object lock = new Object();
@ -96,7 +96,7 @@ final class DelayedClientTransport2 implements ManagedClientTransport {
* @param channelExecutor all listener callbacks of the delayed transport will be run from this * @param channelExecutor all listener callbacks of the delayed transport will be run from this
* ChannelExecutor. * ChannelExecutor.
*/ */
DelayedClientTransport2(Executor defaultAppExecutor, ChannelExecutor channelExecutor) { DelayedClientTransport(Executor defaultAppExecutor, ChannelExecutor channelExecutor) {
this.defaultAppExecutor = defaultAppExecutor; this.defaultAppExecutor = defaultAppExecutor;
this.channelExecutor = channelExecutor; this.channelExecutor = channelExecutor;
} }
@ -188,13 +188,16 @@ final class DelayedClientTransport2 implements ManagedClientTransport {
return newStream(method, headers, CallOptions.DEFAULT, StatsTraceContext.NOOP); return newStream(method, headers, CallOptions.DEFAULT, StatsTraceContext.NOOP);
} }
// Caller must call channelExecutor.drain() outside of lock because this method may schedule /**
// tasks on channelExecutor * Caller must call {@code channelExecutor.drain()} outside of lock because this method may
* schedule tasks on channelExecutor.
*/
@GuardedBy("lock") @GuardedBy("lock")
private PendingStream createPendingStream(MethodDescriptor<?, ?> method, Metadata headers, private PendingStream createPendingStream(
CallOptions callOptions, StatsTraceContext statsTraceCtx) { MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions,
PendingStream pendingStream = new PendingStream(method, headers, callOptions, StatsTraceContext statsTraceCtx) {
statsTraceCtx); PendingStream pendingStream =
new PendingStream(method, headers, callOptions, statsTraceCtx);
pendingStreams.add(pendingStream); pendingStreams.add(pendingStream);
if (pendingStreams.size() == 1) { if (pendingStreams.size() == 1) {
channelExecutor.executeLater(reportTransportInUse); channelExecutor.executeLater(reportTransportInUse);

4
core/src/main/java/io/grpc/internal/GrpcUtil.java
View File

@ -40,8 +40,8 @@ import com.google.common.base.Stopwatch;
import com.google.common.base.Supplier; import com.google.common.base.Supplier;
import com.google.common.util.concurrent.MoreExecutors; import com.google.common.util.concurrent.MoreExecutors;
import com.google.common.util.concurrent.ThreadFactoryBuilder; import com.google.common.util.concurrent.ThreadFactoryBuilder;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.Subchannel; import io.grpc.LoadBalancer.Subchannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.Status; import io.grpc.Status;
import io.grpc.internal.SharedResourceHolder.Resource; import io.grpc.internal.SharedResourceHolder.Resource;

2
core/src/main/java/io/grpc/internal/InUseStateAggregator2.java → core/src/main/java/io/grpc/internal/InUseStateAggregator.java
View File

@ -38,7 +38,7 @@ import javax.annotation.concurrent.NotThreadSafe;
* Aggregates the in-use state of a set of objects. * Aggregates the in-use state of a set of objects.
*/ */
@NotThreadSafe @NotThreadSafe
abstract class InUseStateAggregator2<T> { abstract class InUseStateAggregator<T> {
private final HashSet<T> inUseObjects = new HashSet<T>(); private final HashSet<T> inUseObjects = new HashSet<T>();

4
core/src/main/java/io/grpc/internal/InternalSubchannel.java
View File

@ -118,8 +118,8 @@ final class InternalSubchannel implements WithLogId {
new ArrayList<ConnectionClientTransport>(); new ArrayList<ConnectionClientTransport>();
// Must only be used from channelExecutor // Must only be used from channelExecutor
private final InUseStateAggregator2<ConnectionClientTransport> inUseStateAggregator = private final InUseStateAggregator<ConnectionClientTransport> inUseStateAggregator =
new InUseStateAggregator2<ConnectionClientTransport>() { new InUseStateAggregator<ConnectionClientTransport>() {
@Override @Override
void handleInUse() { void handleInUse() {
callback.onInUse(InternalSubchannel.this); callback.onInUse(InternalSubchannel.this);

42
core/src/main/java/io/grpc/internal/ManagedChannelImpl2.java → core/src/main/java/io/grpc/internal/ManagedChannelImpl.java
View File

@ -51,9 +51,9 @@ import io.grpc.CompressorRegistry;
import io.grpc.ConnectivityStateInfo; import io.grpc.ConnectivityStateInfo;
import io.grpc.DecompressorRegistry; import io.grpc.DecompressorRegistry;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.ManagedChannel; import io.grpc.ManagedChannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor;
@ -81,8 +81,8 @@ import javax.annotation.concurrent.ThreadSafe;
/** A communication channel for making outgoing RPCs. */ /** A communication channel for making outgoing RPCs. */
@ThreadSafe @ThreadSafe
public final class ManagedChannelImpl2 extends ManagedChannel implements WithLogId { public final class ManagedChannelImpl extends ManagedChannel implements WithLogId {
private static final Logger log = Logger.getLogger(ManagedChannelImpl2.class.getName()); private static final Logger log = Logger.getLogger(ManagedChannelImpl.class.getName());
// Matching this pattern means the target string is a URI target or at least intended to be one. // Matching this pattern means the target string is a URI target or at least intended to be one.
// A URI target must be an absolute hierarchical URI. // A URI target must be an absolute hierarchical URI.
@ -102,7 +102,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
private final String target; private final String target;
private final NameResolver.Factory nameResolverFactory; private final NameResolver.Factory nameResolverFactory;
private final Attributes nameResolverParams; private final Attributes nameResolverParams;
private final LoadBalancer2.Factory loadBalancerFactory; private final LoadBalancer.Factory loadBalancerFactory;
private final ClientTransportFactory transportFactory; private final ClientTransportFactory transportFactory;
private final Executor executor; private final Executor executor;
private final ObjectPool<? extends Executor> executorPool; private final ObjectPool<? extends Executor> executorPool;
@ -140,7 +140,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
// null when channel is in idle mode. Must be assigned from channelExecutor. // null when channel is in idle mode. Must be assigned from channelExecutor.
@Nullable @Nullable
private LoadBalancer2 loadBalancer; private LoadBalancer loadBalancer;
// Must be assigned from channelExecutor. null if channel is in idle mode. // Must be assigned from channelExecutor. null if channel is in idle mode.
@Nullable @Nullable
@ -155,7 +155,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
private final Set<InternalSubchannel> oobChannels = new HashSet<InternalSubchannel>(1, .75f); private final Set<InternalSubchannel> oobChannels = new HashSet<InternalSubchannel>(1, .75f);
// reprocess() must be run from channelExecutor // reprocess() must be run from channelExecutor
private final DelayedClientTransport2 delayedTransport; private final DelayedClientTransport delayedTransport;
// Shutdown states. // Shutdown states.
// //
@ -234,8 +234,8 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
// Must be accessed from channelExecutor // Must be accessed from channelExecutor
@VisibleForTesting @VisibleForTesting
final InUseStateAggregator2<Object> inUseStateAggregator = final InUseStateAggregator<Object> inUseStateAggregator =
new InUseStateAggregator2<Object>() { new InUseStateAggregator<Object>() {
@Override @Override
void handleInUse() { void handleInUse() {
exitIdleMode(); exitIdleMode();
@ -381,9 +381,9 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
} }
}; };
ManagedChannelImpl2(String target, BackoffPolicy.Provider backoffPolicyProvider, ManagedChannelImpl(String target, BackoffPolicy.Provider backoffPolicyProvider,
NameResolver.Factory nameResolverFactory, Attributes nameResolverParams, NameResolver.Factory nameResolverFactory, Attributes nameResolverParams,
LoadBalancer2.Factory loadBalancerFactory, ClientTransportFactory transportFactory, LoadBalancer.Factory loadBalancerFactory, ClientTransportFactory transportFactory,
DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry, DecompressorRegistry decompressorRegistry, CompressorRegistry compressorRegistry,
ObjectPool<ScheduledExecutorService> timerServicePool, ObjectPool<ScheduledExecutorService> timerServicePool,
ObjectPool<? extends Executor> executorPool, ObjectPool<? extends Executor> oobExecutorPool, ObjectPool<? extends Executor> executorPool, ObjectPool<? extends Executor> oobExecutorPool,
@ -398,7 +398,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
this.executorPool = checkNotNull(executorPool, "executorPool"); this.executorPool = checkNotNull(executorPool, "executorPool");
this.oobExecutorPool = checkNotNull(oobExecutorPool, "oobExecutorPool"); this.oobExecutorPool = checkNotNull(oobExecutorPool, "oobExecutorPool");
this.executor = checkNotNull(executorPool.getObject(), "executor"); this.executor = checkNotNull(executorPool.getObject(), "executor");
this.delayedTransport = new DelayedClientTransport2(this.executor, this.channelExecutor); this.delayedTransport = new DelayedClientTransport(this.executor, this.channelExecutor);
this.delayedTransport.start(delayedTransportListener); this.delayedTransport.start(delayedTransportListener);
this.backoffPolicyProvider = backoffPolicyProvider; this.backoffPolicyProvider = backoffPolicyProvider;
this.transportFactory = this.transportFactory =
@ -473,7 +473,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
* cancelled. * cancelled.
*/ */
@Override @Override
public ManagedChannelImpl2 shutdown() { public ManagedChannelImpl shutdown() {
log.log(Level.FINE, "[{0}] shutdown() called", getLogId()); log.log(Level.FINE, "[{0}] shutdown() called", getLogId());
if (!shutdown.compareAndSet(false, true)) { if (!shutdown.compareAndSet(false, true)) {
return this; return this;
@ -495,7 +495,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
* return {@code false} immediately after this method returns. * return {@code false} immediately after this method returns.
*/ */
@Override @Override
public ManagedChannelImpl2 shutdownNow() { public ManagedChannelImpl shutdownNow() {
log.log(Level.FINE, "[{0}] shutdownNow() called", getLogId()); log.log(Level.FINE, "[{0}] shutdownNow() called", getLogId());
shutdown(); shutdown();
delayedTransport.shutdownNow(SHUTDOWN_NOW_STATUS); delayedTransport.shutdownNow(SHUTDOWN_NOW_STATUS);
@ -547,7 +547,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
CallOptions callOptions) { CallOptions callOptions) {
Executor executor = callOptions.getExecutor(); Executor executor = callOptions.getExecutor();
if (executor == null) { if (executor == null) {
executor = ManagedChannelImpl2.this.executor; executor = ManagedChannelImpl.this.executor;
} }
StatsTraceContext statsTraceCtx = StatsTraceContext.newClientContext( StatsTraceContext statsTraceCtx = StatsTraceContext.newClientContext(
method.getFullMethodName(), statsFactory, stopwatchSupplier); method.getFullMethodName(), statsFactory, stopwatchSupplier);
@ -588,8 +588,8 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
} }
} }
private class LbHelperImpl extends LoadBalancer2.Helper { private class LbHelperImpl extends LoadBalancer.Helper {
LoadBalancer2 lb; LoadBalancer lb;
final NameResolver nr; final NameResolver nr;
LbHelperImpl(NameResolver nr) { LbHelperImpl(NameResolver nr) {
@ -702,7 +702,7 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
@Override @Override
public String getAuthority() { public String getAuthority() {
return ManagedChannelImpl2.this.authority(); return ManagedChannelImpl.this.authority();
} }
@Override @Override
@ -733,8 +733,8 @@ public final class ManagedChannelImpl2 extends ManagedChannel implements WithLog
} }
private class NameResolverListenerImpl implements NameResolver.Listener { private class NameResolverListenerImpl implements NameResolver.Listener {
final LoadBalancer2 balancer; final LoadBalancer balancer;
final LoadBalancer2.Helper helper; final LoadBalancer.Helper helper;
NameResolverListenerImpl(LbHelperImpl helperImpl) { NameResolverListenerImpl(LbHelperImpl helperImpl) {
this.balancer = helperImpl.lb; this.balancer = helperImpl.lb;

12
core/src/main/java/io/grpc/internal/OobChannel.java
View File

@ -41,9 +41,9 @@ import io.grpc.CallOptions;
import io.grpc.ClientCall; import io.grpc.ClientCall;
import io.grpc.ConnectivityStateInfo; import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.ManagedChannel; import io.grpc.ManagedChannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor;
@ -58,7 +58,7 @@ import java.util.logging.Logger;
import javax.annotation.concurrent.ThreadSafe; import javax.annotation.concurrent.ThreadSafe;
/** /**
* A ManagedChannel backed by a single {@link InternalSubchannel} and used for {@link LoadBalancer2} * A ManagedChannel backed by a single {@link InternalSubchannel} and used for {@link LoadBalancer}
* to its own RPC needs. * to its own RPC needs.
*/ */
@ThreadSafe @ThreadSafe
@ -71,7 +71,7 @@ final class OobChannel extends ManagedChannel implements WithLogId {
private final LogId logId = LogId.allocate(getClass().getName()); private final LogId logId = LogId.allocate(getClass().getName());
private final StatsContextFactory statsFactory; private final StatsContextFactory statsFactory;
private final String authority; private final String authority;
private final DelayedClientTransport2 delayedTransport; private final DelayedClientTransport delayedTransport;
private final ObjectPool<? extends Executor> executorPool; private final ObjectPool<? extends Executor> executorPool;
private final Executor executor; private final Executor executor;
private final ScheduledExecutorService deadlineCancellationExecutor; private final ScheduledExecutorService deadlineCancellationExecutor;
@ -100,7 +100,7 @@ final class OobChannel extends ManagedChannel implements WithLogId {
this.deadlineCancellationExecutor = checkNotNull( this.deadlineCancellationExecutor = checkNotNull(
deadlineCancellationExecutor, "deadlineCancellationExecutor"); deadlineCancellationExecutor, "deadlineCancellationExecutor");
this.stopwatchSupplier = checkNotNull(stopwatchSupplier, "stopwatchSupplier"); this.stopwatchSupplier = checkNotNull(stopwatchSupplier, "stopwatchSupplier");
this.delayedTransport = new DelayedClientTransport2(executor, channelExecutor); this.delayedTransport = new DelayedClientTransport(executor, channelExecutor);
this.delayedTransport.start(new ManagedClientTransport.Listener() { this.delayedTransport.start(new ManagedClientTransport.Listener() {
@Override @Override
public void transportShutdown(Status s) { public void transportShutdown(Status s) {

6
core/src/main/java/io/grpc/internal/SubchannelImpl.java
View File

@ -31,14 +31,14 @@
package io.grpc.internal; package io.grpc.internal;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import javax.annotation.Nullable; import javax.annotation.Nullable;
/** /**
* The base interface of the Subchannels returned by {@link * The base interface of the Subchannels returned by {@link
* io.grpc.LoadBalancer2.Helper#createSubchannel}. * io.grpc.LoadBalancer.Helper#createSubchannel}.
*/ */
abstract class SubchannelImpl extends LoadBalancer2.Subchannel { abstract class SubchannelImpl extends LoadBalancer.Subchannel {
/** /**
* Same as {@link InternalSubchannel#obtainActiveTransport}. * Same as {@link InternalSubchannel#obtainActiveTransport}.
*/ */

39
core/src/main/java/io/grpc/util/RoundRobinLoadBalancerFactory2.java → core/src/main/java/io/grpc/util/RoundRobinLoadBalancerFactory.java
View File

@ -41,10 +41,10 @@ import io.grpc.Attributes;
import io.grpc.ConnectivityStateInfo; import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.ExperimentalApi; import io.grpc.ExperimentalApi;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.Subchannel; import io.grpc.LoadBalancer.Subchannel;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.NameResolver; import io.grpc.NameResolver;
import io.grpc.ResolvedServerInfo; import io.grpc.ResolvedServerInfo;
@ -64,35 +64,30 @@ import javax.annotation.Nullable;
import javax.annotation.concurrent.GuardedBy; import javax.annotation.concurrent.GuardedBy;
/** /**
* A {@link LoadBalancer2} that provides round-robin load balancing mechanism over the * A {@link LoadBalancer} that provides round-robin load balancing mechanism over the
* addresses from the {@link NameResolver}. The sub-lists received from the name resolver * addresses from the {@link NameResolver}. The sub-lists received from the name resolver
* are considered to be an {@link EquivalentAddressGroup} and each of these sub-lists is * are considered to be an {@link EquivalentAddressGroup} and each of these sub-lists is
* what is then balanced across. * what is then balanced across.
*
* <p><strong>TECHNICAL PREVIEW: </strong>The name of this class is temporary. It will be renamed to
* {@code RoundRobinLoadBalancerFactory} during the
* <a href="https://github.com/grpc/grpc-java/issues/2656" target="_blank">transition to LBv2</a>.
* You should use it only if you want to experiment the LBv2 code path.
*/ */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771") @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1771")
public class RoundRobinLoadBalancerFactory2 extends LoadBalancer2.Factory { public class RoundRobinLoadBalancerFactory extends LoadBalancer.Factory {
private static final RoundRobinLoadBalancerFactory2 INSTANCE = private static final RoundRobinLoadBalancerFactory INSTANCE =
new RoundRobinLoadBalancerFactory2(); new RoundRobinLoadBalancerFactory();
private RoundRobinLoadBalancerFactory2() { private RoundRobinLoadBalancerFactory() {
} }
public static RoundRobinLoadBalancerFactory2 getInstance() { public static RoundRobinLoadBalancerFactory getInstance() {
return INSTANCE; return INSTANCE;
} }
@Override @Override
public LoadBalancer2 newLoadBalancer(LoadBalancer2.Helper helper) { public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
return new RoundRobinLoadBalancer(helper); return new RoundRobinLoadBalancer(helper);
} }
@VisibleForTesting @VisibleForTesting
static class RoundRobinLoadBalancer extends LoadBalancer2 { static class RoundRobinLoadBalancer extends LoadBalancer {
private final Helper helper; private final Helper helper;
private final Map<EquivalentAddressGroup, Subchannel> subchannels = private final Map<EquivalentAddressGroup, Subchannel> subchannels =
new HashMap<EquivalentAddressGroup, Subchannel>(); new HashMap<EquivalentAddressGroup, Subchannel>();
@ -101,13 +96,13 @@ public class RoundRobinLoadBalancerFactory2 extends LoadBalancer2.Factory {
static final Attributes.Key<AtomicReference<ConnectivityStateInfo>> STATE_INFO = static final Attributes.Key<AtomicReference<ConnectivityStateInfo>> STATE_INFO =
Attributes.Key.of("state-info"); Attributes.Key.of("state-info");
public RoundRobinLoadBalancer(Helper helper) { RoundRobinLoadBalancer(Helper helper) {
this.helper = helper; this.helper = checkNotNull(helper, "helper");
} }
@Override @Override
public void handleResolvedAddresses(List<ResolvedServerInfoGroup> servers, public void handleResolvedAddresses(
Attributes attributes) { List<ResolvedServerInfoGroup> servers, Attributes attributes) {
Set<EquivalentAddressGroup> currentAddrs = subchannels.keySet(); Set<EquivalentAddressGroup> currentAddrs = subchannels.keySet();
Set<EquivalentAddressGroup> latestAddrs = Set<EquivalentAddressGroup> latestAddrs =
resolvedServerInfoGroupToEquivalentAddressGroup(servers); resolvedServerInfoGroupToEquivalentAddressGroup(servers);
@ -237,7 +232,7 @@ public class RoundRobinLoadBalancerFactory2 extends LoadBalancer2.Factory {
} }
@VisibleForTesting @VisibleForTesting
static class Picker extends SubchannelPicker { static final class Picker extends SubchannelPicker {
@Nullable @Nullable
private final Status status; private final Status status;
private final List<Subchannel> list; private final List<Subchannel> list;

16
core/src/test/java/io/grpc/PickFirstLoadBalancer2Test.java → core/src/test/java/io/grpc/PickFirstLoadBalancerTest.java
View File

@ -44,11 +44,11 @@ import static org.mockito.Mockito.verifyNoMoreInteractions;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
import com.google.common.collect.Lists; import com.google.common.collect.Lists;
import io.grpc.LoadBalancer2.Helper; import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.Subchannel; import io.grpc.LoadBalancer.Subchannel;
import io.grpc.PickFirstBalancerFactory2.PickFirstBalancer; import io.grpc.PickFirstBalancerFactory.PickFirstBalancer;
import io.grpc.PickFirstBalancerFactory2.Picker; import io.grpc.PickFirstBalancerFactory.Picker;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.util.List; import java.util.List;
import org.junit.Before; import org.junit.Before;
@ -62,9 +62,9 @@ import org.mockito.Mock;
import org.mockito.MockitoAnnotations; import org.mockito.MockitoAnnotations;
/** Unit test for {@link PickFirstBalancerFactory2}. */ /** Unit test for {@link PickFirstBalancerFactory}. */
@RunWith(JUnit4.class) @RunWith(JUnit4.class)
public class PickFirstLoadBalancer2Test { public class PickFirstLoadBalancerTest {
private PickFirstBalancer loadBalancer; private PickFirstBalancer loadBalancer;
private List<ResolvedServerInfoGroup> servers = Lists.newArrayList(); private List<ResolvedServerInfoGroup> servers = Lists.newArrayList();
private List<SocketAddress> socketAddresses = Lists.newArrayList(); private List<SocketAddress> socketAddresses = Lists.newArrayList();
@ -96,7 +96,7 @@ public class PickFirstLoadBalancer2Test {
when(mockHelper.createSubchannel(any(EquivalentAddressGroup.class), any(Attributes.class))) when(mockHelper.createSubchannel(any(EquivalentAddressGroup.class), any(Attributes.class)))
.thenReturn(mockSubchannel); .thenReturn(mockSubchannel);
loadBalancer = (PickFirstBalancer) PickFirstBalancerFactory2.getInstance().newLoadBalancer( loadBalancer = (PickFirstBalancer) PickFirstBalancerFactory.getInstance().newLoadBalancer(
mockHelper); mockHelper);
} }

4
core/src/test/java/io/grpc/internal/AbstractManagedChannelImplBuilderTest.java
View File

@ -78,11 +78,11 @@ public class AbstractManagedChannelImplBuilderTest {
builder.getIdleTimeoutMillis()); builder.getIdleTimeoutMillis());
builder.idleTimeout(Long.MAX_VALUE, TimeUnit.DAYS); builder.idleTimeout(Long.MAX_VALUE, TimeUnit.DAYS);
assertEquals(ManagedChannelImpl2.IDLE_TIMEOUT_MILLIS_DISABLE, builder.getIdleTimeoutMillis()); assertEquals(ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE, builder.getIdleTimeoutMillis());
builder.idleTimeout(AbstractManagedChannelImplBuilder.IDLE_MODE_MAX_TIMEOUT_DAYS, builder.idleTimeout(AbstractManagedChannelImplBuilder.IDLE_MODE_MAX_TIMEOUT_DAYS,
TimeUnit.DAYS); TimeUnit.DAYS);
assertEquals(ManagedChannelImpl2.IDLE_TIMEOUT_MILLIS_DISABLE, builder.getIdleTimeoutMillis()); assertEquals(ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE, builder.getIdleTimeoutMillis());
try { try {
builder.idleTimeout(0, TimeUnit.SECONDS); builder.idleTimeout(0, TimeUnit.SECONDS);

10
core/src/test/java/io/grpc/internal/DelayedClientTransport2Test.java → core/src/test/java/io/grpc/internal/DelayedClientTransportTest.java
View File

@ -51,8 +51,8 @@ import static org.mockito.Mockito.when;
import io.grpc.Attributes; import io.grpc.Attributes;
import io.grpc.CallOptions; import io.grpc.CallOptions;
import io.grpc.IntegerMarshaller; import io.grpc.IntegerMarshaller;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor;
import io.grpc.MethodDescriptor.MethodType; import io.grpc.MethodDescriptor.MethodType;
@ -76,10 +76,10 @@ import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer; import org.mockito.stubbing.Answer;
/** /**
* Unit tests for {@link DelayedClientTransport2}. * Unit tests for {@link DelayedClientTransport}.
*/ */
@RunWith(JUnit4.class) @RunWith(JUnit4.class)
public class DelayedClientTransport2Test { public class DelayedClientTransportTest {
@Mock private ManagedClientTransport.Listener transportListener; @Mock private ManagedClientTransport.Listener transportListener;
@Mock private SubchannelPicker mockPicker; @Mock private SubchannelPicker mockPicker;
@Mock private SubchannelImpl mockSubchannel; @Mock private SubchannelImpl mockSubchannel;
@ -117,7 +117,7 @@ public class DelayedClientTransport2Test {
private final FakeClock fakeExecutor = new FakeClock(); private final FakeClock fakeExecutor = new FakeClock();
private final DelayedClientTransport2 delayedTransport = new DelayedClientTransport2( private final DelayedClientTransport delayedTransport = new DelayedClientTransport(
fakeExecutor.getScheduledExecutorService(), new ChannelExecutor()); fakeExecutor.getScheduledExecutorService(), new ChannelExecutor());
@Before public void setUp() { @Before public void setUp() {

8
core/src/test/java/io/grpc/internal/ManagedChannelImplGetNameResolverTest.java
View File

@ -43,7 +43,7 @@ import org.junit.Test;
import org.junit.runner.RunWith; import org.junit.runner.RunWith;
import org.junit.runners.JUnit4; import org.junit.runners.JUnit4;
/** Unit tests for {@link ManagedChannelImpl2#getNameResolver}. */ /** Unit tests for {@link ManagedChannelImpl#getNameResolver}. */
@RunWith(JUnit4.class) @RunWith(JUnit4.class)
public class ManagedChannelImplGetNameResolverTest { public class ManagedChannelImplGetNameResolverTest {
private static final Attributes NAME_RESOLVER_PARAMS = private static final Attributes NAME_RESOLVER_PARAMS =
@ -121,7 +121,7 @@ public class ManagedChannelImplGetNameResolverTest {
} }
}; };
try { try {
ManagedChannelImpl2.getNameResolver( ManagedChannelImpl.getNameResolver(
"foo.googleapis.com:8080", nameResolverFactory, NAME_RESOLVER_PARAMS); "foo.googleapis.com:8080", nameResolverFactory, NAME_RESOLVER_PARAMS);
fail("Should fail"); fail("Should fail");
} catch (IllegalArgumentException e) { } catch (IllegalArgumentException e) {
@ -131,7 +131,7 @@ public class ManagedChannelImplGetNameResolverTest {
private void testValidTarget(String target, String expectedUriString, URI expectedUri) { private void testValidTarget(String target, String expectedUriString, URI expectedUri) {
Factory nameResolverFactory = new FakeNameResolverFactory(expectedUri.getScheme()); Factory nameResolverFactory = new FakeNameResolverFactory(expectedUri.getScheme());
FakeNameResolver nameResolver = (FakeNameResolver) ManagedChannelImpl2.getNameResolver( FakeNameResolver nameResolver = (FakeNameResolver) ManagedChannelImpl.getNameResolver(
target, nameResolverFactory, NAME_RESOLVER_PARAMS); target, nameResolverFactory, NAME_RESOLVER_PARAMS);
assertNotNull(nameResolver); assertNotNull(nameResolver);
assertEquals(expectedUri, nameResolver.uri); assertEquals(expectedUri, nameResolver.uri);
@ -142,7 +142,7 @@ public class ManagedChannelImplGetNameResolverTest {
Factory nameResolverFactory = new FakeNameResolverFactory("dns"); Factory nameResolverFactory = new FakeNameResolverFactory("dns");
try { try {
FakeNameResolver nameResolver = (FakeNameResolver) ManagedChannelImpl2.getNameResolver( FakeNameResolver nameResolver = (FakeNameResolver) ManagedChannelImpl.getNameResolver(
target, nameResolverFactory, NAME_RESOLVER_PARAMS); target, nameResolverFactory, NAME_RESOLVER_PARAMS);
fail("Should have failed, but got resolver with " + nameResolver.uri); fail("Should have failed, but got resolver with " + nameResolver.uri);
} catch (IllegalArgumentException e) { } catch (IllegalArgumentException e) {

20
core/src/test/java/io/grpc/internal/ManagedChannelImpl2IdlenessTest.java → core/src/test/java/io/grpc/internal/ManagedChannelImplIdlenessTest.java
View File

@ -52,11 +52,11 @@ import io.grpc.CompressorRegistry;
import io.grpc.DecompressorRegistry; import io.grpc.DecompressorRegistry;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.IntegerMarshaller; import io.grpc.IntegerMarshaller;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer2.Helper; import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.Subchannel; import io.grpc.LoadBalancer.Subchannel;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.ManagedChannel; import io.grpc.ManagedChannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor;
@ -90,14 +90,14 @@ import org.mockito.MockitoAnnotations;
* Unit tests for {@link ManagedChannelImpl}'s idle mode. * Unit tests for {@link ManagedChannelImpl}'s idle mode.
*/ */
@RunWith(JUnit4.class) @RunWith(JUnit4.class)
public class ManagedChannelImpl2IdlenessTest { public class ManagedChannelImplIdlenessTest {
private final FakeClock timer = new FakeClock(); private final FakeClock timer = new FakeClock();
private final FakeClock executor = new FakeClock(); private final FakeClock executor = new FakeClock();
private final FakeClock oobExecutor = new FakeClock(); private final FakeClock oobExecutor = new FakeClock();
private static final String AUTHORITY = "fakeauthority"; private static final String AUTHORITY = "fakeauthority";
private static final String USER_AGENT = "fakeagent"; private static final String USER_AGENT = "fakeagent";
private static final long IDLE_TIMEOUT_SECONDS = 30; private static final long IDLE_TIMEOUT_SECONDS = 30;
private ManagedChannelImpl2 channel; private ManagedChannelImpl channel;
private final MethodDescriptor<String, Integer> method = private final MethodDescriptor<String, Integer> method =
MethodDescriptor.<String, Integer>newBuilder() MethodDescriptor.<String, Integer>newBuilder()
@ -115,8 +115,8 @@ public class ManagedChannelImpl2IdlenessTest {
@Mock private ObjectPool<Executor> executorPool; @Mock private ObjectPool<Executor> executorPool;
@Mock private ObjectPool<Executor> oobExecutorPool; @Mock private ObjectPool<Executor> oobExecutorPool;
@Mock private ClientTransportFactory mockTransportFactory; @Mock private ClientTransportFactory mockTransportFactory;
@Mock private LoadBalancer2 mockLoadBalancer; @Mock private LoadBalancer mockLoadBalancer;
@Mock private LoadBalancer2.Factory mockLoadBalancerFactory; @Mock private LoadBalancer.Factory mockLoadBalancerFactory;
@Mock private NameResolver mockNameResolver; @Mock private NameResolver mockNameResolver;
@Mock private NameResolver.Factory mockNameResolverFactory; @Mock private NameResolver.Factory mockNameResolverFactory;
@Mock private ClientCall.Listener<Integer> mockCallListener; @Mock private ClientCall.Listener<Integer> mockCallListener;
@ -136,7 +136,7 @@ public class ManagedChannelImpl2IdlenessTest {
.newNameResolver(any(URI.class), any(Attributes.class))) .newNameResolver(any(URI.class), any(Attributes.class)))
.thenReturn(mockNameResolver); .thenReturn(mockNameResolver);
channel = new ManagedChannelImpl2("fake://target", new FakeBackoffPolicyProvider(), channel = new ManagedChannelImpl("fake://target", new FakeBackoffPolicyProvider(),
mockNameResolverFactory, Attributes.EMPTY, mockLoadBalancerFactory, mockNameResolverFactory, Attributes.EMPTY, mockLoadBalancerFactory,
mockTransportFactory, DecompressorRegistry.getDefaultInstance(), mockTransportFactory, DecompressorRegistry.getDefaultInstance(),
CompressorRegistry.getDefaultInstance(), timerServicePool, executorPool, oobExecutorPool, CompressorRegistry.getDefaultInstance(), timerServicePool, executorPool, oobExecutorPool,

42
core/src/test/java/io/grpc/internal/ManagedChannelImpl2Test.java → core/src/test/java/io/grpc/internal/ManagedChannelImplTest.java
View File

@ -71,11 +71,11 @@ import io.grpc.Context;
import io.grpc.DecompressorRegistry; import io.grpc.DecompressorRegistry;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.IntegerMarshaller; import io.grpc.IntegerMarshaller;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer2.Helper; import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.Subchannel; import io.grpc.LoadBalancer.Subchannel;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.ManagedChannel; import io.grpc.ManagedChannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor;
@ -116,9 +116,9 @@ import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock; import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer; import org.mockito.stubbing.Answer;
/** Unit tests for {@link ManagedChannelImpl2}. */ /** Unit tests for {@link ManagedChannelImpl}. */
@RunWith(JUnit4.class) @RunWith(JUnit4.class)
public class ManagedChannelImpl2Test { public class ManagedChannelImplTest {
private static final List<ClientInterceptor> NO_INTERCEPTOR = private static final List<ClientInterceptor> NO_INTERCEPTOR =
Collections.<ClientInterceptor>emptyList(); Collections.<ClientInterceptor>emptyList();
private static final Attributes NAME_RESOLVER_PARAMS = private static final Attributes NAME_RESOLVER_PARAMS =
@ -147,16 +147,16 @@ public class ManagedChannelImpl2Test {
@Rule public final ExpectedException thrown = ExpectedException.none(); @Rule public final ExpectedException thrown = ExpectedException.none();
private ManagedChannelImpl2 channel; private ManagedChannelImpl channel;
private Helper helper; private Helper helper;
@Captor @Captor
private ArgumentCaptor<Status> statusCaptor; private ArgumentCaptor<Status> statusCaptor;
@Captor @Captor
private ArgumentCaptor<StatsTraceContext> statsTraceCtxCaptor; private ArgumentCaptor<StatsTraceContext> statsTraceCtxCaptor;
@Mock @Mock
private LoadBalancer2.Factory mockLoadBalancerFactory; private LoadBalancer.Factory mockLoadBalancerFactory;
@Mock @Mock
private LoadBalancer2 mockLoadBalancer; private LoadBalancer mockLoadBalancer;
@Captor @Captor
private ArgumentCaptor<ConnectivityStateInfo> stateInfoCaptor; private ArgumentCaptor<ConnectivityStateInfo> stateInfoCaptor;
@Mock @Mock
@ -188,11 +188,11 @@ public class ManagedChannelImpl2Test {
private void createChannel( private void createChannel(
NameResolver.Factory nameResolverFactory, List<ClientInterceptor> interceptors) { NameResolver.Factory nameResolverFactory, List<ClientInterceptor> interceptors) {
channel = new ManagedChannelImpl2(target, new FakeBackoffPolicyProvider(), channel = new ManagedChannelImpl(target, new FakeBackoffPolicyProvider(),
nameResolverFactory, NAME_RESOLVER_PARAMS, mockLoadBalancerFactory, nameResolverFactory, NAME_RESOLVER_PARAMS, mockLoadBalancerFactory,
mockTransportFactory, DecompressorRegistry.getDefaultInstance(), mockTransportFactory, DecompressorRegistry.getDefaultInstance(),
CompressorRegistry.getDefaultInstance(), timerServicePool, executorPool, oobExecutorPool, CompressorRegistry.getDefaultInstance(), timerServicePool, executorPool, oobExecutorPool,
timer.getStopwatchSupplier(), ManagedChannelImpl2.IDLE_TIMEOUT_MILLIS_DISABLE, userAgent, timer.getStopwatchSupplier(), ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE, userAgent,
interceptors, statsCtxFactory); interceptors, statsCtxFactory);
// Force-exit the initial idle-mode // Force-exit the initial idle-mode
channel.exitIdleMode(); channel.exitIdleMode();
@ -365,7 +365,7 @@ public class ManagedChannelImpl2Test {
assertTrue(nameResolverFactory.resolvers.get(0).shutdown); assertTrue(nameResolverFactory.resolvers.get(0).shutdown);
// call should have been aborted by delayed transport // call should have been aborted by delayed transport
executor.runDueTasks(); executor.runDueTasks();
verify(mockCallListener).onClose(same(ManagedChannelImpl2.SHUTDOWN_NOW_STATUS), verify(mockCallListener).onClose(same(ManagedChannelImpl.SHUTDOWN_NOW_STATUS),
any(Metadata.class)); any(Metadata.class));
} else { } else {
// LoadBalancer and NameResolver are still running. // LoadBalancer and NameResolver are still running.
@ -389,7 +389,7 @@ public class ManagedChannelImpl2Test {
if (shutdownNow) { if (shutdownNow) {
// Channel shutdownNow() all subchannels after shutting down LoadBalancer // Channel shutdownNow() all subchannels after shutting down LoadBalancer
verify(mockTransport).shutdownNow(ManagedChannelImpl2.SHUTDOWN_NOW_STATUS); verify(mockTransport).shutdownNow(ManagedChannelImpl.SHUTDOWN_NOW_STATUS);
} else { } else {
verify(mockTransport, never()).shutdownNow(any(Status.class)); verify(mockTransport, never()).shutdownNow(any(Status.class));
} }
@ -755,7 +755,7 @@ public class ManagedChannelImpl2Test {
// shutdown() has a delay // shutdown() has a delay
sub1.shutdown(); sub1.shutdown();
timer.forwardTime(ManagedChannelImpl2.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS - 1, TimeUnit.SECONDS); timer.forwardTime(ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS - 1, TimeUnit.SECONDS);
sub1.shutdown(); sub1.shutdown();
verify(transportInfo1.transport, never()).shutdown(); verify(transportInfo1.transport, never()).shutdown();
timer.forwardTime(1, TimeUnit.SECONDS); timer.forwardTime(1, TimeUnit.SECONDS);
@ -991,12 +991,12 @@ public class ManagedChannelImpl2Test {
@Test @Test
public void uriPattern() { public void uriPattern() {
assertTrue(ManagedChannelImpl2.URI_PATTERN.matcher("a:/").matches()); assertTrue(ManagedChannelImpl.URI_PATTERN.matcher("a:/").matches());
assertTrue(ManagedChannelImpl2.URI_PATTERN.matcher("Z019+-.:/!@ #~ ").matches()); assertTrue(ManagedChannelImpl.URI_PATTERN.matcher("Z019+-.:/!@ #~ ").matches());
assertFalse(ManagedChannelImpl2.URI_PATTERN.matcher("a/:").matches()); // "/:" not matched assertFalse(ManagedChannelImpl.URI_PATTERN.matcher("a/:").matches()); // "/:" not matched
assertFalse(ManagedChannelImpl2.URI_PATTERN.matcher("0a:/").matches()); // '0' not matched assertFalse(ManagedChannelImpl.URI_PATTERN.matcher("0a:/").matches()); // '0' not matched
assertFalse(ManagedChannelImpl2.URI_PATTERN.matcher("a,:/").matches()); // ',' not matched assertFalse(ManagedChannelImpl.URI_PATTERN.matcher("a,:/").matches()); // ',' not matched
assertFalse(ManagedChannelImpl2.URI_PATTERN.matcher(" a:/").matches()); // space not matched assertFalse(ManagedChannelImpl.URI_PATTERN.matcher(" a:/").matches()); // space not matched
} }
/** /**

24
core/src/test/java/io/grpc/util/RoundRobinLoadBalancer2Test.java → core/src/test/java/io/grpc/util/RoundRobinLoadBalancerTest.java
View File

@ -34,7 +34,7 @@ package io.grpc.util;
import static com.google.common.truth.Truth.assertThat; import static com.google.common.truth.Truth.assertThat;
import static io.grpc.ConnectivityState.IDLE; import static io.grpc.ConnectivityState.IDLE;
import static io.grpc.ConnectivityState.READY; import static io.grpc.ConnectivityState.READY;
import static io.grpc.util.RoundRobinLoadBalancerFactory2.RoundRobinLoadBalancer.STATE_INFO; import static io.grpc.util.RoundRobinLoadBalancerFactory.RoundRobinLoadBalancer.STATE_INFO;
import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull; import static org.junit.Assert.assertNull;
import static org.mockito.Matchers.any; import static org.mockito.Matchers.any;
@ -53,15 +53,15 @@ import io.grpc.Attributes;
import io.grpc.ConnectivityState; import io.grpc.ConnectivityState;
import io.grpc.ConnectivityStateInfo; import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer2.Helper; import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer2.Subchannel; import io.grpc.LoadBalancer.Subchannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.ResolvedServerInfo; import io.grpc.ResolvedServerInfo;
import io.grpc.ResolvedServerInfoGroup; import io.grpc.ResolvedServerInfoGroup;
import io.grpc.Status; import io.grpc.Status;
import io.grpc.util.RoundRobinLoadBalancerFactory2.Picker; import io.grpc.util.RoundRobinLoadBalancerFactory.Picker;
import io.grpc.util.RoundRobinLoadBalancerFactory2.RoundRobinLoadBalancer; import io.grpc.util.RoundRobinLoadBalancerFactory.RoundRobinLoadBalancer;
import java.net.SocketAddress; import java.net.SocketAddress;
import java.util.Collections; import java.util.Collections;
import java.util.List; import java.util.List;
@ -79,9 +79,9 @@ import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock; import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer; import org.mockito.stubbing.Answer;
/** Unit test for {@link RoundRobinLoadBalancerFactory2}. */ /** Unit test for {@link RoundRobinLoadBalancerFactory}. */
@RunWith(JUnit4.class) @RunWith(JUnit4.class)
public class RoundRobinLoadBalancer2Test { public class RoundRobinLoadBalancerTest {
private RoundRobinLoadBalancer loadBalancer; private RoundRobinLoadBalancer loadBalancer;
private Map<ResolvedServerInfoGroup, EquivalentAddressGroup> servers = Maps.newHashMap(); private Map<ResolvedServerInfoGroup, EquivalentAddressGroup> servers = Maps.newHashMap();
private Map<EquivalentAddressGroup, Subchannel> subchannels = Maps.newLinkedHashMap(); private Map<EquivalentAddressGroup, Subchannel> subchannels = Maps.newLinkedHashMap();
@ -117,7 +117,7 @@ public class RoundRobinLoadBalancer2Test {
} }
}); });
loadBalancer = (RoundRobinLoadBalancer) RoundRobinLoadBalancerFactory2.getInstance() loadBalancer = (RoundRobinLoadBalancer) RoundRobinLoadBalancerFactory.getInstance()
.newLoadBalancer(mockHelper); .newLoadBalancer(mockHelper);
} }
@ -323,7 +323,7 @@ public class RoundRobinLoadBalancer2Test {
Status error = Status.NOT_FOUND.withDescription("nameResolutionError"); Status error = Status.NOT_FOUND.withDescription("nameResolutionError");
loadBalancer.handleNameResolutionError(error); loadBalancer.handleNameResolutionError(error);
verify(mockHelper).updatePicker(pickerCaptor.capture()); verify(mockHelper).updatePicker(pickerCaptor.capture());
LoadBalancer2.PickResult pickResult = pickerCaptor.getValue().pickSubchannel(Attributes.EMPTY, LoadBalancer.PickResult pickResult = pickerCaptor.getValue().pickSubchannel(Attributes.EMPTY,
new Metadata()); new Metadata());
assertNull(pickResult.getSubchannel()); assertNull(pickResult.getSubchannel());
assertEquals(error, pickResult.getStatus()); assertEquals(error, pickResult.getStatus());
@ -341,12 +341,12 @@ public class RoundRobinLoadBalancer2Test {
any(Attributes.class)); any(Attributes.class));
verify(mockHelper, times(2)).updatePicker(pickerCaptor.capture()); verify(mockHelper, times(2)).updatePicker(pickerCaptor.capture());
LoadBalancer2.PickResult pickResult = pickerCaptor.getValue().pickSubchannel(Attributes.EMPTY, LoadBalancer.PickResult pickResult = pickerCaptor.getValue().pickSubchannel(Attributes.EMPTY,
new Metadata()); new Metadata());
assertEquals(readySubchannel, pickResult.getSubchannel()); assertEquals(readySubchannel, pickResult.getSubchannel());
assertEquals(Status.OK.getCode(), pickResult.getStatus().getCode()); assertEquals(Status.OK.getCode(), pickResult.getStatus().getCode());
LoadBalancer2.PickResult pickResult2 = pickerCaptor.getValue().pickSubchannel(Attributes.EMPTY, LoadBalancer.PickResult pickResult2 = pickerCaptor.getValue().pickSubchannel(Attributes.EMPTY,
new Metadata()); new Metadata());
assertEquals(readySubchannel, pickResult2.getSubchannel()); assertEquals(readySubchannel, pickResult2.getSubchannel());
verifyNoMoreInteractions(mockHelper); verifyNoMoreInteractions(mockHelper);

14
grpclb/src/main/java/io/grpc/grpclb/GrpclbLoadBalancer2.java → grpclb/src/main/java/io/grpc/grpclb/GrpclbLoadBalancer.java
View File

@ -43,7 +43,7 @@ import com.google.common.annotations.VisibleForTesting;
import io.grpc.Attributes; import io.grpc.Attributes;
import io.grpc.ConnectivityStateInfo; import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.ManagedChannel; import io.grpc.ManagedChannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.ResolvedServerInfoGroup; import io.grpc.ResolvedServerInfoGroup;
@ -67,13 +67,13 @@ import java.util.logging.Logger;
import javax.annotation.Nullable; import javax.annotation.Nullable;
/** /**
* A {@link LoadBalancer2} that uses the GRPCLB protocol. * A {@link LoadBalancer} that uses the GRPCLB protocol.
* *
* <p>Optionally, when requested by the naming system, will delegate the work to a local pick-first * <p>Optionally, when requested by the naming system, will delegate the work to a local pick-first
* or round-robin balancer. * or round-robin balancer.
*/ */
class GrpclbLoadBalancer2 extends LoadBalancer2 implements WithLogId { class GrpclbLoadBalancer extends LoadBalancer implements WithLogId {
private static final Logger logger = Logger.getLogger(GrpclbLoadBalancer2.class.getName()); private static final Logger logger = Logger.getLogger(GrpclbLoadBalancer.class.getName());
@VisibleForTesting @VisibleForTesting
static final SubchannelPicker BUFFER_PICKER = new SubchannelPicker() { static final SubchannelPicker BUFFER_PICKER = new SubchannelPicker() {
@ -104,7 +104,7 @@ class GrpclbLoadBalancer2 extends LoadBalancer2 implements WithLogId {
// If not null, all work is delegated to it. // If not null, all work is delegated to it.
@Nullable @Nullable
private LoadBalancer2 delegate; private LoadBalancer delegate;
private LbPolicy lbPolicy; private LbPolicy lbPolicy;
/////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////
@ -128,7 +128,7 @@ class GrpclbLoadBalancer2 extends LoadBalancer2 implements WithLogId {
// A null element indicate a simulated error for throttling purpose // A null element indicate a simulated error for throttling purpose
private List<EquivalentAddressGroup> roundRobinList = Collections.emptyList(); private List<EquivalentAddressGroup> roundRobinList = Collections.emptyList();
GrpclbLoadBalancer2(Helper helper, Factory pickFirstBalancerFactory, GrpclbLoadBalancer(Helper helper, Factory pickFirstBalancerFactory,
Factory roundRobinBalancerFactory) { Factory roundRobinBalancerFactory) {
this.helper = checkNotNull(helper, "helper"); this.helper = checkNotNull(helper, "helper");
this.serviceName = checkNotNull(helper.getAuthority(), "helper returns null authority"); this.serviceName = checkNotNull(helper.getAuthority(), "helper returns null authority");
@ -450,7 +450,7 @@ class GrpclbLoadBalancer2 extends LoadBalancer2 implements WithLogId {
} }
@VisibleForTesting @VisibleForTesting
LoadBalancer2 getDelegate() { LoadBalancer getDelegate() {
return delegate; return delegate;
} }

28
grpclb/src/main/java/io/grpc/grpclb/GrpclbLoadBalancerFactory2.java → grpclb/src/main/java/io/grpc/grpclb/GrpclbLoadBalancerFactory.java
View File

@ -32,36 +32,32 @@
package io.grpc.grpclb; package io.grpc.grpclb;
import io.grpc.ExperimentalApi; import io.grpc.ExperimentalApi;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.PickFirstBalancerFactory2; import io.grpc.PickFirstBalancerFactory;
import io.grpc.util.RoundRobinLoadBalancerFactory2; import io.grpc.util.RoundRobinLoadBalancerFactory;
/** /**
* A factory for {@link LoadBalancer2}s that uses the GRPCLB protocol. * A factory for {@link LoadBalancer}s that uses the GRPCLB protocol.
*
* <p><strong>TECHNICAL PREVIEW: </strong>The name of this class is temporary. It will be renamed to
* {@code GrpclbLoadBalancerFactory} during <a href="https://github.com/grpc/grpc-java/issues/2656">
* transition to LBv2</a>. You should use it only if you want to experiment the LBv2 code path.
* *
* <p><b>Experimental:</b>This only works with the GRPCLB load-balancer service, which is not * <p><b>Experimental:</b>This only works with the GRPCLB load-balancer service, which is not
* available yet. Right now it's only good for internal testing. * available yet. Right now it's only good for internal testing.
*/ */
@ExperimentalApi("https://github.com/grpc/grpc-java/issues/1782") @ExperimentalApi("https://github.com/grpc/grpc-java/issues/1782")
public class GrpclbLoadBalancerFactory2 extends LoadBalancer2.Factory { public class GrpclbLoadBalancerFactory extends LoadBalancer.Factory {
private static final GrpclbLoadBalancerFactory2 instance = new GrpclbLoadBalancerFactory2(); private static final GrpclbLoadBalancerFactory instance = new GrpclbLoadBalancerFactory();
private GrpclbLoadBalancerFactory2() { private GrpclbLoadBalancerFactory() {
} }
public static GrpclbLoadBalancerFactory2 getInstance() { public static GrpclbLoadBalancerFactory getInstance() {
return instance; return instance;
} }
@Override @Override
public LoadBalancer2 newLoadBalancer(LoadBalancer2.Helper helper) { public LoadBalancer newLoadBalancer(LoadBalancer.Helper helper) {
return new GrpclbLoadBalancer2( return new GrpclbLoadBalancer(
helper, PickFirstBalancerFactory2.getInstance(), helper, PickFirstBalancerFactory.getInstance(),
RoundRobinLoadBalancerFactory2.getInstance()); RoundRobinLoadBalancerFactory.getInstance());
} }
} }

40
grpclb/src/test/java/io/grpc/grpclb/GrpclbLoadBalancer2Test.java → grpclb/src/test/java/io/grpc/grpclb/GrpclbLoadBalancerTest.java
View File

@ -62,11 +62,11 @@ import io.grpc.Attributes;
import io.grpc.CallOptions; import io.grpc.CallOptions;
import io.grpc.ConnectivityStateInfo; import io.grpc.ConnectivityStateInfo;
import io.grpc.EquivalentAddressGroup; import io.grpc.EquivalentAddressGroup;
import io.grpc.LoadBalancer2; import io.grpc.LoadBalancer;
import io.grpc.LoadBalancer2.Helper; import io.grpc.LoadBalancer.Helper;
import io.grpc.LoadBalancer2.PickResult; import io.grpc.LoadBalancer.PickResult;
import io.grpc.LoadBalancer2.Subchannel; import io.grpc.LoadBalancer.Subchannel;
import io.grpc.LoadBalancer2.SubchannelPicker; import io.grpc.LoadBalancer.SubchannelPicker;
import io.grpc.ManagedChannel; import io.grpc.ManagedChannel;
import io.grpc.Metadata; import io.grpc.Metadata;
import io.grpc.MethodDescriptor; import io.grpc.MethodDescriptor;
@ -76,8 +76,8 @@ import io.grpc.ResolvedServerInfoGroup;
import io.grpc.Status; import io.grpc.Status;
import io.grpc.StatusRuntimeException; import io.grpc.StatusRuntimeException;
import io.grpc.grpclb.GrpclbConstants.LbPolicy; import io.grpc.grpclb.GrpclbConstants.LbPolicy;
import io.grpc.grpclb.GrpclbLoadBalancer2.ErrorPicker; import io.grpc.grpclb.GrpclbLoadBalancer.ErrorPicker;
import io.grpc.grpclb.GrpclbLoadBalancer2.RoundRobinPicker; import io.grpc.grpclb.GrpclbLoadBalancer.RoundRobinPicker;
import io.grpc.inprocess.InProcessChannelBuilder; import io.grpc.inprocess.InProcessChannelBuilder;
import io.grpc.inprocess.InProcessServerBuilder; import io.grpc.inprocess.InProcessServerBuilder;
import io.grpc.internal.SerializingExecutor; import io.grpc.internal.SerializingExecutor;
@ -105,9 +105,9 @@ import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock; import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer; import org.mockito.stubbing.Answer;
/** Unit tests for {@link GrpclbLoadBalancer2}. */ /** Unit tests for {@link GrpclbLoadBalancer}. */
@RunWith(JUnit4.class) @RunWith(JUnit4.class)
public class GrpclbLoadBalancer2Test { public class GrpclbLoadBalancerTest {
private static final Attributes.Key<String> RESOLUTION_ATTR = private static final Attributes.Key<String> RESOLUTION_ATTR =
Attributes.Key.of("resolution-attr"); Attributes.Key.of("resolution-attr");
private static final String SERVICE_AUTHORITY = "api.google.com"; private static final String SERVICE_AUTHORITY = "api.google.com";
@ -159,14 +159,14 @@ public class GrpclbLoadBalancer2Test {
new SerializingExecutor(MoreExecutors.directExecutor()); new SerializingExecutor(MoreExecutors.directExecutor());
private final Metadata headers = new Metadata(); private final Metadata headers = new Metadata();
@Mock @Mock
private LoadBalancer2.Factory pickFirstBalancerFactory; private LoadBalancer.Factory pickFirstBalancerFactory;
@Mock @Mock
private LoadBalancer2 pickFirstBalancer; private LoadBalancer pickFirstBalancer;
@Mock @Mock
private LoadBalancer2.Factory roundRobinBalancerFactory; private LoadBalancer.Factory roundRobinBalancerFactory;
@Mock @Mock
private LoadBalancer2 roundRobinBalancer; private LoadBalancer roundRobinBalancer;
private GrpclbLoadBalancer2 balancer; private GrpclbLoadBalancer balancer;
@SuppressWarnings("unchecked") @SuppressWarnings("unchecked")
@Before @Before
@ -240,7 +240,7 @@ public class GrpclbLoadBalancer2Test {
} }
}).when(helper).runSerialized(any(Runnable.class)); }).when(helper).runSerialized(any(Runnable.class));
when(helper.getAuthority()).thenReturn(SERVICE_AUTHORITY); when(helper.getAuthority()).thenReturn(SERVICE_AUTHORITY);
balancer = new GrpclbLoadBalancer2(helper, pickFirstBalancerFactory, roundRobinBalancerFactory); balancer = new GrpclbLoadBalancer(helper, pickFirstBalancerFactory, roundRobinBalancerFactory);
} }
@After @After
@ -290,7 +290,7 @@ public class GrpclbLoadBalancer2Test {
@Test @Test
public void bufferPicker() { public void bufferPicker() {
assertEquals(PickResult.withNoResult(), assertEquals(PickResult.withNoResult(),
GrpclbLoadBalancer2.BUFFER_PICKER.pickSubchannel(Attributes.EMPTY, headers)); GrpclbLoadBalancer.BUFFER_PICKER.pickSubchannel(Attributes.EMPTY, headers));
} }
@Test @Test
@ -553,11 +553,11 @@ public class GrpclbLoadBalancer2Test {
assertEquals(new EquivalentAddressGroup(backends1.get(1)), subchannel2.getAddresses()); assertEquals(new EquivalentAddressGroup(backends1.get(1)), subchannel2.getAddresses());
// Before any subchannel is READY, a buffer picker will be provided // Before any subchannel is READY, a buffer picker will be provided
inOrder.verify(helper).updatePicker(same(GrpclbLoadBalancer2.BUFFER_PICKER)); inOrder.verify(helper).updatePicker(same(GrpclbLoadBalancer.BUFFER_PICKER));
deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(CONNECTING)); deliverSubchannelState(subchannel1, ConnectivityStateInfo.forNonError(CONNECTING));
deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(CONNECTING)); deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(CONNECTING));
inOrder.verify(helper, times(2)).updatePicker(same(GrpclbLoadBalancer2.BUFFER_PICKER)); inOrder.verify(helper, times(2)).updatePicker(same(GrpclbLoadBalancer.BUFFER_PICKER));
// Let subchannels be connected // Let subchannels be connected
deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY)); deliverSubchannelState(subchannel2, ConnectivityStateInfo.forNonError(READY));
@ -724,7 +724,7 @@ public class GrpclbLoadBalancer2Test {
eq(new EquivalentAddressGroup(backends.get(0))), any(Attributes.class)); eq(new EquivalentAddressGroup(backends.get(0))), any(Attributes.class));
inOrder.verify(helper).createSubchannel( inOrder.verify(helper).createSubchannel(
eq(new EquivalentAddressGroup(backends.get(1))), any(Attributes.class)); eq(new EquivalentAddressGroup(backends.get(1))), any(Attributes.class));
inOrder.verify(helper).updatePicker(same(GrpclbLoadBalancer2.BUFFER_PICKER)); inOrder.verify(helper).updatePicker(same(GrpclbLoadBalancer.BUFFER_PICKER));
inOrder.verifyNoMoreInteractions(); inOrder.verifyNoMoreInteractions();
} }
@ -806,7 +806,7 @@ public class GrpclbLoadBalancer2Test {
Subchannel subchannel = subchannels[i]; Subchannel subchannel = subchannels[i];
if (subchannel == null) { if (subchannel == null) {
assertSame("list[" + i + "] should be drop", assertSame("list[" + i + "] should be drop",
GrpclbLoadBalancer2.THROTTLED_RESULT, picker.list.get(i)); GrpclbLoadBalancer.THROTTLED_RESULT, picker.list.get(i));
} else { } else {
assertEquals("list[" + i + "] should be Subchannel", assertEquals("list[" + i + "] should be Subchannel",
subchannel, picker.list.get(i).getSubchannel()); subchannel, picker.list.get(i).getSubchannel());