parallel

README

Parallel Module

This is a Go module that provides a way to execute tasks concurrently and handle their results in parallel.

Introduction

The module defines a Parallel struct, which is a parallel executor that receives tasks, executes them concurrently, and handles their results. The tasks and the results are typed using Go's generics feature, allowing you to use any type for them.

This module also provides two context structs: ExecuteContext and ResultContext, which carry the relevant information and provide utility methods to the handler functions during execution and result handling, respectively.

Features

• Task execution and result handling in parallel.
• Customizable task execution and result handling logic.
• Shared value map for communication between tasks.
• Execution cancellation support.

Usage

First, import the module:

import "github.com/yourusername/parallel"

Initialization

Create a new Parallel executor by providing execute and result handler functions:

p := parallel.NewParallel(executeHandler, resultHandler)

Here, executeHandler is a function that receives an ExecuteContext and returns a result and an error. And resultHandler is a function that receives a ResultContext.

Customization

You can set the maximum number of concurrent execute handlers that can run in parallel using SetMaxExecuteNum method:

p.SetMaxExecuteNum(10) // Set to 10 concurrent tasks.

Execution

Before executing tasks, you need to call ReadyExecute to initialize the executor:

p.ReadyExecute()

Then, you can add tasks to be executed using Execute method:

p.Execute(target)

To wait for all tasks to complete, use:

p.Wait()

To cancel the execution, use:

p.Cancel()

Example

Here is a simple example of how to use this module:

executeHandler := func(ctx *parallel.ExecuteContext[int, string]) (*string, error) {
	// Do some work with ctx.Target().
	// ...
	result := fmt.Sprintf("Result for target: %v", ctx.Target())
	return &result, nil
}

resultHandler := func(ctx *parallel.ResultContext[int, string]) {
	if ctx.Error() != nil {
		fmt.Println("An error occurred:", ctx.Error())
	} else {
		fmt.Println("Result for target", ctx.Target(), "is", ctx.Result())
	}
}

p := parallel.NewParallel(executeHandler, resultHandler)
p.SetMaxExecuteNum(10)
p.ReadyExecute()

for i := 0; i < 100; i++ {
	p.Execute(i)
}

p.Wait()

In this example, we create a new Parallel executor that performs some work on integers and produces a string as a result. It then waits for all tasks to complete before exiting.

License

This project is licensed under the MIT License - see the LICENSE.md file for details.

Contributing

Please read CONTRIBUTING.md for details on our code of conduct, and the process for submitting pull requests.

Documentation

Index

• type ExecuteContext
  • func (ctx *ExecuteContext[TARGET, RESULT]) Cancel()
  • func (ctx *ExecuteContext[TARGET, RESULT]) SetValue(key string, value any)
  • func (ctx *ExecuteContext[TARGET, RESULT]) Target() TARGET
• type Parallel
  • func NewParallel[TARGET any, RESULT any](executeHandler func(ctx *ExecuteContext[TARGET, RESULT]) (*RESULT, error), ...) *Parallel[TARGET, RESULT]
  • func (p *Parallel[TARGET, RESULT]) Cancel()
  • func (p *Parallel[TARGET, RESULT]) Execute(target TARGET)
  • func (p *Parallel[TARGET, RESULT]) IsExecuting() bool
  • func (p *Parallel[TARGET, RESULT]) ReadyExecute()
  • func (p *Parallel[TARGET, RESULT]) SetMaxExecuteNum(num int)
  • func (p *Parallel[TARGET, RESULT]) Wait()
• type Result
• type ResultContext
  • func (ctx *ResultContext[TARGET, RESULT]) Cancel()
  • func (ctx *ResultContext[TARGET, RESULT]) Error() error
  • func (ctx *ResultContext[TARGET, RESULT]) Result() RESULT
  • func (ctx *ResultContext[TARGET, RESULT]) Target() TARGET
  • func (ctx *ResultContext[TARGET, RESULT]) Value(key string) any

Types

type ExecuteContext

type ExecuteContext[TARGET any, RESULT any] struct {
	// contains filtered or unexported fields
}

ExecuteContext is the context provided to the executeHandler function. It contains the target to be processed, a shared value map, and a boolean indicating if the execution is running.

func (*ExecuteContext[TARGET, RESULT]) Cancel

func (ctx *ExecuteContext[TARGET, RESULT]) Cancel()

Cancel cancels the execution.

func (*ExecuteContext[TARGET, RESULT]) SetValue

func (ctx *ExecuteContext[TARGET, RESULT]) SetValue(key string, value any)

SetValue sets a shared value to the context.

func (*ExecuteContext[TARGET, RESULT]) Target

func (ctx *ExecuteContext[TARGET, RESULT]) Target() TARGET

Target

type Parallel

type Parallel[TARGET any, RESULT any] struct {
	// contains filtered or unexported fields
}

Parallel

func NewParallel

func NewParallel[TARGET any, RESULT any](executeHandler func(ctx *ExecuteContext[TARGET, RESULT]) (*RESULT, error), resultHandler func(ctx *ResultContext[TARGET, RESULT])) *Parallel[TARGET, RESULT]

NewParallel creates a new parallel executor with custom executeHandler and resultHandler.

func (*Parallel[TARGET, RESULT]) Cancel

func (p *Parallel[TARGET, RESULT]) Cancel()

Cancel cancels the execution. It switches the isExecuting flag to false.

func (*Parallel[TARGET, RESULT]) Execute

func (p *Parallel[TARGET, RESULT]) Execute(target TARGET)

Execute adds a target to be executed by the parallel executor. If the execution is not running, the function will simply return without adding the target.

func (*Parallel[TARGET, RESULT]) IsExecuting

func (p *Parallel[TARGET, RESULT]) IsExecuting() bool

IsExecuting returns a boolean indicating whether the executor is currently executing tasks. It uses the atomic Load method to safely check the isExecuting flag.

func (*Parallel[TARGET, RESULT]) ReadyExecute

func (p *Parallel[TARGET, RESULT]) ReadyExecute()

ReadyExecute initializes the execution channels and starts the background goroutines. It will only be executed once. If it's called multiple times, the function will simply return after the first call.

func (*Parallel[TARGET, RESULT]) SetMaxExecuteNum

func (p *Parallel[TARGET, RESULT]) SetMaxExecuteNum(num int)

SetMaxExecuteNum sets the maximum number of concurrent execute handlers that can run in parallel. If the provided number is less than 1, it defaults to the number of CPUs available on the system (runtime.NumCPU()). This function panics if the provided number is less than 0.

func (*Parallel[TARGET, RESULT]) Wait

func (p *Parallel[TARGET, RESULT]) Wait()

Wait waits for all the tasks to be executed and results to be processed.

type Result

type Result[RESULT any] struct {
	Value RESULT
	Err   error
}

type ResultContext

type ResultContext[TARGET any, RESULT any] struct {
	// contains filtered or unexported fields
}

ResultContext is the context provided to the resultHandler function. It contains the processed target, the result, a shared value map, an error if one occurred, and a boolean indicating if the execution is running.

func (*ResultContext[TARGET, RESULT]) Cancel

func (ctx *ResultContext[TARGET, RESULT]) Cancel()

Cancel cancels the execution. It switches the isExecuting flag to false.

func (*ResultContext[TARGET, RESULT]) Error

func (ctx *ResultContext[TARGET, RESULT]) Error() error

Error returns the error that occurred during processing, if any.

func (*ResultContext[TARGET, RESULT]) Result

func (ctx *ResultContext[TARGET, RESULT]) Result() RESULT

Result returns the result of processing the target.

func (*ResultContext[TARGET, RESULT]) Target

func (ctx *ResultContext[TARGET, RESULT]) Target() TARGET

Target returns the processed target from the result context.

func (*ResultContext[TARGET, RESULT]) Value

func (ctx *ResultContext[TARGET, RESULT]) Value(key string) any

Value returns the shared value associated with the provided key after execution. If the key is not found in the shared value map, it returns nil.