Merge pull request #1 from netbirdio/feature/add-latency

Add latency calculation to binding requests

Author: pascal-fischer

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 {
@@ -40,9 +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)
+	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{}
@@ -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 {

agent_handlers.go

@@ -23,6 +23,12 @@ func (a *Agent) OnCandidate(f func(Candidate)) error {
 	return nil
 }
 
+// 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
+}
+
 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) onSuccessfulSelectedPairBindingResponse(p *CandidatePair) {
+	if h, ok := a.onSuccessfulSelectedPairBindingResponseHdlr.Load().(func(*CandidatePair)); ok {
+		h(p)
+	}
+}
+
 func (a *Agent) candidatePairRoutine() {
 	for p := range a.chanCandidatePair {
 		a.onSelectedCandidatePairChange(p)

candidatepair.go

@@ -5,6 +5,7 @@ package ice
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/pion/stun/v2"
 )
@@ -18,12 +19,17 @@ 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 {
 	iceRoleControlling       bool
 	Remote                   Candidate
 	Local                    Candidate
+	latency                  time.Duration
+	lastBindingRequest       time.Time
+	lastBindingTransactionID TransactionID
 	bindingRequestCount      uint16
 	state                    CandidatePairState
 	nominated                bool
@@ -100,3 +106,21 @@ 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 TransactionID) {
+	p.lastBindingRequest = time.Now()
+	p.lastBindingTransactionID = transactionID
+}
+
+func (p *CandidatePair) markBindingResponse(transactionID TransactionID) bool {
+	if p.lastBindingRequest.IsZero() || transactionID != p.lastBindingTransactionID {
+		return false
+	}
+
+	p.latency = time.Since(p.lastBindingRequest)
+	return true
+}
+
+func (p *CandidatePair) Latency() time.Duration {
+	return p.latency
+}

selection.go

@@ -144,6 +144,11 @@ func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remo
 	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) {
@@ -230,6 +235,11 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot
 		return
 	}
 
+	ok = p.markBindingResponse(m.TransactionID)
+	if ok {
+		s.agent.onSuccessfulSelectedPairBindingResponse(p)
+	}
+
 	p.state = CandidatePairStateSucceeded
 	s.log.Tracef("Found valid candidate pair: %s", p)
 	if p.nominateOnBindingSuccess {