diff --git a/agent.go b/agent.go index 9751847..91b19fb 100644 --- a/agent.go +++ b/agent.go @@ -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 } } diff --git a/agent_config.go b/agent_config.go index 5fb3b4f..41071e8 100644 --- a/agent_config.go +++ b/agent_config.go @@ -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 { diff --git a/agent_test.go b/agent_test.go index 44113cd..0f9bba0 100644 --- a/agent_test.go +++ b/agent_test.go @@ -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) diff --git a/connectivity_vnet_test.go b/connectivity_vnet_test.go index efbf2e5..e8f6c6e 100644 --- a/connectivity_vnet_test.go +++ b/connectivity_vnet_test.go @@ -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) diff --git a/transport_test.go b/transport_test.go index 08eb368..3e6c84b 100644 --- a/transport_test.go +++ b/transport_test.go @@ -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) {