fix: producer not close after shutdown (#355)

* fix: producer close not release memory and monirtor

Author: crimson-gao

producer/io_thread_pool.go

@@ -33,18 +33,14 @@ func (threadPool *IoThreadPool) addTask(batch *ProducerBatch) {
 func (threadPool *IoThreadPool) start(ioWorkerWaitGroup *sync.WaitGroup, ioThreadPoolwait *sync.WaitGroup) {
 	defer ioThreadPoolwait.Done()
 	for task := range threadPool.taskCh {
-		if task == nil {
-			level.Info(threadPool.logger).Log("msg", "All cache tasks in the thread pool have been successfully sent")
-			threadPool.stopped.Store(true)
-			return
-		}
-
 		threadPool.ioworker.startSendTask(ioWorkerWaitGroup)
 		go func(producerBatch *ProducerBatch) {
 			defer threadPool.ioworker.closeSendTask(ioWorkerWaitGroup)
 			threadPool.ioworker.sendToServer(producerBatch)
 		}(task)
 	}
+	level.Info(threadPool.logger).Log("msg", "All cache tasks in the thread pool have been successfully sent")
+	threadPool.stopped.Store(true)
 }
 
 func (threadPool *IoThreadPool) ShutDown() {

producer/monitor.go

@@ -1,6 +1,7 @@
 package producer
 
 import (
+	"sync"
 	"sync/atomic"
 	"time"
 
@@ -23,14 +24,23 @@ type ProducerMetrics struct {
 
 type ProducerMonitor struct {
 	metrics atomic.Value // *ProducerMetrics
+	stopCh  chan struct{}
+	wg      sync.WaitGroup
 }
 
 func newProducerMonitor() *ProducerMonitor {
-	m := &ProducerMonitor{}
+	m := &ProducerMonitor{
+		stopCh: make(chan struct{}),
+	}
 	m.metrics.Store(&ProducerMetrics{})
 	return m
 }
 
+func (m *ProducerMonitor) Stop() {
+	close(m.stopCh)
+	m.wg.Wait()
+}
+
 func (m *ProducerMonitor) recordSuccess(sendBegin time.Time, sendEnd time.Time) {
 	metrics := m.metrics.Load().(*ProducerMetrics)
 	metrics.sendBatch.AddSample(float64(sendEnd.Sub(sendBegin).Microseconds()))
@@ -72,17 +82,24 @@ func (m *ProducerMonitor) getAndResetMetrics() *ProducerMetrics {
 }
 
 func (m *ProducerMonitor) reportThread(reportInterval time.Duration, logger log.Logger) {
+	defer m.wg.Done()
 	ticker := time.NewTicker(reportInterval)
-	for range ticker.C {
-		metrics := m.getAndResetMetrics()
-		level.Info(logger).Log("msg", "report status",
-			"sendBatch", metrics.sendBatch.String(),
-			"retryCount", metrics.retryCount.Load(),
-			"createBatch", metrics.createBatch.Load(),
-			"onSuccess", metrics.onSuccess.String(),
-			"onFail", metrics.onFail.String(),
-			"waitMemory", metrics.waitMemory.String(),
-			"waitMemoryFailCount", metrics.waitMemoryFailCount.Load(),
-		)
+	for {
+		select {
+		case <-ticker.C:
+			metrics := m.getAndResetMetrics()
+			level.Info(logger).Log("msg", "report status",
+				"sendBatch", metrics.sendBatch.String(),
+				"retryCount", metrics.retryCount.Load(),
+				"createBatch", metrics.createBatch.Load(),
+				"onSuccess", metrics.onSuccess.String(),
+				"onFail", metrics.onFail.String(),
+				"waitMemory", metrics.waitMemory.String(),
+				"waitMemoryFailCount", metrics.waitMemoryFailCount.Load(),
+			)
+		case <-m.stopCh:
+			ticker.Stop()
+			return
+		}
 	}
 }

producer/producer.go

@@ -288,12 +288,14 @@ func (producer *Producer) Start() {
 	producer.ioThreadPoolWaitGroup.Add(1)
 	go producer.threadPool.start(producer.ioWorkerWaitGroup, producer.ioThreadPoolWaitGroup)
 	if !producer.producerConfig.DisableRuntimeMetrics {
+		producer.monitor.wg.Add(1)
 		go producer.monitor.reportThread(time.Minute, producer.logger)
 	}
 }
 
 // Limited closing transfer parameter nil, safe closing transfer timeout time, timeout Ms parameter in milliseconds
 func (producer *Producer) Close(timeoutMs int64) error {
+	producer.monitor.Stop()
 	startCloseTime := time.Now()
 	producer.sendCloseProdcerSignal()
 	producer.moverWaitGroup.Wait()
@@ -310,6 +312,7 @@ func (producer *Producer) Close(timeoutMs int64) error {
 }
 
 func (producer *Producer) SafeClose() {
+	producer.monitor.Stop()
 	producer.sendCloseProdcerSignal()
 	producer.moverWaitGroup.Wait()
 	level.Info(producer.logger).Log("msg", "Mover close finish")