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.
This commit is contained in:
Easwar Swaminathan
2019-11-04 14:22:16 -08:00
committed by GitHub
parent 88bf070301
commit 583401aac5
3 changed files with 205 additions and 62 deletions

View File

@ -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)
}
}