From f02eeef0f629d2336e745ad15a3c4b7306704bf9 Mon Sep 17 00:00:00 2001 From: kevin Date: Thu, 6 Aug 2020 23:06:26 +0800 Subject: [PATCH] add subset algorithm --- .../internal/consistentbalancer_test.go | 2 +- .../internal/roundrobinbalancer_test.go | 4 +- core/hash/consistenthash_test.go | 2 +- core/mathx/entropy.go | 7 +- core/mathx/entropy_test.go | 2 +- core/mathx/unstable_test.go | 2 +- .../balancer/roundrobin/roundrobin.go | 65 ------------------- rpcx/internal/codes/{error.go => accept.go} | 0 rpcx/internal/resolver/resolver.go | 3 +- rpcx/internal/resolver/subset.go | 14 ++++ rpcx/internal/resolver/subset_test.go | 65 +++++++++++++++++++ 11 files changed, 93 insertions(+), 73 deletions(-) delete mode 100644 rpcx/internal/balancer/roundrobin/roundrobin.go rename rpcx/internal/codes/{error.go => accept.go} (100%) create mode 100644 rpcx/internal/resolver/subset.go create mode 100644 rpcx/internal/resolver/subset_test.go diff --git a/core/discov/internal/consistentbalancer_test.go b/core/discov/internal/consistentbalancer_test.go index 567c1358..1f71d72d 100644 --- a/core/discov/internal/consistentbalancer_test.go +++ b/core/discov/internal/consistentbalancer_test.go @@ -119,7 +119,7 @@ func TestConsistent_next(t *testing.T) { m[val]++ } - entropy := mathx.CalcEntropy(m, total) + entropy := mathx.CalcEntropy(m) assert.Equal(t, size, len(m)) assert.True(t, entropy > .95) diff --git a/core/discov/internal/roundrobinbalancer_test.go b/core/discov/internal/roundrobinbalancer_test.go index ef3037e1..80a1b896 100644 --- a/core/discov/internal/roundrobinbalancer_test.go +++ b/core/discov/internal/roundrobinbalancer_test.go @@ -237,7 +237,7 @@ func TestRoundRobin_initialize(t *testing.T) { for k, v := range m { mi[k] = v } - entropy := mathx.CalcEntropy(mi, total) + entropy := mathx.CalcEntropy(mi) assert.True(t, entropy > .95) } @@ -265,7 +265,7 @@ func TestRoundRobin_next(t *testing.T) { m[val]++ } - entropy := mathx.CalcEntropy(m, total) + entropy := mathx.CalcEntropy(m) assert.Equal(t, size, len(m)) assert.True(t, entropy > .95) diff --git a/core/hash/consistenthash_test.go b/core/hash/consistenthash_test.go index de344c63..cc35380b 100644 --- a/core/hash/consistenthash_test.go +++ b/core/hash/consistenthash_test.go @@ -47,7 +47,7 @@ func TestConsistentHash(t *testing.T) { for k, v := range keys { mi[k] = v } - entropy := mathx.CalcEntropy(mi, requestSize) + entropy := mathx.CalcEntropy(mi) assert.True(t, entropy > .95) } diff --git a/core/mathx/entropy.go b/core/mathx/entropy.go index 0b3c7f22..f426f2c8 100644 --- a/core/mathx/entropy.go +++ b/core/mathx/entropy.go @@ -2,9 +2,14 @@ package mathx import "math" -func CalcEntropy(m map[interface{}]int, total int) float64 { +func CalcEntropy(m map[interface{}]int) float64 { var entropy float64 + var total int + for _, v := range m { + total += v + } + for _, v := range m { proba := float64(v) / float64(total) entropy -= proba * math.Log2(proba) diff --git a/core/mathx/entropy_test.go b/core/mathx/entropy_test.go index 87130cff..33e1a905 100644 --- a/core/mathx/entropy_test.go +++ b/core/mathx/entropy_test.go @@ -13,5 +13,5 @@ func TestCalcEntropy(t *testing.T) { for i := 0; i < total; i++ { m[i] = count } - assert.True(t, CalcEntropy(m, total*count) > .99) + assert.True(t, CalcEntropy(m) > .99) } diff --git a/core/mathx/unstable_test.go b/core/mathx/unstable_test.go index e9bb53c0..2db5fb53 100644 --- a/core/mathx/unstable_test.go +++ b/core/mathx/unstable_test.go @@ -65,7 +65,7 @@ func TestUnstable_Distribution(t *testing.T) { for k, v := range m { mi[k] = v } - entropy := CalcEntropy(mi, total) + entropy := CalcEntropy(mi) assert.True(t, len(m) > 1) assert.True(t, entropy > 0.95) } diff --git a/rpcx/internal/balancer/roundrobin/roundrobin.go b/rpcx/internal/balancer/roundrobin/roundrobin.go deleted file mode 100644 index 0e54ec6c..00000000 --- a/rpcx/internal/balancer/roundrobin/roundrobin.go +++ /dev/null @@ -1,65 +0,0 @@ -package roundrobin - -import ( - "context" - "math/rand" - "sync" - "time" - - "google.golang.org/grpc/balancer" - "google.golang.org/grpc/balancer/base" - "google.golang.org/grpc/resolver" -) - -const Name = "rr" - -func init() { - balancer.Register(newBuilder()) -} - -type roundRobinPickerBuilder struct{} - -func newBuilder() balancer.Builder { - return base.NewBalancerBuilder(Name, new(roundRobinPickerBuilder)) -} - -func (b *roundRobinPickerBuilder) Build(readySCs map[resolver.Address]balancer.SubConn) balancer.Picker { - if len(readySCs) == 0 { - return base.NewErrPicker(balancer.ErrNoSubConnAvailable) - } - - rand.Seed(time.Now().UnixNano()) - picker := &roundRobinPicker{ - index: rand.Int(), - } - - for addr, conn := range readySCs { - picker.conns = append(picker.conns, &subConn{ - addr: addr, - conn: conn, - }) - } - - return picker -} - -type roundRobinPicker struct { - conns []*subConn - index int - lock sync.Mutex -} - -func (p *roundRobinPicker) Pick(ctx context.Context, info balancer.PickInfo) ( - conn balancer.SubConn, done func(balancer.DoneInfo), err error) { - p.lock.Lock() - defer p.lock.Unlock() - - p.index = (p.index + 1) % len(p.conns) - return p.conns[p.index].conn, func(info balancer.DoneInfo) { - }, nil -} - -type subConn struct { - addr resolver.Address - conn balancer.SubConn -} diff --git a/rpcx/internal/codes/error.go b/rpcx/internal/codes/accept.go similarity index 100% rename from rpcx/internal/codes/error.go rename to rpcx/internal/codes/accept.go diff --git a/rpcx/internal/resolver/resolver.go b/rpcx/internal/resolver/resolver.go index c4b64f75..fc4f089d 100644 --- a/rpcx/internal/resolver/resolver.go +++ b/rpcx/internal/resolver/resolver.go @@ -12,6 +12,7 @@ import ( const ( DiscovScheme = "discov" EndpointSep = "," + subsetSize = 36 ) var builder discovBuilder @@ -36,7 +37,7 @@ func (b *discovBuilder) Build(target resolver.Target, cc resolver.ClientConn, op update := func() { var addrs []resolver.Address - for _, val := range sub.Values() { + for _, val := range subset(sub.Values(), subsetSize) { addrs = append(addrs, resolver.Address{ Addr: val, }) diff --git a/rpcx/internal/resolver/subset.go b/rpcx/internal/resolver/subset.go new file mode 100644 index 00000000..3c6c8f06 --- /dev/null +++ b/rpcx/internal/resolver/subset.go @@ -0,0 +1,14 @@ +package resolver + +import "math/rand" + +func subset(set []string, sub int) []string { + rand.Shuffle(len(set), func(i, j int) { + set[i], set[j] = set[j], set[i] + }) + if len(set) <= sub { + return set + } else { + return set[:sub] + } +} diff --git a/rpcx/internal/resolver/subset_test.go b/rpcx/internal/resolver/subset_test.go new file mode 100644 index 00000000..f9410a7a --- /dev/null +++ b/rpcx/internal/resolver/subset_test.go @@ -0,0 +1,65 @@ +package resolver + +import ( + "strconv" + "testing" + + "zero/core/mathx" + + "github.com/stretchr/testify/assert" +) + +func TestSubset(t *testing.T) { + tests := []struct { + name string + set int + sub int + }{ + { + name: "more vals to subset", + set: 100, + sub: 36, + }, + { + name: "less vals to subset", + set: 100, + sub: 200, + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + var vals []string + for i := 0; i < test.set; i++ { + vals = append(vals, strconv.Itoa(i)) + } + + m := make(map[interface{}]int) + for i := 0; i < 1000; i++ { + set := subset(append([]string(nil), vals...), test.sub) + for _, val := range set { + m[val]++ + } + } + + assert.True(t, mathx.CalcEntropy(m) > 0.95) + }) + } +} + +func TestSubsetLess(t *testing.T) { + var vals []string + for i := 0; i < 100; i++ { + vals = append(vals, strconv.Itoa(i)) + } + + m := make(map[interface{}]int) + for i := 0; i < 1000; i++ { + set := subset(append([]string(nil), vals...), 200) + for _, val := range set { + m[val]++ + } + } + + assert.True(t, mathx.CalcEntropy(m) > 0.95) +}