From cddf081b8fee274f07e704668cb346b50381e8d3 Mon Sep 17 00:00:00 2001 From: Sergey Stepanov Date: Fri, 6 Oct 2023 01:12:27 +0300 Subject: [PATCH] Use locks in router with rooms --- pkg/com/com.go | 1 + pkg/com/map.go | 29 +++++++++++++++++++++++++---- pkg/com/map_test.go | 4 ++-- pkg/coordinator/worker.go | 2 +- pkg/coordinator/workerhandlers.go | 2 +- pkg/worker/coordinatorhandlers.go | 12 ++++++------ pkg/worker/room/room.go | 31 +++++++++++++++++++------------ pkg/worker/worker.go | 6 +++++- 8 files changed, 60 insertions(+), 27 deletions(-) diff --git a/pkg/com/com.go b/pkg/com/com.go index d6cc5ca7..5a738039 100644 --- a/pkg/com/com.go +++ b/pkg/com/com.go @@ -15,6 +15,7 @@ func NewNetMap[K comparable, T NetClient[K]]() NetMap[K, T] { func (m *NetMap[K, T]) Add(client T) bool { return m.Put(client.Id(), client) } func (m *NetMap[K, T]) Remove(client T) { m.Map.Remove(client.Id()) } +func (m *NetMap[K, T]) RemoveL(client T) int { return m.Map.RemoveL(client.Id()) } func (m *NetMap[K, T]) Reset() { m.Map = Map[K, T]{m: make(map[K]T, 10)} } func (m *NetMap[K, T]) RemoveDisconnect(client T) { client.Disconnect(); m.Remove(client) } diff --git a/pkg/com/map.go b/pkg/com/map.go index a144c0e3..6a4df33a 100644 --- a/pkg/com/map.go +++ b/pkg/com/map.go @@ -1,6 +1,9 @@ package com -import "sync" +import ( + "fmt" + "sync" +) // Map defines a concurrent-safe map structure. // Keep in mind that the underlying map structure will grow indefinitely. @@ -9,7 +12,7 @@ type Map[K comparable, V any] struct { mu sync.Mutex } -func (m *Map[K, _]) Has(key K) bool { _, ok := m.Find(key); return ok } +func (m *Map[K, _]) Has(key K) bool { _, ok := m.Contains(key); return ok } func (m *Map[_, _]) Len() int { m.mu.Lock(); defer m.mu.Unlock(); return len(m.m) } func (m *Map[K, V]) Pop(key K) V { m.mu.Lock() @@ -26,9 +29,22 @@ func (m *Map[K, V]) Put(key K, v V) bool { return ok } func (m *Map[K, _]) Remove(key K) { m.mu.Lock(); delete(m.m, key); m.mu.Unlock() } +func (m *Map[K, _]) RemoveL(key K) int { + m.mu.Lock() + delete(m.m, key) + k := len(m.m) + m.mu.Unlock() + return k +} +func (m *Map[K, V]) String() string { + m.mu.Lock() + s := fmt.Sprintf("%v", m.m) + m.mu.Unlock() + return s +} -// Find returns the first value found and a boolean flag if its found or not. -func (m *Map[K, V]) Find(key K) (v V, ok bool) { +// Contains returns the first value found and a boolean flag if its found or not. +func (m *Map[K, V]) Contains(key K) (v V, ok bool) { m.mu.Lock() defer m.mu.Unlock() if vv, ok := m.m[key]; ok { @@ -37,6 +53,11 @@ func (m *Map[K, V]) Find(key K) (v V, ok bool) { return v, false } +func (m *Map[K, V]) Find(key K) V { + v, _ := m.Contains(key) + return v +} + // FindBy searches the first key-value with the provided predicate function. func (m *Map[K, V]) FindBy(fn func(v V) bool) (v V, ok bool) { m.mu.Lock() diff --git a/pkg/com/map_test.go b/pkg/com/map_test.go index fb2b19f7..8de5c306 100644 --- a/pkg/com/map_test.go +++ b/pkg/com/map_test.go @@ -17,11 +17,11 @@ func TestMap_Base(t *testing.T) { if !m.Has(k) { t.Errorf("should have the key %v, %v", k, m.m) } - v, ok := m.Find(k) + v, ok := m.Contains(k) if v != 0 && !ok { t.Errorf("should have the key %v and ok, %v %v", k, ok, m.m) } - _, ok = m.Find(k + 1) + _, ok = m.Contains(k + 1) if ok { t.Errorf("should not find anything, %v %v", ok, m.m) } diff --git a/pkg/coordinator/worker.go b/pkg/coordinator/worker.go index dd0221fb..32b8f147 100644 --- a/pkg/coordinator/worker.go +++ b/pkg/coordinator/worker.go @@ -29,7 +29,7 @@ type RegionalClient interface { } type HasUserRegistry interface { - Find(com.Uid) (*User, bool) + Find(com.Uid) *User } func NewWorker(sock *com.Connection, handshake api.ConnectionRequest[com.Uid], log *logger.Logger) *Worker { diff --git a/pkg/coordinator/workerhandlers.go b/pkg/coordinator/workerhandlers.go index 97a7839d..6f50d126 100644 --- a/pkg/coordinator/workerhandlers.go +++ b/pkg/coordinator/workerhandlers.go @@ -14,7 +14,7 @@ func (w *Worker) HandleCloseRoom(rq api.CloseRoomRequest) { } func (w *Worker) HandleIceCandidate(rq api.WebrtcIceCandidateRequest[com.Uid], users HasUserRegistry) error { - if usr, ok := users.Find(rq.Id); ok { + if usr := users.Find(rq.Id); usr != nil { usr.SendWebrtcIceCandidate(rq.Candidate) } else { w.log.Warn().Str("id", rq.Id.String()).Msg("unknown session") diff --git a/pkg/worker/coordinatorhandlers.go b/pkg/worker/coordinatorhandlers.go index 159f3dd3..4d3548db 100644 --- a/pkg/worker/coordinatorhandlers.go +++ b/pkg/worker/coordinatorhandlers.go @@ -49,8 +49,10 @@ func (c *coordinator) HandleWebrtcInit(rq api.WebrtcInitRequest[com.Uid], w *Wor } user := room.NewGameSession(rq.Id, peer) // use user uid from the coordinator - w.router.AddUser(user) c.log.Info().Msgf("Peer connection: %s", user.Id()) + c.log.Debug().Msgf("Users before add: %v", w.router.Users()) + w.router.AddUser(user) + c.log.Debug().Msgf("Users after add: %v", w.router.Users()) return api.Out{Payload: sdp} } @@ -118,11 +120,7 @@ func (c *coordinator) HandleGameStart(rq api.StartGameRequest[com.Uid], w *Worke // make the room r = room.NewRoom[*room.GameSession](uid, app, w.router.Users(), m) - r.HandleClose = func() { - w.router.Close() - c.CloseRoom(uid) - w.log.Debug().Msgf("Closed room %v", uid) - } + r.HandleClose = func() { c.CloseRoom(uid) } w.router.SetRoom(r) c.log.Info().Str("room", r.Id()).Str("game", game.Name).Msg("New room") @@ -148,8 +146,10 @@ func (c *coordinator) HandleTerminateSession(rq api.TerminateSessionRequest[com. // HandleQuitGame handles cases when a user manually exits the game. func (c *coordinator) HandleQuitGame(rq api.GameQuitRequest[com.Uid], w *Worker) { + w.log.Debug().Msgf("Users before remove: %v", w.router.Users()) if user := w.router.FindUser(rq.Id); user != nil { w.router.Remove(user) + w.log.Debug().Msgf("Users after remove: %v", w.router.Users()) } } diff --git a/pkg/worker/room/room.go b/pkg/worker/room/room.go index 0acbcf02..a7cfe123 100644 --- a/pkg/worker/room/room.go +++ b/pkg/worker/room/room.go @@ -1,6 +1,10 @@ package room -import "github.com/giongto35/cloud-game/v3/pkg/worker/caged/app" +import ( + "sync" + + "github.com/giongto35/cloud-game/v3/pkg/worker/caged/app" +) type MediaPipe interface { // Destroy frees all allocated resources. @@ -19,10 +23,9 @@ type MediaPipe interface { type SessionManager[T Session] interface { Add(T) bool - Find(string) (T, bool) + Find(string) T ForEach(func(T)) - Len() int - Remove(T) + RemoveL(T) int // Reset used for proper cleanup of the resources if needed. Reset() } @@ -93,32 +96,36 @@ func (r *Room[T]) Close() { type Router[T Session] struct { room *Room[T] users SessionManager[T] + mu sync.Mutex } func (r *Router[T]) AddUser(user T) { r.users.Add(user) } + func (r *Router[T]) Close() { + r.mu.Lock() if r.room != nil { r.room.Close() r.room = nil } + r.mu.Unlock() } + func (r *Router[T]) FindRoom(id string) *Room[T] { + r.mu.Lock() + defer r.mu.Unlock() if r.room != nil && r.room.Id() == id { return r.room } return nil } -func (r *Router[T]) FindUser(uid Uid) T { sess, _ := r.users.Find(uid.Id()); return sess } + func (r *Router[T]) Remove(user T) { - r.users.Remove(user) - if r.users.Len() == 0 { - if r.room != nil { - r.room.Close() - } - r.users.Reset() + if left := r.users.RemoveL(user); left == 0 { + r.Close() } } -func (r *Router[T]) SetRoom(room *Room[T]) { r.room = room } +func (r *Router[T]) FindUser(uid Uid) T { return r.users.Find(uid.Id()) } +func (r *Router[T]) SetRoom(room *Room[T]) { r.mu.Lock(); r.room = room; r.mu.Unlock() } func (r *Router[T]) Users() SessionManager[T] { return r.users } type AppSession struct { diff --git a/pkg/worker/worker.go b/pkg/worker/worker.go index 56fc8e03..c74375db 100644 --- a/pkg/worker/worker.go +++ b/pkg/worker/worker.go @@ -28,7 +28,11 @@ type Worker struct { storage cloud.Storage } -func (w *Worker) Reset() { w.router.Close() } +func (w *Worker) Reset() { + w.log.Debug().Msgf("Users before close: %v", w.router.Users()) + w.router.Close() + w.log.Debug().Msgf("Users after close: %v", w.router.Users()) +} const retry = 10 * time.Second