Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add a general purpose unbounded buffer implementation (#3099)
This PR moves the unbounded buffer implementation found in
`scStateUpdateBuffer` to the internal package. It also makes the buffer
work with `interface{}` type.
This addresses a TODO in the existing code. This will also help with the
eventual `BalancerManager` implementation which will supersede the
`ccBalancerWrapper` implementation found in balancer_conn_wrappers.go.- Loading branch information
Showing
3 changed files
with
205 additions
and
62 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,78 @@ | ||
| /* | ||
| * Copyright 2019 gRPC authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| */ | ||
|
|
||
| // Package buffer provides an implementation of an unbounded buffer. | ||
| package buffer | ||
|
|
||
| import "sync" | ||
|
|
||
| // Unbounded is an implementation of an unbounded buffer which does not use | ||
| // extra goroutines. This is typically used for passing updates from one entity | ||
| // to another within gRPC. | ||
| // | ||
| // All methods on this type are thread-safe and don't block on anything except | ||
| // the underlying mutex used for synchronization. | ||
| type Unbounded struct { | ||
| c chan interface{} | ||
| mu sync.Mutex | ||
| backlog []interface{} | ||
| } | ||
|
|
||
| // NewUnbounded returns a new instance of Unbounded. | ||
| func NewUnbounded() *Unbounded { | ||
| return &Unbounded{c: make(chan interface{}, 1)} | ||
| } | ||
|
|
||
| // Put adds t to the unbounded buffer. | ||
| func (b *Unbounded) Put(t interface{}) { | ||
| b.mu.Lock() | ||
| if len(b.backlog) == 0 { | ||
| select { | ||
| case b.c <- t: | ||
| b.mu.Unlock() | ||
| return | ||
| default: | ||
| } | ||
| } | ||
| b.backlog = append(b.backlog, t) | ||
| b.mu.Unlock() | ||
| } | ||
|
|
||
| // Load sends the earliest buffered data, if any, onto the read channel | ||
| // returned by Get(). Users are expected to call this every time they read a | ||
| // value from the read channel. | ||
| func (b *Unbounded) Load() { | ||
| b.mu.Lock() | ||
| if len(b.backlog) > 0 { | ||
| select { | ||
| case b.c <- b.backlog[0]: | ||
| b.backlog[0] = nil | ||
| b.backlog = b.backlog[1:] | ||
| default: | ||
| } | ||
| } | ||
| b.mu.Unlock() | ||
| } | ||
|
|
||
| // Get returns a read channel on which values added to the buffer, via Put(), | ||
| // are sent on. | ||
| // | ||
| // Upon reading a value from this channel, users are expected to call Load() to | ||
| // send the next buffered value onto the channel if there is any. | ||
| func (b *Unbounded) Get() <-chan interface{} { | ||
| return b.c | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| /* | ||
| * Copyright 2019 gRPC authors. | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| * | ||
| */ | ||
|
|
||
| package buffer | ||
|
|
||
| import ( | ||
| "reflect" | ||
| "sort" | ||
| "sync" | ||
| "testing" | ||
| ) | ||
|
|
||
| const ( | ||
| numWriters = 10 | ||
| numWrites = 10 | ||
| ) | ||
|
|
||
| // wantReads contains the set of values expected to be read by the reader | ||
| // goroutine in the tests. | ||
| var wantReads []int | ||
|
|
||
| func init() { | ||
| for i := 0; i < numWriters; i++ { | ||
| for j := 0; j < numWrites; j++ { | ||
| wantReads = append(wantReads, i) | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // TestSingleWriter starts one reader and one writer goroutine and makes sure | ||
| // that the reader gets all the value added to the buffer by the writer. | ||
| func TestSingleWriter(t *testing.T) { | ||
| ub := NewUnbounded() | ||
| reads := []int{} | ||
|
|
||
| var wg sync.WaitGroup | ||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| ch := ub.Get() | ||
| for i := 0; i < numWriters*numWrites; i++ { | ||
| r := <-ch | ||
| reads = append(reads, r.(int)) | ||
| ub.Load() | ||
| } | ||
| }() | ||
|
|
||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| for i := 0; i < numWriters; i++ { | ||
| for j := 0; j < numWrites; j++ { | ||
| ub.Put(i) | ||
| } | ||
| } | ||
| }() | ||
|
|
||
| wg.Wait() | ||
| if !reflect.DeepEqual(reads, wantReads) { | ||
| t.Errorf("reads: %#v, wantReads: %#v", reads, wantReads) | ||
| } | ||
| } | ||
|
|
||
| // TestMultipleWriters starts multiple writers and one reader goroutine and | ||
| // makes sure that the reader gets all the data written by all writers. | ||
| func TestMultipleWriters(t *testing.T) { | ||
| ub := NewUnbounded() | ||
| reads := []int{} | ||
|
|
||
| var wg sync.WaitGroup | ||
| wg.Add(1) | ||
| go func() { | ||
| defer wg.Done() | ||
| ch := ub.Get() | ||
| for i := 0; i < numWriters*numWrites; i++ { | ||
| r := <-ch | ||
| reads = append(reads, r.(int)) | ||
| ub.Load() | ||
| } | ||
| }() | ||
|
|
||
| wg.Add(numWriters) | ||
| for i := 0; i < numWriters; i++ { | ||
| go func(index int) { | ||
| defer wg.Done() | ||
| for j := 0; j < numWrites; j++ { | ||
| ub.Put(index) | ||
| } | ||
| }(i) | ||
| } | ||
|
|
||
| wg.Wait() | ||
| sort.Ints(reads) | ||
| if !reflect.DeepEqual(reads, wantReads) { | ||
| t.Errorf("reads: %#v, wantReads: %#v", reads, wantReads) | ||
| } | ||
| } |