Flow - FBP / pipelines / workers pool

Package flow provides support for very basic FBP / pipelines. It helps to structure multistage processing as a set of independent handlers communicating via channels. The typical use case is for ETL (extract, transform, load) type of processing. Package flow doesn't introduce any high-level abstraction and keeps everything in the hand of the user.

Package pool provides a simplified version of flow suitable for cases with a single-handler flows.

Details about flow package

• Each handler represents an async stage. It consumes data from an input channel and publishes results to an output channel.
• Each handler runs in a separate goroutine.
• User must implement Handler functions and add it to the Flow.
• Each handler usually creates an output channel, reads from the input channel, processes data, sends results to the output channel and closes the output channel.
• Processing sequence determined by the order of those handlers.
• Any Handler can run in multiple concurrent goroutines (workers) by using the Parallel decorator.
• FanOut allows to pass multiple handlers in broadcast mode, i.e., each handler gets every input record. Outputs from these handlers merged into single output channel.
• Processing error detected as return error value from user's handler func. Such error interrupts all other running handlers gracefully and won't keep any goroutine running/leaking.
• Each Flow object can be executed only once.
• Handler should handle context cancellation as a termination signal.

Install and update

go get -u github.com/go-pkgz/flow

Example of the flow's handler

// ReaderHandler creates flow.Handler, reading strings from any io.Reader
func ReaderHandler(reader io.Reader) Handler {
	return func(ctx context.Context, ch chan any) (chan any, func() error) {
		metrics := flow.GetMetrics(ctx) // metrics collects how many records read with "read" key.

		readerCh := make(chan any, 1000)
		readerFn := func() error {
			defer close(readerCh)

			scanner := bufio.NewScanner(reader)
			for scanner.Scan() {

				select {
				case readerCh <- scanner.Text():
					metrics.Inc("read")
				case <-ctx.Done():
					return ctx.Err()
				}
			}
			return errors.Wrap(scanner.Err(), "scanner failed")
		}
		return readerCh, readerFn
	}
}

Usage of the flow package

for complete example see example

// flow illustrates the use of a Flow for concurrent pipeline running each handler in separate goroutine.
func ExampleFlow_flow() {

	f := New() // create new empty Flow
	f.Add(     // add handlers. Note: handlers can be added directly in New

		// first handler, generate 100 initial values.
		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
			out = make(chan any, 100) // example of non-async handler
			for i := 1; i <= 100; i++ {
				out <- i
			}
			close(out)      // each handler has to close out channel
			return out, nil // no runnable function for non-async handler
		},

		// second handler - picks odd numbers only and multiply
		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
			out = make(chan any) // async handler makes its out channel
			runFn = func() error {
				defer close(out) // handler should close out channel
				for e := range in {
					val := e.(int)
					if val%2 == 0 {
						continue
					}
					f.Metrics().Inc("passed") // increment user-define metric "passed"

					// send result to the next stage with flow.Send helper. Also checks for cancellation
					if err := Send(ctx, out, val*rand.Int()); err != nil {
						return err
					}
				}
				return nil
			}
			return out, runFn
		},

		// final handler - sum all numbers
		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
			out = make(chan any, 1)
			runFn = func() error {
				defer close(out)
				sum := 0
				for {
					select {
					case e, more := <-in:
						if !more {
							out <- sum //send result
							return nil
						}
						val := e.(int)
						sum += val

					case <-ctx.Done():
						return ctx.Err()
					}
				}
			}
			return out, runFn
		},
	)

	f.Go() // activate flow

	// wait for all handlers to complete
	if err := f.Wait(); err == nil {
		fmt.Printf("all done, result=%v, passed=%d", <-f.Channel(), f.Metrics().Get("passed"))
	}
}

// illustrates the use of a Flow for concurrent pipeline running some handlers in parallel way.
func ExampleFlow_parallel() {

	f := New() // create new empty Flow

	// make flow with mixed singles and parallel handlers and activate
	f.Add(

		// generate 100 initial values in single handler
		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
			out = make(chan any, 100) // example of non-async handler
			for i := 1; i <= 100; i++ {
				out <- i
			}
			close(out)      // each handler has to close out channel
			return out, nil // no runnable function for non-async handler
		},

		// multiple all numbers in 10 parallel handlers
		f.Parallel(10, func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
			out = make(chan any) // async handler makes its out channel
			runFn = func() error {
				defer close(out) // handler should close out channel
				for e := range in {
					val := e.(int)
					select {
					case out <- val * rand.Int(): // send result to the next stage
					case <-ctx.Done(): // check for cancellation
						return ctx.Err()
					}
				}
				return nil
			}
			return out, runFn
		}),

		// print all numbers
		func(ctx context.Context, in chan any) (out chan any, runFn func() error) {
			runFn = func() error {
				defer close(out)
				sum := 0
				for e := range in {
					val := e.(int)
					sum += val
					select {
					case <-ctx.Done():
						return ctx.Err()
					default:
					}
				}
				fmt.Printf("all done, result=%d", sum)
				return nil
			}
			return out, runFn
		},
	)

	// wait for all handlers to complete
	if err := f.Wait(); err == nil {
		fmt.Printf("all done, result=%v", <-f.Channel())
	}
}

Details about pool package

Pool package provides thin implementation of workers pool. In addition to the default "run a func in multiple goroutines" mode, it also provides an optional support of chunked workers. In this mode each key, detected by user-provide func, guaranteed to be processed by the same worker. Such mode needed for stateful flows where each set of input records has to be processed sequentially and some state should be kept. Another thing pool allows to define is the batch size. This one is a simple performance optimization collecting input request into a buffer and send them to worker channel in batches (slices) instead of per-submit call.

Options:

• ChunkFn - the function returns string identifying the chunk
• Batch - sets batch size (default 1)
• ResChanSize sets the size of the results channel (default 0, i.e. unbuffered)
• WorkerChanSize sets the size of workers channels (default 0, i.e. unbuffered)
• ContinueOnError allows workers continuation after error occurred
• OnCompletion sets callback called for each worker after its input channel closed and drained. A worker terminated early, by an error with the default fail-fast or by a canceled context, doesn't get it.

worker function

Worker function passed by user and runs in multiple workers (goroutines) concurrently. This is the function: type WorkerFn func(ctx context.Context, inpRec any, sender SenderFn, store WorkerStore) error

It takes inp parameter, does the job and optionally send result(s) with SenderFn to the common results channel. Error will terminate all workers unless ContinueOnError set.

Note: workerFn can be stateful, collect anything it needs and sends 0 or more results by calling SenderFn one or more times.

worker store

Each worker gets WorkerStore and can be used as per-worker storage for any intermediate results. The store is not synchronised internally, it doesn't need to be as only the owning worker touches it.

type WorkerStore interface {
	Set(key string, val any)
	Get(key string) (any, bool)
	GetInt(key string) int
	GetFloat(key string) float64
	GetString(key string) string
	GetBool(key string) bool
	Keys() []string
	Delete(key string)
}

alternatively state can be kept outside of workers as a slice of values and accessed by worker ID.

usage

    p := pool.New(8, func(ctx context.Context, v any, sendFn pool.SenderFn, ws pool.WorkerStore) error {
        // worker function gets input v, processes it and sends result(s) to the common results channel

        input, ok := v.(string) // in this case it gets string as input
        if !ok {
            return errors.New("incorrect input type")
        }
        // do something with input
        // ...

        count := ws.GetInt("something") // access worker-local var

        if err := sendFn(input + " processed"); err != nil { // send the result
            return err
        }
        if err := sendFn(input + " processed again"); err != nil { // 0 or more results per input
            return err
        }
        pool.Metrics(ctx).Inc("counter")
        ws.Set("something", count+1) // keep worker-local things
        return nil
    })

    cursor, err := p.Go(ctx) // start all workers in 8 goroutines and get back result's cursor
    if err != nil {
        return err
    }

    // submit values (producer side)
    go func() {
        p.Submit("something")
        p.Submit("something else")
        p.Close() // indicates completion of all inputs
    }()

    var v any
    for cursor.Next(ctx, &v) {
        log.Print(v)  // print value
    }

    if cursor.Err() != nil { // error happened
        return cursor.Err()
    }

    // metrics the same as for flow
    metrics := p.Metrics()
    log.Print(metrics.Get("counter"))

Instead of the Next loop everything can be read from the cursor at once:

    res, err := cursor.All(ctx)