From bf57064619928604f11d2435e258c3a26cc46eb8 Mon Sep 17 00:00:00 2001 From: Hugo Arregui Date: Tue, 14 May 2019 16:20:17 -0300 Subject: [PATCH] Improve nomination This implements a basic validation schema using a checklist. We try every pair at least maxTries, and mark it as failed if we don't get a success response after that many requests. Once we get a success response, we check if it belongs to the best candidate available so far, if it does we nominate it, otherwise we continue. Also, after a given timeout, if no candidate has been nominated, we simply choose the best valid candidate we got so far (if no candidate is valid, we mark the connection as failed). Finally, the nomination request also has a maximum of maxTries, we mark the connection as failed if after that many attempt we fail to get a success response. --- agent.go | 232 +++++++++++++++++++++++++++++++++------------- agent_test.go | 14 ++- candidatepair.go | 29 +++++- selection.go | 117 ++++++++++++++++++++--- transport_test.go | 9 +- 5 files changed, 311 insertions(+), 90 deletions(-) diff --git a/agent.go b/agent.go index a60ace3..c18e032 100644 --- a/agent.go +++ b/agent.go @@ -6,7 +6,6 @@ import ( "fmt" "math/rand" "net" - "sort" "sync" "sync/atomic" "time" @@ -26,6 +25,24 @@ const ( // defaultConnectionTimeout used to declare a connection dead defaultConnectionTimeout = 30 * time.Second + // timeout for candidate selection, after this time, the best candidate is used + defaultCandidateSelectionTimeout = 10 * time.Second + + // wait time before nominating a host candidate + defaultHostAcceptanceMinWait = 0 + + // wait time before nominating a srflx candidate + defaultSrflxAcceptanceMinWait = 500 * time.Millisecond + + // wait time before nominating a prflx candidate + defaultPrflxAcceptanceMinWait = 1000 * time.Millisecond + + // wait time before nominating a relay candidate + defaultRelayAcceptanceMinWait = 2000 * time.Millisecond + + // max binding request before considering a pair failed + defaultMaxBindingRequests = 7 + // the number of bytes that can be buffered before we start to error maxBufferSize = 1000 * 1000 // 1MB @@ -37,18 +54,6 @@ var ( defaultCandidateTypes = []CandidateType{CandidateTypeHost, CandidateTypeServerReflexive, CandidateTypeRelay} ) -type candidatePairs []*candidatePair - -func (cp candidatePairs) Len() int { return len(cp) } -func (cp candidatePairs) Swap(i, j int) { cp[i], cp[j] = cp[j], cp[i] } - -type byPairPriority struct{ candidatePairs } - -// NB: Reverse sort so our candidates start at highest priority -func (bp byPairPriority) Less(i, j int) bool { - return bp.candidatePairs[i].Priority() > bp.candidatePairs[j].Priority() -} - type bindingRequest struct { transactionID [stun.TransactionIDSize]byte destination net.Addr @@ -81,6 +86,14 @@ type Agent struct { haveStarted atomic.Value isControlling bool + maxBindingRequests uint16 + + candidateSelectionTimeout time.Duration + hostAcceptanceMinWait time.Duration + srflxAcceptanceMinWait time.Duration + prflxAcceptanceMinWait time.Duration + relayAcceptanceMinWait time.Duration + portmin uint16 portmax uint16 @@ -105,10 +118,9 @@ type Agent struct { remotePwd string remoteCandidates map[NetworkType][]Candidate + checklist []*candidatePair selector pairCandidateSelector selectedPair *candidatePair - validPairs candidatePairs - urls []*URL networkTypes []NetworkType @@ -177,6 +189,25 @@ type AgentConfig struct { // task loop handles things like sending keepAlives. This is only value for testing // keepAlive behavior should be modified with KeepaliveInterval and ConnectionTimeout taskLoopInterval time.Duration + + // MaxBindingRequests is the max amount of binding requests the agent will send + // over a candidate pair for validation or nomination, if after MaxBindingRequests + // the candidate is yet to answer a binding request or a nomination we set the pair as failed + MaxBindingRequests *uint16 + + // CandidatesSelectionTimeout specify a timeout for selecting candidates, if no nomination has happen + // before this timeout, once hit we will nominate the best valid candidate available, + // or mark the connection as failed if no valid candidate is available + CandidateSelectionTimeout *time.Duration + + // HostAcceptanceMinWait specify a minimum wait time before selecting host candidates + HostAcceptanceMinWait *time.Duration + // HostAcceptanceMinWait specify a minimum wait time before selecting srflx candidates + SrflxAcceptanceMinWait *time.Duration + // HostAcceptanceMinWait specify a minimum wait time before selecting prflx candidates + PrflxAcceptanceMinWait *time.Duration + // HostAcceptanceMinWait specify a minimum wait time before selecting relay candidates + RelayAcceptanceMinWait *time.Duration } // NewAgent creates a new Agent @@ -215,6 +246,42 @@ func NewAgent(config *AgentConfig) (*Agent, error) { } a.haveStarted.Store(false) + if config.MaxBindingRequests == nil { + a.maxBindingRequests = defaultMaxBindingRequests + } else { + a.maxBindingRequests = *config.MaxBindingRequests + } + + if config.CandidateSelectionTimeout == nil { + a.candidateSelectionTimeout = defaultCandidateSelectionTimeout + } else { + a.candidateSelectionTimeout = *config.CandidateSelectionTimeout + } + + if config.HostAcceptanceMinWait == nil { + a.hostAcceptanceMinWait = defaultHostAcceptanceMinWait + } else { + a.hostAcceptanceMinWait = *config.HostAcceptanceMinWait + } + + if config.SrflxAcceptanceMinWait == nil { + a.srflxAcceptanceMinWait = defaultSrflxAcceptanceMinWait + } else { + a.srflxAcceptanceMinWait = *config.SrflxAcceptanceMinWait + } + + if config.PrflxAcceptanceMinWait == nil { + a.prflxAcceptanceMinWait = defaultPrflxAcceptanceMinWait + } else { + a.prflxAcceptanceMinWait = *config.PrflxAcceptanceMinWait + } + + if config.RelayAcceptanceMinWait == nil { + a.relayAcceptanceMinWait = defaultRelayAcceptanceMinWait + } else { + a.relayAcceptanceMinWait = *config.RelayAcceptanceMinWait + } + // Make sure the buffer doesn't grow indefinitely. // NOTE: We actually won't get anywhere close to this limit. // SRTP will constantly read from the endpoint and drop packets if it's full. @@ -299,16 +366,30 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP a.log.Debugf("Started agent: isControlling? %t, remoteUfrag: %q, remotePwd: %q", isControlling, remoteUfrag, remotePwd) return a.run(func(agent *Agent) { + agent.isControlling = isControlling + agent.remoteUfrag = remoteUfrag + agent.remotePwd = remotePwd + + a.checklist = make([]*candidatePair, 0) + for networkType, localCandidates := range a.localCandidates { + if remoteCandidates, ok := a.remoteCandidates[networkType]; ok { + + for _, localCandidate := range localCandidates { + for _, remoteCandidate := range remoteCandidates { + a.addPair(localCandidate, remoteCandidate) + } + } + + } + } + if isControlling { a.selector = &controllingSelector{agent: a, log: a.log} } else { a.selector = &controlledSelector{agent: a, log: a.log} } - a.selector.Start() - agent.isControlling = isControlling - agent.remoteUfrag = remoteUfrag - agent.remotePwd = remotePwd + a.selector.Start() agent.updateConnectionState(ConnectionStateChecking) @@ -332,32 +413,6 @@ func (a *Agent) updateConnectionState(newState ConnectionState) { } } -func (a *Agent) findValidPair(local, remote Candidate) *candidatePair { - for _, p := range a.validPairs { - if p.local == local && p.remote == remote { - return p - } - } - return nil -} - -func (a *Agent) addValidPair(local, remote Candidate) *candidatePair { - p := a.findValidPair(local, remote) - if p != nil { - a.log.Tracef("Candidate pair is already valid: %s", p) - return p - } - - p = newCandidatePair(local, remote, a.isControlling) - a.log.Tracef("Found valid candidate pair: %s", p) - - // keep track of pairs with succesfull bindings since any of them - // can be used for communication until the final pair is selected: - // https://tools.ietf.org/html/draft-ietf-ice-rfc5245bis-20#section-12 - a.validPairs = append(a.validPairs, p) - return p -} - func (a *Agent) setSelectedPair(p *candidatePair) { a.log.Tracef("Set selected candidate pair: %s", p) // Notify when the selected pair changes @@ -370,12 +425,67 @@ func (a *Agent) setSelectedPair(p *candidatePair) { a.onConnectedOnce.Do(func() { close(a.onConnected) }) } -func (a *Agent) getBestValidPair() *candidatePair { - if len(a.validPairs) == 0 { - return nil +func (a *Agent) pingAllCandidates() { + for _, p := range a.checklist { + if p.state != candidatePairStateChecking { + continue + } + + if p.bindingRequestCount > a.maxBindingRequests { + a.log.Tracef("max requests reached for pair %s, marking it as failed\n", p) + p.state = candidatePairStateFailed + } else { + a.selector.PingCandidate(p.local, p.remote) + p.bindingRequestCount++ + } } - sort.Sort(byPairPriority{a.validPairs}) - return a.validPairs[0] +} + +func (a *Agent) getBestAvailableCandidatePair() *candidatePair { + var best *candidatePair + for _, p := range a.checklist { + if p.state == candidatePairStateFailed { + continue + } + + if best == nil { + best = p + } else if best.Priority() < p.Priority() { + best = p + } + } + return best +} + +func (a *Agent) getBestValidCandidatePair() *candidatePair { + var best *candidatePair + for _, p := range a.checklist { + if p.state != candidatePairStateValid { + continue + } + + if best == nil { + best = p + } else if best.Priority() < p.Priority() { + best = p + } + } + return best +} + +func (a *Agent) addPair(local, remote Candidate) *candidatePair { + p := newCandidatePair(local, remote, a.isControlling) + a.checklist = append(a.checklist, p) + return p +} + +func (a *Agent) findPair(local, remote Candidate) *candidatePair { + for _, p := range a.checklist { + if p.local == local && p.remote == remote { + return p + } + } + return nil } // A task is a @@ -455,22 +565,6 @@ func (a *Agent) checkKeepalive() { } } -// pingAllCandidates sends STUN Binding Requests to all candidates -// Note: the caller should hold the agent lock. -func (a *Agent) pingAllCandidates() { - for networkType, localCandidates := range a.localCandidates { - if remoteCandidates, ok := a.remoteCandidates[networkType]; ok { - - for _, localCandidate := range localCandidates { - for _, remoteCandidate := range remoteCandidates { - a.selector.PingCandidate(localCandidate, remoteCandidate) - } - } - - } - } -} - // AddRemoteCandidate adds a new remote candidate func (a *Agent) AddRemoteCandidate(c Candidate) error { return a.run(func(agent *Agent) { @@ -498,6 +592,12 @@ func (a *Agent) addRemoteCandidate(c Candidate) { } } } + + if localCandidates, ok := a.localCandidates[c.NetworkType()]; ok { + for _, localCandidate := range localCandidates { + a.addPair(localCandidate, c) + } + } } // GetLocalCandidates returns the local candidates diff --git a/agent_test.go b/agent_test.go index bc010c7..8e96244 100644 --- a/agent_test.go +++ b/agent_test.go @@ -35,11 +35,11 @@ func TestPairSearch(t *testing.T) { t.Fatalf("Error constructing ice.Agent") } - if len(a.validPairs) != 0 { + if len(a.checklist) != 0 { t.Fatalf("TestPairSearch is only a valid test if a.validPairs is empty on construction") } - cp := a.getBestValidPair() + cp := a.getBestAvailableCandidatePair() if cp != nil { t.Fatalf("No Candidate pairs should exist") @@ -110,8 +110,14 @@ func TestPairPriority(t *testing.T) { } for _, remote := range []Candidate{relayRemote, srflxRemote, prflxRemote, hostRemote} { - a.addValidPair(hostLocal, remote) - bestPair := a.getBestValidPair() + p := a.findPair(hostLocal, remote) + + if p == nil { + p = a.addPair(hostLocal, remote) + } + + p.state = candidatePairStateValid + bestPair := a.getBestValidCandidatePair() if bestPair.String() != (&candidatePair{remote: remote, local: hostLocal}).String() { t.Fatalf("Unexpected bestPair %s (expected remote: %s)", bestPair, remote) } diff --git a/candidatepair.go b/candidatepair.go index 82a9a24..a5af55f 100644 --- a/candidatepair.go +++ b/candidatepair.go @@ -6,19 +6,42 @@ import ( "github.com/pion/stun" ) +type candidatePairState int + +const ( + candidatePairStateChecking candidatePairState = iota + 1 + candidatePairStateFailed + candidatePairStateValid +) + +func (c candidatePairState) String() string { + switch c { + case candidatePairStateChecking: + return "checking" + case candidatePairStateFailed: + return "failed" + case candidatePairStateValid: + return "valid" + } + return "Unknown candidate pair state" +} + func newCandidatePair(local, remote Candidate, controlling bool) *candidatePair { return &candidatePair{ iceRoleControlling: controlling, remote: remote, local: local, + state: candidatePairStateChecking, } } // candidatePair represents a combination of a local and remote candidate type candidatePair struct { - iceRoleControlling bool - remote Candidate - local Candidate + iceRoleControlling bool + remote Candidate + local Candidate + bindingRequestCount uint16 + state candidatePairState } func (p *candidatePair) String() string { diff --git a/selection.go b/selection.go index 8cfa305..ebbb629 100644 --- a/selection.go +++ b/selection.go @@ -2,6 +2,7 @@ package ice import ( "net" + "time" "github.com/pion/logging" "github.com/pion/stun" @@ -16,12 +17,51 @@ type pairCandidateSelector interface { } type controllingSelector struct { - agent *Agent - nominatedPair *candidatePair - log logging.LeveledLogger + startTime time.Time + agent *Agent + nominatedPair *candidatePair + nominationRequestCount uint16 + log logging.LeveledLogger } func (s *controllingSelector) Start() { + s.startTime = time.Now() + go func() { + time.Sleep(s.agent.candidateSelectionTimeout) + err := s.agent.run(func(a *Agent) { + if s.nominatedPair == nil { + p := s.agent.getBestValidCandidatePair() + if p == nil { + s.log.Trace("check timeout reached and no valid candidate pair found, marking connection as failed") + s.agent.updateConnectionState(ConnectionStateFailed) + } else { + s.log.Tracef("check timeout reached, nominating (%s, %s)", p.local.String(), p.remote.String()) + s.nominatedPair = p + s.nominatePair(p) + } + } + }) + + if err != nil { + s.log.Errorf("error processing checkCandidatesTimeout handler %v", err.Error()) + } + }() +} + +func (s *controllingSelector) isNominatable(c Candidate) bool { + switch { + case c.Type() == CandidateTypeHost: + return time.Since(s.startTime).Nanoseconds() > s.agent.hostAcceptanceMinWait.Nanoseconds() + case c.Type() == CandidateTypeServerReflexive: + return time.Since(s.startTime).Nanoseconds() > s.agent.srflxAcceptanceMinWait.Nanoseconds() + case c.Type() == CandidateTypePeerReflexive: + return time.Since(s.startTime).Nanoseconds() > s.agent.prflxAcceptanceMinWait.Nanoseconds() + case c.Type() == CandidateTypeRelay: + return time.Since(s.startTime).Nanoseconds() > s.agent.relayAcceptanceMinWait.Nanoseconds() + } + + s.log.Errorf("isNominatable invalid candidate type %s", c.Type().String()) + return false } func (s *controllingSelector) ContactCandidates() { @@ -32,8 +72,21 @@ func (s *controllingSelector) ContactCandidates() { s.agent.checkKeepalive() } case s.nominatedPair != nil: + if s.nominationRequestCount > s.agent.maxBindingRequests { + s.log.Trace("max nomination requests reached, setting the connection state to failed") + s.agent.updateConnectionState(ConnectionStateFailed) + return + } s.nominatePair(s.nominatedPair) default: + p := s.agent.getBestValidCandidatePair() + if p != nil && s.isNominatable(p.local) && s.isNominatable(p.remote) { + s.log.Tracef("Nominatable pair found, nominating (%s, %s)", p.local.String(), p.remote.String()) + s.nominatedPair = p + s.nominatePair(p) + return + } + s.log.Trace("pinging all candidates") s.agent.pingAllCandidates() } @@ -60,15 +113,29 @@ func (s *controllingSelector) nominatePair(pair *candidatePair) { s.log.Tracef("ping STUN (nominate candidate pair) from %s to %s\n", pair.local.String(), pair.remote.String()) s.agent.sendBindingRequest(msg, pair.local, pair.remote) + s.nominationRequestCount++ } func (s *controllingSelector) HandleBindingRequest(m *stun.Message, local, remote Candidate) { s.agent.sendBindingSuccess(m, local, remote) - p := s.agent.findValidPair(local, remote) - if p != nil && s.nominatedPair == nil && s.agent.selectedPair == nil { - s.nominatedPair = p - s.nominatePair(p) + p := s.agent.findPair(local, remote) + + if p == nil { + s.agent.addPair(local, remote) + return + } + + if p.state == candidatePairStateValid && s.nominatedPair == nil && s.agent.selectedPair == nil { + bestPair := s.agent.getBestAvailableCandidatePair() + if bestPair == nil { + s.log.Tracef("No best pair available\n") + } else if bestPair.Equal(p) && s.isNominatable(p.local) && s.isNominatable(p.remote) { + s.log.Tracef("The candidate (%s, %s) is the best candidate available, marking it as nominated\n", + p.local.String(), p.remote.String()) + s.nominatedPair = p + s.nominatePair(p) + } } } @@ -89,8 +156,16 @@ func (s *controllingSelector) HandleSucessResponse(m *stun.Message, local, remot } s.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String()) - p := s.agent.addValidPair(local, remote) + p := s.agent.findPair(local, remote) + if p == nil { + // This shouldn't happen + s.log.Error("Success response from invalid candidate pair") + return + } + + p.state = candidatePairStateValid + s.log.Tracef("Found valid candidate pair: %s", p) if pendingRequest.isUseCandidate && s.agent.selectedPair == nil { s.agent.setSelectedPair(p) } @@ -173,15 +248,31 @@ func (s *controlledSelector) HandleSucessResponse(m *stun.Message, local, remote } s.log.Tracef("inbound STUN (SuccessResponse) from %s to %s", remote.String(), local.String()) - s.agent.addValidPair(local, remote) + + p := s.agent.findPair(local, remote) + if p == nil { + // This shouldn't happen + s.log.Error("Success response from invalid candidate pair") + return + } + + p.state = candidatePairStateValid + s.log.Tracef("Found valid candidate pair: %s", p) } func (s *controlledSelector) HandleBindingRequest(m *stun.Message, local, remote Candidate) { - if m.Contains(stun.AttrUseCandidate) { - // https://tools.ietf.org/html/rfc8445#section-7.3.1.5 - p := s.agent.findValidPair(local, remote) + useCandidate := m.Contains(stun.AttrUseCandidate) - if p != nil { + p := s.agent.findPair(local, remote) + + if p == nil { + p = s.agent.addPair(local, remote) + } + + if useCandidate { + // https://tools.ietf.org/html/rfc8445#section-7.3.1.5 + + if p.state == candidatePairStateValid { // If the state of this pair is Succeeded, it means that the check // previously sent by this pair produced a successful response and // generated a valid pair (Section 7.2.5.3.2). The agent sets the diff --git a/transport_test.go b/transport_test.go index 13dfad7..ca5e95f 100644 --- a/transport_test.go +++ b/transport_test.go @@ -334,10 +334,11 @@ func copyCandidate(o Candidate) Candidate { case *CandidateHost: return &CandidateHost{ candidateBase{ - networkType: orig.networkType, - ip: orig.ip, - port: orig.port, - component: orig.component, + candidateType: orig.candidateType, + networkType: orig.networkType, + ip: orig.ip, + port: orig.port, + component: orig.component, }, } default: