concur

README

concur 🏎️

concur is a lightweight, type-safe, production-ready Go library that implements the Fan-Out / Fan-In concurrency pattern using Go Generics.

It distributes resource-heavy workloads across a controlled pool of parallel workers and multiplexes their results back into a single fanned-in stream, completely preventing unbounded goroutine leaks, data races, and memory spikes.

---

✨ Features

• Strict Type Safety: Built using Go Generics—no slow interface{} reflections or dynamic runtime type-casting.
• Bounded Scaling: Enforces a strict, maximum worker pool size to protect system resources.
• Configurable Error Strategies: Choose whether an error or panic halts the entire pipeline instantly (StopOnError: true) or lets processing continue (StopOnError: false).
• Context-Aware: Native support for context.Context cancellation and timeouts.
• Automated Panic Recovery: Built-in catch routines capture user function panics, seamlessly translating them into manageable Go errors.
• Zero Signature Friction: Accepts a single robust API format. Users can wrap any function signature using standard Go closures without writing tedious wrapper boilerplate.

---

💎 Why Use concur? (The Pros)

Implementing manual fan-out/fan-in pipelines requires writing complex boilerplates using channels, sync.WaitGroup, context tracking, and select blocks. concur abstracts this safely.

• 🛡️ Shielded Memory Allocation (Bounded Scaling): Instead of spinning up an arbitrary number of goroutines that could trigger out-of-memory (OOM) crashes under high loads, it enforces a strict maximum worker pool boundary.
• ⚡ Native Type Safety (Zero Reflection): Built fully using Go Generics. It enforces static compile-time type check safety without resorting to slow runtime interface{} type casting or reflection.
• 🛑 Advanced Error Interception: You can toggle the StopOnError mode. If a single item fails, it immediately triggers an internal structural context cancellation to stop all other active workers, avoiding wasted compute.
• 🫧 Zero Goroutine Leaks: The engine guarantees that all worker sub-routines cleanly exit and internal contexts are entirely torn down, even if down-stream channels stop reading data early.
• 🧩 Adaptable to "Any Function Signature": By packing complex arguments into single custom input/output structs or leveraging direct function closures, you can map any business function signature cleanly into the pipeline.

---

🛠️ Installation

go get github.com/jigneshsatam/concur

---

🏎️ Core Usage Example

Here is how to map a slice of custom structural inputs across a pool of parallel workers while using inline closure adaptation and continuing past isolated items errors.

package main

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

	"github.com/jigneshsatam/concur"
)

// Wrap your multiple function input arguments
type ProcessInput struct {
	TaskID   int
	TargetIP string
}

// Wrap your multiple function return values
type ProcessOutput struct {
	BytesWritten int
	IsVerified   bool
}

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

	// 1. Populate the work stream queue
	inputChan := make(chan ProcessInput, 3)
	inputChan <- ProcessInput{TaskID: 1, TargetIP: "192.168.1.5"}
	inputChan <- ProcessInput{TaskID: 2, TargetIP: "10.0.0.1"} // Malformed error item
	inputChan <- ProcessInput{TaskID: 3, TargetIP: "192.168.1.9"}
	close(inputChan)

	// 2. Configure behavior (0 workers automatically defaults to 4)
	opts := concur.Options{
		Workers:     0,
		StopOnError: false, // Continue executing the queue if an item errors out
	}

	// 3. Fire up the fanned-out pool using a Closure Adapter
	resultsStream := concur.Process(ctx, inputChan, opts, func(ctx context.Context, item ProcessInput) (ProcessOutput, error) {
		if item.TargetIP == "10.0.0.1" {
			return ProcessOutput{}, errors.New("network routing rejection")
		}
		return ProcessOutput{BytesWritten: 128, IsVerified: true}, nil
	})

	// 4. Consume fanned-in aggregated results natively
	for res := range resultsStream {
		if res.Err != nil {
			fmt.Printf("[❌ Error/Panic Caught]: %v\n", res.Err)
			continue
		}
		fmt.Printf("[✅ Success] Bytes: %d | Validated: %t\n", res.Value.BytesWritten, res.Value.IsVerified)
	}
}

---

📂 Repository Structure & Examples

The repository is structured as a step-by-step learning path, ranging from basic introduction programs to advanced multi-stage data stream systems:

examples/
|── 01_basic/
│        └── main.go # Pure onboarding (Squaring numbers in parallel)
|── 02_variations/
│        └── main.go # Master reference layout for all 12 combinations
|── 03_advanced_pipeline/
        └── main.go # Advanced multi-stage streaming pipeline 🔥

You can execute any of these scenarios locally from the root of your workspace:

Run the basic introduction program

go run ./examples/01_basic/main.go

Run the 12-signature syntax cheat-sheet reference block

go run ./examples/02_variations/main.go

Run the complex multi-stage streaming pipeline

go run ./examples/03_advanced_pipeline/main.go

---

🧩 Adapting All 12 Signature Variations

Because concur uses Go's type-inference engine, you never need to write messy bracketed types like [int, string]. You can adapt all 12 combinations of inputs and outputs directly at the call-site using standard Go anonymous closures:

Group 1: Signatures with Context & Item (ctx, item)

1. Full Match: func(ctx, In) (Out, error)

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (string, error) {
	return myFunc(ctx, item)
})

2. No Error: func(ctx, In) Out

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (string, error) {
	return myFunc(ctx, item), nil
})

3. Side-Effect Only with Error: func(ctx, In) error

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (struct{}, error) {
	return struct{}{}, myFunc(ctx, item)
})

4. Pure Side-Effect: func(ctx, In)

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (struct{}, error) {
	myFunc(ctx, item)
	return struct{}{}, nil
})

---

Group 2: Signatures with Item Only (item)

5. Standard Go Return: func(In) (Out, error)

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (string, error) {
	return myFunc(item)
})

6. Pure Transformation: func(In) Out

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (string, error) {
	return myFunc(item), nil
})

7. Side-Effect with Error: func(In) error

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (struct{}, error) {
	return struct{}{}, myFunc(item)
})

8. Local Side-Effect: func(In)

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (struct{}, error) {
	myFunc(item)
	return struct{}{}, nil
})

---

Group 3: Standalone Background Tasks ()

(Note: When using these, the channel stream acts strictly as a worker queue counter to trigger background operations).

9. Task with Result and Error: func() (Out, error)

out := concur.Process(ctx, in, opts, func(ctx context.Context, _ int) (string, error) {
	return myFunc()
})

10. Task with Result Only: func() Out

out := concur.Process(ctx, in, opts, func(ctx context.Context, _ int) (string, error) {
	return myFunc(), nil
})

11. Task with Error Only: func() error

out := concur.Process(ctx, in, opts, func(ctx context.Context, _ int) (struct{}, error) {
	return struct{}{}, myFunc()
})

12. Pure Standalone Job: func()

out := concur.Process(ctx, in, opts, func(ctx context.Context, _ int) (struct{}{}, error) {
	myFunc()
	return struct{}{}, nil
})

---

⚡ Panic Safety Guardrails

Unexpected runtime panics inside your closure workloads will not bring down your entire Go application server infrastructure.

concur wraps executions in an inner recovery loop. If a closure panics, the worker captures the payload, normalizes it into a standard Go error structure format (worker panicked: <reason>), and pushes it down the fanned-in result stream:

opts := concur.Options{StopOnError: false} // Keep other workers alive if one panics

out := concur.Process(ctx, in, opts, func(ctx context.Context, item int) (int, error) {
	if item == 42 {
		panic("malformed byte payload crash!")
	}
	return item * 2, nil
})

for res := range out {
	if res.Err != nil {
		// Output: "worker panicked: malformed byte payload crash!"
		fmt.Println("Handled cleanly:", res.Err)
	}
}

---

🧪 Testing

To run unit tests along with Go's automated Data Race Detector:

go test -v -race ./...

📊 Benchmarking

To benchmark the parallel processing gains of concur against a traditional sequential execution loop, run:

go test -bench=. -benchmem

Documentation

Overview

Package concur is a lightweight, type-safe, production-ready Go library that implements the Fan-Out / Fan-In concurrency pattern using Go Generics.

It distributes resource-heavy workloads across a controlled pool of parallel workers and multiplexes their results back into a single fanned-in stream, completely preventing unbounded goroutine leaks and memory spikes.

Core Concepts

The pipeline relies on a few fundamental paradigms to guarantee safe executions:

• Bounded Scaling: Enforces a strict, maximum worker pool size to protect system memory.
• Short-Circuit Error Propagation: Halts the entire pipeline instantly upon encountering the first item failure if configured via options.
• Automated Panic Recovery: Built-in catch routines shield your infrastructure by translating unexpected worker runtime panics into standardized, manageable Go errors.

Usage with Closure Adaptation

Because the primary Process function utilizes a highly comprehensive function signature, users can leverage standard Go anonymous closures at the call-site to adapt any custom worker function format (e.g., functions without a context parameter, functions that don't return errors, or pure background tasks) completely bypassing messy bracketed type syntax.

A simple example using inline closure adaptation:

inputChan := make(chan int, 3)
// ... populate and close channel ...

opts := concur.Options{Workers: 4, StopOnError: false}
results := concur.Process(ctx, inputChan, opts, func(ctx context.Context, item int) (string, error) {
	return fmt.Sprintf("Processed: %d", item), nil
})

for res := range results {
	if res.Err != nil {
		log.Printf("Worker task failed: %v", res.Err)
		continue
	}
	fmt.Println(res.Value)
}

Index

• func Process[In any, Out any](ctx context.Context, in <-chan In, opts Options, workerFn WorkerFunc[In, Out]) <-chan Result[Out]
• type Options
• type Result
• type WorkerFunc

Functions

func Process

func Process[In any, Out any](
	ctx context.Context,
	in <-chan In,
	opts Options,
	workerFn WorkerFunc[In, Out],
) <-chan Result[Out]

Process runs a pool of workers to concurrently process data using the provided options.

Types

type Options

type Options struct {
	// Workers sets the maximum size of the execution goroutine pool.
	// If left at 0, it defaults to 4 workers.
	Workers int

	// If true, the first error (or panic) halts the entire pipeline instantly.
	// If false, errored items pass through, and processing continues.
	StopOnError bool
}

Options configures the operational pipeline behavior.

type Result

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

Result wraps the fanned-in worker output data and its error state.

type WorkerFunc

type WorkerFunc[In any, Out any] func(ctx context.Context, item In) (Out, error)

WorkerFunc accepts a context and an input item, returning a result and an error.