Go module that processes work concurrently and returns output in a channel in the order of input

<a href="https://github.com/tejzpr/ordered-concurrently/actions/workflows/tests.yml"><img src="https://github.com/tejzpr/ordered-concurrently/actions/workflows/tests.yml/badge.svg" alt="Tests"/></a>
[![codecov](https://codecov.io/gh/tejzpr/ordered-concurrently/branch/master/graph/badge.svg?token=6WIXWRO3EW)](https://codecov.io/gh/tejzpr/ordered-concurrently)
[![Go Reference](https://pkg.go.dev/badge/github.com/tejzpr/ordered-concurrently.svg)](https://pkg.go.dev/github.com/tejzpr/ordered-concurrently)
[![Gitpod ready-to-code](https://img.shields.io/badge/Gitpod-ready--to--code-blue?logo=gitpod)](https://gitpod.io/#https://github.com/tejzpr/ordered-concurrently)
[![Go Report Card](https://goreportcard.com/badge/github.com/tejzpr/ordered-concurrently)](https://goreportcard.com/report/github.com/tejzpr/ordered-concurrently)

# Ordered Concurrently
A library for parallel processing with ordered output in Go. This module processes work concurrently / in parallel and returns output in a channel in the order of input. It is useful in concurrently / parallelly processing items in a queue, and get output in the order provided by the queue.

# Usage 
## Get Module
```go
go get github.com/tejzpr/ordered-concurrently/v3
```
## Import Module in your source code
```go
import concurrently "github.com/tejzpr/ordered-concurrently/v3" 
```
## Create a work function by implementing WorkFunction interface
```go
// Create a type based on your input to the work function
type loadWorker int

// The work that needs to be performed
// The input type should implement the WorkFunction interface
func (w loadWorker) Run(ctx context.Context) interface{} {
	time.Sleep(time.Millisecond * time.Duration(rand.Intn(100)))
	return w * 2
}
```
## Demo 
[Go Playground](https://go.dev/play/p/60b_x0YHzYu)

## Run
### Example - 1
```go
func main() {
	max := 10
	inputChan := make(chan concurrently.WorkFunction)
	ctx := context.Background()
	output := concurrently.Process(ctx, inputChan, &concurrently.Options{PoolSize: 10, OutChannelBuffer: 10})
	go func() {
		for work := 0; work < max; work++ {
			inputChan <- loadWorker(work)
		}
		close(inputChan)
	}()
	for out := range output {
		log.Println(out.Value)
	}
}
```
### Example - 2 - Process unknown number of inputs
```go
func main() {
	inputChan := make(chan concurrently.WorkFunction, 10)
	ctx := context.Background()
	output := concurrently.Process(ctx, inputChan, &concurrently.Options{PoolSize: 10, OutChannelBuffer: 10})

	ticker := time.NewTicker(100 * time.Millisecond)
	done := make(chan bool)
	wg := &sync.WaitGroup{}
	go func() {
		input := 0
		for {
			select {
			case <-done:
				return
			case <-ticker.C:
				inputChan <- loadWorker(input)
				wg.Add(1)
				input++
			default:
			}
		}
	}()

	var res []loadWorker
	go func() {
		for out := range output {
			res = append(res, out.Value.(loadWorker))
			wg.Done()
		}
	}()

	time.Sleep(1600 * time.Millisecond)
	ticker.Stop()
	done <- true
	close(inputChan)
	wg.Wait()

	// Check if output is sorted
	isSorted := sort.SliceIsSorted(res, func(i, j int) bool {
		return res[i] < res[j]
	})
	if !isSorted {
		log.Println("output is not sorted")
	}
}
```
# Credits
1.  [u/justinisrael](https://www.reddit.com/user/justinisrael/) for inputs on improving resource usage.
2.  [mh-cbon](https://github.com/mh-cbon) for identifying potential [deadlocks](https://github.com/tejzpr/ordered-concurrently/issues/2).


ordered-concurrently

FIFO Pipeline which parallels execution on each stage while maintaining the order of messages and results

Parapipe - paralleling pipeline
===============================

[![Mentioned in Awesome Go](https://awesome.re/mentioned-badge-flat.svg)](https://github.com/avelino/awesome-go)
[![tests](https://github.com/nazar256/parapipe/actions/workflows/tests.yml/badge.svg)](https://github.com/nazar256/parapipe/actions/workflows/tests.yml)
[![linters](https://github.com/nazar256/parapipe/actions/workflows/linters.yml/badge.svg)](https://github.com/nazar256/parapipe/actions/workflows/linters.yml)
[![coverage](https://codecov.io/gh/nazar256/parapipe/branch/main/graph/badge.svg?token=N6NI66KPXG)](https://codecov.io/gh/nazar256/parapipe)
[![Go Report Card](https://goreportcard.com/badge/github.com/nazar256/parapipe)](https://goreportcard.com/report/github.com/nazar256/parapipe)
[![GoDoc](https://godoc.org/github.com/nazar256/parapipe?status.svg)](https://godoc.org/github.com/nazar256/parapipe)

The library provides a zero-dependency non-blocking buffered FIFO-pipeline 
for structuring the code and vertically scaling your app. 
Unlike regular pipeline examples you may find on the internet - parapipe executes everything on each step concurrently,
yet maintaining the output order. Although, this library does not use any locks or mutexes. Just pure channels.

When to use
-----------

* processed data can be divided in chunks (messages), and the flow may consist of one or more stages
* data should be processed concurrently (scaled vertically)
* the order of processing messages must be maintained

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

```
go get -u github.com/nazar256/parapipe@latest
```

Usage
-----

1. Create a pipeline with first step. Processing callback is generic (so as the pipeline). 
It may receive and return any type of data, but the second return value should always be a boolean.

```go
concurrency := runtime.NumCPU()     // how many messages to process concurrently for each pipe
pipeline := parapipe.NewPipeline(concurrency, func(msg YourInputType) (YourOutputType, bool) {
    // do something and generate a new value "someValue"
    shouldProceedWithNextStep := true
    return someValue, shouldProceedWithNextStep
})
```

2. Add pipes - call `Attach()` function one or more times to add steps to the pipeline
```go
p1 := parapipe.NewPipeline(runtime.NumCPU(), func(msg int) (int, bool) {
    time.Sleep(30 * time.Millisecond)
    return msg + 1000, true
})
p2 := parapipe.Attach(p1, parapipe.NewPipeline(concurrency, func(msg int) (string, bool) {
    time.Sleep(30 * time.Millisecond)
    return strconv.Itoa(msg), true
}))

// final pipeline you are going to work with (push messages and read output)
pipeline := parapipe.Attach(p2, parapipe.NewPipeline(concurrency, func(msg string) (string, bool) {
    time.Sleep(30 * time.Millisecond)
    return "#" + msg, true
}))
```

3. Get "out" channel when all pipes are added and read results from it
```go
for result := range pipeline.Out() {
    // do something with the result
}
```
It's **important** to drain the pipeline (read everything from "out") even when the pipeline won't produce any viable result. 
It could be stuck otherwise.

4. Push values for processing into the pipeline:
```go
pipeline.Push("something")
```   

5. Close pipeline to after the last message. This will cleanup its resources and close its output channel. 
   It's not recommended closing pipeline using `defer` because you may not want to hang output util defer is executed.
```go
pipeline.Close()
```   

### Circuit breaking

In some cases (errors) there could be impossible to process a message, thus there is no way to pass it further.
In such case just return `false` as a second return value from the step processing callback. 
The first value will be ignored.

```go
pipeline.Pipe(4, func(inputValue InputType) (OutputType, bool) {
    someValue, err := someOperation(inputValue)
    if err != nil {
		// handle the error
		// slog.Error("error when calling someOperation", "err", err)
        return someValue, false
    }
    return someValue, true
})
// ...
for result := range pipeline.Out() {
    // do something with the result
}
```

### Performance

Parapipe makes use of generics and channels.
Overall it should be performant enough for most of the cases.
It has zero heap allocations in hot code, thus generates little load for garbage collector.
However, it uses channels under the hood and is bottlenecked mostly by the channel operations which are several
writes and reads per each message.

Examples
--------

### AMQP middleware

Parapipe can be handful when you need to process messages in the middle concurrently, yet maintaining their order.

See the [working example of using parapipe in AMQP client](http://github.com/nazar256/go-amqp-sniffer/blob/a5c5db375dc68a2e83c24686e4e57a63cf08c80b/sniffer/sniffer.go#L49-L108).

### Other examples

With parapipe you can:

  * in your API respond a long JSON-feed as stream, retrieve, enrich and marshal each object concurrently, in maintained order and return them to the client
  * fetch and merge entries from different sources as one stream
  * structure your API controllers or handlers
  * processing heavy files in effective way


parapipe

An implementation of pipelines with fan-in and fan-out

[![GoDoc](https://godoc.org/github.com/hyfather/pipeline?status.svg)](https://godoc.org/github.com/hyfather/pipeline)
[![Build Status](https://travis-ci.org/hyfather/pipeline.svg?branch=master)](https://travis-ci.org/hyfather/pipeline)
[![cover.run](https://cover.run/go/github.com/hyfather/pipeline.svg?style=flat&tag=golang-1.10)](https://cover.run/go?tag=golang-1.10&repo=github.com%2Fhyfather%2Fpipeline)
[![Go Report Card](https://goreportcard.com/badge/github.com/hyfather/pipeline)](https://goreportcard.com/report/github.com/hyfather/pipeline)

# pipeline

This package provides a simplistic implementation of Go pipelines
as outlined in [Go Concurrency Patterns: Pipelines and cancellation.](https://blog.golang.org/pipelines)

# Docs
GoDoc available [here.](https://godoc.org/github.com/hyfather/pipeline)

# Example Usage

```
import "github.com/hyfather/pipeline"

p := pipeline.New()
p.AddStageWithFanOut(myStage, 10)
p.AddStageWithFanOut(anotherStage, 100)
doneChan := p.Run(inChan)

<- doneChan
```

More comprehensive examples can be found [here.](./examples)


pipeline

Generic pipeline functions for concurrent processing

[![Go Reference](https://pkg.go.dev/badge/github.com/nxdir-s/pipelines.svg)](https://pkg.go.dev/github.com/nxdir-s/pipelines)
[![Go Report Card](https://goreportcard.com/badge/github.com/nxdir-s/pipelines)](https://goreportcard.com/report/github.com/nxdir-s/pipelines)
[![Go Coverage](https://github.com/nxdir-s/pipelines/wiki/coverage.svg)](https://raw.githack.com/wiki/nxdir-s/pipelines/coverage.html)

# pipelines

Pipelines contains generic functions that help with concurrent processing

## Usage

A pipeline can be created from a slice or map

```go
stream := pipelines.StreamSlice(ctx, data)
```

Or from a generator function

```go
func GenerateData(ctx context.Context) int { return rand.Intn(10) }

stream := pipelines.GenerateStream(ctx, GenerateData)
```

### FanOut

`FanOut` can be used to process data concurrently. Useful for I/O bound processes, but it can be used in any situation where you have a slice or map of data and want to introduce concurrency

```go
const MaxFan int = 3

fanOutChannels := pipelines.FanOut(ctx, stream, ProcessFunc, MaxFan)
```

### FanIn

`FanIn` can be used to merge data into one channel

```go
fanInData := pipelines.FanIn(ctx, fanOutChannels...)
```

## Example

```go
package main

import (
    "context"
    "fmt"
    "math/rand"
    "os"
    "os/signal"
    "strconv"
    "time"

    "github.com/nxdir-s/pipelines"
)

const (
    MaxFan int = 3
)

func GenerateData(ctx context.Context) int {
    return rand.Intn(5)
}

func Process(ctx context.Context, timeout int) string {
    select {
    case <-ctx.Done():
        return "context cancelled"
    case <-time.After(time.Second * time.Duration(timeout)):
        return "slept for " + strconv.Itoa(timeout) + " seconds!"
    }
}

func Read(ctx context.Context, messages <-chan string) {
    for msg := range messages {
        select {
        case <-ctx.Done():
            return
        default:
            fmt.Fprintf(os.Stdout, "%s\n", msg)
        }
    }
}

func main() {
    ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt)
    defer cancel()

    stream := pipelines.GenerateStream(ctx, GenerateData)
    fanOutChannels := pipelines.FanOut(ctx, stream, Process, MaxFan)
    messages := pipelines.FanIn(ctx, fanOutChannels...)

    go Read(ctx, messages)

    select {
    case <-ctx.Done():
        fmt.Fprint(os.Stdout, "context canceled, exiting...\n")
        os.Exit(0)
    }
}
```


The Best Go Libraries For Data Structures and Algorithms - Pipes (4)

ordered-concurrently

parapipe

pipeline

pipelines