diff --git a/cmd/cli/client/client.go b/cmd/cli/client/client.go index 7f3e9be7d..25390cf4e 100644 --- a/cmd/cli/client/client.go +++ b/cmd/cli/client/client.go @@ -19,6 +19,7 @@ import ( "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/pkg/rtc" + "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/livekit-server/proto/livekit" ) @@ -32,8 +33,8 @@ type RTCClient struct { wsLock sync.Mutex ctx context.Context cancel context.CancelFunc - connected bool - iceConnected bool + connected utils.AtomicFlag + iceConnected utils.AtomicFlag me *webrtc.MediaEngine // optional, populated only when receiving tracks subscribedTracks map[string][]*webrtc.TrackRemote localParticipant *livekit.ParticipantInfo @@ -103,7 +104,7 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) { return } - if !c.connected { + if !c.connected.Get() { c.lock.Lock() defer c.lock.Unlock() // not connected, save to pending @@ -123,7 +124,7 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) { }) peerConn.OnNegotiationNeeded(func() { - if !c.iceConnected { + if !c.iceConnected.Get() { return } c.requestNegotiation() @@ -145,9 +146,9 @@ func NewRTCClient(conn *websocket.Conn) (*RTCClient, error) { } } - initialConnect := !c.iceConnected + initialConnect := !c.iceConnected.Get() c.pendingTrackWriters = nil - c.iceConnected = true + c.iceConnected.TrySet(true) if initialConnect && c.OnConnected != nil { go c.OnConnected() @@ -271,7 +272,7 @@ func (c *RTCClient) WaitUntilConnected() error { case <-ctx.Done(): return errors.New("could not connect after timeout") case <-time.After(10 * time.Millisecond): - if c.iceConnected { + if c.iceConnected.Get() { return nil } } @@ -318,8 +319,8 @@ func (c *RTCClient) RemoteParticipants() []*livekit.ParticipantInfo { } func (c *RTCClient) Stop() { - c.connected = false - c.iceConnected = false + c.connected.TrySet(false) + c.iceConnected.TrySet(false) c.conn.Close() c.PeerConn.Close() c.cancel() @@ -387,7 +388,7 @@ func (c *RTCClient) AddTrack(track *webrtc.TrackLocalStaticSample, path string) writer = NewTrackWriter(c.ctx, track, path) // write tracks only after ICE connectivity - if c.iceConnected { + if c.iceConnected.Get() { err = writer.Start() } else { c.pendingTrackWriters = append(c.pendingTrackWriters, writer) @@ -474,17 +475,18 @@ func (c *RTCClient) handleAnswer(desc webrtc.SessionDescription) error { return err } - if !c.connected { - c.connected = true - - // add all the pending items - c.lock.Lock() - for _, ic := range c.pendingCandidates { - c.SendIceCandidate(ic) - } - c.pendingCandidates = nil - c.lock.Unlock() + if !c.connected.TrySet(true) { + // already connected + return nil } + + // add all the pending items + c.lock.Lock() + for _, ic := range c.pendingCandidates { + c.SendIceCandidate(ic) + } + c.pendingCandidates = nil + c.lock.Unlock() return nil } diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 475f6a08d..7c669e755 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -33,7 +33,7 @@ var ( type MediaTrack struct { id string participantId string - muted bool + muted utils.AtomicFlag ssrc webrtc.SSRC name string @@ -93,7 +93,7 @@ func (t *MediaTrack) Name() string { } func (t *MediaTrack) IsMuted() bool { - return t.muted + return t.muted.Get() } func (t *MediaTrack) OnClose(f func()) { @@ -251,7 +251,7 @@ func (t *MediaTrack) forwardRTPWorker() { // "track", t.ID()) // when track is muted, it's "disabled" on the client side, and will still be sending black frames // when our metadata is updated as such, we shortcircuit forwarding of black frames - if t.muted { + if t.muted.Get() { continue } diff --git a/pkg/rtc/mediatrack_internal_test.go b/pkg/rtc/mediatrack_internal_test.go index 95460f4f0..0b2624609 100644 --- a/pkg/rtc/mediatrack_internal_test.go +++ b/pkg/rtc/mediatrack_internal_test.go @@ -41,7 +41,7 @@ func TestForwardRTP(t *testing.T) { t.Run("muted tracks do not forward data", func(t *testing.T) { mt := newMediaTrackWithReceiver() - mt.muted = true + mt.muted.TrySet(true) dt := &typesfakes.FakeDownTrack{} mt.downtracks["test"] = dt @@ -89,7 +89,6 @@ func newMediaTrackWithReceiver() *MediaTrack { return &MediaTrack{ id: utils.NewGuid(utils.TrackPrefix), participantId: "PAtest", - muted: false, kind: livekit.TrackType_VIDEO, codec: webrtc.RTPCodecParameters{}, rtcpCh: utils.NewCalmChannel(5), diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 8af30a982..01e7b140b 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -4,6 +4,7 @@ import ( "context" "io" "sync" + "sync/atomic" "time" "github.com/bep/debounce" @@ -40,7 +41,7 @@ type ParticipantImpl struct { cancel context.CancelFunc mediaEngine *webrtc.MediaEngine name string - state livekit.ParticipantInfo_State + state atomic.Value // livekit.ParticipantInfo_State rtcpCh *utils.CalmChannel subscribedTracks map[string][]*sfu.DownTrack // publishedTracks that participant is publishing @@ -90,7 +91,6 @@ func NewParticipant(participantId, name string, pc types.PeerConnection, rs rout cancel: cancel, rtcpCh: utils.NewCalmChannel(50), subscribedTracks: make(map[string][]*sfu.DownTrack), - state: livekit.ParticipantInfo_JOINING, lock: sync.RWMutex{}, negotiationCond: sync.NewCond(&sync.Mutex{}), publishedTracks: make(map[string]types.PublishedTrack, 0), @@ -98,6 +98,7 @@ func NewParticipant(participantId, name string, pc types.PeerConnection, rs rout mediaEngine: me, debouncedNegotiate: debounce.New(negotiationFrequency), } + participant.state.Store(livekit.ParticipantInfo_JOINING) pc.OnTrack(participant.onMediaTrack) @@ -156,11 +157,12 @@ func (p *ParticipantImpl) Name() string { } func (p *ParticipantImpl) State() livekit.ParticipantInfo_State { - return p.state + return p.state.Load().(livekit.ParticipantInfo_State) } func (p *ParticipantImpl) IsReady() bool { - return p.state == livekit.ParticipantInfo_JOINED || p.state == livekit.ParticipantInfo_ACTIVE + state := p.State() + return state == livekit.ParticipantInfo_JOINED || state == livekit.ParticipantInfo_ACTIVE } func (p *ParticipantImpl) RTCPChan() *utils.CalmChannel { @@ -171,7 +173,7 @@ func (p *ParticipantImpl) ToProto() *livekit.ParticipantInfo { info := &livekit.ParticipantInfo{ Sid: p.id, Name: p.name, - State: p.state, + State: p.State(), } p.lock.RLock() @@ -205,7 +207,7 @@ func (p *ParticipantImpl) OnClose(callback func(types.Participant)) { // Answer an offer from remote participant, used when clients make the initial connection func (p *ParticipantImpl) Answer(sdp webrtc.SessionDescription) (answer webrtc.SessionDescription, err error) { - if p.state != livekit.ParticipantInfo_JOINING && p.negotiationState != negotiationStateClient { + if p.State() != livekit.ParticipantInfo_JOINING && p.negotiationState != negotiationStateClient { // not in a valid state to continue err = ErrUnexpectedNegotiation return @@ -240,7 +242,7 @@ func (p *ParticipantImpl) Answer(sdp webrtc.SessionDescription) (answer webrtc.S }, }) - if p.state == livekit.ParticipantInfo_JOINING { + if p.State() == livekit.ParticipantInfo_JOINING { p.updateState(livekit.ParticipantInfo_JOINED) } return @@ -323,14 +325,15 @@ func (p *ParticipantImpl) Start() { func (p *ParticipantImpl) Close() error { if p.ctx.Err() != nil { - return p.ctx.Err() + // already closed + return nil } + p.updateState(livekit.ParticipantInfo_DISCONNECTED) p.onICECandidate = nil p.peerConn.OnDataChannel(nil) p.peerConn.OnICECandidate(nil) p.peerConn.OnNegotiationNeeded(nil) p.peerConn.OnTrack(nil) - p.updateState(livekit.ParticipantInfo_DISCONNECTED) p.responseSink.Close() if p.onClose != nil { p.onClose(p) @@ -382,7 +385,9 @@ func (p *ParticipantImpl) SendJoinResponse(roomInfo *livekit.Room, otherParticip } func (p *ParticipantImpl) SendParticipantUpdate(participants []*livekit.ParticipantInfo) error { - + if !p.IsReady() { + return nil + } return p.responseSink.WriteMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Update{ Update: &livekit.ParticipantUpdate{ @@ -403,8 +408,7 @@ func (p *ParticipantImpl) SetTrackMuted(trackId string, muted bool) { switch t := track.(type) { case *MediaTrack: - updated = t.muted != muted - t.muted = muted + updated = t.muted.TrySet(muted) } if updated && p.onTrackUpdated != nil { @@ -441,7 +445,7 @@ func (p *ParticipantImpl) scheduleNegotiate() { // initiates server-driven negotiation by creating an offer func (p *ParticipantImpl) negotiate() { - if p.state == livekit.ParticipantInfo_DISCONNECTED { + if p.State() == livekit.ParticipantInfo_DISCONNECTED { // skip when disconnected return } @@ -484,11 +488,11 @@ func (p *ParticipantImpl) negotiate() { } func (p *ParticipantImpl) updateState(state livekit.ParticipantInfo_State) { - if state == p.state { + oldState := p.State() + if state == oldState { return } - oldState := p.state - p.state = state + p.state.Store(state) logger.Debugw("updating participant state", "state", state.String(), "participant", p.ID()) if p.onStateChange != nil { go func() { diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index ceb59374b..b74d03a2f 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/pkg/rtc/participant_internal_test.go @@ -38,9 +38,8 @@ func TestIsReady(t *testing.T) { for _, test := range tests { t.Run(test.state.String(), func(t *testing.T) { - p := &ParticipantImpl{ - state: test.state, - } + p := &ParticipantImpl{} + p.state.Store(test.state) assert.Equal(t, test.ready, p.IsReady()) }) } diff --git a/pkg/rtc/room.go b/pkg/rtc/room.go index 45cab7f61..78fcd846b 100644 --- a/pkg/rtc/room.go +++ b/pkg/rtc/room.go @@ -8,6 +8,7 @@ import ( "github.com/livekit/livekit-server/pkg/logger" "github.com/livekit/livekit-server/pkg/rtc/types" + "github.com/livekit/livekit-server/pkg/utils" "github.com/livekit/livekit-server/proto/livekit" ) @@ -18,7 +19,7 @@ type Room struct { // map of participantId -> Participant participants map[string]types.Participant hasJoined bool - isClosed bool + isClosed utils.AtomicFlag onClose func() } @@ -44,7 +45,7 @@ func (r *Room) GetParticipants() []types.Participant { } func (r *Room) Join(participant types.Participant) error { - if r.isClosed { + if r.isClosed.Get() { return ErrRoomClosed } @@ -125,7 +126,7 @@ func (r *Room) RemoveParticipant(id string) { // Close the room if all participants had left, or it's still empty past timeout func (r *Room) CloseIfEmpty() { - if r.isClosed { + if r.isClosed.Get() { return } @@ -140,8 +141,7 @@ func (r *Room) CloseIfEmpty() { elapsed := uint32(time.Now().Unix() - r.CreationTime) logger.Infow("comparing elapsed", "elapsed", elapsed, "timeout", r.EmptyTimeout) if r.hasJoined || (r.EmptyTimeout > 0 && elapsed >= r.EmptyTimeout) { - r.isClosed = true - if r.onClose != nil { + if r.isClosed.TrySet(true) && r.onClose != nil { r.onClose() } }