Merge pull request #3 from lxzan/dev

完善单元测试和边界处理

Author: lxzan

go.mod

@@ -3,7 +3,7 @@ module github.com/lxzan/concurrency
 go 1.20
 
 require (
-	github.com/lxzan/dao v1.1.7
+	github.com/lxzan/dao v1.1.12
 	github.com/pkg/errors v0.9.1
 	github.com/stretchr/testify v1.8.4
 )

go.sum

@@ -1,7 +1,7 @@
 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/lxzan/dao v1.1.7 h1:I049e67buJIpr4QJ/vJbHSjKMLN4ZJlSMeK3Rq+CJl8=
-github.com/lxzan/dao v1.1.7/go.mod h1:5ChTIo7RSZ4upqRo16eicJ3XdJWhGwgMIsyuGLMUofM=
+github.com/lxzan/dao v1.1.12 h1:TMvCwhFVzZV6c9upFxXXoiPD5wDKIYgzIYoC5KE//yc=
+github.com/lxzan/dao v1.1.12/go.mod h1:5ChTIo7RSZ4upqRo16eicJ3XdJWhGwgMIsyuGLMUofM=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=

groups/group.go

@@ -3,10 +3,11 @@ package groups
 import (
 	"context"
 	"errors"
-	"github.com/lxzan/concurrency/internal"
 	"sync"
 	"sync/atomic"
 	"time"
+
+	"github.com/lxzan/concurrency/internal"
 )
 
 const (
@@ -149,7 +150,10 @@ func (c *Group[T]) Update(f func()) {
 
 // Start 启动并等待所有任务执行完成
 func (c *Group[T]) Start() error {
-	var taskTotal = int64(c.Len())
+	c.mu.Lock()
+	var taskTotal = c.taskTotal
+	c.mu.Unlock()
+
 	if taskTotal == 0 {
 		return nil
 	}

groups/group_test.go

@@ -1,12 +1,13 @@
 package groups
 
 import (
-	"github.com/pkg/errors"
-	"github.com/stretchr/testify/assert"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/pkg/errors"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestNewTaskGroup(t *testing.T) {
@@ -103,4 +104,136 @@ func TestNewTaskGroup(t *testing.T) {
 		as.Error(err)
 		as.ElementsMatch(arr, []int{1, 3})
 	})
+
+	t.Run("push after start", func(t *testing.T) {
+		// 测试在Start之后Push任务的行为
+		ctl := New[int](WithConcurrency(2))
+		ctl.Push(1, 2)
+		var processed = make([]int, 0)
+		var mu = sync.Mutex{}
+		ctl.OnMessage = func(args int) error {
+			mu.Lock()
+			processed = append(processed, args)
+			mu.Unlock()
+			time.Sleep(10 * time.Millisecond)
+			return nil
+		}
+
+		go func() {
+			time.Sleep(5 * time.Millisecond)
+			ctl.Push(3, 4) // 在Start之后Push
+		}()
+
+		err := ctl.Start()
+		as.NoError(err)
+		// 验证所有任务都被处理
+		as.GreaterOrEqual(len(processed), 2)
+	})
+
+	t.Run("multiple errors", func(t *testing.T) {
+		// 测试多个任务都出错的情况
+		ctl := New[int](WithConcurrency(2))
+		ctl.Push(1, 2, 3)
+		ctl.OnMessage = func(args int) error {
+			return errors.Errorf("error %d", args)
+		}
+		err := ctl.Start()
+		as.Error(err)
+		// 验证错误被正确收集
+		as.Contains(err.Error(), "error")
+	})
+
+	t.Run("concurrent push", func(t *testing.T) {
+		// 测试并发Push
+		ctl := New[int](WithConcurrency(4))
+		var wg sync.WaitGroup
+		for i := 0; i < 10; i++ {
+			wg.Add(1)
+			go func(n int) {
+				defer wg.Done()
+				ctl.Push(n)
+			}(i)
+		}
+		wg.Wait()
+
+		var processed = int64(0)
+		ctl.OnMessage = func(args int) error {
+			atomic.AddInt64(&processed, 1)
+			return nil
+		}
+		err := ctl.Start()
+		as.NoError(err)
+		as.Equal(int64(10), processed)
+	})
+
+	t.Run("zero concurrency", func(t *testing.T) {
+		// 测试并发度为0的情况（应该使用默认值）
+		ctl := New[int](WithConcurrency(0))
+		ctl.Push(1, 2, 3)
+		var processed = int64(0)
+		ctl.OnMessage = func(args int) error {
+			atomic.AddInt64(&processed, 1)
+			return nil
+		}
+		err := ctl.Start()
+		as.NoError(err)
+		as.Equal(int64(3), processed)
+	})
+
+	t.Run("cancel before start", func(t *testing.T) {
+		// 测试在Start之前Cancel
+		ctl := New[int]()
+		ctl.Push(1, 2, 3)
+		ctl.Cancel()
+		var processed = int64(0)
+		ctl.OnMessage = func(args int) error {
+			atomic.AddInt64(&processed, 1)
+			return nil
+		}
+		_ = ctl.Start()
+		// Cancel后任务应该不会执行
+		as.Equal(int64(0), processed)
+		// 由于任务被取消，可能会超时或返回错误
+	})
+
+	t.Run("update deadlock check", func(t *testing.T) {
+		// 测试Update方法不会导致死锁
+		ctl := New[int](WithConcurrency(2))
+		ctl.Push(1, 2, 3, 4, 5)
+		var processed = int64(0)
+		ctl.OnMessage = func(args int) error {
+			ctl.Update(func() {
+				processed++
+			})
+			return nil
+		}
+		err := ctl.Start()
+		as.NoError(err)
+		as.Equal(int64(5), processed)
+	})
+
+	t.Run("task count accuracy", func(t *testing.T) {
+		// 测试任务计数准确性
+		ctl := New[int](WithConcurrency(2))
+		ctl.Push(1, 2, 3, 4, 5)
+		var processed = int64(0)
+		ctl.OnMessage = func(args int) error {
+			atomic.AddInt64(&processed, 1)
+			return nil
+		}
+		err := ctl.Start()
+		as.NoError(err)
+		as.Equal(int64(5), processed)
+	})
+
+	t.Run("len accuracy", func(t *testing.T) {
+		// 测试Len方法的准确性
+		ctl := New[int]()
+		as.Equal(0, ctl.Len())
+		ctl.Push(1, 2, 3)
+		as.Equal(3, ctl.Len())
+		ctl.Start()
+		// Start后队列应该为空
+		as.Equal(0, ctl.Len())
+	})
 }

queues/multiple_queue.go

@@ -53,5 +53,8 @@ func (c *multipleQueue) Stop(ctx context.Context) error {
 		}(c.qs[i])
 	}
 	wg.Wait()
-	return err.Load().err
+	if e := err.Load(); e != nil {
+		return e.err
+	}
+	return nil
 }

queues/queue_test.go

@@ -2,12 +2,13 @@ package queues
 
 import (
 	"context"
-	"github.com/lxzan/concurrency/logs"
-	"github.com/stretchr/testify/assert"
 	"sync"
 	"sync/atomic"
 	"testing"
 	"time"
+
+	"github.com/lxzan/concurrency/logs"
+	"github.com/stretchr/testify/assert"
 )
 
 func TestSingleQueue(t *testing.T) {
@@ -171,4 +172,120 @@ func TestMultiQueue(t *testing.T) {
 		q.Stop(context.Background())
 		q.Push(func() {})
 	})
+
+	t.Run("push after stop", func(t *testing.T) {
+		// 测试Stop后Push的行为
+		q := New(WithConcurrency(2))
+		var processed = int64(0)
+		q.Push(func() {
+			atomic.AddInt64(&processed, 1)
+		})
+		q.Stop(context.Background())
+		as.Equal(int64(1), processed)
+
+		// Stop后Push应该被忽略
+		q.Push(func() {
+			atomic.AddInt64(&processed, 1)
+		})
+		time.Sleep(50 * time.Millisecond)
+		as.Equal(int64(1), processed) // 应该还是1，因为新任务被忽略
+	})
+
+	t.Run("concurrent push", func(t *testing.T) {
+		// 测试并发Push
+		q := New(WithConcurrency(4))
+		var processed = int64(0)
+		var wg sync.WaitGroup
+		for i := 0; i < 100; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				q.Push(func() {
+					atomic.AddInt64(&processed, 1)
+				})
+			}()
+		}
+		wg.Wait()
+		q.Stop(context.Background())
+		as.Equal(int64(100), processed)
+	})
+
+	t.Run("zero concurrency", func(t *testing.T) {
+		// 测试并发度为0的情况
+		q := New(WithConcurrency(0))
+		var processed = int64(0)
+		q.Push(func() {
+			atomic.AddInt64(&processed, 1)
+		})
+		q.Stop(context.Background())
+		as.Equal(int64(1), processed)
+	})
+
+	t.Run("len accuracy", func(t *testing.T) {
+		// 测试Len方法的准确性
+		q := New(WithConcurrency(2))
+		as.Equal(0, q.Len())
+		q.Push(func() {})
+		q.Push(func() {})
+		// 由于有并发执行，Len可能为0或更小
+		time.Sleep(10 * time.Millisecond)
+		q.Stop(context.Background())
+		as.Equal(0, q.Len())
+	})
+
+	t.Run("stop with empty queue", func(t *testing.T) {
+		// 测试空队列Stop
+		q := New()
+		err := q.Stop(context.Background())
+		as.NoError(err)
+	})
+
+	t.Run("stop multiple times", func(t *testing.T) {
+		// 测试多次Stop
+		q := New()
+		err1 := q.Stop(context.Background())
+		err2 := q.Stop(context.Background())
+		as.NoError(err1)
+		as.NoError(err2)
+	})
+
+	t.Run("context cancel during stop", func(t *testing.T) {
+		// 测试Stop时context被取消
+		q := New(WithConcurrency(1))
+		q.Push(func() {
+			time.Sleep(200 * time.Millisecond)
+		})
+		ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
+		defer cancel()
+		err := q.Stop(ctx)
+		as.Error(err)
+	})
+
+	t.Run("multiple queue error handling", func(t *testing.T) {
+		// 测试多队列错误处理
+		q := New(WithSharding(4), WithConcurrency(1), WithTimeout(50*time.Millisecond))
+		// 添加一些会超时的任务
+		for i := 0; i < 8; i++ {
+			q.Push(func() {
+				time.Sleep(200 * time.Millisecond)
+			})
+		}
+		err := q.Stop(context.Background())
+		// 应该返回超时错误
+		as.Error(err)
+	})
+
+	t.Run("multiple queue all success", func(t *testing.T) {
+		// 测试所有队列都成功的情况
+		q := New(WithSharding(4), WithConcurrency(2))
+		var processed = int64(0)
+		for i := 0; i < 20; i++ {
+			q.Push(func() {
+				atomic.AddInt64(&processed, 1)
+			})
+		}
+		err := q.Stop(context.Background())
+		as.NoError(err)
+		as.Equal(int64(20), processed)
+	})
 }