Merge pull request #2184 from calmh/mc

Multicast double whammy!
This commit is contained in:
Audrius Butkevicius
2015-08-23 14:28:44 +01:00
89 changed files with 7758 additions and 61 deletions

View File

@@ -18,7 +18,11 @@ type Interface interface {
Recv() ([]byte, net.Addr)
}
func genericReader(conn *net.UDPConn, outbox chan<- recv) {
type readerFrom interface {
ReadFrom([]byte) (int, net.Addr, error)
}
func genericReader(conn readerFrom, outbox chan<- recv) {
bs := make([]byte, 65536)
for {
n, addr, err := conn.ReadFrom(bs)

View File

@@ -6,38 +6,65 @@
package beacon
import "net"
import (
"errors"
"fmt"
"net"
"golang.org/x/net/ipv6"
)
type Multicast struct {
conn *net.UDPConn
conn *ipv6.PacketConn
addr *net.UDPAddr
intf *net.Interface
inbox chan []byte
outbox chan recv
intfs []net.Interface
}
func NewMulticast(addr, ifname string) (*Multicast, error) {
func NewMulticast(addr string) (*Multicast, error) {
gaddr, err := net.ResolveUDPAddr("udp6", addr)
if err != nil {
return nil, err
}
intf, err := net.InterfaceByName(ifname)
conn, err := net.ListenPacket("udp6", fmt.Sprintf("[::]:%d", gaddr.Port))
if err != nil {
return nil, err
}
conn, err := net.ListenMulticastUDP("udp6", intf, gaddr)
if err != nil {
return nil, err
}
b := &Multicast{
conn: conn,
addr: gaddr,
intf: intf,
inbox: make(chan []byte),
outbox: make(chan recv, 16),
}
go genericReader(b.conn, b.outbox)
intfs, err := net.Interfaces()
if err != nil {
return nil, err
}
p := ipv6.NewPacketConn(conn)
joined := 0
for _, intf := range intfs {
err := p.JoinGroup(&intf, &net.UDPAddr{IP: gaddr.IP})
if debug {
if err != nil {
l.Debugln("IPv6 join", intf.Name, "failed:", err)
} else {
l.Debugln("IPv6 join", intf.Name, "success")
}
}
joined++
}
if joined == 0 {
return nil, errors.New("no multicast interfaces available")
}
b := &Multicast{
conn: p,
addr: gaddr,
inbox: make(chan []byte),
outbox: make(chan recv, 16),
intfs: intfs,
}
go genericReader(ipv6ReaderAdapter{b.conn}, b.outbox)
go b.writer()
return b, nil
@@ -53,14 +80,30 @@ func (b *Multicast) Recv() ([]byte, net.Addr) {
}
func (b *Multicast) writer() {
addr := *b.addr
addr.Zone = b.intf.Name
wcm := &ipv6.ControlMessage{
HopLimit: 1,
}
for bs := range b.inbox {
_, err := b.conn.WriteTo(bs, &addr)
if err != nil && debug {
l.Debugln(err, "on write to", addr)
} else if debug {
l.Debugf("sent %d bytes to %s", len(bs), addr.String())
for _, intf := range b.intfs {
wcm.IfIndex = intf.Index
_, err := b.conn.WriteTo(bs, wcm, b.addr)
if err != nil && debug {
l.Debugln(err, "on write to", b.addr)
} else if debug {
l.Debugf("sent %d bytes to %v on %s", len(bs), b.addr, intf.Name)
}
}
}
}
// This makes ReadFrom on an *ipv6.PacketConn behave like ReadFrom on a
// net.PacketConn.
type ipv6ReaderAdapter struct {
c *ipv6.PacketConn
}
func (i ipv6ReaderAdapter) ReadFrom(bs []byte) (int, net.Addr, error) {
n, _, src, err := i.c.ReadFrom(bs)
return n, src, err
}

View File

@@ -218,7 +218,7 @@ type OptionsConfiguration struct {
GlobalAnnEnabled bool `xml:"globalAnnounceEnabled" json:"globalAnnounceEnabled" default:"true"`
LocalAnnEnabled bool `xml:"localAnnounceEnabled" json:"localAnnounceEnabled" default:"true"`
LocalAnnPort int `xml:"localAnnouncePort" json:"localAnnouncePort" default:"21025"`
LocalAnnMCAddr string `xml:"localAnnounceMCAddr" json:"localAnnounceMCAddr" default:"[ff32::5222]:21026"`
LocalAnnMCAddr string `xml:"localAnnounceMCAddr" json:"localAnnounceMCAddr" default:"[ff12::8384]:21027"`
RelayServers []string `xml:"relayServer" json:"relayServers" default:"dynamic+https://relays.syncthing.net"`
MaxSendKbps int `xml:"maxSendKbps" json:"maxSendKbps"`
MaxRecvKbps int `xml:"maxRecvKbps" json:"maxRecvKbps"`
@@ -504,6 +504,11 @@ func convertV11V12(cfg *Configuration) {
}
}
// Use new multicast group
if cfg.Options.LocalAnnMCAddr == "[ff32::5222]:21026" {
cfg.Options.LocalAnnMCAddr = "[ff12::8384]:21027"
}
cfg.Version = 12
}

View File

@@ -36,7 +36,7 @@ func TestDefaultValues(t *testing.T) {
GlobalAnnEnabled: true,
LocalAnnEnabled: true,
LocalAnnPort: 21025,
LocalAnnMCAddr: "[ff32::5222]:21026",
LocalAnnMCAddr: "[ff12::8384]:21027",
RelayServers: []string{"dynamic+https://relays.syncthing.net"},
MaxSendKbps: 0,
MaxRecvKbps: 0,

View File

@@ -14,7 +14,6 @@ import (
"io"
"net"
"net/url"
"runtime"
"sort"
"time"
@@ -101,38 +100,13 @@ func (d *Discoverer) startLocalIPv4Broadcasts(localPort int) {
}
func (d *Discoverer) startLocalIPv6Multicasts(localMCAddr string) {
intfs, err := net.Interfaces()
mb, err := beacon.NewMulticast(localMCAddr)
if err != nil {
if debug {
l.Debugln("discover: interfaces:", err)
}
l.Infoln("Local discovery over IPv6 unavailable")
return
}
v6Intfs := 0
for _, intf := range intfs {
// Interface flags seem to always be 0 on Windows
if runtime.GOOS != "windows" && (intf.Flags&net.FlagUp == 0 || intf.Flags&net.FlagMulticast == 0) {
continue
}
mb, err := beacon.NewMulticast(localMCAddr, intf.Name)
if err != nil {
if debug {
l.Debugln("discover: Start local v6:", err)
}
continue
}
d.beacons = append(d.beacons, mb)
go d.recvAnnouncements(mb)
v6Intfs++
}
if v6Intfs == 0 {
l.Infoln("Local discovery over IPv6 unavailable")
}
d.beacons = append(d.beacons, mb)
go d.recvAnnouncements(mb)
}
func (d *Discoverer) StartGlobal(servers []string, extPort uint16) {