syncthing/discover/discover.go

401 lines
8.7 KiB
Go
Raw Normal View History

2014-06-01 20:50:14 +00:00
// Copyright (C) 2014 Jakob Borg and other contributors. All rights reserved.
// Use of this source code is governed by an MIT-style license that can be
// found in the LICENSE file.
2013-12-15 10:43:31 +00:00
package discover
import (
2014-02-20 16:40:15 +00:00
"encoding/hex"
"errors"
2013-12-15 10:43:31 +00:00
"fmt"
2014-05-11 22:55:43 +00:00
"io"
2013-12-15 10:43:31 +00:00
"net"
"sync"
"time"
2013-12-24 16:10:49 +00:00
2014-05-15 03:29:18 +00:00
"github.com/calmh/syncthing/beacon"
2013-12-24 16:10:49 +00:00
"github.com/calmh/syncthing/buffers"
2013-12-15 10:43:31 +00:00
)
type Discoverer struct {
myID string
listenAddrs []string
localBcastIntv time.Duration
globalBcastIntv time.Duration
2014-05-15 03:29:18 +00:00
beacon *beacon.Beacon
registry map[string][]string
registryLock sync.RWMutex
extServer string
2014-04-18 11:20:42 +00:00
extPort uint16
localBcastTick <-chan time.Time
forcedBcastTick chan time.Time
extAnnounceOK bool
extAnnounceOKmut sync.Mutex
2013-12-22 21:29:23 +00:00
}
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
)
2013-12-22 22:13:51 +00:00
// We tolerate a certain amount of errors because we might be running on
2013-12-18 17:29:15 +00:00
// laptops that sleep and wake, have intermittent network connectivity, etc.
// When we hit this many errors in succession, we stop.
const maxErrors = 30
func NewDiscoverer(id string, addresses []string, localPort int) (*Discoverer, error) {
b, err := beacon.New(localPort)
if err != nil {
return nil, err
}
2013-12-15 10:43:31 +00:00
disc := &Discoverer{
myID: id,
listenAddrs: addresses,
localBcastIntv: 30 * time.Second,
globalBcastIntv: 1800 * time.Second,
beacon: b,
registry: make(map[string][]string),
}
2013-12-15 10:43:31 +00:00
go disc.recvAnnouncements()
return disc, nil
}
func (d *Discoverer) StartLocal() {
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) {
d.extServer = server
2014-04-18 11:20:42 +00:00
d.extPort = extPort
go d.sendExternalAnnouncements()
}
func (d *Discoverer) ExtAnnounceOK() bool {
d.extAnnounceOKmut.Lock()
defer d.extAnnounceOKmut.Unlock()
return d.extAnnounceOK
}
2014-05-02 06:53:19 +00:00
func (d *Discoverer) Lookup(node string) []string {
d.registryLock.Lock()
addr, ok := d.registry[node]
d.registryLock.Unlock()
if ok {
return addr
} else if len(d.extServer) != 0 {
// We might want to cache this, but not permanently so it needs some intelligence
return d.externalLookup(node)
}
return nil
}
func (d *Discoverer) Hint(node string, addrs []string) {
resAddrs := resolveAddrs(addrs)
d.registerNode(nil, Node{
ID: node,
Addresses: resAddrs,
})
}
func (d *Discoverer) All() map[string][]string {
d.registryLock.RLock()
nodes := make(map[string][]string, len(d.registry))
for node, addrs := range d.registry {
addrsCopy := make([]string, len(addrs))
copy(addrsCopy, addrs)
nodes[node] = addrsCopy
}
d.registryLock.RUnlock()
return nodes
}
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 = AnnounceV2{
2014-05-11 22:55:43 +00:00
Magic: AnnouncementMagicV2,
This: Node{d.myID, addrs},
}
return pkt.MarshalXDR()
}
func (d *Discoverer) sendLocalAnnouncements() {
2014-05-11 22:55:43 +00:00
var addrs = resolveAddrs(d.listenAddrs)
var pkt = AnnounceV2{
Magic: AnnouncementMagicV2,
This: Node{d.myID, addrs},
}
2014-03-28 10:04:48 +00:00
for {
2014-05-11 22:55:43 +00:00
pkt.Extra = nil
d.registryLock.RLock()
for node, addrs := range d.registry {
if len(pkt.Extra) == 16 {
break
}
anode := Node{node, resolveAddrs(addrs)}
pkt.Extra = append(pkt.Extra, anode)
}
d.registryLock.RUnlock()
d.beacon.Send(pkt.MarshalXDR())
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() {
remote, err := net.ResolveUDPAddr("udp", d.extServer)
2013-12-22 21:29:23 +00:00
if err != nil {
2014-05-15 00:08:56 +00:00
l.Warnf("Global discovery: %v; no external announcements", err)
2013-12-22 21:29:23 +00:00
return
}
conn, err := net.ListenUDP("udp", nil)
if err != nil {
2014-05-15 00:08:56 +00:00
l.Warnf("Global discovery: %v; no external announcements", err)
return
}
2014-04-18 11:20:42 +00:00
var buf []byte
if d.extPort != 0 {
var pkt = AnnounceV2{
2014-05-11 22:55:43 +00:00
Magic: AnnouncementMagicV2,
This: Node{d.myID, []Address{{Port: d.extPort}}},
2014-04-18 11:20:42 +00:00
}
buf = pkt.MarshalXDR()
} else {
buf = d.announcementPkt()
}
for {
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 node 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 {
time.Sleep(d.globalBcastIntv)
} else {
time.Sleep(60 * time.Second)
2014-04-16 13:06:54 +00:00
}
2013-12-15 10:43:31 +00:00
}
}
func (d *Discoverer) recvAnnouncements() {
2014-03-28 10:04:48 +00:00
for {
buf, addr := d.beacon.Recv()
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 announcement:\n%s", hex.Dump(buf))
2014-02-20 16:40:15 +00:00
}
var pkt AnnounceV2
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
}
2014-03-09 07:58:03 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l.Debugf("discover: parsed announcement: %#v", pkt)
2014-02-20 16:40:15 +00:00
}
2014-05-11 22:55:43 +00:00
var newNode bool
if pkt.This.ID != d.myID {
n := d.registerNode(addr, pkt.This)
newNode = newNode || n
for _, node := range pkt.Extra {
if node.ID != d.myID {
n := d.registerNode(nil, node)
newNode = newNode || n
}
2013-12-15 10:43:31 +00:00
}
2014-05-11 22:55:43 +00:00
}
if newNode {
select {
case d.forcedBcastTick <- time.Now():
}
2013-12-15 10:43:31 +00:00
}
}
}
2014-05-11 22:55:43 +00:00
func (d *Discoverer) registerNode(addr net.Addr, node Node) bool {
var addrs []string
for _, a := range node.Addresses {
var nodeAddr string
if len(a.IP) > 0 {
nodeAddr = fmt.Sprintf("%s:%d", net.IP(a.IP), a.Port)
addrs = append(addrs, nodeAddr)
} else if addr != nil {
ua := addr.(*net.UDPAddr)
ua.Port = int(a.Port)
nodeAddr = ua.String()
addrs = append(addrs, nodeAddr)
}
}
if len(addrs) == 0 {
if debug {
2014-05-15 00:08:56 +00:00
l.Debugln("discover: no valid address for", node.ID)
2014-05-11 22:55:43 +00:00
}
}
if debug {
2014-05-15 00:08:56 +00:00
l.Debugf("discover: register: %s -> %#v", node.ID, addrs)
2014-05-11 22:55:43 +00:00
}
d.registryLock.Lock()
_, seen := d.registry[node.ID]
d.registry[node.ID] = addrs
d.registryLock.Unlock()
return !seen
}
2014-02-20 16:40:15 +00:00
func (d *Discoverer) externalLookup(node string) []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-02-20 16:40:15 +00:00
buf := QueryV2{QueryMagicV2, node}.MarshalXDR()
_, 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
}
2014-02-20 16:40:15 +00:00
buffers.Put(buf)
2013-12-22 21:29:23 +00:00
2014-05-11 22:55:43 +00:00
buf = buffers.Get(2048)
2013-12-24 16:10:49 +00:00
defer buffers.Put(buf)
2013-12-22 21:29:23 +00:00
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 node 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 AnnounceV2
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-03-09 07:58:03 +00:00
if debug {
2014-05-15 00:08:56 +00:00
l.Debugf("discover: parsed external: %#v", pkt)
2013-12-22 21:29:23 +00:00
}
2013-12-24 16:10:49 +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 {
2014-05-02 06:53:19 +00:00
nodeAddr := fmt.Sprintf("%s:%d", net.IP(a.IP), a.Port)
2014-02-20 16:40:15 +00:00
addrs = append(addrs, nodeAddr)
}
return addrs
2013-12-22 21:29:23 +00:00
}
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
}