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: