Merge pull request #59 from matrix-org/track-related-improvements

More elegant handling of the RTCPs

Author: daniel-abramov

pkg/conference/data_channel_message_processor.go

@@ -19,7 +19,7 @@ func (c *Conference) processSelectDCMessage(participant *Participant, msg event.
 	if len(tracks) != len(msg.Start) {
 		for _, expected := range msg.Start {
 			found := slices.IndexFunc(tracks, func(track *webrtc.TrackLocalStaticRTP) bool {
-				return track.StreamID() == expected.StreamID && track.ID() == expected.TrackID
+				return track.ID() == expected.TrackID
 			})
 
 			if found == -1 {

pkg/conference/matrix_message_processor.go

@@ -65,7 +65,7 @@ func (c *Conference) onNewParticipant(participantID ParticipantID, inviteEvent *
 			logger:          logger,
 			remoteSessionID: inviteEvent.SenderSessionID,
 			streamMetadata:  inviteEvent.SDPStreamMetadata,
-			publishedTracks: make(map[event.SFUTrackDescription]PublishedTrack),
+			publishedTracks: make(map[string]PublishedTrack),
 		}
 
 		c.participants[participantID] = participant

pkg/conference/messsage_processor.go

@@ -63,7 +63,7 @@ func (c *Conference) processPeerMessage(message common.Message[ParticipantID, pe
 	case peer.DataChannelAvailable:
 		c.processDataChannelAvailableMessage(participant, msg)
 	case peer.RTCPReceived:
-		c.processForwardRTCPMessage(msg)
+		c.processRTCPPackets(msg)
 	default:
 		c.logger.Errorf("Unknown message type: %T", msg)
 	}

pkg/conference/participant.go

@@ -22,9 +22,9 @@ type ParticipantID struct {
 
 type PublishedTrack struct {
 	track *webrtc.TrackLocalStaticRTP
-	// The time when we sent the last PLI to the sender. We store this to avoid
-	// spamming the sender.
-	lastPLITimestamp time.Time
+	// The timestamp at which we are allowed to send the FIR or PLI request. We don't want to send them
+	// too often, so we introduce some trivial rate limiting to not "enforce" too many key frames.
+	canSendKeyframeAt time.Time
 }
 
 // Participant represents a participant in the conference.
@@ -34,7 +34,7 @@ type Participant struct {
 	peer            *peer.Peer[ParticipantID]
 	remoteSessionID id.SessionID
 	streamMetadata  event.CallSDPStreamMetadata
-	publishedTracks map[event.SFUTrackDescription]PublishedTrack
+	publishedTracks map[string]PublishedTrack
 }
 
 func (p *Participant) asMatrixRecipient() signaling.MatrixRecipient {

pkg/conference/peer_message_processor.go

@@ -21,26 +21,19 @@ func (c *Conference) processLeftTheCallMessage(participant *Participant, msg pee
 
 func (c *Conference) processNewTrackPublishedMessage(participant *Participant, msg peer.NewTrackPublished) {
 	participant.logger.Infof("Published new track: %s", msg.Track.ID())
-	key := event.SFUTrackDescription{
-		StreamID: msg.Track.StreamID(),
-		TrackID:  msg.Track.ID(),
-	}
 
-	if _, ok := participant.publishedTracks[key]; ok {
-		c.logger.Errorf("Track already published: %v", key)
+	if _, ok := participant.publishedTracks[msg.Track.ID()]; ok {
+		c.logger.Errorf("Track already published: %v", msg.Track.ID())
 		return
 	}
 
-	participant.publishedTracks[key] = PublishedTrack{track: msg.Track}
+	participant.publishedTracks[msg.Track.ID()] = PublishedTrack{track: msg.Track}
 	c.resendMetadataToAllExcept(participant.id)
 }
 
 func (c *Conference) processPublishedTrackFailedMessage(participant *Participant, msg peer.PublishedTrackFailed) {
 	participant.logger.Infof("Failed published track: %s", msg.Track.ID())
-	delete(participant.publishedTracks, event.SFUTrackDescription{
-		StreamID: msg.Track.StreamID(),
-		TrackID:  msg.Track.ID(),
-	})
+	delete(participant.publishedTracks, msg.Track.ID())
 
 	for _, otherParticipant := range c.participants {
 		if otherParticipant.id == participant.id {
@@ -114,13 +107,14 @@ func (c *Conference) processDataChannelAvailableMessage(participant *Participant
 	})
 }
 
-func (c *Conference) processForwardRTCPMessage(msg peer.RTCPReceived) {
+func (c *Conference) processRTCPPackets(msg peer.RTCPReceived) {
+	const sendKeyFrameInterval = 500 * time.Millisecond
+
 	for _, participant := range c.participants {
-		for _, publishedTrack := range participant.publishedTracks {
-			if publishedTrack.track.StreamID() == msg.StreamID && publishedTrack.track.ID() == msg.TrackID {
-				err := participant.peer.WriteRTCP(msg.Packets, msg.StreamID, msg.TrackID, publishedTrack.lastPLITimestamp)
-				if err == nil {
-					publishedTrack.lastPLITimestamp = time.Now()
+		if published, ok := participant.publishedTracks[msg.TrackID]; ok {
+			if published.canSendKeyframeAt.Before(time.Now()) {
+				if err := participant.peer.WriteRTCP(msg.TrackID, msg.Packets); err == nil {
+					published.canSendKeyframeAt = time.Now().Add(sendKeyFrameInterval)
 				}
 			}
 		}

pkg/conference/state.go

@@ -59,6 +59,7 @@ func (c *Conference) removeParticipant(participantID ParticipantID) {
 	for _, publishedTrack := range participant.publishedTracks {
 		obsoleteTracks = append(obsoleteTracks, publishedTrack.track)
 	}
+
 	for _, otherParticipant := range c.participants {
 		otherParticipant.peer.UnsubscribeFrom(obsoleteTracks)
 	}
@@ -98,7 +99,7 @@ func (c *Conference) getTracks(identifiers []event.SFUTrackDescription) []*webrt
 	for _, participant := range c.participants {
 		// Check if this participant has any of the tracks that we're looking for.
 		for _, identifier := range identifiers {
-			if track, ok := participant.publishedTracks[identifier]; ok {
+			if track, ok := participant.publishedTracks[identifier.TrackID]; ok {
 				tracks = append(tracks, track.track)
 			}
 		}

pkg/peer/messages.go

@@ -1,7 +1,6 @@
 package peer
 
 import (
-	"github.com/pion/rtcp"
 	"github.com/pion/webrtc/v3"
 	"maunium.net/go/mautrix/event"
 )
@@ -41,7 +40,13 @@ type DataChannelMessage struct {
 type DataChannelAvailable struct{}
 
 type RTCPReceived struct {
-	Packets  []rtcp.Packet
-	StreamID string
-	TrackID  string
+	TrackID string
+	Packets []RTCPPacketType
 }
+
+type RTCPPacketType int
+
+const (
+	PictureLossIndicator RTCPPacketType = iota + 1
+	FullIntraRequest
+)

pkg/peer/peer.go

@@ -23,7 +23,7 @@ var (
 	ErrDataChannelNotAvailable    = errors.New("data channel is not available")
 	ErrDataChannelNotReady        = errors.New("data channel is not ready")
 	ErrCantSubscribeToTrack       = errors.New("can't subscribe to track")
-	ErrCantWriteRTCP              = errors.New("can't write RTCP")
+	ErrTrackNotFound              = errors.New("track not found")
 )
 
 // A wrapped representation of the peer connection (single peer in the call).
@@ -106,71 +106,56 @@ func (p *Peer[ID]) SubscribeTo(track *webrtc.TrackLocalStaticRTP) error {
 			packets, _, err := rtpSender.ReadRTCP()
 			if err != nil {
 				if errors.Is(err, io.ErrClosedPipe) || errors.Is(err, io.EOF) {
+					p.logger.WithError(err).Warn("failed to read RTCP on track")
 					return
 				}
+			}
 
-				p.logger.WithError(err).Warn("failed to read RTCP on track")
+			// We only want to inform others about PLIs and FIRs. We skip the rest of the packets for now.
+			toForward := []RTCPPacketType{}
+			for _, packet := range packets {
+				// TODO: Should we also handle NACKs?
+				switch packet.(type) {
+				case *rtcp.PictureLossIndication:
+					toForward = append(toForward, PictureLossIndicator)
+				case *rtcp.FullIntraRequest:
+					toForward = append(toForward, FullIntraRequest)
+				}
 			}
 
-			p.sink.Send(RTCPReceived{Packets: packets, TrackID: track.ID(), StreamID: track.StreamID()})
+			p.sink.Send(RTCPReceived{Packets: toForward, TrackID: track.ID()})
 		}
 	}()
 
 	return nil
 }
 
-func (p *Peer[ID]) WriteRTCP(packets []rtcp.Packet, streamID string, trackID string, lastPLITimestamp time.Time) error {
-	const minimalPLIInterval = time.Millisecond * 500
-
-	packetsToSend := []rtcp.Packet{}
-	var mediaSSRC uint32
+// Writes the specified packets to the `trackID`.
+func (p *Peer[ID]) WriteRTCP(trackID string, packets []RTCPPacketType) error {
+	// Find the right track.
 	receivers := p.peerConnection.GetReceivers()
 	receiverIndex := slices.IndexFunc(receivers, func(receiver *webrtc.RTPReceiver) bool {
-		return receiver.Track().ID() == trackID && receiver.Track().StreamID() == streamID
+		return receiver.Track().ID() == trackID
 	})
-
 	if receiverIndex == -1 {
-		p.logger.Error("failed to find track to write RTCP on")
-		return ErrCantWriteRTCP
-	} else {
-		mediaSSRC = uint32(receivers[receiverIndex].Track().SSRC())
+		return ErrTrackNotFound
 	}
 
-	for _, packet := range packets {
-		switch typedPacket := packet.(type) {
-		// We mung the packets here, so that the SSRCs match what the
-		// receiver expects:
-		// The media SSRC is the SSRC of the media about which the packet is
-		// reporting; therefore, we mung it to be the SSRC of the publishing
-		// participant's track. Without this, it would be SSRC of the SFU's
-		// track which isn't right
-		case *rtcp.PictureLossIndication:
-			// Since we sometimes spam the sender with PLIs, make sure we don't send
-			// them way too often
-			if time.Now().UnixNano()-lastPLITimestamp.UnixNano() < minimalPLIInterval.Nanoseconds() {
-				continue
-			}
-
-			typedPacket.MediaSSRC = mediaSSRC
-			packetsToSend = append(packetsToSend, typedPacket)
-		case *rtcp.FullIntraRequest:
-			typedPacket.MediaSSRC = mediaSSRC
-			packetsToSend = append(packetsToSend, typedPacket)
-		}
-
-		packetsToSend = append(packetsToSend, packet)
-	}
-
-	if len(packetsToSend) != 0 {
-		if err := p.peerConnection.WriteRTCP(packetsToSend); err != nil {
-			if !errors.Is(err, io.ErrClosedPipe) {
-				p.logger.WithError(err).Error("failed to write RTCP on track")
-				return err
-			}
+	// The ssrc that we must use when sending the RTCP packet.
+	// Otherwise the peer won't understand where the packet comes from.
+	ssrc := uint32(receivers[receiverIndex].Track().SSRC())
+
+	toSend := make([]rtcp.Packet, len(packets))
+	for i, packet := range packets {
+		switch packet {
+		case PictureLossIndicator:
+			toSend[i] = &rtcp.PictureLossIndication{MediaSSRC: ssrc}
+		case FullIntraRequest:
+			toSend[i] = &rtcp.FullIntraRequest{MediaSSRC: ssrc}
 		}
 	}
 
-	return nil
+	return p.peerConnection.WriteRTCP(toSend)
 }
 
 // Unsubscribes from the given list of tracks.