feat: add ParallelComplete function.

Author: ghosind

parallel.go

@@ -2,6 +2,7 @@ package async
 
 import (
 	"context"
+	"sync"
 
 	"github.com/ghosind/utils"
 )
@@ -105,3 +106,86 @@ func runTaskInParallel(
 		}
 	}
 }
+
+// ParallelComplete runs the functions asynchronously with the specified concurrency limitation. It
+// returns an error array and a boolean value to indicate whether any function panics or returns an
+// error, and you can get the error details from the error array by the indices of the functions in
+// the parameter list. It will return until all of the functions are finished.
+//
+// The number of concurrency must be greater than or equal to 0, and it means no concurrency
+// limitation if the number is 0.
+func ParallelComplete(concurrency int, funcs ...AsyncFn) ([]error, bool) {
+	return parallelComplete(context.Background(), concurrency, funcs...)
+}
+
+// ParallelCompleteWithContext runs the functions asynchronously with the specified concurrency
+// limitation and the context. It returns an error array and a boolean value to indicate whether
+// any function panics or returns an error, and you can get the error details from the error array
+// by the indices of the functions in the parameter list. It will return until all of the functions
+// are finished.
+//
+// The number of concurrency must be greater than or equal to 0, and it means no concurrency
+// limitation if the number is 0.
+func ParallelCompleteWithContext(
+	ctx context.Context,
+	concurrency int,
+	funcs ...AsyncFn,
+) ([]error, bool) {
+	return parallelComplete(ctx, concurrency, funcs...)
+}
+
+// parallelComplete runs the functions asynchronously with the specified concurrency until all of
+// the functions are finished.
+func parallelComplete(parent context.Context, concurrency int, funcs ...AsyncFn) ([]error, bool) {
+	// the number of concurrency should be 0 (no limitation) or greater than 0.
+	if concurrency < 0 {
+		panic(ErrInvalidConcurrency)
+	}
+
+	errs := make([]error, len(funcs))
+	hasError := false
+
+	if len(funcs) == 0 {
+		return errs, hasError
+	}
+
+	ctx := getContext(parent)
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(funcs))
+
+	var conch chan struct{} // channel for concurrency limit
+	// no concurrency limitation if the value of the number is 0
+	if concurrency > 0 {
+		conch = make(chan struct{}, concurrency)
+	}
+
+	for i := 0; i < len(funcs); i++ {
+		go func(n int) {
+			defer wg.Done()
+
+			if conch != nil {
+				conch <- struct{}{}
+			}
+
+			fn := funcs[n]
+			childCtx, childCanFunc := context.WithCancel(ctx)
+			defer childCanFunc()
+
+			if err := utils.Try(func() error {
+				return fn(childCtx)
+			}); err != nil {
+				errs[n] = err
+				hasError = true
+			}
+
+			if conch != nil {
+				<-conch
+			}
+		}(i)
+	}
+
+	wg.Wait()
+
+	return errs, hasError
+}

parallel_test.go

@@ -137,3 +137,149 @@ func TestParallelWithTimedOutContext(t *testing.T) {
 	}
 	a.EqualNow(finishedNum, 2)
 }
+
+func TestParallelComplete(t *testing.T) {
+	a := assert.New(t)
+
+	errs, hasError := ParallelComplete(0)
+	a.NotTrueNow(hasError)
+	a.EqualNow(errs, []error{})
+
+	a.PanicNow(func() {
+		ParallelComplete(-1)
+	})
+}
+
+func TestParallelCompleteWithoutConcurrencyLimit(t *testing.T) {
+	a := assert.New(t)
+
+	funcs := make([]AsyncFn, 0, 5)
+	for i := 0; i < 5; i++ {
+		funcs = append(funcs, func(ctx context.Context) error {
+			time.Sleep(100 * time.Millisecond)
+			return nil
+		})
+	}
+
+	start := time.Now()
+	errs, hasError := ParallelComplete(0, funcs...)
+	dur := time.Since(start)
+	a.NotTrueNow(hasError)
+	a.EqualNow(errs, []error{nil, nil, nil, nil, nil})
+	a.TrueNow(dur-100*time.Millisecond < 30*time.Millisecond) // allow 30ms deviation
+}
+
+func TestParallelCompleteWithConcurrencyLimit(t *testing.T) {
+	a := assert.New(t)
+
+	funcs := make([]AsyncFn, 0, 5)
+	for i := 0; i < 5; i++ {
+		funcs = append(funcs, func(ctx context.Context) error {
+			time.Sleep(100 * time.Millisecond)
+			return nil
+		})
+	}
+
+	start := time.Now()
+	errs, hasError := ParallelComplete(2, funcs...)
+	dur := time.Since(start)
+	a.NotTrueNow(hasError)
+	a.EqualNow(errs, []error{nil, nil, nil, nil, nil})
+	a.TrueNow(dur-300*time.Millisecond < 30*time.Millisecond) // allow 30ms deviation
+}
+
+func TestParallelCompleteWithFailedTask(t *testing.T) {
+	a := assert.New(t)
+
+	expectedErr := errors.New("expected error")
+
+	funcs := make([]AsyncFn, 0, 5)
+	for i := 0; i < 5; i++ {
+		n := i
+		funcs = append(funcs, func(ctx context.Context) error {
+			if n == 2 {
+				time.Sleep(50 * time.Millisecond)
+				return expectedErr
+			} else {
+				time.Sleep(100 * time.Millisecond)
+			}
+			return nil
+		})
+	}
+
+	errs, hasError := ParallelComplete(0, funcs...)
+	a.TrueNow(hasError)
+	a.EqualNow(errs, []error{nil, nil, expectedErr, nil, nil})
+}
+
+func TestParallelCompleteWithContext(t *testing.T) {
+	a := assert.New(t)
+
+	funcs := make([]AsyncFn, 0, 5)
+	res := make([]bool, 5)
+	for i := 0; i < 5; i++ {
+		n := i
+		funcs = append(funcs, func(ctx context.Context) error {
+			time.Sleep(100 * time.Millisecond)
+			res[n] = true
+			return nil
+		})
+	}
+
+	errs, hasError := ParallelCompleteWithContext(context.Background(), 2, funcs...)
+	a.NotTrueNow(hasError)
+	a.EqualNow(errs, []error{nil, nil, nil, nil, nil})
+
+	finishedNum := 0
+	for _, v := range res {
+		if v {
+			finishedNum++
+		}
+	}
+	a.EqualNow(finishedNum, 5)
+}
+
+func TestParallelCompleteWithTimedOutContext(t *testing.T) {
+	a := assert.New(t)
+
+	timeoutErr := errors.New("timed out")
+
+	funcs := make([]AsyncFn, 0, 5)
+	res := make([]bool, 5)
+	for i := 0; i < 5; i++ {
+		n := i
+		funcs = append(funcs, func(ctx context.Context) error {
+			timer := time.NewTimer(100 * time.Millisecond)
+
+			select {
+			case <-ctx.Done():
+				return timeoutErr
+			case <-timer.C:
+				res[n] = true
+				return nil
+			}
+		})
+	}
+
+	ctx, canFunc := context.WithTimeout(context.Background(), 150*time.Millisecond)
+	defer canFunc()
+
+	errs, hasError := ParallelCompleteWithContext(ctx, 2, funcs...)
+	a.TrueNow(hasError)
+
+	numErrors := 0
+	for _, e := range errs {
+		if errors.Is(e, timeoutErr) {
+			numErrors++
+		}
+	}
+	a.EqualNow(numErrors, 3)
+
+	finishedNum := 0
+	for _, v := range res {
+		if v {
+			finishedNum++
+		}
+	}
+	a.EqualNow(finishedNum, 2)
+}