From 77a072692cafe2820f20925894f1f52f9535fc05 Mon Sep 17 00:00:00 2001 From: Atsushi Watanabe Date: Sat, 4 Jul 2020 14:47:30 +0900 Subject: [PATCH] Fix deadlock in SelectedCandidatePairChange Call handler in different routine. --- agent.go | 78 ++++++++++++++++++++++++++++++++------------------- agent_test.go | 50 ++++++++++++++++++++++++++++++++- 2 files changed, 98 insertions(+), 30 deletions(-) diff --git a/agent.go b/agent.go index 9b96fd2..9751847 100644 --- a/agent.go +++ b/agent.go @@ -113,8 +113,9 @@ type Agent struct { done chan struct{} err atomicError - chanCandidate chan Candidate - chanState chan ConnectionState + chanCandidate chan Candidate + chanCandidatePair chan *candidatePair + chanState chan ConnectionState loggerFactory logging.LoggerFactory log logging.LeveledLogger @@ -210,6 +211,7 @@ func (a *Agent) taskLoop() { close(a.chanState) close(a.chanCandidate) + close(a.chanCandidatePair) }() for { @@ -271,27 +273,28 @@ func NewAgent(config *AgentConfig) (*Agent, error) { startedCtx, startedFn := context.WithCancel(context.Background()) a := &Agent{ - chanTask: make(chan task), - chanState: make(chan ConnectionState), - chanCandidate: make(chan Candidate), - tieBreaker: globalMathRandomGenerator.Uint64(), - lite: config.Lite, - gatheringState: GatheringStateNew, - connectionState: ConnectionStateNew, - localCandidates: make(map[NetworkType][]Candidate), - remoteCandidates: make(map[NetworkType][]Candidate), - urls: config.Urls, - networkTypes: config.NetworkTypes, - onConnected: make(chan struct{}), - buffer: packetio.NewBuffer(), - done: make(chan struct{}), - startedCh: startedCtx.Done(), - startedFn: startedFn, - portmin: config.PortMin, - portmax: config.PortMax, - loggerFactory: loggerFactory, - log: log, - net: config.Net, + chanTask: make(chan task), + chanState: make(chan ConnectionState), + chanCandidate: make(chan Candidate), + chanCandidatePair: make(chan *candidatePair), + tieBreaker: globalMathRandomGenerator.Uint64(), + lite: config.Lite, + gatheringState: GatheringStateNew, + connectionState: ConnectionStateNew, + localCandidates: make(map[NetworkType][]Candidate), + remoteCandidates: make(map[NetworkType][]Candidate), + urls: config.Urls, + networkTypes: config.NetworkTypes, + onConnected: make(chan struct{}), + buffer: packetio.NewBuffer(), + done: make(chan struct{}), + startedCh: startedCtx.Done(), + startedFn: startedFn, + portmin: config.PortMin, + portmax: config.PortMax, + loggerFactory: loggerFactory, + log: log, + net: config.Net, mDNSMode: mDNSMode, mDNSName: mDNSName, @@ -369,10 +372,8 @@ func (a *Agent) OnCandidate(f func(Candidate)) error { } func (a *Agent) onSelectedCandidatePairChange(p *candidatePair) { - if p != nil { - if h, ok := a.onSelectedCandidatePairChangeHdlr.Load().(func(Candidate, Candidate)); ok { - h(p.local, p.remote) - } + if h, ok := a.onSelectedCandidatePairChangeHdlr.Load().(func(Candidate, Candidate)); ok { + h(p.local, p.remote) } } @@ -389,6 +390,17 @@ func (a *Agent) onConnectionStateChange(s ConnectionState) { } func (a *Agent) startOnConnectionStateChangeRoutine() { + go func() { + for { + // CandidatePair and ConnectionState are usually changed at once. + // Blocking one by the other one causes deadlock. + p, isOpen := <-a.chanCandidatePair + if !isOpen { + return + } + a.onSelectedCandidatePairChange(p) + } + }() go func() { for { select { @@ -521,8 +533,6 @@ func (a *Agent) updateConnectionState(newState ConnectionState) { func (a *Agent) setSelectedPair(p *candidatePair) { a.log.Tracef("Set selected candidate pair: %s", p) - // Notify when the selected pair changes - a.onSelectedCandidatePairChange(p) if p == nil { var nilPair *candidatePair @@ -535,6 +545,16 @@ func (a *Agent) setSelectedPair(p *candidatePair) { a.updateConnectionState(ConnectionStateConnected) + // Notify when the selected pair changes + if p != nil { + a.afterRun(func(ctx context.Context) { + select { + case a.chanCandidatePair <- p: + case <-ctx.Done(): + } + }) + } + // Signal connected a.onConnectedOnce.Do(func() { close(a.onConnected) }) } diff --git a/agent_test.go b/agent_test.go index 881d47e..3a75f0f 100644 --- a/agent_test.go +++ b/agent_test.go @@ -1542,7 +1542,7 @@ func TestCloseInConnectionStateCallback(t *testing.T) { assert.NoError(t, bAgent.Close()) } -func TestAgentRestartInConnectionStateCallback(t *testing.T) { +func TestRunTaskInConnectionStateCallback(t *testing.T) { report := test.CheckRoutines(t) defer report() @@ -1569,6 +1569,8 @@ func TestAgentRestartInConnectionStateCallback(t *testing.T) { isComplete := make(chan interface{}) err = aAgent.OnConnectionStateChange(func(c ConnectionState) { if c == ConnectionStateConnected { + _, _, errCred := aAgent.GetLocalUserCredentials() + assert.NoError(t, errCred) assert.NoError(t, aAgent.Restart("", "")) close(isComplete) } @@ -1583,3 +1585,49 @@ func TestAgentRestartInConnectionStateCallback(t *testing.T) { assert.NoError(t, aAgent.Close()) assert.NoError(t, bAgent.Close()) } + +func TestRunTaskInSelectedCandidatePairChangeCallback(t *testing.T) { + report := test.CheckRoutines(t) + defer report() + + lim := test.TimeOut(time.Second * 5) + defer lim.Stop() + + oneSecond := time.Second + KeepaliveInterval := time.Duration(0) + + cfg := &AgentConfig{ + Urls: []*URL{}, + NetworkTypes: supportedNetworkTypes, + DisconnectedTimeout: &oneSecond, + FailedTimeout: &oneSecond, + KeepaliveInterval: &KeepaliveInterval, + taskLoopInterval: 50 * time.Millisecond, + } + + aAgent, err := NewAgent(cfg) + check(err) + bAgent, err := NewAgent(cfg) + check(err) + + isComplete := make(chan interface{}) + if err = aAgent.OnSelectedCandidatePairChange(func(Candidate, Candidate) { + _, _, errCred := aAgent.GetLocalUserCredentials() + assert.NoError(t, errCred) + }); err != nil { + t.Error(err) + } + if err = aAgent.OnConnectionStateChange(func(c ConnectionState) { + if c == ConnectionStateConnected { + close(isComplete) + } + }); err != nil { + t.Error(err) + } + + connect(aAgent, bAgent) + + <-isComplete + assert.NoError(t, aAgent.Close()) + assert.NoError(t, bAgent.Close()) +}