Roomserver per-room input parallelisation (Postgres) (#1289)

* Per-room input mutex

* GetMembership should use transaction when assigning state key NID

* Actually use writer transactions rather than ignoring them

* Limit per-room mutexes to Postgres

* Flip the check in InputRoomEvents

roomserver/internal/api.go

@@ -20,7 +20,7 @@ type RoomserverInternalAPI struct {
 	ServerName           gomatrixserverlib.ServerName
 	KeyRing              gomatrixserverlib.JSONVerifier
 	FedClient            *gomatrixserverlib.FederationClient
-	OutputRoomEventTopic string     // Kafka topic for new output room events
-	mutex                sync.Mutex // Protects calls to processRoomEvent
+	OutputRoomEventTopic string   // Kafka topic for new output room events
+	mutexes              sync.Map // room ID -> *sync.Mutex, protects calls to processRoomEvent
 	fsAPI                fsAPI.FederationSenderInternalAPI
 }

roomserver/internal/input.go

@@ -18,6 +18,7 @@ package internal
 import (
 	"context"
 	"encoding/json"
+	"sync"
 
 	"github.com/Shopify/sarama"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -71,13 +72,18 @@ func (r *RoomserverInternalAPI) InputRoomEvents(
 	request *api.InputRoomEventsRequest,
 	response *api.InputRoomEventsResponse,
 ) (err error) {
-	// We lock as processRoomEvent can only be called once at a time
-	r.mutex.Lock()
-	defer r.mutex.Unlock()
-	for i := range request.InputRoomEvents {
+	for i, e := range request.InputRoomEvents {
+		roomID := "global"
+		if r.DB.SupportsConcurrentRoomInputs() {
+			roomID = e.Event.RoomID()
+		}
+		mutex, _ := r.mutexes.LoadOrStore(roomID, &sync.Mutex{})
+		mutex.(*sync.Mutex).Lock()
 		if response.EventID, err = r.processRoomEvent(ctx, request.InputRoomEvents[i]); err != nil {
+			mutex.(*sync.Mutex).Unlock()
 			return err
 		}
+		mutex.(*sync.Mutex).Unlock()
 	}
 	return nil
 }