Trying out different concurrency things

transport/transport.go

@@ -30,25 +30,65 @@ func init() {
 	prometheus.MustRegister(MetricFlowMessageCount)
 }
 
+type TransportDispatchMethod int
+
+const (
+	TransportDispatchLinear     TransportDispatchMethod = iota
+	TransportDispatchWorkerPool TransportDispatchMethod = iota
+	TransportDispatchGoroutine  TransportDispatchMethod = iota
+)
+
 type Transport struct {
-	Destinations []destination.Destination
-	Enrichers    []enricher.Enricher
+	Destinations   []destination.Destination
+	Enrichers      []enricher.Enricher
+	workerPool     *WorkerPool[*goflowpb.FlowMessage]
+	DispatchMethod TransportDispatchMethod
+}
+
+func NewTransport(dispatchMethod TransportDispatchMethod, destinations []destination.Destination, enrichers []enricher.Enricher) *Transport {
+	t := &Transport{
+		Destinations:   destinations,
+		Enrichers:      enrichers,
+		DispatchMethod: dispatchMethod,
+	}
+	if t.DispatchMethod == TransportDispatchWorkerPool {
+		t.workerPool = NewWorkerPool(100, t.messageWorkerPublish)
+		t.workerPool.Start()
+	}
+	return t
 }
 
 func (s *Transport) Publish(fmsgs []*goflowpb.FlowMessage) {
 	MetricFlowMessageBatchCount.Inc()
-	for _, fmsg := range fmsgs {
-		MetricFlowMessageCount.Inc()
-		ffmsg := s.FormatFlowMessage(fmsg)
-		for _, enricher := range s.Enrichers {
-			ffmsg = enricher.Process(ffmsg)
+	switch s.DispatchMethod {
+	case TransportDispatchLinear:
+		for _, fmsg := range fmsgs {
+			s.messageWorkerPublish(fmsg)
+		}
+	case TransportDispatchWorkerPool:
+		for _, fmsg := range fmsgs {
+			s.workerPool.Push(fmsg)
 		}
-		for _, destination := range s.Destinations {
-			destination.Publish(ffmsg)
+	case TransportDispatchGoroutine:
+		for i := range fmsgs {
+			go func(fmesg *goflowpb.FlowMessage) {
+				s.messageWorkerPublish(fmesg)
+			}(fmsgs[i])
 		}
 	}
 }
 
+func (s *Transport) messageWorkerPublish(fmsg *goflowpb.FlowMessage) {
+	MetricFlowMessageCount.Inc()
+	ffmsg := s.FormatFlowMessage(fmsg)
+	for _, enricher := range s.Enrichers {
+		ffmsg = enricher.Process(ffmsg)
+	}
+	for _, destination := range s.Destinations {
+		destination.Publish(ffmsg)
+	}
+}
+
 func (s *Transport) FormatFlowMessage(fmsg *goflowpb.FlowMessage) map[string]interface{} {
 	msg := make(map[string]interface{})
 

transport/worker_pool.go

@@ -0,0 +1,39 @@
+package transport
+
+import (
+	"runtime"
+	"sync"
+)
+
+type WorkerPool[V any] struct {
+	WorkerCount    int
+	MessageHandler func(V)
+	MessageChannel chan V
+	wg             sync.WaitGroup
+}
+
+func NewWorkerPool[V any](messageBuffer int, handler func(V)) *WorkerPool[V] {
+	workerCount := runtime.NumCPU()
+	messageChannel := make(chan V, workerCount*messageBuffer)
+	return &WorkerPool[V]{
+		WorkerCount:    workerCount,
+		MessageHandler: handler,
+		MessageChannel: messageChannel,
+	}
+}
+
+func (wp *WorkerPool[V]) Start() {
+	for i := 0; i < wp.WorkerCount; i++ {
+		wp.wg.Add(1)
+		go func(i int) {
+			defer wp.wg.Done()
+			for msg := range wp.MessageChannel {
+				wp.MessageHandler(msg)
+			}
+		}(i)
+	}
+}
+
+func (wp *WorkerPool[V]) Push(message V) {
+	wp.MessageChannel <- message
+}