dandri/livekit
1
0
Fork 0
mirror of https://github.com/livekit/livekit.git synced 2026-08-05 06:29:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add option to issue full reconnect on a publication error. (#1214)

Browse Source 
* Add option to issue full reconnect on a publication error.

Leaving the publication error timeout at 30 seconds as there
are some publications taking long. Also, there are cases
where the peer connection fails after 30 seconds. The peer
connection failure happens after publication error is detected.
But, 30 seconds is a good amount of time for publication to establish.

* prevent recursive lock
This commit is contained in:
Raja Subramanian
2022-12-06 14:46:59 +05:30
committed by GitHub
parent 6f27d683a9
commit 6bd5504bff
6 changed files with 90 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/config/config.go
+3
View File
@@ -85,6 +85,9 @@ type RTCConfig struct {
// for testing, disable UDP
ForceTCP bool `yaml:"force_tcp,omitempty"`
// force a reconnect on a publication error
ReconnectOnPublicationError *bool `yaml:"reconnect_on_publication_error,omitempty"`
}
type TURNServer struct {
pkg/config/config_test.go
+7 -4
View File
@@ -42,10 +42,11 @@ func TestGeneratedFlags(t *testing.T) {
app.Flags = append(app.Flags, generatedFlags...)
set := flag.NewFlagSet("test", 0)
set.Bool("rtc.use_ice_lite", true, "") // bool
set.String("redis.address", "localhost:6379", "") // string
set.Uint("prometheus_port", 9999, "") // uint32
set.Bool("rtc.allow_tcp_fallback", true, "") // pointer
set.Bool("rtc.use_ice_lite", true, "") // bool
set.String("redis.address", "localhost:6379", "") // string
set.Uint("prometheus_port", 9999, "") // uint32
set.Bool("rtc.allow_tcp_fallback", true, "") // pointer
set.Bool("rtc.reconnect_on_publication_error", true, "") // pointer
c := cli.NewContext(app, set, nil)
conf, err := NewConfig("", true, c, nil)
@@ -56,4 +57,6 @@ func TestGeneratedFlags(t *testing.T) {
require.Equal(t, uint32(9999), conf.PrometheusPort)
require.NotNil(t, conf.RTC.AllowTCPFallback)
require.True(t, *conf.RTC.AllowTCPFallback)
require.NotNil(t, conf.RTC.ReconnectOnPublicationError)
require.True(t, *conf.RTC.ReconnectOnPublicationError)
}
pkg/rtc/participant.go
+44 -29
View File
@@ -64,30 +64,31 @@ type SubscribeRequest struct {
}
type ParticipantParams struct {
Identity livekit.ParticipantIdentity
Name livekit.ParticipantName
SID livekit.ParticipantID
Config *WebRTCConfig
Sink routing.MessageSink
AudioConfig config.AudioConfig
VideoConfig config.VideoConfig
ProtocolVersion types.ProtocolVersion
Telemetry telemetry.TelemetryService
PLIThrottleConfig config.PLIThrottleConfig
CongestionControlConfig config.CongestionControlConfig
EnabledCodecs []*livekit.Codec
Logger logger.Logger
SimTracks map[uint32]SimulcastTrackInfo
Grants *auth.ClaimGrants
InitialVersion uint32
ClientConf *livekit.ClientConfiguration
ClientInfo ClientInfo
Region string
Migration bool
AdaptiveStream bool
AllowTCPFallback bool
TURNSEnabled bool
GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo
Identity livekit.ParticipantIdentity
Name livekit.ParticipantName
SID livekit.ParticipantID
Config *WebRTCConfig
Sink routing.MessageSink
AudioConfig config.AudioConfig
VideoConfig config.VideoConfig
ProtocolVersion types.ProtocolVersion
Telemetry telemetry.TelemetryService
PLIThrottleConfig config.PLIThrottleConfig
CongestionControlConfig config.CongestionControlConfig
EnabledCodecs []*livekit.Codec
Logger logger.Logger
SimTracks map[uint32]SimulcastTrackInfo
Grants *auth.ClaimGrants
InitialVersion uint32
ClientConf *livekit.ClientConfiguration
ClientInfo ClientInfo
Region string
Migration bool
AdaptiveStream bool
AllowTCPFallback bool
TURNSEnabled bool
GetParticipantInfo func(pID livekit.ParticipantID) *livekit.ParticipantInfo
ReconnectOnPublicationError bool
}
type ParticipantImpl struct {
@@ -198,6 +199,8 @@ func NewParticipant(params ParticipantParams) (*ParticipantImpl, error) {
p.grants = params.Grants
p.SetResponseSink(params.Sink)
p.supervisor.OnPublicationError(p.onPublicationError)
var err error
// keep last participants and when updates were sent
if p.updateCache, err = lru.New(128); err != nil {
@@ -1484,7 +1487,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l
} else {
p.pendingTracks[req.Cid].trackInfos = append(p.pendingTracks[req.Cid].trackInfos, ti)
}
p.params.Logger.Debugw("pending track queued", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
p.params.Logger.Infow("pending track queued", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
return nil
}
@@ -1492,7 +1495,7 @@ func (p *ParticipantImpl) addPendingTrackLocked(req *livekit.AddTrackRequest) *l
p.supervisor.SetPublicationMute(livekit.TrackID(ti.Sid), ti.Muted)
p.pendingTracks[req.Cid] = &pendingTrackInfo{trackInfos: []*livekit.TrackInfo{ti}}
p.params.Logger.Debugw("pending track added", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
p.params.Logger.Infow("pending track added", "trackID", ti.Sid, "track", ti.String(), "request", req.String())
return ti
}
@@ -1947,13 +1950,12 @@ func (p *ParticipantImpl) GetCachedDownTrack(trackID livekit.TrackID) (*webrtc.R
return nil, sfu.DownTrackState{}
}
func (p *ParticipantImpl) onAnyTransportNegotiationFailed() {
p.params.Logger.Infow("negotiation failed, starting full reconnect")
func (p *ParticipantImpl) issueFullReconnect(reason types.ParticipantCloseReason) {
_ = p.writeMessage(&livekit.SignalResponse{
Message: &livekit.SignalResponse_Leave{
Leave: &livekit.LeaveRequest{
CanReconnect: true,
Reason: types.ParticipantCloseReasonNegotiateFailed.ToDisconnectReason(),
Reason: reason.ToDisconnectReason(),
},
},
})
@@ -1963,6 +1965,19 @@ func (p *ParticipantImpl) onAnyTransportNegotiationFailed() {
p.supervisor.Stop()
}
func (p *ParticipantImpl) onPublicationError(trackID livekit.TrackID) {
p.params.Logger.Infow("publication failed", "trackID", trackID)
if p.params.ReconnectOnPublicationError {
p.params.Logger.Infow("starting full reconnect")
p.issueFullReconnect(types.ParticipantCloseReasonPublicationError)
}
}
func (p *ParticipantImpl) onAnyTransportNegotiationFailed() {
p.params.Logger.Infow("negotiation failed, starting full reconnect")
p.issueFullReconnect(types.ParticipantCloseReasonNegotiateFailed)
}
func (p *ParticipantImpl) EnqueueSubscribeTrack(trackID livekit.TrackID, isRelayed bool, f func(sub types.LocalParticipant) error) bool {
// do not queue subscription is participant is already closed/disconnected
if p.isClosed.Load() || p.State() == livekit.ParticipantInfo_DISCONNECTED {
pkg/rtc/supervisor/participant_supervisor.go
+24
View File
@@ -27,6 +27,8 @@ type ParticipantSupervisor struct {
subscriptions map[livekit.TrackID]types.OperationMonitor
isStopped atomic.Bool
onPublicationError func(trackID livekit.TrackID)
}
func NewParticipantSupervisor(params ParticipantSupervisorParams) *ParticipantSupervisor {
@@ -45,6 +47,20 @@ func (p *ParticipantSupervisor) Stop() {
p.isStopped.Store(true)
}
func (p *ParticipantSupervisor) OnPublicationError(f func(trackID livekit.TrackID)) {
p.lock.Lock()
defer p.lock.Unlock()
p.onPublicationError = f
}
func (p *ParticipantSupervisor) getOnPublicationError() func(trackID livekit.TrackID) {
p.lock.RLock()
defer p.lock.RUnlock()
return p.onPublicationError
}
func (p *ParticipantSupervisor) SetPublisherPeerConnectionConnected(isConnected bool) {
p.lock.Lock()
p.isPublisherConnected = isConnected
@@ -141,11 +157,13 @@ func (p *ParticipantSupervisor) checkState() {
}
func (p *ParticipantSupervisor) checkPublications() {
var erroredPublications []livekit.TrackID
var removablePublications []livekit.TrackID
p.lock.RLock()
for trackID, pm := range p.publications {
if err := pm.Check(); err != nil {
p.params.Logger.Errorw("supervisor error on publication", err, "trackID", trackID)
erroredPublications = append(erroredPublications, trackID)
} else {
if pm.IsIdle() {
removablePublications = append(removablePublications, trackID)
@@ -159,6 +177,12 @@ func (p *ParticipantSupervisor) checkPublications() {
delete(p.publications, trackID)
}
p.lock.Unlock()
if onPublicationError := p.getOnPublicationError(); onPublicationError != nil {
for _, trackID := range erroredPublications {
onPublicationError(trackID)
}
}
}
func (p *ParticipantSupervisor) checkSubscriptions() {
pkg/rtc/types/interfaces.go
+6 -3
View File
@@ -84,6 +84,7 @@ const (
ParticipantCloseReasonNegotiateFailed
ParticipantCloseReasonMigrationRequested
ParticipantCloseReasonOvercommitted
ParticipantCloseReasonPublicationError
)
func (p ParticipantCloseReason) String() string {
@@ -122,10 +123,12 @@ func (p ParticipantCloseReason) String() string {
return "SIMULATE_SERVER_LEAVE"
case ParticipantCloseReasonNegotiateFailed:
return "NEGOTIATE_FAILED"
case ParticipantCloseReasonOvercommitted:
return "OVERCOMMITTED"
case ParticipantCloseReasonMigrationRequested:
return "MIGRATION_REQUESTED"
case ParticipantCloseReasonOvercommitted:
return "OVERCOMMITTED"
case ParticipantCloseReasonPublicationError:
return "PUBLICATION_ERROR"
default:
return fmt.Sprintf("%d", int(p))
}
@@ -156,7 +159,7 @@ func (p ParticipantCloseReason) ToDisconnectReason() livekit.DisconnectReason {
return livekit.DisconnectReason_SERVER_SHUTDOWN
case ParticipantCloseReasonOvercommitted:
return livekit.DisconnectReason_SERVER_SHUTDOWN
case ParticipantCloseReasonNegotiateFailed:
case ParticipantCloseReasonNegotiateFailed, ParticipantCloseReasonPublicationError:
return livekit.DisconnectReason_STATE_MISMATCH
default:
// the other types will map to unknown reason
pkg/service/roommanager.go
+6
View File
@@ -280,6 +280,11 @@ func (r *RoomManager) StartSession(
if r.config.RTC.AllowTCPFallback != nil {
allowFallback = *r.config.RTC.AllowTCPFallback
}
// default do not force full reconnect on a publication error
reconnectOnPublicationError := false
if r.config.RTC.ReconnectOnPublicationError != nil {
reconnectOnPublicationError = *r.config.RTC.ReconnectOnPublicationError
}
participant, err = rtc.NewParticipant(rtc.ParticipantParams{
Identity: pi.Identity,
Name: pi.Name,
@@ -307,6 +312,7 @@ func (r *RoomManager) StartSession(
}
return nil
},
ReconnectOnPublicationError: reconnectOnPublicationError,
})
if err != nil {
return err