diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index 2e2d5df5c..a6d6fef8d 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -12,7 +12,7 @@ import ( // a router of messages on the same node, basic implementation for local testing type LocalRouter struct { currentNode LocalNode - lock sync.Mutex + lock sync.RWMutex // channels for each participant requestChannels map[string]*MessageChannel responseChannels map[string]*MessageChannel diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 50be984aa..2e38c2192 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -208,9 +208,9 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK // when it's not reconnecting, we do not want to re-use the same response sink // the previous rtc worker thread is still consuming off of it. // we'll want to sever the connection and switch to the new one - r.lock.Lock() + r.lock.RLock() requestChan, ok := r.requestChannels[participantKey] - r.lock.Unlock() + r.lock.RUnlock() if ok { requestChan.Close() } @@ -334,9 +334,9 @@ func (r *RedisRouter) redisWorker() { func (r *RedisRouter) handleSignalMessage(sm *livekit.SignalNodeMessage) error { connectionId := sm.ConnectionId - r.lock.Lock() + r.lock.RLock() resSink := r.responseChannels[connectionId] - r.lock.Unlock() + r.lock.RUnlock() // if a client closed the channel, then sent more messages after that, if resSink == nil { @@ -371,9 +371,9 @@ func (r *RedisRouter) handleRTCMessage(rm *livekit.RTCNodeMessage) error { } case *livekit.RTCNodeMessage_Request: - r.lock.Lock() + r.lock.RLock() requestChan := r.requestChannels[pKey] - r.lock.Unlock() + r.lock.RUnlock() if err := requestChan.WriteMessage(rmb.Request); err != nil { return err }