Add a per-room mutex to federationapi when processing transactions (#1810)

* Add a per-room mutex to federationapi when processing transactions

This has numerous benefits:
 - Prevents us doing lots of state resolutions in busy rooms. Previously, room forks would always result
   in a state resolution being performed immediately, without checking if we were already doing this in
   a different transaction. Now they will queue up, resulting in fewer calls to `/state_ids`, `/g_m_e`, etc.
 - Prevents memory usage from growing too large as a result and potentially OOMing.

And costs:
 - High traffic rooms will be slightly slower due to head-of-line blocking from other servers,
   though this has always been an issue as roomserver has a per-room mutex already.

* Fix unit tests

* Correct mutex lock ordering

1 files changed, 38 insertions, 0 deletions

diff --git a/internal/mutex.go b/internal/mutex.go
new file mode 100644
index 00000000..3d36cdac
--- /dev/null
+++ b/internal/mutex.go
@@ -0,0 +1,38 @@
+package internal
+
+import "sync"
+
+type MutexByRoom struct {
+	mu       *sync.Mutex // protects the map
+	roomToMu map[string]*sync.Mutex
+}
+
+func NewMutexByRoom() *MutexByRoom {
+	return &MutexByRoom{
+		mu:       &sync.Mutex{},
+		roomToMu: make(map[string]*sync.Mutex),
+	}
+}
+
+func (m *MutexByRoom) Lock(roomID string) {
+	m.mu.Lock()
+	roomMu := m.roomToMu[roomID]
+	if roomMu == nil {
+		roomMu = &sync.Mutex{}
+	}
+	m.roomToMu[roomID] = roomMu
+	m.mu.Unlock()
+	// don't lock inside m.mu else we can deadlock
+	roomMu.Lock()
+}
+
+func (m *MutexByRoom) Unlock(roomID string) {
+	m.mu.Lock()
+	roomMu := m.roomToMu[roomID]
+	if roomMu == nil {
+		panic("MutexByRoom: Unlock before Lock")
+	}
+	m.mu.Unlock()
+
+	roomMu.Unlock()
+}