octoleo/syncthing
1
0
Fork 0
mirror of https://github.com/octoleo/syncthing.git synced 2024-11-12 16:26:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Mechanism to stop external announcement routine

Browse Source
This commit is contained in:
Jakob Borg 2014-08-14 12:44:49 +02:00
parent f80f5b3bda
commit 798c4aef9a
1 changed files with 47 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

58
discover/discover.go
View File

@ -24,12 +24,15 @@ type Discoverer struct {
listenAddrs []string listenAddrs []string
localBcastIntv time.Duration localBcastIntv time.Duration
globalBcastIntv time.Duration globalBcastIntv time.Duration
errorRetryIntv time.Duration
beacon *beacon.Beacon beacon *beacon.Beacon
registry map[protocol.NodeID][]string registry map[protocol.NodeID][]string
registryLock sync.RWMutex registryLock sync.RWMutex
extServer string extServer string
extPort uint16 extPort uint16
localBcastTick <-chan time.Time localBcastTick <-chan time.Time
stopGlobal chan struct{}
globalWG sync.WaitGroup
forcedBcastTick chan time.Time forcedBcastTick chan time.Time
extAnnounceOK bool extAnnounceOK bool
extAnnounceOKmut sync.Mutex extAnnounceOKmut sync.Mutex
@ -54,6 +57,7 @@ func NewDiscoverer(id protocol.NodeID, addresses []string, localPort int) (*Disc
listenAddrs: addresses, listenAddrs: addresses,
localBcastIntv: 30 * time.Second, localBcastIntv: 30 * time.Second,
globalBcastIntv: 1800 * time.Second, globalBcastIntv: 1800 * time.Second,
errorRetryIntv: 60 * time.Second,
beacon: b, beacon: b,
registry: make(map[protocol.NodeID][]string), registry: make(map[protocol.NodeID][]string),
} }
@ -70,11 +74,20 @@ func (d *Discoverer) StartLocal() {
} }
func (d *Discoverer) StartGlobal(server string, extPort uint16) { func (d *Discoverer) StartGlobal(server string, extPort uint16) {
// Wait for any previous announcer to stop before starting a new one.
d.globalWG.Wait()
d.extServer = server d.extServer = server
d.extPort = extPort d.extPort = extPort
d.stopGlobal = make(chan struct{})
d.globalWG.Add(1)
go d.sendExternalAnnouncements() go d.sendExternalAnnouncements()
} }
func (d *Discoverer) StopGlobal() {
close(d.stopGlobal)
d.globalWG.Wait()
}
func (d *Discoverer) ExtAnnounceOK() bool { func (d *Discoverer) ExtAnnounceOK() bool {
d.extAnnounceOKmut.Lock() d.extAnnounceOKmut.Lock()
defer d.extAnnounceOKmut.Unlock() defer d.extAnnounceOKmut.Unlock()
@ -173,20 +186,19 @@ func (d *Discoverer) sendLocalAnnouncements() {
} }
func (d *Discoverer) sendExternalAnnouncements() { func (d *Discoverer) sendExternalAnnouncements() {
// this should go in the Discoverer struct defer d.globalWG.Done()
errorRetryIntv := 60 * time.Second
remote, err := net.ResolveUDPAddr("udp", d.extServer) remote, err := net.ResolveUDPAddr("udp", d.extServer)
for err != nil { for err != nil {
l.Warnf("Global discovery: %v; trying again in %v", err, errorRetryIntv) l.Warnf("Global discovery: %v; trying again in %v", err, d.errorRetryIntv)
time.Sleep(errorRetryIntv) time.Sleep(d.errorRetryIntv)
remote, err = net.ResolveUDPAddr("udp", d.extServer) remote, err = net.ResolveUDPAddr("udp", d.extServer)
} }
conn, err := net.ListenUDP("udp", nil) conn, err := net.ListenUDP("udp", nil)
for err != nil { for err != nil {
l.Warnf("Global discovery: %v; trying again in %v", err, errorRetryIntv) l.Warnf("Global discovery: %v; trying again in %v", err, d.errorRetryIntv)
time.Sleep(errorRetryIntv) time.Sleep(d.errorRetryIntv)
conn, err = net.ListenUDP("udp", nil) conn, err = net.ListenUDP("udp", nil)
} }
@ -201,7 +213,10 @@ func (d *Discoverer) sendExternalAnnouncements() {
buf = d.announcementPkt() buf = d.announcementPkt()
} }
for { var bcastTick = time.Tick(d.globalBcastIntv)
var errTick <-chan time.Time
sendOneAnnouncement := func() {
var ok bool var ok bool
if debug { if debug {
@ -230,11 +245,32 @@ func (d *Discoverer) sendExternalAnnouncements() {
d.extAnnounceOKmut.Unlock() d.extAnnounceOKmut.Unlock()
if ok { if ok {
time.Sleep(d.globalBcastIntv) errTick = nil
} else { } else if errTick != nil {
time.Sleep(errorRetryIntv) errTick = time.Tick(d.errorRetryIntv)
} }
} }
// Announce once, immediately
sendOneAnnouncement()
loop:
for {
select {
case <-d.stopGlobal:
break loop
case <-errTick:
sendOneAnnouncement()
case <-bcastTick:
sendOneAnnouncement()
}
}
if debug {
l.Debugln("discover: stopping global")
}
} }
func (d *Discoverer) recvAnnouncements() { func (d *Discoverer) recvAnnouncements() {
@ -295,7 +331,7 @@ func (d *Discoverer) registerNode(addr net.Addr, node Node) bool {
} }
} }
if debug { if debug {
l.Debugf("discover: register: %s -> %#v", node.ID, addrs) l.Debugf("discover: register: %v -> %#v", node.ID, addrs)
} }
var id protocol.NodeID var id protocol.NodeID
copy(id[:], node.ID) copy(id[:], node.ID)