3 * Copyright 2016 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
26 "google.golang.org/grpc/codes"
27 "google.golang.org/grpc/credentials"
28 "google.golang.org/grpc/grpclog"
29 "google.golang.org/grpc/naming"
30 "google.golang.org/grpc/status"
33 // Address represents a server the client connects to.
35 // Deprecated: please use package balancer.
37 // Addr is the server address on which a connection will be established.
39 // Metadata is the information associated with Addr, which may be used
40 // to make load balancing decision.
44 // BalancerConfig specifies the configurations for Balancer.
46 // Deprecated: please use package balancer.
47 type BalancerConfig struct {
48 // DialCreds is the transport credential the Balancer implementation can
49 // use to dial to a remote load balancer server. The Balancer implementations
50 // can ignore this if it does not need to talk to another party securely.
51 DialCreds credentials.TransportCredentials
52 // Dialer is the custom dialer the Balancer implementation can use to dial
53 // to a remote load balancer server. The Balancer implementations
54 // can ignore this if it doesn't need to talk to remote balancer.
55 Dialer func(context.Context, string) (net.Conn, error)
58 // BalancerGetOptions configures a Get call.
60 // Deprecated: please use package balancer.
61 type BalancerGetOptions struct {
62 // BlockingWait specifies whether Get should block when there is no
67 // Balancer chooses network addresses for RPCs.
69 // Deprecated: please use package balancer.
70 type Balancer interface {
71 // Start does the initialization work to bootstrap a Balancer. For example,
72 // this function may start the name resolution and watch the updates. It will
73 // be called when dialing.
74 Start(target string, config BalancerConfig) error
75 // Up informs the Balancer that gRPC has a connection to the server at
76 // addr. It returns down which is called once the connection to addr gets
78 // TODO: It is not clear how to construct and take advantage of the meaningful error
79 // parameter for down. Need realistic demands to guide.
80 Up(addr Address) (down func(error))
81 // Get gets the address of a server for the RPC corresponding to ctx.
82 // i) If it returns a connected address, gRPC internals issues the RPC on the
83 // connection to this address;
84 // ii) If it returns an address on which the connection is under construction
85 // (initiated by Notify(...)) but not connected, gRPC internals
86 // * fails RPC if the RPC is fail-fast and connection is in the TransientFailure or
89 // * issues RPC on the connection otherwise.
90 // iii) If it returns an address on which the connection does not exist, gRPC
91 // internals treats it as an error and will fail the corresponding RPC.
93 // Therefore, the following is the recommended rule when writing a custom Balancer.
94 // If opts.BlockingWait is true, it should return a connected address or
95 // block if there is no connected address. It should respect the timeout or
96 // cancellation of ctx when blocking. If opts.BlockingWait is false (for fail-fast
97 // RPCs), it should return an address it has notified via Notify(...) immediately
98 // instead of blocking.
100 // The function returns put which is called once the rpc has completed or failed.
101 // put can collect and report RPC stats to a remote load balancer.
103 // This function should only return the errors Balancer cannot recover by itself.
104 // gRPC internals will fail the RPC if an error is returned.
105 Get(ctx context.Context, opts BalancerGetOptions) (addr Address, put func(), err error)
106 // Notify returns a channel that is used by gRPC internals to watch the addresses
107 // gRPC needs to connect. The addresses might be from a name resolver or remote
108 // load balancer. gRPC internals will compare it with the existing connected
109 // addresses. If the address Balancer notified is not in the existing connected
110 // addresses, gRPC starts to connect the address. If an address in the existing
111 // connected addresses is not in the notification list, the corresponding connection
112 // is shutdown gracefully. Otherwise, there are no operations to take. Note that
113 // the Address slice must be the full list of the Addresses which should be connected.
115 Notify() <-chan []Address
116 // Close shuts down the balancer.
120 // RoundRobin returns a Balancer that selects addresses round-robin. It uses r to watch
121 // the name resolution updates and updates the addresses available correspondingly.
123 // Deprecated: please use package balancer/roundrobin.
124 func RoundRobin(r naming.Resolver) Balancer {
125 return &roundRobin{r: r}
128 type addrInfo struct {
133 type roundRobin struct {
136 addrs []*addrInfo // all the addresses the client should potentially connect
138 addrCh chan []Address // the channel to notify gRPC internals the list of addresses the client should connect to.
139 next int // index of the next address to return for Get()
140 waitCh chan struct{} // the channel to block when there is no connected address available
141 done bool // The Balancer is closed.
144 func (rr *roundRobin) watchAddrUpdates() error {
145 updates, err := rr.w.Next()
147 grpclog.Warningf("grpc: the naming watcher stops working due to %v.", err)
152 for _, update := range updates {
155 Metadata: update.Metadata,
160 for _, v := range rr.addrs {
163 grpclog.Infoln("grpc: The name resolver wanted to add an existing address: ", addr)
170 rr.addrs = append(rr.addrs, &addrInfo{addr: addr})
172 for i, v := range rr.addrs {
174 copy(rr.addrs[i:], rr.addrs[i+1:])
175 rr.addrs = rr.addrs[:len(rr.addrs)-1]
180 grpclog.Errorln("Unknown update.Op ", update.Op)
183 // Make a copy of rr.addrs and write it onto rr.addrCh so that gRPC internals gets notified.
184 open := make([]Address, len(rr.addrs))
185 for i, v := range rr.addrs {
189 return ErrClientConnClosing
199 func (rr *roundRobin) Start(target string, config BalancerConfig) error {
203 return ErrClientConnClosing
206 // If there is no name resolver installed, it is not needed to
207 // do name resolution. In this case, target is added into rr.addrs
208 // as the only address available and rr.addrCh stays nil.
209 rr.addrs = append(rr.addrs, &addrInfo{addr: Address{Addr: target}})
212 w, err := rr.r.Resolve(target)
217 rr.addrCh = make(chan []Address, 1)
220 if err := rr.watchAddrUpdates(); err != nil {
228 // Up sets the connected state of addr and sends notification if there are pending
230 func (rr *roundRobin) Up(addr Address) func(error) {
234 for _, a := range rr.addrs {
245 // addr is only one which is connected. Notify the Get() callers who are blocking.
246 if cnt == 1 && rr.waitCh != nil {
250 return func(err error) {
255 // down unsets the connected state of addr.
256 func (rr *roundRobin) down(addr Address, err error) {
259 for _, a := range rr.addrs {
267 // Get returns the next addr in the rotation.
268 func (rr *roundRobin) Get(ctx context.Context, opts BalancerGetOptions) (addr Address, put func(), err error) {
273 err = ErrClientConnClosing
277 if len(rr.addrs) > 0 {
278 if rr.next >= len(rr.addrs) {
284 next = (next + 1) % len(rr.addrs)
292 // Has iterated all the possible address but none is connected.
297 if !opts.BlockingWait {
298 if len(rr.addrs) == 0 {
300 err = status.Errorf(codes.Unavailable, "there is no address available")
303 // Returns the next addr on rr.addrs for failfast RPCs.
304 addr = rr.addrs[rr.next].addr
309 // Wait on rr.waitCh for non-failfast RPCs.
310 if rr.waitCh == nil {
311 ch = make(chan struct{})
326 err = ErrClientConnClosing
330 if len(rr.addrs) > 0 {
331 if rr.next >= len(rr.addrs) {
337 next = (next + 1) % len(rr.addrs)
345 // Has iterated all the possible address but none is connected.
350 // The newly added addr got removed by Down() again.
351 if rr.waitCh == nil {
352 ch = make(chan struct{})
362 func (rr *roundRobin) Notify() <-chan []Address {
366 func (rr *roundRobin) Close() error {
370 return errBalancerClosed
376 if rr.waitCh != nil {
380 if rr.addrCh != nil {
386 // pickFirst is used to test multi-addresses in one addrConn in which all addresses share the same addrConn.
387 // It is a wrapper around roundRobin balancer. The logic of all methods works fine because balancer.Get()
388 // returns the only address Up by resetTransport().
389 type pickFirst struct {