3 * Copyright 2017 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.
25 "google.golang.org/grpc/balancer"
26 "google.golang.org/grpc/connectivity"
27 "google.golang.org/grpc/grpclog"
28 "google.golang.org/grpc/resolver"
31 // scStateUpdate contains the subConn and the new state it changed to.
32 type scStateUpdate struct {
34 state connectivity.State
37 // scStateUpdateBuffer is an unbounded channel for scStateChangeTuple.
38 // TODO make a general purpose buffer that uses interface{}.
39 type scStateUpdateBuffer struct {
42 backlog []*scStateUpdate
45 func newSCStateUpdateBuffer() *scStateUpdateBuffer {
46 return &scStateUpdateBuffer{
47 c: make(chan *scStateUpdate, 1),
51 func (b *scStateUpdateBuffer) put(t *scStateUpdate) {
54 if len(b.backlog) == 0 {
61 b.backlog = append(b.backlog, t)
64 func (b *scStateUpdateBuffer) load() {
67 if len(b.backlog) > 0 {
69 case b.c <- b.backlog[0]:
71 b.backlog = b.backlog[1:]
77 // get returns the channel that the scStateUpdate will be sent to.
79 // Upon receiving, the caller should call load to send another
80 // scStateChangeTuple onto the channel if there is any.
81 func (b *scStateUpdateBuffer) get() <-chan *scStateUpdate {
85 // resolverUpdate contains the new resolved addresses or error if there's
87 type resolverUpdate struct {
88 addrs []resolver.Address
92 // ccBalancerWrapper is a wrapper on top of cc for balancers.
93 // It implements balancer.ClientConn interface.
94 type ccBalancerWrapper struct {
96 balancer balancer.Balancer
97 stateChangeQueue *scStateUpdateBuffer
98 resolverUpdateCh chan *resolverUpdate
102 subConns map[*acBalancerWrapper]struct{}
105 func newCCBalancerWrapper(cc *ClientConn, b balancer.Builder, bopts balancer.BuildOptions) *ccBalancerWrapper {
106 ccb := &ccBalancerWrapper{
108 stateChangeQueue: newSCStateUpdateBuffer(),
109 resolverUpdateCh: make(chan *resolverUpdate, 1),
110 done: make(chan struct{}),
111 subConns: make(map[*acBalancerWrapper]struct{}),
114 ccb.balancer = b.Build(ccb, bopts)
118 // watcher balancer functions sequentially, so the balancer can be implemented
120 func (ccb *ccBalancerWrapper) watcher() {
123 case t := <-ccb.stateChangeQueue.get():
124 ccb.stateChangeQueue.load()
131 ccb.balancer.HandleSubConnStateChange(t.sc, t.state)
132 case t := <-ccb.resolverUpdateCh:
139 ccb.balancer.HandleResolvedAddrs(t.addrs, t.err)
150 for acbw := range scs {
151 ccb.cc.removeAddrConn(acbw.getAddrConn(), errConnDrain)
159 func (ccb *ccBalancerWrapper) close() {
163 func (ccb *ccBalancerWrapper) handleSubConnStateChange(sc balancer.SubConn, s connectivity.State) {
164 // When updating addresses for a SubConn, if the address in use is not in
165 // the new addresses, the old ac will be tearDown() and a new ac will be
166 // created. tearDown() generates a state change with Shutdown state, we
167 // don't want the balancer to receive this state change. So before
168 // tearDown() on the old ac, ac.acbw (acWrapper) will be set to nil, and
169 // this function will be called with (nil, Shutdown). We don't need to call
170 // balancer method in this case.
174 ccb.stateChangeQueue.put(&scStateUpdate{
180 func (ccb *ccBalancerWrapper) handleResolvedAddrs(addrs []resolver.Address, err error) {
181 if ccb.cc.curBalancerName != grpclbName {
182 var containsGRPCLB bool
183 for _, a := range addrs {
184 if a.Type == resolver.GRPCLB {
185 containsGRPCLB = true
190 // The current balancer is not grpclb, but addresses contain grpclb
191 // address. This means we failed to switch to grpclb, most likely
192 // because grpclb is not registered. Filter out all grpclb addresses
193 // from addrs before sending to balancer.
194 tempAddrs := make([]resolver.Address, 0, len(addrs))
195 for _, a := range addrs {
196 if a.Type != resolver.GRPCLB {
197 tempAddrs = append(tempAddrs, a)
204 case <-ccb.resolverUpdateCh:
207 ccb.resolverUpdateCh <- &resolverUpdate{
213 func (ccb *ccBalancerWrapper) NewSubConn(addrs []resolver.Address, opts balancer.NewSubConnOptions) (balancer.SubConn, error) {
215 return nil, fmt.Errorf("grpc: cannot create SubConn with empty address list")
218 defer ccb.mu.Unlock()
219 if ccb.subConns == nil {
220 return nil, fmt.Errorf("grpc: ClientConn balancer wrapper was closed")
222 ac, err := ccb.cc.newAddrConn(addrs, opts)
226 acbw := &acBalancerWrapper{ac: ac}
230 ccb.subConns[acbw] = struct{}{}
234 func (ccb *ccBalancerWrapper) RemoveSubConn(sc balancer.SubConn) {
235 acbw, ok := sc.(*acBalancerWrapper)
240 defer ccb.mu.Unlock()
241 if ccb.subConns == nil {
244 delete(ccb.subConns, acbw)
245 ccb.cc.removeAddrConn(acbw.getAddrConn(), errConnDrain)
248 func (ccb *ccBalancerWrapper) UpdateBalancerState(s connectivity.State, p balancer.Picker) {
250 defer ccb.mu.Unlock()
251 if ccb.subConns == nil {
254 // Update picker before updating state. Even though the ordering here does
255 // not matter, it can lead to multiple calls of Pick in the common start-up
256 // case where we wait for ready and then perform an RPC. If the picker is
257 // updated later, we could call the "connecting" picker when the state is
258 // updated, and then call the "ready" picker after the picker gets updated.
259 ccb.cc.blockingpicker.updatePicker(p)
260 ccb.cc.csMgr.updateState(s)
263 func (ccb *ccBalancerWrapper) ResolveNow(o resolver.ResolveNowOption) {
267 func (ccb *ccBalancerWrapper) Target() string {
271 // acBalancerWrapper is a wrapper on top of ac for balancers.
272 // It implements balancer.SubConn interface.
273 type acBalancerWrapper struct {
278 func (acbw *acBalancerWrapper) UpdateAddresses(addrs []resolver.Address) {
280 defer acbw.mu.Unlock()
282 acbw.ac.tearDown(errConnDrain)
285 if !acbw.ac.tryUpdateAddrs(addrs) {
287 opts := acbw.ac.scopts
289 // Set old ac.acbw to nil so the Shutdown state update will be ignored
292 // TODO(bar) the state transition could be wrong when tearDown() old ac
293 // and creating new ac, fix the transition.
296 acState := acbw.ac.getState()
297 acbw.ac.tearDown(errConnDrain)
299 if acState == connectivity.Shutdown {
303 ac, err := cc.newAddrConn(addrs, opts)
305 grpclog.Warningf("acBalancerWrapper: UpdateAddresses: failed to newAddrConn: %v", err)
312 if acState != connectivity.Idle {
318 func (acbw *acBalancerWrapper) Connect() {
320 defer acbw.mu.Unlock()
324 func (acbw *acBalancerWrapper) getAddrConn() *addrConn {
326 defer acbw.mu.Unlock()