Merge pull request #307 from panjf2000/dev

ver: release 2.9.0

Author: panjf2000

.github/pull_request_template.md

@@ -1,13 +1,5 @@
----
-name: Pull request
-about: Propose changes to the code
-title: ''
-labels: ''
-assignees: ''
----
-
 <!--
-Thank you for contributing to `ants`! Please fill this out to help us make the most of your pull request.
+Thank you for contributing to `ants`! Please fill this out to help us review your pull request more efficiently.
 
 Was this change discussed in an issue first? That can help save time in case the change is not a good fit for the project. Not all pull requests get merged.
 

ants.go

@@ -66,6 +66,12 @@ var (
 	// ErrTimeout will be returned after the operations timed out.
 	ErrTimeout = errors.New("operation timed out")
 
+	// ErrInvalidPoolIndex will be returned when trying to retrieve a pool with an invalid index.
+	ErrInvalidPoolIndex = errors.New("invalid pool index")
+
+	// ErrInvalidLoadBalancingStrategy will be returned when trying to create a MultiPool with an invalid load-balancing strategy.
+	ErrInvalidLoadBalancingStrategy = errors.New("invalid load-balancing strategy")
+
 	// workerChanCap determines whether the channel of a worker should be a buffered channel
 	// to get the best performance. Inspired by fasthttp at
 	// https://github.com/valyala/fasthttp/blob/master/workerpool.go#L139

ants_benchmark_test.go

@@ -25,6 +25,7 @@ package ants
 import (
 	"runtime"
 	"sync"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -47,18 +48,22 @@ func demoPoolFunc(args interface{}) {
 	time.Sleep(time.Duration(n) * time.Millisecond)
 }
 
+var stopLongRunningFunc int32
+
 func longRunningFunc() {
-	for {
+	for atomic.LoadInt32(&stopLongRunningFunc) == 0 {
 		runtime.Gosched()
 	}
 }
 
+var stopLongRunningPoolFunc int32
+
 func longRunningPoolFunc(arg interface{}) {
 	if ch, ok := arg.(chan struct{}); ok {
 		<-ch
 		return
 	}
-	for {
+	for atomic.LoadInt32(&stopLongRunningPoolFunc) == 0 {
 		runtime.Gosched()
 	}
 }
@@ -133,6 +138,24 @@ func BenchmarkAntsPool(b *testing.B) {
 	}
 }
 
+func BenchmarkAntsMultiPool(b *testing.B) {
+	var wg sync.WaitGroup
+	p, _ := NewMultiPool(10, PoolCap/10, RoundRobin, WithExpiryDuration(DefaultExpiredTime))
+	defer p.ReleaseTimeout(DefaultExpiredTime) //nolint:errcheck
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		wg.Add(RunTimes)
+		for j := 0; j < RunTimes; j++ {
+			_ = p.Submit(func() {
+				demoFunc()
+				wg.Done()
+			})
+		}
+		wg.Wait()
+	}
+}
+
 func BenchmarkGoroutinesThroughput(b *testing.B) {
 	for i := 0; i < b.N; i++ {
 		for j := 0; j < RunTimes; j++ {
@@ -165,3 +188,15 @@ func BenchmarkAntsPoolThroughput(b *testing.B) {
 		}
 	}
 }
+
+func BenchmarkAntsMultiPoolThroughput(b *testing.B) {
+	p, _ := NewMultiPool(10, PoolCap/10, RoundRobin, WithExpiryDuration(DefaultExpiredTime))
+	defer p.ReleaseTimeout(DefaultExpiredTime) //nolint:errcheck
+
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		for j := 0; j < RunTimes; j++ {
+			_ = p.Submit(demoFunc)
+		}
+	}
+}

ants_test.go

@@ -985,3 +985,113 @@ func TestDefaultPoolReleaseTimeout(t *testing.T) {
 	err := ReleaseTimeout(2 * time.Second)
 	assert.NoError(t, err)
 }
+
+func TestMultiPool(t *testing.T) {
+	_, err := NewMultiPool(10, -1, 8)
+	assert.ErrorIs(t, err, ErrInvalidLoadBalancingStrategy)
+
+	mp, err := NewMultiPool(10, 5, RoundRobin)
+	testFn := func() {
+		for i := 0; i < 50; i++ {
+			err = mp.Submit(longRunningFunc)
+			assert.NoError(t, err)
+		}
+		assert.EqualValues(t, mp.Waiting(), 0)
+		_, err = mp.WaitingByIndex(-1)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		_, err = mp.WaitingByIndex(11)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		assert.EqualValues(t, 50, mp.Running())
+		_, err = mp.RunningByIndex(-1)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		_, err = mp.RunningByIndex(11)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		assert.EqualValues(t, 0, mp.Free())
+		_, err = mp.FreeByIndex(-1)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		_, err = mp.FreeByIndex(11)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		assert.EqualValues(t, 50, mp.Cap())
+		assert.False(t, mp.IsClosed())
+		for i := 0; i < 10; i++ {
+			n, _ := mp.WaitingByIndex(i)
+			assert.EqualValues(t, 0, n)
+			n, _ = mp.RunningByIndex(i)
+			assert.EqualValues(t, 5, n)
+			n, _ = mp.FreeByIndex(i)
+			assert.EqualValues(t, 0, n)
+		}
+		atomic.StoreInt32(&stopLongRunningFunc, 1)
+		assert.NoError(t, mp.ReleaseTimeout(3*time.Second))
+		assert.Zero(t, mp.Running())
+		assert.True(t, mp.IsClosed())
+		atomic.StoreInt32(&stopLongRunningFunc, 0)
+	}
+	testFn()
+
+	mp.Reboot()
+	testFn()
+
+	mp, err = NewMultiPool(10, 5, LeastTasks)
+	testFn()
+
+	mp.Reboot()
+	testFn()
+
+	mp.Tune(10)
+}
+
+func TestMultiPoolWithFunc(t *testing.T) {
+	_, err := NewMultiPoolWithFunc(10, -1, longRunningPoolFunc, 8)
+	assert.ErrorIs(t, err, ErrInvalidLoadBalancingStrategy)
+
+	mp, err := NewMultiPoolWithFunc(10, 5, longRunningPoolFunc, RoundRobin)
+	testFn := func() {
+		for i := 0; i < 50; i++ {
+			err = mp.Invoke(i)
+			assert.NoError(t, err)
+		}
+		assert.EqualValues(t, mp.Waiting(), 0)
+		_, err = mp.WaitingByIndex(-1)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		_, err = mp.WaitingByIndex(11)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		assert.EqualValues(t, 50, mp.Running())
+		_, err = mp.RunningByIndex(-1)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		_, err = mp.RunningByIndex(11)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		assert.EqualValues(t, 0, mp.Free())
+		_, err = mp.FreeByIndex(-1)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		_, err = mp.FreeByIndex(11)
+		assert.ErrorIs(t, err, ErrInvalidPoolIndex)
+		assert.EqualValues(t, 50, mp.Cap())
+		assert.False(t, mp.IsClosed())
+		for i := 0; i < 10; i++ {
+			n, _ := mp.WaitingByIndex(i)
+			assert.EqualValues(t, 0, n)
+			n, _ = mp.RunningByIndex(i)
+			assert.EqualValues(t, 5, n)
+			n, _ = mp.FreeByIndex(i)
+			assert.EqualValues(t, 0, n)
+		}
+		atomic.StoreInt32(&stopLongRunningPoolFunc, 1)
+		assert.NoError(t, mp.ReleaseTimeout(3*time.Second))
+		assert.Zero(t, mp.Running())
+		assert.True(t, mp.IsClosed())
+		atomic.StoreInt32(&stopLongRunningPoolFunc, 0)
+	}
+	testFn()
+
+	mp.Reboot()
+	testFn()
+
+	mp, err = NewMultiPoolWithFunc(10, 5, longRunningPoolFunc, LeastTasks)
+	testFn()
+
+	mp.Reboot()
+	testFn()
+
+	mp.Tune(10)
+}