slow sql fixed on assettocorsa

Author: jacobjurek

assettocorsa/database/batch.go

@@ -0,0 +1,73 @@
+package database
+
+import (
+	"ac/utils"
+	"sync"
+	"time"
+)
+
+type BatchInserter[T any] struct {
+	mu        sync.Mutex
+	buffer    []T
+	maxSize   int
+	ticker    *time.Ticker
+	stopCh    chan struct{}
+	tableName string
+}
+
+func NewBatchInserter[T any](tableName string, maxSize int, flushInterval time.Duration) *BatchInserter[T] {
+	b := &BatchInserter[T]{
+		buffer:    make([]T, 0, maxSize),
+		maxSize:   maxSize,
+		ticker:    time.NewTicker(flushInterval),
+		stopCh:    make(chan struct{}),
+		tableName: tableName,
+	}
+	go b.run()
+	return b
+}
+
+func (b *BatchInserter[T]) run() {
+	for {
+		select {
+		case <-b.ticker.C:
+			b.Flush()
+		case <-b.stopCh:
+			return
+		}
+	}
+}
+
+func (b *BatchInserter[T]) Add(item T) {
+	b.mu.Lock()
+	b.buffer = append(b.buffer, item)
+	shouldFlush := len(b.buffer) >= b.maxSize
+	b.mu.Unlock()
+
+	if shouldFlush {
+		b.Flush()
+	}
+}
+
+func (b *BatchInserter[T]) Flush() {
+	b.mu.Lock()
+	if len(b.buffer) == 0 {
+		b.mu.Unlock()
+		return
+	}
+	batch := b.buffer
+	b.buffer = make([]T, 0, b.maxSize)
+	b.mu.Unlock()
+
+	if result := DB.CreateInBatches(&batch, len(batch)); result.Error != nil {
+		utils.SugarLogger.Errorf("[DB] Batch insert failed for %s: %s", b.tableName, result.Error)
+		return
+	}
+	utils.SugarLogger.Infof("[DB] Flushed %d %s rows", len(batch), b.tableName)
+}
+
+func (b *BatchInserter[T]) Stop() {
+	b.ticker.Stop()
+	close(b.stopCh)
+	b.Flush()
+}

assettocorsa/main.go

@@ -17,6 +17,10 @@ func main() {
 
 	service.RegisterRincon()
 	database.InitializeDB()
+	service.InitSignalBatch()
+	service.InitPingBatch()
+	defer service.StopSignalBatch()
+	defer service.StopPingBatch()
 	mqtt.InitializeMQTT()
 	service.SubscribeTopics()
 

assettocorsa/service/ping.go

@@ -11,6 +11,16 @@ import (
 	"github.com/gaucho-racing/mapache-go"
 )
 
+var pingBatch *database.BatchInserter[mapache.Ping]
+
+func InitPingBatch() {
+	pingBatch = database.NewBatchInserter[mapache.Ping]("pings", 5000, 1*time.Second)
+}
+
+func StopPingBatch() {
+	pingBatch.Stop()
+}
+
 func HandlePing(vehicleID string, nodeID string, payload []byte) {
 	utils.SugarLogger.Infof("[MQ] Received ping from ac/%s/%s", vehicleID, nodeID)
 	ping := binary.BigEndian.Uint64(payload[:8])
@@ -53,15 +63,6 @@ func GetPing(vehicleID string, micros int) mapache.Ping {
 }
 
 func CreatePing(ping mapache.Ping) error {
-	result := database.DB.Create(&ping)
-	if result.Error != nil {
-		return result.Error
-	}
-	utils.SugarLogger.Infow("[DB] New ping created",
-		"vehicle_id", ping.VehicleID,
-		"ping", ping.Ping,
-		"pong", ping.Pong,
-		"latency", ping.Latency,
-	)
+	pingBatch.Add(ping)
 	return nil
 }

assettocorsa/service/signal.go

@@ -2,12 +2,22 @@ package service
 
 import (
 	"ac/database"
-	"ac/utils"
 	"fmt"
+	"time"
 
 	"github.com/gaucho-racing/mapache-go"
 )
 
+var signalBatch *database.BatchInserter[mapache.Signal]
+
+func InitSignalBatch() {
+	signalBatch = database.NewBatchInserter[mapache.Signal]("signals", 5000, 1*time.Second)
+}
+
+func StopSignalBatch() {
+	signalBatch.Stop()
+}
+
 // signalCallbacks is a list of functions that will be called when a signal is created or updated
 var signalCallbacks = []func(signal mapache.Signal){}
 
@@ -39,14 +49,7 @@ func CreateSignal(signal mapache.Signal) error {
 	if signal.Name == "" {
 		return fmt.Errorf("signal name cannot be empty")
 	}
-	if result := database.DB.Create(&signal); result.Error != nil {
-		return result.Error
-	}
-	utils.SugarLogger.Infow("[DB] Signal inserted",
-		"timestamp", signal.Timestamp,
-		"vehicle_id", signal.VehicleID,
-		"name", signal.Name,
-	)
+	signalBatch.Add(signal)
 	go signalNotify(signal)
 	return nil
 }