From 553df099a0656a0b29934f1a2fc7613e0e462340 Mon Sep 17 00:00:00 2001 From: Atsushi Watanabe Date: Mon, 6 Jul 2020 13:04:52 +0900 Subject: [PATCH] Prevent multiple gathering in parallel Multiple candidate gathering routine were executed in parallel when Restart() called immediately after previous start of gathering. Cancel previous gathering by context to prevent mixing candidates from two routines. --- agent.go | 9 ++++++-- gather.go | 67 ++++++++++++++++++++++++++++++++++++++----------------- 2 files changed, 54 insertions(+), 22 deletions(-) diff --git a/agent.go b/agent.go index 91b19fb..dc16158 100644 --- a/agent.go +++ b/agent.go @@ -112,6 +112,8 @@ type Agent struct { done chan struct{} err atomicError + gatherCandidateCancel func() + chanCandidate chan Candidate chanCandidatePair chan *candidatePair chanState chan ConnectionState @@ -295,6 +297,8 @@ func NewAgent(config *AgentConfig) (*Agent, error) { mDNSName: mDNSName, mDNSConn: mDNSConn, + gatherCandidateCancel: func() {}, + forceCandidateContact: make(chan bool, 1), interfaceFilter: config.InterfaceFilter, @@ -777,8 +781,8 @@ func (a *Agent) addRemoteCandidate(c Candidate) { a.requestConnectivityCheck() } -func (a *Agent) addCandidate(c Candidate, candidateConn net.PacketConn) error { - return a.run(a.context(), func(ctx context.Context, agent *Agent) { +func (a *Agent) addCandidate(ctx context.Context, c Candidate, candidateConn net.PacketConn) error { + return a.run(ctx, func(ctx context.Context, agent *Agent) { c.start(a, candidateConn, a.startedCh) set := a.localCandidates[c.NetworkType()] @@ -866,6 +870,7 @@ func (a *Agent) Close() error { close(done) }) + a.gatherCandidateCancel() a.err.Store(ErrClosed) close(a.done) diff --git a/gather.go b/gather.go index d608f98..3ff7bdc 100644 --- a/gather.go +++ b/gather.go @@ -67,14 +67,18 @@ func (a *Agent) GatherCandidates() error { return } - go a.gatherCandidates() + a.gatherCandidateCancel() // Cancel previous gathering routine + ctx, cancel := context.WithCancel(ctx) + a.gatherCandidateCancel = cancel + + go a.gatherCandidates(ctx) }); runErr != nil { return runErr } return gatherErr } -func (a *Agent) gatherCandidates() { +func (a *Agent) gatherCandidates(ctx context.Context) { if err := a.setGatheringState(GatheringStateGathering); err != nil { a.log.Warnf("failed to set gatheringState to GatheringStateGathering: %v", err) return @@ -84,16 +88,30 @@ func (a *Agent) gatherCandidates() { for _, t := range a.candidateTypes { switch t { case CandidateTypeHost: - a.gatherCandidatesLocal(a.networkTypes) + wg.Add(1) + go func() { + a.gatherCandidatesLocal(ctx, a.networkTypes) + wg.Done() + }() case CandidateTypeServerReflexive: - a.gatherCandidatesSrflx(a.urls, a.networkTypes, &wg) + wg.Add(1) + go func() { + a.gatherCandidatesSrflx(ctx, a.urls, a.networkTypes) + wg.Done() + }() if a.extIPMapper != nil && a.extIPMapper.candidateType == CandidateTypeServerReflexive { - a.gatherCandidatesSrflxMapped(a.networkTypes, &wg) + wg.Add(1) + go func() { + a.gatherCandidatesSrflxMapped(ctx, a.networkTypes) + wg.Done() + }() } case CandidateTypeRelay: - if err := a.gatherCandidatesRelay(a.urls, &wg); err != nil { - a.log.Errorf("Failed to gather relay candidates: %v\n", err) - } + wg.Add(1) + go func() { + a.gatherCandidatesRelay(ctx, a.urls) + wg.Done() + }() } } // Block until all STUN and TURN URLs have been gathered (or timed out) @@ -104,7 +122,7 @@ func (a *Agent) gatherCandidates() { } } -func (a *Agent) gatherCandidatesLocal(networkTypes []NetworkType) { +func (a *Agent) gatherCandidatesLocal(ctx context.Context, networkTypes []NetworkType) { localIPs, err := localInterfaces(a.net, a.interfaceFilter, networkTypes) if err != nil { a.log.Warnf("failed to iterate local interfaces, host candidates will not be gathered %s", err) @@ -154,7 +172,7 @@ func (a *Agent) gatherCandidatesLocal(networkTypes []NetworkType) { } } - if err := a.addCandidate(c, conn); err != nil { + if err := a.addCandidate(ctx, c, conn); err != nil { if closeErr := c.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -164,7 +182,10 @@ func (a *Agent) gatherCandidatesLocal(networkTypes []NetworkType) { } } -func (a *Agent) gatherCandidatesSrflxMapped(networkTypes []NetworkType, wg *sync.WaitGroup) { +func (a *Agent) gatherCandidatesSrflxMapped(ctx context.Context, networkTypes []NetworkType) { + var wg sync.WaitGroup + defer wg.Wait() + for _, networkType := range networkTypes { network := networkType.String() wg.Add(1) @@ -201,7 +222,7 @@ func (a *Agent) gatherCandidatesSrflxMapped(networkTypes []NetworkType, wg *sync return } - if err := a.addCandidate(c, conn); err != nil { + if err := a.addCandidate(ctx, c, conn); err != nil { if closeErr := c.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -211,7 +232,10 @@ func (a *Agent) gatherCandidatesSrflxMapped(networkTypes []NetworkType, wg *sync } } -func (a *Agent) gatherCandidatesSrflx(urls []*URL, networkTypes []NetworkType, wg *sync.WaitGroup) { +func (a *Agent) gatherCandidatesSrflx(ctx context.Context, urls []*URL, networkTypes []NetworkType) { + var wg sync.WaitGroup + defer wg.Wait() + for _, networkType := range networkTypes { for i := range urls { wg.Add(1) @@ -254,7 +278,7 @@ func (a *Agent) gatherCandidatesSrflx(urls []*URL, networkTypes []NetworkType, w return } - if err := a.addCandidate(c, conn); err != nil { + if err := a.addCandidate(ctx, c, conn); err != nil { if closeErr := c.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -265,16 +289,21 @@ func (a *Agent) gatherCandidatesSrflx(urls []*URL, networkTypes []NetworkType, w } } -func (a *Agent) gatherCandidatesRelay(urls []*URL, wg *sync.WaitGroup) error { +func (a *Agent) gatherCandidatesRelay(ctx context.Context, urls []*URL) { + var wg sync.WaitGroup + defer wg.Wait() + network := NetworkTypeUDP4.String() // TODO IPv6 for i := range urls { switch { case urls[i].Scheme != SchemeTypeTURN && urls[i].Scheme != SchemeTypeTURNS: continue case urls[i].Username == "": - return ErrUsernameEmpty + a.log.Errorf("Failed to gather relay candidates: %v", ErrUsernameEmpty) + return case urls[i].Password == "": - return ErrPasswordEmpty + a.log.Errorf("Failed to gather relay candidates: %v", ErrPasswordEmpty) + return } wg.Add(1) @@ -397,7 +426,7 @@ func (a *Agent) gatherCandidatesRelay(urls []*URL, wg *sync.WaitGroup) error { return } - if err := a.addCandidate(candidate, relayConn); err != nil { + if err := a.addCandidate(ctx, candidate, relayConn); err != nil { if closeErr := candidate.close(); closeErr != nil { a.log.Warnf("Failed to close candidate: %v", closeErr) } @@ -405,6 +434,4 @@ func (a *Agent) gatherCandidatesRelay(urls []*URL, wg *sync.WaitGroup) error { } }(*urls[i]) } - - return nil }