Merge pull request #174 from tylertreat/gosched_changes

RM-21797 Remove Gosched call limiting

queue/ring.go

@@ -87,7 +87,6 @@ func (rb *RingBuffer) Offer(item interface{}) (bool, error) {
 func (rb *RingBuffer) put(item interface{}, offer bool) (bool, error) {
 	var n *node
 	pos := atomic.LoadUint64(&rb.queue)
-	i := 0
 L:
 	for {
 		if atomic.LoadUint64(&rb.disposed) == 1 {
@@ -111,12 +110,7 @@ L:
 			return false, nil
 		}
 
-		if i == 10000 {
-			runtime.Gosched() // free up the cpu before the next iteration
-			i = 0
-		} else {
-			i++
-		}
+		runtime.Gosched() // free up the cpu before the next iteration
 	}
 
 	n.data = item
@@ -141,7 +135,6 @@ func (rb *RingBuffer) Poll(timeout time.Duration) (interface{}, error) {
 	var (
 		n     *node
 		pos   = atomic.LoadUint64(&rb.dequeue)
-		i     = 0
 		start time.Time
 	)
 	if timeout > 0 {
@@ -170,12 +163,7 @@ L:
 			return nil, ErrTimeout
 		}
 
-		if i == 10000 {
-			runtime.Gosched() // free up the cpu before the next iteration
-			i = 0
-		} else {
-			i++
-		}
+		runtime.Gosched() // free up the cpu before the next iteration
 	}
 	data := n.data
 	n.data = nil

queue/ring_test.go

@@ -328,6 +328,44 @@ func BenchmarkRBLifeCycle(b *testing.B) {
 	wg.Wait()
 }
 
+func BenchmarkRBLifeCycleContention(b *testing.B) {
+	rb := NewRingBuffer(64)
+
+	var wwg sync.WaitGroup
+	var rwg sync.WaitGroup
+	wwg.Add(10)
+	rwg.Add(10)
+
+	for i := 0; i < 10; i++ {
+		go func() {
+			for {
+				_, err := rb.Get()
+				if err == ErrDisposed {
+					rwg.Done()
+					return
+				} else {
+					assert.Nil(b, err)
+				}
+			}
+		}()
+	}
+
+	b.ResetTimer()
+
+	for i := 0; i < 10; i++ {
+		go func() {
+			for j := 0; j < b.N; j++ {
+				rb.Put(j)
+			}
+			wwg.Done()
+		}()
+	}
+
+	wwg.Wait()
+	rb.Dispose()
+	rwg.Wait()
+}
+
 func BenchmarkRBPut(b *testing.B) {
 	rb := NewRingBuffer(uint64(b.N))