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