Add round trip time measurement to candidate pair

Use the round trip time measurement to populate RTT fields in
CandidatePairStats.

Atomic and tests

agent.go

@@ -973,16 +973,16 @@ func (a *Agent) invalidatePendingBindingRequests(filterTime time.Time) {
 
 // Assert that the passed TransactionID is in our pendingBindingRequests and returns the destination
 // If the bindingRequest was valid remove it from our pending cache
-func (a *Agent) handleInboundBindingSuccess(id [stun.TransactionIDSize]byte) (bool, *bindingRequest) {
+func (a *Agent) handleInboundBindingSuccess(id [stun.TransactionIDSize]byte) (bool, *bindingRequest, time.Duration) {
  a.invalidatePendingBindingRequests(time.Now())
  for i := range a.pendingBindingRequests {
  if a.pendingBindingRequests[i].transactionID == id {
  validBindingRequest := a.pendingBindingRequests[i]
  a.pendingBindingRequests = append(a.pendingBindingRequests[:i], a.pendingBindingRequests[i+1:]...)
- return true, &validBindingRequest
+ return true, &validBindingRequest, time.Since(validBindingRequest.timestamp)
  }
  }
- return false, nil
+ return false, nil, 0
 }
 
 // handleInbound processes STUN traffic from a remote candidate

agent_stats.go

@@ -29,14 +29,14 @@ func (a *Agent) GetCandidatePairsStats() []CandidatePairStats {
  // FirstRequestTimestamp time.Time
  // LastRequestTimestamp time.Time
  // LastResponseTimestamp time.Time
- // TotalRoundTripTime float64
- // CurrentRoundTripTime float64
+ TotalRoundTripTime: cp.TotalRoundTripTime(),
+ CurrentRoundTripTime: cp.CurrentRoundTripTime(),
  // AvailableOutgoingBitrate float64
  // AvailableIncomingBitrate float64
  // CircuitBreakerTriggerCount uint32
  // RequestsReceived uint64
  // RequestsSent uint64
- // ResponsesReceived uint64
+ ResponsesReceived: cp.ResponsesReceived(),
  // ResponsesSent uint64
  // RetransmissionsReceived uint64
  // RetransmissionsSent uint64

agent_test.go

@@ -721,6 +721,10 @@ func TestCandidatePairStats(t *testing.T) {
  p := a.findPair(hostLocal, prflxRemote)
  p.state = CandidatePairStateFailed
 
+ for i := 0; i < 10; i++ {
+ p.UpdateRoundTripTime(time.Duration(i+1) * time.Second)
+ }
+
  stats := a.GetCandidatePairsStats()
  if len(stats) != 4 {
  t.Fatal("expected 4 candidate pairs stats")
@@ -766,6 +770,23 @@ func TestCandidatePairStats(t *testing.T) {
  t.Fatalf("expected host-prflx pair to have state failed, it has state %s instead",
  prflxPairStat.State.String())
  }
+
+ expectedCurrentRoundTripTime := time.Duration(10) * time.Second
+ if prflxPairStat.CurrentRoundTripTime != expectedCurrentRoundTripTime.Seconds() {
+ t.Fatalf("expected current round trip time to be %f, it is %f instead",
+ expectedCurrentRoundTripTime.Seconds(), prflxPairStat.CurrentRoundTripTime)
+ }
+
+ expectedTotalRoundTripTime := time.Duration(55) * time.Second
+ if prflxPairStat.TotalRoundTripTime != expectedTotalRoundTripTime.Seconds() {
+ t.Fatalf("expected total round trip time to be %f, it is %f instead",
+ expectedTotalRoundTripTime.Seconds(), prflxPairStat.TotalRoundTripTime)
+ }
+
+ if prflxPairStat.ResponsesReceived != 10 {
+ t.Fatalf("expected responses received to be 10, it is %d instead",
+ prflxPairStat.ResponsesReceived)
+ }
 }
 
 func TestLocalCandidateStats(t *testing.T) {

candidatepair.go

@@ -5,6 +5,8 @@ package ice
 
 import (
  "fmt"
+ "sync/atomic"
+ "time"
 
  "github.com/pion/stun/v3"
 )
@@ -28,6 +30,11 @@ type CandidatePair struct {
  state CandidatePairState
  nominated bool
  nominateOnBindingSuccess bool
+
+ // stats
+ currentRoundTripTime atomic.Pointer[time.Duration]
+ totalRoundTripTime atomic.Pointer[time.Duration]
+ responsesReceived uint64
 }
 
 func (p *CandidatePair) String() string {
@@ -100,3 +107,46 @@ func (a *Agent) sendSTUN(msg *stun.Message, local, remote Candidate) {
  a.log.Tracef("Failed to send STUN message: %s", err)
  }
 }
+
+// UpdateRoundTripTime sets the current round time of this pair and
+// accumulates total round trip time and responses received
+func (p *CandidatePair) UpdateRoundTripTime(rtt time.Duration) {
+ p.currentRoundTripTime.Store(&rtt)
+
+ prevTotalRoundTripTime := p.totalRoundTripTime.Load()
+ totalRoundTripTime := rtt
+ if prevTotalRoundTripTime != nil {
+ totalRoundTripTime += *prevTotalRoundTripTime
+ }
+ p.totalRoundTripTime.CompareAndSwap(prevTotalRoundTripTime, &totalRoundTripTime)
+
+ atomic.AddUint64(&p.responsesReceived, 1)
+}
+
+// CurrentRoundTripTime returns the current round trip time in seconds
+// https://www.w3.org/TR/webrtc-stats/#dom-rtcicecandidatepairstats-currentroundtriptime
+func (p *CandidatePair) CurrentRoundTripTime() float64 {
+ crtt := p.currentRoundTripTime.Load()
+ if crtt != nil {
+ return crtt.Seconds()
+ }
+
+ return 0
+}
+
+// TotalRoundTripTime returns the current round trip time in seconds
+// https://www.w3.org/TR/webrtc-stats/#dom-rtcicecandidatepairstats-totalroundtriptime
+func (p *CandidatePair) TotalRoundTripTime() float64 {
+ trtt := p.totalRoundTripTime.Load()
+ if trtt != nil {
+ return trtt.Seconds()
+ }
+
+ return 0
+}
+
+// ResponsesReceived returns the total number of connectivity responses received
+// https://www.w3.org/TR/webrtc-stats/#dom-rtcicecandidatepairstats-responsesreceived
+func (p *CandidatePair) ResponsesReceived() uint64 {
+ return atomic.LoadUint64(&p.responsesReceived)
+}

selection.go

@@ -120,7 +120,7 @@ func (s *controllingSelector) HandleBindingRequest(m *stun.Message, local, remot
 }
 
 func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remote Candidate, remoteAddr net.Addr) {
- ok, pendingRequest := s.agent.handleInboundBindingSuccess(m.TransactionID)
+ ok, pendingRequest, rtt := s.agent.handleInboundBindingSuccess(m.TransactionID)
  if !ok {
  s.log.Warnf("Discard message from (%s), unknown TransactionID 0x%x", remote, m.TransactionID)
  return
@@ -149,6 +149,8 @@ func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remo
  if pendingRequest.isUseCandidate && s.agent.getSelectedPair() == nil {
  s.agent.setSelectedPair(p)
  }
+
+ p.UpdateRoundTripTime(rtt)
 }
 
 func (s *controllingSelector) PingCandidate(local, remote Candidate) {
@@ -211,7 +213,7 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot
  // request with an appropriate error code response (e.g., 400)
  // [RFC5389].
 
- ok, pendingRequest := s.agent.handleInboundBindingSuccess(m.TransactionID)
+ ok, pendingRequest, rtt := s.agent.handleInboundBindingSuccess(m.TransactionID)
  if !ok {
  s.log.Warnf("Discard message from (%s), unknown TransactionID 0x%x", remote, m.TransactionID)
  return
@@ -245,6 +247,8 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot
  s.log.Tracef("Ignore nominate new pair %s, already nominated pair %s", p, selectedPair)
  }
  }
+
+ p.UpdateRoundTripTime(rtt)
 }
 
 func (s *controlledSelector) HandleBindingRequest(m *stun.Message, local, remote Candidate) {