fix sessions manager thread safety.

This commit is contained in:
m1k1o 2021-02-15 15:41:08 +01:00
parent 321e52ee4f
commit 595259b30c
2 changed files with 32 additions and 2 deletions

View File

@ -35,6 +35,7 @@ This app uses Web RTC to stream a desktop inside of a docker container. This is
- Fixed minor gst pipeline bug. - Fixed minor gst pipeline bug.
- Locked screen only for users, admins can still join. - Locked screen only for users, admins can still join.
- Fixed h264 pipelines bugs (by @mbattista). - Fixed h264 pipelines bugs (by @mbattista).
- Fixed sessions manager thread safety by adding mutexes (caused panic in rare edge cases).
### Misc ### Misc
- Custom docker workflow. - Custom docker workflow.

View File

@ -2,6 +2,7 @@ package session
import ( import (
"fmt" "fmt"
"sync"
"github.com/kataras/go-events" "github.com/kataras/go-events"
"github.com/rs/zerolog" "github.com/rs/zerolog"
@ -22,6 +23,7 @@ func New(remote types.RemoteManager) *SessionManager {
} }
type SessionManager struct { type SessionManager struct {
mu sync.Mutex
logger zerolog.Logger logger zerolog.Logger
host string host string
remote types.RemoteManager remote types.RemoteManager
@ -39,13 +41,14 @@ func (manager *SessionManager) New(id string, admin bool, socket types.WebSocket
connected: false, connected: false,
} }
manager.mu.Lock()
manager.members[id] = session manager.members[id] = session
manager.emmiter.Emit("created", id, session)
if manager.remote.Streaming() != true && len(manager.members) > 0 { if manager.remote.Streaming() != true && len(manager.members) > 0 {
manager.remote.StartStream() manager.remote.StartStream()
} }
manager.mu.Unlock()
manager.emmiter.Emit("created", id, session)
return session return session
} }
@ -58,16 +61,23 @@ func (manager *SessionManager) IsHost(id string) bool {
} }
func (manager *SessionManager) SetHost(id string) error { func (manager *SessionManager) SetHost(id string) error {
manager.mu.Lock()
_, ok := manager.members[id] _, ok := manager.members[id]
manager.mu.Unlock()
if ok { if ok {
manager.host = id manager.host = id
manager.emmiter.Emit("host", id) manager.emmiter.Emit("host", id)
return nil return nil
} }
return fmt.Errorf("invalid session id %s", id) return fmt.Errorf("invalid session id %s", id)
} }
func (manager *SessionManager) GetHost() (types.Session, bool) { func (manager *SessionManager) GetHost() (types.Session, bool) {
manager.mu.Lock()
defer manager.mu.Unlock()
host, ok := manager.members[manager.host] host, ok := manager.members[manager.host]
return host, ok return host, ok
} }
@ -79,16 +89,25 @@ func (manager *SessionManager) ClearHost() {
} }
func (manager *SessionManager) Has(id string) bool { func (manager *SessionManager) Has(id string) bool {
manager.mu.Lock()
defer manager.mu.Unlock()
_, ok := manager.members[id] _, ok := manager.members[id]
return ok return ok
} }
func (manager *SessionManager) Get(id string) (types.Session, bool) { func (manager *SessionManager) Get(id string) (types.Session, bool) {
manager.mu.Lock()
defer manager.mu.Unlock()
session, ok := manager.members[id] session, ok := manager.members[id]
return session, ok return session, ok
} }
func (manager *SessionManager) Admins() []*types.Member { func (manager *SessionManager) Admins() []*types.Member {
manager.mu.Lock()
defer manager.mu.Unlock()
members := []*types.Member{} members := []*types.Member{}
for _, session := range manager.members { for _, session := range manager.members {
if !session.connected || !session.admin { if !session.connected || !session.admin {
@ -100,10 +119,14 @@ func (manager *SessionManager) Admins() []*types.Member {
members = append(members, member) members = append(members, member)
} }
} }
return members return members
} }
func (manager *SessionManager) Members() []*types.Member { func (manager *SessionManager) Members() []*types.Member {
manager.mu.Lock()
defer manager.mu.Unlock()
members := []*types.Member{} members := []*types.Member{}
for _, session := range manager.members { for _, session := range manager.members {
if !session.connected { if !session.connected {
@ -119,6 +142,7 @@ func (manager *SessionManager) Members() []*types.Member {
} }
func (manager *SessionManager) Destroy(id string) error { func (manager *SessionManager) Destroy(id string) error {
manager.mu.Lock()
session, ok := manager.members[id] session, ok := manager.members[id]
if ok { if ok {
err := session.destroy() err := session.destroy()
@ -127,11 +151,13 @@ func (manager *SessionManager) Destroy(id string) error {
if manager.remote.Streaming() != false && len(manager.members) <= 0 { if manager.remote.Streaming() != false && len(manager.members) <= 0 {
manager.remote.StopStream() manager.remote.StopStream()
} }
manager.mu.Unlock()
manager.emmiter.Emit("destroyed", id, session) manager.emmiter.Emit("destroyed", id, session)
return err return err
} }
manager.mu.Unlock()
return nil return nil
} }
@ -140,6 +166,9 @@ func (manager *SessionManager) Clear() error {
} }
func (manager *SessionManager) Broadcast(v interface{}, exclude interface{}) error { func (manager *SessionManager) Broadcast(v interface{}, exclude interface{}) error {
manager.mu.Lock()
defer manager.mu.Unlock()
for id, session := range manager.members { for id, session := range manager.members {
if !session.connected { if !session.connected {
continue continue