feat: refactor AllCompleted and ParallelCompleted.

Author: ghosind

all.go

@@ -2,8 +2,6 @@ package async
 
 import (
 	"context"
-	"sync"
-	"sync/atomic"
 )
 
 // All executes the functions asynchronously until all functions have been finished. If some
@@ -98,44 +96,14 @@ func allCompleted(
 	parent context.Context,
 	funcs ...AsyncFn,
 ) ([][]any, error) {
-	validateAsyncFuncs(funcs...)
-
-	out := make([][]any, len(funcs))
-	if len(funcs) == 0 {
-		return out, nil
-	}
-
-	parent = getContext(parent)
-
-	errs := make([]error, len(funcs))
-	errNum := atomic.Int32{}
-
-	wg := sync.WaitGroup{}
-	wg.Add(len(funcs))
-
-	for i := 0; i < len(funcs); i++ {
-		go func(n int) {
-			fn := funcs[n]
-
-			childCtx, childCanFunc := context.WithCancel(parent)
-			defer childCanFunc()
-			defer wg.Done()
-
-			ret, err := invokeAsyncFn(fn, childCtx, nil)
-			if err != nil {
-				errNum.Add(1)
-				errs[n] = err
-			}
-			out[n] = ret
-		}(i)
-	}
-
-	wg.Wait()
-	if errNum.Load() == 0 {
-		return out, nil
-	}
+	paralleler := builtinPool.Get().(*Paralleler)
+	defer func() {
+		builtinPool.Put(paralleler)
+	}()
 
-	err := convertErrorListToExecutionErrors(errs, int(errNum.Load()))
+	paralleler.
+		WithContext(parent).
+		Add(funcs...)
 
-	return out, err
+	return paralleler.RunCompleted()
 }

parallel.go

@@ -2,8 +2,6 @@ package async
 
 import (
 	"context"
-	"sync"
-	"sync/atomic"
 )
 
 // Parallel runs the functions asynchronously with the specified concurrency limitation. It will
@@ -95,62 +93,15 @@ func parallelCompleted(
 	concurrency int,
 	funcs ...AsyncFn,
 ) ([][]any, error) {
-	// the number of concurrency should be 0 (no limitation) or greater than 0.
-	if concurrency < 0 {
-		panic(ErrInvalidConcurrency)
-	}
-	validateAsyncFuncs(funcs...)
-
-	out := make([][]any, len(funcs))
-
-	if len(funcs) == 0 {
-		return out, nil
-	}
-
-	ctx := getContext(parent)
-	errs := make([]error, len(funcs))
-	errNum := atomic.Int32{}
-
-	wg := sync.WaitGroup{}
-	wg.Add(len(funcs))
-
-	var conch chan empty // channel for concurrency limit
-	// no concurrency limitation if the value of the number is 0
-	if concurrency > 0 {
-		conch = make(chan empty, concurrency)
-	}
-
-	for i := 0; i < len(funcs); i++ {
-		if conch != nil {
-			conch <- empty{}
-		}
-
-		go func(n int) {
-			defer wg.Done()
-
-			fn := funcs[n]
-			childCtx, childCanFunc := context.WithCancel(ctx)
-			defer childCanFunc()
-
-			ret, err := invokeAsyncFn(fn, childCtx, nil)
-			if err != nil {
-				errs[n] = err
-				errNum.Add(1)
-			}
-			out[n] = ret
-
-			if conch != nil {
-				<-conch
-			}
-		}(i)
-	}
-
-	wg.Wait()
-	if errNum.Load() == 0 {
-		return out, nil
-	}
+	paralleler := builtinPool.Get().(*Paralleler)
+	defer func() {
+		builtinPool.Put(paralleler)
+	}()
 
-	err := convertErrorListToExecutionErrors(errs, int(errNum.Load()))
+	paralleler.
+		WithContext(parent).
+		WithConcurrency(concurrency).
+		Add(funcs...)
 
-	return out, err
+	return paralleler.RunCompleted()
 }