func (p *Pipeline) Context() context.Context {
	return p.ctx
}
Context returns this pipeline's context.
package pipeline
import (
	"context"
	"runtime"
	"sync"
)
type Node interface {
	TryMerge(node Node) (merged bool)
	Begin(p *Pipeline, index int, dataSize *int) (keep bool)
	Feed(p *Pipeline, index int, seqNo int, data interface{})
	End()
}
type Pipeline struct {
	mutex        sync.RWMutex
	err          error
	ctx          context.Context
	cancel       context.CancelFunc
	source       Source
	nodes        []Node
	nofBatches   int
	batchInc     int
	maxBatchSize int
}
func (p *Pipeline) Err() (err error) {
	p.mutex.RLock()
	err = p.err
	p.mutex.RUnlock()
	return err
}
func (p *Pipeline) SetErr(err error) bool {
	p.mutex.Lock()
	if p.err == nil {
		p.err = err
		p.mutex.Unlock()
		p.cancel()
		return true
	}
	p.mutex.Unlock()
	return false
}