A high-performance and low-cost goroutine pool in Go.

ants

Simple in-memory job queue for Golang using worker-based dispatching.

[![GoDoc](http://godoc.org/github.com/mborders/artifex?status.png)](http://godoc.org/github.com/mborders/artifex)
[![Build Status](https://travis-ci.org/mborders/artifex.svg?branch=master)](https://travis-ci.org/mborders/artifex)
[![Go Report Card](https://goreportcard.com/badge/github.com/mborders/artifex)](https://goreportcard.com/report/github.com/mborders/artifex)
[![codecov](https://codecov.io/gh/mborders/artifex/branch/master/graph/badge.svg)](https://codecov.io/gh/mborders/artifex)

# artifex

Simple in-memory job queue for Golang using worker-based dispatching

Documentation here: https://godoc.org/github.com/mborders/artifex

Cron jobs use the robfig/cron library: https://godoc.org/github.com/robfig/cron

## Example Usage

```go
// 10 workers, 100 max in job queue
d := artifex.NewDispatcher(10, 100)
d.Start()

d.Dispatch(func() {
  // do something
})

err := d.DispatchIn(func() {
  // do something in 500ms
}, time.Millisecond*500)

// Returns a DispatchTicker
dt, err := d.DispatchEvery(func() {
  // do something every 250ms
}, time.Millisecond*250)

// Stop a given DispatchTicker
dt.Stop()

// Returns a DispatchCron
dc, err := d.DispatchCron(func() {
  // do something every 1s
}, "*/1 * * * * *")

// Stop a given DispatchCron
dc.Stop()

// Stop a dispatcher and all its workers/tickers
d.Stop()
```


artifex

An asynchronous task package with async/await style for Go.

# Async
Async is an async/await like task package for Go

![License](https://img.shields.io/badge/license-MIT-green.svg)
[![Tests](https://github.com/yaitoo/async/actions/workflows/tests.yml/badge.svg)](https://github.com/yaitoo/async/actions/workflows/tests.yml)
[![Go Reference](https://pkg.go.dev/badge/github.com/yaitoo/async.svg)](https://pkg.go.dev/github.com/yaitoo/async)
[![Codecov](https://codecov.io/gh/yaitoo/async/branch/main/graph/badge.svg)](https://codecov.io/gh/yaitoo/async)
[![GitHub Release](https://img.shields.io/github/v/release/yaitoo/async)](https://github.com/yaitoo/sqle/blob/main/CHANGELOG.md)
[![Go Report Card](https://goreportcard.com/badge/yaitoo/async)](http://goreportcard.com/report/yaitoo/async)


## Features
- Wait/WaitAny/WaitN for `Task` and `Action`
- `context.Context` with `timeout`, `cancel`  support
- Works with generic instead of `interface{}`

## Tutorials
see more examples on [tasks](./waiter_test.go), [actions](./awaiter_test.go) or [go.dev](https://go.dev/play/p/7jgcRltbwts)

### Install async
- install latest commit from `main` branch
```
go get github.com/yaitoo/async@main
```

- install latest release
```
go get github.com/yaitoo/async@latest
```

### Wait 
wait all tasks to completed.

```
t := async.New[int](func(ctx context.Context) (int, error) {
		return 1, nil
	}, func(ctx context.Context) (int, error) {
		return 2, nil
	})

result, err, taskErrs := t.Wait(context.Background())


fmt.Println(result)  //[1,2] or [2,1]
fmt.Println(err) // nil
fmt.Println(taskErrs) //nil


```


### WaitAny
wait any task to completed

```
t := async.New[int](func(ctx context.Context) (int, error) {
    time.Sleep(2 * time.Second)
		return 1, nil
	}, func(ctx context.Context) (int, error) {
		return 2, nil
	})

result, err, taskErrs := t.WaitAny(context.Background())

fmt.Println(result)  //2
fmt.Println(err) //nil
fmt.Println(taskErrs) //nil

```

### WaitN
wait N tasks to completed. 

```
t := async.New[int](func(ctx context.Context) (int, error) {
    time.Sleep(2 * time.Second)
		return 1, nil
	}, func(ctx context.Context) (int, error) {
		return 2, nil
	}, func(ctx context.Context) (int, error) {
		return 3, nil
	})

result, err, taskErrs := t.WaitN(context.Background(),2)


fmt.Println(result)  //[2,3] or [3,2]
fmt.Println(err) //nil
fmt.Println(taskErrs) //nil

```

### Timeout
cancel all tasks if it is timeout. 
```
 t := async.New[int](func(ctx context.Context) (int, error) {
		time.Sleep(2 * time.Second)
		return 1, nil
	}, func(ctx context.Context) (int, error) {
		time.Sleep(2 * time.Second)
		return 2, nil
	})

	ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
	defer cancel()

	result, err, tasks := t.WaitAny(ctx)
	//result, err, tasks := t.Wait(ctx)

	
	fmt.Println(result) //nil
	fmt.Println(err) // context.DeadlineExceeded
	fmt.Println(taskErrs) //nil
```

### Cancel
manually cancel all tasks.

```
t := async.New[int](func(ctx context.Context) (int, error) {
    time.Sleep(2 * time.Second)
		return 1, nil
	}, func(ctx context.Context) (int, error) {
     time.Sleep(2 * time.Second)
		return 2, nil
	})

ctx, cancel := context.WithCancel(context.Background())
go func(){
  time.Sleep(1 * time.Second)
  cancel()
}()

//result, err, taskErrs := t.WaitAny(ctx)
 result, err, taskErrs := t.Wait(ctx)


fmt.Println(result)  //nil
fmt.Println(err) // context.Cancelled
fmt.Println(taskErrs) // nil


```


## Contributing
Contributions are welcome! If you're interested in contributing, please feel free to [contribute](CONTRIBUTING.md)


## License
[MIT License](LICENSE)

async

An alternative sync library for Go (Future, Promise, Locks).

<div align="center" style="margin:0 !important;"><img src="./docs/images/async.png" /></div>
<div align="center">
  <a href="https://github.com/reugn/async/actions/workflows/build.yml"><img src="https://github.com/reugn/async/actions/workflows/build.yml/badge.svg"></a>
  <a href="https://pkg.go.dev/github.com/reugn/async"><img src="https://pkg.go.dev/badge/github.com/reugn/async"></a>
  <a href="https://goreportcard.com/report/github.com/reugn/async"><img src="https://goreportcard.com/badge/github.com/reugn/async"></a>
  <a href="https://codecov.io/gh/reugn/async"><img src="https://codecov.io/gh/reugn/async/branch/master/graph/badge.svg"></a>
</div>
<br/>
Async is a synchronization and asynchronous computation package for Go.

## Overview
* **ConcurrentMap** - Implements the generic `async.Map` interface in a thread-safe manner by delegating load/store operations to the underlying `sync.Map`.
* **ShardedMap** - Implements the generic `async.Map` interface in a thread-safe manner, delegating load/store operations to one of the underlying `async.SynchronizedMap`s (shards), using a key hash to calculate the shard number.
* **Future** - A placeholder object for a value that may not yet exist.
* **Promise** - While futures are defined as a type of read-only placeholder object created for a result which doesn’t yet exist, a promise can be thought of as a writable, single-assignment container, which completes a future.
* **Executor** - A worker pool for executing asynchronous tasks, where each submission returns a Future instance representing the result of the task.
* **Task** - A data type for controlling possibly lazy and asynchronous computations.
* **Once** - An object similar to sync.Once having the Do method taking `f func() (T, error)` and returning `(T, error)`.
* **Value** - An object similar to atomic.Value, but without the consistent type constraint.
* **WaitGroupContext** - A WaitGroup with the `context.Context` support for graceful unblocking.
* **ReentrantLock** - A mutex that allows goroutines to enter into the lock on a resource more than once.
* **PriorityLock** - A non-reentrant mutex that allows for the specification of lock acquisition priority.

## Examples
Can be found in the examples directory/tests.

## License
Licensed under the MIT License.


A safe way to execute functions asynchronously, recovering them in case of panic.

## Async

[![Build Status](https://travis-ci.org/StudioSol/async.svg?branch=master)](https://travis-ci.org/StudioSol/async)
[![codecov](https://codecov.io/gh/StudioSol/async/branch/master/graph/badge.svg)](https://codecov.io/gh/StudioSol/async)
[![Go Report Card](https://goreportcard.com/badge/github.com/StudioSol/async)](https://goreportcard.com/report/github.com/StudioSol/async)
[![GoDoc](https://godoc.org/github.com/StudioSol/async?status.svg)](https://godoc.org/github.com/StudioSol/async)

Provides a safe way to execute functions asynchronously, recovering them in case of panic. It also provides an error stack aiming to facilitate fail causes discovery, and a simple way to control execution flow without `WaitGroup`.

### Usage
```go
var (
    user   User
    songs  []Songs
    photos []Photos
)

err := async.Run(ctx,
    func(ctx context.Context) error {
        user, err = user.Get(ctx, id)
        return err
    },
    func(ctx context.Context) error {
        songs, err = song.GetByUserID(ctx, id)
        return err
    },
    func(ctx context.Context) error {
        photos, err = photo.GetByUserID(ctx, id)
        return err
    },
)

if err != nil {
    log.Error(err)
}
```

You can also limit the number of asynchronous tasks

```go
runner := async.NewRunner(tasks...).WithLimit(3)
if err := runner.Run(ctx); err != nil { 
    log.Error(e)
}
```


AsyncJob is an asynchronous queue job manager with light code, clear and speed.

[![Go](https://github.com/lab210-dev/async-job/actions/workflows/go.yml/badge.svg)](https://github.com/lab210-dev/async-job/actions/workflows/go.yml)
![GitHub go.mod Go version](https://img.shields.io/github/go-mod/go-version/lab210-dev/async-job)
[![Go Report Card](https://goreportcard.com/badge/github.com/lab210-dev/async-job)](https://goreportcard.com/report/github.com/lab210-dev/async-job)
[![codecov](https://codecov.io/gh/lab210-dev/async-job/branch/main/graph/badge.svg?token=3JRL5ZLSIH)](https://codecov.io/gh/lab210-dev/async-job)
[![License](https://img.shields.io/badge/license-MIT-blue.svg)](https://github.com/lab210-dev/async-job/blob/main/LICENSE)
# Overview

AsyncJob is an asynchronous job manager with light code, clear and speed. I hope so ! 😬

## Features

- [x] AsyncJob is a simple asynchronous job manager.
- [x] Full code coverage
- [x] Async queue
- [x] Define the number of asynchronous tasks (default: runtime.NumCPU())
- [x] Handling of managed and unmanaged errors
- [x] Provide a simple ETA
- [x] Full code description

### Usage

```go
package main

import (
	"github.com/lab210-dev/async-job"
	"log"
)

func main() {
	// Create a new AsyncJob
	asj := asyncjob.New[string]()

	// Set the number of asynchronous tasks (default: runtime.NumCPU())
	asj.SetWorkers(2)

	// Listen to the progress status
	asj.OnProgress(func(progress asyncjob.Progress) {
            log.Printf("Progress: %s\n", progress.String())
	})

	// Run all jobs 
	err := asj.Run(func(job asyncjob.Job[string]) error {
            // receive the job in job data function
            // if err return or panic, the job will be marked as failed and all progress will be canceled
            return nil
	}, []string{"Hello", "World"})

	// if a job returns an error, it stops the process
	if err != nil {
            log.Fatal(err)
	}
}
```

## 💡 For better performance

Using a modulo to reduce the eta display (fast example)

```go
package main

import (
	"github.com/lab210-dev/async-job"
	"log"
	"time"
)

func main() {
	// create slice of jobs
	var list []time.Duration
	for i := 1; i <= 100; i++ {
		list = append(list, time.Duration(1)*time.Millisecond)
	}
	err := asyncjob.New[time.Duration]().
		SetWorkers(2).
		OnProgress(func(progress asyncjob.Progress) {
			// Eta will be displayed every 10 jobs
			if progress.Current()%10 != 0 {
				return
			}
			// print the eta
		    log.Printf("Progress: %s\n", progress.String())
		}).
		Run(func(job asyncjob.Job[time.Duration]) error {
			// slow down the job
			time.Sleep(job.Data())
			return nil
		}, list)
	
	// if a job returns an error, it stops the process
	if err != nil {
		log.Fatal(err)
	}
}
```

## 🤝 Contributions
Contributors to the package are encouraged to help improve the code.

async-job

Flexible mechanism to make execution flow interruptible.

> # 🚧 breaker [![Awesome Go][awesome.icon]][awesome.page]
>
> Flexible mechanism to make execution flow interruptible.

[![Build][build.icon]][build.page]
[![Documentation][docs.icon]][docs.page]
[![Quality][quality.icon]][quality.page]
[![Template][template.icon]][template.page]
[![Coverage][coverage.icon]][coverage.page]
[![Mirror][mirror.icon]][mirror.page]

## 💡 Idea

The breaker carries a cancellation signal to interrupt an action execution.

```go
var NewYear = time.Time{}.AddDate(time.Now().Year(), 0, 0)

interrupter := breaker.Multiplex(
	breaker.BreakByContext(context.WithTimeout(req.Context(), time.Minute)),
	breaker.BreakByDeadline(NewYear),
	breaker.BreakBySignal(os.Interrupt),
)
defer interrupter.Close()

<-interrupter.Done() // wait context cancellation, timeout or interrupt signal
```

A full description of the idea is available [here][design.page].

## 🏆 Motivation

I have to make modules [github.com/kamilsk/retry/v5][retry]:

```go
if err := retry.Retry(breaker.BreakByTimeout(time.Minute), action); err != nil {
	log.Fatal(err)
}
```

and [github.com/kamilsk/semaphore/v5][semaphore]:

```go
if err := semaphore.Acquire(breaker.BreakByTimeout(time.Minute), 5); err != nil {
	log.Fatal(err)
}
```

more consistent and reliable.

Additionally, I want to implement a Graceful Shutdown on the same mechanism.

## 🤼‍♂️ How to

### Do HTTP request with retries

```go
interrupter := breaker.Multiplex(
	breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
	breaker.BreakByTimeout(timeout),
)
defer interrupter.Close()

ctx := breaker.ToContext(interrupter)
ctx = context.WithValue(ctx, header, "...")

req, err := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil)
if err != nil {
	panic(err)
}

var resp *http.Response
action := func(ctx context.Context) (err error) {
	req = req.Clone(ctx)

	source := ctx.Value(header).(string)
	req.Header.Set(header, source)

	resp, err = http.DefaultClient.Do(req)
	return err
}

if err := retry.Do(ctx, action); err != nil {
	panic(err)
}
```

<details>
  <summary>Full example</summary>

```go
package main

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"syscall"
	"time"

	"github.com/kamilsk/breaker"
	"github.com/kamilsk/retry/v5"
)

func main() {
	const (
		header  = "X-Message"
		timeout = time.Minute
	)

	server := httptest.NewServer(http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {
		time.Sleep(timeout / 10)
		_, _ = rw.Write([]byte(req.Header.Get(header)))
	}))
	defer server.Close()

	interrupter := breaker.Multiplex(
		breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
		breaker.BreakByTimeout(timeout),
	)
	defer interrupter.Close()

	ctx := breaker.ToContext(interrupter)
	ctx = context.WithValue(ctx, header, "flexible mechanism to make execution flow interruptible")

	req, err := http.NewRequestWithContext(ctx, http.MethodGet, server.URL, nil)
	if err != nil {
		panic(err)
	}

	var resp *http.Response
	action := func(ctx context.Context) (err error) {
		req = req.Clone(ctx)

		source := ctx.Value(header).(string)
		req.Header.Set(header, source)

		resp, err = http.DefaultClient.Do(req)
		return err
	}

	if err := retry.Do(ctx, action); err != nil {
		fmt.Println("error:", err)
		return
	}
	_, _ = io.Copy(os.Stdout, resp.Body)
}
```

[Play it](https://play.golang.org/p/nsvS79E7VRA)!
</details>

### Graceful Shutdown HTTP server

```go
interrupter := breaker.Multiplex(
	breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
	breaker.BreakByTimeout(timeout),
)
defer interrupter.Close()

server := http.Server{
	BaseContext: func(net.Listener) context.Context {
		return breaker.ToContext(interrupter)
	},
}
go func() {
	if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
		log.Fatal(err)
	}
}()

<-interrupter.Done()
if errors.Is(interrupter.Err(), breaker.Interrupted) {
	if err := server.Shutdown(context.TODO()); err != nil {
		panic(err)
	}
}
```

<details>
  <summary>Full example</summary>

```go
package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"net"
	"net/http"
	"os"
	"syscall"
	"time"

	"github.com/kamilsk/breaker"
)

func main() {
	const timeout = time.Minute

	interrupter := breaker.Multiplex(
		breaker.BreakBySignal(os.Interrupt, syscall.SIGINT, syscall.SIGTERM),
		breaker.BreakByTimeout(timeout),
	)
	defer interrupter.Close()

	server := http.Server{
		Addr:    ":8080",
		Handler: http.HandlerFunc(func(rw http.ResponseWriter, req *http.Request) {}),
		BaseContext: func(net.Listener) context.Context {
			return breaker.ToContext(interrupter)
		},
	}
	go func() {
		if err := server.ListenAndServe(); !errors.Is(err, http.ErrServerClosed) {
			log.Fatal(err)
		}
	}()

	<-interrupter.Done()
	if err := interrupter.Err(); errors.Is(err, breaker.Interrupted) {
		if err := server.Shutdown(context.TODO()); err != nil {
			panic(err)
		}
	}
	fmt.Println("graceful shutdown")
}
```

[Play it](https://play.golang.org/p/D7-nqT-ncR0)!
</details>

## 🧩 Integration

The library uses [SemVer](https://semver.org) for versioning, and it is not
[BC](https://en.wikipedia.org/wiki/Backward_compatibility)-safe through major releases.
You can use [go modules](https://github.com/golang/go/wiki/Modules) to manage its version.

```bash
$ go get github.com/kamilsk/breaker@latest
```

## 🤲 Outcomes

### Console tool to execute commands for a limited time

The example shows how to execute console commands for ten minutes.

```bash
$ date
# Thu Jan  7 21:02:21
$ breakit after 10m -- server run --port=8080
$ breakit ps
# +--------------------------+----------------------------+----------+----------+
# | Process                  | Status                     | Since    | Until    |
# +--------------------------+----------------------------+----------+----------+
# | server run --port=8080   | exit 1; panic: database... | 21:02:36 | -        |
# +--------------------------+----------------------------+----------+----------+
# |                          |                            |    Total |        1 |
# +--------------------------+----------------------------+----------+----------+
$ breakit after 10m -- database run --port=5432
$ breakit after 10m delay 5s -- server run --port=8080
$ breakit ps
# +--------------------------+----------------------------+----------+----------+
# | Process                  | Status                     | Since    | Until    |
# +--------------------------+----------------------------+----------+----------+
# | database run --port=5432 | running                    | 21:04:09 | 21:14:09 |
# | server run --port=8080   | delayed                    | 21:04:30 | 21:14:25 |
# +--------------------------+----------------------------+----------+----------+
# |                          |                            |    Total |        2 |
# +--------------------------+----------------------------+----------+----------+
```

See more details [here][cli].

<p align="right">made with ❤️ for everyone</p>

[awesome.page]:     https://github.com/avelino/awesome-go#goroutines
[awesome.icon]:     https://awesome.re/mentioned-badge.svg
[build.page]:       https://travis-ci.com/kamilsk/breaker
[build.icon]:       https://travis-ci.com/kamilsk/breaker.svg?branch=master
[coverage.page]:    https://codeclimate.com/github/kamilsk/breaker/test_coverage
[coverage.icon]:    https://api.codeclimate.com/v1/badges/1d703de640b4c6cfcd6f/test_coverage
[design.page]:      https://www.notion.so/octolab/breaker-77116e98fda74c28bd64e42bd440bbf3?r=0b753cbf767346f5a6fd51194829a2f3
[docs.page]:        https://pkg.go.dev/github.com/kamilsk/breaker
[docs.icon]:        https://img.shields.io/badge/docs-pkg.go.dev-blue
[promo.page]:       https://github.com/kamilsk/breaker
[quality.page]:     https://goreportcard.com/report/github.com/kamilsk/breaker
[quality.icon]:     https://goreportcard.com/badge/github.com/kamilsk/breaker
[template.page]:    https://github.com/octomation/go-module
[template.icon]:    https://img.shields.io/badge/template-go--module-blue
[mirror.page]:      https://bitbucket.org/kamilsk/breaker
[mirror.icon]:      https://img.shields.io/badge/mirror-bitbucket-blue

[cli]:              https://github.com/octolab/breakit
[retry]:            https://github.com/kamilsk/retry
[semaphore]:        https://github.com/kamilsk/semaphore

[_]:                https://img.shields.io/sourcegraph/rrc/github.com/kamilsk/breaker


breaker

Transform your function to return channels for easy and powerful parallel processing.

# channelify

[![GitHub stars](https://img.shields.io/github/stars/ddelizia/channelify.svg?style=social&label=Star&maxAge=2592000)](https://GitHub.com/ddelizia/channelify/stargazers/) 
[![GitHub forks](https://img.shields.io/github/forks/ddelizia/channelify.svg?style=social&label=Fork&maxAge=2592000)](https://GitHub.com/ddelizia/channelify/network/) 

[![PkgGoDev](https://pkg.go.dev/badge/github.com/ddelizia/channelify)](https://pkg.go.dev/github.com/ddelizia/channelify)
[![Test Actions Status](https://github.com/ddelizia/channelify/workflows/Test/badge.svg)](https://github.com/ddelizia/channelify/actions)
[![Go Report Card](https://goreportcard.com/badge/github.com/ddelizia/channelify)](https://goreportcard.com/report/github.com/ddelizia/channelify)
[![MIT license](https://img.shields.io/badge/License-MIT-blue.svg)](https://lbesson.mit-license.org/)


This library helps you to transform any function into a function that returns a any type to a function that return such types within a channel. This is useful to run in parallel multiple functions and have control on the returned values.

Channelify uses go routines to parallelize the execution of the functions. 

The idea comes from Javascript Promisify utility that transforms a callback into a promise.

## Installation

```
go get github.com/ddelizia/channelify
```


## Usage example

Here an example of transforming a simple function in channel so you can execute multiple functions in parallel:

```go
fn := func () string {
    time.Sleep(time.Second * 3)
    return "hello"
}

ch1 := Channelify(fn)
ch2 := Channelify(fn)
chV1 := ch1.(func () chan string)()
chV2 := ch2.(func () chan string)()

v1, v2 := <- chV1, <- chV2
```

If your functions returns multiple values you can use as follow:

```go
fn1 := func (hello string) (string, error)  {
    time.Sleep(time.Second * 2)
    fmt.Println(hello)
    return hello, nil
}

fn2 := func (hello string) (string, error)  {
    time.Sleep(time.Second * 3)
    fmt.Println(hello)
    return hello, nil
}

ch1 := Channelify(fn1)
ch2 := Channelify(fn2)
chV1, chE1 := ch1.(func (string) (chan string, chan error))("hello1")
chV2, chE2 := ch2.(func (string) (chan string, chan error))("hello2")

v1, e1, v2, e2 := <- chV1, <- chE1, <- chV2, <- chE2

fmt.Print(v1, e1, v2, e2)
```

# Contributing

 * Fork it
 * Create your feature branch (`git checkout -b my-new-feature`)
 * Commit your changes (`git commit -am 'Add some feature'`)
 * Push to the branch (`git push origin my-new-feature`)
 * Create new Pull Request




channelify

`conc` is your toolbelt for structured concurrency in go, making common tasks easier and safer.

![conch](https://user-images.githubusercontent.com/12631702/210295964-785cc63d-d697-420c-99ff-f492eb81dec9.svg)

# `conc`: better structured concurrency for go

[![Go Reference](https://pkg.go.dev/badge/github.com/sourcegraph/conc.svg)](https://pkg.go.dev/github.com/sourcegraph/conc)
[![Sourcegraph](https://img.shields.io/badge/view%20on-sourcegraph-A112FE?logo=data:image/png;base64,iVBORw0KGgoAAAANSUhEUgAAADIAAAAyCAYAAAAeP4ixAAAEZklEQVRoQ+2aXWgUZxSG3292sxtNN43BhBakFPyhxSujRSxiU1pr7SaGXqgUxOIEW0IFkeYighYUxAuLUlq0lrq2iCDpjWtmFVtoG6QVNOCFVShVLyxIk0DVjZLMxt3xTGTccd2ZOd/8JBHci0CY9zvnPPN+/7sCIXwKavOwAcy2QgngQiIztDSE0OwQlDPYR1ebiaH6J5kZChyfW12gRG4QVgGTBfMchMbFP9Sn5nlZL2D0JjLD6710lc+z0NfqSGTXQRQ4bX07Mq423yoBL3OSyHSvUxirMuaEvgbJWrdcvkHMoJwxYuq4INUhyuWvQa1jvdMGxAvCxJlyEC9XOBCWL04wwRzpbDoDQ7wfZJzIQLi5Eggk6DiRhZgWIAbE3NrM4A3LPT8Q7UgqAqLqTmLSHLGPkyzG/qXEczhd0q6RH+zaSBfaUoc4iQx19pIClIscrTkNZzG6gd7qMY6eC2Hqyo705ZfTf+eqJmhMzcSbYtQpOXc92ZsZjLVAL4YNUQbJ5Ttg4CQrQdGYj44Xr9m1XJCzmZusFDJOWNpHjmh5x624a2ZFtOKDVL+uNo2TuXE3bZQQZUf8gtgqP31uI94Z/rMqix+IGiRfWw3xN9dCgVx+L3WrHm4Dju6PXz/EkjuXJ6R+IGgyOE1TbZqTq9y1eo0EZo7oMo1ktPu3xjHvuiLT5AFNszUyDULtWpzE2/fEsey8O5TbWuGWwxrs5rS7nFNMWJrNh2No74s9Ec4vRNmRRzPXMP19fBMSVsGcOJ98G8N3Wl2gXcbTjbX7vUBxLaeASDQCm5Cu/0E2tvtb0Ea+BowtskFD0wvlc6Rf2M+Jx7dTu7ubFr2dnKDRaMQe2v/tcIrNB7FH0O50AcrBaApmRDVwFO31ql3pD8QW4dP0feNwl/Q+kFEtRyIGyaWXnpy1OO0qNJWHo1y6iCmAGkBb/Ru+HenDWIF2mo4r8G+tRRzoniSn2uqFLxANhe9LKHVyTbz6egk9+x5w5fK6ulSNNMhZ/Feno+GebLZV6isTTa6k5qNl5RnZ5u56Ib6SBvFzaWBBVFZzvnERWlt/Cg4l27XChLCqFyLekjhy6xJyoytgjPf7opIB8QPx7sYFiMXHPGt76m741MhCKMZfng0nBOIjmoJPsLqWHwgFpe6V6qtfcopxveR2Oy+J0ntIN/zCWkf8QNAJ7y6d8Bq4lxLc2/qJl5K7t432XwcqX5CrI34gzATWuYILQtdQPyePDK3iuOekCR3Efjhig1B1Uq5UoXEEoZX7d1q535J5S9VOeFyYyEBku5XTMXXKQTToX5Rg7OI44nbW5oKYeYK4EniMeF0YFNSmb+grhc84LyRCEP1/OurOcipCQbKxDeK2V5FcVyIDMQvsgz5gwFhcWWwKyRlvQ3gv29RwWoDYAbIofNyBxI9eDlQ+n3YgsgCWnr4MStGXQXmv9pF2La/k3OccV54JEBM4yp9EsXa/3LfO0dGPcYq0Y7DfZB8nJzZw2rppHgKgVHs8L5wvRwAAAABJRU5ErkJggg==)](https://sourcegraph.com/github.com/sourcegraph/conc)
[![Go Report Card](https://goreportcard.com/badge/github.com/sourcegraph/conc)](https://goreportcard.com/report/github.com/sourcegraph/conc)
[![codecov](https://codecov.io/gh/sourcegraph/conc/branch/main/graph/badge.svg?token=MQZTEA1QWT)](https://codecov.io/gh/sourcegraph/conc)
[![Discord](https://img.shields.io/badge/discord-chat-%235765F2)](https://discord.gg/bvXQXmtRjN)

`conc` is your toolbelt for structured concurrency in go, making common tasks
easier and safer.

```sh
go get github.com/sourcegraph/conc
```

# At a glance

- Use [`conc.WaitGroup`](https://pkg.go.dev/github.com/sourcegraph/conc#WaitGroup) if you just want a safer version of `sync.WaitGroup`
- Use [`pool.Pool`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool) if you want a concurrency-limited task runner
- Use [`pool.ResultPool`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#ResultPool) if you want a concurrent task runner that collects task results
- Use [`pool.(Result)?ErrorPool`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#ErrorPool) if your tasks are fallible
- Use [`pool.(Result)?ContextPool`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#ContextPool) if your tasks should be canceled on failure
- Use [`stream.Stream`](https://pkg.go.dev/github.com/sourcegraph/conc/stream#Stream) if you want to process an ordered stream of tasks in parallel with serial callbacks
- Use [`iter.Map`](https://pkg.go.dev/github.com/sourcegraph/conc/iter#Map) if you want to concurrently map a slice
- Use [`iter.ForEach`](https://pkg.go.dev/github.com/sourcegraph/conc/iter#ForEach) if you want to concurrently iterate over a slice
- Use [`panics.Catcher`](https://pkg.go.dev/github.com/sourcegraph/conc/panics#Catcher) if you want to catch panics in your own goroutines

All pools are created with
[`pool.New()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#New)
or
[`pool.NewWithResults[T]()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#NewWithResults),
then configured with methods:

- [`p.WithMaxGoroutines()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool.MaxGoroutines) configures the maximum number of goroutines in the pool
- [`p.WithErrors()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool.WithErrors) configures the pool to run tasks that return errors
- [`p.WithContext(ctx)`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#Pool.WithContext) configures the pool to run tasks that should be canceled on first error
- [`p.WithFirstError()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#ErrorPool.WithFirstError) configures error pools to only keep the first returned error rather than an aggregated error
- [`p.WithCollectErrored()`](https://pkg.go.dev/github.com/sourcegraph/conc/pool#ResultContextPool.WithCollectErrored) configures result pools to collect results even when the task errored

# Goals

The main goals of the package are:
1) Make it harder to leak goroutines
2) Handle panics gracefully
3) Make concurrent code easier to read

## Goal #1: Make it harder to leak goroutines

A common pain point when working with goroutines is cleaning them up. It's
really easy to fire off a `go` statement and fail to properly wait for it to
complete.

`conc` takes the opinionated stance that all concurrency should be scoped.
That is, goroutines should have an owner and that owner should always
ensure that its owned goroutines exit properly.

In `conc`, the owner of a goroutine is always a `conc.WaitGroup`. Goroutines
are spawned in a `WaitGroup` with `(*WaitGroup).Go()`, and
`(*WaitGroup).Wait()` should always be called before the `WaitGroup` goes out
of scope.

In some cases, you might want a spawned goroutine to outlast the scope of the
caller. In that case, you could pass a `WaitGroup` into the spawning function.

```go
func main() {
    var wg conc.WaitGroup
    defer wg.Wait()

    startTheThing(&wg)
}

func startTheThing(wg *conc.WaitGroup) {
    wg.Go(func() { ... })
}
```

For some more discussion on why scoped concurrency is nice, check out [this
blog
post](https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/).

## Goal #2: Handle panics gracefully

A frequent problem with goroutines in long-running applications is handling
panics. A goroutine spawned without a panic handler will crash the whole process
on panic. This is usually undesirable.

However, if you do add a panic handler to a goroutine, what do you do with the
panic once you catch it? Some options:
1) Ignore it
2) Log it
3) Turn it into an error and return that to the goroutine spawner
4) Propagate the panic to the goroutine spawner

Ignoring panics is a bad idea since panics usually mean there is actually
something wrong and someone should fix it.

Just logging panics isn't great either because then there is no indication to the spawner
that something bad happened, and it might just continue on as normal even though your
program is in a really bad state.

Both (3) and (4) are reasonable options, but both require the goroutine to have
an owner that can actually receive the message that something went wrong. This
is generally not true with a goroutine spawned with `go`, but in the `conc`
package, all goroutines have an owner that must collect the spawned goroutine.
In the conc package, any call to `Wait()` will panic if any of the spawned goroutines
panicked. Additionally, it decorates the panic value with a stacktrace from the child
goroutine so that you don't lose information about what caused the panic.

Doing this all correctly every time you spawn something with `go` is not
trivial and it requires a lot of boilerplate that makes the important parts of
the code more difficult to read, so `conc` does this for you.

<table>
<tr>
<th><code>stdlib</code></th>
<th><code>conc</code></th>
</tr>
<tr>
<td>

```go
type caughtPanicError struct {
    val   any
    stack []byte
}

func (e *caughtPanicError) Error() string {
    return fmt.Sprintf(
        "panic: %q\n%s",
        e.val,
        string(e.stack)
    )
}

func main() {
    done := make(chan error)
    go func() {
        defer func() {
            if v := recover(); v != nil {
                done <- &caughtPanicError{
                    val: v,
                    stack: debug.Stack()
                }
            } else {
                done <- nil
            }
        }()
        doSomethingThatMightPanic()
    }()
    err := <-done
    if err != nil {
        panic(err)
    }
}
```
</td>
<td>

```go
func main() {
    var wg conc.WaitGroup
    wg.Go(doSomethingThatMightPanic)
    // panics with a nice stacktrace
    wg.Wait()
}
```
</td>
</tr>
</table>

## Goal #3: Make concurrent code easier to read

Doing concurrency correctly is difficult. Doing it in a way that doesn't
obfuscate what the code is actually doing is more difficult. The `conc` package
attempts to make common operations easier by abstracting as much boilerplate
complexity as possible.

Want to run a set of concurrent tasks with a bounded set of goroutines? Use
`pool.New()`. Want to process an ordered stream of results concurrently, but
still maintain order? Try `stream.New()`. What about a concurrent map over
a slice? Take a peek at `iter.Map()`.

Browse some examples below for some comparisons with doing these by hand.

# Examples

Each of these examples forgoes propagating panics for simplicity. To see
what kind of complexity that would add, check out the "Goal #2" header above.

Spawn a set of goroutines and waiting for them to finish:

<table>
<tr>
<th><code>stdlib</code></th>
<th><code>conc</code></th>
</tr>
<tr>
<td>

```go
func main() {
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            // crashes on panic!
            doSomething()
        }()
    }
    wg.Wait()
}
```
</td>
<td>

```go
func main() {
    var wg conc.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Go(doSomething)
    }
    wg.Wait()
}
```
</td>
</tr>
</table>

Process each element of a stream in a static pool of goroutines:

<table>
<tr>
<th><code>stdlib</code></th>
<th><code>conc</code></th>
</tr>
<tr>
<td>

```go
func process(stream chan int) {
    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for elem := range stream {
                handle(elem)
            }
        }()
    }
    wg.Wait()
}
```
</td>
<td>

```go
func process(stream chan int) {
    p := pool.New().WithMaxGoroutines(10)
    for elem := range stream {
        elem := elem
        p.Go(func() {
            handle(elem)
        })
    }
    p.Wait()
}
```
</td>
</tr>
</table>

Process each element of a slice in a static pool of goroutines:

<table>
<tr>
<th><code>stdlib</code></th>
<th><code>conc</code></th>
</tr>
<tr>
<td>

```go
func process(values []int) {
    feeder := make(chan int, 8)

    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for elem := range feeder {
                handle(elem)
            }
        }()
    }

    for _, value := range values {
        feeder <- value
    }
    close(feeder)
    wg.Wait()
}
```
</td>
<td>

```go
func process(values []int) {
    iter.ForEach(values, handle)
}
```
</td>
</tr>
</table>

Concurrently map a slice:

<table>
<tr>
<th><code>stdlib</code></th>
<th><code>conc</code></th>
</tr>
<tr>
<td>

```go
func concMap(
    input []int,
    f func(int) int,
) []int {
    res := make([]int, len(input))
    var idx atomic.Int64

    var wg sync.WaitGroup
    for i := 0; i < 10; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()

            for {
                i := int(idx.Add(1) - 1)
                if i >= len(input) {
                    return
                }

                res[i] = f(input[i])
            }
        }()
    }
    wg.Wait()
    return res
}
```
</td>
<td>

```go
func concMap(
    input []int,
    f func(*int) int,
) []int {
    return iter.Map(input, f)
}
```
</td>
</tr>
</table>

Process an ordered stream concurrently:


<table>
<tr>
<th><code>stdlib</code></th>
<th><code>conc</code></th>
</tr>
<tr>
<td>

```go
func mapStream(
    in chan int,
    out chan int,
    f func(int) int,
) {
    tasks := make(chan func())
    taskResults := make(chan chan int)

    // Worker goroutines
    var workerWg sync.WaitGroup
    for i := 0; i < 10; i++ {
        workerWg.Add(1)
        go func() {
            defer workerWg.Done()
            for task := range tasks {
                task()
            }
        }()
    }

    // Ordered reader goroutines
    var readerWg sync.WaitGroup
    readerWg.Add(1)
    go func() {
        defer readerWg.Done()
        for result := range taskResults {
            item := <-result
            out <- item
        }
    }()

    // Feed the workers with tasks
    for elem := range in {
        resultCh := make(chan int, 1)
        taskResults <- resultCh
        tasks <- func() {
            resultCh <- f(elem)
        }
    }

    // We've exhausted input.
    // Wait for everything to finish
    close(tasks)
    workerWg.Wait()
    close(taskResults)
    readerWg.Wait()
}
```
</td>
<td>

```go
func mapStream(
    in chan int,
    out chan int,
    f func(int) int,
) {
    s := stream.New().WithMaxGoroutines(10)
    for elem := range in {
        elem := elem
        s.Go(func() stream.Callback {
            res := f(elem)
            return func() { out <- res }
        })
    }
    s.Wait()
}
```
</td>
</tr>
</table>

# Status

This package is currently pre-1.0. There are likely to be minor breaking
changes before a 1.0 release as we stabilize the APIs and tweak defaults.
Please open an issue if you have questions, concerns, or requests that you'd
like addressed before the 1.0 release. Currently, a 1.0 is targeted for 
March 2023.


conc

A pool built around exec.Cmd that spins up a given number of processes in advance and attaches stdin and stdout to them when needed. Very similar to FastCGI or Apache Prefork MPM but works for any command.

# ExecPool

[![License](https://img.shields.io/badge/license-mit-green.svg)](https://github.com/hexdigest/execpool/blob/master/LICENSE)
[![Build](https://github.com/hexdigest/execpool/actions/workflows/go.yml/badge.svg)](https://github.com/hexdigest/execpool/actions/workflows/go.yml)
[![Coverage Status](https://coveralls.io/repos/github/hexdigest/execpool/badge.svg?branch=main)](https://coveralls.io/github/hexdigest/execpool?branch=main)
[![Go Report Card](https://goreportcard.com/badge/github.com/hexdigest/execpool)](https://goreportcard.com/report/github.com/hexdigest/execpool)
[![GoDoc](https://godoc.org/github.com/hexdigest/execpool?status.svg)](http://godoc.org/github.com/hexdigest/execpool)
[![Release](https://img.shields.io/github/release/hexdigest/execpool.svg)](https://github.com/hexdigest/execpool/releases/latest)
[![Awesome](https://cdn.rawgit.com/sindresorhus/awesome/d7305f38d29fed78fa85652e3a63e154dd8e8829/media/badge.svg)](https://github.com/avelino/awesome-go#goroutines)

Why? Sometimes when you implement Go services you have to integrate with third party command line tools. 
You can do it by simply using [exec.Command](https://golang.org/pkg/os/exec/#Command) but the problem with this approach is 
that your service has to wait while a new process is being loaded into the memory and started. Sometimes this can drastically 
increase the latency of your service. ExecPool works similarly to [FastCGI](https://en.wikipedia.org/wiki/FastCGI) but it can
wrap any regular process. It spins up a given number of processes in advance and when it's time to handle a request from the
user your service just attaches stdin to an existing process from the pool. Basically the execpool helps you trade memory for latency.

# Usage

```go
package main

import (
	"fmt"
	"io/ioutil"
	"log"
	"os/exec"
	"strings"

	"github.com/hexdigest/execpool"
)

func main() {
	cmd := exec.Command("grep", "none")

	//spin up 100 processes of grep
	pool, err := execpool.New(cmd, 100)

	rc := pool.Exec(strings.NewReader("this makes sense\nthis is nonesense"))
	b, err := ioutil.ReadAll(rc)
	if err != nil {
		log.Fatalf("failed to read from stdout: %v", err)
	}

	// this is nonesense
	fmt.Println(string(b))
}
```

# Benchmark

This benchmark compares "standard" approach with exec.Command and execpool.Pool by running
grep 100 times. For heavier processes you can expect a bigger difference.
```
make benchmark
goos: darwin
goarch: amd64
pkg: github.com/hexdigest/execpool
cpu: Intel(R) Core(TM) i7-6820HQ CPU @ 2.70GHz
BenchmarkNew-8               100            941753 ns/op
BenchmarkCmd-8               100           2386990 ns/op
```



execpool

![Gopher Floc Control](https://github.com/workanator/repo-faces/blob/main/go-floc/main.png)

# go-floc
Floc: Orchestrate goroutines with ease.

[![Go Reference](https://pkg.go.dev/badge/github.com/workanator/go-floc.svg)](https://pkg.go.dev/github.com/workanator/go-floc)
[![Build Status](https://travis-ci.org/workanator/go-floc.svg?branch=main)](https://travis-ci.org/workanator/go-floc)
[![Coverage Status](https://coveralls.io/repos/github/workanator/go-floc/badge.svg?branch=main)](https://coveralls.io/github/workanator/go-floc?branch=main)
[![Go Report Card](https://goreportcard.com/badge/github.com/workanator/go-floc)](https://goreportcard.com/report/github.com/workanator/go-floc)
[![Join the chat at https://gitter.im/go-floc/Lobby](https://badges.gitter.im/go-floc/Lobby.svg)](https://gitter.im/go-floc/Lobby?utm_source=badge&utm_medium=badge&utm_campaign=pr-badge&utm_content=badge)
[![License](https://img.shields.io/dub/l/vibe-d.svg)](https://github.com/workanator/go-floc/blob/main/LICENSE)

The goal of the project is to make the process of running goroutines in parallel
and synchronizing them easy.

## Announcements

`v3` released! The new version keeps the same design as `v2` but it distributed now as a Go module and therefore
it has breaking changes.

## Migration from v2 to v3

1. Install the package with `go get github.com/workanator/go-floc/v3`
2. Replace all old paths `gopkg.in/workanator/go-floc` to `github.com/workanator/go-floc/v3`
3. All types from `errors` sub-package moved to the base package `go-floc`. So please, remove all imports of
   `"gopkg.in/workanator/go-floc/errors"` if any and fix type names, e.g. `errors.ErrTimeout` -> `floc.ErrTimeout`

## Installation and requirements

The package requires Go v1.12 or later.

To install the package use `go get github.com/workanator/go-floc/v3`

## Documentation and examples

Please refer [Godoc reference](https://pkg.go.dev/github.com/workanator/go-floc)
of the package for more details.

Some examples are available at the Godoc reference. Additional examples can
be found in [go-floc-showcase](https://github.com/workanator/go-floc-showcase?branch=v3).

## Features

- Easy to use functional interface.
- Simple parallelism and synchronization of jobs.
- As little overhead as possible, in comparison to direct use of goroutines
and sync primitives.
- Provide better control over execution with one entry point and one exit
point.

## Introduction

Floc introduces some terms which are widely used through the package.

### Flow

Flow is the overall process which can be controlled through `floc.Flow`. Flow
can be canceled or completed with any arbitrary data at any point of execution.
Flow has only one enter point and only one exit point.

```go
// Design the job
flow := run.Sequence(do, something, here, ...)

// The enter point: Run the job
result, data, err := floc.Run(flow)

// The exit point: Check the result of the job.
if err != nil {
	// Handle the error
} else if result.IsCompleted() {
	// Handle the success
} else {
	// Handle other cases
}
```

### Job

Job in Floc is a smallest piece of flow. The prototype of job function is
`floc.Job`. Each job can read/write data with `floc.Context` and control
the flow with `floc.Control`.

`Cancel()`, `Complete()`, `Fail()` methods of `floc.Flow` has permanent effect.
Once finished flow cannot be canceled or completed anymore. Calling `Fail` and
returning error from job is almost equal.

```go
func ValidateContentLength(ctx floc.Context, ctrl floc.Control) error {
  request := ctx.Value("request").(http.Request)

  // Cancel the flow with error if request body size is too big
  if request.ContentLength > MaxContentLength {
    return errors.New("content is too big")
  }
  
  return nil
}
```

## Example

Lets have some fun and write a simple example which calculates some statistics
on text given.

```go
const Text = `Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed
  do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim
  veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo
  consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum
  dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident,
  sunt in culpa qui officia deserunt mollit anim id est laborum.`

type myKey int
const keyStatistics myKey = 1

var sanitizeWordRe = regexp.MustCompile(`\W`)

type Statistics struct {
  Words      []string
  Characters int
  Occurrence map[string]int
}

// Split to words and sanitize them
SplitToWords := func(ctx floc.Context, ctrl floc.Control) error {
  statistics := ctx.Value(keyStatistics).(*Statistics)

  statistics.Words = strings.Split(Text, " ")
  for i, word := range statistics.Words {
    statistics.Words[i] = sanitizeWordRe.ReplaceAllString(word, "")
  }
  
  return nil
}

// Count and sum the number of characters in the each word
CountCharacters := func(ctx floc.Context, ctrl floc.Control) error {
  statistics := ctx.Value(keyStatistics).(*Statistics)

  for _, word := range statistics.Words {
    statistics.Characters += len(word)
  }
  
  return nil
}

// Count the number of unique words
CountUniqueWords := func(ctx floc.Context, ctrl floc.Control) error {
  statistics := ctx.Value(keyStatistics).(*Statistics)

  statistics.Occurrence = make(map[string]int)
  for _, word := range statistics.Words {
    statistics.Occurrence[word] = statistics.Occurrence[word] + 1
  }
  
  return nil
}

// Print result
PrintResult := func(ctx floc.Context, ctrl floc.Control) error {
  statistics := ctx.Value(keyStatistics).(*Statistics)

  fmt.Printf("Words Total       : %d\n", len(statistics.Words))
  fmt.Printf("Unique Word Count : %d\n", len(statistics.Occurrence))
  fmt.Printf("Character Count   : %d\n", statistics.Characters)
  
  return nil
}

// Design the flow and run it
flow := run.Sequence(
  SplitToWords,
  run.Parallel(
    CountCharacters,
    CountUniqueWords,
  ),
  PrintResult,
)

ctx := floc.NewContext()
ctx.AddValue(keyStatistics, new(Statistics))

ctrl := floc.NewControl(ctx)

_, _, err := floc.RunWith(ctx, ctrl, flow)
if err != nil {
	panic(err)
}

// Output:
// Words Total       : 64
// Unique Word Count : 60
// Character Count   : 370
```

## Contributing

Please found information about contributing in
[CONTRIBUTING.md](CONTRIBUTING.md)
and the list of braves who spent their priceless time and effort to make the project better in
[CONTRIBUTORS.md](CONTRIBUTORS.md).



go-floc

# Goflow

[![GoDoc](http://godoc.org/github.com/kamildrazkiewicz/go-flow?status.svg)](http://godoc.org/github.com/kamildrazkiewicz/go-flow) [![License](https://img.shields.io/github/license/mashape/apistatus.svg?maxAge=2592000)](https://raw.githubusercontent.com/kamildrazkiewicz/go-flow/master/LICENSE) [![Release](https://img.shields.io/github/release/kamildrazkiewicz/go-flow.svg?label=Release)](http://godoc.org/github.com/kamildrazkiewicz/go-flow) [![GoReport](https://goreportcard.com/badge/github.com/kamildrazkiewicz/go-flow)](https://goreportcard.com/report/github.com/kamildrazkiewicz/go-flow) [![Travis](https://travis-ci.org/kamildrazkiewicz/go-flow.svg?branch=master)](https://travis-ci.org/kamildrazkiewicz/go-flow) [![Coverage](http://coveralls.io/repos/github/kamildrazkiewicz/go-flow/badge.svg?branch=master)](https://coveralls.io/github/kamildrazkiewicz/go-flow?branch=master)

Goflow is a simply package to control goroutines execution order based on dependencies. It works similar to ```async.auto``` from [node.js async package](https://github.com/caolan/async), but for Go.

![](http://i.imgur.com/Rej4XAC.png)

## Install

Install the package with:

```bash
go get github.com/kamildrazkiewicz/go-flow
```

Import it with:

```go
import "github.com/kamildrazkiewicz/go-flow"
```

and use `goflow` as the package name inside the code.

## Example

```go
package main

import (
	"fmt"
	"github.com/kamildrazkiewicz/go-flow"
	"time"
)

func main() {
	f1 := func(r map[string]interface{}) (interface{}, error) {
		fmt.Println("function1 started")
		time.Sleep(time.Millisecond * 1000)
		return 1, nil
	}

	f2 := func(r map[string]interface{}) (interface{}, error) {
		time.Sleep(time.Millisecond * 1000)
		fmt.Println("function2 started", r["f1"])
		return "some results", nil
	}

	f3 := func(r map[string]interface{}) (interface{}, error) {
		fmt.Println("function3 started", r["f1"])
		return nil, nil
	}

	f4 := func(r map[string]interface{}) (interface{}, error) {
		fmt.Println("function4 started", r)
		return nil, nil
	}

	res, err := goflow.New().
		Add("f1", nil, f1).
		Add("f2", []string{"f1"}, f2).
		Add("f3", []string{"f1"}, f3).
		Add("f4", []string{"f2", "f3"}, f4).
		Do()

	fmt.Println(res, err)
}


```

Output will be:
```
function1 started
function3 started 1
function2 started 1
function4 started map[f2:some results f3:<nil>]
map[f1:1 f2:some results f3:<nil> f4:<nil>] <nil>
```


go-flow

A highly performant and easy to use Goroutine pool for Go.

<p align="center"> 
    <img width="150" height="150" src="internal/static/GoHiveIcon.png" alt="">
    <h1 align="center">gohive</h1>
    <p align="center">Package gohive implements a simple and easy to use goroutine pool for Go<p>
    <p align="center"><a href="https://travis-ci.org/loveleshsharma/gohive"><img src="https://travis-ci.org/loveleshsharma/gohive.svg?branch=master" /></a>
    <a href="https://goreportcard.com/report/github.com/loveleshsharma/gohive"><img src="https://goreportcard.com/badge/github.com/loveleshsharma/gohive" /></a>
    <a href="https://codecov.io/gh/loveleshsharma/gohive"><img src="https://codecov.io/gh/loveleshsharma/gohive/branch/master/graph/badge.svg" /></a>
    <a href="https://godoc.org/github.com/loveleshsharma/gohive"><img src="https://godoc.org/github.com/loveleshsharma/gohive?status.svg" /></a>
    <a href="https://github.com/avelino/awesome-go#goroutines"><img src="https://awesome.re/mentioned-badge.svg" /></a>
    </p>
</p>

## Features

- Pool can be created with a specific size as per the requirement
- Accepts tasks which implements Runner interface
- Uses channels to accepts tasks and gets them executed via workers
- Uses synchronization among workers to avoid race conditions

## Installation
Use ```go get``` to install and update:
```go
$ go get -u github.com/loveleshsharma/gohive
```

## Usage

- Create an instance of Pool type first

```go
hive := gohive.NewFixedPool(5)
```

- Invoke the Submit() function and pass the task to execute

```go
hive.Submit(object Runner)
```
Submit function accepts a Runner object as an argument, which it passes to the pool if a worker is available, otherwise it will wait for the worker to be available

- To close the pool we can invoke the Close() function

```go
hive.Close()
```
Once the pool is closed, we cannot assign any task to it

## Example

Let's get into a full program where we can see how to use the gohive package in order to execute many goroutines simultaneously

```go
package main

import (
   "fmt"
   "github.com/loveleshsharma/gohive"
   "sync"
)

func main() {
   var wg sync.WaitGroup
   pool := gohive.NewFixedPool(5)

   for i := 1; i <= 20; i++ {
      if err := pool.Submit(NewMyStruct(i, &wg)); err != nil {
         fmt.Println("error: ", err)
         break
      }
   }

   wg.Wait()
}

type MyStruct struct {
   num int
   wg  *sync.WaitGroup
}

func NewMyStruct(num int, wg *sync.WaitGroup) MyStruct {
   myStruct := MyStruct{
      num: num,
      wg:  wg,
   }
   wg.Add(1)
   return myStruct
}

func (s MyStruct) Run() {
   defer s.wg.Done()
   val := s.num
   fact := s.num
   for i := s.num - 1; i > 0; i-- {
      fact *= i
   }

   fmt.Printf("Factorial of %d: %d\n", val, fact)
}


```
<B>Important : </B> Always keep sync.WaitGroup in your struct and put ```defer wg.Done()``` as the first statement of your Run() function. It will wait for your task to complete.

###
TODO
1. Maintain a waiting queue to stop blocking submit method when all goroutines are busy.
2. Submitting priority tasks which takes priority over other tasks.
3. Handling panics inside goroutines to prevent them from crashing.
4. Implement dynamic pool which will scale the number of goroutines as per requirement and scales down when they are idle.
5. Submitting multiple tasks together.

gohive

asynchronous simple function utilities, for managing execution of closures and callbacks.

⚙️ gollback
================
[![Build Status](https://travis-ci.org/vardius/gollback.svg?branch=master)](https://travis-ci.org/vardius/gollback)
[![Go Report Card](https://goreportcard.com/badge/github.com/vardius/gollback)](https://goreportcard.com/report/github.com/vardius/gollback)
[![codecov](https://codecov.io/gh/vardius/gollback/branch/master/graph/badge.svg)](https://codecov.io/gh/vardius/gollback)
[![](https://godoc.org/github.com/vardius/gollback?status.svg)](https://pkg.go.dev/github.com/vardius/gollback)
[![license](https://img.shields.io/github/license/mashape/apistatus.svg)](https://github.com/vardius/gollback/blob/master/LICENSE.md)

<img align="right" height="180px" src="https://github.com/vardius/gorouter/blob/master/website/src/static/img/logo.png?raw=true" alt="logo" />

gollback - Go asynchronous simple function utilities, for managing execution of closures and callbacks

📖 ABOUT
==================================================
Contributors:

* [Rafał Lorenz](http://rafallorenz.com)

Want to contribute ? Feel free to send pull requests!

Have problems, bugs, feature ideas?
We are using the github [issue tracker](https://github.com/vardius/gollback/issues) to manage them.

## 📚 Documentation

For __examples__ **visit [godoc#pkg-examples](http://godoc.org/github.com/vardius/gollback#pkg-examples)**

For **GoDoc** reference, **visit [pkg.go.dev](https://pkg.go.dev/github.com/vardius/gollback)**

🚏 HOW TO USE
==================================================

## 🚅 Benchmark
**CPU: 3,3 GHz Intel Core i7**

**RAM: 16 GB 2133 MHz LPDDR3**

```bash
➜  gollback git:(master) ✗ go test -bench=. -cpu=4 -benchmem
goos: darwin
goarch: amd64
pkg: github.com/vardius/gollback
BenchmarkRace-4    	  566022	      2608 ns/op	     663 B/op	       5 allocs/op
BenchmarkAll-4     	 5052489	       241 ns/op	      42 B/op	       1 allocs/op
BenchmarkRetry-4   	206430384	         5.93 ns/op	       0 B/op	       0 allocs/op
PASS
ok  	github.com/vardius/gollback	31.741s
```

## Race
> Race method returns a response as soon as one of the callbacks in an iterable resolves with the value that is not an error, otherwise last error is returne
```go
package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/vardius/gollback"
)

func main() {
	r, err := gollback.Race(
		context.Background(),
		func(ctx context.Context) (interface{}, error) {
			time.Sleep(3 * time.Second)
			return 1, nil
		},
		func(ctx context.Context) (interface{}, error) {
			return nil, errors.New("failed")
		},
		func(ctx context.Context) (interface{}, error) {
			return 3, nil
		},
	)
}
```

## All
> All method returns when all of the callbacks passed as an iterable have finished, returned responses and errors are ordered according to callback order
```go
package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/vardius/gollback"
)

func main() {
	rs, errs := gollback.All(
		context.Background(),
		func(ctx context.Context) (interface{}, error) {
			time.Sleep(3 * time.Second)
			return 1, nil
		},
		func(ctx context.Context) (interface{}, error) {
			return nil, errors.New("failed")
		},
		func(ctx context.Context) (interface{}, error) {
			return 3, nil
		},
	)
}
```

## Retry
> Retry method retries callback given amount of times until it executes without an error, when retries = 0 it will retry infinitely
```go
package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/vardius/gollback"
)

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	// Will retry infinitely until timeouts by context (after 5 seconds)
	res, err := gollback.Retry(ctx, 0, func(ctx context.Context) (interface{}, error) {
		return nil, errors.New("failed")
	})

	// Will retry 5 times or will timeout by context (after 5 seconds)
	res, err := gollback.Retry(ctx, 5, func(ctx context.Context) (interface{}, error) {
		return nil, errors.New("failed")
	})
}
```

📜 [License](LICENSE.md)
-------

This package is released under the MIT license. See the complete license in the package


gollback

Gowl is a process management and process monitoring tool at once. An infinite worker pool gives you the ability to control the pool and processes and monitor their status.

# Gowl

![build](https://github.com/hmdsefi/gowl/actions/workflows/build.yml/badge.svg)
[![codecov](https://codecov.io/gh/hmdsefi/gowl/branch/master/graph/badge.svg?token=1TYYX8IBR0)](https://codecov.io/gh/hmdsefi/gowl)
[![Go Report Card](https://goreportcard.com/badge/github.com/hamed-yousefi/gowl)](https://goreportcard.com/report/github.com/hamed-yousefi/gowl)
[![FOSSA Status](https://app.fossa.com/api/projects/custom%2B24403%2Fgithub.com%2Fhamed-yousefi%2Fgowl.svg?type=shield)](https://app.fossa.com/projects/custom%2B24403%2Fgithub.com%2Fhamed-yousefi%2Fgowl?ref=badge_shield)
[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go) 
[![Go Reference](https://pkg.go.dev/badge/github.com/hamed-yousefi/gowl.svg)](https://pkg.go.dev/github.com/hamed-yousefi/gowl)
<div  align="center"><img src="https://github.com/hamed-yousefi/gowl/blob/master/docs/images/process-pool.png" width="450" ></div>
Gowl is a process management and process monitoring tool at once.
An infinite worker pool gives you the ability to control the pool and processes
and monitor their status.

## Table of Contents

* [Install](#Install)
* [How to use](#How-to-use)
    * [Pool](#Pooling)
      * [Start](#Start)
      * [Register process](#Register-process)
      * [Kill process](#Kill-process)
      * [Close](#Close)
    * [Monitor](#Monitor)
* [License](#License)

## Install

Using Gowl is easy. First, use `go get` to install the latest version of the library. This command will install
the `gowl` along with library and its dependencies:

```shell
go get -u github.com/hamed-yousefi/gowl
```

Next, include Gowl in your application:

```go
import "github.com/hamed-yousefi/gowl"
```

## How to use

Gowl has three main parts. Process, Pool, and Monitor. The process is the smallest part of this project. The process is
the part of code that the developer must implement. To do that, Gowl provides an interface to inject outside code into
the pool. The process interface is as follows:

```go
Process interface {
Start(ctx context.Context) error
Name() string
PID() PID
}
```

The process interface has three methods. The Start function contains the user codes, and the pool workers use this
function to run the process. The Name function returns the process name, and the monitor uses this function to provide
reports. The PID function returns process id. The process id is unique in the entire pool, and it will use by the pool
and monitor.

If properly done it should respond to the context cancellation (if needed)

Let's take a look at an example:

```go
Document struct {
   content string
   hash string
}

func (d *Document) Start(ctx context.Context) error {
   hasher := sha1.New()
   hasher.Write(bv)
   h.hash = base64.URLEncoding.EncodeToString(hasher.Sum(nil))
}

func (d *Document) Name() string {
   return "hashing-process"
}

func (d *Document) PID() PID {
   return "p-1"
}

func (d *Document) Hash() string {
   return h.hash
}
```

As you can see, in this example, `Document` implements the Process interface. So now we can register it into the pool.

### Pool

Creating Gowl pool is very easy. You must use the `NewPool(size int)`
function and pass the pool size to this function. Pool size indicates the worker numbers in and the underlying queue
size that workers consume process from it. Look at the following example:

```go
pool := gowl.NewPool(4)
```

In this example, Gowl will create a new instance of a Pool object with four workers and an underlying queue with the
size of four.

#### Start

To start the Gowl, you must call the `Start()` method of the pool object. It will begin to create the workers, and
workers start listening to the queue to consume process.

#### Register process

To register processes to the pool, you must use the `Register(args ...process)`
method. Pass the processes to the register method, and it will create a new publisher to publish the process list to the
queue. You can call multiple times when Gowl pool is running.

#### Kill process

One of the most remarkable features of Gowl is the ability to control the process after registered it into the pool. You
can kill a process before any worker runs it, this also works after the job have started if it consideres the context
cancellation. Killing a process is simple, and you need the process id to do it.

```go
pool.Kill(PID("p-909"))
```

#### Close

Gowl is an infinite worker pool. However, you should have control over the pool and decide when you want to start it,
register a new process on it, kill a process, and `close` the pool and terminate the workers. Gowl gives you this option
to close the pool by the `Close()` method of the Pool object.

## Monitor

Every process management tool needs a monitoring system to expose the internal stats to the outside world. Gowl gives
you a monitoring API to see processes and workers stats.

You can get the Monitor instance by calling the `Monitor()` method of the Pool. The monitor object is as follows:

```go
Monitor interface {
   PoolStatus() pool.Status
   Error(PID) error
   WorkerList() []WorkerName
   WorkerStatus(name WorkerName) worker.Status
   ProcessStats(pid PID) ProcessStats
}
```

The Monitor gives you this opportunity to get the Pool status, process error, worker list, worker status, and process
stats. Wis Monitor API, you can create your monitoring app with ease. The following example is using Monitor API to
present the stats in the console in real-time.

![process-monitoring](https://github.com/hamed-yousefi/gowl/blob/master/docs/images/process-monitoring.gif)

Also, you can use the Monitor API to show worker status in the console:

![worker-monitoring](https://github.com/hamed-yousefi/gowl/blob/master/docs/images/worker-monitoring.gif)

## License

MIT License, please see [LICENSE](https://github.com/hamed-yousefi/gowl/blob/master/LICENSE) for details.


gowl

goworker is a Go-based background worker.

# goworker

![Build](https://github.com/benmanns/goworker/workflows/Go/badge.svg)
[![GoDoc](https://godoc.org/github.com/benmanns/goworker?status.svg)](https://godoc.org/github.com/benmanns/goworker)

goworker is a Resque-compatible, Go-based background worker. It allows you to push jobs into a queue using an expressive language like Ruby while harnessing the efficiency and concurrency of Go to minimize job latency and cost.

goworker workers can run alongside Ruby Resque clients so that you can keep all but your most resource-intensive jobs in Ruby.

## Installation

To install goworker, use

```sh
go get github.com/benmanns/goworker
```

to install the package, and then from your worker

```go
import "github.com/benmanns/goworker"
```

## Getting Started

To create a worker, write a function matching the signature

```go
func(string, ...interface{}) error
```

and register it using

```go
goworker.Register("MyClass", myFunc)
```

Here is a simple worker that prints its arguments:

```go
package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func myFunc(queue string, args ...interface{}) error {
	fmt.Printf("From %s, %v\n", queue, args)
	return nil
}

func init() {
	goworker.Register("MyClass", myFunc)
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}
```

To create workers that share a database pool or other resources, use a closure to share variables.

```go
package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func newMyFunc(uri string) (func(queue string, args ...interface{}) error) {
	foo := NewFoo(uri)
	return func(queue string, args ...interface{}) error {
		foo.Bar(args)
		return nil
	}
}

func init() {
	goworker.Register("MyClass", newMyFunc("http://www.example.com/"))
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}
```

Here is a simple worker with settings:

```go
package main

import (
	"fmt"
	"github.com/benmanns/goworker"
)

func myFunc(queue string, args ...interface{}) error {
	fmt.Printf("From %s, %v\n", queue, args)
	return nil
}

func init() {
	settings := goworker.WorkerSettings{
		URI:            "redis://localhost:6379/",
		Connections:    100,
		Queues:         []string{"myqueue", "delimited", "queues"},
		UseNumber:      true,
		ExitOnComplete: false,
		Concurrency:    2,
		Namespace:      "resque:",
		Interval:       5.0,
	}
	goworker.SetSettings(settings)
	goworker.Register("MyClass", myFunc)
}

func main() {
	if err := goworker.Work(); err != nil {
		fmt.Println("Error:", err)
	}
}
```

goworker worker functions receive the queue they are serving and a slice of interfaces. To use them as parameters to other functions, use Go type assertions to convert them into usable types.

```go
// Expecting (int, string, float64)
func myFunc(queue, args ...interface{}) error {
	idNum, ok := args[0].(json.Number)
	if !ok {
		return errorInvalidParam
	}
	id, err := idNum.Int64()
	if err != nil {
		return errorInvalidParam
	}
	name, ok := args[1].(string)
	if !ok {
		return errorInvalidParam
	}
	weightNum, ok := args[2].(json.Number)
	if !ok {
		return errorInvalidParam
	}
	weight, err := weightNum.Float64()
	if err != nil {
		return errorInvalidParam
	}
	doSomething(id, name, weight)
	return nil
}
```

For testing, it is helpful to use the `redis-cli` program to insert jobs onto the Redis queue:

```sh
redis-cli -r 100 RPUSH resque:queue:myqueue '{"class":"MyClass","args":["hi","there"]}'
```

will insert 100 jobs for the `MyClass` worker onto the `myqueue` queue. It is equivalent to:

```ruby
class MyClass
  @queue = :myqueue
end

100.times do
  Resque.enqueue MyClass, ['hi', 'there']
end
```

or

```golang
goworker.Enqueue(&goworker.Job{
    Queue: "myqueue",
    Payload: goworker.Payload{
        Class: "MyClass",
        Args: []interface{}{"hi", "there"},
    },
})
```

## Flags

There are several flags which control the operation of the goworker client.

* `-queues="comma,delimited,queues"` — This is the only required flag. The recommended practice is to separate your Resque workers from your goworkers with different queues. Otherwise, Resque worker classes that have no goworker analog will cause the goworker process to fail the jobs. Because of this, there is no default queue, nor is there a way to select all queues (à la Resque's `*` queue). If you have multiple queues you can assign them weights. A queue with a weight of 2 will be checked twice as often as a queue with a weight of 1: `-queues='high=2,low=1'`.
* `-interval=5.0` — Specifies the wait period between polling if no job was in the queue the last time one was requested.
* `-concurrency=25` — Specifies the number of concurrently executing workers. This number can be as low as 1 or rather comfortably as high as 100,000, and should be tuned to your workflow and the availability of outside resources.
* `-connections=2` — Specifies the maximum number of Redis connections that goworker will consume between the poller and all workers. There is not much performance gain over two and a slight penalty when using only one. This is configurable in case you need to keep connection counts low for cloud Redis providers who limit plans on `maxclients`.
* `-uri=redis://localhost:6379/` — Specifies the URI of the Redis database from which goworker polls for jobs. Accepts URIs of the format `redis://user:pass@host:port/db` or `unix:///path/to/redis.sock`. The flag may also be set by the environment variable `$($REDIS_PROVIDER)` or `$REDIS_URL`. E.g. set `$REDIS_PROVIDER` to `REDISTOGO_URL` on Heroku to let the Redis To Go add-on configure the Redis database.
* `-namespace=resque:` — Specifies the namespace from which goworker retrieves jobs and stores stats on workers.
* `-exit-on-complete=false` — Exits goworker when there are no jobs left in the queue. This is helpful in conjunction with the `time` command to benchmark different configurations.

You can also configure your own flags for use within your workers. Be sure to set them before calling `goworker.Main()`. It is okay to call `flags.Parse()` before calling `goworker.Main()` if you need to do additional processing on your flags.

## Signal Handling in goworker

To stop goworker, send a `QUIT`, `TERM`, or `INT` signal to the process. This will immediately stop job polling. There can be up to `$CONCURRENCY` jobs currently running, which will continue to run until they are finished.

## Failure Modes

Like Resque, goworker makes no guarantees about the safety of jobs in the event of process shutdown. Workers must be both idempotent and tolerant to loss of the job in the event of failure.

If the process is killed with a `KILL` or by a system failure, there may be one job that is currently in the poller's buffer that will be lost without any representation in either the queue or the worker variable.

If you are running goworker on a system like Heroku, which sends a `TERM` to signal a process that it needs to stop, ten seconds later sends a `KILL` to force the process to stop, your jobs must finish within 10 seconds or they may be lost. Jobs will be recoverable from the Redis database under

```
resque:worker:<hostname>:<process-id>-<worker-id>:<queues>
```

as a JSON object with keys `queue`, `run_at`, and `payload`, but the process is manual. Additionally, there is no guarantee that the job in Redis under the worker key has not finished, if the process is killed before goworker can flush the update to Redis.

## Contributing

1. [Fork it](https://github.com/benmanns/goworker/fork)
2. Create your feature branch (`git checkout -b my-new-feature`)
3. Commit your changes (`git commit -am 'Add some feature'`)
4. Push to the branch (`git push origin my-new-feature`)
5. Create new Pull Request


goworker

gowp is concurrency limiting goroutine pool.

[![Build Status](https://travis-ci.org/xxjwxc/gowp.svg?branch=master)](https://travis-ci.org/xxjwxc/gowp)
[![Go Report Card](https://goreportcard.com/badge/github.com/xxjwxc/gowp)](https://goreportcard.com/report/github.com/xxjwxc/gowp)
[![codecov](https://codecov.io/gh/xxjwxc/gowp/branch/master/graph/badge.svg)](https://codecov.io/gh/xxjwxc/gowp)
[![GoDoc](https://godoc.org/github.com/xxjwxc/gowp?status.svg)](https://godoc.org/github.com/xxjwxc/gowp)
[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go)
<a href="https://hellogithub.com/repository/0f15a9cd05d14fec9e9ceafa2f5ae984" target="_blank"><img src="https://abroad.hellogithub.com/v1/widgets/recommend.svg?rid=0f15a9cd05d14fec9e9ceafa2f5ae984&claim_uid=6IkmGoxiuN4LWFn&theme=small" alt="Featured｜HelloGitHub" /></a>

## golang worker pool

### [中文说明](README_cn.md)

- Concurrency limiting goroutine pool. 
- Limits the concurrency of task execution, not the number of tasks queued. 
- Never blocks submitting tasks, no matter how many tasks are queued.
- Support timeout
- Support through security queues [queue](https://github.com/xxjwxc/public/tree/master/myqueue)


## Installation

The simplest way to install the library is to run:

```
$ go get github.com/xxjwxc/gowp
```


### Support the maximum number of tasks, put them in the workpool and wait for them to be completed

## Example

```go
package main

import (
	"fmt"
	"time"

	"github.com/xxjwxc/gowp/workpool"
)

func main() {
	wp := workpool.New(10)     // Set the maximum number of threads
	for i := 0; i < 20; i++ { // Open 20 requests 
		ii := i
		wp.Do(func() error {
			for j := 0; j < 10; j++ { // 0-10 values per print
				fmt.Println(fmt.Sprintf("%v->\t%v", ii, j))
				time.Sleep(1 * time.Second)
			}
			//time.Sleep(1 * time.Second)
			return nil
		})
	}

	wp.Wait()
	fmt.Println("down")
}

```

### Support for error return

```go
package main

import (
	"fmt"
	"time"

	"github.com/xxjwxc/gowp/workpool"
)

func main() {
	wp := workpool.New(10)             // Set the maximum number of threads
	for i := 0; i < 20; i++ { 
		ii := i
		wp.Do(func() error {
			for j := 0; j < 10; j++ { // 0-10 values per print
				fmt.Println(fmt.Sprintf("%v->\t%v", ii, j))
				if ii == 1 {
					return errors.Cause(errors.New("my test err")) // have err return
				}
				time.Sleep(1 * time.Second)
			}

			return nil
		})
	}

	err := wp.Wait()
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println("down")
	}
```

### Supporting judgement of completion (non-blocking)

```go
package main

import (
	"fmt"
	"time"

	"github.com/xxjwxc/gowp/workpool"
)

func main() {
	wp := workpool.New(5)              // Set the maximum number of threads
	for i := 0; i < 10; i++ { 
		//	ii := i
		wp.Do(func() error {
			for j := 0; j < 5; j++ { 
				//fmt.Println(fmt.Sprintf("%v->\t%v", ii, j))
				time.Sleep(1 * time.Second)
			}
			return nil
		})

		fmt.Println(wp.IsDone())
	}
	wp.Wait()
	fmt.Println(wp.IsDone())
	fmt.Println("down")
}
```

### Support synchronous waiting for results

```go
package main

import (
	"fmt"
	"time"

	"github.com/xxjwxc/gowp/workpool"
)

func main() {
	wp := workpool.New(5) // Set the maximum number of threads
	for i := 0; i < 10; i++ { 
		ii := i
		wp.DoWait(func() error {
			for j := 0; j < 5; j++ { 
				fmt.Println(fmt.Sprintf("%v->\t%v", ii, j))
				// if ii == 1 {
				// 	return errors.New("my test err")
				// }
				time.Sleep(1 * time.Second)
			}

			return nil
			//time.Sleep(1 * time.Second)
			//return errors.New("my test err")
		})
	}

	err := wp.Wait()
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println("down")
}
```

### Support timeout exit

```go
package main

import (
	"fmt"
	"time"
	"time"
	"github.com/xxjwxc/gowp/workpool"
)

func main() {
	wp := workpool.New(5)              // Set the maximum number of threads
		wp.SetTimeout(time.Millisecond) // set max timeout
	for i := 0; i < 10; i++ { 
		ii := i
		wp.DoWait(func() error {
			for j := 0; j < 5; j++ {
				fmt.Println(fmt.Sprintf("%v->\t%v", ii, j))
				time.Sleep(1 * time.Second)
			}

			return nil
		})
	}

	err := wp.Wait()
	if err != nil {
		fmt.Println(err)
	}
	fmt.Println("down")
}
```


## limiter(cache)

```go
package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/xxjwxc/gowp/limiter"
)

func main() {
	limiter := limiter.NewLimiter(limiter.WithLimit(10), limiter.WithNamespace("test"), limiter.WithTsTimeout(true) /*, limiter.WithRedis(res)*/)
	var wg sync.WaitGroup
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			token, _ := limiter.Acquire(10) // get
			fmt.Println(token)

			time.Sleep(1 * time.Second)
			limiter.Release(token) 
		}()
	}
	wg.Wait()
	fmt.Println("down")
}
```
## limiter(redis)

```go
package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/xxjwxc/gowp/limiter"
	"github.com/xxjwxc/public/myredis"
)

func main() {
	conf := myredis.InitRedis(myredis.WithAddr("127.0.0.1:6379"), myredis.WithPwd("123456"), myredis.WithGroupName("test"))
	res, err := myredis.NewRedis(conf)
	if err != nil {
		fmt.Println(err)
		return
	}

	limiter := limiter.NewLimiter(limiter.WithRedis(res), limiter.WithLimit(10), limiter.WithNamespace("test") /*, limiter.WithRedis(res)*/)
	var wg sync.WaitGroup
	for i := 0; i < 20; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			token, _ := limiter.Acquire(10) 
			fmt.Println(token)

			time.Sleep(1 * time.Second)
			limiter.Release(token) 
		}()
	}
	wg.Wait()
	fmt.Println("down")
}

```


gowp

manages a resizeable pool of context-aware goroutines to bound concurrency.

# gpool - a generic context-aware resizable goroutines pool to bound concurrency.


[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go#goroutines)
<a>
  <img src="https://img.shields.io/github/v/tag/sherifabdlnaby/gpool?label=release&amp;sort=semver">
</a>
[![](https://godoc.org/github.com/sherifabdlnaby/gpool?status.svg)](http://godoc.org/github.com/sherifabdlnaby/gpool)
[![Go Report Card](https://goreportcard.com/badge/github.com/sherifabdlnaby/gpool)](https://goreportcard.com/report/github.com/sherifabdlnaby/gpool)
[![Build Status](https://travis-ci.org/sherifabdlnaby/gpool.svg?branch=func)](https://travis-ci.org/sherifabdlnaby/gpool)
[![codecov](https://codecov.io/gh/sherifabdlnaby/gpool/branch/func/graph/badge.svg)](https://codecov.io/gh/sherifabdlnaby/gpool)
[![License](https://img.shields.io/badge/License-MIT-blue.svg)](https://github.com/SherifAbdlNaby/gpool/blob/master/LICENSE)

## Installation
``` bash
$ go get github.com/sherifabdlnaby/gpool
```
``` go
import "github.com/sherifabdlnaby/gpool"
```

## Introduction

Easily manages a resizeable pool of context aware goroutines to bound concurrency, A **`Job`** is **Enqueued** to the pool and only **`N`** jobs can be processing concurrently at the same time.

- A Job is simply a `func(){}`,  when you `Enqueue(..)` a job, the enqueue call will return *ONCE* the job has **started** processing. Otherwise if the pool is full it will **block** until:
  1. pool has room for the job.
  2. job's `context` is canceled.
  3. the pool is stopped.

- The Pool can be re-sized using `Resize()` that will resize the pool in a concurrent safe-way. `Resize` can enlarge the pool and any blocked enqueue will unblock after pool is resized, in case of shrinking the pool `resize` will not affect any already processing/waiting jobs.

- Enqueuing a Job will return error `nil` once a job starts, `ErrPoolClosed` if the pool is closed, or **the context's error if the job's context is canceled while blocking waiting for the pool.**

- Stopping the Pool using `pool.Stop()` will **wait** for all processing jobs to finish before returning, it will also unblock any **blocked** job enqueues (enqueues will return ErrPoolClosed).

- `Start`, `Stop`, and `Resize(N)` are all concurrent safe and can be called from multiple goroutines, subsequent calls of Start or Stop has no effect unless called interchangeably.

further documentation at : [![](https://godoc.org/github.com/sherifabdlnaby/gpool?status.svg)](http://godoc.org/github.com/sherifabdlnaby/gpool)

------------------------------------------------------

## Usage

- Create new pool
    ```
    pool, err := gpool.NewPool(concurrency)
    ```
- Enqueue a job
    ```
        job := func() {
            time.Sleep(2000 * time.Millisecond)
            fmt.Println("did some work")
        }

        // Enqueue Job
        err := pool.Enqueue(ctx, job)
    ```
    A call to `pool.Enqueue()` will return `nil` if `job` started processing, blocks if the pool is full, `ctx.Err()` if context was canceled while waiting/blocking, or finally `ErrPoolClosed` if the pool stopped or was never started.
- Resize the pool
    ```
    err = pool.Resize(size)
    ```
    Will live change the size of the pool, If new size is larger, waiting job enqueues from another goroutines will be unblocked to fit the new size, and if new size is smaller, any new enqueues will block until the current size of the pool is less than the new one.
- Stop the pool
    ```
    pool.Stop()
    ```
    - ALL Blocked/Waiting jobs will return immediately.

    - Stop() WILL Block until all running jobs is done.

- Different types of "Enqueues"
    - `Enqueue(ctx, job)`          returns ONCE the job has started executing (not after job finishes/return)

    - `EnqueueAndWait(ctx, job)`   returns ONCE the job has started **and** finished executing.

    - `TryEnqueue(job)`            will not block if the pool is full, returns `true` ONCE the job has started executing and `false` if pool is full.

    - `TryEnqueueAndWait(job)`     will not block if the pool is full, returns `true` ONCE the job has started **and** finished executing. and `false` if pool is full.


------------------------------------------------------

## Benchmarks

``` bash
$ go test -bench=. -cpu=2 -benchmem
```

```
go test -bench=. -cpu=2 -benchmem
goos: darwin
goarch: amd64
pkg: github.com/sherifabdlnaby/gpool
BenchmarkThroughput/PoolSize[2]-2                 853724               730 ns/op             125 B/op          2 allocs/op
BenchmarkThroughput/PoolSize[10]-2               3647638               329 ns/op              10 B/op          0 allocs/op
BenchmarkThroughput/PoolSize[100]-2              4869789               248 ns/op               0 B/op          0 allocs/op
BenchmarkThroughput/PoolSize[1000]-2             4320566               280 ns/op               0 B/op          0 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[2]BulkJobs[2]-2             530328              2146 ns/op             275 B/op          5 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[2]BulkJobs[100]-2            10000            112478 ns/op           12737 B/op        239 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[2]BulkJobs[1000]-2            1069           1109384 ns/op          126943 B/op       2380 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[10]BulkJobs[2]-2           1417808               844 ns/op              58 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[10]BulkJobs[100]-2           30556             39187 ns/op            1764 B/op         33 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[10]BulkJobs[1000]-2           3012            385652 ns/op           15737 B/op        295 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[100]BulkJobs[2]-2          1986571               609 ns/op              16 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[100]BulkJobs[100]-2          41235             29004 ns/op              37 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[100]BulkJobs[1000]-2          4080            290791 ns/op             188 B/op          4 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[1000]BulkJobs[2]-2         1963564               605 ns/op              16 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[1000]BulkJobs[100]-2         42000             28442 ns/op              16 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[1000]BulkJobs[1000]-2         4333            284865 ns/op              17 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[10000]BulkJobs[2]-2        1963168               611 ns/op              16 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[10000]BulkJobs[100]-2        42238             28419 ns/op              16 B/op          1 allocs/op
BenchmarkBulkJobs_UnderLimit/PoolSize[10000]BulkJobs[1000]-2        4171            283981 ns/op              29 B/op          1 allocs/op
```


**BenchmarkOneThroughput/PoolSize[S]**                    = Enqueue Async Jobs ( Will not wait for result ) in a Pool of size = `S`

**BenchmarkBulkJobs/PoolSize[S]BulkJobs[J]**              = Enqueue `J` Jobs In Pool of size `S` at a time where `J` < `S`


------------------------------------------------------

## Examples

### Example 1 - Simple Job Enqueue
```go
func main() {
    concurrency := 2

    // Create and start pool.
    pool, _ := gpool.NewPool(concurrency)
    defer pool.Stop()

    // Create JOB
    resultChan1 := make(chan int)
    ctx := context.Background()
    job := func() {
        time.Sleep(2000 * time.Millisecond)
        resultChan1 <- 1337
    }

    // Enqueue Job
    err1 := pool.Enqueue(ctx, job)
    if err1 != nil {
        log.Printf("Job was not enqueued. Error: [%s]", err1.Error())
        return
    }

    log.Printf("Job Enqueued and started processing")
    log.Printf("Job Done, Received: %v", <-resultChan1)
}
```
----------

### Example 2 - Enqueue A Job with Timeout
```go

func main() {
  concurrency := 2

  // Create and start pool.
  pool, _ := gpool.NewPool(concurrency)
  defer pool.Stop()

  // Create JOB
  resultChan := make(chan int)
  ctx := context.Background()
  job := func() {
    resultChan <- 1337
  }

  // Enqueue 2 Jobs to fill pool (Will not finish unless we pull result from resultChan)
  _ = pool.Enqueue(ctx, job)
  _ = pool.Enqueue(ctx, job)


  ctxWithTimeout, _ := context.WithTimeout(ctx, 1000 * time.Millisecond)

  // Will block for 1 second only because of Timeout
  err1 := pool.Enqueue(ctxWithTimeout, job)

  if err1 != nil {
    log.Printf("Job was not enqueued. Error: [%s]", err1.Error())
  }

  log.Printf("Job 1 Done, Received: %v", <-resultChan)
  log.Printf("Job 2 Done, Received: %v", <-resultChan)
}
```

----------

### Example 3
``` go
// size Workers / Concurrent jobs of the Pool
const concurrency = 2

func main() {
  pool, _ := gpool.NewPool(concurrency)
  defer pool.Stop()

  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()

  go func() {
    for i := 0; i < 10; i++ {

      // Small Interval for more readable output
      time.Sleep(500 * time.Millisecond)

      go func(i int) {
        x := make(chan int, 1)

        log.Printf("Job [%v] Enqueueing", i)

        err := pool.Enqueue(ctx, func() {
          time.Sleep(2000 * time.Millisecond)
          x <- i
        })

        if err != nil {
          log.Printf("Job [%v] was not enqueued. [%s]", i, err.Error())
          return
        }

        log.Printf("Job [%v] Enqueue-ed ", i)

        log.Printf("Job [%v] Receieved, Result: [%v]", i, <-x)
      }(i)
    }
  }()

  // Uncomment to demonstrate ctx cancel of jobs.
  //time.Sleep(100 * time.Millisecond)
  //cancel()

  time.Sleep(5000 * time.Millisecond)

  fmt.Println("Stopping...")

  pool.Stop()

  fmt.Println("Stopped")

  fmt.Println("Sleeping for couple of seconds so canceled job have a chance to print out their status")

  time.Sleep(4000 * time.Millisecond)
}
```
#### Output
```
2019/01/08 20:15:39 Job [0] Enqueueing
2019/01/08 20:15:39 Job [0] Enqueue-ed
2019/01/08 20:15:39 Job [1] Enqueueing
2019/01/08 20:15:39 Job [1] Enqueue-ed
2019/01/08 20:15:40 Job [2] Enqueueing
2019/01/08 20:15:40 Job [3] Enqueueing
2019/01/08 20:15:41 Job [0] Receieved, Result: [0]
2019/01/08 20:15:41 Job [2] Enqueue-ed
2019/01/08 20:15:41 Job [4] Enqueueing
2019/01/08 20:15:41 Job [3] Enqueue-ed
2019/01/08 20:15:41 Job [1] Receieved, Result: [1]
2019/01/08 20:15:41 Job [5] Enqueueing
2019/01/08 20:15:42 Job [6] Enqueueing
2019/01/08 20:15:42 Job [7] Enqueueing
2019/01/08 20:15:43 Job [4] Enqueue-ed
2019/01/08 20:15:43 Job [2] Receieved, Result: [2]
2019/01/08 20:15:43 Job [8] Enqueueing
Stopping...
2019/01/08 20:15:43 Job [7] was not enqueued. [pool is closed]
2019/01/08 20:15:43 Job [5] was not enqueued. [pool is closed]
2019/01/08 20:15:43 Job [6] was not enqueued. [pool is closed]
2019/01/08 20:15:43 Job [3] Receieved, Result: [3]
2019/01/08 20:15:43 Job [8] was not enqueued. [pool is closed]
2019/01/08 20:15:43 Job [9] Enqueueing
Stopped
2019/01/08 20:15:45 Job [4] Receieved, Result: [4]
Sleeping for couple of seconds so canceled job have a chance to print out their status
2019/01/08 20:15:45 Job [9] was not enqueued. [pool is closed]

Process finished with exit code 0
```

# License
[MIT License](https://raw.githubusercontent.com/sherifabdlnaby/gpool/blob/master/LICENSE)
Copyright (c) 2019 Sherif Abdel-Naby

# Contribution

PR(s) are Open and Welcomed.

gpool

# grpool
[![Build Status](https://travis-ci.org/ivpusic/grpool.svg?branch=master)](https://travis-ci.org/ivpusic/grpool)

Lightweight Goroutine pool

Clients can submit jobs. Dispatcher takes job, and sends it to first available worker.
When worker is done with processing job, will be returned back to worker pool.

Number of workers and Job queue size is configurable.

## Docs
https://godoc.org/github.com/ivpusic/grpool

## Installation
```
go get github.com/ivpusic/grpool
```

## Simple example
```Go
package main

import (
  "fmt"
  "runtime"
  "time"

  "github.com/ivpusic/grpool"
)

func main() {
  // number of workers, and size of job queue
  pool := grpool.NewPool(100, 50)

  // release resources used by pool
  defer pool.Release()

  // submit one or more jobs to pool
  for i := 0; i < 10; i++ {
    count := i

    pool.JobQueue <- func() {
      fmt.Printf("I am worker! Number %d\n", count)
    }
  }

  // dummy wait until jobs are finished
  time.Sleep(1 * time.Second)
}
```

## Example with waiting jobs to finish
```Go
package main

import (
  "fmt"
  "runtime"

  "github.com/ivpusic/grpool"
)

func main() {
  // number of workers, and size of job queue
  pool := grpool.NewPool(100, 50)
  defer pool.Release()

  // how many jobs we should wait
  pool.WaitCount(10)

  // submit one or more jobs to pool
  for i := 0; i < 10; i++ {
    count := i

    pool.JobQueue <- func() {
      // say that job is done, so we can know how many jobs are finished
      defer pool.JobDone()

      fmt.Printf("hello %d\n", count)
    }
  }

  // wait until we call JobDone for all jobs
  pool.WaitAll()
}
```

## License
*MIT*


grpool

A process controller used to control the execution and return strategies of multiple goroutines.

# Hands

[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go) 
[![Go Report Card](https://goreportcard.com/badge/github.com/duanckham/hands)](https://goreportcard.com/report/github.com/duanckham/hands)
[![Build Status](https://travis-ci.com/duanckham/hands.svg?branch=master)](https://travis-ci.com/duanckham/hands)
[![codecov](https://codecov.io/gh/duanckham/hands/branch/master/graph/badge.svg)](https://codecov.io/gh/duanckham/hands)
[![LICENSE](https://img.shields.io/github/license/duanckham/hands.svg)](https://github.com/duanckham/hands/blob/master/LICENSE)

#### “Dedicated to Brother Chang”

Hands is a process controller used to control the execution and return strategies of multiple goroutines.

## Getting started

### A simple example

```go
n := 0
controller := hands.New()

controller.Do(func(ctx context.Context) error {
  n++
  return nil
})

err := controller.Run()
if err != nil {
  // ...
}

fmt.Println(n)

// Output:
// 1
```

Use the `Do` method to add a task, use the `Run` method to start the task(s).

## TaskOption

`TaskOption` is used to set some metadata for the task.

### `func Priority(priority int32) TaskOption`

Use the `Priority` method to set a priority for a task. The higher the priority, the higher the execution order.

(`hands.P()` is an alias for `hands.Priority()`.)

```go
controller := New()

controller.Do(func(ctx context.Context) error {
  fmt.Println("3")
  return nil
}, hands.P(1))

controller.Do(func(ctx context.Context) error {
  fmt.Println("2")
  return nil
}, hands.P(2))

controller.Do(func(ctx context.Context) error {
  fmt.Println("1")
  return nil
}, hands.P(3))

controller.Run()

// Output:
// 1
// 2
// 3
```

## HandOption

HandOption is used to control the execution strategy of the task.

### `func Fastest() HandOption`

`Fastest()`: When a task is completed, return immediately.

```go
n := 0
controller := hands.New()

controller.Do(func(ctx context.Context) error {
  time.Sleep(time.Duration(10) * time.Millisecond)
  n += 1
  return nil
})

controller.Do(func(ctx context.Context) error {
  n += 2
  return nil
})

controller.Run(hands.Fastest())

fmt.Println(n)

// Output:
// 2
```

### `func Percentage(percentage float32) HandOption`

When a certain percentage of tasks are executed, the results are returned.

```go
n := 0
controller := hands.New()

controller.Do(func(ctx context.Context) error {
  n++
  return nil
})

controller.Do(func(ctx context.Context) error {
  n++
  return nil
})

controller.Do(func(ctx context.Context) error {
  n++
  return nil
})

controller.Do(func(ctx context.Context) error {
  n++
  return nil
})

controller.Run(hands.Percentage(0.5))

fmt.Println(n)

// Output:
// 2
```

### `func Between(l, r int32) HandOption`

`Between()`: Only execute tasks with a priority within the specified range.

```go
n := 0
controller := hands.New()

controller.Do(func(ctx context.Context) error {
  n += 1
  return nil
}, hands.P(1))

controller.Do(func(ctx context.Context) error {
  n += 2
  return nil
}, hands.P(2))

controller.Do(func(ctx context.Context) error {
  n += 3
  return nil
}, hands.P(3))

controller.Do(func(ctx context.Context) error {
  n += 4
  return nil
}, hands.P(4))

controller.Run(hands.Between(2, 3))

fmt.Println(n)

// Output:
// 5
```

*Note*: If the use the `controller.Run()` method, tasks outside the `Between()` will not be executed, you can use the `controller.RunAll()` method to allow other priority tasks to be executed asynchronously.

```go
...
controller.RunAll(hands.Between(2, 3))

fmt.Println(n)
time.Sleep(time.Duration(10) * time.Millisecond)
fmt.Println(n)

// Output:
// 5
// 10
```

### `func In(in []int32) HandOption`

`In()`: Only execute tasks in the specified priority list.

```go
n := 0
controller := hands.New()

controller.Do(func(ctx context.Context) error {
  n += 1
  return nil
}, hands.P(1))

controller.Do(func(ctx context.Context) error {
  n += 2
  return nil
}, hands.P(2))

controller.Do(func(ctx context.Context) error {
  n += 3
  return nil
}, hands.P(3))

controller.Do(func(ctx context.Context) error {
  n += 4
  return nil
}, hands.P(4))

controller.Run(hands.In([]int32{2, 4}))

fmt.Println(n)

// Output:
// 6
```

Yes, the `controller.RunAll()` method can also be used here.

```go
...
controller.RunAll(hands.In([]int32{2, 4}))

fmt.Println(n)
time.Sleep(time.Duration(10) * time.Millisecond)
fmt.Println(n)

// Output:
// 6
// 10
```

### `func WithContext(ctx context.Context) HandOption`

Make the task use the specified context.

```go
c, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond)
defer cancel()

controller := hands.New()

controller.Do(func(ctx context.Context) error {
  time.Sleep(time.Duration(100) * time.Millisecond)
  return nil
})

err := controller.Run(hands.WithContext(c))

fmt.Println(err.Error())

// Output:
// context deadline exceeded
```

## Callback after all tasks have been executed

```go
n := 0
controller := hands.New()

controller.Do(func(ctx context.Context) error {
  time.Sleep(time.Duration(10) * time.Millisecond)
  n++
  return nil
})

controller.Do(func(ctx context.Context) error {
  time.Sleep(time.Duration(10) * time.Millisecond)
  n++
  return nil
})

controller.Do(func(ctx context.Context) error {
  n += 5
  return nil
})

// Here.
controller.Done(func() {
  // `True`
  assert.Equal(t, n, 7)
})

controller.RunAll(Fastest())

// `True`
assert.Equal(t, n, 5)
time.Sleep(time.Duration(500) * time.Millisecond)
```

---

## License
[![FOSSA Status](https://app.fossa.com/api/projects/git%2Bgithub.com%2Fduanckham%2Fhands.svg?type=large)](https://app.fossa.com/projects/git%2Bgithub.com%2Fduanckham%2Fhands?ref=badge_large)


hands

Hunch provides functions like: `All`, `First`, `Retry`, `Waterfall` etc., that makes asynchronous flow control more intuitive.

<a href="https://github.com/AaronJan/Hunch">
	<img width="160" src="https://user-images.githubusercontent.com/4630940/59684078-ea9d8c80-920b-11e9-9c99-85051dcb8a04.jpg" alt="Housekeeper" title="Hunch" align="right"/>
</a>

![GitHub tag (latest SemVer)](https://img.shields.io/github/tag/aaronjan/hunch.svg)
![Build status](https://github.com/aaronjan/hunch/workflows/Go/badge.svg?branch=master)
[![codecov](https://codecov.io/gh/AaronJan/Hunch/branch/master/graph/badge.svg)](https://codecov.io/gh/AaronJan/Hunch)
[![Go Report Card](https://goreportcard.com/badge/github.com/aaronjan/hunch)](https://goreportcard.com/report/github.com/aaronjan/hunch)
![GitHub](https://img.shields.io/github/license/aaronjan/hunch.svg)
[![GoDoc](https://godoc.org/github.com/aaronjan/hunch?status.svg)](https://godoc.org/github.com/aaronjan/hunch)

# Hunch

Hunch provides functions like: `All`, `First`, `Retry`, `Waterfall` etc., that makes asynchronous flow control more intuitive.

## About Hunch

Go have several concurrency patterns, here're some articles:

* https://blog.golang.org/pipelines
* https://blog.golang.org/context
* https://blog.golang.org/go-concurrency-patterns-timing-out-and
* https://blog.golang.org/advanced-go-concurrency-patterns

But nowadays, using the `context` package is the most powerful pattern.

So base on `context`, Hunch provides functions that can help you deal with complex asynchronous logics with ease.

## Usage

### Installation

#### `go get`

```shell
$ go get -u -v github.com/aaronjan/hunch
```

#### `go mod` (Recommended)

```go
import "github.com/aaronjan/hunch"
```

```shell
$ go mod tidy
```

### Types

```go
type Executable func(context.Context) (interface{}, error)

type ExecutableInSequence func(context.Context, interface{}) (interface{}, error)
```

### API

#### All

```go
func All(parentCtx context.Context, execs ...Executable) ([]interface{}, error) 
```

All returns all the outputs from all Executables, order guaranteed.

##### Examples

```go
ctx := context.Background()
r, err := hunch.All(
    ctx,
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(300 * time.Millisecond)
        return 1, nil
    },
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(200 * time.Millisecond)
        return 2, nil
    },
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(100 * time.Millisecond)
        return 3, nil
    },
)

fmt.Println(r, err)
// Output:
// [1 2 3] <nil>
```

#### Take

```go
func Take(parentCtx context.Context, num int, execs ...Executable) ([]interface{}, error)
```

Take returns the first `num` values outputted by the Executables.

##### Examples

```go
ctx := context.Background()
r, err := hunch.Take(
    ctx,
    // Only need the first 2 values.
    2,
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(300 * time.Millisecond)
        return 1, nil
    },
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(200 * time.Millisecond)
        return 2, nil
    },
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(100 * time.Millisecond)
        return 3, nil
    },
)

fmt.Println(r, err)
// Output:
// [3 2] <nil>
```

#### Last

```go
func Last(parentCtx context.Context, num int, execs ...Executable) ([]interface{}, error)
```

Last returns the last `num` values outputted by the Executables.

##### Examples

```go
ctx := context.Background()
r, err := hunch.Last(
    ctx,
    // Only need the last 2 values.
    2,
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(300 * time.Millisecond)
        return 1, nil
    },
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(200 * time.Millisecond)
        return 2, nil
    },
    func(ctx context.Context) (interface{}, error) {
        time.Sleep(100 * time.Millisecond)
        return 3, nil
    },
)

fmt.Println(r, err)
// Output:
// [2 1] <nil>
```

#### Waterfall

```go
func Waterfall(parentCtx context.Context, execs ...ExecutableInSequence) (interface{}, error)
```

Waterfall runs `ExecutableInSequence`s one by one, passing previous result to next Executable as input. When an error occurred, it stop the process then returns the error. When the parent Context canceled, it returns the `Err()` of it immediately.

##### Examples

```go
ctx := context.Background()
r, err := hunch.Waterfall(
    ctx,
    func(ctx context.Context, n interface{}) (interface{}, error) {
        return 1, nil
    },
    func(ctx context.Context, n interface{}) (interface{}, error) {
        return n.(int) + 1, nil
    },
    func(ctx context.Context, n interface{}) (interface{}, error) {
        return n.(int) + 1, nil
    },
)

fmt.Println(r, err)
// Output:
// 3 <nil>
```

#### Retry

```go
func Retry(parentCtx context.Context, retries int, fn Executable) (interface{}, error)
```

Retry attempts to get a value from an Executable instead of an Error. It will keeps re-running the Executable when failed no more than `retries` times. Also, when the parent Context canceled, it returns the `Err()` of it immediately.

##### Examples

```go
count := 0
getStuffFromAPI := func() (int, error) {
    if count == 5 {
        return 1, nil
    }
    count++

    return 0, fmt.Errorf("timeout")
}

ctx := context.Background()
r, err := hunch.Retry(
    ctx,
    10,
    func(ctx context.Context) (interface{}, error) {
        rs, err := getStuffFromAPI()

        return rs, err
    },
)

fmt.Println(r, err, count)
// Output:
// 1 <nil> 5
```

## Credits

Heavily inspired by [Async](https://github.com/caolan/async/) and [ReactiveX](http://reactivex.io/).

## Licence

[Apache 2.0](https://www.apache.org/licenses/LICENSE-2.0)


Hunch

Provides an unlimited job queue and concurrent worker pools.

# kyoo: A Go library providing an unlimited job queue and concurrent worker pools

![Build](https://gitlab.com/dirkaholic/kyoo/badges/master/pipeline.svg) ![Coverage](https://gitlab.com/dirkaholic/kyoo/badges/master/coverage.svg) [![Go Report Card](https://goreportcard.com/badge/github.com/dirkaholic/kyoo)](https://goreportcard.com/report/github.com/dirkaholic/kyoo)

## About

kyoo is the phonetic transcription of the word queue. It provides a job queue that can hold as much jobs as
resources are available on the running system.

The queue has the following characteristics:
* No limit of jobs to be queued (only limited by system resources = memory)
* Concurrent processing of jobs using worker pools
* When stopping queue, pending jobs are still processed

The library contains a simple `Job` interface and a simple `FuncExecutorJob` that
just executes a given function and implements that interface. With that nearly all
kinds of workloads should be processable already but of course it is possible to add
custom implementations of the `Job` interface.

Possible use cases for the library are:
* Consumers for message queues like RabbitMQ or Amazon SQS
* Processing web server requests offloading time extensive work into background jobs
* All kinds of backend processing jobs like image optimization, etc.

## Example

The following example shows a simple http server offloading jobs to the jobqueue that is constantly
processed in the background.

```go
package main

import (
	"fmt"
	"log"
	"net/http"
	"runtime"
	"time"

	jobqueue "github.com/dirkaholic/kyoo"
	"github.com/dirkaholic/kyoo/job"
)

var queue *jobqueue.JobQueue

func handler(w http.ResponseWriter, r *http.Request) {
	queue.Submit(&job.FuncExecutorJob{Func: func() error {
		return doTheHeavyBackgroundWork(r.URL.Path)
	}})
	fmt.Printf("%s - submitted %s !!\n", time.Now().String(), r.URL.Path)

	fmt.Fprint(w, "Job added to queue.")
}

func main() {
	queue = jobqueue.NewJobQueue(runtime.NumCPU() * 2)
	queue.Start()

	http.HandleFunc("/", handler)
	log.Fatal(http.ListenAndServe(":8080", nil))
}

func doTheHeavyBackgroundWork(path string) error {
	time.Sleep(2 * time.Second)
	fmt.Printf("%s - processed %s !!\n", time.Now().String(), path)
	return nil
}
```

Test the offloading by sending a bunch of http requests to the server 

```bash
$ for i in {1..10}; do http http://127.0.0.1:8080/test/$i; done
````

The output on http server side should be similar like this

```bash
2020-01-09 21:36:36.156277 +0100 CET m=+5.733617272 - submitted /test/1 !!
2020-01-09 21:36:36.443521 +0100 CET m=+6.020861136 - submitted /test/2 !!
2020-01-09 21:36:36.730535 +0100 CET m=+6.307874793 - submitted /test/3 !!
2020-01-09 21:36:37.021405 +0100 CET m=+6.598744533 - submitted /test/4 !!
2020-01-09 21:36:37.311973 +0100 CET m=+6.889312431 - submitted /test/5 !!
2020-01-09 21:36:37.609868 +0100 CET m=+7.187208115 - submitted /test/6 !!
2020-01-09 21:36:37.895222 +0100 CET m=+7.472561850 - submitted /test/7 !!
2020-01-09 21:36:38.160524 +0100 CET m=+7.737863891 - processed /test/1 !!
2020-01-09 21:36:38.171491 +0100 CET m=+7.748830724 - submitted /test/8 !!
2020-01-09 21:36:38.445832 +0100 CET m=+8.023171514 - processed /test/2 !!
2020-01-09 21:36:38.448423 +0100 CET m=+8.025762679 - submitted /test/9 !!
2020-01-09 21:36:38.730541 +0100 CET m=+8.307880933 - submitted /test/10 !!
2020-01-09 21:36:38.735158 +0100 CET m=+8.312497505 - processed /test/3 !!
2020-01-09 21:36:39.024788 +0100 CET m=+8.602128093 - processed /test/4 !!
2020-01-09 21:36:39.315991 +0100 CET m=+8.893331115 - processed /test/5 !!
2020-01-09 21:36:39.614848 +0100 CET m=+9.192187633 - processed /test/6 !!
2020-01-09 21:36:39.896692 +0100 CET m=+9.474031970 - processed /test/7 !!
2020-01-09 21:36:40.175952 +0100 CET m=+9.753291345 - processed /test/8 !!
2020-01-09 21:36:40.451877 +0100 CET m=+10.029216847 - processed /test/9 !!
2020-01-09 21:36:40.734289 +0100 CET m=+10.311628415 - processed /test/10 !!
```

## More examples

* [SQS worker](examples/sqsworker)


kyoo

Drop-in alternative to `sync/errgroup`, limited to a pool of N worker goroutines.


[![Actions Status](https://github.com/neilotoole/errgroup/workflows/Go/badge.svg)](https://github.com/neilotoole/errgroup/actions?query=workflow%3AGo)
[![Go Report Card](https://goreportcard.com/badge/neilotoole/errgroup)](https://goreportcard.com/report/neilotoole/errgroup)
[![release](https://img.shields.io/badge/release-v0.1.5-green.svg)](https://github.com/neilotoole/errgroup/releases/tag/v0.1.5)
[![Coverage](https://gocover.io/_badge/github.com/neilotoole/errgroup)](https://gocover.io/github.com/neilotoole/errgroup)
[![GoDoc](https://godoc.org/github.com/golang/gddo?status.svg)](https://pkg.go.dev/github.com/neilotoole/errgroup)
[![license](https://img.shields.io/github/license/neilotoole/errgroup)](./LICENSE)

# neilotoole/errgroup
`neilotoole/errgroup` is a drop-in alternative to Go's wonderful
[`sync/errgroup`](https://pkg.go.dev/golang.org/x/sync/errgroup) but
limited to `N` goroutines. This is useful for interaction with rate-limited
APIs, databases, and the like.


> **Note**
> The `sync/errgroup` package [now](https://github.com/neilotoole/errgroup/issues/14) has a [Group.SetLimit](https://pkg.go.dev/golang.org/x/sync/errgroup#Group.SetLimit) method,
> which eliminates the need for `neilotoole/errgroup`. This package will no longer be maintained. Use `sync/errgroup` instead.


## Overview
In effect, `neilotoole/errgroup` is `sync/errgroup` but with a worker pool
of `N` goroutines. The exported API is identical but for an additional
function `WithContextN`, which allows the caller
to specify the maximum number of goroutines (`numG`) and the capacity
of the queue channel (`qSize`) used to hold work before it is picked
up by a worker goroutine. The zero `Group` and the `Group` returned
by `WithContext` have `numG` and `qSize` equal to `runtime.NumCPU`.


## Usage
The exported API of this package mirrors the `sync/errgroup` package.
The only change needed is the import path of the package, from:

```go
import (
  "golang.org/x/sync/errgroup"
)
```

to

```go
import (
  "github.com/neilotoole/errgroup"
)
```

Then use in the normal manner. See the [godoc](https://pkg.go.dev/github.com/neilotoole/errgroup) for more.

```go
g, ctx := errgroup.WithContext(ctx)
g.Go(func() error {
    // do something
    return nil
})

err := g.Wait()
```

Many users will have no need to tweak the `numG` and `qCh` params. However, benchmarking
may suggest particular values for your workload. For that you'll need `WithContextN`:

```go
numG, qSize := 8, 4
g, ctx := errgroup.WithContextN(ctx, numG, qSize)

```

## Performance
The motivation for creating `neilotoole/errgroup` was to provide rate-limiting while
maintaining the lovely `sync/errgroup` semantics. Sacrificing some
performance vs `sync/errgroup` was assumed. However, benchmarking
suggests that this implementation can be more effective than `sync/errgroup` 
when tuned for a specific workload.

Below is a selection of benchmark results. How to read this: a workload is _X_ tasks
of _Y_ complexity. The workload is executed for:
 
- `sync/errgroup`, listed as `sync_errgroup`
- a non-parallel implementation (`sequential`)
- various `{numG, qSize}` configurations of `neilotoole/errgroup`, listed as `errgroupn_{numG}_{qSize}`

```
BenchmarkGroup_Short/complexity_5/tasks_50/errgroupn_default_16_16-16         	   25574	     46867 ns/op	     688 B/op	      12 allocs/op
BenchmarkGroup_Short/complexity_5/tasks_50/errgroupn_4_4-16                   	   24908	     48926 ns/op	     592 B/op	      12 allocs/op
BenchmarkGroup_Short/complexity_5/tasks_50/errgroupn_16_4-16                  	   24895	     48313 ns/op	     592 B/op	      12 allocs/op
BenchmarkGroup_Short/complexity_5/tasks_50/errgroupn_32_4-16                  	   24853	     48284 ns/op	     592 B/op	      12 allocs/op
BenchmarkGroup_Short/complexity_5/tasks_50/sync_errgroup-16                   	   18784	     65826 ns/op	    1858 B/op	      55 allocs/op
BenchmarkGroup_Short/complexity_5/tasks_50/sequential-16                      	   10000	    111483 ns/op	       0 B/op	       0 allocs/op

BenchmarkGroup_Short/complexity_20/tasks_50/errgroupn_default_16_16-16        	    3745	    325993 ns/op	    1168 B/op	      27 allocs/op
BenchmarkGroup_Short/complexity_20/tasks_50/errgroupn_4_4-16                  	    5186	    227034 ns/op	    1072 B/op	      27 allocs/op
BenchmarkGroup_Short/complexity_20/tasks_50/errgroupn_16_4-16                 	    3970	    312816 ns/op	    1076 B/op	      27 allocs/op
BenchmarkGroup_Short/complexity_20/tasks_50/errgroupn_32_4-16                 	    3715	    320757 ns/op	    1073 B/op	      27 allocs/op
BenchmarkGroup_Short/complexity_20/tasks_50/sync_errgroup-16                  	    2739	    432093 ns/op	    1862 B/op	      55 allocs/op
BenchmarkGroup_Short/complexity_20/tasks_50/sequential-16                     	    2306	    520947 ns/op	       0 B/op	       0 allocs/op

BenchmarkGroup_Short/complexity_40/tasks_250/errgroupn_default_16_16-16       	     354	   3602666 ns/op	    1822 B/op	      47 allocs/op
BenchmarkGroup_Short/complexity_40/tasks_250/errgroupn_4_4-16                 	     420	   2468605 ns/op	    1712 B/op	      47 allocs/op
BenchmarkGroup_Short/complexity_40/tasks_250/errgroupn_16_4-16                	     334	   3581349 ns/op	    1716 B/op	      47 allocs/op
BenchmarkGroup_Short/complexity_40/tasks_250/errgroupn_32_4-16                	     310	   3890316 ns/op	    1712 B/op	      47 allocs/op
BenchmarkGroup_Short/complexity_40/tasks_250/sync_errgroup-16                 	     253	   4740462 ns/op	    8303 B/op	     255 allocs/op
BenchmarkGroup_Short/complexity_40/tasks_250/sequential-16                    	     200	   5924693 ns/op	       0 B/op	       0 allocs/op
```

The overall impression is that `neilotoole/errgroup` can provide higher
throughput than `sync/errgroup` for these (CPU-intensive) workloads,
sometimes significantly so. As always, these benchmark results should
not be taken as gospel: your results may vary.


## Design Note
Why require an explicit `qSize` limit?

If the number of calls to `Group.Go` results in `qCh` becoming
full, the `Go` method will block until worker goroutines relieve `qCh`.
This behavior is in contrast to `sync/errgroup`'s `Go` method, which doesn't block.
While `neilotoole/errgroup` aims to be as much of a behaviorally similar
"drop-in" alternative to `sync/errgroup` as possible, this blocking behavior
is a conscious deviation.

Noting that the capacity of `qCh` is controlled by `qSize`, it's probable an
alternative implementation could be built that uses a (growable) slice
acting - if `qCh` is full - as a buffer for functions passed to `Go`.
Consideration of this potential design led to this [issue](https://github.com/golang/go/issues/20352)
regarding _unlimited capacity channels_, or perhaps better characterized
in this particular case as "_growable capacity channels_". If such a
feature existed in the language, it's possible that this implementation might
have taken advantage of it, at least in the first-pass release (benchmarking notwithstanding).
However benchmarking seems to suggest that a relatively
small `qSize` has performance benefits for some workloads, so it's possible
that the explicit `qSize` requirement is a better design choice regardless.


neilotoole/errgroup

# nursery: structured concurrency in Go
[![GoDoc](https://godoc.org/github.com/arunsworld/nursery?status.svg)](https://godoc.org/github.com/arunsworld/nursery)
[![GoReportCard](https://goreportcard.com/badge/github.com/arunsworld/nursery)](https://goreportcard.com/badge/github.com/arunsworld/nursery)
[![CircleCI](https://circleci.com/gh/arunsworld/nursery.svg?style=svg)](https://circleci.com/gh/arunsworld/nursery)
<a href='https://github.com/jpoles1/gopherbadger' target='_blank'>![gopherbadger-tag-do-not-edit](https://img.shields.io/badge/Go%20Coverage-100%25-brightgreen.svg?longCache=true&style=flat)</a>

```go
RunConcurrently(
    // Job 1
    func(context.Context, chan error) {
        time.Sleep(time.Millisecond * 10)
        log.Println("Job 1 done...")
    },
    // Job 2
    func(context.Context, chan error) {
        time.Sleep(time.Millisecond * 5)
        log.Println("Job 2 done...")
    },
)
log.Println("All jobs done...")
```

## Installation
```bash
go get -u github.com/arunsworld/nursery
```

[Notes on structured concurrency, or: Go statement considered harmful](https://vorpus.org/blog/notes-on-structured-concurrency-or-go-statement-considered-harmful/#nurseries-a-structured-replacement-for-go-statements) is an article that compares the dangers of goto with the go statement.

While I don't necessarily agree with the entire content I can appreciate that even with Go's high-level abstraction of concurrency using Goroutines, Channels & the select statement it is possible to end up with unreadable code, deadlocks, leaked goroutines, race conditions and poor error handling.

Implementing a higher-level abstraction for the use-cases mentioned is very straightforward in Go and this simple package provides just that.

The following functions are provided:
* `RunConcurrently(jobs ...ConcurrentJob) error`: takes an array of `ConcurrentJob`s and runs them concurrently ensuring that all jobs are completed before the call terminates. If all jobs terminate cleanly error is nil; otherwise the first non-nil error is returned.
* `RunConcurrentlyWithContext(parentCtx context.Context, jobs ...ConcurrentJob) error`: is the RunConcurrently behavior but additionally wraps a context that's passed in allowing cancellations of the parentCtx to get propagated.
* `RunMultipleCopiesConcurrently(copies int, job ConcurrentJob) error`: makes copies of the given job and runs them concurrently. This is useful for cases where we want to execute multiple slow consumers taking jobs from a channel until the job is finished. The channel itself can be fed by a producer that is run concurrently with the job running the consumers. Each job's context is also passed an unique index with key `nursery.JobID` - a 0 based int - that maybe used as a job identity if required.
* `RunMultipleCopiesConcurrentlyWithContext(ctx context.Context, copies int, job ConcurrentJob) error`: is the RunMultipleCopiesConcurrently behavior with a context that allows cancellation to be propagated to the jobs.
* `RunUntilFirstCompletion(jobs ...ConcurrentJob) error`: takes an array of `ConcurrentJob`s and runs them concurrently but terminates after the completion of the earliest completing job. A key point here is that despite early termination it blocks until all jobs have terminated (ie. released any used resources). If all jobs terminate cleanly error is nil; otherwise the first non-nil error is returned.
* `RunUntilFirstCompletionWithContext(parentCtx context.Context, jobs ...ConcurrentJob) error`: is the RunUntilFirstCompletion behavior but additionally wraps a context that's passed in allowing cancellations of the parentCtx to get propagated.
* `RunConcurrentlyWithTimeout(timeout time.Duration, jobs ...ConcurrentJob) error`: is similar in behavior to `RunConcurrently` except it also takes a timeout and can cause the function to terminate earlier if timeout has expired. As before we wait for all jobs to have cleanly terminated.
* `RunUntilFirstCompletionWithTimeout(timeout time.Duration, jobs ...ConcurrentJob) error`: is similar in behavior to `RunUntilFirstCompletion` with an additional timeout clause.

`ConcurrentJob` is a simple function that takes a context and error channel. We need to ensure that we're listening to the `Done()` channel on context and if invoked to clean-up resources and bail out. Errors are to be published to the error channel for proper handling.

Note: while this package simplifies the semantics of defining and executing concurrent code it cannot protect against bad concurrent programming such as using shared resources across jobs leading to data corruption or panics due to race conditions.

You may also be interested in reading [Structured Concurrency in Go](https://medium.com/@arunsworld/structured-concurrency-in-go-b800c7c4434e).

The library includes a utility function: `IsContextDone(context.Context)` to check if the passed in context is done or not. This can be used as a guard clause in a for loop within a ConcurrentJob using the passed in context to decide whether to stop processing and return or continue.


nursery

## Parallel fn

* Run functions in parallel.
* Limit the number of goroutines running at the same time.

## Installation
```bash
go get -u github.com/rafaeljesus/parallel-fn
```

## Usage
### Run
```go
package main

import (
  	"errors"

  	"github.com/rafaeljesus/parallel-fn"
)

func main() {
	timeout := time.After(2 * time.Second)
        fn1 := func() error { return nil }
        fn2 := func() error { return errors.New("BOOM!") }

	for {
		select {
		case err := <-Run(fn1, fn2):
                	// catch errors
		case <-timeout:
      			// handle timeout
		}
	}
}
```

### RunLimit
```go
package main

import (
  	"errors"

  	"github.com/rafaeljesus/parallel-fn"
)

func main() {
	timeout := time.After(2 * time.Second)
  	fn1 := func() error { return nil }
  	fn2 := func() error { return errors.New("BOOM!") }
  	fn3 := func() error { nil }
  	fn4 := func() error { nil }

	for {
		select {
		case err := <-RunLimit(2, fn1, fn2, fn3, fn4):
      			// catch errors
		case <-timeout:
      			// handle timeout
		}
	}
}
```

## Contributing
- Fork it
- Create your feature branch (`git checkout -b my-new-feature`)
- Commit your changes (`git commit -am 'Add some feature'`)
- Push to the branch (`git push origin my-new-feature`)
- Create new Pull Request

## Badges

[![Build Status](https://circleci.com/gh/rafaeljesus/parallel-fn.svg?style=svg)](https://circleci.com/gh/rafaeljesus/parallel-fn)
[![Go Report Card](https://goreportcard.com/badge/github.com/rafaeljesus/parallel-fn)](https://goreportcard.com/report/github.com/rafaeljesus/parallel-fn)
[![Go Doc](https://godoc.org/github.com/rafaeljesus/parallel-fn?status.svg)](https://godoc.org/github.com/rafaeljesus/parallel-fn)

---

> GitHub [@rafaeljesus](https://github.com/rafaeljesus) &nbsp;&middot;&nbsp;
> Medium [@_jesus_rafael](https://medium.com/@_jesus_rafael) &nbsp;&middot;&nbsp;
> Twitter [@_jesus_rafael](https://twitter.com/_jesus_rafael)


parallel-fn

Minimalistic and High-performance goroutine worker pool written in Go.

<a title="Codecov" target="_blank" href="https://github.com/alitto/pond/actions"><img alt="Build status" src="https://github.com/alitto/pond/actions/workflows/main.yml/badge.svg?branch=master&event=push"/></a>
<a title="Codecov" target="_blank" href="https://codecov.io/gh/alitto/pond"><img src="https://codecov.io/gh/alitto/pond/branch/master/graph/badge.svg"/></a>
<a title="Release" target="_blank" href="https://github.com/alitto/pond/releases"><img src="https://img.shields.io/github/v/release/alitto/pond"/></a>
<a title="Go Report Card" target="_blank" href="https://goreportcard.com/report/github.com/alitto/pond"><img src="https://goreportcard.com/badge/github.com/alitto/pond"/></a>

# pond
Minimalistic and High-performance goroutine worker pool written in Go

## Motivation

This library is meant to provide a simple way to limit concurrency when executing some function over a limited resource or service.

Some common scenarios include:

 - Executing queries against a Database with a limited no. of connections
 - Sending HTTP requests to a a rate/concurrency limited API

## Features:

- Zero dependencies
- Create pools with fixed or dynamic size
- Worker goroutines are only created when needed (backpressure detection) and automatically purged after being idle for some time (configurable)
- Minimalistic APIs for:
  - Creating worker pools with fixed or dynamic size
  - Submitting tasks to a pool in a fire-and-forget fashion
  - Submitting tasks to a pool and waiting for them to complete
  - Submitting tasks to a pool with a deadline
  - Submitting a group of tasks and waiting for them to complete
  - Submitting a group of tasks associated to a Context
  - Getting the number of running workers (goroutines)
  - Stopping a worker pool
- Task panics are handled gracefully (configurable panic handler)
- Supports Non-blocking and Blocking task submission modes (buffered / unbuffered)
- Very high performance and efficient resource usage under heavy workloads, even outperforming unbounded goroutines in some scenarios (See [benchmarks](https://github.com/alitto/pond-benchmarks))
- Configurable pool resizing strategy, with 3 presets for common scenarios: Eager, Balanced and Lazy. 
- Complete pool metrics such as number of running workers, tasks waiting in the queue [and more](#metrics--monitoring).
- **New (since v1.7.0)**: configurable parent context and graceful shutdown with deadline.
- [API reference](https://pkg.go.dev/github.com/alitto/pond)

## How to install

```bash
go get -u github.com/alitto/pond
```

## How to use

### Worker pool with dynamic size

``` go
package main

import (
	"fmt"

	"github.com/alitto/pond"
)

func main() {

	// Create a buffered (non-blocking) pool that can scale up to 100 workers
	// and has a buffer capacity of 1000 tasks
	pool := pond.New(100, 1000)

	// Submit 1000 tasks
	for i := 0; i < 1000; i++ {
		n := i
		pool.Submit(func() {
			fmt.Printf("Running task #%d\n", n)
		})
	}

	// Stop the pool and wait for all submitted tasks to complete
	pool.StopAndWait()
}
```

### Worker pool with fixed size

``` go
package main

import (
	"fmt"

	"github.com/alitto/pond"
)

func main() {

	// Create an unbuffered (blocking) pool with a fixed 
	// number of workers
	pool := pond.New(10, 0, pond.MinWorkers(10))

	// Submit 1000 tasks
	for i := 0; i < 1000; i++ {
		n := i
		pool.Submit(func() {
			fmt.Printf("Running task #%d\n", n)
		})
	}

	// Stop the pool and wait for all submitted tasks to complete
	pool.StopAndWait()
}
```

### Submitting a group of tasks

``` go
package main

import (
	"fmt"

	"github.com/alitto/pond"
)

func main() {

	// Create a pool
	pool := pond.New(10, 1000)
	defer pool.StopAndWait()

	// Create a task group
	group := pool.Group()

	// Submit a group of tasks
	for i := 0; i < 20; i++ {
		n := i
		group.Submit(func() {
			fmt.Printf("Running group task #%d\n", n)
		})
	}

	// Wait for all tasks in the group to complete
	group.Wait()
}
```

### Submitting a group of tasks associated to a context (**since v1.8.0**)

This feature provides synchronization, error propagation, and Context cancelation for subtasks of a common task. Similar to `errgroup.Group` from [`golang.org/x/sync/errgroup`](https://pkg.go.dev/golang.org/x/sync/errgroup) package with concurrency bounded by the worker pool.

``` go
package main

import (
	"context"
	"fmt"
	"net/http"

	"github.com/alitto/pond"
)

func main() {

	// Create a worker pool
	pool := pond.New(10, 1000)
	defer pool.StopAndWait()

	// Create a task group associated to a context
	group, ctx := pool.GroupContext(context.Background())

	var urls = []string{
		"https://www.golang.org/",
		"https://www.google.com/",
		"https://www.github.com/",
	}

	// Submit tasks to fetch each URL
	for _, url := range urls {
		url := url
		group.Submit(func() error {
			req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
			resp, err := http.DefaultClient.Do(req)
			if err == nil {
				resp.Body.Close()
			}
			return err
		})
	}

	// Wait for all HTTP requests to complete.
	err := group.Wait()
	if err != nil {
		fmt.Printf("Failed to fetch URLs: %v", err)
	} else {
		fmt.Println("Successfully fetched all URLs")
	}
}
```

### Pool Configuration Options

#### MinWorkers

Specifies the minimum number of worker goroutines that must be running at any given time. These goroutines are started when the pool is created. The default value is 0. Example:

``` go
// This will create a pool with 5 running worker goroutines 
pool := pond.New(10, 1000, pond.MinWorkers(5))
```

#### IdleTimeout

Defines how long to wait before removing idle worker goroutines from the pool. The default value is 5 seconds. Example:

``` go
// This will create a pool that will remove workers 100ms after they become idle 
pool := pond.New(10, 1000, pond.IdleTimeout(100 * time.Millisecond))
``` 

#### PanicHandler

Allows to configure a custom function to handle panics thrown by tasks submitted to the pool. The default handler just writes a message to standard output using `fmt.Printf` with the following contents: `Worker exits from a panic: [panic] \n Stack trace: [stack trace]`).  Example:

```go
// Custom panic handler function
panicHandler := func(p interface{}) {
	fmt.Printf("Task panicked: %v", p)
}

// This will create a pool that will handle panics using a custom panic handler
pool := pond.New(10, 1000, pond.PanicHandler(panicHandler)))
```

#### Strategy

Configures the strategy used to resize the pool when backpressure is detected. You can create a custom strategy by implementing the `pond.ResizingStrategy` interface or choose one of the 3 presets:

- **Eager**: maximizes responsiveness at the expense of higher resource usage, which can reduce throughput under certain conditions. This strategy is meant for worker pools that will operate at a small percentage of their capacity most of the time and may occasionally receive bursts of tasks. This is the default strategy.
- **Balanced**: tries to find a balance between responsiveness and throughput. It's suitable for general purpose worker pools or those that will operate close to 50% of their capacity most of the time.
- **Lazy**: maximizes throughput at the expense of responsiveness. This strategy is meant for worker pools that will operate close to their max. capacity most of the time.

``` go
// Example: create pools with different resizing strategies
eagerPool := pond.New(10, 1000, pond.Strategy(pond.Eager()))
balancedPool := pond.New(10, 1000, pond.Strategy(pond.Balanced()))
lazyPool := pond.New(10, 1000, pond.Strategy(pond.Lazy()))
```

#### Context

Configures a parent context on this pool to stop all workers when it is cancelled. The default value `context.Background()`. Example:

``` go
// This creates a pool that is stopped when myCtx is cancelled 
pool := pond.New(10, 1000, pond.Context(myCtx))
``` 

### Resizing strategies

The following chart illustrates the behaviour of the different pool resizing strategies as the number of submitted tasks increases. Each line represents the number of worker goroutines in the pool (pool size) and the x-axis reflects the number of submitted tasks (cumulative). 

![Pool resizing strategies behaviour](./docs/strategies.svg)

As the name suggests, the "Eager" strategy always spawns an extra worker when there are no idles, which causes the pool to grow almost linearly with the number of submitted tasks. On the other end, the "Lazy" strategy creates one worker every N submitted tasks, where N is the maximum number of available CPUs ([GOMAXPROCS](https://golang.org/pkg/runtime/#GOMAXPROCS)). The "Balanced" strategy represents a middle ground between the previous two because it creates a worker every N/2 submitted tasks.

### Stopping a pool

There are 3 methods available to stop a pool and release associated resources:
- `pool.Stop()`: stop accepting new tasks and signal all workers to stop processing new tasks. Tasks being processed by workers will continue until completion unless the process is terminated.
- `pool.StopAndWait()`: stop accepting new tasks and wait until all running and queued tasks have completed before returning.
- `pool.StopAndWaitFor(deadline time.Duration)`: similar to `StopAndWait` but with a deadline to prevent waiting indefinitely.

### Metrics & monitoring

Each worker pool instance exposes useful metrics that can be queried through the following methods:

- `pool.RunningWorkers() int`: Current number of running workers
- `pool.IdleWorkers() int`: Current number of idle workers
- `pool.MinWorkers() int`: Minimum number of worker goroutines
- `pool.MaxWorkers() int`: Maxmimum number of worker goroutines
- `pool.MaxCapacity() int`: Maximum number of tasks that can be waiting in the queue at any given time (queue capacity)
- `pool.SubmittedTasks() uint64`: Total number of tasks submitted since the pool was created
- `pool.WaitingTasks() uint64`: Current number of tasks in the queue that are waiting to be executed
- `pool.SuccessfulTasks() uint64`: Total number of tasks that have successfully completed their exection since the pool was created
- `pool.FailedTasks() uint64`: Total number of tasks that completed with panic since the pool was created
- `pool.CompletedTasks() uint64`: Total number of tasks that have completed their exection either successfully or with panic since the pool was created

In our [Prometheus example](./examples/prometheus/prometheus.go) we showcase how to configure collectors for these metrics and expose them to Prometheus.

## Examples

- [Creating a worker pool with dynamic size](./examples/dynamic_size/dynamic_size.go)
- [Creating a worker pool with fixed size](./examples/fixed_size/fixed_size.go)
- [Creating a worker pool with a Context](./examples/pool_context/pool_context.go)
- [Exporting worker pool metrics to Prometheus](./examples/prometheus/prometheus.go)
- [Submitting a group of tasks](./examples/task_group/task_group.go)
- [Submitting a group of tasks associated to a context](./examples/group_context/group_context.go)

## API Reference

Full API reference is available at https://pkg.go.dev/github.com/alitto/pond

## Benchmarks

See [Benchmarks](https://github.com/alitto/pond-benchmarks).

## Resources

Here are some of the resources which have served as inspiration when writing this library:

- http://marcio.io/2015/07/handling-1-million-requests-per-minute-with-golang/
- https://brandur.org/go-worker-pool
- https://gobyexample.com/worker-pools
- https://github.com/panjf2000/ants
- https://github.com/gammazero/workerpool

## Contribution & Support

Feel free to send a pull request if you consider there's something which can be improved. Also, please open up an issue if you run into a problem when using this library or just have a question.

pond

Limited consumer goroutine or unlimited goroutine pool for easier goroutine handling and cancellation.

Package pool
============

![Project status](https://img.shields.io/badge/version-3.1.1-green.svg)
[![Build Status](https://semaphoreci.com/api/v1/joeybloggs/pool/branches/v3/badge.svg)](https://semaphoreci.com/joeybloggs/pool)
[![Coverage Status](https://coveralls.io/repos/go-playground/pool/badge.svg?branch=v3&service=github)](https://coveralls.io/github/go-playground/pool?branch=v3)
[![Go Report Card](https://goreportcard.com/badge/gopkg.in/go-playground/pool.v3)](https://goreportcard.com/report/gopkg.in/go-playground/pool.v3)
[![GoDoc](https://godoc.org/gopkg.in/go-playground/pool.v3?status.svg)](https://godoc.org/gopkg.in/go-playground/pool.v3)
![License](https://img.shields.io/dub/l/vibe-d.svg)

Package pool implements a limited consumer goroutine or unlimited goroutine pool for easier goroutine handling and cancellation. 

Features:

-    Dead simple to use and makes no assumptions about how you will use it.
-    Automatic recovery from consumer goroutines which returns an error to the results

Pool v2 advantages over Pool v1:

- Up to 300% faster due to lower contention ( BenchmarkSmallRun used to take 3 seconds, now 1 second )
- Cancels are much faster
- Easier to use, no longer need to know the # of Work Units to be processed.
- Pool can now be used as a long running/globally defined pool if desired ( v1 Pool was only good for one run )
- Supports single units of work as well as batching
- Pool can easily be reset after a Close() or Cancel() for reuse.
- Multiple Batches can be run and even cancelled on the same Pool.
- Supports individual Work Unit cancellation.

Pool v3 advantages over Pool v2:

- Objects are not interfaces allowing for less breaking changes going forward.
- Now there are 2 Pool types, both completely interchangeable, a limited worker pool and unlimited pool.
- Simpler usage of Work Units, instead of `<-work.Done` now can do `work.Wait()`

Installation
------------

Use go get.

	go get gopkg.in/go-playground/pool.v3

Then import the pool package into your own code.

	import "gopkg.in/go-playground/pool.v3"


Important Information READ THIS!
------

- It is recommended that you cancel a pool or batch from the calling function and not inside of the Unit of Work, it will work fine, however because of the goroutine scheduler and context switching it may not cancel as soon as if called from outside.
- When Batching DO NOT FORGET TO CALL batch.QueueComplete(), if you do the Batch WILL deadlock
- It is your responsibility to call WorkUnit.IsCancelled() to check if it's cancelled after a blocking operation like waiting for a connection from a pool. (optional)

Usage and documentation
------

Please see http://godoc.org/gopkg.in/go-playground/pool.v3 for detailed usage docs.

##### Examples:

both Limited Pool and Unlimited Pool have the same signatures and are completely interchangeable.

Per Unit Work
```go
package main

import (
	"fmt"
	"time"

	"gopkg.in/go-playground/pool.v3"
)

func main() {

	p := pool.NewLimited(10)
	defer p.Close()

	user := p.Queue(getUser(13))
	other := p.Queue(getOtherInfo(13))

	user.Wait()
	if err := user.Error(); err != nil {
		// handle error
	}

	// do stuff with user
	username := user.Value().(string)
	fmt.Println(username)

	other.Wait()
	if err := other.Error(); err != nil {
		// handle error
	}

	// do stuff with other
	otherInfo := other.Value().(string)
	fmt.Println(otherInfo)
}

func getUser(id int) pool.WorkFunc {

	return func(wu pool.WorkUnit) (interface{}, error) {

		// simulate waiting for something, like TCP connection to be established
		// or connection from pool grabbed
		time.Sleep(time.Second * 1)

		if wu.IsCancelled() {
			// return values not used
			return nil, nil
		}

		// ready for processing...

		return "Joeybloggs", nil
	}
}

func getOtherInfo(id int) pool.WorkFunc {

	return func(wu pool.WorkUnit) (interface{}, error) {

		// simulate waiting for something, like TCP connection to be established
		// or connection from pool grabbed
		time.Sleep(time.Second * 1)

		if wu.IsCancelled() {
			// return values not used
			return nil, nil
		}

		// ready for processing...

		return "Other Info", nil
	}
}
```

Batch Work
```go
package main

import (
	"fmt"
	"time"

	"gopkg.in/go-playground/pool.v3"
)

func main() {

	p := pool.NewLimited(10)
	defer p.Close()

	batch := p.Batch()

	// for max speed Queue in another goroutine
	// but it is not required, just can't start reading results
	// until all items are Queued.

	go func() {
		for i := 0; i < 10; i++ {
			batch.Queue(sendEmail("email content"))
		}

		// DO NOT FORGET THIS OR GOROUTINES WILL DEADLOCK
		// if calling Cancel() it calles QueueComplete() internally
		batch.QueueComplete()
	}()

	for email := range batch.Results() {

		if err := email.Error(); err != nil {
			// handle error
			// maybe call batch.Cancel()
		}

		// use return value
		fmt.Println(email.Value().(bool))
	}
}

func sendEmail(email string) pool.WorkFunc {
	return func(wu pool.WorkUnit) (interface{}, error) {

		// simulate waiting for something, like TCP connection to be established
		// or connection from pool grabbed
		time.Sleep(time.Second * 1)

		if wu.IsCancelled() {
			// return values not used
			return nil, nil
		}

		// ready for processing...

		return true, nil // everything ok, send nil, error if not
	}
}
```

Benchmarks
------
###### Run on MacBook Pro (Retina, 15-inch, Late 2013) 2.6 GHz Intel Core i7 16 GB 1600 MHz DDR3 using Go 1.6.2

run with 1, 2, 4,8 and 16 cpu to show it scales well...16 is double the # of logical cores on this machine.

NOTE: Cancellation times CAN vary depending how busy your system is and how the goroutine scheduler is but 
worse case I've seen is 1s to cancel instead of 0ns

```go
go test -cpu=1,2,4,8,16 -bench=. -benchmem=true
PASS
BenchmarkLimitedSmallRun              	       1	1002492008 ns/op	    3552 B/op	      55 allocs/op
BenchmarkLimitedSmallRun-2            	       1	1002347196 ns/op	    3568 B/op	      55 allocs/op
BenchmarkLimitedSmallRun-4            	       1	1010533571 ns/op	    4720 B/op	      73 allocs/op
BenchmarkLimitedSmallRun-8            	       1	1008883324 ns/op	    4080 B/op	      63 allocs/op
BenchmarkLimitedSmallRun-16           	       1	1002317677 ns/op	    3632 B/op	      56 allocs/op
BenchmarkLimitedSmallCancel           	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedSmallCancel-2         	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedSmallCancel-4         	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedSmallCancel-8         	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedSmallCancel-16        	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedLargeCancel           	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedLargeCancel-2         	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedLargeCancel-4         	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedLargeCancel-8         	 1000000	      1006 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedLargeCancel-16        	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkLimitedOverconsumeLargeRun   	       1	4027153081 ns/op	   36176 B/op	     572 allocs/op
BenchmarkLimitedOverconsumeLargeRun-2 	       1	4003489261 ns/op	   32336 B/op	     512 allocs/op
BenchmarkLimitedOverconsumeLargeRun-4 	       1	4005579847 ns/op	   34128 B/op	     540 allocs/op
BenchmarkLimitedOverconsumeLargeRun-8 	       1	4004639857 ns/op	   34992 B/op	     553 allocs/op
BenchmarkLimitedOverconsumeLargeRun-16	       1	4022695297 ns/op	   36864 B/op	     532 allocs/op
BenchmarkLimitedBatchSmallRun         	       1	1000785511 ns/op	    6336 B/op	      94 allocs/op
BenchmarkLimitedBatchSmallRun-2       	       1	1001459945 ns/op	    4480 B/op	      65 allocs/op
BenchmarkLimitedBatchSmallRun-4       	       1	1002475371 ns/op	    6672 B/op	      99 allocs/op
BenchmarkLimitedBatchSmallRun-8       	       1	1002498902 ns/op	    4624 B/op	      67 allocs/op
BenchmarkLimitedBatchSmallRun-16      	       1	1002202273 ns/op	    5344 B/op	      78 allocs/op
BenchmarkUnlimitedSmallRun            	       1	1002361538 ns/op	    3696 B/op	      59 allocs/op
BenchmarkUnlimitedSmallRun-2          	       1	1002230293 ns/op	    3776 B/op	      60 allocs/op
BenchmarkUnlimitedSmallRun-4          	       1	1002148953 ns/op	    3776 B/op	      60 allocs/op
BenchmarkUnlimitedSmallRun-8          	       1	1002120679 ns/op	    3584 B/op	      57 allocs/op
BenchmarkUnlimitedSmallRun-16         	       1	1001698519 ns/op	    3968 B/op	      63 allocs/op
BenchmarkUnlimitedSmallCancel         	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedSmallCancel-2       	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedSmallCancel-4       	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedSmallCancel-8       	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedSmallCancel-16      	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedLargeCancel         	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedLargeCancel-2       	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedLargeCancel-4       	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedLargeCancel-8       	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedLargeCancel-16      	2000000000	         0.00 ns/op	       0 B/op	       0 allocs/op
BenchmarkUnlimitedLargeRun            	       1	1001631711 ns/op	   40352 B/op	     603 allocs/op
BenchmarkUnlimitedLargeRun-2          	       1	1002603908 ns/op	   38304 B/op	     586 allocs/op
BenchmarkUnlimitedLargeRun-4          	       1	1001452975 ns/op	   38192 B/op	     584 allocs/op
BenchmarkUnlimitedLargeRun-8          	       1	1005382882 ns/op	   35200 B/op	     537 allocs/op
BenchmarkUnlimitedLargeRun-16         	       1	1001818482 ns/op	   37056 B/op	     566 allocs/op
BenchmarkUnlimitedBatchSmallRun       	       1	1002391247 ns/op	    4240 B/op	      63 allocs/op
BenchmarkUnlimitedBatchSmallRun-2     	       1	1010313222 ns/op	    4688 B/op	      70 allocs/op
BenchmarkUnlimitedBatchSmallRun-4     	       1	1008364651 ns/op	    4304 B/op	      64 allocs/op
BenchmarkUnlimitedBatchSmallRun-8     	       1	1001858192 ns/op	    4448 B/op	      66 allocs/op
BenchmarkUnlimitedBatchSmallRun-16    	       1	1001228000 ns/op	    4320 B/op	      64 allocs/op
```
To put some of these benchmarks in perspective:

- BenchmarkLimitedSmallRun did 10 seconds worth of processing in 1.002492008s
- BenchmarkLimitedSmallCancel ran 20 jobs, cancelled on job 6 and and ran in 0s
- BenchmarkLimitedLargeCancel ran 1000 jobs, cancelled on job 6 and and ran in 0s
- BenchmarkLimitedOverconsumeLargeRun ran 100 jobs using 25 workers in 4.027153081s


License
------
Distributed under MIT License, please see license file in code for more details.


pool

Go concurrency with channel transformations. No boilerplate, type safety, batching and error handling.

# Rill [![GoDoc](https://pkg.go.dev/badge/github.com/destel/rill)](https://pkg.go.dev/github.com/destel/rill) [![Go Report Card](https://goreportcard.com/badge/github.com/destel/rill)](https://goreportcard.com/report/github.com/destel/rill) [![codecov](https://codecov.io/gh/destel/rill/graph/badge.svg?token=252K8OQ7E1)](https://codecov.io/gh/destel/rill) [![Mentioned in Awesome Go](https://awesome.re/mentioned-badge.svg)](https://github.com/avelino/awesome-go) 
Rill (noun: a small stream) is a Go toolkit that offers a collection of easy-to-use functions for concurrency, streaming,
batching and pipeline construction. It abstracts away the complexities of concurrency, removes boilerplate, 
provides a structured way to handle errors and allows developers to focus on core logic.
Whether you need to perform a basic concurrent ForEach or construct a complex multi-stage processing pipeline,
Rill has got you covered.

  


## Key Features
- **Easy to Use**: the complexity of managing goroutines, channels, wait groups, and atomics is abstracted away
- **Easy to Integrate**: seamlessly integrates into existing projects without any setup or configuration
- **Concurrent**: provides control over the level of concurrency for all operations
- **Error Handling**: provides a structured way to handle errors in concurrent applications
- **Streaming**: handles real-time data streams or large datasets with a minimal memory footprint
- **Modular**: allows composing functions to create custom pipelines and higher-order operations
- **Batching**: simplifies organizing and processing data in batches
- **Order Preservation**: provides functions that maintain the original order of data during concurrent processing
- **Efficient Resource Use**: ensures goroutine pool sizes and memory allocations are independent of input size
- **Generic**: all operations are type-safe and can be used with any data type



## Motivation
Rill might look like an iterator or functional programming library, but 
at its core it's a concurrency library made specifically for Go channels.

There is a consensus in the Go community that functional programming style operations like Map, Filter, ForEach and others are 
not idiomatic in Go, and that basic for-loops are better, faster, and more concise. This is true for slices, 
but for channels, the complexity can quickly escalate beyond a basic for-loop as requirements are added:

- A basic for-range loop is sufficient to iterate over a channel
- Adding concurrency requires goroutines and a WaitGroup
- Adding error handling means replacing WaitGroup with ErrGroup
- Controlling concurrency level is often done with a semaphore
- If the above approach becomes a bottleneck, worker pools must be introduced and managed manually
- For a multi-stage pipeline, everything must be manually managed at each stage, causing complexity to grow non-linearly
- Features like batching or ordered fan-in require even more complex orchestration and synchronization

These increasing levels of complexity introduce several challenges. While tools like channels, ErrGroups or semaphores are powerful on their own, 
combining them into a more complex logic, can lead to code with lots of boilerplate that's difficult to write, read, and maintain. 

Rill was born out of the desire to remove code duplication and to encapsulate all this complexity in a library with a 
simple, composable, and expressive API. The introduction of generics in Go 1.18 opened the door to creating 
functional-style operations on channels, providing a natural way to achieve this goal.



## Example Usage
Consider an example application that loads users from an API concurrently,
updates their status to active and saves them back, 
while controlling the level of concurrency for each operation.

[Full runnable example](https://pkg.go.dev/github.com/destel/rill#example-package)

```go
func main() {
	// In case of early exit this will cancel the user fetching,
	// which in turn will terminate the entire pipeline.	
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Start with a stream of user ids
	ids := rill.FromSlice([]int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}, nil)

	// Read users from the API.
	// Concurrency = 3
	users := rill.Map(ids, 3, func(id int) (*User, error) {
		return getUser(ctx, id)
	})

	// Activate users.
	// Concurrency = 2
	err := rill.ForEach(users, 2, func(u *User) error {
		if u.IsActive {
			fmt.Printf("User %d is already active\n", u.ID)
			return nil
		}

		u.IsActive = true
		return saveUser(ctx, u)
	})

	fmt.Println("Error:", err)
}
```


## Installation
```bash
go get -u github.com/destel/rill
```


## Testing Strategy
Rill has a test coverage of over 95%, with testing focused on:
- **Correctness**: ensuring that functions produce accurate results at different levels of concurrency
- **Concurrency**: confirming that correct number of goroutines is spawned and utilized
- **Ordering**: ensuring that ordered versions of functions preserve the order, while basic versions do not




## Design Philosophy
At the heart of rill lies a simple yet powerful concept: operating on channels of wrapped values, encapsulated by the **Try** container.
This allows to propagate both values and errors through the pipeline, ensuring that errors are handled correctly at each stage.
Such wrapped channels can be created manually or through utilities like **FromSlice** or **FromChan**, and then transformed via non-blocking 
functions like **Map** or **Filter**. Finally, the transformed stream can be consumed by a blocking function such as 
**ForEach**, **Reduce** or **MapReduce**

One of the key features of Rill is the ability to control the level of concurrency for almost all operations through the **n** parameter.
This is possible due to the channel and goroutine orchestration that library does under the hood. Rill's built-in functions manage
worker pools internally, making the number of goroutines and allocations independent of the input size.

Finally, rill is designed to be modular and extensible. Most functions take streams as input and return transformed streams as output. 
It's easy to create custom reusable higher-order operations and pipelines by combining existing ones.




## Batching
Batching is a common pattern in concurrent processing, especially when dealing with external services or databases.
Rill provides a **Batch** function that transforms a stream of items into a stream of batches of a specified size. It's also possible 
to specify a timeout, after which the batch is emitted even if it's not full. This is useful for keeping an application reactive
when input stream is slow or sparse.

Consider a modification of the previous example, where list of ids is streamed from a remote file,
and users are fetched from the API in batches.

[Full runnable example](https://pkg.go.dev/github.com/destel/rill#example-package-Batching)

```go
func main() {
	// In case of early exit this will cancel the file streaming,
	// which in turn will terminate the entire pipeline.
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Stream a file with user ids as an io.Reader
	reader, err := downloadFile(ctx, "http://example.com/user_ids1.txt")
	if err != nil {
		fmt.Println("Error:", err)
		return
	}

	// Transform the reader into a stream of lines
	lines := streamLines(reader)

	// Parse lines as integers
	// Concurrency = 3
	ids := rill.Map(lines, 3, func(line string) (int, error) {
		return strconv.Atoi(line)
	})

	// Group IDs into batches of 5 for bulk processing
	idBatches := rill.Batch(ids, 5, 1*time.Second)

	// Fetch users for each batch of IDs
	// Concurrency = 3
	userBatches := rill.Map(idBatches, 3, func(ids []int) ([]*User, error) {
		return getUsers(ctx, ids...)
	})

	// Transform batches back into a stream of users
	users := rill.Unbatch(userBatches)

	// Activate users.
	// Concurrency = 2
	err = rill.ForEach(users, 2, func(u *User) error {
		if u.IsActive {
			fmt.Printf("User %d is already active\n", u.ID)
			return nil
		}

		u.IsActive = true
		return saveUser(ctx, u)
	})

	fmt.Println("Error:", err)	
}
```



## Fan-in and Fan-out
Go channels support both Fan-in and Fan-out patterns, meaning that multiple goroutines can write to a single channel (fan-in)
or read from a single channel (fan-out). On top of that Rill adds a Merge function that can be used to combine multiple streams into a single one.

Consider a basic example application that concurrently sends messages through multiple servers, then collects the results
into a single stream and handles errors.

[Full runnable example](https://pkg.go.dev/github.com/destel/rill#example-package-FanIn_FanOut)

```go
func main() {
	messages := rill.FromSlice([]string{
		"message1", "message2", "message3", "message4", "message5",
		"message6", "message7", "message8", "message9", "message10",
	}, nil)

	// Fan-out the messages to three servers
	results1 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server1")
	})

	results2 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server2")
	})

	results3 := rill.Map(messages, 2, func(message string) (string, error) {
		return message, sendMessage(message, "server3")
	})

	// Fan-in the results from all servers into a single stream
	results := rill.Merge(results1, results2, results3)

	// Handle errors
	err := rill.Err(results)
	fmt.Println("Error:", err)
}
```

## Errors, Termination and Contexts
Error handling can be non-trivial in concurrent applications. Rill simplifies this by providing a structured error handling approach.
Usually rill pipelines consist of zero or more non-blocking stages that transform the input stream, 
and one blocking stage that returns the results. General rule is: any error happening anywhere in the pipeline is 
propagated down to the final stage, where it's caught by some blocking function and returned to the caller.

Rill provides several blocking functions out of the box:

- **ForEach:** Concurrently applies a user function to each item in the stream. 
  [Example](https://pkg.go.dev/github.com/destel/rill#example-ForEach)
- **ToSlice:** Collects all stream items into a slice.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-ToSlice)
- **ToSeq2:** Converts a stream into an iterator of value-error pairs.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-ToSeq2)
- **Reduce:** Concurrently reduces the stream to a single value, using a user provided reducer function.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-Reduce)
- **MapReduce:** Performs a concurrent MapReduce operation one the stream, reducing it to Go map,
  using user provided mapper and reducer functions.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-MapReduce)
- **All:** Concurrently checks if all items in the stream satisfy a user provided condition.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-All)
- **Any:** Concurrently checks if at least one item in the stream satisfies a user provided condition.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-Any)
- **First:** Returns the first item or error encountered in the stream.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-First)
- **Err:** Checks if there's an error somewhere in the stream an returns it.
  [Example](https://pkg.go.dev/github.com/destel/rill#example-Err)


All blocking functions share a common behavior. In case of an early termination (before reaching the end of the input stream), 
such functions initiate background draining of the remaining items. This is done to prevent goroutine leaks by ensuring that 
all goroutines feeding the stream are allowed to complete. 
The input stream should not be used anymore after calling a blocking function. 

It also possible to use a for-range loop instead of a blocking function to consume the stream. 
In this case, the caller would be responsible for draining the stream in case of an early termination. 
See more details in the package documentation.

Rill is context-agnostic, meaning that it does not enforce any specific context usage. 
However, it's recommended to make user-defined pipeline stages context-aware.
This is especially important for the initial stage, as it allows to finish background draining
process, described above, faster.

In the example below the printOddSquares function initiates a pipeline that depends on a context.
When an error occurs in one of the pipeline stages, it propagates down the pipeline, causing an early return, 
context cancellation (via defer) and resource cleanup.

[Full runnable example](https://pkg.go.dev/github.com/destel/rill#example-package-Context)

```go
func main() {
	ctx := context.Background()

	err := printOddSquares(ctx)
	fmt.Println("Error:", err)
}

func printOddSquares(ctx context.Context) error {
	ctx, cancel := context.WithCancel(ctx)
	defer cancel()

	numbers := infiniteNumberStream(ctx)

	odds := rill.Filter(numbers, 3, func(x int) (bool, error) {
		if x == 20 {
			return false, fmt.Errorf("early exit")
		}
		return x%2 == 1, nil
	})

	return rill.ForEach(odds, 3, func(x int) error {
		fmt.Println(x * x)
		return nil
	})
}
```



## Order Preservation
In concurrent applications, maintaining the original sequence of processed items is challenging due to the nature of parallel execution. 
When values are read from an input stream, concurrently processed through a function **f**, and written to an output stream, their order might not 
match the order of the input. To address this, rill provides ordered versions of its core functions, such as **OrderedMap** or **OrderedFilter**. 
These ensure that if value **x** precedes value **y** in the input channel, then **f(x)** will precede **f(y)** in the output, 
preserving the original order. It's important to note that these ordered functions have a small overhead compared to their unordered counterparts, 
due to more advanced orchestration and synchronization happening under the hood.

Order preservation is vital in scenarios where the sequence of data impacts the outcome, such as time-series data processing. 
Take, for instance, an application that retrieves daily temperature measurements over a specific period and calculates the change 
in temperature from one day to the next. Such application can benefit from concurrent data fetching, but need fetched data
to be processed in the correct order.

[Full runnable example](https://pkg.go.dev/github.com/destel/rill#example-package-Ordering)

```go
type Measurement struct {
    Date time.Time
    Temp float64
}

func main() {
	city := "New York"
	endDate := time.Now()
	startDate := endDate.AddDate(0, 0, -30)

	// Create a stream of all days between startDate and endDate
	days := make(chan rill.Try[time.Time])
	go func() {
		defer close(days)
		for date := startDate; date.Before(endDate); date = date.AddDate(0, 0, 1) {
			days <- rill.Wrap(date, nil)
		}
	}()

	// Fetch the temperature for each day from the API
	// Concurrency = 10; Ordered
	measurements := rill.OrderedMap(days, 10, func(date time.Time) (Measurement, error) {
		temp, err := getTemperature(city, date)
		return Measurement{Date: date, Temp: temp}, err
	})

	// Iterate over the measurements, calculate and print changes.
	// Concurrency = 1; Ordered
	prev := Measurement{Temp: math.NaN()}
	err := rill.ForEach(measurements, 1, func(m Measurement) error {
		change := m.Temp - prev.Temp
		prev = m

		fmt.Printf("%s: %.1f°C (change %+.1f°C)\n", m.Date.Format("2006-01-02"), m.Temp, change)
		return nil
	})

	fmt.Println("Error:", err)
}
```

## Go 1.23 Iterators
Starting from Go 1.23, the language supports *range over function*, allowing users to define custom iterators 
for use in for-range loops. This feature enables Rill to integrate seamlessly with existing iterator-based functions
in the standard library and third-party packages.

Rill provides **FromSeq** and **FromSeq2** functions to convert an iterator into
a stream. Additionally, there's a **ToSeq2** function to convert a stream back into an iterator.

**ToSeq2** can be a good alternative to **ForEach** when concurrency is not needed. 
It gives more control and performs all necessary cleanup and draining, even if the loop is terminated early using *break* or *return*.

[Full runnable example](https://pkg.go.dev/github.com/destel/rill#example-ToSeq2)

```go
func main() {
	nums := rill.FromSeq2(genPositive(40))
	squares := rill.Map(nums, 4, func(x int) (int, error) {
		return x * x, nil
	})
	for val, err := range rill.ToSeq2(squares) {
		fmt.Println(val, err)
	}
}

func genPositive(to int) iter.Seq2[int, error] {
	return func(yield func(i int, err error) bool) {
		for i := 1; i <= to; i++ {
			if !yield(i, nil) {
				return
			}
		}
	}
}
```




## Performance
Rill's performance is primarily bounded by the performance of channel operations. As a result, applications that
already use channels and channel-based concurrency patterns can expect minimal overhead when adopting Rill.
Additionally, Rill does fewer allocations compared to some traditional concurrency patterns that spawn a goroutine 
for each channel item and use a semaphore to control the level of concurrency. For more details, refer to the [benchmarks](https://github.com/destel/rill/wiki/Benchmarks).

This makes Rill well-suited for a wide range of tasks, especially I/O-bound workloads where the overhead of channel operations is typically negligible.


rill

`routine` is a `ThreadLocal` for go library. It encapsulates and provides some easy-to-use, non-competitive, high-performance `goroutine` context access interfaces, which can help you access coroutine context information more gracefully.

# routine

[![Build Status](https://github.com/timandy/routine/actions/workflows/build.yml/badge.svg)](https://github.com/timandy/routine/actions)
[![Codecov](https://codecov.io/gh/timandy/routine/branch/main/graph/badge.svg)](https://app.codecov.io/gh/timandy/routine)
[![Go Report Card](https://goreportcard.com/badge/github.com/timandy/routine)](https://goreportcard.com/report/github.com/timandy/routine)
[![Documentation](https://pkg.go.dev/badge/github.com/timandy/routine.svg)](https://pkg.go.dev/github.com/timandy/routine)
[![Release](https://img.shields.io/github/release/timandy/routine.svg)](https://github.com/timandy/routine/releases)
[![License](https://img.shields.io/github/license/timandy/routine.svg)](https://github.com/timandy/routine/blob/main/LICENSE)

> [中文版](README_zh.md)

`routine` encapsulates and provides some easy-to-use, non-competitive, high-performance `goroutine` context access interfaces, which can help you access coroutine context information more gracefully.

# Introduce

From the very beginning of its design, the `Golang` language has spared no effort to shield the concept of coroutine context from developers, including the acquisition of coroutine `goid`, the state of coroutine within the process, and the storage of coroutine context.

If you have used other languages such as `C++`, `Java` and so on, then you must be familiar with `ThreadLocal`, but after starting to use `Golang`, you will be deeply confused and distressed by the lack of convenient functions like `ThreadLocal`.

Of course, you can choose to use `Context`, which carries all the context information, appears in the first input parameter of all functions, and then shuttles around your system.

And the core goal of `routine` is to open up another way: Introduce `goroutine local storage` to the `Golang` world.

# Usage & Demo

This chapter briefly introduces how to install and use the `routine` library.

## Install

```bash
go get github.com/timandy/routine
```

## Use `goid`

The following code simply demonstrates the use of `routine.Goid()`:

```go
package main

import (
	"fmt"
	"time"

	"github.com/timandy/routine"
)

func main() {
	goid := routine.Goid()
	fmt.Printf("cur goid: %v\n", goid)
	go func() {
		goid := routine.Goid()
		fmt.Printf("sub goid: %v\n", goid)
	}()

	// Wait for the sub-coroutine to finish executing.
	time.Sleep(time.Second)
}
```

In this example, the `main` function starts a new coroutine, so `Goid()` returns the main coroutine `1` and the child coroutine `6`:

```text
cur goid: 1
sub goid: 6
```

## Use `ThreadLocal`

The following code briefly demonstrates `ThreadLocal`'s creation, setting, getting, spreading across coroutines, etc.:

```go
package main

import (
	"fmt"
	"time"

	"github.com/timandy/routine"
)

var threadLocal = routine.NewThreadLocal[string]()
var inheritableThreadLocal = routine.NewInheritableThreadLocal[string]()

func main() {
	threadLocal.Set("hello world")
	inheritableThreadLocal.Set("Hello world2")
	fmt.Println("threadLocal:", threadLocal.Get())
	fmt.Println("inheritableThreadLocal:", inheritableThreadLocal.Get())

	// The child coroutine cannot read the previously assigned "hello world".
	go func() {
		fmt.Println("threadLocal in goroutine:", threadLocal.Get())
		fmt.Println("inheritableThreadLocal in goroutine:", inheritableThreadLocal.Get())
	}()

	// However, a new sub-coroutine can be started via the Go/GoWait/GoWaitResult function, and all inheritable variables of the current coroutine can be passed automatically.
	routine.Go(func() {
		fmt.Println("threadLocal in goroutine by Go:", threadLocal.Get())
		fmt.Println("inheritableThreadLocal in goroutine by Go:", inheritableThreadLocal.Get())
	})

	// You can also create a task via the WrapTask/WrapWaitTask/WrapWaitResultTask function, and all inheritable variables of the current coroutine can be automatically captured.
	task := routine.WrapTask(func() {
		fmt.Println("threadLocal in task by WrapTask:", threadLocal.Get())
		fmt.Println("inheritableThreadLocal in task by WrapTask:", inheritableThreadLocal.Get())
	})
	go task.Run()

	// Wait for the sub-coroutine to finish executing.
	time.Sleep(time.Second)
}
```

The execution result is:

```text
threadLocal: hello world
inheritableThreadLocal: Hello world2
threadLocal in goroutine:
inheritableThreadLocal in goroutine:
threadLocal in goroutine by Go:
inheritableThreadLocal in goroutine by Go: Hello world2
threadLocal in task by WrapTask:
inheritableThreadLocal in task by WrapTask: Hello world2
```

# API

This chapter introduces in detail all the interfaces encapsulated by the `routine` library, as well as their core functions and implementation methods.

## `Goid() int64`

Get the `goid` of the current `goroutine`.

It can be obtained directly through assembly code under `386`, `amd64`, `armv6`, `armv7`, `arm64`, `loong64`, `mips`, `mipsle`, `mips64`, `mips64le`, `ppc64`, `ppc64le`, `riscv64`, `s390x`, `wasm` architectures. This operation has extremely high performance and the time-consuming is usually only one-fifth of `rand.Int()`.

## `NewThreadLocal[T any]() ThreadLocal[T]`

Create a new `ThreadLocal[T]` instance with the initial value stored with the default value of type `T`.

## `NewThreadLocalWithInitial[T any](supplier Supplier[T]) ThreadLocal[T]`

Create a new `ThreadLocal[T]` instance with the initial value stored as the return value of the method `supplier()`.

## `NewInheritableThreadLocal[T any]() ThreadLocal[T]`

Create a new `ThreadLocal[T]` instance with the initial value stored with the default value of type `T`.
When a new coroutine is started via `Go()`, `GoWait()` or `GoWaitResult()`, the value of the current coroutine is copied to the new coroutine.
When a new task is created via `WrapTask()`, `WrapWaitTask()` or `WrapWaitResultTask()`, the value of the current coroutine is captured to the new task.

## `NewInheritableThreadLocalWithInitial[T any](supplier Supplier[T]) ThreadLocal[T]`

Create a new `ThreadLocal[T]` instance with the initial value stored as the return value of the method `supplier()`.
When a new coroutine is started via `Go()`, `GoWait()` or `GoWaitResult()`, the value of the current coroutine is copied to the new coroutine.
When a new task is created via `WrapTask()`, `WrapWaitTask()` or `WrapWaitResultTask()`, the value of the current coroutine is captured to the new task.

## `WrapTask(fun Runnable) FutureTask[any]`

Create a new task and capture the `inheritableThreadLocals` from the current goroutine.
This function returns a `FutureTask` instance, but the return task will not run automatically.
You can run it in a sub-goroutine or goroutine-pool by `FutureTask.Run()` method, wait by `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method.
When the returned task run `panic` will be caught and error stack will be printed, the `panic` will be trigger again when calling `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method.

## `WrapWaitTask(fun CancelRunnable) FutureTask[any]`

Create a new task and capture the `inheritableThreadLocals` from the current goroutine.
This function returns a `FutureTask` instance, but the return task will not run automatically.
You can run it in a sub-goroutine or goroutine-pool by `FutureTask.Run()` method, wait by `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method.
When the returned task run `panic` will be caught, the `panic` will be trigger again when calling `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method.

## `WrapWaitResultTask[TResult any](fun CancelCallable[TResult]) FutureTask[TResult]`

Create a new task and capture the `inheritableThreadLocals` from the current goroutine.
This function returns a `FutureTask` instance, but the return task will not run automatically.
You can run it in a sub-goroutine or goroutine-pool by `FutureTask.Run()` method, wait and get result by `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method.
When the returned task run `panic` will be caught, the `panic` will be trigger again when calling `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method.

## `Go(fun Runnable)`

Start a new coroutine and automatically copy all contextual `inheritableThreadLocals` data of the current coroutine to the new coroutine.
Any `panic` while the child coroutine is executing will be caught and the stack automatically printed.

## `GoWait(fun CancelRunnable) FutureTask[any]`

Start a new coroutine and automatically copy all contextual `inheritableThreadLocals` data of the current coroutine to the new coroutine.
You can wait for the sub-coroutine to finish executing through the `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method that returns a value.
Any `panic` while the child coroutine is executing will be caught and thrown again when `FutureTask.Get()` or `FutureTask.GetWithTimeout()` is called.

## `GoWaitResult[TResult any](fun CancelCallable[TResult]) FutureTask[TResult]`

Start a new coroutine and automatically copy all contextual `inheritableThreadLocals` data of the current coroutine to the new coroutine.
You can wait for the sub-coroutine to finish executing and get the return value through the `FutureTask.Get()` or `FutureTask.GetWithTimeout()` method of the return value.
Any `panic` while the child coroutine is executing will be caught and thrown again when `FutureTask.Get()` or `FutureTask.GetWithTimeout()` is called.

[More API Documentation](https://pkg.go.dev/github.com/timandy/routine#section-documentation)

# Garbage Collection

`routine` allocates a `thread` structure for each coroutine, which stores context variable information related to the coroutine.

A pointer to this structure is stored on the `g.labels` field of the coroutine structure.

When the coroutine finishes executing and exits, `g.labels` will be set to `nil`, no longer referencing the `thread` structure.

The `thread` structure will be collected at the next `GC`.

If the data stored in `thread` is not additionally referenced, these data will be collected together.

# Support Grid

|                | **`darwin`** | **`linux`** | **`windows`** | **`freebsd`** | **`js`** |                |
|---------------:|:------------:|:-----------:|:-------------:|:-------------:|:--------:|:---------------|
|      **`386`** |              |      ✅      |       ✅       |       ✅       |          | **`386`**      |
|    **`amd64`** |      ✅       |      ✅      |       ✅       |       ✅       |          | **`amd64`**    |
|    **`armv6`** |              |      ✅      |               |               |          | **`armv6`**    |
|    **`armv7`** |              |      ✅      |               |               |          | **`armv7`**    |
|    **`arm64`** |      ✅       |      ✅      |               |               |          | **`arm64`**    |
|  **`loong64`** |              |      ✅      |               |               |          | **`loong64`**  |
|     **`mips`** |              |      ✅      |               |               |          | **`mips`**     |
|   **`mipsle`** |              |      ✅      |               |               |          | **`mipsle`**   |
|   **`mips64`** |              |      ✅      |               |               |          | **`mips64`**   |
| **`mips64le`** |              |      ✅      |               |               |          | **`mips64le`** |
|    **`ppc64`** |              |      ✅      |               |               |          | **`ppc64`**    |
|  **`ppc64le`** |              |      ✅      |               |               |          | **`ppc64le`**  |
|  **`riscv64`** |              |      ✅      |               |               |          | **`riscv64`**  |
|    **`s390x`** |              |      ✅      |               |               |          | **`s390x`**    |
|     **`wasm`** |              |             |               |               |    ✅     | **`wasm`**     |
|                | **`darwin`** | **`linux`** | **`windows`** | **`freebsd`** | **`js`** |                |

✅: Supported

# Thanks

Thanks to all [contributors](https://github.com/timandy/routine/graphs/contributors) for their contributions!

# *License*

`routine` is released under the [Apache License 2.0](LICENSE).

```
Copyright 2021-2024 TimAndy

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    https://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
```


routine

go routine control with context, support: Main, Go, Pool and some useful Executors.

routine
===
[![GoDoc](https://godoc.org/github.com/x-mod/routine?status.svg)](https://godoc.org/github.com/x-mod/routine) [![Go Report Card](https://goreportcard.com/badge/github.com/x-mod/routine)](https://goreportcard.com/report/github.com/x-mod/routine) [![Build Status](https://travis-ci.org/x-mod/routine.svg?branch=master)](https://travis-ci.org/x-mod/routine) [![Version](https://img.shields.io/github/tag/x-mod/routine.svg)](https://github.com/x-mod/routine/releases) [![Coverage Status](https://coveralls.io/repos/github/x-mod/routine/badge.svg?branch=master)](https://coveralls.io/github/x-mod/routine?branch=master)


## Routine Architecture

![](docs/routine.png)

## Quick Start

````go
package main

import (
	"log"
	"context"

	"github.com/x-mod/routine"
)

func main(){
	if err := routine.Main(
		context.TODO(),
		routine.Command("echo", routine.ARG("hello routine!")),
	); err != nil {
		log.Fatal(err)
	}
}
````

Or you can just clone the repo, then running `go run quickstart/main.go`.

## Main Routine

The most functional feature is providing the `Main` function abstraction, you can use the `routine.Main` to wrap your main function logic very quickly.

````go
package main

import (
	"context"
	"github.com/x-mod/routine"
)

func MainGo(ctx context.Context) error {
	log.Println("this is the Main Go func")
	return nil
}
func ChildGo(ctx context.Context) error {
	log.Println("this is the Child Go func")
	return nil
}
func prepareGo(ctx context.Context) error {
	log.Println("this is the prepare Go func")
	return nil
}
func cleanupGo(ctx context.Context) error {
	log.Println("this is the Clean Go func")
	return nil
}

func main(){
	log.Println(
		routine.Main(
			context.TODO(),
			//main Go
			routine.ExecutorFunc(MainGo),
			//prpare Go
			routine.Prepare(routine.ExecutorFunc(prepareGo)),
			//cleanup Go
			routine.Cleanup(routine.ExecutorFunc(cleanupGo)),
			routine.Go(routine.ExecutorFunc(ChildGo)),//child Go
			routine.Go(routine.ExecutorFunc(ChildGo)),
			routine.Go(routine.ExecutorFunc(ChildGo)),
			//signals
			routine.Signal(syscall.SIGINT, routine.SigHandler(func() {
				os.Exit(1)
			})),
		),
	)
}

````

## Routine

create and control your own routine by `routine.New`.

````go

import "github.com/x-mod/routine"

err := routine.New(opts...).Execute(ctx)

````

## Executors

The package provides many useful executor adapters for you:

- guarantee
- timeout & deadline
- retry & repeat
- concurrent
- crontab
- parallel & sequence
- command
- profiling

with these executor adapters, you can building the most complex goroutine logic.

````go

import "github.com/x-mod/routine"

//timeout
timeout := routine.Timeout(time.Minute, exec)

//retry
retry := routine.Retry(3, exec)

//repeat
repeat := routine.Repeat(10, time.Second, exec)

//concurrent
concurrent := routine.Concurrent(4, exec)

//schedule executor
crontab := routine.Crontab("* * * * *", exec)

//command
command := routine.Command("echo", routine.ARG("hello routine!"))

//parallel
parallel := routine.Parallel(exec1, exec2, exec3, ...)

//sequence
sequece := routine.Append(exec1, exec2, exec3, ...)
````

# Enjoy

More details, please check the [example](example/main.go) and trace it.

````bash
$: go run example/main.go

# trace go routine & tasks
$: go tool trace  trace.out
````

Then you can check the tasks like this:

![](docs/usetasks.png)


Semaphore pattern implementation with timeout of lock/unlock operations based on channel and context.

> # 🚦 semaphore
>
> Semaphore pattern implementation with timeout of lock/unlock operations.

[![Build][icon_build]][page_build]
[![Quality][icon_quality]][page_quality]
[![Documentation][icon_docs]][page_docs]
[![Coverage][icon_coverage]][page_coverage]
[![Awesome][icon_awesome]][page_awesome]

## 💡 Idea

The semaphore provides API to control access to a shared resource by multiple goroutines or limit throughput.

```go
releaser, err := semaphore.Acquire(breaker.BreakByTimeout(time.Second))
if err != nil {
	// timeout exceeded
}
defer releaser.Release()
```

Full description of the idea is available [here][design].

## 🏆 Motivation

...

## 🤼‍♂️ How to

### Quick start

```go
limiter := semaphore.New(1000)

http.HandleFunc("/", func(rw http.ResponseWriter, req *http.Request) {
	if _, err := limiter.Acquire(
		breaker.BreakByContext(
			context.WithTimeout(req.Context(), time.Second),
		),
	); err != nil {
		http.Error(rw, http.StatusText(http.StatusTooManyRequests), http.StatusTooManyRequests)
		return
	}
	defer limiter.Release()

	// handle request
})

log.Fatal(http.ListenAndServe(":80", http.DefaultServeMux))
```

## 🧩 Integration

The library uses [SemVer](https://semver.org) for versioning, and it is not
[BC](https://en.wikipedia.org/wiki/Backward_compatibility)-safe through major releases.
You can use [go modules](https://github.com/golang/go/wiki/Modules) or
[dep](https://golang.github.io/dep/) to manage its version.

The **[master][legacy]** is a feature frozen branch for versions **4.3.x** and no longer maintained.

```bash
$ dep ensure -add github.com/kamilsk/semaphore@4.3.1
```

The **[v4][]** branch is a continuation of the **[master][legacy]** branch for versions **v4.4.x**
to better integration with [go modules](https://github.com/golang/go/wiki/Modules).

```bash
$ go get -u github.com/kamilsk/semaphore/v4@v4.3.1
```

The **[v5][]** branch is an actual development branch.

```bash
$ go get -u github.com/kamilsk/semaphore    # inside GOPATH and for old Go versions

$ go get -u github.com/kamilsk/semaphore/v5 # inside Go module, works well since Go 1.11

$ dep ensure -add github.com/kamilsk/semaphore@v5.0.0-rc1
```

Version **v5** focused on integration with the 🚧 [breaker][] package.

## 🤲 Outcomes

### Console tool for command execution in parallel

This example shows how to execute many console commands in parallel.

```bash
$ semaphore create 2
$ semaphore add -- docker build
$ semaphore add -- vagrant up
$ semaphore add -- ansible-playbook
$ semaphore wait --timeout=1m --notify
```

[![asciicast][cli.preview]][cli.demo]

See more details [here][cli].

---

made with ❤️ for everyone

[icon_awesome]:     https://cdn.rawgit.com/sindresorhus/awesome/d7305f38d29fed78fa85652e3a63e154dd8e8829/media/badge.svg
[icon_build]:       https://travis-ci.org/kamilsk/semaphore.svg?branch=v5
[icon_coverage]:    https://api.codeclimate.com/v1/badges/0261f2170c785702034f/test_coverage
[icon_docs]:        https://godoc.org/github.com/kamilsk/semaphore?status.svg
[icon_quality]:     https://goreportcard.com/badge/github.com/kamilsk/semaphore

[page_awesome]:     https://github.com/avelino/awesome-go#goroutines
[page_build]:       https://travis-ci.org/kamilsk/semaphore
[page_coverage]:    https://codeclimate.com/github/kamilsk/semaphore/test_coverage
[page_docs]:        https://godoc.org/github.com/kamilsk/semaphore
[page_quality]:     https://goreportcard.com/report/github.com/kamilsk/semaphore

[breaker]:          https://github.com/kamilsk/breaker
[cli]:              https://github.com/kamilsk/parallel
[cli.demo]:         https://asciinema.org/a/136111
[cli.preview]:      https://asciinema.org/a/136111.png
[design]:           https://www.notion.so/octolab/semaphore-7d5ebf715d0141d1a8fa045c7966be3b?r=0b753cbf767346f5a6fd51194829a2f3
[egg]:              https://github.com/kamilsk/egg
[promo]:            https://github.com/kamilsk/semaphore

[legacy]:           https://github.com/kamilsk/semaphore/tree/master
[v4]:               https://github.com/kamilsk/semaphore/tree/v4
[v5]:               https://github.com/kamilsk/semaphore/projects/6

[tmp.docs]:         https://nicedoc.io/kamilsk/semaphore?theme=dark
[tmp.history]:      https://github.githistory.xyz/kamilsk/semaphore/blob/v5/README.md


The Best Go Libraries For Goroutines - Tools for managing and working with Goroutines. (31)

ants

artifex

async

async

async

async-job

breaker

channelify

conc

execpool

go-floc

go-flow

gohive

gollback

gowl

goworker

gowp

gpool

grpool

hands

Hunch

kyoo

neilotoole/errgroup

nursery

parallel-fn

pond

pool

rill

routine

routine

semaphore