Merge pull request #25 from platinummonkey/fixes-2

Fixes bug for blocking limiter with timeout = 0, adds testing for pools

Author: platinummonkey

.gitmodules

@@ -1,2 +1,2 @@
-// Package go_concurrency_limits provides primitives for concurrency control in complex systems.
-package go_concurrency_limits
+// Package provides primitives for concurrency control in complex systems.
+package main

doc.go

@@ -22,7 +22,7 @@ type contextKey string
 const testContextKey contextKey = "jobID"
 
 type resource struct {
-	counter  *int64
+	counter *int64
 }
 
 func (r *resource) poll(ctx context.Context) (bool, error) {
@@ -112,7 +112,7 @@ func main() {
 	wg := sync.WaitGroup{}
 
 	// spin up 10*l consumers
-	wg.Add(10*l)
+	wg.Add(10 * l)
 	for i := 0; i < 10*l; i++ {
 		go func(c int) {
 			for i := 0; i < 5; i++ {

examples/example_concurrent_loading/main.go

@@ -21,7 +21,6 @@ type StreamServerResponseClassifier func(
 	ctx context.Context, req interface{}, info *golangGrpc.StreamServerInfo, err error,
 ) ResponseType
 
-
 func defaultStreamClientResponseClassifier(
 	ctx context.Context,
 	req interface{},
@@ -47,15 +46,15 @@ func defaultStreamServerResponseClassifier(
 }
 
 type streamInterceptorConfig struct {
-	recvName                string
-	sendName                string
-	tags                    []string
-	recvLimiter             core.Limiter
-	sendLimiter             core.Limiter
+	recvName                            string
+	sendName                            string
+	tags                                []string
+	recvLimiter                         core.Limiter
+	sendLimiter                         core.Limiter
 	recvLimitExceededResponseClassifier LimitExceededResponseClassifier
 	sendLimitExceededResponseClassifier LimitExceededResponseClassifier
-	serverResponseClassifer StreamServerResponseClassifier
-	clientResponseClassifer StreamClientResponseClassifier
+	serverResponseClassifer             StreamServerResponseClassifier
+	clientResponseClassifer             StreamClientResponseClassifier
 }
 
 // StreamInterceptorOption represents an option that can be passed to the stream
@@ -157,4 +156,3 @@ func WithStreamServerResponseTypeClassifier(classifier StreamServerResponseClass
 		cfg.serverResponseClassifer = classifier
 	}
 }
-

grpc/option_streaming.go

@@ -3,11 +3,11 @@ package limiter
 import (
 	"context"
 	"fmt"
-	"github.com/platinummonkey/go-concurrency-limits/limit"
 	"sync"
 	"time"
 
 	"github.com/platinummonkey/go-concurrency-limits/core"
+	"github.com/platinummonkey/go-concurrency-limits/limit"
 )
 
 const longBlockingTimeout = time.Hour * 24 * 30 * 12 * 100 // 100 years
@@ -32,26 +32,38 @@ func newTimeoutWaiter(c *sync.Cond, timeout time.Duration) *timeoutWaiter {
 
 func (w *timeoutWaiter) start() {
 	// start two routines, one runner to signal, another blocking to wait and call unblock
+	var wg sync.WaitGroup
+	wg.Add(2)
 	go func() {
+		wg.Done()
 		w.run()
 	}()
 	go func() {
+		wg.Done()
 		w.c.L.Lock()
-		defer w.c.L.Unlock()
 		w.c.Wait()
+		w.c.L.Unlock()
 		w.unblock()
 	}()
+	wg.Wait()
 }
 
 func (w *timeoutWaiter) run() {
+	if w.timeout > 0 {
+		select {
+		case <-w.closerSig:
+			close(w.timeoutSig)
+			return
+		case <-time.After(w.timeout):
+			// call unblock
+			close(w.timeoutSig)
+			return
+		}
+	}
 	select {
 	case <-w.closerSig:
 		close(w.timeoutSig)
 		return
-	case <-time.After(w.timeout):
-		// call unblock
-		close(w.timeoutSig)
-		return
 	}
 }
 
@@ -99,20 +111,21 @@ func NewBlockingLimiter(
 
 // tryAcquire will block when attempting to acquire a token
 func (l *BlockingLimiter) tryAcquire(ctx context.Context) (core.Listener, bool) {
-	l.c.L.Lock()
-	defer l.c.L.Unlock()
 	for {
+		l.c.L.Lock()
 		// if the deadline has passed, fail quickly
 		deadline, deadlineSet := ctx.Deadline()
 		if deadlineSet && time.Now().UTC().After(deadline) {
 			l.logger.Debugf("deadline passed ctx=%v", time.Now().UTC().After(deadline), ctx)
+			l.c.L.Unlock()
 			return nil, false
 		}
 
 		// try to acquire a new token and return immediately if successful
 		listener, ok := l.delegate.Acquire(ctx)
 		if ok && listener != nil {
 			l.logger.Debugf("delegate returned a listener ctx=%v", ctx)
+			l.c.L.Unlock()
 			return listener, true
 		}
 
@@ -130,6 +143,7 @@ func (l *BlockingLimiter) tryAcquire(ctx context.Context) (core.Listener, bool)
 		}
 
 		// block until we timeout
+		l.c.L.Unlock()
 		timeoutWaiter := newTimeoutWaiter(l.c, timeout)
 		timeoutWaiter.start()
 		l.logger.Debugf("Blocking waiting for release or timeout ctx=%v", ctx)

limiter/blocking.go

@@ -34,7 +34,7 @@ func NewDeadlineLimiter(
 		logger:   logger,
 		delegate: delegate,
 		c:        sync.NewCond(&mu),
-		deadline:  deadline,
+		deadline: deadline,
 	}
 }
 
@@ -67,7 +67,6 @@ func (l *DeadlineLimiter) tryAcquire(ctx context.Context) (listener core.Listene
 	}
 }
 
-
 // Acquire a token from the limiter.  Returns `nil, false` 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.
@@ -90,5 +89,3 @@ func (l *DeadlineLimiter) Acquire(ctx context.Context) (listener core.Listener,
 func (l DeadlineLimiter) String() string {
 	return fmt.Sprintf("DeadlineLimiter{delegate=%v}", l.delegate)
 }
-
-

limiter/deadline.go

@@ -3,6 +3,8 @@ package limiter
 import (
 	"context"
 	"strings"
+	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -16,23 +18,22 @@ func TestDeadlineLimiter(t *testing.T) {
 	t.Run("Unblocked", func(t2 *testing.T) {
 		asrt := assert.New(t2)
 		l := limit.NewSettableLimit("test", 10, nil)
-		noopLogger := limit.NoopLimitLogger{}
 		defaultLimiter, err := NewDefaultLimiter(
 			l,
 			defaultMinWindowTime,
 			defaultMaxWindowTime,
 			defaultMinRTTThreshold,
 			defaultWindowSize,
 			strategy.NewSimpleStrategy(10),
-			noopLogger,
+			nil,
 			core.EmptyMetricRegistryInstance,
 		)
 		if !asrt.NoError(err) {
 			asrt.FailNow("")
 		}
 		asrt.NotNil(defaultLimiter)
-		deadline := time.Now().Add(time.Second*15)
-		deadlineLimiter := NewDeadlineLimiter(defaultLimiter, deadline, noopLogger)
+		deadline := time.Now().Add(time.Second * 15)
+		deadlineLimiter := NewDeadlineLimiter(defaultLimiter, deadline, nil)
 		// stringer
 		asrt.True(strings.Contains(deadlineLimiter.String(), "DeadlineLimiter{delegate=DefaultLimiter{"))
 
@@ -71,7 +72,7 @@ func TestDeadlineLimiter(t *testing.T) {
 			asrt.FailNow("")
 		}
 		asrt.NotNil(defaultLimiter)
-		deadline := time.Now().Add(time.Second*1)
+		deadline := time.Now().Add(time.Second * 1)
 		deadlineLimiter := NewDeadlineLimiter(defaultLimiter, deadline, noopLogger)
 
 		i := 1
@@ -94,5 +95,48 @@ func TestDeadlineLimiter(t *testing.T) {
 		asrt.True(deadlineFound, "expected deadline to be reached but not after %d attempts", i)
 		asrt.Equal(2, i, "expected deadline to be exceeded on second attempt")
 	})
-}
 
+	t.Run("limit reached", func(t2 *testing.T) {
+		asrt := assert.New(t2)
+		l := limit.NewFixedLimit("test", 1, nil)
+		noopLogger := limit.BuiltinLimitLogger{}
+		defaultLimiter, err := NewDefaultLimiter(
+			l,
+			defaultMinWindowTime,
+			defaultMaxWindowTime,
+			defaultMinRTTThreshold,
+			defaultWindowSize,
+			strategy.NewSimpleStrategy(1),
+			noopLogger,
+			core.EmptyMetricRegistryInstance,
+		)
+		if !asrt.NoError(err) {
+			asrt.FailNow("")
+		}
+		asrt.NotNil(defaultLimiter)
+		deadline := time.Now().Add(time.Second * 1)
+		deadlineLimiter := NewDeadlineLimiter(defaultLimiter, deadline, noopLogger)
+
+		var deadlineFound atomic.Value
+		deadlineFound.Store(false)
+		var wg sync.WaitGroup
+		wg.Add(3)
+
+		for i := 0; i < 3; i++ {
+			go func(c int) {
+				defer wg.Done()
+				listener, ok := deadlineLimiter.Acquire(context.Background())
+				if ok && listener != nil {
+					time.Sleep(time.Second)
+					listener.OnSuccess()
+				} else {
+					deadlineFound.Store(true)
+				}
+			}(i)
+			time.Sleep(time.Millisecond * 5)
+		}
+		wg.Wait()
+
+		asrt.True(deadlineFound.Load().(bool), "expected deadline limit to be reached but not after 2 attempts")
+	})
+}

limiter/deadline_test.go

@@ -10,7 +10,8 @@ import (
 )
 
 func ExampleFixedPool() {
-	var JobKey = "job_id"
+	type JobKey string
+	var JobKeyID = JobKey("job_id")
 
 	l := 1000 // limit to 1000 concurrent requests.
 	// create a new pool
@@ -35,7 +36,7 @@ func ExampleFixedPool() {
 	for i := 0; i <= l*3; i++ {
 		go func(c int) {
 			defer wg.Done()
-			ctx := context.WithValue(context.Background(), JobKey, c)
+			ctx := context.WithValue(context.Background(), JobKeyID, c)
 			// this will block until timeout or token was acquired.
 			listener, ok := pool.Acquire(ctx)
 			if !ok {

metric_registry/doc.go

@@ -12,7 +12,8 @@ import (
 )
 
 func ExamplePool() {
-	var JobKey = "job_id"
+	type JobKey string
+	var JobKeyID = JobKey("job_id")
 
 	l := 1000 // limit to adjustable 1000 concurrent requests.
 	delegateLimit := limit.NewDefaultAIMLimit(
@@ -54,7 +55,7 @@ func ExamplePool() {
 	for i := 0; i <= l*3; i++ {
 		go func(c int) {
 			defer wg.Done()
-			ctx := context.WithValue(context.Background(), JobKey, c)
+			ctx := context.WithValue(context.Background(), JobKeyID, c)
 			// this will block until timeout or token was acquired.
 			listener, ok := pool.Acquire(ctx)
 			if !ok {

patterns/pool/example_fixed_pool_test.go

@@ -10,7 +10,8 @@ import (
 )
 
 func ExampleLIFOFixedPool() {
-	var JobKey = "job_id"
+	type JobKey string
+	var JobKeyID = JobKey("job_id")
 
 	l := 1000 // limit to 1000 concurrent requests.
 	// create a new pool
@@ -36,7 +37,7 @@ func ExampleLIFOFixedPool() {
 	for i := 0; i <= l*3; i++ {
 		go func(c int) {
 			defer wg.Done()
-			ctx := context.WithValue(context.Background(), JobKey, c)
+			ctx := context.WithValue(context.Background(), JobKeyID, c)
 			// this will block until timeout or token was acquired.
 			listener, ok := pool.Acquire(ctx)
 			if !ok {

patterns/pool/example_generic_pool_test.go

@@ -53,7 +53,7 @@ func NewFixedPool(
 		metricRegistry = core.EmptyMetricRegistryInstance
 	}
 
-	limitStrategy := strategy.NewSimpleStrategy(fixedLimit)
+	limitStrategy := strategy.NewPreciseStrategy(fixedLimit)
 	defaultLimiter, err := limiter.NewDefaultLimiter(
 		limit.NewFixedLimit(
 			name,

patterns/pool/example_lifo_fixed_pool_test.go

@@ -0,0 +1,45 @@
+package pool
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestFixedPool(t *testing.T) {
+	asrt := assert.New(t)
+	p, err := NewFixedPool(
+		"test-fixed-pool",
+		10,
+		-1,
+		-1,
+		-1,
+		-1,
+		0,
+		nil,
+		nil,
+	)
+	asrt.NoError(err)
+
+	asrt.Equal(10, p.Limit())
+
+	var wg sync.WaitGroup
+	for i := 0; i < 20; i++ {
+		wg.Add(1)
+		go func(c int) {
+			defer wg.Done()
+			l, _ := p.Acquire(context.WithValue(context.Background(), testKeyID, fmt.Sprint(c)))
+			log.Printf("acquired now, sleeping - %d\n", c)
+			time.Sleep(time.Millisecond * 100)
+			l.OnSuccess()
+			log.Printf("no longer acquired, released - %d\n", c)
+		}(i)
+		time.Sleep(time.Millisecond * 5)
+	}
+	wg.Wait()
+}