fix synchronization issues, use AtomicFlag/atomic.Value for shared fields

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
 }
 

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
 		}
 

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),

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() {

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())
 		})
 	}

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()
 		}
 	}