concurrentloop

README

concurrentloop

concurrentloop provides a function to call a function concurrently powered by generics, channels, and goroutines.

Install

$ go get github.com/thalesfsp/concurrentloop

Specific version

Example: $ go get github.com/thalesfsp/concurrentloop@v1.2.0

Usage

See example_test.go, and concurrentloop_test.go file.

Documentation

Run $ make doc or check out online.

Development

Check out CONTRIBUTION.

Release

1. Update CHANGELOG accordingly.
2. Once changes from MR are merged.
3. Tag and release.

Roadmap

Check out CHANGELOG.

Documentation

Overview

Package concurrentloop provides a function to call a function concurrently powered by generics, channels, and goroutines.

Index

• Constants
• func ExecuteCh[T any](ctx context.Context, fns []ExecuteFunc[T]) chan ResultCh[T]
• func Flatten2D[T any](data [][]T) []T
• func RemoveZeroValues[T any](removeZeroValues bool, results []T) []T
• func SplitSlice[T any](items []T, batchSize int) [][]T
• type Errors
  • func Execute[T any](ctx context.Context, fns []ExecuteFunc[T]) ([]T, Errors)
  • func Map[T any, Result any](ctx context.Context, items []T, f MapFunc[T, Result], opts ...Func) ([]Result, Errors)
  • func MapDone[T any, Result any](ctx context.Context, items []T, f MapFuncCh[T, Result], opts ...Func) ([]Result, Errors)
  • func MapM[T any, Result any](ctx context.Context, itemsMap map[string]T, f MapMFunc[T, Result], ...) ([]Result, Errors)
  • func (e Errors) Error() string
• type ExecuteFunc
• type Func
  • func WithBatchSize(concurrency int) Func
  • func WithLimit(limit int) Func
  • func WithRandomDelayTime(minDelay, maxDelay int, d time.Duration) Func
  • func WithRemoveZeroValues(remove bool) Func
• type MapFunc
• type MapFuncCh
• type MapMFunc
• type Option
• type ResultCh

Constants

const Name = "concurrentloop"

Name of the package.

Functions

func ExecuteCh

func ExecuteCh[T any](ctx context.Context, fns []ExecuteFunc[T]) chan ResultCh[T]

ExecuteCh calls the `fns` concurrently.

NOTE: It's the caller's responsibility to close the channel.

func Flatten2D

func Flatten2D[T any](data [][]T) []T

Flatten2D takes a 2D slice and returns a 1D slice containing all the elements.

func RemoveZeroValues

func RemoveZeroValues[T any](removeZeroValues bool, results []T) []T

RemoveZeroValues removes zero values from the results.

func SplitSlice

func SplitSlice[T any](items []T, batchSize int) [][]T

SplitSlice splits a slice into batches of a given size.

Types

type Errors

type Errors []error

Errors is a slice of errors.

func Execute

func Execute[T any](ctx context.Context, fns []ExecuteFunc[T]) ([]T, Errors)

Execute calls the `fns` concurrently, and returns the results and any errors that occurred. The function blocks until all executions have completed.

NOTE: Order is may preserved.

func Map

func Map[T any, Result any](
	ctx context.Context,
	items []T,
	f MapFunc[T, Result],
	opts ...Func,
) ([]Result, Errors)

Map concurrently applies a function `f` to each element in the slice `items` and returns the resulting slice and any errors that occurred. `f` should be of type MapFunc.

func MapDone

func MapDone[T any, Result any](
	ctx context.Context,
	items []T,
	f MapFuncCh[T, Result],
	opts ...Func,
) ([]Result, Errors)

MapDone concurrently applies a function `f` to each element in the slice `items` and returns the resulting slice and any errors that occurred. `f` should be of type MapFunc. It also takes a channel `done` to signal early termination.

The function takes an optional number of `Func` options that allow you to customize the behavior of the function.

If an error occurs during execution of `f`, it is stored and returned along with the results. The order of the results matches the order of the input slice.

If any of the operations are cancelled by the context or through the cancelCh, the function will return immediately. MapDone is a generic concurrent mapping function that processes slices of items in parallel while providing control over concurrency, delays, and result handling.

Flow of the function: 1. Initialize options and concurrency controls 2. Set up error handling and result tracking 3. Configure random delay if specified 4. For each input item:

• Check for early termination signals
• Apply random delay if configured
• Check processing limits
• Acquire semaphore slot
• Launch goroutine to process item:
• Process item with provided function
• Handle errors
• Store results
• Update counters

5. Wait for all processing to complete 6. Return results and any errors

Key Concepts: - Generics: The function uses type parameters (T, Result) to work with any data types - Concurrency: Uses goroutines for parallel processing - Synchronization:

• semaphore: Limits number of concurrent goroutines
• WaitGroup: Tracks completion of all goroutines
• Mutex: Protects shared resources (error slice)

- Context: Handles cancellation and timeouts - Atomic Operations: Thread-safe counting of processed results - Channels: Used for signaling early termination - Error Handling: Collects and returns errors from all goroutines

Type Parameters:

• T: The input type of items to be processed
• Result: The output type after processing each item

Parameters:

• ctx: Context for cancellation and timeout control
• items: Slice of input items to be processed
• f: Function that processes each item and returns a Result
• opts: Optional configuration functions to modify default behavior

Returns:

• []Result: Slice of processed results
• Errors: Any errors encountered during processing

func MapM

func MapM[T any, Result any](
	ctx context.Context,
	itemsMap map[string]T,
	f MapMFunc[T, Result],
	opts ...Func,
) ([]Result, Errors)

MapM concurrently applies a function `f` to each element in the map `itemMaps` and returns the resulting slice and any errors that occurred. `f` should be of type MapMFunc.

func (Errors) Error

func (e Errors) Error() string

Error returns a string representation of the combined errors in the `Errors` slice, separated by commas. This method satisfies the `error` interface.

type ExecuteFunc

type ExecuteFunc[T any] func(context.Context) (T, error)

ExecuteFunc is the type of the function that will be executed concurrently for each element in a slice of type `T`. The function takes a `context.Context` and a value of type `T`, and returns a value of type `Result` and an error value.

type Func

type Func func(o Option) Option

Func allows to specify message's options.

func WithBatchSize

func WithBatchSize(concurrency int) Func

WithBatchSize sets the size of the batch.

func WithLimit

func WithLimit(limit int) Func

WithLimit sets the max amount of results to collect before stopping the loop.

func WithRandomDelayTime

func WithRandomDelayTime(minDelay, maxDelay int, d time.Duration) Func

WithRandomDelayTime sets the random delay time between each iteration.

func WithRemoveZeroValues

func WithRemoveZeroValues(remove bool) Func

WithRemoveZeroValues if set to true removes zero values from the results.

type MapFunc

type MapFunc[T any, Result any] func(ctx context.Context, item T) (Result, error)

MapFunc is the type of the function that will be executed concurrently for each element in a slice of type `T`. The function takes a `context.Context` and a value of type `T`, and returns a value of type `Result` and an error value.

type MapFuncCh

type MapFuncCh[T any, Result any] func(ctx context.Context, item T, done chan<- struct{}) (Result, error)

type MapMFunc

type MapMFunc[T any, Result any] func(ctx context.Context, key string, item T) (Result, error)

MapMFunc is the type of the function that will be executed concurrently for each element in the map.

type Option

type Option struct {
	// BatchSize is the size of the batch.
	BatchSize int

	// The max amount of results to collect before
	Limit int

	// RandomDelayTimeDuration is the unit of the duration (Second, Millisecond, etc.)
	RandomDelayTimeDuration time.Duration

	// RandomDelayTimeMax is the upper limit.
	RandomDelayTimeMax int

	// RandomDelayTimeMin is the lower limit.
	RandomDelayTimeMin int

	// RemoveZeroValues indicates whether to remove zero values from the results.
	RemoveZeroValues bool
}

Option for the concurrent loop.

type ResultCh

type ResultCh[T any] struct {
	Error  error
	Index  int
	Output T
}

ResultCh receives the result from the channel.