Merge pull request #79 from marcoferrer/queue-impl-parity

Queue impl parity

Author: platinummonkey

core/dep.go

@@ -21,10 +21,10 @@ const (
 	MetricWindowMinRTT = "window.min_rtt"
 	// MetricWindowQueueSize represents the name of the metric for the Window's Queue Size
 	MetricWindowQueueSize = "window.queue_size"
-	// MetricLifoQueueSize represents the name of the metric for the size of a lifo queue
-	MetricLifoQueueSize = "lifo.queue_size"
-	// MetricLifoQueueLimit represents the name of the metric for the max size of a lifo queue
-	MetricLifoQueueLimit = "lifo.queue_limit"
+	// MetricQueueSize represents the name of the metric for the size of a lifo queue
+	MetricQueueSize = "queue_size"
+	// MetricQueueLimit represents the name of the metric for the max size of a lifo queue
+	MetricQueueLimit = "queue_limit"
 )
 
 // PrefixMetricWithName will prefix a given name with the metric name in the form "<name>.<metric>"

limit/aimd.go

@@ -88,7 +88,7 @@ func (l *AIMDLimit) OnSample(startTime int64, rtt int64, inFlight int, didDrop b
 	l.commonSampler.Sample(rtt, inFlight, didDrop)
 
 	if didDrop {
-		l.limit = int(math.Max(1, math.Min(float64(l.limit-1), float64(int(float64(l.limit)*l.backOffRatio)))))
+		l.limit = int(math.Max(1, math.Min(float64(l.limit-1), float64(l.limit)*l.backOffRatio)))
 		l.notifyListeners(l.limit)
 	} else if inFlight >= l.limit {
 		l.limit += l.increaseBy

limiter/fifo_blocking.go

@@ -1,218 +1,43 @@
 package limiter
 
 import (
-	"container/list"
-	"context"
-	"fmt"
-	"sync"
 	"time"
 
 	"github.com/platinummonkey/go-concurrency-limits/core"
 )
 
-type fifoElement struct {
-	ctx         context.Context
-	releaseChan chan core.Listener
-}
-
-func (e *fifoElement) setListener(listener core.Listener) {
-	select {
-	case e.releaseChan <- listener:
-		// noop
-	default:
-		// timeout has expired
-	}
-}
-
-type fifoQueue struct {
-	q  *list.List
-	mu sync.RWMutex
-}
-
-func (q *fifoQueue) len() int {
-	q.mu.RLock()
-	defer q.mu.RUnlock()
-	return q.q.Len()
-}
-
-func (q *fifoQueue) push(ctx context.Context) (*list.Element, chan core.Listener) {
-	q.mu.Lock()
-	defer q.mu.Unlock()
-	releaseChan := make(chan core.Listener, 1)
-	fe := fifoElement{
-		ctx:         ctx,
-		releaseChan: releaseChan,
-	}
-	e := q.q.PushBack(fe)
-	return e, releaseChan
-}
-
-func (q *fifoQueue) pop() *fifoElement {
-	q.mu.Lock()
-	defer q.mu.Unlock()
-	e := q.q.Front()
-	if e != nil {
-		fe := e.Value.(fifoElement)
-		q.q.Remove(e)
-		return &fe
-	}
-	return nil
-}
-
-func (q *fifoQueue) peek() *fifoElement {
-	q.mu.RLock()
-	defer q.mu.RUnlock()
-	e := q.q.Front()
-	if e != nil {
-		fe := q.q.Front().Value.(fifoElement)
-		return &fe
-	}
-	return nil
-}
-
-func (q *fifoQueue) remove(event *list.Element) {
-	if event == nil {
-		return
-	}
-	q.mu.Lock()
-	defer q.mu.Unlock()
-	q.q.Remove(event)
-}
-
-// FifoBlockingListener implements a blocking listener for the FifoBlockingListener
-type FifoBlockingListener struct {
-	delegateListener core.Listener
-	limiter          *FifoBlockingLimiter
-}
-
-func (l *FifoBlockingListener) unblock() {
-	l.limiter.mu.Lock()
-	defer l.limiter.mu.Unlock()
-	if l.limiter.backlog.len() > 0 {
-		event := l.limiter.backlog.peek()
-		if event != nil {
-			listener, ok := l.limiter.delegate.Acquire(event.ctx)
-			if ok && listener != nil {
-				nextEvent := l.limiter.backlog.pop()
-				nextEvent.setListener(listener)
-			}
-		}
-		// otherwise: still can't acquire the limit.  unblock will be called again next time the limit is released.
-	}
-}
-
-// OnDropped is called to indicate the request failed and was dropped due to being rejected by an external limit or
-// hitting a timeout.  Loss based Limit implementations will likely do an aggressive reducing in limit when this
-// happens.
-func (l *FifoBlockingListener) OnDropped() {
-	l.delegateListener.OnDropped()
-	l.unblock()
-}
-
-// OnIgnore is called to indicate the operation failed before any meaningful RTT measurement could be made and
-// should be ignored to not introduce an artificially low RTT.
-func (l *FifoBlockingListener) OnIgnore() {
-	l.delegateListener.OnIgnore()
-	l.unblock()
-}
-
-// OnSuccess is called as a notification that the operation succeeded and internally measured latency should be
-// used as an RTT sample.
-func (l *FifoBlockingListener) OnSuccess() {
-	l.delegateListener.OnSuccess()
-	l.unblock()
-}
-
 // FifoBlockingLimiter implements a Limiter that blocks the caller when the limit has been reached.  This strategy
 // ensures the resource is properly protected but favors availability over latency by not fast failing requests when
 // the limit has been reached.  To help keep success latencies low and minimize timeouts any blocked requests are
 // processed in last in/first out order.
 //
 // Use this limiter only when the concurrency model allows the limiter to be blocked.
+// Deprecated in favor of QueueBlockingLimiter
 type FifoBlockingLimiter struct {
-	delegate          core.Limiter
-	maxBacklogSize    int
-	maxBacklogTimeout time.Duration
-
-	backlog fifoQueue
-	c       *sync.Cond
-	mu      sync.RWMutex
+	*QueueBlockingLimiter
 }
 
 // NewFifoBlockingLimiter will create a new FifoBlockingLimiter
+// Deprecated, use NewQueueBlockingLimiterFromConfig instead
 func NewFifoBlockingLimiter(
 	delegate core.Limiter,
 	maxBacklogSize int,
 	maxBacklogTimeout time.Duration,
 ) *FifoBlockingLimiter {
-	if maxBacklogSize <= 0 {
-		maxBacklogSize = 100
-	}
-	if maxBacklogTimeout == 0 {
-		maxBacklogTimeout = time.Millisecond * 1000
-	}
-	mu := sync.Mutex{}
+
 	return &FifoBlockingLimiter{
-		delegate:          delegate,
-		maxBacklogSize:    maxBacklogSize,
-		maxBacklogTimeout: maxBacklogTimeout,
-		backlog: fifoQueue{
-			q: list.New(),
-		},
-		c: sync.NewCond(&mu),
+		NewQueueBlockingLimiterFromConfig(delegate, QueueLimiterConfig{
+			Ordering:          OrderingFIFO,
+			MaxBacklogSize:    maxBacklogSize,
+			MaxBacklogTimeout: maxBacklogTimeout,
+		}),
 	}
 }
 
 // NewFifoBlockingLimiterWithDefaults will create a new FifoBlockingLimiter with default values.
+// Deprecated, use NewQueueBlockingLimiterWithDefaults instead
 func NewFifoBlockingLimiterWithDefaults(
 	delegate core.Limiter,
 ) *FifoBlockingLimiter {
 	return NewFifoBlockingLimiter(delegate, 100, time.Millisecond*1000)
 }
-
-func (l *FifoBlockingLimiter) tryAcquire(ctx context.Context) core.Listener {
-	// Try to acquire a token and return immediately if successful
-	listener, ok := l.delegate.Acquire(ctx)
-	if ok && listener != nil {
-		return listener
-	}
-
-	// Restrict backlog size so the queue doesn't grow unbounded during an outage
-	if l.backlog.len() >= l.maxBacklogSize {
-		return nil
-	}
-
-	// Create a holder for a listener and block until a listener is released by another
-	// operation.  Holders will be unblocked in FIFO order
-	event, eventReleaseChan := l.backlog.push(ctx)
-	select {
-	case listener = <-eventReleaseChan:
-		return listener
-	case <-time.After(l.maxBacklogTimeout):
-		// Remove the holder from the backlog.  This item is likely to be at the end of the
-		// list so do a remove to minimize the number of items to traverse
-		l.backlog.remove(event)
-		return nil
-	}
-}
-
-// Acquire a token from the limiter.  Returns an Optional.empty() if the limit has been exceeded.
-// If acquired the caller must call one of the Listener methods when the operation has been completed to release
-// the count.
-//
-// context Context for the request. The context is used by advanced strategies such as LookupPartitionStrategy.
-func (l *FifoBlockingLimiter) Acquire(ctx context.Context) (core.Listener, bool) {
-	delegateListener := l.tryAcquire(ctx)
-	if delegateListener == nil {
-		return nil, false
-	}
-	return &FifoBlockingListener{
-		delegateListener: delegateListener,
-		limiter:          l,
-	}, true
-}
-
-func (l *FifoBlockingLimiter) String() string {
-	return fmt.Sprintf("FifoBlockingLimiter{delegate=%v, maxBacklogSize=%d, maxBacklogTimeout=%v}",
-		l.delegate, l.maxBacklogSize, l.maxBacklogTimeout)
-}

limiter/fifo_blocking_test.go

@@ -1,7 +1,6 @@
 package limiter
 
 import (
-	"container/list"
 	"context"
 	"strings"
 	"sync"
@@ -14,88 +13,6 @@ import (
 	"github.com/platinummonkey/go-concurrency-limits/strategy"
 )
 
-type testFifoQueueContextKey int
-
-func TestFifoQueue(t *testing.T) {
-	t.Parallel()
-	asrt := assert.New(t)
-	q := fifoQueue{
-		q: list.New(),
-	}
-
-	asrt.Equal(0, q.len())
-	el := q.peek()
-	asrt.Nil(el)
-	asrt.Nil(q.pop())
-
-	ctx1 := context.WithValue(context.Background(), testFifoQueueContextKey(1), 1)
-	q.push(ctx1)
-
-	el = q.peek()
-	asrt.Equal(1, q.len())
-	asrt.NotNil(el)
-	asrt.Equal(ctx1, el.ctx)
-
-	// add a 2nd
-	ctx2 := context.WithValue(context.Background(), testFifoQueueContextKey(2), 2)
-	q.push(ctx2)
-
-	// make sure it's still FIFO
-	el = q.peek()
-	asrt.Equal(2, q.len())
-	asrt.NotNil(el)
-	asrt.Equal(ctx1, el.ctx)
-
-	// pop off
-	el = q.pop()
-	asrt.NotNil(el)
-	asrt.Equal(ctx1, el.ctx)
-
-	// check that we only have one again
-	el = q.peek()
-	asrt.Equal(1, q.len())
-	asrt.NotNil(el)
-	asrt.Equal(ctx2, el.ctx)
-
-	// add a 2nd & 3rd
-	ctx3 := context.WithValue(context.Background(), testFifoQueueContextKey(3), 3)
-	el3, _ := q.push(ctx3)
-	ctx4 := context.WithValue(context.Background(), testFifoQueueContextKey(4), 4)
-	q.push(ctx4)
-
-	// remove the middle
-	q.remove(el3)
-	el = q.peek()
-	asrt.Equal(2, q.len())
-	asrt.NotNil(el)
-	asrt.Equal(ctx2, el.ctx)
-	asrt.Equal(ctx4, q.q.Back().Value.(fifoElement).ctx)
-}
-
-func TestFifoBlockingListener(t *testing.T) {
-	t.Parallel()
-	delegateLimiter, _ := NewDefaultLimiterWithDefaults(
-		"",
-		strategy.NewSimpleStrategy(20),
-		limit.NoopLimitLogger{},
-		core.EmptyMetricRegistryInstance,
-	)
-	limiter := NewFifoBlockingLimiterWithDefaults(delegateLimiter)
-	delegateListener, _ := delegateLimiter.Acquire(context.Background())
-	listener := FifoBlockingListener{
-		delegateListener: delegateListener,
-		limiter:          limiter,
-	}
-	listener.OnSuccess()
-	listener.OnIgnore()
-	listener.OnDropped()
-}
-
-type acquiredListenerFifo struct {
-	id       int
-	listener core.Listener
-}
-
 func TestFifoBlockingLimiter(t *testing.T) {
 	t.Parallel()
 
@@ -110,7 +27,7 @@ func TestFifoBlockingLimiter(t *testing.T) {
 		)
 		limiter := NewFifoBlockingLimiterWithDefaults(delegateLimiter)
 		asrt.NotNil(limiter)
-		asrt.True(strings.Contains(limiter.String(), "FifoBlockingLimiter{delegate=DefaultLimiter{"))
+		asrt.True(strings.Contains(limiter.String(), "QueueBlockingLimiter{delegate=DefaultLimiter{"))
 	})
 
 	t.Run("NewFifoBlockingLimiter", func(t2 *testing.T) {
@@ -124,7 +41,7 @@ func TestFifoBlockingLimiter(t *testing.T) {
 		)
 		limiter := NewFifoBlockingLimiter(delegateLimiter, -1, 0)
 		asrt.NotNil(limiter)
-		asrt.True(strings.Contains(limiter.String(), "FifoBlockingLimiter{delegate=DefaultLimiter{"))
+		asrt.True(strings.Contains(limiter.String(), "QueueBlockingLimiter{delegate=DefaultLimiter{"))
 	})
 
 	t.Run("Acquire", func(t2 *testing.T) {