syncthing/internal/discover/discover.go

513 lines
12 KiB
Go
Raw Normal View History

2014-11-16 20:13:20 +00:00
// Copyright (C) 2014 The Syncthing Authors.
2014-09-29 19:43:32 +00:00
//
// This program is free software: you can redistribute it and/or modify it
// under the terms of the GNU General Public License as published by the Free
// Software Foundation, either version 3 of the License, or (at your option)
// any later version.
//
// This program is distributed in the hope that it will be useful, but WITHOUT
// ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
// FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
// more details.
//
// You should have received a copy of the GNU General Public License along
// with this program. If not, see <http://www.gnu.org/licenses/>.
2014-06-01 20:50:14 +00:00
2013-12-15 10:43:31 +00:00
package discover
import (
"bytes"
2014-02-20 16:40:15 +00:00
"encoding/hex"
"errors"
2014-05-11 22:55:43 +00:00
"io"
2013-12-15 10:43:31 +00:00
"net"
"strconv"
2013-12-15 10:43:31 +00:00
"sync"
"time"
2013-12-24 16:10:49 +00:00
2014-09-22 19:42:11 +00:00
"github.com/syncthing/syncthing/internal/beacon"
"github.com/syncthing/syncthing/internal/events"
"github.com/syncthing/syncthing/internal/protocol"
2013-12-15 10:43:31 +00:00
)
type Discoverer struct {
myID protocol.DeviceID
listenAddrs []string
localBcastIntv time.Duration
globalBcastIntv time.Duration
errorRetryIntv time.Duration
cacheLifetime time.Duration
broadcastBeacon beacon.Interface
multicastBeacon beacon.Interface
registry map[protocol.DeviceID][]CacheEntry
registryLock sync.RWMutex
extServer string
2014-04-18 11:20:42 +00:00
extPort uint16
localBcastTick <-chan time.Time
stopGlobal chan struct{}
globalWG sync.WaitGroup
forcedBcastTick chan time.Time
extAnnounceOK bool
extAnnounceOKmut sync.Mutex
2013-12-22 21:29:23 +00:00
}
type CacheEntry struct {
Address string
Seen time.Time
}
2014-02-20 16:40:15 +00:00
var (
2014-02-24 12:29:30 +00:00
ErrIncorrectMagic = errors.New("incorrect magic number")
2014-02-20 16:40:15 +00:00
)
func NewDiscoverer(id protocol.DeviceID, addresses []string) *Discoverer {
return &Discoverer{
myID: id,
listenAddrs: addresses,
localBcastIntv: 30 * time.Second,
globalBcastIntv: 1800 * time.Second,
errorRetryIntv: 60 * time.Second,
cacheLifetime: 5 * time.Minute,
registry: make(map[protocol.DeviceID][]CacheEntry),
}
}
2013-12-15 10:43:31 +00:00
func (d *Discoverer) StartLocal(localPort int, localMCAddr string) {
if localPort > 0 {
bb, err := beacon.NewBroadcast(localPort)
if err != nil {
if debug {
l.Debugln(err)
}
l.Infoln("Local discovery over IPv4 unavailable")
} else {
d.broadcastBeacon = bb
go d.recvAnnouncements(bb)
}
}
if len(localMCAddr) > 0 {
mb, err := beacon.NewMulticast(localMCAddr)
if err != nil {
if debug {
l.Debugln(err)
}
l.Infoln("Local discovery over IPv6 unavailable")
} else {
d.multicastBeacon = mb
go d.recvAnnouncements(mb)
}
}
if d.broadcastBeacon == nil && d.multicastBeacon == nil {
l.Warnln("Local discovery unavailable")
} else {
d.localBcastTick = time.Tick(d.localBcastIntv)
d.forcedBcastTick = make(chan time.Time)
go d.sendLocalAnnouncements()
}
}
2014-04-18 11:20:42 +00:00
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
2014-04-18 11:20:42 +00:00
d.extPort = extPort
d.stopGlobal = make(chan struct{})
d.globalWG.Add(1)
go d.sendExternalAnnouncements()
}
func (d *Discoverer) StopGlobal() {
if d.stopGlobal != nil {
close(d.stopGlobal)
d.globalWG.Wait()
}
}
func (d *Discoverer) ExtAnnounceOK() bool {
d.extAnnounceOKmut.Lock()
defer d.extAnnounceOKmut.Unlock()
return d.extAnnounceOK
}
func (d *Discoverer) Lookup(device protocol.DeviceID) []string {
d.registryLock.RLock()
cached := d.filterCached(d.registry[device])
d.registryLock.RUnlock()
2014-05-02 06:53:19 +00:00
if len(cached) > 0 {
addrs := make([]string, len(cached))
for i := range cached {
addrs[i] = cached[i].Address
}
return addrs
2014-05-02 06:53:19 +00:00
} else if len(d.extServer) != 0 {
addrs := d.externalLookup(device)
cached = make([]CacheEntry, len(addrs))
for i := range addrs {
cached[i] = CacheEntry{
Address: addrs[i],
Seen: time.Now(),
}
}
d.registryLock.Lock()
d.registry[device] = cached
d.registryLock.Unlock()
2014-05-02 06:53:19 +00:00
}
return nil
}
func (d *Discoverer) Hint(device string, addrs []string) {
resAddrs := resolveAddrs(addrs)
var id protocol.DeviceID
id.UnmarshalText([]byte(device))
d.registerDevice(nil, Device{
Addresses: resAddrs,
ID: id[:],
})
}
func (d *Discoverer) All() map[protocol.DeviceID][]CacheEntry {
d.registryLock.RLock()
devices := make(map[protocol.DeviceID][]CacheEntry, len(d.registry))
for device, addrs := range d.registry {
addrsCopy := make([]CacheEntry, len(addrs))
copy(addrsCopy, addrs)
devices[device] = addrsCopy
}
d.registryLock.RUnlock()
return devices
}
func (d *Discoverer) announcementPkt() []byte {
var addrs []Address
for _, astr := range d.listenAddrs {
addr, err := net.ResolveTCPAddr("tcp", astr)
if err != nil {
2014-05-15 00:08:56 +00:00
l.Warnln("%v: not announcing %s", err, astr)
continue
} else if debug {
2014-05-15 00:08:56 +00:00
l.Debugf("discover: announcing %s: %#v", astr, addr)
}
if len(addr.IP) == 0 || addr.IP.IsUnspecified() {
addrs = append(addrs, Address{Port: uint16(addr.Port)})
} else if bs := addr.IP.To4(); bs != nil {
addrs = append(addrs, Address{IP: bs, Port: uint16(addr.Port)})
} else if bs := addr.IP.To16(); bs != nil {
addrs = append(addrs, Address{IP: bs, Port: uint16(addr.Port)})
}
}
var pkt = Announce{
Magic: AnnouncementMagic,
This: Device{d.myID[:], addrs},
}
2014-10-21 06:40:05 +00:00
return pkt.MustMarshalXDR()
}
func (d *Discoverer) sendLocalAnnouncements() {
2014-05-11 22:55:43 +00:00
var addrs = resolveAddrs(d.listenAddrs)
var pkt = Announce{
Magic: AnnouncementMagic,
This: Device{d.myID[:], addrs},
2014-05-11 22:55:43 +00:00
}
2014-10-21 06:40:05 +00:00
msg := pkt.MustMarshalXDR()
2014-03-28 10:04:48 +00:00
for {
if d.multicastBeacon != nil {
d.multicastBeacon.Send(msg)
}
if d.broadcastBeacon != nil {
d.broadcastBeacon.Send(msg)
}
select {
case <-d.localBcastTick:
case <-d.forcedBcastTick:
}
}
2013-12-22 21:29:23 +00:00
}
2013-12-15 10:43:31 +00:00
func (d *Discoverer) sendExternalAnnouncements() {
defer d.globalWG.Done()
remote, err := net.ResolveUDPAddr("udp", d.extServer)
for err != nil {
l.Warnf("Global discovery: %v; trying again in %v", err, d.errorRetryIntv)
time.Sleep(d.errorRetryIntv)
remote, err = net.ResolveUDPAddr("udp", d.extServer)
2013-12-22 21:29:23 +00:00
}
conn, err := net.ListenUDP("udp", nil)
for err != nil {
l.Warnf("Global discovery: %v; trying again in %v", err, d.errorRetryIntv)
time.Sleep(d.errorRetryIntv)
conn, err = net.ListenUDP("udp", nil)
}
2014-04-18 11:20:42 +00:00
var buf []byte
if d.extPort != 0 {
var pkt = Announce{
Magic: AnnouncementMagic,
This: Device{d.myID[:], []Address{{Port: d.extPort}}},
2014-04-18 11:20:42 +00:00
}
2014-10-21 06:40:05 +00:00
buf = pkt.MustMarshalXDR()
2014-04-18 11:20:42 +00:00
} else {
buf = d.announcementPkt()
}
var bcastTick = time.Tick(d.globalBcastIntv)
var errTick <-chan time.Time
sendOneAnnouncement := func() {
var ok bool
2014-03-09 07:58:03 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l.Debugf("discover: send announcement -> %v\n%s", remote, hex.Dump(buf))
}
_, err := conn.WriteTo(buf, remote)
2013-12-15 10:43:31 +00:00
if err != nil {
2014-05-15 00:08:56 +00:00
if debug {
l.Debugln("discover: warning:", err)
}
ok = false
2013-12-18 17:29:15 +00:00
} else {
// Verify that the announce server responds positively for our device ID
2014-04-16 13:06:54 +00:00
time.Sleep(1 * time.Second)
res := d.externalLookup(d.myID)
if debug {
2014-05-15 00:08:56 +00:00
l.Debugln("discover: external lookup check:", res)
}
ok = len(res) > 0
}
d.extAnnounceOKmut.Lock()
d.extAnnounceOK = ok
d.extAnnounceOKmut.Unlock()
if ok {
errTick = nil
} else if errTick != nil {
errTick = time.Tick(d.errorRetryIntv)
}
}
2014-08-12 22:29:29 +00:00
// Announce once, immediately
sendOneAnnouncement()
loop:
for {
select {
case <-d.stopGlobal:
break loop
case <-errTick:
sendOneAnnouncement()
case <-bcastTick:
sendOneAnnouncement()
2014-04-16 13:06:54 +00:00
}
2013-12-15 10:43:31 +00:00
}
if debug {
l.Debugln("discover: stopping global")
}
2013-12-15 10:43:31 +00:00
}
func (d *Discoverer) recvAnnouncements(b beacon.Interface) {
2014-03-28 10:04:48 +00:00
for {
buf, addr := b.Recv()
2013-12-22 21:29:23 +00:00
2014-03-09 07:58:03 +00:00
if debug {
l.Debugf("discover: read announcement from %s:\n%s", addr, hex.Dump(buf))
2014-02-20 16:40:15 +00:00
}
var pkt Announce
2014-03-28 10:04:48 +00:00
err := pkt.UnmarshalXDR(buf)
2014-05-11 22:55:43 +00:00
if err != nil && err != io.EOF {
2013-12-15 10:43:31 +00:00
continue
}
var newDevice bool
if bytes.Compare(pkt.This.ID, d.myID[:]) != 0 {
newDevice = d.registerDevice(addr, pkt.This)
2014-05-11 22:55:43 +00:00
}
if newDevice {
2014-05-11 22:55:43 +00:00
select {
case d.forcedBcastTick <- time.Now():
}
2013-12-15 10:43:31 +00:00
}
}
}
func (d *Discoverer) registerDevice(addr net.Addr, device Device) bool {
var id protocol.DeviceID
copy(id[:], device.ID)
d.registryLock.RLock()
current := d.filterCached(d.registry[id])
d.registryLock.RUnlock()
orig := current
for _, a := range device.Addresses {
var deviceAddr string
2014-05-11 22:55:43 +00:00
if len(a.IP) > 0 {
deviceAddr = net.JoinHostPort(net.IP(a.IP).String(), strconv.Itoa(int(a.Port)))
2014-05-11 22:55:43 +00:00
} else if addr != nil {
ua := addr.(*net.UDPAddr)
ua.Port = int(a.Port)
deviceAddr = ua.String()
2014-05-11 22:55:43 +00:00
}
for i := range current {
if current[i].Address == deviceAddr {
current[i].Seen = time.Now()
goto done
}
2014-05-11 22:55:43 +00:00
}
current = append(current, CacheEntry{
Address: deviceAddr,
Seen: time.Now(),
})
done:
2014-05-11 22:55:43 +00:00
}
2014-05-11 22:55:43 +00:00
if debug {
l.Debugf("discover: register: %v -> %v", id, current)
2014-05-11 22:55:43 +00:00
}
2014-05-11 22:55:43 +00:00
d.registryLock.Lock()
d.registry[id] = current
2014-05-11 22:55:43 +00:00
d.registryLock.Unlock()
2014-07-17 11:38:36 +00:00
if len(current) > len(orig) {
addrs := make([]string, len(current))
for i := range current {
addrs[i] = current[i].Address
}
events.Default.Log(events.DeviceDiscovered, map[string]interface{}{
2014-09-28 11:05:25 +00:00
"device": id.String(),
"addrs": addrs,
2014-07-17 11:38:36 +00:00
})
}
return len(current) > len(orig)
2014-05-11 22:55:43 +00:00
}
func (d *Discoverer) externalLookup(device protocol.DeviceID) []string {
2014-01-26 13:28:41 +00:00
extIP, err := net.ResolveUDPAddr("udp", d.extServer)
2013-12-23 02:35:05 +00:00
if err != nil {
2014-05-15 00:08:56 +00:00
if debug {
l.Debugf("discover: %v; no external lookup", err)
}
2014-02-20 16:40:15 +00:00
return nil
2013-12-23 02:35:05 +00:00
}
conn, err := net.DialUDP("udp", nil, extIP)
2013-12-22 22:13:51 +00:00
if err != nil {
2014-05-15 00:08:56 +00:00
if debug {
l.Debugf("discover: %v; no external lookup", err)
}
2014-02-20 16:40:15 +00:00
return nil
2013-12-22 21:29:23 +00:00
}
2013-12-24 16:10:49 +00:00
defer conn.Close()
2013-12-22 21:29:23 +00:00
2013-12-24 16:10:49 +00:00
err = conn.SetDeadline(time.Now().Add(5 * time.Second))
2013-12-23 02:35:05 +00:00
if err != nil {
2014-05-15 00:08:56 +00:00
if debug {
l.Debugf("discover: %v; no external lookup", err)
}
2014-02-20 16:40:15 +00:00
return nil
2013-12-23 02:35:05 +00:00
}
2014-10-21 06:40:05 +00:00
buf := Query{QueryMagic, device[:]}.MustMarshalXDR()
2014-02-20 16:40:15 +00:00
_, err = conn.Write(buf)
2013-12-24 16:10:49 +00:00
if err != nil {
2014-05-15 00:08:56 +00:00
if debug {
l.Debugf("discover: %v; no external lookup", err)
}
2014-02-20 16:40:15 +00:00
return nil
2013-12-24 16:10:49 +00:00
}
2013-12-22 21:29:23 +00:00
2014-06-18 21:57:22 +00:00
buf = make([]byte, 2048)
2013-12-24 16:10:49 +00:00
n, err := conn.Read(buf)
if err != nil {
2013-12-24 16:15:21 +00:00
if err, ok := err.(net.Error); ok && err.Timeout() {
// Expected if the server doesn't know about requested device ID
2014-02-20 16:40:15 +00:00
return nil
2013-12-24 16:15:21 +00:00
}
2014-05-15 00:08:56 +00:00
if debug {
l.Debugf("discover: %v; no external lookup", err)
}
2014-02-20 16:40:15 +00:00
return nil
2013-12-24 16:10:49 +00:00
}
2013-12-22 21:29:23 +00:00
2014-03-09 07:58:03 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l.Debugf("discover: read external:\n%s", hex.Dump(buf[:n]))
2014-02-20 16:40:15 +00:00
}
var pkt Announce
2014-02-20 16:40:15 +00:00
err = pkt.UnmarshalXDR(buf[:n])
2014-05-11 22:55:43 +00:00
if err != nil && err != io.EOF {
2014-05-15 00:08:56 +00:00
if debug {
l.Debugln("discover:", err)
}
2014-02-20 16:40:15 +00:00
return nil
2013-12-24 16:10:49 +00:00
}
2013-12-22 21:29:23 +00:00
2014-02-20 16:40:15 +00:00
var addrs []string
2014-05-11 22:55:43 +00:00
for _, a := range pkt.This.Addresses {
deviceAddr := net.JoinHostPort(net.IP(a.IP).String(), strconv.Itoa(int(a.Port)))
addrs = append(addrs, deviceAddr)
2014-02-20 16:40:15 +00:00
}
return addrs
2013-12-22 21:29:23 +00:00
}
2014-05-11 22:55:43 +00:00
func (d *Discoverer) filterCached(c []CacheEntry) []CacheEntry {
for i := 0; i < len(c); {
if ago := time.Since(c[i].Seen); ago > d.cacheLifetime {
if debug {
l.Debugf("removing cached address %s: seen %v ago", c[i].Address, ago)
}
c[i] = c[len(c)-1]
c = c[:len(c)-1]
} else {
i++
}
}
return c
}
2014-05-11 22:55:43 +00:00
func addrToAddr(addr *net.TCPAddr) Address {
if len(addr.IP) == 0 || addr.IP.IsUnspecified() {
return Address{Port: uint16(addr.Port)}
} else if bs := addr.IP.To4(); bs != nil {
return Address{IP: bs, Port: uint16(addr.Port)}
} else if bs := addr.IP.To16(); bs != nil {
return Address{IP: bs, Port: uint16(addr.Port)}
}
return Address{}
}
func resolveAddrs(addrs []string) []Address {
var raddrs []Address
for _, addrStr := range addrs {
addrRes, err := net.ResolveTCPAddr("tcp", addrStr)
if err != nil {
continue
}
addr := addrToAddr(addrRes)
if len(addr.IP) > 0 {
raddrs = append(raddrs, addr)
2014-05-16 14:28:52 +00:00
} else {
raddrs = append(raddrs, Address{Port: addr.Port})
2014-05-11 22:55:43 +00:00
}
}
return raddrs
}