Parallelize mattermost room loading; mutexes

1 files changed, 31 insertions, 1 deletions

diff --git a/appservice/db.go b/appservice/db.go
index 9c3280c..34fc046 100644
--- a/appservice/db.go
+++ b/appservice/db.go
@@ -2,12 +2,14 @@ package appservice
 
 import (
 	"fmt"
+	"sync"
 
 	"github.com/jinzhu/gorm"
 	_ "github.com/jinzhu/gorm/dialects/mysql"
 	_ "github.com/jinzhu/gorm/dialects/postgres"
 	_ "github.com/jinzhu/gorm/dialects/sqlite"
 	log "github.com/sirupsen/logrus"
+	"golang.org/x/crypto/blake2b"
 
 	"git.deuxfleurs.fr/Deuxfleurs/easybridge/connector"
 	"git.deuxfleurs.fr/Deuxfleurs/easybridge/mxlib"
@@ -84,6 +86,20 @@ type DbPmRoomMap struct {
 	MxRoomID string `gorm:"index:mxroomoid"`
 }
 
+// ---- Simple locking mechanism
+
+var dbLocks [256]sync.Mutex
+
+func dbLockSlot(key string) {
+	slot := blake2b.Sum512([]byte(key))[0]
+	dbLocks[slot].Lock()
+}
+
+func dbUnlockSlot(key string) {
+	slot := blake2b.Sum512([]byte(key))[0]
+	dbLocks[slot].Unlock()
+}
+
 // ----
 
 func dbCacheGet(key string) string {
@@ -101,7 +117,9 @@ func dbCachePut(key string, value string) {
 }
 
 func dbCacheTestAndSet(key string, value string) bool {
-	// TODO make this really an atomic operation
+	dbLockSlot(key)
+	defer dbUnlockSlot(key)
+
 	// True if value was changed, false if was already set
 	if dbCacheGet(key) != value {
 		dbCachePut(key, value)
@@ -111,6 +129,10 @@ func dbCacheTestAndSet(key string, value string) bool {
 }
 
 func dbGetMxRoom(protocol string, roomId connector.RoomID) (string, error) {
+	slot_key := fmt.Sprintf("room: %s / %s", protocol, roomId)
+	dbLockSlot(slot_key)
+	defer dbUnlockSlot(slot_key)
+
 	var room DbRoomMap
 
 	// Check if room exists in our mapping,
@@ -148,6 +170,10 @@ func dbGetMxRoom(protocol string, roomId connector.RoomID) (string, error) {
 }
 
 func dbGetMxPmRoom(protocol string, them connector.UserID, themMxId string, usMxId string, usAccount string) (string, error) {
+	slot_key := fmt.Sprintf("pmroom: %s / %s / %s / %s", protocol, usMxId, usAccount, them)
+	dbLockSlot(slot_key)
+	defer dbUnlockSlot(slot_key)
+
 	var room DbPmRoomMap
 
 	must_create := db.First(&room, DbPmRoomMap{
@@ -186,6 +212,10 @@ func dbGetMxPmRoom(protocol string, them connector.UserID, themMxId string, usMx
 }
 
 func dbGetMxUser(protocol string, userId connector.UserID) (string, error) {
+	slot_key := fmt.Sprintf("user: %s / %s", protocol, userId)
+	dbLockSlot(slot_key)
+	defer dbUnlockSlot(slot_key)
+
 	var user DbUserMap
 
 	must_create := db.First(&user, DbUserMap{