[xcontainer] move DelayQueue to queue

Author: leaxoy

go.mod

@@ -5,12 +5,11 @@ go 1.13
 require (
 	github.com/bwmarrin/snowflake v0.3.0
 	github.com/davecgh/go-spew v1.1.1
-	github.com/go-redsync/redsync v1.4.1
+	github.com/go-redsync/redsync v1.4.2
 	github.com/gofrs/uuid v3.2.0+incompatible
 	github.com/gomodule/redigo v2.0.0+incompatible
 	github.com/google/btree v1.0.0
 	github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e // indirect
-	github.com/smartystreets-prototypes/go-disruptor v0.0.0-20200316140655-c96477fd7a6a
 	github.com/spaolacci/murmur3 v1.1.0
 	github.com/spf13/afero v1.2.2
 	github.com/stretchr/testify v1.5.1

go.sum

@@ -3,8 +3,8 @@ github.com/bwmarrin/snowflake v0.3.0/go.mod h1:NdZxfVWX+oR6y2K0o6qAYv6gIOP9rjG0/
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/go-redsync/redsync v1.4.1 h1:HsNhNnEF+56PJEKQtn3sji97BK0uqJfskOUenoUsBus=
-github.com/go-redsync/redsync v1.4.1/go.mod h1:my8/M5YL986u2jBMtZTLkBIgBsKNNSixJWzWwISH6Uw=
+github.com/go-redsync/redsync v1.4.2 h1:KADEZ2rlaHMZWnlkthQCxfGP+8ZWwJLiSjOYN3mntKA=
+github.com/go-redsync/redsync v1.4.2/go.mod h1:my8/M5YL986u2jBMtZTLkBIgBsKNNSixJWzWwISH6Uw=
 github.com/gofrs/uuid v3.2.0+incompatible h1:y12jRkkFxsd7GpqdSZ+/KCs/fJbqpEXSGd4+jfEaewE=
 github.com/gofrs/uuid v3.2.0+incompatible/go.mod h1:b2aQJv3Z4Fp6yNu3cdSllBxTCLRxnplIgP/c0N/04lM=
 github.com/gomodule/redigo v2.0.0+incompatible h1:K/R+8tc58AaqLkqG2Ol3Qk+DR/TlNuhuh457pBFPtt0=
@@ -22,8 +22,6 @@ github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e h1:fD57ERR4JtEqsWb
 github.com/niemeyer/pretty v0.0.0-20200227124842-a10e7caefd8e/go.mod h1:zD1mROLANZcx1PVRCS0qkT7pwLkGfwJo4zjcN/Tysno=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/smartystreets-prototypes/go-disruptor v0.0.0-20200316140655-c96477fd7a6a h1:mHEYm/fBGwtGwgIW/tlprwUd2syvcue8oosLVuQscic=
-github.com/smartystreets-prototypes/go-disruptor v0.0.0-20200316140655-c96477fd7a6a/go.mod h1:slFCjqF2v0VgmCeB+J4uEy0d7HAgLkgEjVrG0DPO67M=
 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.2.2 h1:5jhuqJyZCZf2JRofRvN/nIFgIWNzPa3/Vz8mYylgbWc=

iter/ints.go iter/numberic.go

@@ -34,3 +34,32 @@ func (s *intSliceIterator) Next() (interface{}, bool) {
 	}
 	return item, true
 }
+
+type Int64Slice types.Int64Slice
+
+func (i Int64Slice) Iter() Iterator {
+	ch := make(chan int64, 1)
+	go func(ch chan int64) {
+		for _, x := range i {
+			ch <- x
+		}
+	}(ch)
+	return &int64SliceIterator{next: ch}
+}
+
+type int64SliceIterator struct {
+	next    chan int64
+	stopped int32
+}
+
+func (s *int64SliceIterator) Next() (interface{}, bool) {
+	if atomic.LoadInt32(&s.stopped) == 1 {
+		return nil, false
+	}
+	item, ok := <-s.next
+	if !ok {
+		atomic.StoreInt32(&s.stopped, 1)
+		return nil, false
+	}
+	return item, true
+}

xcontainer/queue/block_queue.go

@@ -0,0 +1,46 @@
+package queue
+
+import (
+	"time"
+)
+
+type BlockQueue struct {
+	capacity int
+	ch       chan interface{}
+}
+
+func (q *BlockQueue) Capacity() int {
+	return q.capacity
+}
+
+func (q *BlockQueue) Push(x interface{}, timeout time.Duration) bool {
+	t := time.NewTimer(timeout)
+	select {
+	case q.ch <- x:
+		return true
+	case <-t.C:
+		return false
+	}
+}
+
+func (q *BlockQueue) Pop(timeout time.Duration) (interface{}, bool) {
+	t := time.NewTimer(timeout)
+	select {
+	case <-t.C:
+		return nil, false
+	case x := <-q.ch:
+		return x, true
+	}
+}
+
+func (q *BlockQueue) BlockPop() (interface{}, bool) {
+	val, ok := <-q.ch
+	return val, ok
+}
+
+func NewBlockingQueue(n int) *BlockQueue {
+	return &BlockQueue{
+		capacity: n,
+		ch:       make(chan interface{}, n),
+	}
+}

xcontainer/delay_queue/delay_queue.go xcontainer/queue/delay_queue.go

@@ -1,4 +1,4 @@
-package delay_queue
+package queue
 
 import (
 	"math"
@@ -8,13 +8,13 @@ import (
 	"github.com/go-board/x-go/xcontainer/priority_queue"
 )
 
-type Task struct {
+type DelayTask struct {
 	data interface{}
 	at   time.Time
 }
 
-func (t Task) Compare(o types.Comparable) types.Ordering {
-	oo := o.(*Task)
+func (t DelayTask) Compare(o types.Comparable) types.Ordering {
+	oo := o.(*DelayTask)
 	if t.at.Before(oo.at) {
 		return types.OrderingLess
 	}
@@ -24,28 +24,28 @@ func (t Task) Compare(o types.Comparable) types.Ordering {
 	return types.OrderingGreater
 }
 
-func NewTask(v interface{}, at time.Time) *Task {
-	return &Task{data: v, at: at}
+func NewTimedTask(v interface{}, at time.Time) *DelayTask {
+	return &DelayTask{data: v, at: at}
 }
 
-func NewDelayTask(v interface{}, d time.Duration) *Task {
-	return NewTask(v, time.Now().Add(d))
+func NewDelayTask(v interface{}, d time.Duration) *DelayTask {
+	return NewTimedTask(v, time.Now().Add(d))
 }
 
 type DelayQueue struct {
 	q      *priority_queue.PriorityQueue
 	notify chan struct{}
 }
 
-func New() *DelayQueue {
+func NewDelayQueue() *DelayQueue {
 	return &DelayQueue{
 		q:      priority_queue.NewComparablePriorityQueue(false),
 		notify: make(chan struct{}, 1),
 	}
 }
 
 // Push new data into delay queue.
-func (q *DelayQueue) Push(task *Task) {
+func (q *DelayQueue) Push(task *DelayTask) {
 	q.q.Push(task)
 	// this should not block
 	select {
@@ -65,7 +65,7 @@ func (q *DelayQueue) popNearest() (interface{}, time.Duration, bool) {
 	if task == nil {
 		return nil, time.Duration(math.MaxInt64), false
 	}
-	t := task.(*Task)
+	t := task.(*DelayTask)
 	now := time.Now()
 	duration := t.at.Sub(now)
 	if !t.at.After(now) {
@@ -87,6 +87,7 @@ func (q *DelayQueue) BlockPop() interface{} {
 			}
 			return v
 		}
+		// block until timeout or new element pushed
 		select {
 		case <-time.NewTimer(duration).C:
 		case <-q.notify:

xcontainer/queue/queue_test.go

@@ -1,99 +1,25 @@
 package queue
 
 import (
-	"sync/atomic"
 	"testing"
 	"time"
 
-	"github.com/smartystreets-prototypes/go-disruptor"
 	"github.com/stretchr/testify/require"
 )
 
 func TestBlockingQueue(t *testing.T) {
 	t.Run("put", func(t *testing.T) {
 		q := NewBlockingQueue(1)
-		require.Equal(t, true, q.Put(1, time.Second), "should not block when putting first element")
-		require.Equal(t, false, q.Put(2, time.Second), "should block when putting second element")
+		require.Equal(t, true, q.Push(1, time.Second), "should not block when putting first element")
+		require.Equal(t, false, q.Push(2, time.Second), "should block when putting second element")
 	})
 	t.Run("get", func(t *testing.T) {
 		q := NewBlockingQueue(1)
-		require.Equal(t, true, q.Put(1, time.Second), "should not block when putting first element")
-		x, ok := q.Get(time.Second)
+		require.Equal(t, true, q.Push(1, time.Second), "should not block when putting first element")
+		x, ok := q.Pop(time.Second)
 		require.Equal(t, true, ok, "should get first element")
 		require.Equal(t, 1, x, "first element should be 1")
-		_, ok = q.Get(time.Second)
+		_, ok = q.Pop(time.Second)
 		require.Equal(t, false, ok, "should get nothing when get again")
 	})
 }
-
-type testItem struct {
-	i int
-}
-
-const RingBufferCapacity = 1 << 14 // must be a power of 2
-const RingBufferMask = RingBufferCapacity - 1
-
-// this instance will be shared among producers and consumers of this application
-var ringBuffer = [RingBufferCapacity]int{}
-
-type disruptorConsumer struct{}
-
-func (d disruptorConsumer) Consume(lower, upper int64) {
-	for sequence := lower; sequence <= upper; sequence++ {
-
-		_ = ringBuffer[sequence&RingBufferMask] // see performance note on producer sample above
-
-		// handle the incoming message with your application code
-	}
-}
-
-func BenchmarkDisruptor(b *testing.B) {
-	b.Run("put", func(b *testing.B) {
-		writer, _ := disruptor.New(
-			disruptor.WithCapacity(1<<14),
-			disruptor.WithConsumerGroup(disruptorConsumer{}))
-		b.ResetTimer()
-		b.ReportAllocs()
-		i := int64(0)
-		b.RunParallel(func(pb *testing.PB) {
-			for pb.Next() {
-				reservation := writer.Reserve(1)
-				ringBuffer[atomic.AddInt64(&i, 1)&RingBufferMask] = 42 // example of incoming value from a network operation such as HTTP, TCP, UDP, etc.
-				writer.Commit(reservation, reservation)
-			}
-		})
-	})
-}
-
-func BenchmarkChan(b *testing.B) {
-	b.Run("get", func(b *testing.B) {
-		b.ReportAllocs()
-		b.ResetTimer()
-		ch := make(chan testItem, 10000)
-		go func() {
-			for {
-				ch <- testItem{i: 0}
-			}
-		}()
-		b.RunParallel(func(pb *testing.PB) {
-			for pb.Next() {
-				<-ch
-			}
-		})
-	})
-	b.Run("put", func(b *testing.B) {
-		b.ReportAllocs()
-		b.ResetTimer()
-		ch := make(chan testItem, 10000)
-		go func() {
-			for {
-				<-ch
-			}
-		}()
-		b.RunParallel(func(pb *testing.PB) {
-			for pb.Next() {
-				ch <- testItem{i: 0}
-			}
-		})
-	})
-}