From a065a015925690a78074b166ebb9c0e3461d5385 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Tue, 26 Jan 2021 17:38:47 -0800 Subject: [PATCH] fixed panic when client connected abruptly --- cmd/cli/client/client.go | 2 ++ magefile.go | 2 +- pkg/routing/errors.go | 1 + pkg/routing/messagechannel.go | 12 +++++++++--- pkg/routing/redis.go | 5 +++++ pkg/rtc/participant.go | 6 +++++- pkg/rtc/participant_internal_test.go | 21 +++++++++++++++++++++ pkg/rtc/room.go | 4 ++-- pkg/rtc/room_test.go | 9 +++++++-- pkg/service/roommanager.go | 2 +- 10 files changed, 54 insertions(+), 10 deletions(-) diff --git a/cmd/cli/client/client.go b/cmd/cli/client/client.go index 6b1efadcf..166ed61c2 100644 --- a/cmd/cli/client/client.go +++ b/cmd/cli/client/client.go @@ -335,6 +335,8 @@ func (c *RTCClient) ResumeLogs() { } func (c *RTCClient) SendRequest(msg *livekit.SignalRequest) error { + c.lock.Lock() + defer c.lock.Unlock() payload, err := protojson.Marshal(msg) if err != nil { return err diff --git a/magefile.go b/magefile.go index ee9fdce2e..9d450d847 100644 --- a/magefile.go +++ b/magefile.go @@ -157,7 +157,7 @@ func Test() error { // run all thests including integration func TestAll() error { mg.Deps(Proto) - cmd := exec.Command("go", "test", "./...") + cmd := exec.Command("go", "test", "./...", "-count=1") connectStd(cmd) return cmd.Run() } diff --git a/pkg/routing/errors.go b/pkg/routing/errors.go index 5eb3053f5..1f040d878 100644 --- a/pkg/routing/errors.go +++ b/pkg/routing/errors.go @@ -8,4 +8,5 @@ var ( ErrNoAvailableNodes = errors.New("could not find any available nodes") ErrIncorrectNodeForRoom = errors.New("incorrect node for the current room") errInvalidRouterMessage = errors.New("invalid router message") + ErrChannelClosed = errors.New("channel closed") ) diff --git a/pkg/routing/messagechannel.go b/pkg/routing/messagechannel.go index 0a635a069..3dfd99bfa 100644 --- a/pkg/routing/messagechannel.go +++ b/pkg/routing/messagechannel.go @@ -5,13 +5,15 @@ import ( ) type MessageChannel struct { - msgChan chan proto.Message - onClose func() + msgChan chan proto.Message + isClosed bool + onClose func() } func NewMessageChannel() *MessageChannel { return &MessageChannel{ - msgChan: make(chan proto.Message, 1), + // allow some buffer to avoid blocked writes + msgChan: make(chan proto.Message, 2), } } @@ -20,6 +22,9 @@ func (m *MessageChannel) OnClose(f func()) { } func (m *MessageChannel) WriteMessage(msg proto.Message) error { + if m.isClosed { + return ErrChannelClosed + } m.msgChan <- msg return nil } @@ -29,6 +34,7 @@ func (m *MessageChannel) ReadChan() <-chan proto.Message { } func (m *MessageChannel) Close() { + m.isClosed = true close(m.msgChan) if m.onClose != nil { m.onClose() diff --git a/pkg/routing/redis.go b/pkg/routing/redis.go index 9d7f595a2..ee4e40010 100644 --- a/pkg/routing/redis.go +++ b/pkg/routing/redis.go @@ -69,6 +69,7 @@ type RedisSink struct { rc *redis.Client nodeId string participantId string + isClosed bool once sync.Once onClose func() } @@ -83,12 +84,16 @@ func NewRedisSink(rc *redis.Client, nodeId, participantId string) *RedisSink { } func (s *RedisSink) WriteMessage(msg proto.Message) error { + if s.isClosed { + return ErrChannelClosed + } return publishRouterMessage(s.rc, s.nodeId, s.participantId, msg) } func (s *RedisSink) Close() { s.once.Do(func() { publishRouterMessage(s.rc, s.nodeId, s.participantId, &livekit.EndSession{}) + s.isClosed = true if s.onClose != nil { s.onClose() } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 9265dc6ac..c8ea115e4 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -320,7 +320,6 @@ func (p *ParticipantImpl) Close() error { if p.ctx.Err() != nil { return p.ctx.Err() } - close(p.rtcpCh) p.onICECandidate = nil p.peerConn.OnDataChannel(nil) p.peerConn.OnICECandidate(nil) @@ -332,6 +331,7 @@ func (p *ParticipantImpl) Close() error { p.onClose(p) } p.cancel() + close(p.rtcpCh) return p.peerConn.Close() } @@ -434,6 +434,10 @@ func (p *ParticipantImpl) scheduleNegotiate() { // initiates server-driven negotiation by creating an offer func (p *ParticipantImpl) negotiate() { + if p.state == livekit.ParticipantInfo_DISCONNECTED { + // skip when disconnected + return + } p.negotiationCond.L.Lock() for p.negotiationState != negotiationStateNone { p.negotiationCond.Wait() diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index c0fec8293..bf8424be4 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -5,6 +5,7 @@ import ( "github.com/stretchr/testify/assert" + "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/routing/routingfakes" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/rtc/types/typesfakes" @@ -68,6 +69,26 @@ func TestTrackPublishEvents(t *testing.T) { assert.True(t, updated) } +// after disconnection, things should continue to function and not panic +func TestDisconnectTiming(t *testing.T) { + t.Run("negotiate doesn't fail after channel closed", func(t *testing.T) { + p := newParticipantForTest("test") + msg := routing.NewMessageChannel() + p.responseSink = msg + go func() { + for msg := range msg.ReadChan() { + t.Log("received message from chan", msg) + } + }() + track := &typesfakes.FakePublishedTrack{} + p.handleTrackPublished(track) + + // close channel and then try to negotiate + msg.Close() + p.negotiate() + }) +} + func newParticipantForTest(name string) *ParticipantImpl { p, _ := NewParticipant( utils.NewGuid(utils.ParticipantPrefix), diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index a7c3fb487..7ce35ff49 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -194,8 +194,8 @@ func (r *Room) broadcastParticipantState(p types.Participant) { updates := ToProtoParticipants([]types.Participant{p}) for _, op := range r.participants { - // skip itself - if p.ID() == op.ID() { + // skip itself && closed participants + if p.ID() == op.ID() || p.State() == livekit.ParticipantInfo_DISCONNECTED { continue } diff --git a/pkg/rtc/room_test.go b/pkg/rtc/room_test.go index f810a0806..bb2fa57b4 100644 --- a/pkg/rtc/room_test.go +++ b/pkg/rtc/room_test.go @@ -56,22 +56,27 @@ func TestRoomJoin(t *testing.T) { }) t.Run("participant state change is broadcasted to others", func(t *testing.T) { - rm := newRoomWithParticipants(t, 1) + rm := newRoomWithParticipants(t, numParticipants) participants := rm.GetParticipants() p := participants[0].(*typesfakes.FakeParticipant) + disconnectedParticipant := participants[1].(*typesfakes.FakeParticipant) + disconnectedParticipant.StateReturns(livekit.ParticipantInfo_DISCONNECTED) rm.RemoveParticipant(p.ID()) p.OnStateChangeArgsForCall(0)(p, livekit.ParticipantInfo_ACTIVE) time.Sleep(defaultDelay) + numUpdates := 0 for _, op := range participants { - if op == p { + if op == p || op == disconnectedParticipant { assert.Zero(t, p.SendParticipantUpdateCallCount()) continue } fakeP := op.(*typesfakes.FakeParticipant) assert.Equal(t, 1, fakeP.SendParticipantUpdateCallCount()) + numUpdates += 1 } + assert.Equal(t, numParticipants-2, numUpdates) }) t.Run("cannot exceed max participants", func(t *testing.T) { diff --git a/pkg/service/roommanager.go b/pkg/service/roommanager.go index c6c823a21..2aca97026 100644 --- a/pkg/service/roommanager.go +++ b/pkg/service/roommanager.go @@ -121,7 +121,7 @@ func (r *RoomManager) Cleanup() error { return nil } -// starts WebRTC session when a new participant is connected +// starts WebRTC session when a new participant is connected, takes place on RTC node func (r *RoomManager) StartSession(roomName, participantId, participantName string, requestSource routing.MessageSource, responseSink routing.MessageSink) { room, err := r.getOrCreateRoom(roomName) if err != nil {