Refactor processor pipeline dispatch

.goreleaser.yml

@@ -49,6 +49,7 @@ dockers:
       - "v{{ .Major }}.{{ .Minor }}"
       - latest
     extra_files:
+      - .git
       - cmd
       - config
       - Gopkg.lock

config/test/smoke_in.yaml

@@ -53,4 +53,4 @@ output:
         topic: benthos_topic
         client_id: benthos_output
 logger:
-  log_level: INFO
+  level: INFO

config/test/smoke_out.yaml

@@ -61,4 +61,4 @@ input:
 output:
   type: http_server
 logger:
-  log_level: INFO
+  level: INFO

lib/pipeline/pool.go

@@ -97,8 +97,14 @@ func (p *Pool) loop() {
 				}
 			}()
 			for {
-				t, open := <-w.TransactionChan()
-				if !open {
+				var t types.Transaction
+				var open bool
+				select {
+				case t, open = <-w.TransactionChan():
+					if !open {
+						return
+					}
+				case <-p.closeChan:
 					return
 				}
 				select {

lib/pipeline/processor.go

@@ -38,7 +38,8 @@ import (
 // The processor will read from a source, perform some processing, and then
 // either propagate a new message or drop it.
 type Processor struct {
-	running int32
+	running     int32
+	dispatchers int32
 
 	log   log.Modular
 	stats metrics.Type
@@ -50,6 +51,9 @@ type Processor struct {
 
 	messagesIn <-chan types.Transaction
 
+	mSndSucc metrics.StatCounter
+	mSndErr  metrics.StatCounter
+
 	closeChan chan struct{}
 	closed    chan struct{}
 }
@@ -62,9 +66,12 @@ func NewProcessor(
 ) *Processor {
 	return &Processor{
 		running:       1,
+		dispatchers:   1,
 		msgProcessors: msgProcessors,
 		log:           log.NewModule(".pipeline.processor"),
 		stats:         stats,
+		mSndSucc:      stats.GetCounter("pipeline.processor.send.success"),
+		mSndErr:       stats.GetCounter("pipeline.processor.send.error"),
 		messagesOut:   make(chan types.Transaction),
 		responsesIn:   make(chan types.Response),
 		closeChan:     make(chan struct{}),
@@ -77,21 +84,17 @@ func NewProcessor(
 // loop is the processing loop of this pipeline.
 func (p *Processor) loop() {
 	defer func() {
-		atomic.StoreInt32(&p.running, 0)
-
-		close(p.messagesOut)
-		close(p.closed)
+		if atomic.AddInt32(&p.dispatchers, -1) == 0 {
+			close(p.messagesOut)
+			close(p.closed)
+		}
 	}()
 
 	var (
 		mProcCount   = p.stats.GetCounter("pipeline.processor.count")
 		mProcDropped = p.stats.GetCounter("pipeline.processor.dropped")
-		mSndSucc     = p.stats.GetCounter("pipeline.processor.send.success")
-		mSndErr      = p.stats.GetCounter("pipeline.processor.send.error")
 	)
 
-	throt := throttle.New(throttle.OptCloseChan(p.closeChan))
-
 	var open bool
 	for atomic.LoadInt32(&p.running) == 1 {
 		var tran types.Transaction
@@ -127,72 +130,142 @@ func (p *Processor) loop() {
 			continue
 		}
 
-		var skipAcks int64
-		sendMsg := func(m types.Message) {
-			resChan := make(chan types.Response)
-			transac := types.NewTransaction(m, resChan)
+		if len(resultMsgs) > 1 {
+			p.dispatchMessages(resultMsgs, tran.ResponseChan)
+		} else {
+			p.dispatchMessage(resultMsgs[0], tran.ResponseChan)
+		}
+	}
+}
 
-			for {
-				select {
-				case p.messagesOut <- transac:
-				case <-p.closeChan:
-					return
-				}
+// dispatchMessage attempts to send a single message result of processors over
+// the shared messages channel. This send is retried until success.
+func (p *Processor) dispatchMessage(m types.Message, ogResChan chan<- types.Response) {
+	resChan := make(chan types.Response)
+	transac := types.NewTransaction(m, resChan)
 
-				var res types.Response
-				var open bool
-				select {
-				case res, open = <-resChan:
-					if !open {
-						return
-					}
-				case <-p.closeChan:
-					return
-				}
+	var res types.Response
 
-				if skipAck := res.SkipAck(); res.Error() == nil || skipAck {
-					if skipAck {
-						atomic.AddInt64(&skipAcks, 1)
-					}
-					mSndSucc.Incr(1)
-					return
-				}
-				mSndErr.Incr(1)
-				if !throt.Retry() {
+	select {
+	case p.messagesOut <- transac:
+	case <-p.closeChan:
+		return
+	}
+
+	atomic.AddInt32(&p.dispatchers, 1)
+	go func() {
+		defer func() {
+			if atomic.AddInt32(&p.dispatchers, -1) == 0 {
+				close(p.messagesOut)
+				close(p.closed)
+			}
+		}()
+
+		throt := throttle.New(throttle.OptCloseChan(p.closeChan))
+
+	sendLoop:
+		for {
+			var open bool
+			select {
+			case res, open = <-resChan:
+				if !open {
 					return
 				}
+			case <-p.closeChan:
+				return
 			}
-		}
 
-		if len(resultMsgs) > 1 {
-			wg := sync.WaitGroup{}
-			wg.Add(len(resultMsgs))
-
-			for _, msg := range resultMsgs {
-				go func(m types.Message) {
-					defer wg.Done()
-					sendMsg(m)
-				}(msg)
+			if res.Error() == nil {
+				p.mSndSucc.Incr(1)
+				break sendLoop
 			}
 
-			wg.Wait()
-		} else {
-			sendMsg(resultMsgs[0])
-		}
-		throt.Reset()
+			p.mSndErr.Incr(1)
+			if !throt.Retry() {
+				return
+			}
 
-		var res types.Response
-		if skipAcks == int64(len(resultMsgs)) {
-			res = response.NewUnack()
-		} else {
-			res = response.NewAck()
+			select {
+			case p.messagesOut <- transac:
+			case <-p.closeChan:
+				return
+			}
 		}
 
 		select {
-		case tran.ResponseChan <- res:
+		case ogResChan <- res:
 		case <-p.closeChan:
 			return
 		}
+	}()
+}
+
+// dispatchMessages attempts to send a multiple messages results of processors
+// over the shared messages channel. This send is retried until success.
+func (p *Processor) dispatchMessages(msgs []types.Message, ogResChan chan<- types.Response) {
+	throt := throttle.New(throttle.OptCloseChan(p.closeChan))
+
+	var skipAcks int64
+	sendMsg := func(m types.Message) {
+		resChan := make(chan types.Response)
+		transac := types.NewTransaction(m, resChan)
+
+		for {
+			select {
+			case p.messagesOut <- transac:
+			case <-p.closeChan:
+				return
+			}
+
+			var res types.Response
+			var open bool
+			select {
+			case res, open = <-resChan:
+				if !open {
+					return
+				}
+			case <-p.closeChan:
+				return
+			}
+
+			if skipAck := res.SkipAck(); res.Error() == nil || skipAck {
+				if skipAck {
+					atomic.AddInt64(&skipAcks, 1)
+				}
+				p.mSndSucc.Incr(1)
+				return
+			}
+			p.mSndErr.Incr(1)
+			if !throt.Retry() {
+				return
+			}
+		}
+	}
+
+	wg := sync.WaitGroup{}
+	wg.Add(len(msgs))
+
+	for _, msg := range msgs {
+		go func(m types.Message) {
+			defer wg.Done()
+			sendMsg(m)
+		}(msg)
+	}
+
+	wg.Wait()
+	throt.Reset()
+
+	var res types.Response
+	if skipAcks == int64(len(msgs)) {
+		res = response.NewUnack()
+	} else {
+		res = response.NewAck()
+	}
+
+	select {
+	case ogResChan <- res:
+	case <-p.closeChan:
+		return
 	}
 }
 

resources/docker/benchmark/benthos.yaml

@@ -36,7 +36,7 @@ output:
 
 logger:
   prefix: service
-  log_level: INFO
+  level: INFO
 
 metrics:
   type: prometheus

resources/docker/benchmark/extract_data.yaml

@@ -14,4 +14,4 @@ output:
 
 logger:
   prefix: service
-  log_level: INFO
+  level: INFO

resources/docker/benchmark/inject_data.yaml

@@ -36,4 +36,4 @@ output:
 
 logger:
   prefix: service
-  log_level: INFO
+  level: INFO

resources/docker/profiling/benthos.yaml

@@ -39,7 +39,7 @@ output:
 
 logger:
   prefix: service
-  log_level: INFO
+  level: INFO
 
 metrics:
   type: prometheus