[fanout] Organize fan-out pipeline

messaging/fanout.go

@@ -8,45 +8,39 @@ import (
 	"sync/atomic"
 )
 
-const (
-	MaxWorkers   = 32
-	MaxQueueSize = 128
+var (
+	log, _ = zap.NewDevelopment()
 )
 
-var (
-	log, _         = zap.NewDevelopment()
+const (
+	MaxWorkers      = 16
+	MaxQueueSize    = 512
+	MasterQueueSize = MaxQueueSize * MaxWorkers
 )
 
 type Pipeline struct {
-	workers []*worker
+	workers map[int]*worker
 	chain   chan interface{}
-	running *uint32
 }
 
 func (p *Pipeline) Start() {
-	distributeToChannels := func(ch chan interface{}, cs []*worker) {
-		writer := cs[0] //first worker must stream as default
+	go func(pipe *Pipeline) {
 		for {
-			for _, c := range cs {
-				expectationWorkers := uint32(len(ch)/(MaxQueueSize/MaxWorkers)) + 1
-				runningWorker := atomic.LoadUint32(p.running)
-				if c.index > runningWorker+1 || c.index > expectationWorkers {
+			expectationWorkers := len(pipe.chain)
+			if expectationWorkers > MaxWorkers {
+				expectationWorkers = expectationWorkers % MaxWorkers
+			}
+			for _, c := range pipe.workers {
+				if expectationWorkers < int(c.index) {
 					break
 				}
 				select {
-				case val := <-ch:
-					writer = c
-					if c.debug {
-						log.Info("Worker receiving", zap.Any("index", writer.index), zap.Any("running", runningWorker), zap.Any("no# workers", expectationWorkers))
-					}
-					go writer.stream(val)
+				case val := <-pipe.chain:
+					go c.stream(val)
 				}
 			}
 		}
-	}
-}
-
-go distributeToChannels(p.chain, p.workers)
+	}(p)
 }
 
 func (p *Pipeline) Dispatch(msg interface{}) {
@@ -56,23 +50,19 @@ func (p *Pipeline) Dispatch(msg interface{}) {
 type DispatcherBuilder func() Dispatcher
 
 func NewPipeline(d DispatcherBuilder, idle uint32, debug bool) *Pipeline {
-	wk := make([]*worker, 0, MaxWorkers)
-	ch := make(chan interface{}, 4096)
-	pipe := &Pipeline{chain: ch, running: new(uint32)}
+	ch := make(chan interface{}, MasterQueueSize)
+	wk := make(map[int]*worker)
 	for i := 0; i < MaxWorkers; i++ {
-		wk = append(wk,
-			&worker{
-				index:      uint32(i + 1),
-				chain:      make(chan interface{}, MaxQueueSize),
-				mutex:      new(sync.Mutex),
-				debug:      debug,
-				idle:       idle,
-				Dispatcher: d(),
-				broker:     pipe.running,
-			})
+		wk[i] = &worker{
+			index:      uint32(i + 1),
+			chain:      make(chan interface{}, MaxQueueSize),
+			mutex:      new(sync.Mutex),
+			debug:      debug,
+			idle:       idle,
+			Dispatcher: d(),
+		}
 	}
-	pipe.workers = wk
-	return pipe
+	return &Pipeline{workers: wk, chain: ch}
 }
 
 type Dispatcher interface {
@@ -88,7 +78,6 @@ type worker struct {
 	chain   chan interface{}
 	debug   bool
 	idle    uint32
-	broker  *uint32
 	Dispatcher
 }
 
@@ -97,14 +86,7 @@ func (c *worker) stream(val interface{}) {
 	if !c.running {
 		c.mutex.Lock()
 		c.running = true
-		atomic.AddUint32(c.broker, 1)
-		defer atomic.AddUint32(c.broker, ^uint32(1 - 1))
 		ctx, cancel := context.WithCancel(context.Background())
-		err := c.Before(ctx)
-
-		if err != nil {
-			log.Error("can not start worker", zap.Error(err))
-		}
 		defer func(w *worker, cancel context.CancelFunc) {
 			if w.debug {
 				log.Info("Worker leaving", zap.Any("index", w.index), zap.Any("idle", w.idle))
@@ -117,6 +99,11 @@ func (c *worker) stream(val interface{}) {
 			w.mutex.Unlock()
 			w.running = false
 		}(c, cancel)
+		err := c.Before(ctx)
+
+		if err != nil {
+			log.Error("can not start worker", zap.Error(err))
+		}
 		var idle uint32 = 0
 		for {
 			select {
@@ -140,9 +127,6 @@ func (c *worker) stream(val interface{}) {
 					if i > c.idle {
 						return
 					}
-					if c.debug {
-						log.Info("Idle", zap.Any("worker index", c.index), zap.Any("idle", idle))
-					}
 				}
 			}
 		}