first commit

Author: sysulq

.gitignore

@@ -0,0 +1,101 @@
+package aperture
+
+import (
+	"math"
+
+	"github.com/hnlq715/go-loadbalance"
+	"github.com/hnlq715/go-loadbalance/p2c/leastloaded"
+	"google.golang.org/grpc/balancer"
+)
+
+type Aperture struct {
+	localID       string
+	localPeers    []string
+	localPeersMap map[string]int
+	remotePeers   []interface{}
+	logicalWidth  int
+
+	p2c loadbalance.P2C
+}
+
+const (
+	defaultLogicalWidth int = 1
+)
+
+func New() loadbalance.Aperture {
+	return &Aperture{
+		logicalWidth:  defaultLogicalWidth,
+		localPeers:    make([]string, 0),
+		localPeersMap: make(map[string]int),
+		remotePeers:   make([]interface{}, 0),
+		p2c:           leastloaded.New(),
+	}
+}
+
+func (a *Aperture) SetLogicalWidth(width int) {
+	a.logicalWidth = width
+	a.rebuild()
+}
+
+func (a *Aperture) SetLocalPeerID(id string) {
+	a.localID = id
+	a.rebuild()
+}
+
+func (a *Aperture) SetLocalPeers(localPeers []string) {
+	a.localPeers = localPeers
+	for idx, local := range localPeers {
+		a.localPeersMap[local] = idx
+	}
+
+	a.rebuild()
+}
+
+func (a *Aperture) SetRemotePeers(remotePeers []interface{}) {
+	a.remotePeers = remotePeers
+	a.rebuild()
+}
+
+func (a *Aperture) Next() (interface{}, func(balancer.DoneInfo)) {
+	return a.p2c.Next()
+}
+
+func (a *Aperture) rebuild() {
+	if len(a.localPeers) == 0 {
+		return
+	}
+
+	if len(a.remotePeers) == 0 {
+		return
+	}
+
+	idx, ok := a.localPeersMap[a.localID]
+	if !ok {
+		return
+	}
+
+	localWidth := floatOne / float64(len(a.localPeers))
+	remoteWidth := floatOne / float64(len(a.remotePeers))
+
+	width := dApertureWidth(localWidth, remoteWidth, a.logicalWidth)
+
+	offset := float64(idx) * width
+
+	ring := NewRing(len(a.remotePeers))
+	idxes := ring.Slice(offset, width)
+	a.p2c = leastloaded.New()
+
+	for _, idx := range idxes {
+		weight := ring.Weight(idx, offset, width)
+		a.p2c.Add(a.remotePeers[idx], weight)
+	}
+}
+
+func dApertureWidth(localWidth, remoteWidth float64, logicalAperture int) float64 {
+	unitWidth := localWidth
+	unitAperture := float64(logicalAperture) * remoteWidth
+	n := math.Ceil(unitAperture / unitWidth)
+	width := n * unitWidth
+
+	return math.Min(floatOne, width)
+}

aperture/aperture.go

@@ -0,0 +1,95 @@
+package aperture_test
+
+import (
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/hnlq715/go-loadbalance/aperture"
+	"github.com/stretchr/testify/assert"
+	"google.golang.org/grpc/balancer"
+)
+
+func TestAperture(t *testing.T) {
+	t.Run("0 item", func(t *testing.T) {
+		ll := aperture.New()
+		item, done := ll.Next()
+		done(balancer.DoneInfo{})
+		assert.Nil(t, item)
+	})
+
+	t.Run("1 item", func(t *testing.T) {
+		ll := aperture.New()
+		ll.SetLocalPeers([]string{"1"})
+		ll.SetRemotePeers([]interface{}{"8"})
+		ll.SetLocalPeerID("1")
+
+		item, done := ll.Next()
+		done(balancer.DoneInfo{})
+		assert.Equal(t, "8", item)
+	})
+
+	t.Run("3 items", func(t *testing.T) {
+		ll := aperture.New()
+		ll.SetLocalPeers([]string{"1", "2", "3"})
+		ll.SetRemotePeers([]interface{}{"8", "9", "10"})
+		ll.SetLocalPeerID("1")
+
+		item, done := ll.Next()
+		done(balancer.DoneInfo{})
+		assert.Equal(t, "8", item)
+
+		ll.SetLocalPeerID("2")
+
+		item, done = ll.Next()
+		done(balancer.DoneInfo{})
+		assert.Equal(t, "9", item)
+
+		ll.SetLocalPeerID("3")
+
+		item, done = ll.Next()
+		done(balancer.DoneInfo{})
+		assert.Equal(t, "10", item)
+	})
+
+	t.Run("count", func(t *testing.T) {
+		ll := aperture.New()
+		ll.SetLocalPeers([]string{"1", "2", "3"})
+		ll.SetRemotePeers([]interface{}{"8", "9", "10", "11", "12"})
+		ll.SetLocalPeerID("1")
+		ll.SetLogicalWidth(2)
+
+		countMap := make(map[interface{}]int)
+
+		totalCount := 10000
+		wg := sync.WaitGroup{}
+		wg.Add(totalCount)
+
+		mu := sync.Mutex{}
+		for i := 0; i < totalCount; i++ {
+			go func() {
+				defer wg.Done()
+				item, done := ll.Next()
+				time.Sleep(1 * time.Second)
+				done(balancer.DoneInfo{})
+
+				mu.Lock()
+				countMap[item]++
+				mu.Unlock()
+			}()
+		}
+
+		wg.Wait()
+
+		total := 0
+		for _, count := range countMap {
+			total += count
+		}
+		assert.Less(t, 2990, countMap["8"])
+		assert.Less(t, 2990, countMap["9"])
+		assert.Less(t, 2990, countMap["10"])
+		assert.Less(t, 990, countMap["11"])
+
+		assert.Equal(t, totalCount, total)
+	})
+}

aperture/aperture_test.go

@@ -0,0 +1,87 @@
+package aperture
+
+import (
+	"math"
+)
+
+type Ring struct {
+	size      int
+	unitWidth float64
+}
+
+const (
+	floatOne float64 = 1.0
+	intOne   int     = 1
+)
+
+func NewRing(size int) *Ring {
+	return &Ring{
+		size:      size,
+		unitWidth: floatOne / float64(size),
+	}
+}
+
+func (r *Ring) Range(offset, width float64) int {
+	begin := r.Index(offset)
+	end := r.Index(math.Mod(offset+width, 1.0))
+
+	if begin == end && width > r.unitWidth {
+		return r.size
+	} else if begin == end {
+		return intOne
+	}
+
+	beginWeight := r.Weight(begin, offset, width)
+	endWeight := r.Weight(end, offset, width)
+
+	adjustedBegin := begin
+	if beginWeight <= 0 {
+		adjustedBegin++
+	}
+
+	adjustedEnd := end
+	if endWeight > 0 {
+		adjustedEnd++
+	}
+
+	diff := adjustedEnd - adjustedBegin
+	if diff < 0 {
+		return diff + r.size
+	}
+
+	return diff
+}
+
+func (r *Ring) Slice(offset, width float64) []int {
+	seq := make([]int, 0)
+	i := r.Index(offset)
+	rr := r.Range(offset, width)
+
+	for rr > 0 {
+		idx := i % r.size
+		seq = append(seq, idx)
+		i++
+		rr--
+	}
+
+	return seq
+}
+
+func (r *Ring) Index(offset float64) int {
+	return int(math.Floor(offset*float64(r.size))) % r.size
+}
+
+func (r *Ring) Weight(index int, offset, width float64) float64 {
+	ab := float64(index) * r.unitWidth
+	if ab+1 < offset+width {
+		ab++
+	}
+
+	ae := ab + r.unitWidth
+	return intersect(ab, ae, offset, offset+width) / r.unitWidth
+}
+
+func intersect(b0, e0, b1, e1 float64) float64 {
+	len := math.Min(e0, e1) - math.Max(b0, b1)
+	return math.Max(0, len)
+}

aperture/ring.go

@@ -0,0 +1,69 @@
+package aperture
+
+import (
+	"math"
+	"testing"
+
+	"gotest.tools/assert"
+)
+
+func TestRing(t *testing.T) {
+	r := NewRing(3)
+
+	len := float64(3)
+	width := 1.0 / len
+
+	offset := math.Mod(0*width, 1.0)
+	assert.DeepEqual(t, []int{0}, r.Slice(offset, width))
+
+	offset = math.Mod(1*width, 1.0)
+	assert.DeepEqual(t, []int{1}, r.Slice(offset, width))
+
+	offset = math.Mod(2*width, 1.0)
+	assert.DeepEqual(t, []int{2}, r.Slice(offset, width))
+}
+
+func TestRing1(t *testing.T) {
+	r := NewRing(5)
+
+	len := float64(3)
+	width := 1.0 / len
+
+	offset := math.Mod(0*width, 1.0)
+	assert.DeepEqual(t, []int{0, 1}, r.Slice(offset, width))
+
+	offset = math.Mod(1*width, 1.0)
+	assert.DeepEqual(t, []int{1, 2, 3}, r.Slice(offset, width))
+
+	offset = math.Mod(2*width, 1.0)
+	assert.DeepEqual(t, []int{3, 4}, r.Slice(offset, width))
+
+}
+
+func TestRing2(t *testing.T) {
+	r := NewRing(5)
+
+	len := float64(3)
+	width := 1.0 / len
+
+	offset := float64(0) * width
+	assert.Equal(t, float64(10), math.Round(r.Weight(0, offset, width)*10))
+	assert.Equal(t, float64(7), math.Round(r.Weight(1, offset, width)*10))
+	assert.Equal(t, float64(0), math.Round(r.Weight(2, offset, width)*10))
+	assert.Equal(t, float64(0), math.Round(r.Weight(3, offset, width)*10))
+	assert.Equal(t, float64(0), math.Round(r.Weight(4, offset, width)*10))
+
+	offset = float64(1) * width
+	assert.Equal(t, float64(0), math.Round(r.Weight(0, offset, width)*10))
+	assert.Equal(t, float64(3), math.Round(r.Weight(1, offset, width)*10))
+	assert.Equal(t, float64(10), math.Round(r.Weight(2, offset, width)*10))
+	assert.Equal(t, float64(3), math.Round(r.Weight(3, offset, width)*10))
+	assert.Equal(t, float64(0), math.Round(r.Weight(5, offset, width)*10))
+
+	offset = float64(2) * width
+	assert.Equal(t, float64(0), math.Round(r.Weight(0, offset, width)*10))
+	assert.Equal(t, float64(0), math.Round(r.Weight(1, offset, width)*10))
+	assert.Equal(t, float64(0), math.Round(r.Weight(2, offset, width)*10))
+	assert.Equal(t, float64(7), math.Round(r.Weight(3, offset, width)*10))
+	assert.Equal(t, float64(10), math.Round(r.Weight(4, offset, width)*10))
+}

aperture/ring_test.go

@@ -0,0 +1,14 @@
+module github.com/hnlq715/go-loadbalance
+
+go 1.14
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/golang/protobuf v1.4.2 // indirect
+	github.com/kr/pretty v0.2.0 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/stretchr/testify v1.6.1
+	google.golang.org/grpc v1.31.0
+	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
+	gotest.tools v2.2.0+incompatible
+)