func (p *Pipeline) SyncAdd(c context.Context, key string, value interface{}) {
	ch, msg := p.add(c, key, value)
	ch <- msg
}
SyncAdd sync add a value to channal, channel shard in split method
package pipeline
import (
	"context"
	"errors"
	"strconv"
	"sync"
	"time"
	"github.com/mapgoo-lab/atreus/pkg/net/metadata"
	"github.com/mapgoo-lab/atreus/pkg/stat/metric"
	xtime "github.com/mapgoo-lab/atreus/pkg/time"
)
var ErrFull = errors.New("channel full")
const _metricNamespace = "sync"
const _metricSubSystem = "pipeline"
var (
	_metricCount = metric.NewCounterVec(&metric.CounterVecOpts{
		Namespace: _metricNamespace,
		Subsystem: _metricSubSystem,
		Name:      "process_count",
		Help:      "process count",
		Labels:    []string{"name", "chan"},
	})
	_metricChanLen = metric.NewGaugeVec(&metric.GaugeVecOpts{
		Namespace: _metricNamespace,
		Subsystem: _metricSubSystem,
		Name:      "chan_len",
		Help:      "channel length",
		Labels:    []string{"name", "chan"},
	})
)
type message struct {
	key   string
	value interface{}
}
type Pipeline struct {
	Do          func(c context.Context, index int, values map[string][]interface{})
	Split       func(key string) int
	chans       []chan *message
	mirrorChans []chan *message
	config      *Config
	wait        sync.WaitGroup
	name        string
}
type Config struct {
	MaxSize int
	Interval xtime.Duration
	Buffer int
	Worker int
	Name string
}
func (c *Config) fix() {
	if c.MaxSize <= 0 {
		c.MaxSize = 1000
	}
	if c.Interval <= 0 {
		c.Interval = xtime.Duration(time.Second)
	}
	if c.Buffer <= 0 {
		c.Buffer = 1000
	}
	if c.Worker <= 0 {
		c.Worker = 10
	}
	if c.Name == "" {
		c.Name = "anonymous"
	}
}
func NewPipeline(config *Config) (res *Pipeline) {
	if config == nil {
		config = &Config{}
	}
	config.fix()
	res = &Pipeline{
		chans:       make([]chan *message, config.Worker),
		mirrorChans: make([]chan *message, config.Worker),
		config:      config,
		name:        config.Name,
	}
	for i := 0; i < config.Worker; i++ {
		res.chans[i] = make(chan *message, config.Buffer)
		res.mirrorChans[i] = make(chan *message, config.Buffer)
	}
	return
}
func (p *Pipeline) Start() {
	if p.Do == nil {
		panic("pipeline: do func is nil")
	}
	if p.Split == nil {
		panic("pipeline: split func is nil")
	}
	var mirror bool
	p.wait.Add(len(p.chans) + len(p.mirrorChans))
	for i, ch := range p.chans {
		go p.mergeproc(mirror, i, ch)
	}
	mirror = true
	for i, ch := range p.mirrorChans {
		go p.mergeproc(mirror, i, ch)
	}
}