From bc851c52845959aaebad808e77e0da612a1e1e9d Mon Sep 17 00:00:00 2001 From: Pascal Fischer Date: Fri, 15 Mar 2024 18:20:59 +0100 Subject: [PATCH 1/3] use binding requests to calculate latency for each candidate pair --- agent.go | 14 ++++++++++---- agent_handlers.go | 12 ++++++++++++ candidatepair.go | 21 +++++++++++++++++++++ selection.go | 6 ++++++ 4 files changed, 49 insertions(+), 4 deletions(-) diff --git a/agent.go b/agent.go index 571cd5b..a268382 100644 --- a/agent.go +++ b/agent.go @@ -15,8 +15,6 @@ import ( "sync/atomic" "time" - atomicx "github.com/pion/ice/v3/internal/atomic" - stunx "github.com/pion/ice/v3/internal/stun" "github.com/pion/logging" "github.com/pion/mdns" "github.com/pion/stun/v2" @@ -25,6 +23,9 @@ import ( "github.com/pion/transport/v3/stdnet" "github.com/pion/transport/v3/vnet" "golang.org/x/net/proxy" + + atomicx "github.com/pion/ice/v3/internal/atomic" + stunx "github.com/pion/ice/v3/internal/stun" ) type bindingRequest struct { @@ -43,6 +44,7 @@ type Agent struct { onConnectionStateChangeHdlr atomic.Value // func(ConnectionState) onSelectedCandidatePairChangeHdlr atomic.Value // func(Candidate, Candidate) onCandidateHdlr atomic.Value // func(Candidate) + onSuccessfulBindingResponseHdlr atomic.Value // func(Candidate) // State owned by the taskLoop onConnected chan struct{} @@ -648,7 +650,8 @@ func (a *Agent) checkKeepalive() { if (a.keepaliveInterval != 0) && ((time.Since(selectedPair.Local.LastSent()) > a.keepaliveInterval) || - (time.Since(selectedPair.Remote.LastReceived()) > a.keepaliveInterval)) { + (time.Since(selectedPair.Remote.LastReceived()) > a.keepaliveInterval) || + (time.Since(selectedPair.lastBindingRequest) > a.keepaliveInterval)) { // We use binding request instead of indication to support refresh consent schemas // see https://tools.ietf.org/html/rfc7675 a.selector.PingCandidate(selectedPair.Local, selectedPair.Remote) @@ -991,6 +994,9 @@ func (a *Agent) sendBindingRequest(m *stun.Message, local, remote Candidate) { isUseCandidate: m.Contains(stun.AttrUseCandidate), }) + p := a.findPair(local, remote) + p.markBindingRequest(m.TransactionID) + a.sendSTUN(m, local, remote) } @@ -1174,7 +1180,7 @@ func (a *Agent) GetSelectedCandidatePair() (*CandidatePair, error) { return nil, err } - return &CandidatePair{Local: local, Remote: remote}, nil + return &CandidatePair{Local: local, Remote: remote, latency: selectedPair.Latency()}, nil } func (a *Agent) getSelectedPair() *CandidatePair { diff --git a/agent_handlers.go b/agent_handlers.go index c5a5ec0..427b438 100644 --- a/agent_handlers.go +++ b/agent_handlers.go @@ -23,6 +23,12 @@ func (a *Agent) OnCandidate(f func(Candidate)) error { return nil } +// OnSuccessfulBindingResponse sets a handler that is fired when a successful binding response is received +func (a *Agent) OnSuccessfulBindingResponse(f func(*CandidatePair)) error { + a.onSuccessfulBindingResponseHdlr.Store(f) + return nil +} + func (a *Agent) onSelectedCandidatePairChange(p *CandidatePair) { if h, ok := a.onSelectedCandidatePairChangeHdlr.Load().(func(Candidate, Candidate)); ok { h(p.Local, p.Remote) @@ -41,6 +47,12 @@ func (a *Agent) onConnectionStateChange(s ConnectionState) { } } +func (a *Agent) onSuccessfulBindingResponse(p *CandidatePair) { + if h, ok := a.onSuccessfulBindingResponseHdlr.Load().(func(*CandidatePair)); ok { + h(p) + } +} + func (a *Agent) candidatePairRoutine() { for p := range a.chanCandidatePair { a.onSelectedCandidatePairChange(p) diff --git a/candidatepair.go b/candidatepair.go index 93470fe..4ebd56c 100644 --- a/candidatepair.go +++ b/candidatepair.go @@ -5,6 +5,7 @@ package ice import ( "fmt" + "time" "github.com/pion/stun/v2" ) @@ -24,6 +25,9 @@ type CandidatePair struct { iceRoleControlling bool Remote Candidate Local Candidate + latency time.Duration + lastBindingRequest time.Time + lastBindingTransactionID [12]byte bindingRequestCount uint16 state CandidatePairState nominated bool @@ -100,3 +104,20 @@ func (a *Agent) sendSTUN(msg *stun.Message, local, remote Candidate) { a.log.Tracef("Failed to send STUN message: %s", err) } } + +func (p *CandidatePair) markBindingRequest(transactionID [12]byte) { + p.lastBindingRequest = time.Now() + p.lastBindingTransactionID = transactionID +} + +func (p *CandidatePair) markBindingResponse(transactionID [12]byte) { + if p.lastBindingRequest.IsZero() || transactionID != p.lastBindingTransactionID { + return + } + + p.latency = time.Since(p.lastBindingRequest) +} + +func (p *CandidatePair) Latency() time.Duration { + return p.latency +} diff --git a/selection.go b/selection.go index e6e1fac..53b4b36 100644 --- a/selection.go +++ b/selection.go @@ -139,6 +139,9 @@ func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remo return } + p.markBindingResponse(m.TransactionID) + s.agent.onSuccessfulBindingResponse(p) + p.state = CandidatePairStateSucceeded s.log.Tracef("Found valid candidate pair: %s", p) if pendingRequest.isUseCandidate && s.agent.getSelectedPair() == nil { @@ -230,6 +233,9 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot return } + p.markBindingResponse(m.TransactionID) + s.agent.onSuccessfulBindingResponse(p) + p.state = CandidatePairStateSucceeded s.log.Tracef("Found valid candidate pair: %s", p) if p.nominateOnBindingSuccess { From e72a50fcb64e1c8d9a127e7e91337c43e07700cd Mon Sep 17 00:00:00 2001 From: Pascal Fischer Date: Fri, 15 Mar 2024 18:46:35 +0100 Subject: [PATCH 2/3] fire callback method only on selected pair binding --- agent.go | 8 ++++---- agent_handlers.go | 10 +++++----- candidatepair.go | 5 +++-- selection.go | 14 +++++++++----- 4 files changed, 21 insertions(+), 16 deletions(-) diff --git a/agent.go b/agent.go index a268382..d30281f 100644 --- a/agent.go +++ b/agent.go @@ -41,10 +41,10 @@ type Agent struct { afterRunFn []func(ctx context.Context) muAfterRun sync.Mutex - onConnectionStateChangeHdlr atomic.Value // func(ConnectionState) - onSelectedCandidatePairChangeHdlr atomic.Value // func(Candidate, Candidate) - onCandidateHdlr atomic.Value // func(Candidate) - onSuccessfulBindingResponseHdlr atomic.Value // func(Candidate) + onConnectionStateChangeHdlr atomic.Value // func(ConnectionState) + onSelectedCandidatePairChangeHdlr atomic.Value // func(Candidate, Candidate) + onCandidateHdlr atomic.Value // func(Candidate) + onSuccessfulSelectedPairBindingResponseHdlr atomic.Value // func(*Candidate) // State owned by the taskLoop onConnected chan struct{} diff --git a/agent_handlers.go b/agent_handlers.go index 427b438..7ceb6e7 100644 --- a/agent_handlers.go +++ b/agent_handlers.go @@ -23,9 +23,9 @@ func (a *Agent) OnCandidate(f func(Candidate)) error { return nil } -// OnSuccessfulBindingResponse sets a handler that is fired when a successful binding response is received -func (a *Agent) OnSuccessfulBindingResponse(f func(*CandidatePair)) error { - a.onSuccessfulBindingResponseHdlr.Store(f) +// OnSuccessfulSelectedPairBindingResponse sets a handler that is fired when a successful binding response is received for the selected candidate pair +func (a *Agent) OnSuccessfulSelectedPairBindingResponse(f func(*CandidatePair)) error { + a.onSuccessfulSelectedPairBindingResponseHdlr.Store(f) return nil } @@ -47,8 +47,8 @@ func (a *Agent) onConnectionStateChange(s ConnectionState) { } } -func (a *Agent) onSuccessfulBindingResponse(p *CandidatePair) { - if h, ok := a.onSuccessfulBindingResponseHdlr.Load().(func(*CandidatePair)); ok { +func (a *Agent) onSuccessfulSelectedPairBindingResponse(p *CandidatePair) { + if h, ok := a.onSuccessfulSelectedPairBindingResponseHdlr.Load().(func(*CandidatePair)); ok { h(p) } } diff --git a/candidatepair.go b/candidatepair.go index 4ebd56c..fd36366 100644 --- a/candidatepair.go +++ b/candidatepair.go @@ -110,12 +110,13 @@ func (p *CandidatePair) markBindingRequest(transactionID [12]byte) { p.lastBindingTransactionID = transactionID } -func (p *CandidatePair) markBindingResponse(transactionID [12]byte) { +func (p *CandidatePair) markBindingResponse(transactionID [12]byte) bool { if p.lastBindingRequest.IsZero() || transactionID != p.lastBindingTransactionID { - return + return false } p.latency = time.Since(p.lastBindingRequest) + return true } func (p *CandidatePair) Latency() time.Duration { diff --git a/selection.go b/selection.go index 53b4b36..bfbd5a9 100644 --- a/selection.go +++ b/selection.go @@ -139,14 +139,16 @@ func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remo return } - p.markBindingResponse(m.TransactionID) - s.agent.onSuccessfulBindingResponse(p) - p.state = CandidatePairStateSucceeded s.log.Tracef("Found valid candidate pair: %s", p) if pendingRequest.isUseCandidate && s.agent.getSelectedPair() == nil { s.agent.setSelectedPair(p) } + + ok = p.markBindingResponse(m.TransactionID) + if ok && s.agent.getSelectedPair() == p { + s.agent.onSuccessfulSelectedPairBindingResponse(p) + } } func (s *controllingSelector) PingCandidate(local, remote Candidate) { @@ -233,8 +235,10 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot return } - p.markBindingResponse(m.TransactionID) - s.agent.onSuccessfulBindingResponse(p) + ok = p.markBindingResponse(m.TransactionID) + if ok { + s.agent.onSuccessfulSelectedPairBindingResponse(p) + } p.state = CandidatePairStateSucceeded s.log.Tracef("Found valid candidate pair: %s", p) From 3be7f17602596edcea470093336c37fd4e1070c8 Mon Sep 17 00:00:00 2001 From: Pascal Fischer Date: Tue, 19 Mar 2024 10:13:22 +0100 Subject: [PATCH 3/3] add TransaktionID type --- candidatepair.go | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/candidatepair.go b/candidatepair.go index fd36366..7dc3885 100644 --- a/candidatepair.go +++ b/candidatepair.go @@ -19,6 +19,8 @@ func newCandidatePair(local, remote Candidate, controlling bool) *CandidatePair } } +type TransactionID [stun.TransactionIDSize]byte + // CandidatePair is a combination of a // local and remote candidate type CandidatePair struct { @@ -27,7 +29,7 @@ type CandidatePair struct { Local Candidate latency time.Duration lastBindingRequest time.Time - lastBindingTransactionID [12]byte + lastBindingTransactionID TransactionID bindingRequestCount uint16 state CandidatePairState nominated bool @@ -105,12 +107,12 @@ func (a *Agent) sendSTUN(msg *stun.Message, local, remote Candidate) { } } -func (p *CandidatePair) markBindingRequest(transactionID [12]byte) { +func (p *CandidatePair) markBindingRequest(transactionID TransactionID) { p.lastBindingRequest = time.Now() p.lastBindingTransactionID = transactionID } -func (p *CandidatePair) markBindingResponse(transactionID [12]byte) bool { +func (p *CandidatePair) markBindingResponse(transactionID TransactionID) bool { if p.lastBindingRequest.IsZero() || transactionID != p.lastBindingTransactionID { return false }