Repository: pion/ice (go)

Issue:
Slowdown because of server reflexive address allocation
### What happened?
Between v0.2.7 and v0.2.8 there was a change in the way server reflexive addresses are allocated. Rather than collecting only one (default) address, the gathering process now actually goes through all of the local interfaces which takes a significant load time.

Here is the part that causes this:
https://github.com/pion/ice/blob/v0.2.8/gather.go#L131


### Proposal
Create an asynchronous process that dispatches a goroutine to collect the server reflexive addresses but only waits for one (default one) to come back before continuing with the execution.

### Requirements

- [x] OnIceCandidate must be called when new candidate is discovered
- [ ] there must be a method for adding new remote candidates (not sure of the method name)
- [x] there should be an option for disabling trickle alltogether and waiting for all candidates

Pull request:
Add trickle support
#### Reference issue
Resolves #51


Patch:
diff --git a/agent.go b/agent.go
index ba66be2..e2d56c1 100644
--- a/agent.go
+++ b/agent.go
@@ -55,6 +55,7 @@ type bindingRequest struct {
 type Agent struct {
 	onConnectionStateChangeHdlr       func(ConnectionState)
 	onSelectedCandidatePairChangeHdlr func(Candidate, Candidate)
+	onCandidateHdlr                   func(Candidate)
 
 	// Used to block double Dial/Accept
 	opened bool
@@ -68,6 +69,7 @@ type Agent struct {
 	// force candidate to be contacted immediately (instead of waiting for connectivityChan)
 	forceCandidateContact chan bool
 
+	trickle         bool
 	tieBreaker      uint64
 	connectionState ConnectionState
 	gatheringState  GatheringState
@@ -78,12 +80,12 @@ type Agent struct {
 	portmin uint16
 	portmax uint16
 
-	//How long should a pair stay quiet before we declare it dead?
-	//0 means never timeout
+	// How long should a pair stay quiet before we declare it dead?
+	// 0 means never timeout
 	connectionTimeout time.Duration
 
-	//How often should we send keepalive packets?
-	//0 means never
+	// How often should we send keepalive packets?
+	// 0 means never
 	keepaliveInterval time.Duration
 
 	// How after should we run our internal taskLoop
@@ -101,6 +103,9 @@ type Agent struct {
 	selectedPair *candidatePair
 	validPairs   candidatePairs
 
+	urls         []*URL
+	networkTypes []NetworkType
+
 	buffer *packetio.Buffer
 
 	// LRU of outbound Binding request Transaction IDs
@@ -139,6 +144,10 @@ type AgentConfig struct {
 	PortMin uint16
 	PortMax uint16
 
+	// Trickle specifies whether or not ice agent should trickle candidates or
+	// work perform synchronous gathering.
+	Trickle bool
+
 	// ConnectionTimeout defaults to 30 seconds when this property is nil.
 	// If the duration is 0, we will never timeout this connection.
 	ConnectionTimeout *time.Duration
@@ -177,11 +186,13 @@ func NewAgent(config *AgentConfig) (*Agent, error) {
 
 	a := &Agent{
 		tieBreaker:             rand.New(rand.NewSource(time.Now().UnixNano())).Uint64(),
-		gatheringState:         GatheringStateComplete, // TODO trickle-ice
+		gatheringState:         GatheringStateNew,
 		connectionState:        ConnectionStateNew,
 		localCandidates:        make(map[NetworkType][]Candidate),
 		remoteCandidates:       make(map[NetworkType][]Candidate),
 		pendingBindingRequests: make([]bindingRequest, 0, maxPendingBindingRequests),
+		urls:                   config.Urls,
+		networkTypes:           config.NetworkTypes,
 
 		localUfrag:  randSeq(16),
 		localPwd:    randSeq(32),
@@ -191,6 +202,7 @@ func NewAgent(config *AgentConfig) (*Agent, error) {
 		done:        make(chan struct{}),
 		portmin:     config.PortMin,
 		portmax:     config.PortMax,
+		trickle:     config.Trickle,
 		log:         loggerFactory.NewLogger("ice"),
 
 		forceCandidateContact: make(chan bool, 1),
@@ -221,12 +233,12 @@ func NewAgent(config *AgentConfig) (*Agent, error) {
 		a.taskLoopInterval = config.taskLoopInterval
 	}
 
-	// Initialize local candidates
-	gatherCandidatesLocal(a, config.NetworkTypes)
-	gatherCandidatesReflective(a, config.Urls, config.NetworkTypes)
-
 	go a.taskLoop()
 
+	// Initialize local candidates
+	if !a.trickle {
+		a.gatherCandidates()
+	}
 	return a, nil
 }
 
@@ -245,6 +257,14 @@ func (a *Agent) OnSelectedCandidatePairChange(f func(Candidate, Candidate)) erro
 	})
 }
 
+// OnCandidate sets a handler that is fired when new candidates gathered. When
+// the gathering process complete the last candidate is nil.
+func (a *Agent) OnCandidate(f func(Candidate)) error {
+	return a.run(func(agent *Agent) {
+		agent.onCandidateHdlr = f
+	})
+}
+
 func (a *Agent) onSelectedCandidatePairChange(p *candidatePair) {
 	if p != nil {
 		if a.onSelectedCandidatePairChangeHdlr != nil {
diff --git a/errors.go b/errors.go
index 96a850a..e39f4ab 100644
--- a/errors.go
+++ b/errors.go
@@ -42,6 +42,13 @@ var (
 	// ErrRemotePwdEmpty indicates agent was started with an empty remote pwd
 	ErrRemotePwdEmpty = errors.New("remote pwd is empty")
 
+	// ErrNoOnCandidateHandler indicates agent was started without OnCandidate
+	// while running in trickle mode.
+	ErrNoOnCandidateHandler = errors.New("no OnCandidate provided")
+
+	// ErrMultipleGatherAttempted indicates GatherCandidates has been called multiple times
+	ErrMultipleGatherAttempted = errors.New("attempting to gather candidates during gathering state")
+
 	// ErrUsernameEmpty indicates agent was give TURN URL with an empty Username
 	ErrUsernameEmpty = errors.New("username is empty")
 
diff --git a/gather.go b/gather.go
index 32bb484..3efec15 100644
--- a/gather.go
+++ b/gather.go
@@ -90,7 +90,55 @@ func listenUDP(portMax, portMin int, network string, laddr *net.UDPAddr) (*net.U
 	return nil, ErrPort
 }
 
-func gatherCandidatesLocal(a *Agent, networkTypes []NetworkType) {
+// GatherCandidates initiates the trickle based gathering process.
+func (a *Agent) GatherCandidates() error {
+	gatherErrChan := make(chan error, 1)
+
+	runErr := a.run(func(agent *Agent) {
+		if a.gatheringState == GatheringStateGathering {
+			gatherErrChan <- ErrMultipleGatherAttempted
+			return
+		} else if a.onCandidateHdlr == nil {
+			gatherErrChan <- ErrNoOnCandidateHandler
+			return
+		}
+
+		go a.gatherCandidates()
+
+		gatherErrChan <- nil
+	})
+	if runErr != nil {
+		return runErr
+	}
+	return <-gatherErrChan
+}
+
+func (a *Agent) gatherCandidates() {
+	gatherStateUpdated := make(chan bool)
+	if err := a.run(func(agent *Agent) {
+		a.gatheringState = GatheringStateGathering
+		close(gatherStateUpdated)
+	}); err != nil {
+		a.log.Warnf("failed to set gatheringState to GatheringStateGathering for gatherCandidates: %v", err)
+		return
+	}
+	<-gatherStateUpdated
+
+	a.gatherCandidatesLocal(a.networkTypes)
+	a.gatherCandidatesSrflx(a.urls, a.networkTypes)
+	if err := a.run(func(agent *Agent) {
+		if a.onCandidateHdlr != nil {
+			go a.onCandidateHdlr(nil)
+		}
+	}); err != nil {
+		a.log.Warnf("Failed to run onCandidateHdlr task: %v\n", err)
+		return
+	}
+	a.gatheringState = GatheringStateComplete
+
+}
+
+func (a *Agent) gatherCandidatesLocal(networkTypes []NetworkType) {
 	localIPs := localInterfaces(networkTypes)
 	for _, ip := range localIPs {
 		for _, network := range supportedNetworks {
@@ -107,16 +155,30 @@ func gatherCandidatesLocal(a *Agent, networkTypes []NetworkType) {
 				continue
 			}
 
-			set := a.localCandidates[c.NetworkType()]
-			set = append(set, c)
-			a.localCandidates[c.NetworkType()] = set
+			if err := a.run(func(agent *Agent) {
+				set := a.localCandidates[c.NetworkType()]
+				set = append(set, c)
+				a.localCandidates[c.NetworkType()] = set
+			}); err != nil {
+				a.log.Warnf("Failed to append to localCandidates: %v\n", err)
+				continue
+			}
 
 			c.start(a, conn)
+
+			if err := a.run(func(agent *Agent) {
+				if a.onCandidateHdlr != nil {
+					go a.onCandidateHdlr(c)
+				}
+			}); err != nil {
+				a.log.Warnf("Failed to run onCandidateHdlr task: %v\n", err)
+				continue
+			}
 		}
 	}
 }
 
-func gatherCandidatesReflective(a *Agent, urls []*URL, networkTypes []NetworkType) {
+func (a *Agent) gatherCandidatesSrflx(urls []*URL, networkTypes []NetworkType) {
 	localIPs := localInterfaces(networkTypes)
 	for _, networkType := range networkTypes {
 		network := networkType.String()
@@ -153,12 +215,26 @@ func gatherCandidatesReflective(a *Agent, urls []*URL, networkTypes []NetworkTyp
 						continue
 					}
 
-					set := a.localCandidates[c.NetworkType()]
-					set = append(set, c)
-					a.localCandidates[c.NetworkType()] = set
+					if err := a.run(func(agent *Agent) {
+						set := a.localCandidates[c.NetworkType()]
+						set = append(set, c)
+						a.localCandidates[c.NetworkType()] = set
+					}); err != nil {
+						a.log.Warnf("Failed to append to localCandidates: %v\n", err)
+						continue
+					}
 
 					c.start(a, conn)
 
+					if err := a.run(func(agent *Agent) {
+						if a.onCandidateHdlr != nil {
+							go a.onCandidateHdlr(c)
+						}
+					}); err != nil {
+						a.log.Warnf("Failed to run onCandidateHdlr task: %v\n", err)
+						continue
+					}
+
 				default:
 					a.log.Warnf("scheme %s is not implemented\n", url.Scheme)
 					continue


Test patch:
diff --git a/agent_test.go b/agent_test.go
index b7cf04a..bc010c7 100644
--- a/agent_test.go
+++ b/agent_test.go
@@ -3,6 +3,7 @@ package ice
 import (
 	"context"
 	"net"
+	"sync"
 	"testing"
 	"time"
 
@@ -318,6 +319,7 @@ func TestConnectivityOnStartup(t *testing.T) {
 
 	cfg := &AgentConfig{
 		Urls:             []*URL{},
+		Trickle:          false,
 		NetworkTypes:     supportedNetworkTypes,
 		taskLoopInterval: time.Hour,
 		LoggerFactory:    logging.NewDefaultLoggerFactory(),
@@ -513,10 +515,14 @@ func TestConnectionStateCallback(t *testing.T) {
 	lim := test.TimeOut(time.Second * 5)
 	defer lim.Stop()
 
+	var wg sync.WaitGroup
+	wg.Add(2)
+
 	timeoutDuration := time.Second
 	KeepaliveInterval := time.Duration(0)
 	cfg := &AgentConfig{
 		Urls:              []*URL{},
+		Trickle:           true,
 		NetworkTypes:      supportedNetworkTypes,
 		ConnectionTimeout: &timeoutDuration,
 		KeepaliveInterval: &KeepaliveInterval,
@@ -527,11 +533,35 @@ func TestConnectionStateCallback(t *testing.T) {
 	if err != nil {
 		t.Error(err)
 	}
+	err = aAgent.OnCandidate(func(candidate Candidate) {
+		if candidate == nil {
+			wg.Done()
+		}
+	})
+	if err != nil {
+		panic(err)
+	}
+	err = aAgent.GatherCandidates()
+	if err != nil {
+		panic(err)
+	}
 
 	bAgent, err := NewAgent(cfg)
 	if err != nil {
 		t.Error(err)
 	}
+	err = bAgent.OnCandidate(func(candidate Candidate) {
+		if candidate == nil {
+			wg.Done()
+		}
+	})
+	if err != nil {
+		panic(err)
+	}
+	err = bAgent.GatherCandidates()
+	if err != nil {
+		panic(err)
+	}
 
 	isChecking := make(chan interface{})
 	isConnected := make(chan interface{})
@@ -553,6 +583,7 @@ func TestConnectionStateCallback(t *testing.T) {
 		t.Error(err)
 	}
 
+	wg.Wait()
 	connect(aAgent, bAgent)
 
 	<-isChecking
@@ -568,3 +599,17 @@ func TestConnectionStateCallback(t *testing.T) {
 
 	<-isClosed
 }
+
+func TestInvalidGather(t *testing.T) {
+	t.Run("Gather with Trickle enable and no OnCandidate should error", func(t *testing.T) {
+		a, err := NewAgent(&AgentConfig{Trickle: true})
+		if err != nil {
+			t.Fatalf("Error constructing ice.Agent")
+		}
+
+		err = a.GatherCandidates()
+		if err != ErrNoOnCandidateHandler {
+			t.Fatal("trickle GatherCandidates succeeded without OnCandidate")
+		}
+	})
+}
diff --git a/transport_test.go b/transport_test.go
index 27f40a8..13dfad7 100644
--- a/transport_test.go
+++ b/transport_test.go
@@ -2,6 +2,7 @@ package ice
 
 import (
 	"context"
+	"sync"
 	"testing"
 	"time"
 
@@ -196,10 +197,16 @@ func pipe() (*Conn, *Conn) {
 	aNotifier, aConnected := onConnected()
 	bNotifier, bConnected := onConnected()
 
-	aAgent, err := NewAgent(&AgentConfig{
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	cfg := &AgentConfig{
 		Urls:         urls,
+		Trickle:      true,
 		NetworkTypes: supportedNetworkTypes,
-	})
+	}
+
+	aAgent, err := NewAgent(cfg)
 	if err != nil {
 		panic(err)
 	}
@@ -207,19 +214,41 @@ func pipe() (*Conn, *Conn) {
 	if err != nil {
 		panic(err)
 	}
-
-	bAgent, err := NewAgent(&AgentConfig{
-		Urls:         urls,
-		NetworkTypes: supportedNetworkTypes,
+	err = aAgent.OnCandidate(func(candidate Candidate) {
+		if candidate == nil {
+			wg.Done()
+		}
 	})
 	if err != nil {
 		panic(err)
 	}
+	err = aAgent.GatherCandidates()
+	if err != nil {
+		panic(err)
+	}
+
+	bAgent, err := NewAgent(cfg)
+	if err != nil {
+		panic(err)
+	}
 	err = bAgent.OnConnectionStateChange(bNotifier)
 	if err != nil {
 		panic(err)
 	}
+	err = bAgent.OnCandidate(func(candidate Candidate) {
+		if candidate == nil {
+			wg.Done()
+		}
+	})
+	if err != nil {
+		panic(err)
+	}
+	err = bAgent.GatherCandidates()
+	if err != nil {
+		panic(err)
+	}
 
+	wg.Wait()
 	aConn, bConn := connect(aAgent, bAgent)
 
 	// Ensure pair selected
@@ -236,12 +265,18 @@ func pipeWithTimeout(iceTimeout time.Duration, iceKeepalive time.Duration) (*Con
 	aNotifier, aConnected := onConnected()
 	bNotifier, bConnected := onConnected()
 
-	aAgent, err := NewAgent(&AgentConfig{
+	var wg sync.WaitGroup
+	wg.Add(2)
+
+	cfg := &AgentConfig{
 		Urls:              urls,
+		Trickle:           true,
 		ConnectionTimeout: &iceTimeout,
 		KeepaliveInterval: &iceKeepalive,
 		NetworkTypes:      supportedNetworkTypes,
-	})
+	}
+
+	aAgent, err := NewAgent(cfg)
 	if err != nil {
 		panic(err)
 	}
@@ -249,21 +284,41 @@ func pipeWithTimeout(iceTimeout time.Duration, iceKeepalive time.Duration) (*Con
 	if err != nil {
 		panic(err)
 	}
-
-	bAgent, err := NewAgent(&AgentConfig{
-		Urls:              urls,
-		ConnectionTimeout: &iceTimeout,
-		KeepaliveInterval: &iceKeepalive,
-		NetworkTypes:      supportedNetworkTypes,
+	err = aAgent.OnCandidate(func(candidate Candidate) {
+		if candidate == nil {
+			wg.Done()
+		}
 	})
 	if err != nil {
 		panic(err)
 	}
+	err = aAgent.GatherCandidates()
+	if err != nil {
+		panic(err)
+	}
+
+	bAgent, err := NewAgent(cfg)
+	if err != nil {
+		panic(err)
+	}
 	err = bAgent.OnConnectionStateChange(bNotifier)
 	if err != nil {
 		panic(err)
 	}
+	err = bAgent.OnCandidate(func(candidate Candidate) {
+		if candidate == nil {
+			wg.Done()
+		}
+	})
+	if err != nil {
+		panic(err)
+	}
+	err = bAgent.GatherCandidates()
+	if err != nil {
+		panic(err)
+	}
 
+	wg.Wait()
 	aConn, bConn := connect(aAgent, bAgent)
 
 	// Ensure pair selected