pipe

README

pipe/s

A pipers bag - generic functions to gain concurrency - batteries included :-)

pipe/s - an extensive and systematic ensemble of concurrency-based mechanisms

• as ready-to-use generic production-grade code

• in different sizes such s and m (mind You: one size never fit's all)

• with comprehensively explained concepts, formulas and terminology - readme

• complemented with lots of 'how-to-do' and 'how-to-use' expamples.

• Follow a (hopefully) enlightening evolution.

• Begin with oversimplified (but popular!) code.

• Encounter dangerous pitfalls and easy-to-do mistakes along the way.

• Improve Your use of vaccines and protective habits.

• Ascend to implementations of some higher-level concepts such as 'load balancer', 'circular crawler' ...

• Eventually encounter the little-known land of concurrent lazy evaluation.

That's where we aim to go here - using go, common sense and playful discipline.

Are You ready for this?

    go get -u github.com/GoLangsam/pipe

Still in doubt? Have a peek at the short prolog.

And the overview is a good place to start...

May what You find here in this repo be a trustworthy and helpful companion along Your journey. Enjoy!

Please feel free and encouraged to suggest, improve, comment or ask - You'll be welcome!

---

Think deep - code happy - be simple - see clear :-)

---

Support on Beerpay

Hey dude! Help me out for a couple of 🍻!

Documentation

Index

• func AnyChan(inp ...Any) (out <-chan Any)
• func AnyChanFuncErr(gen func() (Any, error)) (out <-chan Any)
• func AnyChanFuncNok(gen func() (Any, bool)) (out <-chan Any)
• func AnyChanSlice(inp ...[]Any) (out <-chan Any)
• func AnyDaisyChaiN(inp chan Any, somany int, procs ...func(out chan<- Any, inp <-chan Any)) (out chan Any)
• func AnyDaisyChain(inp chan Any, procs ...func(out chan<- Any, inp <-chan Any)) (out chan Any)
• func AnyDone(inp <-chan Any) (done <-chan struct{})
• func AnyDoneFreq(inp <-chan Any) (freq <-chan map[Any]int64)
• func AnyDoneFreqAttr(inp <-chan Any, attr func(a Any) interface{}) (freq <-chan map[interface{}]int64)
• func AnyDoneFunc(inp <-chan Any, act func(a Any)) (done <-chan struct{})
• func AnyDoneLeave(inp <-chan Any, wg AnyWaiter) (done <-chan struct{})
• func AnyDoneSlice(inp <-chan Any) (done <-chan []Any)
• func AnyDoneWait(out chan<- Any, wg AnyWaiter) (done <-chan struct{})
• func AnyFan2(inp <-chan Any, inps ...Any) (out <-chan Any)
• func AnyFan2Chan(inp <-chan Any, inp2 <-chan Any) (out <-chan Any)
• func AnyFan2FuncErr(inp <-chan Any, gen func() (Any, error)) (out <-chan Any)
• func AnyFan2FuncNok(inp <-chan Any, gen func() (Any, bool)) (out <-chan Any)
• func AnyFan2Slice(inp <-chan Any, inps ...[]Any) (out <-chan Any)
• func AnyFanIn(inps ...<-chan Any) (out <-chan Any)
• func AnyFanIn1(inpS ...<-chan Any) (out <-chan Any)
• func AnyFanIn2(inp1, inp2 <-chan Any) (out <-chan Any)
• func AnyFanOut(inp <-chan Any, size int) (outS [](<-chan Any))
• func AnyFini() func(inp <-chan Any) (done <-chan struct{})
• func AnyFiniFunc(act func(a Any)) func(inp <-chan Any) (done <-chan struct{})
• func AnyFiniLeave(wg AnyWaiter) func(inp <-chan Any) (done <-chan struct{})
• func AnyFiniSlice() func(inp <-chan Any) (done <-chan []Any)
• func AnyFiniWait(wg AnyWaiter) func(out chan<- Any) (done <-chan struct{})
• func AnyFork(inp <-chan Any) (out1, out2 <-chan Any)
• func AnyForkSeen(inp <-chan Any) (new, old <-chan Any)
• func AnyForkSeenAttr(inp <-chan Any, attr func(a Any) interface{}) (new, old <-chan Any)
• func AnyJoin(out chan<- Any, inp ...Any) (done <-chan struct{})
• func AnyJoinChan(out chan<- Any, inp <-chan Any) (done <-chan struct{})
• func AnyJoinSlice(out chan<- Any, inp ...[]Any) (done <-chan struct{})
• func AnyMakeChan() (out chan Any)
• func AnyMerge(less func(i, j Any) bool, inps ...<-chan Any) (out <-chan Any)
• func AnyPair(inp <-chan Any) (out1, out2 <-chan Any)
• func AnyPipeAdjust(inp <-chan Any, sizes ...int) (out <-chan Any)
• func AnyPipeBuffered(inp <-chan Any, cap int) (out <-chan Any)
• func AnyPipeDone(inp <-chan Any) (out <-chan Any, done <-chan struct{})
• func AnyPipeEnter(inp <-chan Any, wg AnyWaiter) (out <-chan Any)
• func AnyPipeFunc(inp <-chan Any, act func(a Any) Any) (out <-chan Any)
• func AnyPipeFuncMany(inp <-chan Any, act func(a Any) Any, many int) (out <-chan Any)
• func AnyPipeLeave(inp <-chan Any, wg AnyWaiter) (out <-chan Any)
• func AnyPipeSeen(inp <-chan Any) (out <-chan Any)
• func AnyPipeSeenAttr(inp <-chan Any, attr func(a Any) interface{}) (out <-chan Any)
• func AnyPlug(inp <-chan Any, stop <-chan struct{}) (out <-chan Any, done <-chan struct{})
• func AnyPlugAfter(inp <-chan Any, after <-chan time.Time) (out <-chan Any, done <-chan struct{})
• func AnySame(same func(a, b Any) bool, inp1, inp2 <-chan Any) (out <-chan bool)
• func AnySendProxy(out chan<- Any, sizes ...int) chan<- Any
• func AnyStrew(inp <-chan Any, size int) (outS [](<-chan Any))
• func AnyTubeAdjust(sizes ...int) (tube func(inp <-chan Any) (out <-chan Any))
• func AnyTubeBuffered(cap int) (tube func(inp <-chan Any) (out <-chan Any))
• func AnyTubeEnter(wg AnyWaiter) (tube func(inp <-chan Any) (out <-chan Any))
• func AnyTubeFunc(act func(a Any) Any) (tube func(inp <-chan Any) (out <-chan Any))
• func AnyTubeLeave(wg AnyWaiter) (tube func(inp <-chan Any) (out <-chan Any))
• func AnyTubeSeen() (tube func(inp <-chan Any) (out <-chan Any))
• func AnyTubeSeenAttr(attr func(a Any) interface{}) (tube func(inp <-chan Any) (out <-chan Any))
• type Any
• type AnyProc
• type AnyWaiter

Functions

func AnyChan

func AnyChan(inp ...Any) (out <-chan Any)

AnyChan returns a channel to receive all inputs before close.

func AnyChanFuncErr

func AnyChanFuncErr(gen func() (Any, error)) (out <-chan Any)

AnyChanFuncErr returns a channel to receive all results of generator `gen` until `err != nil` before close.

func AnyChanFuncNok

func AnyChanFuncNok(gen func() (Any, bool)) (out <-chan Any)

AnyChanFuncNok returns a channel to receive all results of generator `gen` until `!ok` before close.

func AnyChanSlice

func AnyChanSlice(inp ...[]Any) (out <-chan Any)

AnyChanSlice returns a channel to receive all inputs before close.

func AnyDaisyChaiN

func AnyDaisyChaiN(
	inp chan Any,
	somany int,
	procs ...func(out chan<- Any, inp <-chan Any),
) (
	out chan Any,
)

AnyDaisyChaiN returns a channel to receive all inp after having passed `somany` times thru the process(es) (`from` right `into` left) before close.

Note: If `somany` is less than 1 or no `tubes` are provided, `out` shall receive elements from `inp` unaltered (as a convenience), thus making null values useful.

Note: AnyDaisyChaiN(inp, 1, procs) <==> AnyDaisyChain(inp, procs)

func AnyDaisyChain

func AnyDaisyChain(
	inp chan Any,
	procs ...func(out chan<- Any, inp <-chan Any),
) (
	out chan Any,
)

AnyDaisyChain returns a channel to receive all inp after having passed thru the process(es) (`from` right `into` left) before close.

Note: If no `tubes` are provided, `out` shall receive elements from `inp` unaltered (as a convenience), thus making a null value useful.

func AnyDone

func AnyDone(inp <-chan Any) (done <-chan struct{})

AnyDone returns a channel to receive one signal upon close and after `inp` has been drained.

func AnyDoneFreq

func AnyDoneFreq(inp <-chan Any) (freq <-chan map[Any]int64)

AnyDoneFreq returns a channel to receive a frequency histogram (as a `map[Any]int64`) upon close.

func AnyDoneFreqAttr

func AnyDoneFreqAttr(inp <-chan Any, attr func(a Any) interface{}) (freq <-chan map[interface{}]int64)

AnyDoneFreqAttr returns a channel to receive a frequency histogram (as a `map[interface{}]int64`) upon close.

`attr` provides the key to the frequency map. If `nil` is passed as `attr` then Any is used as key.

func AnyDoneFunc

func AnyDoneFunc(inp <-chan Any, act func(a Any)) (done <-chan struct{})

AnyDoneFunc will apply `act` to every `inp` and returns a channel to receive one signal upon close.

func AnyDoneLeave

func AnyDoneLeave(inp <-chan Any, wg AnyWaiter) (done <-chan struct{})

AnyDoneLeave returns a channel to receive one signal after all throughput on `inp` has been registered as departure on the given `sync.WaitGroup` before close.

func AnyDoneSlice

func AnyDoneSlice(inp <-chan Any) (done <-chan []Any)

AnyDoneSlice returns a channel to receive a slice with every Any received on `inp` upon close.

Note: Unlike AnyDone, AnyDoneSlice sends the fully accumulated slice, not just an event, once upon close of inp.

func AnyDoneWait

func AnyDoneWait(out chan<- Any, wg AnyWaiter) (done <-chan struct{})

AnyDoneWait returns a channel to receive one signal after wg.Wait() has returned and out has been closed before close.

Note: Use only *after* You've started flooding the facilities.

func AnyFan2

func AnyFan2(inp <-chan Any, inps ...Any) (out <-chan Any)

AnyFan2 returns a channel to receive everything from `inp` as well as all inputs before close.

func AnyFan2Chan

func AnyFan2Chan(inp <-chan Any, inp2 <-chan Any) (out <-chan Any)

AnyFan2Chan returns a channel to receive everything from `inp` as well as everything from `inp2` before close. Note: AnyFan2Chan is nothing but AnyFanIn2

func AnyFan2FuncErr

func AnyFan2FuncErr(inp <-chan Any, gen func() (Any, error)) (out <-chan Any)

AnyFan2FuncErr returns a channel to receive everything from `inp` as well as all results of generator `gen` until `err != nil` before close.

func AnyFan2FuncNok

func AnyFan2FuncNok(inp <-chan Any, gen func() (Any, bool)) (out <-chan Any)

AnyFan2FuncNok returns a channel to receive everything from `inp` as well as all results of generator `gen` until `!ok` before close.

func AnyFan2Slice

func AnyFan2Slice(inp <-chan Any, inps ...[]Any) (out <-chan Any)

AnyFan2Slice returns a channel to receive everything from `inp` as well as all inputs before close.

func AnyFanIn

func AnyFanIn(inps ...<-chan Any) (out <-chan Any)

AnyFanIn returns a channel to receive all inputs arriving on variadic inps before close.

Note: For each input one go routine is spawned to forward arrivals.

See AnyFanIn1 in `fan-in1` for another implementation.

Ref: https://blog.golang.org/pipelines
Ref: https://github.com/QuentinPerez/go-stuff/channel/Fan-out-Fan-in/main.go

func AnyFanIn1

func AnyFanIn1(inpS ...<-chan Any) (out <-chan Any)

AnyFanIn1 returns a channel to receive all inputs arriving on variadic inps before close.

Note: Only one go routine is used for all receives,
which keeps trying open inputs in round-robin fashion
until all inputs are closed.

See AnyFanIn in `fan-in` for another implementation.

func AnyFanIn2

func AnyFanIn2(inp1, inp2 <-chan Any) (out <-chan Any)

AnyFanIn2 returns a channel to receive all from both `inp1` and `inp2` before close.

func AnyFanOut

func AnyFanOut(inp <-chan Any, size int) (outS [](<-chan Any))

AnyFanOut returns a slice (of size = size) of channels each of which shall receive any inp before close.

func AnyFini

func AnyFini() func(inp <-chan Any) (done <-chan struct{})

AnyFini returns a closure around `AnyDone(_)`.

func AnyFiniFunc

func AnyFiniFunc(act func(a Any)) func(inp <-chan Any) (done <-chan struct{})

AnyFiniFunc returns a closure around `AnyDoneFunc(_, act)`.

func AnyFiniLeave

func AnyFiniLeave(wg AnyWaiter) func(inp <-chan Any) (done <-chan struct{})

AnyFiniLeave returns a closure around `AnyDoneLeave(_, wg)` registering throughput as departure on the given `sync.WaitGroup`.

func AnyFiniSlice

func AnyFiniSlice() func(inp <-chan Any) (done <-chan []Any)

AnyFiniSlice returns a closure around `AnyDoneSlice(_)`.

func AnyFiniWait

func AnyFiniWait(wg AnyWaiter) func(out chan<- Any) (done <-chan struct{})

AnyFiniWait returns a closure around `AnyDoneWait(_, wg)`.

func AnyFork

func AnyFork(inp <-chan Any) (out1, out2 <-chan Any)

AnyFork returns two channels either of which is to receive every result of inp before close.

func AnyForkSeen

func AnyForkSeen(inp <-chan Any) (new, old <-chan Any)

AnyForkSeen returns two channels, `new` and `old`, where `new` is to receive all `inp` not been seen before and `old` all `inp` seen before (internally growing a `sync.Map` to discriminate) until close.

func AnyForkSeenAttr

func AnyForkSeenAttr(inp <-chan Any, attr func(a Any) interface{}) (new, old <-chan Any)

AnyForkSeenAttr returns two channels, `new` and `old`, where `new` is to receive all `inp` whose attribute `attr` has not been seen before and `old` all `inp` seen before (internally growing a `sync.Map` to discriminate) until close.

func AnyJoin

func AnyJoin(out chan<- Any, inp ...Any) (done <-chan struct{})

AnyJoin sends inputs on the given out channel and returns a done channel to receive one signal when inp has been drained

func AnyJoinChan

func AnyJoinChan(out chan<- Any, inp <-chan Any) (done <-chan struct{})

AnyJoinChan sends inputs on the given out channel and returns a done channel to receive one signal when inp has been drained

func AnyJoinSlice

func AnyJoinSlice(out chan<- Any, inp ...[]Any) (done <-chan struct{})

AnyJoinSlice sends inputs on the given out channel and returns a done channel to receive one signal when inp has been drained

func AnyMakeChan

func AnyMakeChan() (out chan Any)

AnyMakeChan returns a new open channel (simply a 'chan Any' that is). Note: No 'Any-producer' is launched here yet! (as is in all the other functions).

This is useful to easily create corresponding variables such as:

var myAnyPipelineStartsHere := AnyMakeChan() // ... lot's of code to design and build Your favourite "myAnyWorkflowPipeline"

// ...
// ... *before* You start pouring data into it, e.g. simply via:
for drop := range water {

myAnyPipelineStartsHere <- drop

}

close(myAnyPipelineStartsHere)

Hint: especially helpful, if Your piping library operates on some hidden (non-exported) type
(or on a type imported from elsewhere - and You don't want/need or should(!) have to care.)

Note: as always (except for AnyPipeBuffer) the channel is unbuffered.

func AnyMerge

func AnyMerge(less func(i, j Any) bool, inps ...<-chan Any) (out <-chan Any)

AnyMerge returns a channel to receive all inputs sorted and free of duplicates. Each input channel needs to be sorted ascending and free of duplicates. The passed binary boolean function `less` defines the applicable order.

Note: If no inputs are given, a closed channel is returned.

func AnyPair

func AnyPair(inp <-chan Any) (out1, out2 <-chan Any)

AnyPair returns a pair of channels to receive every result of inp before close.

Note: Yes, it is a VERY simple fanout - but sometimes all You need.

func AnyPipeAdjust

func AnyPipeAdjust(inp <-chan Any, sizes ...int) (out <-chan Any)

AnyPipeAdjust returns a channel to receive all `inp` buffered by a AnySendProxy process before close.

func AnyPipeBuffered

func AnyPipeBuffered(inp <-chan Any, cap int) (out <-chan Any)

AnyPipeBuffered returns a buffered channel with capacity `cap` to receive all `inp` before close.

func AnyPipeDone

func AnyPipeDone(inp <-chan Any) (out <-chan Any, done <-chan struct{})

AnyPipeDone returns a channel to receive every `inp` before close and a channel to signal this closing.

func AnyPipeEnter

func AnyPipeEnter(inp <-chan Any, wg AnyWaiter) (out <-chan Any)

AnyPipeEnter returns a channel to receive all `inp` and registers throughput as arrival on the given `sync.WaitGroup` until close.

func AnyPipeFunc

func AnyPipeFunc(inp <-chan Any, act func(a Any) Any) (out <-chan Any)

AnyPipeFunc returns a channel to receive every result of action `act` applied to `inp` before close. Note: it 'could' be AnyPipeMap for functional people, but 'map' has a very different meaning in go lang.

func AnyPipeFuncMany

func AnyPipeFuncMany(inp <-chan Any, act func(a Any) Any, many int) (out <-chan Any)

AnyPipeFuncMany returns a channel to receive every result of action `act` applied to `inp` by `many` parallel processing goroutines before close.

ref: database/sql/sql_test.go
ref: cmd/compile/internal/gc/noder.go

func AnyPipeLeave

func AnyPipeLeave(inp <-chan Any, wg AnyWaiter) (out <-chan Any)

AnyPipeLeave returns a channel to receive all `inp` and registers throughput as departure on the given `sync.WaitGroup` until close.

func AnyPipeSeen

func AnyPipeSeen(inp <-chan Any) (out <-chan Any)

AnyPipeSeen returns a channel to receive all `inp` not been seen before while silently dropping everything seen before (internally growing a `sync.Map` to discriminate) until close. Note: AnyPipeFilterNotSeenYet might be a better name, but is fairly long.

func AnyPipeSeenAttr

func AnyPipeSeenAttr(inp <-chan Any, attr func(a Any) interface{}) (out <-chan Any)

AnyPipeSeenAttr returns a channel to receive all `inp` whose attribute `attr` has not been seen before while silently dropping everything seen before (internally growing a `sync.Map` to discriminate) until close. Note: AnyPipeFilterAttrNotSeenYet might be a better name, but is fairly long.

func AnyPlug

func AnyPlug(inp <-chan Any, stop <-chan struct{}) (out <-chan Any, done <-chan struct{})

AnyPlug returns a channel to receive every `inp` before close and a channel to signal this closing. Upon receipt of a stop signal, output is immediately closed, and for graceful termination any remaining input is drained before done is signalled.

func AnyPlugAfter

func AnyPlugAfter(inp <-chan Any, after <-chan time.Time) (out <-chan Any, done <-chan struct{})

AnyPlugAfter returns a channel to receive every `inp` before close and a channel to signal this closing. Upon receipt of a time signal (e.g. from `time.After(...)`), output is immediately closed, and for graceful termination any remaining input is drained before done is signalled.

func AnySame

func AnySame(same func(a, b Any) bool, inp1, inp2 <-chan Any) (out <-chan bool)

AnySame reads values from two channels in lockstep and iff they have the same contents then `true` is sent on the returned bool channel before close.

func AnySendProxy

func AnySendProxy(out chan<- Any, sizes ...int) chan<- Any

AnySendProxy returns a channel to serve as a sending proxy to 'out'. Uses a goroutine to receive values from 'out' and store them in an expanding buffer, so that sending to 'out' never blocks.

Note: the expanding buffer is implemented via "container/ring"

Note: AnySendProxy is kept for the Sieve example and other dynamic use to be discovered even so it does not fit the pipe tube pattern as AnyPipeAdjust does.

func AnyStrew

func AnyStrew(inp <-chan Any, size int) (outS [](<-chan Any))

AnyStrew returns a slice (of size = size) of channels one of which shall receive each inp before close.

func AnyTubeAdjust

func AnyTubeAdjust(sizes ...int) (tube func(inp <-chan Any) (out <-chan Any))

AnyTubeAdjust returns a closure around AnyPipeAdjust (_, sizes ...int).

func AnyTubeBuffered

func AnyTubeBuffered(cap int) (tube func(inp <-chan Any) (out <-chan Any))

AnyTubeBuffered returns a closure around PipeAnyBuffer (_, cap).

func AnyTubeEnter

func AnyTubeEnter(wg AnyWaiter) (tube func(inp <-chan Any) (out <-chan Any))

AnyTubeEnter returns a closure around AnyPipeEnter (_, wg) registering throughput as arrival on the given `sync.WaitGroup`.

func AnyTubeFunc

func AnyTubeFunc(act func(a Any) Any) (tube func(inp <-chan Any) (out <-chan Any))

AnyTubeFunc returns a closure around PipeAnyFunc (_, act).

func AnyTubeLeave

func AnyTubeLeave(wg AnyWaiter) (tube func(inp <-chan Any) (out <-chan Any))

AnyTubeLeave returns a closure around AnyPipeLeave (_, wg) registering throughput as departure on the given `sync.WaitGroup`.

func AnyTubeSeen

func AnyTubeSeen() (tube func(inp <-chan Any) (out <-chan Any))

AnyTubeSeen returns a closure around AnyPipeSeen() (silently dropping every Any seen before).

func AnyTubeSeenAttr

func AnyTubeSeenAttr(attr func(a Any) interface{}) (tube func(inp <-chan Any) (out <-chan Any))

AnyTubeSeenAttr returns a closure around AnyPipeSeenAttr() (silently dropping every Any whose attribute `attr` was seen before).

Types

type Any

type Any generic.Type

Any is the generic type flowing thru the pipe network.

type AnyProc

type AnyProc func(into chan<- Any, from <-chan Any)

AnyProc is the signature of the inner process of any linear pipe-network

Example: the identity proc:

samesame := func(into chan<- Any, from <-chan Any) { into <- <-from } Note: type AnyProc is provided for documentation purpose only. The implementation uses the explicit function signature in order to avoid some genny-related issue.

Note: In https://talks.golang.org/2012/waza.slide#40

Rob Pike uses a AnyProc named `worker`.

type AnyWaiter

type AnyWaiter interface {
	Add(delta int)
	Done()
	Wait()
}

AnyWaiter - as implemented by `*sync.WaitGroup` - attends Flapdoors and keeps counting who enters and who leaves.

Use AnyDoneWait to learn about when the facilities are closed.

Note: You may also use Your provided `*sync.WaitGroup.Wait()` to know when to close the facilities. Just: AnyDoneWait is more convenient as it also closes the primary channel for You.

Just make sure to have _all_ entrances and exits attended, and `Wait()` only *after* You've started flooding the facilities.