feat: add Parallel.

Author: ghosind

error.go

@@ -5,4 +5,7 @@ import "errors"
 var (
 	// ErrContextCanceled to indicate the context was canceled or timed out.
 	ErrContextCanceled error = errors.New("context canceled")
+	// ErrInvalidConcurrency to indicate the number of the concurrency limitation is an invalid
+	// value.
+	ErrInvalidConcurrency error = errors.New("invalid concurrency")
 )

parallel.go

@@ -0,0 +1,107 @@
+package async
+
+import (
+	"context"
+
+	"github.com/ghosind/utils"
+)
+
+// Parallel runs the functions asynchronously with the specified concurrency limitation. It will
+// send a cancel sign to context and terminate immediately if any function returns an error or
+// panic, and also returns the index of the error function in the parameters list.
+//
+// The number of concurrency must be greater than or equal to 0, and it means no concurrency
+// limitation if the number is 0.
+func Parallel(concurrency int, funcs ...AsyncFn) (int, error) {
+	return parallel(context.Background(), concurrency, funcs...)
+}
+
+// ParallelWithContext runs the functions asynchronously with the specified concurrency limitation.
+// It will send a cancel sign to context and terminate immediately if any function returns an error
+// or panic, and also returns the index of the error function in the parameters list. If the
+// context was canceled or timed out before all functions finished executing, it will send a cancel
+// sign to all uncompleted functions, and return the value of the index as -1 with a `context
+// canceled` error.
+//
+// The number of concurrency must be greater than or equal to 0, and it means no concurrency
+// limitation if the number is 0.
+func ParallelWithContext(ctx context.Context, concurrency int, funcs ...AsyncFn) (int, error) {
+	return parallel(ctx, concurrency, funcs...)
+}
+
+// parallel runs the functions asynchronously with the specified concurrency.
+func parallel(parent context.Context, concurrency int, funcs ...AsyncFn) (int, error) {
+	// the number of concurrency should be 0 (no limitation) or greater than 0.
+	if concurrency < 0 {
+		panic(ErrInvalidConcurrency)
+	}
+
+	if len(funcs) == 0 {
+		return -1, nil
+	}
+
+	parent = getContext(parent)
+	ctx, canFunc := context.WithCancel(parent)
+	defer canFunc()
+
+	ch := make(chan executeResult) // channel for result
+	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 runTaskInParallel(ctx, i, funcs[i], conch, ch)
+	}
+
+	finished := 0
+	for finished < len(funcs) {
+		select {
+		case <-parent.Done():
+			return -1, ErrContextCanceled
+		case ret := <-ch:
+			if ret.Error != nil {
+				return ret.Index, ret.Error
+			}
+			finished++
+		}
+	}
+
+	return -1, nil
+}
+
+// runTaskInParallel runs the specified function for Parallel / ParallelWithContext.
+func runTaskInParallel(
+	ctx context.Context,
+	n int,
+	fn AsyncFn,
+	conch chan struct{},
+	ch chan executeResult,
+) {
+	if conch != nil {
+		conch <- struct{}{}
+	}
+
+	childCtx, childCanFunc := context.WithCancel(ctx)
+	defer childCanFunc()
+
+	err := utils.Try(func() error {
+		return fn(childCtx)
+	})
+
+	if conch != nil {
+		<-conch
+	}
+
+	select {
+	case <-ctx.Done():
+		return
+	default:
+		ch <- executeResult{
+			Index: n,
+			Error: err,
+		}
+	}
+}

parallel_test.go

@@ -0,0 +1,139 @@
+package async
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/ghosind/go-assert"
+)
+
+func TestParallel(t *testing.T) {
+	a := assert.New(t)
+
+	index, err := Parallel(0)
+	a.NilNow(err)
+	a.EqualNow(index, -1)
+
+	a.PanicNow(func() {
+		Parallel(-1)
+	})
+}
+
+func TestParallelWithoutConcurrencyLimit(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()
+	index, err := Parallel(0, funcs...)
+	dur := time.Since(start)
+	a.NilNow(err)
+	a.EqualNow(index, -1)
+	a.TrueNow(dur-100*time.Millisecond < 5*time.Millisecond) // allow 5ms deviation
+}
+
+func TestParallelWithConcurrencyLimit(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()
+	index, err := Parallel(2, funcs...)
+	dur := time.Since(start)
+	a.NilNow(err)
+	a.EqualNow(index, -1)
+	a.TrueNow(dur-300*time.Millisecond < 10*time.Millisecond) // allow 10ms deviation
+}
+
+func TestParallelWithFailedTask(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 {
+			time.Sleep(100 * time.Millisecond)
+			if n == 2 {
+				return expectedErr
+			}
+			return nil
+		})
+	}
+
+	index, err := Parallel(2, funcs...)
+	a.EqualNow(err, expectedErr)
+	a.EqualNow(index, 2)
+}
+
+func TestParallelWithContext(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
+		})
+	}
+
+	index, err := ParallelWithContext(context.Background(), 2, funcs...)
+	a.NilNow(err)
+	a.EqualNow(index, -1)
+
+	finishedNum := 0
+	for _, v := range res {
+		if v {
+			finishedNum++
+		}
+	}
+	a.EqualNow(finishedNum, 5)
+}
+
+func TestParallelWithTimedOutContext(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
+		})
+	}
+
+	ctx, canFunc := context.WithTimeout(context.Background(), 150*time.Millisecond)
+	defer canFunc()
+
+	index, err := ParallelWithContext(ctx, 2, funcs...)
+	a.TrueNow(errors.Is(err, ErrContextCanceled))
+	a.EqualNow(index, -1)
+
+	finishedNum := 0
+	for _, v := range res {
+		if v {
+			finishedNum++
+		}
+	}
+	a.EqualNow(finishedNum, 2)
+}