diff --git a/cmd/cli/client/client.go b/cmd/cli/client/client.go index bf369cd3c..f7666918e 100644 --- a/cmd/cli/client/client.go +++ b/cmd/cli/client/client.go @@ -127,19 +127,12 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) { }) c.subscriber.PeerConnection().OnTrack(func(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { - logger.Debugw("track received", "label", track.StreamID(), "id", track.ID(), - "participant", c.localParticipant.Identity) go c.processTrack(track) }) c.subscriber.PeerConnection().OnDataChannel(func(channel *webrtc.DataChannel) { }) - c.publisher.OnNegotiationNeeded(func() { - if !c.iceConnected.Get() { - return - } - c.negotiate() - }) + c.publisher.OnNegotiationNeeded(c.negotiate) c.publisher.PeerConnection().OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) { logger.Debugw("ICE state has changed", "state", connectionState.String(), @@ -329,7 +322,14 @@ func (c *RTCClient) ReadResponse() (*livekit.SignalResponse, error) { // TODO: this function is not thread safe, need to cleanup func (c *RTCClient) SubscribedTracks() map[string][]*webrtc.TrackRemote { - return c.subscribedTracks + // create a copy of this + c.lock.Lock() + defer c.lock.Unlock() + tracks := make(map[string][]*webrtc.TrackRemote, len(c.subscribedTracks)) + for key, val := range c.subscribedTracks { + tracks[key] = val + } + return tracks } func (c *RTCClient) RemoteParticipants() []*livekit.ParticipantInfo { @@ -433,7 +433,7 @@ func (c *RTCClient) AddFileTrack(path string, id string, label string) (writer * return nil, fmt.Errorf("%s has an unsupported extension", filepath.Base(path)) } - logger.Debugw("adding track", + logger.Debugw("adding file track", "mime", mime, ) @@ -506,18 +506,18 @@ func (c *RTCClient) handleAnswer(desc webrtc.SessionDescription) error { return nil } -func (c *RTCClient) negotiate() error { +func (c *RTCClient) negotiate() { logger.Debugw("starting negotiation", "participant", c.localParticipant.Identity) offer, err := c.publisher.PeerConnection().CreateOffer(nil) if err != nil { - return err + return } if err := c.publisher.PeerConnection().SetLocalDescription(offer); err != nil { - return err + return } - return c.SendRequest(&livekit.SignalRequest{ + c.SendRequest(&livekit.SignalRequest{ Message: &livekit.SignalRequest_Offer{ Offer: rtc.ToProtoSessionDescription(offer), }, @@ -526,11 +526,17 @@ func (c *RTCClient) negotiate() error { func (c *RTCClient) processTrack(track *webrtc.TrackRemote) { lastUpdate := time.Time{} - pId, trackId := rtc.UnpackTrackId(track.ID()) + pId := track.StreamID() + trackId := track.ID() c.lock.Lock() c.subscribedTracks[pId] = append(c.subscribedTracks[pId], track) c.lock.Unlock() + logger.Debugw("client added track", "participant", c.localParticipant.Identity, + "source", pId, + "track", trackId, + ) + defer func() { c.lock.Lock() c.subscribedTracks[pId] = funk.Without(c.subscribedTracks[pId], track).([]*webrtc.TrackRemote) diff --git a/pkg/routing/interfaces.go b/pkg/routing/interfaces.go index e709724f4..fbf0638f5 100644 --- a/pkg/routing/interfaces.go +++ b/pkg/routing/interfaces.go @@ -39,9 +39,10 @@ type Router interface { ListNodes() ([]*livekit.Node, error) // participant signal connection is ready to start - StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (reqSink MessageSink, resSource MessageSource, err error) + StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) + // sends a message to RTC node - SendRTCMessage(roomName, identity string, msg *livekit.RTCNodeMessage) error + CreateRTCSink(roomName, identity string) (MessageSink, error) // when a new participant's RTC connection is ready to start OnNewParticipantRTC(callback NewParticipantCallback) diff --git a/pkg/routing/localrouter.go b/pkg/routing/localrouter.go index a4bc05782..3fd593493 100644 --- a/pkg/routing/localrouter.go +++ b/pkg/routing/localrouter.go @@ -4,6 +4,8 @@ import ( "sync" "time" + "github.com/livekit/livekit-server/pkg/logger" + "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/livekit-server/proto/livekit" ) @@ -14,6 +16,9 @@ type LocalRouter struct { // channels for each participant requestChannels map[string]*MessageChannel responseChannels map[string]*MessageChannel + isStarted utils.AtomicFlag + + rtcMessageChan *MessageChannel onNewParticipant NewParticipantCallback onRTCMessage RTCMessageCallback @@ -24,6 +29,7 @@ func NewLocalRouter(currentNode LocalNode) *LocalRouter { currentNode: currentNode, requestChannels: make(map[string]*MessageChannel), responseChannels: make(map[string]*MessageChannel), + rtcMessageChan: NewMessageChannel(), } } @@ -65,10 +71,11 @@ func (r *LocalRouter) ListNodes() ([]*livekit.Node, error) { }, nil } -func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (reqSink MessageSink, resSource MessageSource, err error) { +func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) { // treat it as a new participant connecting if r.onNewParticipant == nil { - return nil, nil, ErrHandlerNotDefined + err = ErrHandlerNotDefined + return } // index channels by roomName | identity @@ -86,15 +93,15 @@ func (r *LocalRouter) StartParticipantSignal(roomName, identity, metadata string // response sink resChan, ) - return reqChan, resChan, nil + return identity, reqChan, resChan, nil } -func (r *LocalRouter) SendRTCMessage(roomName, identity string, msg *livekit.RTCNodeMessage) error { - if r.onRTCMessage == nil { - return nil +func (r *LocalRouter) CreateRTCSink(roomName, identity string) (MessageSink, error) { + if r.rtcMessageChan.isClosed.Get() { + // create a new one + r.rtcMessageChan = NewMessageChannel() } - go r.onRTCMessage(roomName, identity, msg) - return nil + return r.rtcMessageChan, nil } func (r *LocalRouter) OnNewParticipantRTC(callback NewParticipantCallback) { @@ -106,22 +113,62 @@ func (r *LocalRouter) OnRTCMessage(callback RTCMessageCallback) { } func (r *LocalRouter) Start() error { + if !r.isStarted.TrySet(true) { + return nil + } go r.statsWorker() // on local routers, Start doesn't do anything, websocket connections initiate the connections + go r.rtcMessageWorker() return nil } func (r *LocalRouter) Stop() { + + r.rtcMessageChan.Close() } func (r *LocalRouter) statsWorker() { for { + if !r.isStarted.Get() { + return + } // update every 10 seconds <-time.After(statsUpdateInterval) r.currentNode.Stats.UpdatedAt = time.Now().Unix() } } +func (r *LocalRouter) rtcMessageWorker() { + // is a new channel available? if so swap to that one + if !r.isStarted.Get() { + return + } + + // start a new worker after this finished + defer func() { + go r.rtcMessageWorker() + }() + + if r.rtcMessageChan.isClosed.Get() { + // sleep and retry + time.Sleep(time.Second) + } + + // consume messages from + for msg := range r.rtcMessageChan.ReadChan() { + if rtcMsg, ok := msg.(*livekit.RTCNodeMessage); ok { + room, identity, err := parseParticipantKey(rtcMsg.ParticipantKey) + if err != nil { + logger.Errorw("could not process RTC message", "error", err) + continue + } + if r.onRTCMessage != nil { + r.onRTCMessage(room, identity, rtcMsg) + } + } + } +} + func (r *LocalRouter) getOrCreateMessageChannel(target map[string]*MessageChannel, key string) *MessageChannel { r.lock.Lock() defer r.lock.Unlock() diff --git a/pkg/routing/redisrouter.go b/pkg/routing/redisrouter.go index 252407380..b396a4473 100644 --- a/pkg/routing/redisrouter.go +++ b/pkg/routing/redisrouter.go @@ -28,8 +28,6 @@ type RedisRouter struct { ctx context.Context isStarted utils.AtomicFlag - // map of participantKey => RTCNodeSink - rtcSinks map[string]*RTCNodeSink // map of connectionId => SignalNodeSink signalSinks map[string]*SignalNodeSink @@ -41,7 +39,6 @@ func NewRedisRouter(currentNode LocalNode, rc *redis.Client) *RedisRouter { rr := &RedisRouter{ LocalRouter: *NewLocalRouter(currentNode), rc: rc, - rtcSinks: make(map[string]*RTCNodeSink), signalSinks: make(map[string]*SignalNodeSink), } rr.ctx, rr.cancel = context.WithCancel(context.Background()) @@ -127,7 +124,7 @@ func (r *RedisRouter) ListNodes() ([]*livekit.Node, error) { } // signal connection sets up paths to the RTC node, and starts to route messages to that message queue -func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (reqSink MessageSink, resSource MessageSource, err error) { +func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string, reconnect bool) (connectionId string, reqSink MessageSink, resSource MessageSource, err error) { // find the node where the room is hosted at rtcNode, err := r.GetNodeForRoom(roomName) if err != nil { @@ -135,7 +132,7 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string } // create a new connection id - connectionId := utils.NewGuid("CO_") + connectionId = utils.NewGuid("CO_") pKey := participantKey(roomName, identity) // map signal & rtc nodes @@ -143,7 +140,7 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string return } - sink := r.getOrCreateRTCSink(rtcNode, pKey) + sink := NewRTCNodeSink(r.rc, rtcNode, pKey) // sends a message to start session err = sink.WriteMessage(&livekit.StartSession{ @@ -160,19 +157,17 @@ func (r *RedisRouter) StartParticipantSignal(roomName, identity, metadata string // index by connectionId, since there may be multiple connections for the participant resChan := r.getOrCreateMessageChannel(r.responseChannels, connectionId) - return sink, resChan, nil + return connectionId, sink, resChan, nil } -func (r *RedisRouter) SendRTCMessage(roomName, identity string, msg *livekit.RTCNodeMessage) error { +func (r *RedisRouter) CreateRTCSink(roomName, identity string) (MessageSink, error) { pkey := participantKey(roomName, identity) rtcNode, err := r.getParticipantRTCNode(pkey) if err != nil { - return err + return nil, err } - rtcSink := r.getOrCreateRTCSink(rtcNode, pkey) - - return rtcSink.WriteMessage(msg) + return NewRTCNodeSink(r.rc, rtcNode, pkey), nil } func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantKey string) error { @@ -216,7 +211,7 @@ func (r *RedisRouter) startParticipantRTC(ss *livekit.StartSession, participantK } reqChan := r.getOrCreateMessageChannel(r.requestChannels, participantKey) - resSink := r.getOrCreateSignalSink(signalNode, ss.ConnectionId) + resSink := NewSignalNodeSink(r.rc, signalNode, ss.ConnectionId) r.onNewParticipant( ss.RoomName, ss.Identity, @@ -261,44 +256,6 @@ func (r *RedisRouter) setParticipantSignalNode(connectionId, nodeId string) erro return nil } -func (r *RedisRouter) getOrCreateRTCSink(nodeId string, participantKey string) *RTCNodeSink { - r.lock.Lock() - defer r.lock.Unlock() - sink := r.rtcSinks[participantKey] - - if sink != nil { - return sink - } - - sink = NewRTCNodeSink(r.rc, nodeId, participantKey) - sink.OnClose(func() { - r.lock.Lock() - delete(r.rtcSinks, participantKey) - r.lock.Unlock() - }) - r.rtcSinks[participantKey] = sink - return sink -} - -func (r *RedisRouter) getOrCreateSignalSink(nodeId string, connectionId string) *SignalNodeSink { - r.lock.Lock() - defer r.lock.Unlock() - sink := r.signalSinks[connectionId] - - if sink != nil { - return sink - } - - sink = NewSignalNodeSink(r.rc, nodeId, connectionId) - sink.OnClose(func() { - r.lock.Lock() - delete(r.signalSinks, connectionId) - r.lock.Unlock() - }) - r.signalSinks[connectionId] = sink - return sink -} - func (r *RedisRouter) getParticipantRTCNode(participantKey string) (string, error) { return r.rc.Get(r.ctx, participantRTCKey(participantKey)).Result() } diff --git a/pkg/routing/routingfakes/fake_router.go b/pkg/routing/routingfakes/fake_router.go index f74ef99ce..320b79fbb 100644 --- a/pkg/routing/routingfakes/fake_router.go +++ b/pkg/routing/routingfakes/fake_router.go @@ -20,6 +20,20 @@ type FakeRouter struct { clearRoomStateReturnsOnCall map[int]struct { result1 error } + CreateRTCSinkStub func(string, string) (routing.MessageSink, error) + createRTCSinkMutex sync.RWMutex + createRTCSinkArgsForCall []struct { + arg1 string + arg2 string + } + createRTCSinkReturns struct { + result1 routing.MessageSink + result2 error + } + createRTCSinkReturnsOnCall map[int]struct { + result1 routing.MessageSink + result2 error + } GetNodeStub func(string) (*livekit.Node, error) getNodeMutex sync.RWMutex getNodeArgsForCall []struct { @@ -88,19 +102,6 @@ type FakeRouter struct { removeDeadNodesReturnsOnCall map[int]struct { result1 error } - SendRTCMessageStub func(string, string, *livekit.RTCNodeMessage) error - sendRTCMessageMutex sync.RWMutex - sendRTCMessageArgsForCall []struct { - arg1 string - arg2 string - arg3 *livekit.RTCNodeMessage - } - sendRTCMessageReturns struct { - result1 error - } - sendRTCMessageReturnsOnCall map[int]struct { - result1 error - } SetNodeForRoomStub func(string, string) error setNodeForRoomMutex sync.RWMutex setNodeForRoomArgsForCall []struct { @@ -123,7 +124,7 @@ type FakeRouter struct { startReturnsOnCall map[int]struct { result1 error } - StartParticipantSignalStub func(string, string, string, bool) (routing.MessageSink, routing.MessageSource, error) + StartParticipantSignalStub func(string, string, string, bool) (string, routing.MessageSink, routing.MessageSource, error) startParticipantSignalMutex sync.RWMutex startParticipantSignalArgsForCall []struct { arg1 string @@ -132,14 +133,16 @@ type FakeRouter struct { arg4 bool } startParticipantSignalReturns struct { - result1 routing.MessageSink - result2 routing.MessageSource - result3 error + result1 string + result2 routing.MessageSink + result3 routing.MessageSource + result4 error } startParticipantSignalReturnsOnCall map[int]struct { - result1 routing.MessageSink - result2 routing.MessageSource - result3 error + result1 string + result2 routing.MessageSink + result3 routing.MessageSource + result4 error } StopStub func() stopMutex sync.RWMutex @@ -220,6 +223,71 @@ func (fake *FakeRouter) ClearRoomStateReturnsOnCall(i int, result1 error) { }{result1} } +func (fake *FakeRouter) CreateRTCSink(arg1 string, arg2 string) (routing.MessageSink, error) { + fake.createRTCSinkMutex.Lock() + ret, specificReturn := fake.createRTCSinkReturnsOnCall[len(fake.createRTCSinkArgsForCall)] + fake.createRTCSinkArgsForCall = append(fake.createRTCSinkArgsForCall, struct { + arg1 string + arg2 string + }{arg1, arg2}) + stub := fake.CreateRTCSinkStub + fakeReturns := fake.createRTCSinkReturns + fake.recordInvocation("CreateRTCSink", []interface{}{arg1, arg2}) + fake.createRTCSinkMutex.Unlock() + if stub != nil { + return stub(arg1, arg2) + } + if specificReturn { + return ret.result1, ret.result2 + } + return fakeReturns.result1, fakeReturns.result2 +} + +func (fake *FakeRouter) CreateRTCSinkCallCount() int { + fake.createRTCSinkMutex.RLock() + defer fake.createRTCSinkMutex.RUnlock() + return len(fake.createRTCSinkArgsForCall) +} + +func (fake *FakeRouter) CreateRTCSinkCalls(stub func(string, string) (routing.MessageSink, error)) { + fake.createRTCSinkMutex.Lock() + defer fake.createRTCSinkMutex.Unlock() + fake.CreateRTCSinkStub = stub +} + +func (fake *FakeRouter) CreateRTCSinkArgsForCall(i int) (string, string) { + fake.createRTCSinkMutex.RLock() + defer fake.createRTCSinkMutex.RUnlock() + argsForCall := fake.createRTCSinkArgsForCall[i] + return argsForCall.arg1, argsForCall.arg2 +} + +func (fake *FakeRouter) CreateRTCSinkReturns(result1 routing.MessageSink, result2 error) { + fake.createRTCSinkMutex.Lock() + defer fake.createRTCSinkMutex.Unlock() + fake.CreateRTCSinkStub = nil + fake.createRTCSinkReturns = struct { + result1 routing.MessageSink + result2 error + }{result1, result2} +} + +func (fake *FakeRouter) CreateRTCSinkReturnsOnCall(i int, result1 routing.MessageSink, result2 error) { + fake.createRTCSinkMutex.Lock() + defer fake.createRTCSinkMutex.Unlock() + fake.CreateRTCSinkStub = nil + if fake.createRTCSinkReturnsOnCall == nil { + fake.createRTCSinkReturnsOnCall = make(map[int]struct { + result1 routing.MessageSink + result2 error + }) + } + fake.createRTCSinkReturnsOnCall[i] = struct { + result1 routing.MessageSink + result2 error + }{result1, result2} +} + func (fake *FakeRouter) GetNode(arg1 string) (*livekit.Node, error) { fake.getNodeMutex.Lock() ret, specificReturn := fake.getNodeReturnsOnCall[len(fake.getNodeArgsForCall)] @@ -574,69 +642,6 @@ func (fake *FakeRouter) RemoveDeadNodesReturnsOnCall(i int, result1 error) { }{result1} } -func (fake *FakeRouter) SendRTCMessage(arg1 string, arg2 string, arg3 *livekit.RTCNodeMessage) error { - fake.sendRTCMessageMutex.Lock() - ret, specificReturn := fake.sendRTCMessageReturnsOnCall[len(fake.sendRTCMessageArgsForCall)] - fake.sendRTCMessageArgsForCall = append(fake.sendRTCMessageArgsForCall, struct { - arg1 string - arg2 string - arg3 *livekit.RTCNodeMessage - }{arg1, arg2, arg3}) - stub := fake.SendRTCMessageStub - fakeReturns := fake.sendRTCMessageReturns - fake.recordInvocation("SendRTCMessage", []interface{}{arg1, arg2, arg3}) - fake.sendRTCMessageMutex.Unlock() - if stub != nil { - return stub(arg1, arg2, arg3) - } - if specificReturn { - return ret.result1 - } - return fakeReturns.result1 -} - -func (fake *FakeRouter) SendRTCMessageCallCount() int { - fake.sendRTCMessageMutex.RLock() - defer fake.sendRTCMessageMutex.RUnlock() - return len(fake.sendRTCMessageArgsForCall) -} - -func (fake *FakeRouter) SendRTCMessageCalls(stub func(string, string, *livekit.RTCNodeMessage) error) { - fake.sendRTCMessageMutex.Lock() - defer fake.sendRTCMessageMutex.Unlock() - fake.SendRTCMessageStub = stub -} - -func (fake *FakeRouter) SendRTCMessageArgsForCall(i int) (string, string, *livekit.RTCNodeMessage) { - fake.sendRTCMessageMutex.RLock() - defer fake.sendRTCMessageMutex.RUnlock() - argsForCall := fake.sendRTCMessageArgsForCall[i] - return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3 -} - -func (fake *FakeRouter) SendRTCMessageReturns(result1 error) { - fake.sendRTCMessageMutex.Lock() - defer fake.sendRTCMessageMutex.Unlock() - fake.SendRTCMessageStub = nil - fake.sendRTCMessageReturns = struct { - result1 error - }{result1} -} - -func (fake *FakeRouter) SendRTCMessageReturnsOnCall(i int, result1 error) { - fake.sendRTCMessageMutex.Lock() - defer fake.sendRTCMessageMutex.Unlock() - fake.SendRTCMessageStub = nil - if fake.sendRTCMessageReturnsOnCall == nil { - fake.sendRTCMessageReturnsOnCall = make(map[int]struct { - result1 error - }) - } - fake.sendRTCMessageReturnsOnCall[i] = struct { - result1 error - }{result1} -} - func (fake *FakeRouter) SetNodeForRoom(arg1 string, arg2 string) error { fake.setNodeForRoomMutex.Lock() ret, specificReturn := fake.setNodeForRoomReturnsOnCall[len(fake.setNodeForRoomArgsForCall)] @@ -752,7 +757,7 @@ func (fake *FakeRouter) StartReturnsOnCall(i int, result1 error) { }{result1} } -func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 string, arg4 bool) (routing.MessageSink, routing.MessageSource, error) { +func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 string, arg4 bool) (string, routing.MessageSink, routing.MessageSource, error) { fake.startParticipantSignalMutex.Lock() ret, specificReturn := fake.startParticipantSignalReturnsOnCall[len(fake.startParticipantSignalArgsForCall)] fake.startParticipantSignalArgsForCall = append(fake.startParticipantSignalArgsForCall, struct { @@ -769,9 +774,9 @@ func (fake *FakeRouter) StartParticipantSignal(arg1 string, arg2 string, arg3 st return stub(arg1, arg2, arg3, arg4) } if specificReturn { - return ret.result1, ret.result2, ret.result3 + return ret.result1, ret.result2, ret.result3, ret.result4 } - return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3 + return fakeReturns.result1, fakeReturns.result2, fakeReturns.result3, fakeReturns.result4 } func (fake *FakeRouter) StartParticipantSignalCallCount() int { @@ -780,7 +785,7 @@ func (fake *FakeRouter) StartParticipantSignalCallCount() int { return len(fake.startParticipantSignalArgsForCall) } -func (fake *FakeRouter) StartParticipantSignalCalls(stub func(string, string, string, bool) (routing.MessageSink, routing.MessageSource, error)) { +func (fake *FakeRouter) StartParticipantSignalCalls(stub func(string, string, string, bool) (string, routing.MessageSink, routing.MessageSource, error)) { fake.startParticipantSignalMutex.Lock() defer fake.startParticipantSignalMutex.Unlock() fake.StartParticipantSignalStub = stub @@ -793,33 +798,36 @@ func (fake *FakeRouter) StartParticipantSignalArgsForCall(i int) (string, string return argsForCall.arg1, argsForCall.arg2, argsForCall.arg3, argsForCall.arg4 } -func (fake *FakeRouter) StartParticipantSignalReturns(result1 routing.MessageSink, result2 routing.MessageSource, result3 error) { +func (fake *FakeRouter) StartParticipantSignalReturns(result1 string, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) { fake.startParticipantSignalMutex.Lock() defer fake.startParticipantSignalMutex.Unlock() fake.StartParticipantSignalStub = nil fake.startParticipantSignalReturns = struct { - result1 routing.MessageSink - result2 routing.MessageSource - result3 error - }{result1, result2, result3} + result1 string + result2 routing.MessageSink + result3 routing.MessageSource + result4 error + }{result1, result2, result3, result4} } -func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 routing.MessageSink, result2 routing.MessageSource, result3 error) { +func (fake *FakeRouter) StartParticipantSignalReturnsOnCall(i int, result1 string, result2 routing.MessageSink, result3 routing.MessageSource, result4 error) { fake.startParticipantSignalMutex.Lock() defer fake.startParticipantSignalMutex.Unlock() fake.StartParticipantSignalStub = nil if fake.startParticipantSignalReturnsOnCall == nil { fake.startParticipantSignalReturnsOnCall = make(map[int]struct { - result1 routing.MessageSink - result2 routing.MessageSource - result3 error + result1 string + result2 routing.MessageSink + result3 routing.MessageSource + result4 error }) } fake.startParticipantSignalReturnsOnCall[i] = struct { - result1 routing.MessageSink - result2 routing.MessageSource - result3 error - }{result1, result2, result3} + result1 string + result2 routing.MessageSink + result3 routing.MessageSource + result4 error + }{result1, result2, result3, result4} } func (fake *FakeRouter) Stop() { @@ -904,6 +912,8 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} { defer fake.invocationsMutex.RUnlock() fake.clearRoomStateMutex.RLock() defer fake.clearRoomStateMutex.RUnlock() + fake.createRTCSinkMutex.RLock() + defer fake.createRTCSinkMutex.RUnlock() fake.getNodeMutex.RLock() defer fake.getNodeMutex.RUnlock() fake.getNodeForRoomMutex.RLock() @@ -918,8 +928,6 @@ func (fake *FakeRouter) Invocations() map[string][][]interface{} { defer fake.registerNodeMutex.RUnlock() fake.removeDeadNodesMutex.RLock() defer fake.removeDeadNodesMutex.RUnlock() - fake.sendRTCMessageMutex.RLock() - defer fake.sendRTCMessageMutex.RUnlock() fake.setNodeForRoomMutex.RLock() defer fake.setNodeForRoomMutex.RUnlock() fake.startMutex.RLock() diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 3c362a1f6..951a8bc91 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -2,6 +2,7 @@ package rtc import ( "encoding/json" + "fmt" "io" "sync" "sync/atomic" @@ -225,7 +226,7 @@ func (p *ParticipantImpl) HandleOffer(sdp webrtc.SessionDescription) (answer web "participant", p.Identity(), //"sdp", sdp.SDP, ) - err = p.responseSink.WriteMessage(&livekit.SignalResponse{ + p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Answer{ Answer: ToProtoSessionDescription(answer), }, @@ -254,7 +255,7 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac } p.pendingTracks[clientId] = ti - err := p.responseSink.WriteMessage(&livekit.SignalResponse{ + p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_TrackPublished{ TrackPublished: &livekit.TrackPublishedResponse{ Cid: clientId, @@ -262,10 +263,6 @@ func (p *ParticipantImpl) AddTrack(clientId, name string, trackType livekit.Trac }, }, }) - if err != nil { - logger.Errorw("could not write message", "error", err, - "participant", p.identity) - } } // handles a client answer response, with subscriber PC, server initiates the offer @@ -324,9 +321,13 @@ func (p *ParticipantImpl) Close() error { p.subscriber.pc.OnNegotiationNeeded(nil) p.subscriber.pc.OnTrack(nil) p.publisher.pc.OnICECandidate(nil) + // ensure this is synchronized + p.lock.RLock() p.responseSink.Close() - if p.onClose != nil { - p.onClose(p) + onClose := p.onClose + p.lock.RUnlock() + if onClose != nil { + onClose(p) } p.publisher.Close() p.subscriber.Close() @@ -369,7 +370,7 @@ func (p *ParticipantImpl) RemoveSubscriber(participantId string) { // signal connection methods func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticipants []types.Participant) error { // send Join response - return p.responseSink.WriteMessage(&livekit.SignalResponse{ + return p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Join{ Join: &livekit.JoinResponse{ Room: roomInfo, @@ -386,7 +387,7 @@ func (p *ParticipantImpl) SendParticipantUpdate(participants []*livekit.Particip return nil } - return p.responseSink.WriteMessage(&livekit.SignalResponse{ + return p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Update{ Update: &livekit.ParticipantUpdate{ Participants: participants, @@ -400,7 +401,7 @@ func (p *ParticipantImpl) SendActiveSpeakers(speakers []*livekit.SpeakerInfo) er return nil } - return p.responseSink.WriteMessage(&livekit.SignalResponse{ + return p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Speaker{ Speaker: &livekit.ActiveSpeakerUpdate{ Speakers: speakers, @@ -485,20 +486,17 @@ func (p *ParticipantImpl) sendIceCandidate(c *webrtc.ICECandidate, target liveki //logger.Debugw("sending ice candidates") trickle := ToProtoTrickle(ci) trickle.Target = target - err := p.responseSink.WriteMessage(&livekit.SignalResponse{ + p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Trickle{ Trickle: trickle, }, }) - if err != nil { - logger.Errorw("could not send trickle", "err", err, - "participant", p.identity) - } } // initiates server-driven negotiation by creating an offer func (p *ParticipantImpl) negotiate() { if p.State() == livekit.ParticipantInfo_DISCONNECTED { + logger.Debugw("skipping server negotiation", "participant", p.Identity()) // skip when disconnected return } @@ -521,16 +519,12 @@ func (p *ParticipantImpl) negotiate() { "participant", p.Identity(), //"sdp", offer.SDP, ) - err = p.responseSink.WriteMessage(&livekit.SignalResponse{ + + p.writeMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Offer{ Offer: ToProtoSessionDescription(offer), }, }) - if err != nil { - logger.Errorw("could not send offer to participant", - "err", err, - "participant", p.identity) - } } func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) { @@ -548,6 +542,23 @@ func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) { } } +func (p *ParticipantImpl) writeMessage(msg *livekit.SignalResponse) error { + if p.State() == livekit.ParticipantInfo_DISCONNECTED { + return nil + } + sink := p.responseSink + err := sink.WriteMessage(msg) + if err != nil { + logger.Warnw("could not send message to participant", + "error", err, + "id", p.ID(), + "participant", p.identity, + "message", fmt.Sprintf("%T", msg.Message)) + return err + } + return nil +} + // when a new remoteTrack is created, creates a Track and adds it to room func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *webrtc.RTPReceiver) { logger.Debugw("mediaTrack added", "participant", p.Identity(), "remoteTrack", track.ID()) diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index fcf7f2fed..9fe7bd72c 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -8,6 +8,7 @@ import ( "github.com/bep/debounce" "github.com/pion/webrtc/v3" + "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/proto/livekit" ) @@ -111,6 +112,7 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error { state := t.negotiationState.Load().(int) t.negotiationState.Store(negotiationStateNone) if state == negotiationStateServer && t.onNegotiation != nil { + logger.Debugw("negotiating again") // need to negotiate again t.negotiate() } @@ -126,14 +128,14 @@ func (t *PCTransport) negotiate() { t.debouncedNegotiate(func() { state := t.negotiationState.Load().(int) // when there's an ongoing negotiation, let it finish and not disrupt its state - if state != negotiationStateNone { + if state == negotiationStateClient { + logger.Debugw("skipping negotiation, trying again later") t.negotiationState.Store(negotiationStateServer) return } if t.onNegotiation != nil { t.onNegotiation() - // indicate waiting for client t.negotiationState.Store(negotiationStateClient) } diff --git a/pkg/service/roomservice.go b/pkg/service/roomservice.go index 2e5aeb631..41b04042c 100644 --- a/pkg/service/roomservice.go +++ b/pkg/service/roomservice.go @@ -106,7 +106,12 @@ func (s *RoomService) RemoveParticipant(ctx context.Context, req *livekit.RoomPa return } - err = s.roomManager.router.SendRTCMessage(req.Room, participant.Identity, &livekit.RTCNodeMessage{ + rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participant.Identity) + if err != nil { + return + } + defer rtcSink.Close() + err = rtcSink.WriteMessage(&livekit.RTCNodeMessage{ Message: &livekit.RTCNodeMessage_RemoveParticipant{ RemoveParticipant: req, }, @@ -126,7 +131,12 @@ func (s *RoomService) MutePublishedTrack(ctx context.Context, req *livekit.MuteR return } - err = s.roomManager.router.SendRTCMessage(req.Room, participant.Identity, &livekit.RTCNodeMessage{ + rtcSink, err := s.roomManager.router.CreateRTCSink(req.Room, participant.Identity) + if err != nil { + return + } + defer rtcSink.Close() + err = rtcSink.WriteMessage(&livekit.RTCNodeMessage{ Message: &livekit.RTCNodeMessage_MuteTrack{ MuteTrack: req, }, diff --git a/pkg/service/rtcservice.go b/pkg/service/rtcservice.go index 1e0f927fd..49e950e15 100644 --- a/pkg/service/rtcservice.go +++ b/pkg/service/rtcservice.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "net/http" + "strings" "github.com/gorilla/websocket" @@ -84,15 +85,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { } // this needs to be started first *before* using router functions on this node - reqSink, resSource, err := s.router.StartParticipantSignal(roomName, identity, metadata, isReconnect) + connId, reqSink, resSource, err := s.router.StartParticipantSignal(roomName, identity, metadata, isReconnect) if err != nil { handleError(w, http.StatusInternalServerError, "could not start session: "+err.Error()) return } done := make(chan bool, 1) + // function exits when websocket terminates, it'll close the event reading off of response sink as well defer func() { - logger.Infow("WS connection closed", "participant", identity) + logger.Infow("WS connection closed", "participant", identity, "connectionId", connId) reqSink.Close() close(done) }() @@ -109,6 +111,7 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { sigConn := NewWSSignalConnection(conn) logger.Infow("new client WS connected", + "connectionId", connId, "room", rm.Sid, "roomName", rm.Name, "name", identity, @@ -128,14 +131,16 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { return case msg := <-resSource.ReadChan(): if msg == nil { - logger.Errorw("source closed connection", "participant", identity) + logger.Infow("source closed connection", "participant", identity, + "connectionId", connId) return } res, ok := msg.(*livekit.SignalResponse) if !ok { logger.Errorw("unexpected message type", "type", fmt.Sprintf("%T", msg), - "participant", identity) + "participant", identity, + "connectionId", connId) continue } @@ -151,15 +156,17 @@ func (s *RTCService) ServeHTTP(w http.ResponseWriter, r *http.Request) { for { req, err := sigConn.ReadRequest() // normal closure - if err == io.EOF || websocket.IsCloseError(err, websocket.CloseAbnormalClosure, websocket.CloseGoingAway, websocket.CloseNormalClosure) { - return - } else if err != nil { - logger.Errorw("error reading from websocket", "error", err) - return + if err != nil { + if err == io.EOF || strings.HasSuffix(err.Error(), "use of closed network connection") || websocket.IsCloseError(err, websocket.CloseAbnormalClosure, websocket.CloseGoingAway, websocket.CloseNormalClosure) { + return + } else { + logger.Errorw("error reading from websocket", "error", err) + return + } } - - if err = reqSink.WriteMessage(req); err != nil { - logger.Warnw("error writing to request sink", "error", err) + if err := reqSink.WriteMessage(req); err != nil { + logger.Warnw("error writing to request sink", "error", err, + "participant", identity, "connectionId", connId) } } } diff --git a/test/multinode_test.go b/test/multinode_test.go index 4fa64ec84..0672a8c19 100644 --- a/test/multinode_test.go +++ b/test/multinode_test.go @@ -51,7 +51,7 @@ func TestMultiNodeRouting(t *testing.T) { } tr1 := c2.SubscribedTracks()[c1.ID()][0] - assert.Equal(t, "webcam", tr1.StreamID()) + assert.Equal(t, c1.ID(), tr1.StreamID()) return true }) diff --git a/test/singlenode_test.go b/test/singlenode_test.go index b46824058..f1d6b66da 100644 --- a/test/singlenode_test.go +++ b/test/singlenode_test.go @@ -68,7 +68,7 @@ func TestSinglePublisher(t *testing.T) { } tr1 := c2.SubscribedTracks()[c1.ID()][0] - assert.Equal(t, "webcam", tr1.StreamID()) + assert.Equal(t, c1.ID(), tr1.StreamID()) return true }) if !success { @@ -94,6 +94,6 @@ func TestSinglePublisher(t *testing.T) { // ensure that the track ids are generated by server tracks := c3.SubscribedTracks()[c1.ID()] for _, tr := range tracks { - assert.True(t, strings.Contains(tr.ID(), "|TR_"), "track should begin with TR") + assert.True(t, strings.HasPrefix(tr.ID(), "TR_"), "track should begin with TR") } }