netbird/ice
0
0
Fork 0
mirror of https://github.com/netbirdio/ice.git synced 2026-05-22 17:10:58 -07:00
Code Issues Packages Projects Releases Wiki Activity

Agent: Frequent candidate check when connecting

Browse Source 
Previously in the connecting phase, a gathered peer reflexive candidate
which is by default nominated after 500mS, would have to wait 2s
(default taskLoopInterval) before being nominated.
This commit is contained in:
Sam Lancia
2020-07-05 17:45:01 +09:00
committed by Atsushi Watanabe
parent 6b045b828f
commit a4a53ab443
5 changed files with 49 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files
agent.go
+26 -11
View File
@@ -38,8 +38,7 @@ type Agent struct {
onConnected chan struct{}
onConnectedOnce sync.Once
connectivityTicker *time.Ticker
// force candidate to be contacted immediately (instead of waiting for connectivityTicker)
// force candidate to be contacted immediately (instead of waiting for task ticker)
forceCandidateContact chan bool
tieBreaker uint64
@@ -82,8 +81,8 @@ type Agent struct {
// 0 means never
keepaliveInterval time.Duration
// How after should we run our internal taskLoop
taskLoopInterval time.Duration
// How often should we run our internal taskLoop to check for state changes when connecting
checkInterval time.Duration
localUfrag string
localPwd string
@@ -200,10 +199,6 @@ func (a *Agent) taskLoop() {
a.log.Warnf("failed to close buffer: %v", err)
}
if a.connectivityTicker != nil {
a.connectivityTicker.Stop()
}
a.closeMulticastConn()
a.updateConnectionState(ConnectionStateClosed)
@@ -460,9 +455,7 @@ func (a *Agent) startConnectivityChecks(isControlling bool, remoteUfrag, remoteP
agent.updateConnectionState(ConnectionStateChecking)
// TODO this should be dynamic, and grow when the connection is stable
a.requestConnectivityCheck()
agent.connectivityTicker = time.NewTicker(a.taskLoopInterval)
go a.connectivityChecks()
})
}
@@ -502,12 +495,34 @@ func (a *Agent) connectivityChecks() {
}
for {
interval := defaultKeepaliveInterval
updateInterval := func(x time.Duration) {
if x != 0 && (interval == 0 || interval > x) {
interval = x
}
}
switch lastConnectionState {
case ConnectionStateNew, ConnectionStateChecking: // While connecting, check candidates more frequently
updateInterval(a.checkInterval)
case ConnectionStateConnected, ConnectionStateDisconnected:
updateInterval(a.keepaliveInterval)
default:
}
// Ensure we run our task loop as quickly as the minimum of our various configured timeouts
updateInterval(a.disconnectedTimeout)
updateInterval(a.failedTimeout)
t := time.NewTimer(interval)
select {
case <-a.forceCandidateContact:
t.Stop()
contact()
case <-a.connectivityTicker.C:
case <-t.C:
contact()
case <-a.done:
t.Stop()
return
}
}
agent_config.go
+8 -9
View File
@@ -8,8 +8,8 @@ import (
)
const (
// taskLoopInterval is the interval at which the agent performs checks
defaultTaskLoopInterval = 2 * time.Second
// defaultCheckInterval is the interval at which the agent performs candidate checks in the connecting phase
defaultCheckInterval = 200 * time.Millisecond
// keepaliveInterval used to keep candidates alive
defaultKeepaliveInterval = 2 * time.Second
@@ -95,10 +95,9 @@ type AgentConfig struct {
LoggerFactory logging.LoggerFactory
// taskLoopInterval controls how often our internal task loop runs, this
// 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
// checkInterval controls how often our internal task loop runs when
// in the connecting state. Only useful for testing.
checkInterval 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
@@ -205,10 +204,10 @@ func (config *AgentConfig) initWithDefaults(a *Agent) {
a.keepaliveInterval = *config.KeepaliveInterval
}
if config.taskLoopInterval == 0 {
a.taskLoopInterval = defaultTaskLoopInterval
if config.checkInterval == 0 {
a.checkInterval = defaultCheckInterval
} else {
a.taskLoopInterval = config.taskLoopInterval
a.checkInterval = config.checkInterval
}
if config.CandidateTypes == nil || len(config.CandidateTypes) == 0 {
agent_test.go
+11 -18
View File
@@ -238,7 +238,6 @@ func TestHandlePeerReflexive(t *testing.T) {
var config AgentConfig
runAgentTest(t, &config, func(ctx context.Context, a *Agent) {
a.selector = &controllingSelector{agent: a, log: a.log}
a.connectivityTicker = time.NewTicker(a.taskLoopInterval)
hostConfig := CandidateHostConfig{
Network: "udp",
@@ -299,7 +298,6 @@ func TestHandlePeerReflexive(t *testing.T) {
var config AgentConfig
runAgentTest(t, &config, func(ctx context.Context, a *Agent) {
a.selector = &controllingSelector{agent: a, log: a.log}
a.connectivityTicker = time.NewTicker(a.taskLoopInterval)
hostConfig := CandidateHostConfig{
Network: "tcp",
@@ -326,7 +324,6 @@ func TestHandlePeerReflexive(t *testing.T) {
var config AgentConfig
runAgentTest(t, &config, func(ctx context.Context, a *Agent) {
a.selector = &controllingSelector{agent: a, log: a.log}
a.connectivityTicker = time.NewTicker(a.taskLoopInterval)
tID := [stun.TransactionIDSize]byte{}
copy(tID[:], []byte("ABC"))
a.pendingBindingRequests = []bindingRequest{
@@ -393,12 +390,14 @@ func TestConnectivityOnStartup(t *testing.T) {
aNotifier, aConnected := onConnected()
bNotifier, bConnected := onConnected()
KeepaliveInterval := time.Hour
cfg0 := &AgentConfig{
NetworkTypes: supportedNetworkTypes,
MulticastDNSMode: MulticastDNSModeDisabled,
Net: net0,
taskLoopInterval: time.Hour,
KeepaliveInterval: &KeepaliveInterval,
checkInterval: time.Hour,
}
aAgent, err := NewAgent(cfg0)
@@ -406,10 +405,11 @@ func TestConnectivityOnStartup(t *testing.T) {
require.NoError(t, aAgent.OnConnectionStateChange(aNotifier))
cfg1 := &AgentConfig{
NetworkTypes: supportedNetworkTypes,
MulticastDNSMode: MulticastDNSModeDisabled,
Net: net1,
taskLoopInterval: time.Hour,
NetworkTypes: supportedNetworkTypes,
MulticastDNSMode: MulticastDNSModeDisabled,
Net: net1,
KeepaliveInterval: &KeepaliveInterval,
checkInterval: time.Hour,
}
bAgent, err := NewAgent(cfg1)
@@ -617,7 +617,6 @@ func TestInboundValidity(t *testing.T) {
err = a.run(context.Background(), func(ctx context.Context, a *Agent) {
a.selector = &controllingSelector{agent: a, log: a.log}
a.connectivityTicker = time.NewTicker(a.taskLoopInterval)
a.handleInbound(buildMsg(stun.ClassRequest, a.localUfrag+":"+a.remoteUfrag, a.localPwd), local, remote)
if len(a.remoteCandidates) != 1 {
t.Fatal("Binding with valid values was unable to create prflx candidate")
@@ -632,7 +631,6 @@ func TestInboundValidity(t *testing.T) {
var config AgentConfig
runAgentTest(t, &config, func(ctx context.Context, a *Agent) {
a.selector = &controllingSelector{agent: a, log: a.log}
a.connectivityTicker = time.NewTicker(a.taskLoopInterval)
msg, err := stun.Build(stun.BindingRequest, stun.TransactionID,
stun.NewUsername(a.localUfrag+":"+a.remoteUfrag),
stun.NewShortTermIntegrity(a.localPwd),
@@ -732,7 +730,6 @@ func TestConnectionStateCallback(t *testing.T) {
DisconnectedTimeout: &disconnectedDuration,
FailedTimeout: &failedDuration,
KeepaliveInterval: &KeepaliveInterval,
taskLoopInterval: 500 * time.Millisecond,
}
aAgent, err := NewAgent(cfg)
@@ -1276,7 +1273,6 @@ func TestConnectionStateFailedDeleteAllCandidates(t *testing.T) {
DisconnectedTimeout: &oneSecond,
FailedTimeout: &oneSecond,
KeepaliveInterval: &KeepaliveInterval,
taskLoopInterval: 250 * time.Millisecond,
}
aAgent, err := NewAgent(cfg)
@@ -1322,7 +1318,6 @@ func TestConnectionStateConnectingToFailed(t *testing.T) {
DisconnectedTimeout: &oneSecond,
FailedTimeout: &oneSecond,
KeepaliveInterval: &KeepaliveInterval,
taskLoopInterval: 250 * time.Millisecond,
}
aAgent, err := NewAgent(cfg)
@@ -1400,7 +1395,6 @@ func TestAgentRestart(t *testing.T) {
connA, connB := pipe(&AgentConfig{
DisconnectedTimeout: &oneSecond,
FailedTimeout: &oneSecond,
taskLoopInterval: 50 * time.Millisecond,
})
ctx, cancel := context.WithCancel(context.Background())
@@ -1432,7 +1426,6 @@ func TestAgentRestart(t *testing.T) {
connA, connB := pipe(&AgentConfig{
DisconnectedTimeout: &oneSecond,
FailedTimeout: &oneSecond,
taskLoopInterval: 50 * time.Millisecond,
})
connAFirstCandidates := generateCandidateAddressStrings(connA.agent.GetLocalCandidates())
connBFirstCandidates := generateCandidateAddressStrings(connB.agent.GetLocalCandidates())
@@ -1507,7 +1500,7 @@ func TestCloseInConnectionStateCallback(t *testing.T) {
DisconnectedTimeout: &disconnectedDuration,
FailedTimeout: &failedDuration,
KeepaliveInterval: &KeepaliveInterval,
taskLoopInterval: 500 * time.Millisecond,
checkInterval: 500 * time.Millisecond,
}
aAgent, err := NewAgent(cfg)
@@ -1558,7 +1551,7 @@ func TestRunTaskInConnectionStateCallback(t *testing.T) {
DisconnectedTimeout: &oneSecond,
FailedTimeout: &oneSecond,
KeepaliveInterval: &KeepaliveInterval,
taskLoopInterval: 50 * time.Millisecond,
checkInterval: 50 * time.Millisecond,
}
aAgent, err := NewAgent(cfg)
@@ -1602,7 +1595,7 @@ func TestRunTaskInSelectedCandidatePairChangeCallback(t *testing.T) {
DisconnectedTimeout: &oneSecond,
FailedTimeout: &oneSecond,
KeepaliveInterval: &KeepaliveInterval,
taskLoopInterval: 50 * time.Millisecond,
checkInterval: 50 * time.Millisecond,
}
aAgent, err := NewAgent(cfg)
connectivity_vnet_test.go
+2 -2
View File
@@ -475,7 +475,7 @@ func TestDisconnectedToConnected(t *testing.T) {
Net: net0,
DisconnectedTimeout: &disconnectTimeout,
KeepaliveInterval: &keepaliveInterval,
taskLoopInterval: keepaliveInterval,
checkInterval: keepaliveInterval,
})
assert.NoError(t, err)
@@ -485,7 +485,7 @@ func TestDisconnectedToConnected(t *testing.T) {
Net: net1,
DisconnectedTimeout: &disconnectTimeout,
KeepaliveInterval: &keepaliveInterval,
taskLoopInterval: keepaliveInterval,
checkInterval: keepaliveInterval,
})
assert.NoError(t, err)
transport_test.go
+2 -2
View File
@@ -41,7 +41,7 @@ func testTimeout(t *testing.T, c *Conn, timeout time.Duration) {
startedAt := time.Now()
for cnt := time.Duration(0); cnt <= timeout+defaultTaskLoopInterval; cnt += pollrate {
for cnt := time.Duration(0); cnt <= timeout+defaultKeepaliveInterval+pollrate; cnt += pollrate {
<-ticker.C
var cs ConnectionState
@@ -64,7 +64,7 @@ func testTimeout(t *testing.T, c *Conn, timeout time.Duration) {
}
}
}
t.Fatalf("Connection failed to time out in time.")
t.Fatalf("Connection failed to time out in time. (expected timeout: %v)", timeout)
}
func TestTimeout(t *testing.T) {