Compare commits

...
Author SHA1 Message Date
David Wragg f9d25255c6 Changelog entry to accompany the preview release 2015-06-12 21:31:48 +01:00
David Wragg 3e55603f3f Disable MAC cache expiry
When ODP flows are taking care of routing packets, the router might
not hear about a given MAC address for a long time, and so the MAC
cache will expire it.  For now, simply disable expiry of MAC cache
entries.
2015-06-12 21:24:15 +01:00
David Wragg ec779c5ebd Set the MTU of the overlay network
We need to set the MTU of the overlay network accurately, because with
fast datapath we can't inject the ICMP frag-needed packets to cause a
lower path MTU.  Later on we can refine this to try to try to
automatically detect a good cluster-wide MTU value.  But for now we
default to a conservative value that should work on almost all local
networks, with support for manually overriding it where a higher MTU
gives better performance.
2015-06-12 21:24:15 +01:00
David Wragg 62de82f349 Fast datapath 2015-06-12 21:24:10 +01:00
David Wragg 00b3d5bf98 Run weave container with --net=none
This is the simplest thing that kind of works.  It needs cleaning
up and extending to work with all weave features.
2015-06-12 16:22:13 +01:00
David Wragg f522f5b730 Ensure we send a gossip update whenever a new connection requires it
Previously, if LocalConnection run() failed before calling
AddConnection, it would not gossip anything. This was benign, because
the peer information we learn in the handshake is just a placeholder
until the full information arrived via gossip, and if a connection is
not formed, there is nothing else to gossip about.

With short ids, there's a corner case where this is no longer true: If
a new peer connects to us, and it has the same short id as we do, then
we might change our short id.  If we do, then it is important that we
gossip that out, even if the attempt to establish the connection fails
further along.
2015-06-12 16:22:13 +01:00
David Wragg bd185abb29 Introduce PeerShortIDs
These are needed to support vxlan encapsulation.  In a vxlan packet,
the only place we can straightforwardly hold the source and
destination peers identifiers used by the weave router is in the
24-bit VNI (vxlan network identifier) field.  This gives us 12 bits to
indicate a peer. The resulting limit of 4096 peers on a weave network
does not seem like a problem in the near future.  But it does mean
that decentralized allocation makes collisions fairly likely even in
weave networks of sizes that are common today.  Fortunately, the
problem is much more tractable than IPAM because we can detect when
collisions occur and take action to resolve them.

TODO: We already notify the InterHost when a short id -> peer mapping
changes, but we also need to flush any entries in the router MAC cache
that might be affected.
2015-06-12 16:22:13 +01:00
David Wragg 203dfbb3e9 Always include local peer in Peers 2015-06-12 16:22:13 +01:00
David Wragg c09f0f7037 Fold ApplyUpdate and applyUpdate together
In order to do "defer peers.Unlock()". ApplyUpdate now does slightly
more work with the lock held, but the difference is modest.
2015-06-12 16:21:16 +01:00
David Wragg a334059e85 Add NewPeerFromSummary
To reduce the number of places that are explicit about which values
go into a new Peer.
2015-06-12 16:12:59 +01:00
David Wragg a9c710211b Notify the InterHost when routes change 2015-06-12 16:12:59 +01:00
David Wragg 0f888e9735 Flows 2015-06-12 16:12:59 +01:00
David Wragg 17c8a16757 Move ICMP frag-needed handling into sleeveForwarder 2015-06-12 16:12:59 +01:00
David Wragg 4e41f649fb Introduce InterHost / sleeve 2015-06-12 16:12:59 +01:00
David Wragg f8ed319428 Don't inspect the DF flag earlier than we need to 2015-06-12 16:12:58 +01:00
David Wragg ef0153cb0f Introduce IntraHost 2015-06-12 16:12:54 +01:00
David Wragg b77abf480b Move DF indication into LogFrame 2015-06-12 16:11:32 +01:00
David Wragg 2fa5cda2c7 Make the full EthernetDecoder available to LogFrame
So that LogFrame can include information about the decoded frame beyond
the ethernet layer.
2015-06-12 16:11:32 +01:00
David Wragg 225e6871e3 Add a comment to Router's Start method 2015-06-12 16:11:32 +01:00
David Wragg 5d2d857dbe Eliminate the GossipSender Start method 2015-06-12 16:11:32 +01:00
David Wragg d7c4bc04d3 Eliminate ConnectionMaker Start method 2015-06-12 16:11:32 +01:00
David Wragg 32b2bcf01b Eliminate the Routes Start method 2015-06-12 16:11:32 +01:00
David Wragg b1308823d2 Eliminate the LocalPeer Start method 2015-06-12 16:11:32 +01:00
David Wragg 0627ade9de Eliminate MacCache Start method 2015-06-12 16:11:32 +01:00
David Wragg 4d0ca39f97 Eliminate the Start method on LocalConnection 2015-06-12 16:11:32 +01:00
David Wragg 8a54af92ef Remove forwarder Start methods 2015-06-12 16:11:32 +01:00
29 changed files with 3129 additions and 1387 deletions

No files matched your search

+5
View File
@@ -1,3 +1,8 @@
## Fast datapath preview release 20150612
This is a preview release for fast datapath. Its release notes are
in an associated blog post.
## Release 0.11.2
This is a bug fix release, addressing the following issues:
+109 -197
View File
@@ -1,7 +1,6 @@
package router
import (
"encoding/binary"
"encoding/gob"
"fmt"
"log"
@@ -40,25 +39,16 @@ type RemoteConnection struct {
type LocalConnection struct {
sync.RWMutex
RemoteConnection
TCPConn *net.TCPConn
tcpSender TCPSender
remoteUDPAddr *net.UDPAddr
receivedHeartbeat bool
stackFrag bool
effectivePMTU int
SessionKey *[32]byte
heartbeatTCP *time.Ticker
heartbeatTimeout *time.Timer
heartbeatFrame *ForwardedFrame
heartbeat *time.Ticker
fragTest *time.Ticker
forwarder *Forwarder
forwarderDF *ForwarderDF
Decryptor Decryptor
Router *Router
uid uint64
actionChan chan<- ConnectionAction
finished <-chan struct{} // closed to signal that actorLoop has finished
TCPConn *net.TCPConn
tcpSender TCPSender
remoteUDPAddr *net.UDPAddr
SessionKey *[32]byte
heartbeatTCP *time.Ticker
Router *Router
uid uint64
actionChan chan<- ConnectionAction
finished <-chan struct{} // closed to signal that actorLoop has finished
forwarder InterHostForwarder
}
type ConnectionAction func() error
@@ -97,26 +87,22 @@ func (conn *RemoteConnection) String() string {
return fmt.Sprint("Connection ", from, "->", to)
}
func NewLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) *LocalConnection {
// Does not return anything. If the connection is successful, it will
// end up in the local peer's connections map.
func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router, acceptNewPeer bool) {
if connRemote.local != router.Ourself.Peer {
log.Fatal("Attempt to create local connection from a peer which is not ourself")
}
// NB, we're taking a copy of connRemote here.
return &LocalConnection{
actionChan := make(chan ConnectionAction, ChannelSize)
finished := make(chan struct{})
conn := &LocalConnection{
RemoteConnection: *connRemote,
Router: router,
TCPConn: tcpConn,
remoteUDPAddr: udpAddr,
effectivePMTU: DefaultPMTU}
}
// Async. Does not return anything. If the connection is successful,
// it will end up in the local peer's connections map.
func (conn *LocalConnection) Start(acceptNewPeer bool) {
actionChan := make(chan ConnectionAction, ChannelSize)
conn.actionChan = actionChan
finished := make(chan struct{})
conn.finished = finished
actionChan: actionChan,
finished: finished}
go conn.run(actionChan, finished, acceptNewPeer)
}
@@ -133,49 +119,12 @@ func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak {
}
}
// Read by the forwarder processes when in the UDP senders
func (conn *LocalConnection) RemoteUDPAddr() *net.UDPAddr {
conn.RLock()
defer conn.RUnlock()
return conn.remoteUDPAddr
}
func (conn *LocalConnection) Established() bool {
conn.RLock()
defer conn.RUnlock()
return conn.established
}
// Called by forwarder processes, read in Forward (by sniffer and udp
// listener process in router).
func (conn *LocalConnection) setEffectivePMTU(pmtu int) {
conn.Lock()
defer conn.Unlock()
if conn.effectivePMTU != pmtu {
conn.effectivePMTU = pmtu
conn.Log("Effective PMTU set to", pmtu)
}
}
// Called by the connection's actor process, and by the connection's
// TCP received process. StackFrag is read in conn.Forward (called by
// router udp listener and sniffer processes)
func (conn *LocalConnection) setStackFrag(frag bool) {
conn.Lock()
defer conn.Unlock()
conn.stackFrag = frag
}
// Called by the connection's TCP receiver process.
func (conn *LocalConnection) pmtuVerified(pmtu int) {
conn.RLock()
fwd := conn.forwarderDF
conn.RUnlock()
if fwd != nil {
fwd.PMTUVerified(pmtu)
}
}
// Send directly, not via the Actor. If it goes via the Actor we can
// get a deadlock where LocalConnection is blocked talking to
// LocalPeer and LocalPeer is blocked trying send a ProtocolMsg via
@@ -205,71 +154,6 @@ func (conn *LocalConnection) Shutdown(err error) {
go func() { conn.sendAction(func() error { return err }) }()
}
// Async
//
// Heartbeating serves two purposes: a) keeping NAT paths alive, and
// b) updating a remote peer's knowledge of our address, in the event
// it changes (e.g. because NAT paths expired).
func (conn *LocalConnection) ReceivedHeartbeat(remoteUDPAddr *net.UDPAddr, connUID uint64) {
if remoteUDPAddr == nil || connUID != conn.uid {
return
}
conn.sendAction(func() error {
oldRemoteUDPAddr := conn.remoteUDPAddr
old := conn.receivedHeartbeat
conn.Lock()
conn.remoteUDPAddr = remoteUDPAddr
conn.receivedHeartbeat = true
conn.Unlock()
conn.heartbeatTimeout.Reset(HeartbeatTimeout)
if !old {
if err := conn.sendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil {
return err
}
}
if oldRemoteUDPAddr == nil {
return conn.sendFastHeartbeats()
} else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() {
log.Println("Peer", conn.remote, "moved from", old, "to", remoteUDPAddr)
}
return nil
})
}
// Async
func (conn *LocalConnection) SetEstablished() {
conn.sendAction(func() error {
stopTicker(conn.heartbeat)
old := conn.established
conn.Lock()
conn.established = true
conn.Unlock()
if old {
return nil
}
conn.Router.Ourself.ConnectionEstablished(conn)
if err := conn.ensureForwarders(); err != nil {
return err
}
// Send a large frame down the DF channel in order to prompt
// PMTU discovery to start.
conn.Forward(true, &ForwardedFrame{
srcPeer: conn.local,
dstPeer: conn.remote,
frame: PMTUDiscovery},
nil)
conn.heartbeat = time.NewTicker(SlowHeartbeat)
conn.fragTest = time.NewTicker(FragTestInterval)
// avoid initial waits for timers to fire
conn.Forward(true, conn.heartbeatFrame, nil)
conn.setStackFrag(false)
if err := conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
return err
}
return nil
})
}
// Send an actor request to the actorLoop, but don't block if
// actorLoop has exited - see http://blog.golang.org/pipelines for
// pattern
@@ -287,31 +171,52 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
defer func() { conn.shutdown(err) }()
defer close(finished)
dec, err := conn.prepareConnection(acceptNewPeer)
if err != nil {
return
}
// Tell the forwarder that the connection is confirmed
conn.forwarder.SetListener(ConnectionAsForwarderListener{conn})
conn.heartbeatTCP = time.NewTicker(TCPHeartbeat)
go conn.receiveTCP(dec)
err = conn.actorLoop(actionChan)
}
func (conn *LocalConnection) prepareConnection(acceptNewPeer bool) (*gob.Decoder, error) {
tcpConn := conn.TCPConn
tcpConn.SetLinger(0)
enc := gob.NewEncoder(tcpConn)
dec := gob.NewDecoder(tcpConn)
if err = conn.handshake(enc, dec, acceptNewPeer); err != nil {
return
remotePeer, err := conn.handshake(enc, dec, acceptNewPeer)
if err != nil {
return nil, err
}
conn.Log("completed handshake")
if err = conn.setRemote(remotePeer); err != nil {
return nil, err
}
// setRemote may have resulted in the local short id changing.
// So we need to ensure we gossip that, even if we error out
// below.
defer conn.Router.broadcastPeerUpdate(remotePeer)
// The ordering of the following is very important. [1]
if conn.remoteUDPAddr != nil {
if err = conn.ensureForwarders(); err != nil {
return
}
localIP := tcpConn.LocalAddr().(*net.TCPAddr).IP
if conn.forwarder, err = conn.Router.InterHost.MakeForwarder(conn.remote, localIP, conn.remoteUDPAddr, conn.uid, conn.udpCrypto(), conn.sendInterHostControlMessage); err != nil {
return nil, err
}
if err = conn.Router.Ourself.AddConnection(conn); err != nil {
return
return nil, err
}
if err = conn.initHeartbeats(); err != nil {
return
}
go conn.receiveTCP(dec)
err = conn.actorLoop(actionChan)
return dec, nil
}
// [1] Ordering constraints:
@@ -359,19 +264,18 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch
// prevent that completely, since, for example, forwarder can only be
// created when we know the remote UDP address, but it helps to try.
func (conn *LocalConnection) initHeartbeats() error {
conn.heartbeatTCP = time.NewTicker(TCPHeartbeat)
conn.heartbeatTimeout = time.NewTimer(HeartbeatTimeout)
heartbeatFrameBytes := make([]byte, EthernetOverhead+8)
binary.BigEndian.PutUint64(heartbeatFrameBytes[EthernetOverhead:], conn.uid)
conn.heartbeatFrame = &ForwardedFrame{
srcPeer: conn.local,
dstPeer: conn.remote,
frame: heartbeatFrameBytes}
if conn.remoteUDPAddr == nil {
func (conn *LocalConnection) setRemote(toPeer *Peer) error {
toPeer = conn.Router.Peers.FetchWithDefault(toPeer)
switch toPeer {
case nil:
return fmt.Errorf("multiple peers with the same Name")
case conn.local:
conn.remote = toPeer // have to do assigment here to ensure Shutdown releases ref count
return fmt.Errorf("Cannot connect to ourself")
default:
conn.remote = toPeer
return nil
}
return conn.sendFastHeartbeats()
}
func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) {
@@ -381,13 +285,6 @@ func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err
err = action()
case <-conn.heartbeatTCP.C:
err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat)
case <-conn.heartbeatTimeout.C:
err = fmt.Errorf("timed out waiting for UDP heartbeat")
case <-tickerChan(conn.heartbeat):
conn.Forward(true, conn.heartbeatFrame, nil)
case <-tickerChan(conn.fragTest):
conn.setStackFrag(false)
err = conn.sendSimpleProtocolMsg(ProtocolStartFragmentationTest)
}
}
return
@@ -409,21 +306,55 @@ func (conn *LocalConnection) shutdown(err error) {
conn.Router.Ourself.DeleteConnection(conn)
}
if conn.heartbeatTimeout != nil {
conn.heartbeatTimeout.Stop()
stopTicker(conn.heartbeatTCP)
if conn.forwarder != nil {
conn.forwarder.Close()
}
stopTicker(conn.heartbeatTCP)
stopTicker(conn.heartbeat)
stopTicker(conn.fragTest)
// blank out the forwardChan so that the router processes don't
// try to send any more
conn.stopForwarders()
conn.Router.ConnectionMaker.ConnectionTerminated(conn.remoteTCPAddr, err)
}
func (conn *LocalConnection) udpCrypto() InterHostCrypto {
name := conn.local.NameByte
if conn.Router.UsingPassword() {
return InterHostCrypto{
Dec: NewNaClDecryptor(conn.SessionKey, conn.outbound),
Enc: NewNaClEncryptor(name, conn.SessionKey, conn.outbound, false),
EncDF: NewNaClEncryptor(name, conn.SessionKey, conn.outbound, true),
}
} else {
return InterHostCrypto{
Dec: NewNonDecryptor(),
Enc: NewNonEncryptor(name),
EncDF: NewNonEncryptor(name),
}
}
}
func (conn *LocalConnection) sendInterHostControlMessage(msg []byte) error {
return conn.sendProtocolMsg(ProtocolMsg{ProtocolInterHostControlMsg, msg})
}
type ConnectionAsForwarderListener struct{ conn *LocalConnection }
func (l ConnectionAsForwarderListener) Established() {
l.conn.sendAction(func() error {
old := l.conn.established
l.conn.Lock()
l.conn.established = true
l.conn.Unlock()
if !old {
l.conn.Router.Ourself.ConnectionEstablished(l.conn)
}
return nil
})
}
func (l ConnectionAsForwarderListener) Error(err error) {
l.conn.sendAction(func() error { return err })
}
// Helpers
func (conn *LocalConnection) sendSimpleProtocolMsg(tag ProtocolTag) error {
@@ -467,22 +398,8 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) error {
switch tag {
case ProtocolHeartbeat:
case ProtocolConnectionEstablished:
// We sent fast heartbeats to the remote peer, which has now
// received at least one of them and told us via this message.
// We can now consider the connection as established from our
// end.
conn.SetEstablished()
case ProtocolStartFragmentationTest:
conn.Forward(false, &ForwardedFrame{
srcPeer: conn.local,
dstPeer: conn.remote,
frame: FragTest},
nil)
case ProtocolFragmentationReceived:
conn.setStackFrag(true)
case ProtocolPMTUVerified:
conn.pmtuVerified(int(binary.BigEndian.Uint16(payload)))
case ProtocolInterHostControlMsg:
conn.forwarder.ControlMessage(payload)
case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip:
return conn.Router.handleGossip(tag, payload)
default:
@@ -495,13 +412,8 @@ func (conn *LocalConnection) extendReadDeadline() {
conn.TCPConn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2))
}
func (conn *LocalConnection) sendFastHeartbeats() error {
err := conn.ensureForwarders()
if err == nil {
conn.heartbeat = time.NewTicker(FastHeartbeat)
conn.Forward(true, conn.heartbeatFrame, nil) // avoid initial wait
}
return err
func (conn *LocalConnection) Forward(key ForwardPacketKey) FlowOp {
return conn.forwarder.Forward(key)
}
func tickerChan(ticker *time.Ticker) <-chan time.Time {
+5 -7
View File
@@ -34,18 +34,16 @@ type Target struct {
type ConnectionMakerAction func() bool
func NewConnectionMaker(ourself *LocalPeer, peers *Peers, port int) *ConnectionMaker {
return &ConnectionMaker{
actionChan := make(chan ConnectionMakerAction, ChannelSize)
cm := &ConnectionMaker{
ourself: ourself,
peers: peers,
port: port,
cmdLinePeers: make(map[string]*net.TCPAddr),
targets: make(map[string]*Target)}
}
func (cm *ConnectionMaker) Start() {
actionChan := make(chan ConnectionMakerAction, ChannelSize)
cm.actionChan = actionChan
targets: make(map[string]*Target),
actionChan: actionChan}
go cm.queryLoop(actionChan)
return cm
}
func (cm *ConnectionMaker) InitiateConnection(peer string) error {
+27 -23
View File
@@ -9,15 +9,15 @@ import (
)
type EthernetDecoder struct {
eth layers.Ethernet
ip layers.IPv4
Eth layers.Ethernet
IP layers.IPv4
decoded []gopacket.LayerType
parser *gopacket.DecodingLayerParser
}
func NewEthernetDecoder() *EthernetDecoder {
dec := &EthernetDecoder{}
dec.parser = gopacket.NewDecodingLayerParser(layers.LayerTypeEthernet, &dec.eth, &dec.ip)
dec.parser = gopacket.NewDecodingLayerParser(layers.LayerTypeEthernet, &dec.Eth, &dec.IP)
return dec
}
@@ -25,52 +25,56 @@ func (dec *EthernetDecoder) DecodeLayers(data []byte) error {
return dec.parser.DecodeLayers(data, &dec.decoded)
}
func (dec *EthernetDecoder) sendICMPFragNeeded(mtu int, sendFrame func([]byte) error) error {
func (dec *EthernetDecoder) PacketKey() (key PacketKey) {
copy(key.SrcMAC[:], dec.Eth.SrcMAC)
copy(key.DstMAC[:], dec.Eth.DstMAC)
return
}
func (dec *EthernetDecoder) makeICMPFragNeeded(mtu int) ([]byte, error) {
buf := gopacket.NewSerializeBuffer()
opts := gopacket.SerializeOptions{
FixLengths: true,
ComputeChecksums: true}
ipHeaderSize := int(dec.ip.IHL) * 4 // IHL is the number of 32-byte words in the header
payload := gopacket.Payload(dec.ip.BaseLayer.Contents[:ipHeaderSize+8])
ipHeaderSize := int(dec.IP.IHL) * 4 // IHL is the number of 32-byte words in the header
payload := gopacket.Payload(dec.IP.BaseLayer.Contents[:ipHeaderSize+8])
err := gopacket.SerializeLayers(buf, opts,
&layers.Ethernet{
SrcMAC: dec.eth.DstMAC,
DstMAC: dec.eth.SrcMAC,
EthernetType: dec.eth.EthernetType},
SrcMAC: dec.Eth.DstMAC,
DstMAC: dec.Eth.SrcMAC,
EthernetType: dec.Eth.EthernetType},
&layers.IPv4{
Version: 4,
TOS: dec.ip.TOS,
TOS: dec.IP.TOS,
Id: 0,
Flags: 0,
FragOffset: 0,
TTL: 64,
Protocol: layers.IPProtocolICMPv4,
DstIP: dec.ip.SrcIP,
SrcIP: dec.ip.DstIP},
DstIP: dec.IP.SrcIP,
SrcIP: dec.IP.DstIP},
&layers.ICMPv4{
TypeCode: 0x304,
Id: 0,
Seq: uint16(mtu)},
&payload)
if err != nil {
return err
return nil, err
}
log.Printf("Sending ICMP 3,4 (%v -> %v): PMTU= %v\n", dec.ip.DstIP, dec.ip.SrcIP, mtu)
return sendFrame(buf.Bytes())
log.Printf("Sending ICMP 3,4 (%v -> %v): PMTU= %v\n", dec.IP.DstIP, dec.IP.SrcIP, mtu)
return buf.Bytes(), nil
}
var (
// see http://en.wikipedia.org/wiki/Multicast_address#Ethernet
stpMACPrefix = []byte{0x01, 0x80, 0xC2, 0x00, 0x00}
zeroMAC, _ = net.ParseMAC("00:00:00:00:00:00")
zeroMAC, _ = net.ParseMAC("00:00:00:00:00:00")
)
func (dec *EthernetDecoder) DropFrame() bool {
return bytes.Equal(stpMACPrefix, dec.eth.DstMAC[:len(stpMACPrefix)])
func (dec *EthernetDecoder) IsSpecial() bool {
return dec.Eth.Length == 0 && dec.Eth.EthernetType == layers.EthernetTypeLLC &&
bytes.Equal(zeroMAC, dec.Eth.SrcMAC) && bytes.Equal(zeroMAC, dec.Eth.DstMAC)
}
func (dec *EthernetDecoder) IsSpecial() bool {
return dec.eth.Length == 0 && dec.eth.EthernetType == layers.EthernetTypeLLC &&
bytes.Equal(zeroMAC, dec.eth.SrcMAC) && bytes.Equal(zeroMAC, dec.eth.DstMAC)
func (dec *EthernetDecoder) DF() bool {
return len(dec.decoded) == 2 && (dec.IP.Flags&layers.IPv4DontFragment != 0)
}
+593
View File
@@ -0,0 +1,593 @@
package router
import (
"encoding/binary"
"fmt"
"log"
"net"
"sync"
"github.com/weaveworks/go-odp/odp"
)
type thunk func() FlowOp
// A bridgePortID is either an ODP vport or the router. We express it
// like this to use it as a map key.
type bridgePortID struct {
vport odp.VportID
router bool
}
var routerPortID = bridgePortID{router: true}
type FastDatapath struct {
vxlanPort int
// The lock guards rthe FastDatapath state, and also
// synchronises use of the dpif
lock sync.Mutex
dpif *odp.Dpif
dp odp.DatapathHandle
consumingMisses bool
missHandlers map[odp.VportID]func(odp.FlowKeys) thunk
vxlanVportID odp.VportID
localPeer *Peer
peers *Peers
interHostConsumer InterHostConsumer
// Bridge state: A bridge port is represented as a function
// that says how to send out a packet on that port.
bridgePorts map[bridgePortID]func(PacketKey) thunk
bridgeMACs map[MAC]func(PacketKey) thunk
// Only accessed from Miss, so not locked
dec *EthernetDecoder
}
func NewFastDatapath(dpname string, vxlanPort int) (*FastDatapath, error) {
dpif, err := odp.NewDpif()
if err != nil {
return nil, err
}
success := false
defer func() {
if !success {
dpif.Close()
}
}()
dp, err := dpif.LookupDatapath(dpname)
if err != nil {
return nil, err
}
fastdp := &FastDatapath{
vxlanPort: vxlanPort,
dpif: dpif,
dp: dp,
missHandlers: make(map[odp.VportID]func(odp.FlowKeys) thunk),
bridgePorts: make(map[bridgePortID]func(PacketKey) thunk),
bridgeMACs: make(map[MAC]func(PacketKey) thunk),
dec: NewEthernetDecoder(),
}
if err := fastdp.deleteVxlanVports(); err != nil {
return nil, err
}
if err := fastdp.clearFlows(); err != nil {
return nil, err
}
fastdp.vxlanVportID, err = fastdp.dp.CreateVport(
odp.NewVxlanVportSpec("vxlan", uint16(vxlanPort)))
if err != nil {
return nil, err
}
success = true
return fastdp, nil
}
func (fastdp *FastDatapath) consumeMisses() error {
if !fastdp.consumingMisses {
if err := fastdp.dp.ConsumeMisses(fastdp); err != nil {
return err
}
fastdp.consumingMisses = true
}
return nil
}
func (fastdp *FastDatapath) Close() error {
err := fastdp.dpif.Close()
fastdp.dpif = nil
return err
}
func (fastdp *FastDatapath) clearFlows() error {
flows, err := fastdp.dp.EnumerateFlows()
if err != nil {
return err
}
for _, flow := range flows {
err = fastdp.dp.DeleteFlow(flow.FlowSpec)
if err != nil && !odp.IsNoSuchFlowError(err) {
return err
}
}
return nil
}
func (fastdp *FastDatapath) deleteVxlanVports() error {
vports, err := fastdp.dp.EnumerateVports()
if err != nil {
return err
}
for _, vport := range vports {
if vport.Spec.TypeName() != "vxlan" {
continue
}
err = fastdp.dp.DeleteVport(vport.ID)
if err != nil && !odp.IsNoSuchVportError(err) {
return err
}
}
return nil
}
func (fastdp *FastDatapath) Error(err error, stopped bool) {
// XXX fatal if stopped
log.Println("Error while listening on datapath:", err)
}
func (fastdp *FastDatapath) Miss(packet []byte, flowKeys odp.FlowKeys) error {
thunk := fastdp.handleMiss(flowKeys)
if thunk != nil {
fastdp.send(thunk(), packet)
}
return nil
}
func (fastdp *FastDatapath) handleMiss(fks odp.FlowKeys) thunk {
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
ingress := fks[odp.OVS_KEY_ATTR_IN_PORT].(odp.InPortFlowKey).VportID()
log.Println("Got miss", fks, "on port", ingress)
handler := fastdp.missHandlers[ingress]
if handler == nil {
handler = fastdp.makeMissHandler(ingress)
if handler == nil {
return simpleThunk(vetoFlowCreation)
}
fastdp.missHandlers[ingress] = handler
}
return handler(fks)
}
func (fastdp *FastDatapath) makeMissHandler(
ingress odp.VportID) func(odp.FlowKeys) thunk {
// Auto-add netdev and internal vports to the bridge (i.e. not
// vxlan vports)
vport, err := fastdp.dp.LookupVport(ingress)
if err != nil {
log.Println(err)
return nil
}
typ := vport.Spec.TypeName()
if typ != "netdev" && typ != "internal" {
return nil
}
// Construct a bridge port for the netdev, which simply
// outputs packets
bridgePort := func(key PacketKey) thunk {
return simpleThunk(fastdp.odpActions(
odp.NewOutputAction(ingress)))
}
fastdp.bridgePorts[bridgePortID{vport: ingress}] = bridgePort
// Clear flows, in order to recalculate flows for broadcasts
// on the bridge
checkWarn(fastdp.clearFlows())
// Packets from the netdev are processed by the bridge
return func(fks odp.FlowKeys) thunk {
return multiThunk(fastdp.bridge(bridgePortID{vport: ingress},
flowKeysToPacketKey(fks)))
}
}
func flowKeysToPacketKey(fks odp.FlowKeys) PacketKey {
eth := fks[odp.OVS_KEY_ATTR_ETHERNET].(odp.EthernetFlowKey).Key()
return PacketKey{SrcMAC: eth.EthSrc, DstMAC: eth.EthDst}
}
// Send a packet, creating a corresponding ODP flow rule if possible
func (fastdp *FastDatapath) send(fops FlowOp, frame []byte) {
// Gather the actions from actionFlowOps, execute any others
var dec *EthernetDecoder
flow := odp.NewFlowSpec()
createFlow := true
for _, xfop := range FlattenFlowOp(fops) {
switch fop := xfop.(type) {
case interface {
updateFlowSpec(*odp.FlowSpec)
}:
fop.updateFlowSpec(&flow)
case vetoFlowCreationFlowOp:
createFlow = false
default:
// A foreign flow op, so send the packet the
// normal way, decoding the packet lazily.
if dec == nil {
dec = fastdp.dec
dec.DecodeLayers(frame)
createFlow = false
}
if len(dec.decoded) != 0 {
fop.Send(frame, dec, false)
}
}
}
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
if len(flow.Actions) != 0 {
checkWarn(fastdp.dp.Execute(frame, nil, flow.Actions))
}
if createFlow {
log.Println("Creating flow", flow)
checkWarn(fastdp.dp.CreateFlow(flow))
}
}
type odpActionsFlowOp struct {
fastdp *FastDatapath
actions []odp.Action
}
func (fastdp *FastDatapath) odpActions(actions ...odp.Action) FlowOp {
return odpActionsFlowOp{
fastdp: fastdp,
actions: actions,
}
}
func (fop odpActionsFlowOp) updateFlowSpec(flow *odp.FlowSpec) {
flow.AddActions(fop.actions)
}
func (fop odpActionsFlowOp) Send(frame []byte, dec *EthernetDecoder, bc bool) {
fastdp := fop.fastdp
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
checkWarn(fastdp.dp.Execute(frame, nil, fop.actions))
}
type nopFlowOp struct{}
func (nopFlowOp) Send([]byte, *EthernetDecoder, bool) {
// A nopFlowOp just provides a hint about flow creation, it
// doesn't send anything
}
type vetoFlowCreationFlowOp struct {
nopFlowOp
}
var vetoFlowCreation = vetoFlowCreationFlowOp{}
type odpFlowKeyFlowOp struct {
key odp.FlowKey
nopFlowOp
}
func odpFlowKey(key odp.FlowKey) FlowOp {
return odpFlowKeyFlowOp{key: key}
}
func (fop odpFlowKeyFlowOp) updateFlowSpec(flow *odp.FlowSpec) {
flow.AddKey(fop.key)
}
func odpEthernetFlowKey(key PacketKey) FlowOp {
fk := odp.NewEthernetFlowKey()
fk.SetEthSrc(key.SrcMAC)
fk.SetEthDst(key.DstMAC)
return odpFlowKeyFlowOp{key: fk}
}
func simpleThunk(fop FlowOp) thunk {
return func() FlowOp { return fop }
}
func callThunks(thunks []thunk) FlowOp {
if len(thunks) == 1 {
return thunks[0]()
}
mfop := NewMultiFlowOp(false)
for _, thunk := range thunks {
mfop.Add(thunk())
}
return mfop
}
func multiThunk(thunks []thunk) thunk {
return func() FlowOp { return callThunks(thunks) }
}
// IntraHost
func (fastdp *FastDatapath) ConsumeIntraHostPackets(
consumer IntraHostConsumer) error {
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
if fastdp.bridgePorts[routerPortID] != nil {
panic("FastDatapath already has an IntraHostConsumer")
}
fastdp.bridgePorts[routerPortID] = func(key PacketKey) thunk {
return func() FlowOp {
return consumer.CapturedPacket(key)
}
}
return fastdp.consumeMisses()
}
func (fastdp *FastDatapath) InjectPacket(key PacketKey) FlowOp {
return callThunks(func() []thunk {
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
return fastdp.bridge(routerPortID, key)
}())
}
// A trivial bridge implementation
func (fastdp *FastDatapath) bridge(ingress bridgePortID,
key PacketKey) []thunk {
if fastdp.bridgeMACs[key.SrcMAC] == nil {
// Learn the source MAC
fastdp.bridgeMACs[key.SrcMAC] = fastdp.bridgePorts[ingress]
}
// If we know about the destination MAC, deliver it to the
// associated port.
dst := fastdp.bridgeMACs[key.DstMAC]
if dst != nil {
return []thunk{simpleThunk(odpEthernetFlowKey(key)), dst(key)}
}
// Otherwise, it might be a real broadcast, or it might just
// be for a MAC we don't know about yet. Either way, we'll
// broadcast it.
var fop FlowOp
thunks := make([]thunk, 1, len(fastdp.bridgePorts)+2)
if (key.DstMAC[0] & 1) == 0 {
// Not a real broadcast, so doon't create a flow rule.
// If we did, we'd need to clear the flows every time
// we learned a new MAC address, or have a more
// complicated selective invalidation scheme.
fop = vetoFlowCreation
} else {
// A real broadcast
fop = odpEthernetFlowKey(key)
if !ingress.router {
// The rule depends on the ingress vport.
thunks = thunks[:2]
thunks[1] = simpleThunk(odpFlowKey(odp.NewInPortFlowKey(
ingress.vport)))
}
}
thunks[0] = simpleThunk(fop)
// Send to all ports except the one it came in on.
for id, dst := range fastdp.bridgePorts {
if id != ingress {
thunks = append(thunks, dst(key))
}
}
return thunks
}
// InterHost
func (fastdp *FastDatapath) ConsumeInterHostPackets(
localPeer *Peer, peers *Peers, consumer InterHostConsumer) error {
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
if fastdp.missHandlers[fastdp.vxlanVportID] != nil {
panic("FastDatapath already has an InterHostConsumer")
}
fastdp.localPeer = localPeer
fastdp.peers = peers
fastdp.missHandlers[fastdp.vxlanVportID] = func(fks odp.FlowKeys) thunk {
tunnel := fks[odp.OVS_KEY_ATTR_TUNNEL].(odp.TunnelFlowKey).Key()
srcPeer, dstPeer := fastdp.extractPeers(tunnel.TunnelId)
if srcPeer == nil || dstPeer == nil {
return simpleThunk(vetoFlowCreation)
}
key := ForwardPacketKey{
SrcPeer: srcPeer,
DstPeer: dstPeer,
PacketKey: flowKeysToPacketKey(fks),
}
// The resulting flow rule should be restricted to
// packets with the same tunnelID
var tunnelFlowKey odp.TunnelFlowKey
tunnelFlowKey.SetTunnelId(tunnel.TunnelId)
tunnelFlowKey.SetIpv4Src(tunnel.Ipv4Src)
tunnelFlowKey.SetIpv4Dst(tunnel.Ipv4Dst)
return func() FlowOp {
fop := NewMultiFlowOp(false)
fop.Add(odpFlowKey(tunnelFlowKey))
fop.Add(consumer(key))
return fop
}
}
return fastdp.consumeMisses()
}
func (fastdp *FastDatapath) InvalidateRoutes() {
fmt.Println("InvalidateRoutes")
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
checkWarn(fastdp.clearFlows())
}
func (fastdp *FastDatapath) InvalidateShortIDs() {
fmt.Println("InvalidateShortIDs")
fastdp.lock.Lock()
defer fastdp.lock.Unlock()
checkWarn(fastdp.clearFlows())
}
func (fastdp *FastDatapath) extractPeers(tunnelID [8]byte) (*Peer, *Peer) {
if fastdp.peers == nil {
return nil, nil
}
vni := binary.BigEndian.Uint64(tunnelID[:])
srcPeer := fastdp.peers.FetchByShortID(PeerShortID(vni & 0xfff))
dstPeer := fastdp.peers.FetchByShortID(PeerShortID((vni >> 12) & 0xfff))
return srcPeer, dstPeer
}
type FastDatapathForwarder struct {
fastdp *FastDatapath
remotePeer *Peer
localIP [4]byte
sendControlMsg func([]byte) error
lock sync.Mutex
listener InterHostForwarderListener
remoteIP [4]byte
established bool
}
func (fastdp *FastDatapath) MakeForwarder(remotePeer *Peer, localIP net.IP,
remote *net.UDPAddr, connUID uint64, crypto InterHostCrypto,
sendControlMsg func([]byte) error) (InterHostForwarder, error) {
if len(localIP) != 4 {
return nil, fmt.Errorf("local IP address %s is not IPv4", localIP)
}
fwd := &FastDatapathForwarder{
fastdp: fastdp,
remotePeer: remotePeer,
sendControlMsg: sendControlMsg,
}
copy(fwd.localIP[:], localIP)
return fwd, nil
}
func (fwd *FastDatapathForwarder) SetListener(
listener InterHostForwarderListener) {
fwd.lock.Lock()
defer fwd.lock.Unlock()
if listener == nil {
panic("nil listener")
}
fwd.listener = listener
if fwd.established {
listener.Established()
}
fwd.sendControlMsg(fwd.localIP[:])
}
func (fwd *FastDatapathForwarder) ControlMessage(msg []byte) {
fwd.lock.Lock()
defer fwd.lock.Unlock()
if len(msg) != 4 {
if fwd.listener != nil {
fwd.listener.Error(fmt.Errorf("FastDatapath control message wrong length %d", len(msg)))
fwd.listener = nil
}
return
}
if !fwd.established {
copy(fwd.remoteIP[:], msg)
fwd.established = true
if fwd.listener != nil {
fwd.listener.Established()
}
}
}
func (fwd *FastDatapathForwarder) Forward(key ForwardPacketKey) FlowOp {
fwd.lock.Lock()
defer fwd.lock.Unlock()
if !fwd.established {
// Ideally we could just return nil. But then we
// would have to invalidate the resulting flows when
// we learn the remote IP. So for now, just prevent
// flows.
return vetoFlowCreation
}
var sta odp.SetTunnelAction
sta.SetTunnelId(tunnelIDFor(key))
sta.SetIpv4Src(fwd.localIP)
sta.SetIpv4Dst(fwd.remoteIP)
sta.SetTos(0)
sta.SetTtl(64)
sta.SetDf(true)
sta.SetCsum(false)
return fwd.fastdp.odpActions(sta,
odp.NewOutputAction(fwd.fastdp.vxlanVportID))
}
func tunnelIDFor(key ForwardPacketKey) (tunnelID [8]byte) {
src := uint64(key.SrcPeer.ShortID)
dst := uint64(key.DstPeer.ShortID)
binary.BigEndian.PutUint64(tunnelID[:], src|dst<<12)
return
}
func (fwd *FastDatapathForwarder) Close() {
// Ideally we would delete all the relevant flows here. But
// until we do that, it's probably not worth clearing all
// flows.
}
+72
View File
@@ -0,0 +1,72 @@
package router
import "net"
// Just enough flow machinery for the weave router
type MAC [6]byte
func (mac MAC) String() string {
return net.HardwareAddr(mac[:]).String()
}
type PacketKey struct {
SrcMAC MAC
DstMAC MAC
}
type ForwardPacketKey struct {
SrcPeer *Peer
DstPeer *Peer
PacketKey
}
type FlowOp interface {
// The caller must supply an EthernetDecoder specific to this
// thread, which has already been used to decode the frame.
// The broadcast parameter is a hint whether the packet is
// being broadcast.
Send(frame []byte, dec *EthernetDecoder, broadcast bool)
}
type MultiFlowOp struct {
broadcast bool
ops []FlowOp
}
func NewMultiFlowOp(broadcast bool) *MultiFlowOp {
return &MultiFlowOp{broadcast: broadcast}
}
func (mfop *MultiFlowOp) Add(ops ...FlowOp) {
mfop.ops = append(mfop.ops, ops...)
}
func (mfop *MultiFlowOp) Send(frame []byte, dec *EthernetDecoder,
broadcast bool) {
for _, op := range mfop.ops {
if op != nil {
op.Send(frame, dec, mfop.broadcast)
}
}
}
func FlattenFlowOp(fop FlowOp) []FlowOp {
return collectFlowOps(nil, fop)
}
func collectFlowOps(into []FlowOp, fop FlowOp) []FlowOp {
if fop == nil {
return into
}
if mfop, ok := fop.(*MultiFlowOp); ok {
for _, op := range mfop.ops {
into = collectFlowOps(into, op)
}
return into
}
return append(into, fop)
}
-414
View File
@@ -1,414 +0,0 @@
package router
import (
"code.google.com/p/gopacket"
"code.google.com/p/gopacket/layers"
"syscall"
"time"
)
type ForwardedFrame struct {
srcPeer *Peer
dstPeer *Peer
frame []byte
}
type FrameTooBigError struct {
EPMTU int // effective pmtu, i.e. what we tell packet senders
}
func (conn *LocalConnection) ensureForwarders() error {
if conn.forwarder != nil || conn.forwarderDF != nil {
return nil
}
udpSender := NewSimpleUDPSender(conn)
udpSenderDF, err := NewRawUDPSender(conn) // only thing that can error, so do it early
if err != nil {
return err
}
usingPassword := conn.SessionKey != nil
var encryptor, encryptorDF Encryptor
if usingPassword {
encryptor = NewNaClEncryptor(conn.local.NameByte, conn.SessionKey, conn.outbound, false)
encryptorDF = NewNaClEncryptor(conn.local.NameByte, conn.SessionKey, conn.outbound, true)
} else {
encryptor = NewNonEncryptor(conn.local.NameByte)
encryptorDF = NewNonEncryptor(conn.local.NameByte)
}
forwarder := NewForwarder(conn, encryptor, udpSender, DefaultPMTU)
forwarderDF := NewForwarderDF(conn, encryptorDF, udpSenderDF, DefaultPMTU)
effectivePMTU := forwarderDF.unverifiedPMTU
forwarder.Start()
forwarderDF.Start()
// Various fields in the conn struct are read by other processes,
// so we have to use locks.
conn.Lock()
conn.forwarder = forwarder
conn.forwarderDF = forwarderDF
conn.effectivePMTU = effectivePMTU
conn.Unlock()
return nil
}
func (conn *LocalConnection) stopForwarders() {
if conn.forwarder == nil || conn.forwarderDF == nil {
return
}
conn.forwarder.Shutdown()
conn.forwarderDF.Shutdown()
}
// Called from peer.Relay[Broadcast] which is itself invoked from
// router (both UDP listener process and sniffer process). Also called
// from connection's heartbeat process, and from the connection's TCP
// receiver process.
func (conn *LocalConnection) Forward(df bool, frame *ForwardedFrame, dec *EthernetDecoder) error {
conn.RLock()
var (
forwarder = conn.forwarder
forwarderDF = conn.forwarderDF
effectivePMTU = conn.effectivePMTU
stackFrag = conn.stackFrag
)
conn.RUnlock()
if forwarder == nil || forwarderDF == nil {
conn.Log("Cannot forward frame yet - awaiting contact")
return nil
}
// We could use non-blocking channel sends here, i.e. drop frames
// on the floor when the forwarder is busy. This would allow our
// caller - the capturing loop in the router - to read frames more
// quickly when under load, i.e. we'd drop fewer frames on the
// floor during capture. And we could maximise CPU utilisation
// since we aren't stalling a thread. However, a lot of work has
// already been done by the time we get here. Since any packet we
// drop will likely get re-transmitted we end up paying that cost
// multiple times. So it's better to drop things at the beginning
// of our pipeline.
if df {
if !frameTooBig(frame, effectivePMTU) {
forwarderDF.Forward(frame)
return nil
}
return FrameTooBigError{EPMTU: effectivePMTU}
}
if stackFrag || dec == nil || len(dec.decoded) < 2 {
forwarder.Forward(frame)
return nil
}
// Don't have trustworthy stack, so we're going to have to
// send it DF in any case.
if !frameTooBig(frame, effectivePMTU) {
forwarderDF.Forward(frame)
return nil
}
conn.Router.LogFrame("Fragmenting", frame.frame, &dec.eth)
// We can't trust the stack to fragment, we have IP, and we
// have a frame that's too big for the MTU, so we have to
// fragment it ourself.
return fragment(dec.eth, dec.ip, effectivePMTU, frame, func(segFrame *ForwardedFrame) {
forwarderDF.Forward(segFrame)
})
}
func frameTooBig(frame *ForwardedFrame, effectivePMTU int) bool {
// We capture/forward complete ethernet frames. Therefore the
// frame length includes the ethernet header. However, MTUs
// operate at the IP layer and thus do not include the ethernet
// header. To put it another way, when a sender that was told an
// MTU of M sends an IP packet of exactly that length, we will
// capture/forward M + EthernetOverhead bytes of data.
return len(frame.frame) > effectivePMTU+EthernetOverhead
}
func fragment(eth layers.Ethernet, ip layers.IPv4, pmtu int, frame *ForwardedFrame, forward func(*ForwardedFrame)) error {
// We are not doing any sort of NAT, so we don't need to worry
// about checksums of IP payload (eg UDP checksum).
headerSize := int(ip.IHL) * 4
// &^ is bit clear (AND NOT). So here we're clearing the lowest 3
// bits.
maxSegmentSize := (pmtu - headerSize) &^ 7
opts := gopacket.SerializeOptions{
FixLengths: false,
ComputeChecksums: true}
payloadSize := int(ip.Length) - headerSize
payload := ip.BaseLayer.Payload[:payloadSize]
offsetBase := int(ip.FragOffset) << 3
origFlags := ip.Flags
ip.Flags = ip.Flags | layers.IPv4MoreFragments
ip.Length = uint16(headerSize + maxSegmentSize)
if eth.EthernetType == layers.EthernetTypeLLC {
// using LLC, so must set eth length correctly. eth length
// is just the length of the payload
eth.Length = ip.Length
} else {
eth.Length = 0
}
for offset := 0; offset < payloadSize; offset += maxSegmentSize {
var segmentPayload []byte
if len(payload) <= maxSegmentSize {
// last one
segmentPayload = payload
ip.Length = uint16(len(payload) + headerSize)
ip.Flags = origFlags
if eth.EthernetType == layers.EthernetTypeLLC {
eth.Length = ip.Length
} else {
eth.Length = 0
}
} else {
segmentPayload = payload[:maxSegmentSize]
payload = payload[maxSegmentSize:]
}
ip.FragOffset = uint16((offset + offsetBase) >> 3)
buf := gopacket.NewSerializeBuffer()
segPayload := gopacket.Payload(segmentPayload)
err := gopacket.SerializeLayers(buf, opts, &eth, &ip, &segPayload)
if err != nil {
return err
}
// make copies of the frame we received
segFrame := *frame
segFrame.frame = buf.Bytes()
forward(&segFrame)
}
return nil
}
// Forwarder
type Forwarder struct {
conn *LocalConnection
ch chan<- *ForwardedFrame
finished <-chan struct{}
enc Encryptor
udpSender UDPSender
maxPayload int
processSendError func(error) error
}
func NewForwarder(conn *LocalConnection, enc Encryptor, udpSender UDPSender, pmtu int) *Forwarder {
return &Forwarder{
conn: conn,
enc: enc,
udpSender: udpSender,
maxPayload: pmtu - UDPOverhead,
processSendError: func(err error) error { return err }}
}
func (fwd *Forwarder) Start() {
ch := make(chan *ForwardedFrame, ChannelSize)
fwd.ch = ch
finished := make(chan struct{})
fwd.finished = finished
go fwd.run(ch, finished)
}
func (fwd *Forwarder) Shutdown() {
fwd.ch <- nil
}
func (fwd *Forwarder) Forward(frame *ForwardedFrame) {
select {
case fwd.ch <- frame:
case <-fwd.finished:
}
}
func (fwd *Forwarder) run(ch <-chan *ForwardedFrame, finished chan<- struct{}) {
defer fwd.udpSender.Shutdown()
for {
if !fwd.accumulateAndSendFrames(ch, <-ch) {
close(finished)
return
}
}
}
func (fwd *Forwarder) effectiveOverhead() int {
return UDPOverhead + fwd.enc.PacketOverhead() + fwd.enc.FrameOverhead() + EthernetOverhead
}
// Drain the inbound channel of frames, aggregating them into larger
// packets for efficient transmission.
//
// FIXME Depending on the golang scheduler, and the rate at which
// franes get sent to the forwarder, we can be going around this loop
// forever. That is bad since there may be other stuff for us to do,
// i.e. the other branches in the run loop.
func (fwd *Forwarder) accumulateAndSendFrames(ch <-chan *ForwardedFrame, frame *ForwardedFrame) bool {
if frame == nil {
return false
}
if !fwd.appendFrame(frame) {
fwd.logDrop(frame)
// [1] The buffer is empty at this point and therefore we must
// not flush it. The easiest way to accomplish that is simply
// by returning to the surrounding run loop.
return true
}
for {
select {
case frame = <-ch:
if frame == nil {
return false
}
if !fwd.appendFrame(frame) {
fwd.flush()
if !fwd.appendFrame(frame) {
fwd.logDrop(frame)
return true // see [1]
}
}
default:
fwd.flush()
return true
}
}
}
func (fwd *Forwarder) logDrop(frame *ForwardedFrame) {
fwd.conn.Log("Dropping too big frame during forwarding: frame len:", len(frame.frame), "; effective PMTU:", fwd.maxPayload+UDPOverhead-fwd.effectiveOverhead())
}
func (fwd *Forwarder) appendFrame(frame *ForwardedFrame) bool {
frameLen := len(frame.frame)
if fwd.enc.TotalLen()+fwd.enc.FrameOverhead()+frameLen > fwd.maxPayload {
return false
}
fwd.enc.AppendFrame(frame.srcPeer.NameByte, frame.dstPeer.NameByte, frame.frame)
return true
}
func (fwd *Forwarder) flush() {
msg, err := fwd.enc.Bytes()
if err != nil {
fwd.conn.Shutdown(err)
}
err = fwd.processSendError(fwd.udpSender.Send(msg))
if err != nil && PosixError(err) != syscall.ENOBUFS {
fwd.conn.Shutdown(err)
}
}
type ForwarderDF struct {
Forwarder
verifyPMTUTick <-chan time.Time
verifyPMTU chan<- int
pmtuVerifyCount uint
pmtuVerified bool
highestGoodPMTU int
unverifiedPMTU int
lowestBadPMTU int
}
func NewForwarderDF(conn *LocalConnection, enc Encryptor, udpSender UDPSender, pmtu int) *ForwarderDF {
fwd := &ForwarderDF{
Forwarder: Forwarder{
conn: conn,
enc: enc,
udpSender: udpSender,
maxPayload: pmtu - UDPOverhead}}
fwd.Forwarder.processSendError = fwd.processSendError
fwd.unverifiedPMTU = pmtu - fwd.effectiveOverhead()
return fwd
}
func (fwd *ForwarderDF) Start() {
ch := make(chan *ForwardedFrame, ChannelSize)
fwd.ch = ch
finished := make(chan struct{})
fwd.finished = finished
verifyPMTU := make(chan int, ChannelSize)
fwd.verifyPMTU = verifyPMTU
go fwd.run(ch, finished, verifyPMTU)
}
func (fwd *ForwarderDF) PMTUVerified(pmtu int) {
select {
case fwd.verifyPMTU <- pmtu:
case <-fwd.finished:
}
}
func (fwd *ForwarderDF) run(ch <-chan *ForwardedFrame, finished chan<- struct{}, verifyPMTU <-chan int) {
defer fwd.udpSender.Shutdown()
for {
select {
case <-fwd.verifyPMTUTick:
// We only do this case here when we know the
// buffers are all empty so that we don't risk
// appending verify-frames to other data.
fwd.verifyPMTUTick = nil
if fwd.pmtuVerified {
continue
}
if fwd.pmtuVerifyCount > 0 {
fwd.pmtuVerifyCount--
fwd.attemptVerifyEffectivePMTU()
} else {
// we've exceeded the verification
// attempts of the unverifiedPMTU
fwd.lowestBadPMTU = fwd.unverifiedPMTU
fwd.verifyEffectivePMTU((fwd.highestGoodPMTU + fwd.lowestBadPMTU) / 2)
}
case epmtu := <-verifyPMTU:
if fwd.pmtuVerified || epmtu != fwd.unverifiedPMTU {
continue
}
if epmtu+1 < fwd.lowestBadPMTU {
fwd.highestGoodPMTU = fwd.unverifiedPMTU // = epmtu
fwd.verifyEffectivePMTU((fwd.highestGoodPMTU + fwd.lowestBadPMTU) / 2)
} else {
fwd.pmtuVerified = true
fwd.maxPayload = epmtu + fwd.effectiveOverhead() - UDPOverhead
fwd.conn.setEffectivePMTU(epmtu)
fwd.conn.Log("Effective PMTU verified at", epmtu)
}
case frame := <-ch:
if !fwd.accumulateAndSendFrames(ch, frame) {
close(finished)
return
}
}
}
}
func (fwd *ForwarderDF) verifyEffectivePMTU(newUnverifiedPMTU int) {
fwd.unverifiedPMTU = newUnverifiedPMTU
fwd.pmtuVerifyCount = PMTUVerifyAttempts
fwd.attemptVerifyEffectivePMTU()
}
func (fwd *ForwarderDF) attemptVerifyEffectivePMTU() {
fwd.enc.AppendFrame(fwd.conn.local.NameByte, fwd.conn.remote.NameByte,
make([]byte, fwd.unverifiedPMTU+EthernetOverhead))
fwd.flush()
if fwd.verifyPMTUTick == nil {
fwd.verifyPMTUTick = time.After(PMTUVerifyTimeout << (PMTUVerifyAttempts - fwd.pmtuVerifyCount))
}
}
func (fwd *ForwarderDF) processSendError(err error) error {
if mtbe, ok := err.(MsgTooBigError); ok {
newUnverifiedPMTU := mtbe.PMTU - fwd.effectiveOverhead()
if newUnverifiedPMTU < fwd.unverifiedPMTU {
fwd.pmtuVerified = false
fwd.maxPayload = mtbe.PMTU - UDPOverhead
fwd.highestGoodPMTU = 8
fwd.lowestBadPMTU = newUnverifiedPMTU + 1
fwd.conn.setEffectivePMTU(newUnverifiedPMTU)
fwd.verifyEffectivePMTU(newUnverifiedPMTU)
}
return nil
}
return err
}
+4 -7
View File
@@ -44,12 +44,11 @@ type GossipSender struct {
}
func NewGossipSender(send func(GossipData)) *GossipSender {
return &GossipSender{send: send}
}
func (sender *GossipSender) Start() {
sender.cell = make(chan GossipData, 1)
sender := &GossipSender{
send: send,
cell: make(chan GossipData, 1)}
go sender.run()
return sender
}
func (sender *GossipSender) run() {
@@ -239,7 +238,6 @@ func (c *GossipChannel) sendDown(conn Connection, data GossipData) {
conn.(ProtocolSender).SendProtocolMsg(protocolMsg)
})
c.senders[conn] = sender
sender.Start()
}
sender.Send(data)
}
@@ -290,7 +288,6 @@ func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) erro
if !found {
broadcaster = NewGossipSender(func(pending GossipData) { c.sendBroadcast(srcName, pending) })
c.broadcasters[srcName] = broadcaster
broadcaster.Start()
}
broadcaster.Send(update)
return nil
+8 -13
View File
@@ -17,11 +17,8 @@ type mockChannelConnection struct {
// Construct a "passive" Router, i.e. without any goroutines, except
// for Routes and GossipSenders.
func NewTestRouter(name PeerName) *Router {
router := NewRouter(RouterConfig{}, name, "")
// need to create a dummy channel otherwise tests hang on nil
// channels when the Router invoked ConnectionMaker.Refresh
router.ConnectionMaker.actionChan = make(chan ConnectionMakerAction, ChannelSize)
router.Routes.Start()
router := NewRouter(RouterConfig{}, name, "nick")
router.Start()
return router
}
@@ -33,17 +30,15 @@ func (conn *mockChannelConnection) SendProtocolMsg(protocolMsg ProtocolMsg) {
}
func (router *Router) AddTestChannelConnection(r *Router) {
fromName := router.Ourself.Peer.Name
toName := r.Ourself.Peer.Name
fromPeer := NewPeer(fromName, "", router.Ourself.Peer.UID, 0)
toPeer := NewPeer(toName, "", r.Ourself.Peer.UID, 0)
fromPeer := NewPeerFromSummary(router.Ourself.Peer.PeerSummary)
toPeer := NewPeerFromSummary(r.Ourself.Peer.PeerSummary)
r.Peers.FetchWithDefault(fromPeer) // Has side-effect of incrementing refcount
router.Peers.FetchWithDefault(toPeer) //
conn := &mockChannelConnection{RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, r}
router.Ourself.handleAddConnection(conn)
router.broadcastPeerUpdate(conn.Remote())
router.Ourself.handleConnectionEstablished(conn)
router.sendPendingGossip()
}
@@ -73,13 +68,13 @@ func TestGossipTopology(t *testing.T) {
// the routers supplied as arguments, carrying across all UID and
// version information.
func (router *Router) tp(routers ...*Router) *Peer {
peer := NewPeer(router.Ourself.Peer.Name, "", router.Ourself.Peer.UID, 0)
peer := NewPeerFromSummary(router.Ourself.Peer.PeerSummary)
connections := make(map[PeerName]Connection)
for _, r := range routers {
p := NewPeer(r.Ourself.Peer.Name, "", r.Ourself.Peer.UID, r.Ourself.Peer.version)
p := NewPeerFromSummary(r.Ourself.Peer.PeerSummary)
connections[r.Ourself.Peer.Name] = newMockConnection(peer, p)
}
peer.version = router.Ourself.Peer.version
peer.Version = router.Ourself.Peer.Version
peer.connections = connections
return peer
}
+23 -28
View File
@@ -44,7 +44,7 @@ func (fv *FieldValidator) Err() error {
return fv.err
}
func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, acceptNewPeer bool) error {
func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, acceptNewPeer bool) (*Peer, error) {
// We do not need to worry about locking in here as at this point
// the connection is not reachable by any go-routine other than
// ourself. Only when we add this connection to the conn.local
@@ -56,46 +56,56 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep
usingPassword := conn.Router.UsingPassword()
fv, private, err := conn.handshakeSendRecv(localConnID, usingPassword, enc, dec)
if err != nil {
return err
return nil, err
}
nameStr, _ := fv.Value("Name")
nickNameStr, _ := fv.Value("NickName")
uidStr, _ := fv.Value("UID")
shortIDStr, _ := fv.Value("ShortID")
remoteConnIDStr, _ := fv.Value("ConnID")
if err := fv.Err(); err != nil {
return err
return nil, err
}
name, err := PeerNameFromString(nameStr)
if err != nil {
return err
return nil, err
}
if !acceptNewPeer {
if _, found := conn.Router.Peers.Fetch(name); !found {
return fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
return nil, fmt.Errorf("Found unknown remote name: %s at %s", name, conn.remoteTCPAddr)
}
}
if existingConn, found := conn.Router.Ourself.ConnectionTo(name); found && existingConn.Established() {
return fmt.Errorf("Already have connection to %s at %s", existingConn.Remote(), existingConn.RemoteTCPAddr())
return nil, fmt.Errorf("Already have connection to %s at %s", existingConn.Remote(), existingConn.RemoteTCPAddr())
}
uid, err := ParsePeerUID(uidStr)
if err != nil {
return err
return nil, err
}
shortID, err := strconv.ParseUint(shortIDStr, 10, PeerShortIDBits)
if err != nil {
return nil, err
}
remoteConnID, err := strconv.ParseUint(remoteConnIDStr, 10, 64)
if err != nil {
return err
return nil, err
}
conn.uid = localConnID ^ remoteConnID
remotePublicStr, rpErr := fv.Value("PublicKey")
if usingPassword {
if rpErr != nil {
return rpErr
return nil, rpErr
}
remotePublicSlice, rpErr := hex.DecodeString(remotePublicStr)
if rpErr != nil {
return rpErr
return nil, rpErr
}
remotePublic := [32]byte{}
for idx, elem := range remotePublicSlice {
@@ -103,16 +113,14 @@ func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, accep
}
conn.SessionKey = FormSessionKey(&remotePublic, private, conn.Router.Password)
conn.tcpSender = NewEncryptedTCPSender(enc, conn.SessionKey, conn.outbound)
conn.Decryptor = NewNaClDecryptor(conn.SessionKey, conn.outbound)
} else {
if rpErr == nil {
return fmt.Errorf("Remote network is encrypted. Password required.")
return nil, fmt.Errorf("Remote network is encrypted. Password required.")
}
conn.tcpSender = NewSimpleTCPSender(enc)
conn.Decryptor = NewNonDecryptor()
}
return conn.setRemote(NewPeer(name, nickNameStr, uid, 0))
return NewPeer(name, nickNameStr, uid, 0, PeerShortID(shortID)), nil
}
func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword bool, enc *gob.Encoder, dec *gob.Decoder) (*FieldValidator, *[32]byte, error) {
@@ -124,6 +132,7 @@ func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword
"Name": conn.local.Name.String(),
"NickName": conn.local.NickName,
"UID": fmt.Sprint(conn.local.UID),
"ShortID": fmt.Sprint(conn.local.ShortID),
"ConnID": fmt.Sprint(localConnID)}
handshakeRecv := map[string]string{}
@@ -148,17 +157,3 @@ func (conn *LocalConnection) handshakeSendRecv(localConnID uint64, usingPassword
fv.CheckEqual("PeerNameFlavour", PeerNameFlavour)
return fv, private, nil
}
func (conn *LocalConnection) setRemote(toPeer *Peer) error {
toPeer = conn.Router.Peers.FetchWithDefault(toPeer)
switch toPeer {
case nil:
return fmt.Errorf("Connection appears to be with different version of a peer we already know of")
case conn.local:
conn.remote = toPeer // have to do assigment here to ensure Shutdown releases ref count
return fmt.Errorf("Cannot connect to ourself")
default:
conn.remote = toPeer
return nil
}
}
+90
View File
@@ -0,0 +1,90 @@
package router
import (
"net"
)
// Interface to inter-host (i.e. overlay network) packet handling
type InterHost interface {
// Start consuming forwarded packets.
ConsumeInterHostPackets(*Peer, *Peers, InterHostConsumer) error
// The routes have changed, so any cached information should
// be discarded.
InvalidateRoutes()
// A mapping of a short id to a peer has changed
InvalidateShortIDs()
// Form a packet-forwarding connection. The remote UDPAddr
// can be nil if unknown (in which case the implementation
// needs to discover it).
MakeForwarder(remotePeer *Peer, localIP net.IP, remote *net.UDPAddr,
connUID uint64, crypto InterHostCrypto,
sendControlMessage func([]byte) error) (InterHostForwarder, error)
}
// When a consumer is called, the decoder will already have been used
// to decode the frame.
type InterHostConsumer func(ForwardPacketKey) FlowOp
// Crypto settings for a forwarder.
type InterHostCrypto struct {
Dec Decryptor
Enc Encryptor
EncDF Encryptor
}
// All of the machinery to forward packets to a particular peer
type InterHostForwarder interface {
// Register a callback for forwarder state changes.
// side-effect, calling this confirms that the connection is
// really wanted, and so the provider should activate it.
// However, Forward might be called before this is called
// (e.g. on another thread).
SetListener(InterHostForwarderListener)
// Forward a packet across the connection.
Forward(ForwardPacketKey) FlowOp
Close()
// Handle a message from the peer
ControlMessage([]byte)
}
type InterHostForwarderListener interface {
Established()
Error(error)
}
type NullInterHost struct{}
func (NullInterHost) ConsumeInterHostPackets(*Peer, *Peers,
InterHostConsumer) error {
return nil
}
func (NullInterHost) InvalidateRoutes() {
}
func (NullInterHost) InvalidateShortIDs() {
}
func (NullInterHost) MakeForwarder(*Peer, net.IP, *net.UDPAddr, uint64,
InterHostCrypto, func([]byte) error) (InterHostForwarder, error) {
return NullInterHost{}, nil
}
func (NullInterHost) SetListener(InterHostForwarderListener) {
}
func (NullInterHost) Forward(ForwardPacketKey) FlowOp {
return nil
}
func (NullInterHost) Close() {
}
func (NullInterHost) ControlMessage([]byte) {
}
+31
View File
@@ -0,0 +1,31 @@
package router
// Interface to intra-host (i.e. container) packet handling
type IntraHost interface {
// Inject a packet to be delivered locally
InjectPacket(PacketKey) FlowOp
// Start consuming packets
ConsumeIntraHostPackets(IntraHostConsumer) error
}
type IntraHostConsumer interface {
// A locally captured packet. The caller must supply an
// EthernetDecoder specific to this thread, but it should not
// have been used to decode the packet data.
CapturedPacket(PacketKey) FlowOp
}
type NullIntraHost struct{}
func (NullIntraHost) InjectPacket(PacketKey) FlowOp {
return nil
}
func (NullIntraHost) ConsumeIntraHostPackets(IntraHostConsumer) error {
return nil
}
func (NullIntraHost) String() string {
return "<no intra-host networking>"
}
+2 -2
View File
@@ -16,7 +16,7 @@ func (router *Router) StatusJSON(version, encryption string) ([]byte, error) {
Macs *MacCache
Peers *Peers
Routes *Routes
}{version, encryption, router.Ourself.Name.String(), router.Ourself.NickName, fmt.Sprintf("%v", router.Iface), router.Macs, router.Peers, router.Routes})
}{version, encryption, router.Ourself.Name.String(), router.Ourself.NickName, fmt.Sprint(router.IntraHost), router.Macs, router.Peers, router.Routes})
// leaving out ConectionMaker due to async complexities
}
@@ -57,7 +57,7 @@ func (peers *Peers) MarshalJSON() ([]byte, error) {
connections = append(connections, conn)
}
}
ps = append(ps, &p{peer.Name.String(), peer.NickName, peer.UID, peer.version, connections})
ps = append(ps, &p{peer.Name.String(), peer.NickName, peer.UID, peer.Version, connections})
})
return json.Marshal(ps)
}
+46 -44
View File
@@ -18,63 +18,65 @@ type LocalPeer struct {
type LocalPeerAction func()
func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer {
return &LocalPeer{Peer: NewPeer(name, nickName, 0, 0), router: router}
}
func (peer *LocalPeer) Start() {
actionChan := make(chan LocalPeerAction, ChannelSize)
peer.actionChan = actionChan
peer := &LocalPeer{
Peer: NewPeer(name, nickName, randomPeerUID(), 0,
randomPeerShortID()),
router: router,
actionChan: actionChan,
}
go peer.actorLoop(actionChan)
return peer
}
func (peer *LocalPeer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
return peer.Relay(peer.Peer, dstPeer, df, frame, dec)
func (peer *LocalPeer) Forward(dstPeer *Peer, key PacketKey) FlowOp {
return peer.Relay(ForwardPacketKey{
PacketKey: key,
SrcPeer: peer.Peer,
DstPeer: dstPeer,
})
}
func (peer *LocalPeer) Broadcast(df bool, frame []byte, dec *EthernetDecoder) {
peer.RelayBroadcast(peer.Peer, df, frame, dec)
func (peer *LocalPeer) Broadcast(key PacketKey) FlowOp {
return peer.RelayBroadcast(peer.Peer, key)
}
func (peer *LocalPeer) Relay(srcPeer, dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error {
relayPeerName, found := peer.router.Routes.Unicast(dstPeer.Name)
func (peer *LocalPeer) Relay(key ForwardPacketKey) FlowOp {
relayPeerName, found := peer.router.Routes.Unicast(key.DstPeer.Name)
if !found {
// Not necessarily an error as there could be a race with the
// dst disappearing whilst the frame is in flight
log.Println("Received packet for unknown destination:", dstPeer)
log.Println("Received packet for unknown destination:", key.DstPeer)
return nil
}
conn, found := peer.ConnectionTo(relayPeerName)
if !found {
// Again, could just be a race, not necessarily an error
log.Println("Unable to find connection to relay peer", relayPeerName)
return nil
}
return conn.(*LocalConnection).Forward(df, &ForwardedFrame{
srcPeer: srcPeer,
dstPeer: dstPeer,
frame: frame},
dec)
return conn.(*LocalConnection).Forward(key)
}
func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) {
func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, key PacketKey) FlowOp {
nextHops := peer.router.Routes.Broadcast(srcPeer.Name)
if len(nextHops) == 0 {
return
return nil
}
op := NewMultiFlowOp(true)
for _, conn := range peer.ConnectionsTo(nextHops) {
err := conn.(*LocalConnection).Forward(df, &ForwardedFrame{
srcPeer: srcPeer,
dstPeer: conn.Remote(),
frame: frame},
dec)
if err != nil {
if ftbe, ok := err.(FrameTooBigError); ok {
log.Printf("dropping too big DF broadcast frame (%v -> %v): PMTU= %v\n", dec.ip.DstIP, dec.ip.SrcIP, ftbe.EPMTU)
} else {
log.Println(err)
}
}
op.Add(conn.(*LocalConnection).Forward(ForwardPacketKey{
PacketKey: key,
SrcPeer: srcPeer,
DstPeer: conn.Remote(),
}))
}
return op
}
func (peer *LocalPeer) Info() string {
@@ -132,8 +134,7 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err
return err
}
connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), true, false)
connLocal := NewLocalConnection(connRemote, tcpConn, udpAddr, peer.router)
connLocal.Start(acceptNewPeer)
StartLocalConnection(connRemote, tcpConn, udpAddr, peer.router, acceptNewPeer)
return nil
}
@@ -217,7 +218,6 @@ func (peer *LocalPeer) handleAddConnection(conn Connection) error {
conn.Log("connection added (new peer)")
peer.router.SendAllGossipDown(conn)
}
peer.broadcastPeerUpdate(conn.Remote())
return nil
}
@@ -231,7 +231,7 @@ func (peer *LocalPeer) handleConnectionEstablished(conn Connection) {
}
peer.connectionEstablished(conn)
conn.Log("connection fully established")
peer.broadcastPeerUpdate()
peer.router.broadcastPeerUpdate()
}
func (peer *LocalPeer) handleDeleteConnection(conn Connection) {
@@ -250,16 +250,11 @@ func (peer *LocalPeer) handleDeleteConnection(conn Connection) {
// Must do garbage collection first to ensure we don't send out an
// update with unreachable peers (can cause looping)
peer.router.Peers.GarbageCollect()
peer.broadcastPeerUpdate()
peer.router.broadcastPeerUpdate()
}
// helpers
func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
peer.router.Routes.Recalculate()
peer.router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(peer.router.Peers, append(peers, peer.Peer)...))
}
func (peer *LocalPeer) checkConnectionLimit() error {
limit := peer.router.ConnLimit
if 0 != limit && peer.connectionCount() >= limit {
@@ -272,20 +267,20 @@ func (peer *LocalPeer) addConnection(conn Connection) {
peer.Lock()
defer peer.Unlock()
peer.connections[conn.Remote().Name] = conn
peer.version++
peer.Version++
}
func (peer *LocalPeer) deleteConnection(conn Connection) {
peer.Lock()
defer peer.Unlock()
delete(peer.connections, conn.Remote().Name)
peer.version++
peer.Version++
}
func (peer *LocalPeer) connectionEstablished(conn Connection) {
peer.Lock()
defer peer.Unlock()
peer.version++
peer.Version++
}
func (peer *LocalPeer) connectionCount() int {
@@ -293,3 +288,10 @@ func (peer *LocalPeer) connectionCount() int {
defer peer.RUnlock()
return len(peer.connections)
}
func (peer *LocalPeer) setShortID(shortID PeerShortID) {
peer.Lock()
defer peer.Unlock()
peer.ShortID = shortID
peer.Version++
}
+6 -8
View File
@@ -22,14 +22,12 @@ type MacCache struct {
}
func NewMacCache(maxAge time.Duration, onExpiry func(net.HardwareAddr, *Peer)) *MacCache {
return &MacCache{
cache := &MacCache{
table: make(map[uint64]*MacCacheEntry),
maxAge: maxAge,
onExpiry: onExpiry}
}
func (cache *MacCache) Start() {
cache.setExpiryTimer()
return cache
}
func (cache *MacCache) Enter(mac net.HardwareAddr, peer *Peer) bool {
@@ -60,15 +58,15 @@ func (cache *MacCache) Enter(mac net.HardwareAddr, peer *Peer) bool {
return false
}
func (cache *MacCache) Lookup(mac net.HardwareAddr) (*Peer, bool) {
func (cache *MacCache) Lookup(mac net.HardwareAddr) *Peer {
key := macint(mac)
cache.RLock()
defer cache.RUnlock()
entry, found := cache.table[key]
if !found {
return nil, false
return nil
}
return entry.peer, true
return entry.peer
}
func (cache *MacCache) Delete(peer *Peer) bool {
@@ -95,7 +93,7 @@ func (cache *MacCache) String() string {
}
func (cache *MacCache) setExpiryTimer() {
cache.expiryTimer = time.AfterFunc(cache.maxAge/10, func() { cache.expire() })
//cache.expiryTimer = time.AfterFunc(cache.maxAge/10, func() { cache.expire() })
}
func (cache *MacCache) expire() {
+6 -6
View File
@@ -12,7 +12,9 @@ import (
// Add to peers a connection from peers.ourself to p
func (peers *Peers) AddTestConnection(p *Peer) {
toPeer := NewPeer(p.Name, "", p.UID, 0)
summary := p.PeerSummary
summary.Version = 0
toPeer := NewPeerFromSummary(summary)
peers.FetchWithDefault(toPeer) // Has side-effect of incrementing refcount
conn := newMockConnection(peers.ourself.Peer, toPeer)
peers.ourself.addConnection(conn)
@@ -21,11 +23,9 @@ func (peers *Peers) AddTestConnection(p *Peer) {
// Add to peers a connection from p1 to p2
func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) {
fromName := p1.Name
fromPeer := NewPeer(fromName, "", p1.UID, 0)
fromPeer := NewPeerFromSummary(p1.PeerSummary)
fromPeer = peers.FetchWithDefault(fromPeer)
toName := p2.Name
toPeer := NewPeer(toName, "", p2.UID, 0)
toPeer := NewPeerFromSummary(p2.PeerSummary)
toPeer = peers.FetchWithDefault(toPeer)
peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, "", false, false})
}
@@ -68,7 +68,7 @@ func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Co
// Get all the peers from a Peers in a slice
func (peers *Peers) allPeers() []*Peer {
var res []*Peer
for _, peer := range peers.table {
for _, peer := range peers.byName {
res = append(res, peer)
}
return res
+60
View File
@@ -0,0 +1,60 @@
package router
import (
"github.com/weaveworks/go-odp/odp"
)
// ODP admin functionality
func CreateDatapath(dpname string) error {
dpif, err := odp.NewDpif()
if err != nil {
return err
}
defer dpif.Close()
_, err = dpif.CreateDatapath(dpname)
if err != nil && !odp.IsDatapathNameAlreadyExistsError(err) {
return err
}
return nil
}
func DeleteDatapath(dpname string) error {
dpif, err := odp.NewDpif()
if err != nil {
return err
}
defer dpif.Close()
dp, err := dpif.LookupDatapath(dpname)
if err != nil {
if odp.IsNoSuchDatapathError(err) {
return nil
} else {
return err
}
}
return dp.Delete()
}
func AddDatapathInterface(dpname string, ifname string) error {
dpif, err := odp.NewDpif()
if err != nil {
return err
}
defer dpif.Close()
dp, err := dpif.LookupDatapath(dpname)
if err != nil {
return err
}
_, err = dp.CreateVport(odp.NewNetdevVportSpec(ifname))
return err
}
+81 -29
View File
@@ -3,16 +3,38 @@ package router
import (
"code.google.com/p/gopacket/pcap"
"fmt"
"log"
"net"
"sync"
)
type PcapIO struct {
handle *pcap.Handle
type Pcap struct {
iface *net.Interface
bufSz int
// The libpcap handle for writing packets. It's possible thata
// single handle could be used for reading and writing, but
// we'd have to examine the performance implications.
writeHandle *pcap.Handle
// pcap handles are single-threaded, so we need to lock around
// uses of writeHandle.
writeHandleMutex sync.Mutex
}
func NewPcapIO(ifName string, bufSz int) (PacketSourceSink, error) {
pio, err := newPcapIO(ifName, true, 65535, bufSz)
func NewPcap(iface *net.Interface, bufSz int) (IntraHost, error) {
wh, err := newPcapHandle(iface.Name, false, 0, 0)
if err != nil {
return pio, err
return nil, err
}
return &Pcap{iface: iface, bufSz: bufSz, writeHandle: wh}, nil
}
func (p *Pcap) ConsumeIntraHostPackets(consumer IntraHostConsumer) error {
rh, err := newPcapHandle(p.iface.Name, true, 65535, p.bufSz)
if err != nil {
return err
}
// Under Linux, libpcap implements the SetDirection filtering
@@ -20,16 +42,16 @@ func NewPcapIO(ifName string, bufSz int) (PacketSourceSink, error) {
// packets inside the kernel. We do this here rather than in
// newPcapIO because libpcap doesn't like this in combination
// with a 0 snaplen.
err = pio.handle.SetBPFFilter("inbound")
return pio, err
err = rh.SetBPFFilter("inbound")
if err != nil {
return err
}
go p.sniff(rh, consumer)
return nil
}
func NewPcapO(ifName string) (po PacketSink, err error) {
po, err = newPcapIO(ifName, false, 0, 0)
return
}
func newPcapIO(ifName string, promisc bool, snaplen int, bufSz int) (handle *PcapIO, err error) {
func newPcapHandle(ifName string, promisc bool, snaplen int, bufSz int) (handle *pcap.Handle, err error) {
inactive, err := pcap.NewInactiveHandle(ifName)
if err != nil {
return
@@ -60,26 +82,56 @@ func newPcapIO(ifName string, promisc bool, snaplen int, bufSz int) (handle *Pca
if err = inactive.SetBufferSize(bufSz); err != nil {
return
}
active, err := inactive.Activate()
handle, err = inactive.Activate()
if err != nil {
return
}
if err = active.SetDirection(pcap.DirectionIn); err != nil {
return
}
return &PcapIO{handle: active}, nil
}
func (pi *PcapIO) ReadPacket() (data []byte, err error) {
for {
data, _, err = pi.handle.ZeroCopyReadPacketData()
if err == nil || err != pcap.NextErrorTimeoutExpired {
break
}
}
err = handle.SetDirection(pcap.DirectionIn)
return
}
func (po *PcapIO) WritePacket(data []byte) error {
return po.handle.WritePacketData(data)
func (p *Pcap) String() string {
return fmt.Sprint(p.iface, "(via pcap)")
}
func (p *Pcap) InjectPacket(PacketKey) FlowOp {
return p
}
func (p *Pcap) Send(frame []byte, dec *EthernetDecoder, broadcast bool) {
p.writeHandleMutex.Lock()
defer p.writeHandleMutex.Unlock()
checkWarn(p.writeHandle.WritePacketData(frame))
}
func (p *Pcap) sniff(readHandle *pcap.Handle, consumer IntraHostConsumer) {
log.Println("Sniffing traffic on", p.iface)
dec := NewEthernetDecoder()
for {
pkt, _, err := readHandle.ZeroCopyReadPacketData()
if err == pcap.NextErrorTimeoutExpired {
continue
}
checkFatal(err)
dec.DecodeLayers(pkt)
decodedLen := len(dec.decoded)
if decodedLen == 0 {
return
}
if fop := consumer.CapturedPacket(dec.PacketKey()); fop != nil {
// at this point we are handing over the frame
// to forwarders, so we need to make a copy of
// it in order to prevent the next capture
// from overwriting the data
pktLen := len(pkt)
pktCopy := make([]byte, pktLen, pktLen)
copy(pktCopy, pkt)
fop.Send(pktCopy, dec, false)
}
}
}
+45 -16
View File
@@ -8,42 +8,71 @@ import (
type PeerUID uint64
func randomPeerUID() PeerUID {
return PeerUID(randUint64())
}
func ParsePeerUID(s string) (PeerUID, error) {
uid, err := strconv.ParseUint(s, 10, 64)
return PeerUID(uid), err
}
// Short IDs exist for the sake of fast datapath. They are 12 bits,
// randomly assigned, but we detect and recover from collisions. This
// does limit us to 4096 peers, but that should be sufficient for a
// while.
type PeerShortID uint16
const PeerShortIDBits = 12
func randomPeerShortID() PeerShortID {
return PeerShortID(randUint64() & (1<<PeerShortIDBits - 1))
}
type PeerSummary struct {
NameByte []byte
NickName string
UID PeerUID
Version uint64
ShortID PeerShortID
}
type Peer struct {
Name PeerName
NameByte []byte
NickName string
UID PeerUID
version uint64
Name PeerName
PeerSummary
localRefCount uint64 // maintained by Peers
connections map[PeerName]Connection
}
type ConnectionSet map[Connection]struct{}
func NewPeer(name PeerName, nickName string, uid PeerUID, version uint64) *Peer {
if uid == 0 {
uid = PeerUID(randUint64())
}
func NewPeer(name PeerName, nickName string, uid PeerUID, version uint64, shortID PeerShortID) *Peer {
return &Peer{
Name: name,
NameByte: name.Bin(),
NickName: nickName,
UID: uid,
version: version,
Name: name,
PeerSummary: PeerSummary{
NameByte: name.Bin(),
NickName: nickName,
UID: uid,
Version: version,
ShortID: shortID,
},
connections: make(map[PeerName]Connection)}
}
func NewPeerFromSummary(summary PeerSummary) *Peer {
return &Peer{
Name: PeerNameFromBin(summary.NameByte),
PeerSummary: summary,
connections: make(map[PeerName]Connection),
}
}
func (peer *Peer) String() string {
return fmt.Sprint(peer.Name, "(", peer.NickName, ")")
}
func (peer *Peer) Info() string {
return fmt.Sprint(peer.String(), " (v", peer.version, ") (UID ", peer.UID, ")")
return fmt.Sprint(peer.String(), " (v", peer.Version, ") (UID ", peer.UID, ")")
}
// Calculate the routing table from this peer to all peers reachable
@@ -70,7 +99,7 @@ func (peer *Peer) Info() string {
// NB: This function should generally be invoked while holding a read
// lock on Peers and LocalPeer.
func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) {
routes := make(map[PeerName]PeerName)
routes := make(unicastRoutes)
routes[peer.Name] = UnknownPeerName
nextWorklist := []*Peer{peer}
for len(nextWorklist) > 0 {
+293 -117
View File
@@ -5,14 +5,27 @@ import (
"encoding/gob"
"fmt"
"io"
"math/rand"
"sync"
)
type Peers struct {
sync.RWMutex
ourself *LocalPeer
table map[PeerName]*Peer
onGC func(*Peer)
ourself *LocalPeer
onGC func(*Peer)
byName map[PeerName]*Peer
byShortID map[PeerShortID]ShortIDPeers
invalidateShortIDsPending bool
}
type ShortIDPeers struct {
// If we know about a single peer with the short id, this is
// that peer. If there is a coliision, this is the peer with
// the lowest Name.
peer *Peer
// In case of a collision, this holds the other ppeers.
others []*Peer
}
type UnknownPeerError struct {
@@ -25,13 +38,6 @@ type NameCollisionError struct {
type PeerNameSet map[PeerName]struct{}
type PeerSummary struct {
NameByte []byte
NickName string
UID PeerUID
Version uint64
}
type ConnectionSummary struct {
NameByte []byte
RemoteTCPAddr string
@@ -40,34 +46,202 @@ type ConnectionSummary struct {
}
func NewPeers(ourself *LocalPeer, onGC func(*Peer)) *Peers {
return &Peers{
ourself: ourself,
table: make(map[PeerName]*Peer),
onGC: onGC}
peers := &Peers{
ourself: ourself,
byName: make(map[PeerName]*Peer),
byShortID: make(map[PeerShortID]ShortIDPeers),
onGC: onGC}
peers.FetchWithDefault(ourself.Peer)
return peers
}
func (peers *Peers) Fetch(name PeerName) (*Peer, bool) {
peers.RLock()
defer peers.RUnlock()
peer, found := peers.byName[name]
return peer, found // GRRR, why can't I inline this!?
}
func (peers *Peers) FetchByShortID(shortID PeerShortID) *Peer {
peers.RLock()
defer peers.RUnlock()
return peers.byShortID[shortID].peer
}
func (peers *Peers) addByShortID(peer *Peer) {
reassign := false
entry, ok := peers.byShortID[peer.ShortID]
if !ok {
entry = ShortIDPeers{peer: peer}
} else if entry.peer == nil {
// This short ID is free, but was used in the past.
// Because we are reusing it, it's an invalidation
// event.
entry.peer = peer
peers.invalidateShortIDsPending = true
} else if peer.Name < entry.peer.Name {
// Short ID collision, this peer becomes the principal
// peer for the short ID, bumping the previous one
// into others.
if entry.peer == peers.ourself.Peer {
// The bumped peer is peers.ourself, so we
// need to look foor a new short id
reassign = true
}
entry.others = append(entry.others, entry.peer)
entry.peer = peer
peers.invalidateShortIDsPending = true
} else {
// Short ID collision, this peer is secondary
entry.others = append(entry.others, peer)
}
peers.byShortID[peer.ShortID] = entry
if reassign {
peers.reassignLocalShortID()
}
}
func (peers *Peers) deleteByShortID(peer *Peer) {
entry := peers.byShortID[peer.ShortID]
var otherIndex int
if peer != entry.peer {
// peer is secondary, find its index in others
otherIndex = -1
for i, other := range entry.others {
if peer == other {
otherIndex = i
break
}
}
if otherIndex < 0 {
return
}
} else if len(entry.others) != 0 {
// need to find the peer with the lowest name to
// become the new principal one
otherIndex = 0
minName := entry.others[0].Name
for i := 1; i < len(entry.others); i++ {
otherName := entry.others[i].Name
if otherName < minName {
minName = otherName
otherIndex = i
}
}
entry.peer = entry.others[otherIndex]
peers.invalidateShortIDsPending = true
} else {
// This is the last peer with the short id. We clear
// the entry, don't delete it, in order to detect when
// it gets re-used.
peers.byShortID[peer.ShortID] = ShortIDPeers{}
return
}
entry.others[otherIndex] = entry.others[len(entry.others)-1]
entry.others = entry.others[:len(entry.others)-1]
peers.byShortID[peer.ShortID] = entry
}
func (peers *Peers) reassignLocalShortID() {
newShortID, ok := peers.chooseShortID()
if ok {
peers.setLocalShortID(newShortID)
}
// Otherwise we'll try again in garbageColleect
}
func (peers *Peers) setLocalShortID(newShortID PeerShortID) {
peers.deleteByShortID(peers.ourself.Peer)
peers.ourself.setShortID(newShortID)
peers.addByShortID(peers.ourself.Peer)
}
// Choose an available short id at random
func (peers *Peers) chooseShortID() (PeerShortID, bool) {
rng := rand.New(rand.NewSource(int64(randUint64())))
// First, just try picking some short ids at random, and
// seeing if they are available:
for i := 0; i < 10; i++ {
shortID := PeerShortID(rng.Intn(1 << PeerShortIDBits))
if peers.byShortID[shortID].peer == nil {
return shortID, true
}
}
// Looks like most short ids are used. So count the number of
// unused ones, and pick one at random.
available := int(1 << PeerShortIDBits)
for _, entry := range peers.byShortID {
if entry.peer != nil {
available--
}
}
if available == 0 {
// All short ids are used.
return 0, false
}
n := rng.Intn(available)
var i PeerShortID
for {
if peers.byShortID[i].peer == nil {
if n == 0 {
return PeerShortID(i), true
}
n--
}
i++
if i == 0 {
panic("chooseShortID broken")
}
}
}
func (peers *Peers) unlockAndMaybeInvalidateShortIDs() {
pending := peers.invalidateShortIDsPending
peers.invalidateShortIDsPending = false
peers.Unlock()
if pending {
// Some tests have a nil router
router := peers.ourself.router
if router != nil {
router.InterHost.InvalidateShortIDs()
}
}
}
func (peers *Peers) FetchWithDefault(peer *Peer) *Peer {
peers.Lock()
defer peers.Unlock()
if existingPeer, found := peers.table[peer.Name]; found {
defer peers.unlockAndMaybeInvalidateShortIDs()
if existingPeer, found := peers.byName[peer.Name]; found {
if existingPeer.UID != peer.UID {
return nil
}
existingPeer.localRefCount++
return existingPeer
}
peers.table[peer.Name] = peer
peers.byName[peer.Name] = peer
peers.addByShortID(peer)
peer.localRefCount++
return peer
}
func (peers *Peers) Fetch(name PeerName) (*Peer, bool) {
peers.RLock()
defer peers.RUnlock()
peer, found := peers.table[name]
return peer, found // GRRR, why can't I inline this!?
}
func (peers *Peers) Dereference(peer *Peer) {
peers.Lock()
defer peers.Unlock()
@@ -77,54 +251,21 @@ func (peers *Peers) Dereference(peer *Peer) {
func (peers *Peers) ForEach(fun func(*Peer)) {
peers.RLock()
defer peers.RUnlock()
for _, peer := range peers.table {
for _, peer := range peers.byName {
fun(peer)
}
}
// Merge an incoming update with our own topology.
//
// We add peers hitherto unknown to us, and update peers for which the
// update contains a more recent version than known to us. The return
// value is a) a representation of the received update, and b) an
// "improved" update containing just these new/updated elements.
func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) {
peers.Lock()
newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update)
if err != nil {
peers.Unlock()
return nil, nil, err
}
// By this point, we know the update doesn't refer to any peers we
// have no knowledge of. We can now apply the update. Start by
// adding in any new peers into the cache.
for name, newPeer := range newPeers {
peers.table[name] = newPeer
}
// Now apply the updates
newUpdate := peers.applyUpdate(decodedUpdate, decodedConns)
for _, peerRemoved := range peers.garbageCollect() {
delete(newUpdate, peerRemoved.Name)
}
// Don't need to hold peers lock any longer
peers.Unlock()
updateNames := make(PeerNameSet)
for _, peer := range decodedUpdate {
updateNames[peer.Name] = void
}
return updateNames, setFromPeersMap(newUpdate), nil
}
func (peers *Peers) Names() PeerNameSet {
peers.RLock()
defer peers.RUnlock()
return setFromPeersMap(peers.table)
names := make(PeerNameSet)
for name := range peers.byName {
names[name] = void
}
return names
}
func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
@@ -134,7 +275,7 @@ func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
defer peers.RUnlock()
peerList := make([]*Peer, 0, len(names))
for name := range names {
if peer, found := peers.table[name]; found {
if peer, found := peers.byName[name]; found {
if peer == peers.ourself.Peer {
peers.ourself.Peer.Encode(enc)
} else {
@@ -148,7 +289,7 @@ func (peers *Peers) EncodePeers(names PeerNameSet) []byte {
func (peers *Peers) GarbageCollect() []*Peer {
peers.Lock()
defer peers.Unlock()
defer peers.unlockAndMaybeInvalidateShortIDs()
return peers.garbageCollect()
}
@@ -185,22 +326,93 @@ func (peers *Peers) garbageCollect() []*Peer {
peers.ourself.RLock()
_, reached := peers.ourself.Routes(nil, false)
peers.ourself.RUnlock()
for name, peer := range peers.table {
for name, peer := range peers.byName {
if _, found := reached[peer.Name]; !found && peer.localRefCount == 0 {
delete(peers.table, name)
delete(peers.byName, name)
peers.deleteByShortID(peer)
peers.onGC(peer)
removed = append(removed, peer)
}
}
if peers.byShortID[peers.ourself.ShortID].peer != peers.ourself.Peer {
// The local peer doesn't own its short id. Garbage
// collection might have freed some up, so try to reassign.
peers.reassignLocalShortID()
}
return removed
}
func setFromPeersMap(peers map[PeerName]*Peer) PeerNameSet {
names := make(PeerNameSet)
for name := range peers {
names[name] = void
// Merge an incoming update with our own topology.
//
// We add peers hitherto unknown to us, and update peers for which the
// update contains a more recent version than known to us. The return
// value is a) a representation of the received update, and b) an
// "improved" update containing just these new/updated elements.
func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) {
peers.Lock()
defer peers.unlockAndMaybeInvalidateShortIDs()
newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update)
if err != nil {
return nil, nil, err
}
return names
// By this point, we know the update doesn't refer to any peers we
// have no knowledge of. We can now apply the update. Start by
// adding in any new peers into the cache.
for name, newPeer := range newPeers {
peers.byName[name] = newPeer
peers.addByShortID(newPeer)
}
newUpdate := make(PeerNameSet)
for idx, newPeer := range decodedUpdate {
connSummaries := decodedConns[idx]
name := newPeer.Name
// guaranteed to find peer in the peers.byName
peer := peers.byName[name]
if peer != newPeer &&
(peer == peers.ourself.Peer || peer.Version >= newPeer.Version) {
// Nobody but us updates us. And if we know more about a
// peer than what's in the the update, we ignore the
// latter.
continue
}
// If we're here, either it was a new peer, or the update has
// more info about the peer than we do. Either case, we need
// to set version and conns and include the updated peer in
// the outgoing update.
// Can peer have been updated by anyone else in the mean time?
// No - we know that peer is not ourself, so the only prospect
// for an update would be someone else calling
// router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on
// the router.Peers, so there can be no race here.
peer.Version = newPeer.Version
peer.connections = makeConnsMap(peer, connSummaries, peers.byName)
if newPeer.ShortID != peer.ShortID {
peers.deleteByShortID(peer)
peer.ShortID = newPeer.ShortID
peers.addByShortID(peer)
}
newUpdate[name] = void
}
for _, peerRemoved := range peers.garbageCollect() {
delete(newUpdate, peerRemoved.Name)
}
updateNames := make(PeerNameSet)
for _, peer := range decodedUpdate {
updateNames[peer.Name] = void
}
return updateNames, newUpdate, nil
}
func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]ConnectionSummary, err error) {
@@ -220,13 +432,12 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
err = decErr
return
}
name := PeerNameFromBin(peerSummary.NameByte)
newPeer := NewPeer(name, peerSummary.NickName, peerSummary.UID, peerSummary.Version)
newPeer := NewPeerFromSummary(peerSummary)
decodedUpdate = append(decodedUpdate, newPeer)
decodedConns = append(decodedConns, connSummaries)
existingPeer, found := peers.table[name]
existingPeer, found := peers.byName[newPeer.Name]
if !found {
newPeers[name] = newPeer
newPeers[newPeer.Name] = newPeer
} else if existingPeer.UID != newPeer.UID {
err = NameCollisionError{Name: newPeer.Name}
return
@@ -239,7 +450,7 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
if _, found := newPeers[remoteName]; found {
continue
}
if _, found := peers.table[remoteName]; found {
if _, found := peers.byName[remoteName]; found {
continue
}
// Update refers to a peer which we have no knowledge
@@ -251,43 +462,8 @@ func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, de
return
}
func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]ConnectionSummary) map[PeerName]*Peer {
newUpdate := make(map[PeerName]*Peer)
for idx, newPeer := range decodedUpdate {
connSummaries := decodedConns[idx]
name := newPeer.Name
// guaranteed to find peer in the peers.table
peer := peers.table[name]
if peer != newPeer &&
(peer == peers.ourself.Peer || peer.version >= newPeer.version) {
// Nobody but us updates us. And if we know more about a
// peer than what's in the the update, we ignore the
// latter.
continue
}
// If we're here, either it was a new peer, or the update has
// more info about the peer than we do. Either case, we need
// to set version and conns and include the updated peer in
// the outgoing update.
// Can peer have been updated by anyone else in the mean time?
// No - we know that peer is not ourself, so the only prospect
// for an update would be someone else calling
// router.Peers.ApplyUpdate. But ApplyUpdate takes the Lock on
// the router.Peers, so there can be no race here.
peer.version = newPeer.version
peer.connections = makeConnsMap(peer, connSummaries, peers.table)
newUpdate[name] = peer
}
return newUpdate
}
func (peer *Peer) Encode(enc *gob.Encoder) {
checkFatal(enc.Encode(PeerSummary{
peer.NameByte,
peer.NickName,
peer.UID,
peer.version}))
checkFatal(enc.Encode(peer.PeerSummary))
connSummaries := []ConnectionSummary{}
for _, conn := range peer.connections {
@@ -319,11 +495,11 @@ func decodePeer(dec *gob.Decoder) (peerSummary PeerSummary, connSummaries []Conn
return
}
func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, table map[PeerName]*Peer) map[PeerName]Connection {
func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, byName map[PeerName]*Peer) map[PeerName]Connection {
conns := make(map[PeerName]Connection)
for _, connSummary := range connSummaries {
name := PeerNameFromBin(connSummary.NameByte)
remotePeer := table[name]
remotePeer := byName[name]
conn := NewRemoteConnection(peer, remotePeer, connSummary.RemoteTCPAddr, connSummary.Outbound, connSummary.Established)
conns[name] = conn
}
+238 -1
View File
@@ -5,6 +5,7 @@ import (
wt "github.com/weaveworks/weave/testing"
"math/rand"
"testing"
"time"
)
// TODO we should also test:
@@ -19,7 +20,6 @@ import (
func newNode(name PeerName) (*Peer, *Peers) {
peer := NewLocalPeer(name, "", nil)
peers := NewPeers(peer, func(*Peer) {})
peers.FetchWithDefault(peer.Peer)
return peer.Peer, peers
}
@@ -111,3 +111,240 @@ func TestPeersGarbageCollection(t *testing.T) {
ps1.DeleteTestConnection(p3)
checkPeerArray(t, ps1.GarbageCollect(), p3)
}
func TestShortIDCollisions(t *testing.T) {
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
_, peers := newNode(PeerName(1 << PeerShortIDBits))
// Make enough peers that short id collisions are
// overwhelmingly likely
ps := make([]*Peer, 1<<PeerShortIDBits)
for i := 0; i < 1<<PeerShortIDBits; i++ {
ps[i] = NewPeer(PeerName(i), "", PeerUID(i), 0,
PeerShortID(rng.Intn(1<<PeerShortIDBits)))
}
shuffle := func() {
for i := range ps {
j := rng.Intn(i + 1)
ps[i], ps[j] = ps[j], ps[i]
}
}
// Fill peers
shuffle()
for _, p := range ps {
peers.addByShortID(p)
}
// Check invariants
counts := make([]int, 1<<PeerShortIDBits)
saw := func(p *Peer) {
if p != peers.ourself.Peer {
counts[p.UID]++
}
}
for shortID, entry := range peers.byShortID {
if entry.peer == nil {
// no principal peer for this short id, so
// others must be empty
if len(entry.others) != 0 {
t.Fatal()
}
continue
}
if entry.peer.ShortID != shortID {
t.Fatal()
}
saw(entry.peer)
for _, p := range entry.others {
saw(p)
if p.ShortID != shortID {
t.Fatal()
}
// the principal peer should have the lowest name
if p.Name < entry.peer.Name {
t.Fatal()
}
}
}
for _, n := range counts {
if n != 1 {
t.Fatal()
}
}
// Delete all the peers
shuffle()
for _, p := range ps {
peers.deleteByShortID(p)
}
for _, entry := range peers.byShortID {
if entry.peer != nil && entry.peer != peers.ourself.Peer {
t.Fatal()
}
if len(entry.others) != 0 {
t.Fatal()
}
}
}
// Test the easy case of short id reassignment, when few short ids are taken
func TestShortIDReassignmentEasy(t *testing.T) {
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
_, peers := newNode(PeerName(0))
for i := 1; i <= 10; i++ {
peers.FetchWithDefault(NewPeer(PeerName(i), "", PeerUID(i), 0,
PeerShortID(rng.Intn(1<<PeerShortIDBits))))
}
checkShortIDReassignment(t, peers)
}
// Test the hard case of short id reassignment, when most short ids are taken
func TestShortIDReassignmentHard(t *testing.T) {
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
_, peers := newNode(PeerName(1 << PeerShortIDBits))
// Take all short ids
ps := make([]*Peer, 1<<PeerShortIDBits)
for i := 0; i < 1<<PeerShortIDBits; i++ {
ps[i] = NewPeer(PeerName(i), "", PeerUID(i), 0,
PeerShortID(i))
peers.addByShortID(ps[i])
}
// As all short ids are taken, an attempted reassigment won't
// do anything
oldShortID := peers.ourself.ShortID
peers.reassignLocalShortID()
if peers.ourself.ShortID != oldShortID {
t.Fatal()
}
// Free up a few ids
for i := 0; i < 10; i++ {
x := rng.Intn(len(ps))
if ps[x] != nil {
peers.deleteByShortID(ps[x])
ps[x] = nil
}
}
checkShortIDReassignment(t, peers)
}
func checkShortIDReassignment(t *testing.T, peers *Peers) {
oldShortID := peers.ourself.ShortID
peers.reassignLocalShortID()
if peers.ourself.ShortID == oldShortID {
t.Fatal()
}
if peers.byShortID[peers.ourself.ShortID].peer != peers.ourself.Peer {
t.Fatal()
}
}
func TestShortIDInvalidation(t *testing.T) {
_, peers := newNode(PeerName(1 << PeerShortIDBits))
// need to use a short id that is not the local peer's
shortID := peers.ourself.ShortID + 1
assertInvalidateShortIDs := func(expect bool) {
if peers.invalidateShortIDsPending != expect {
t.Fatal()
}
peers.invalidateShortIDsPending = false
}
// The use of a fresh short id does not cause invalidation
a := NewPeer(PeerName(1), "", PeerUID(1), 0, shortID)
peers.addByShortID(a)
assertInvalidateShortIDs(false)
// An addition which does not change the mapping
b := NewPeer(PeerName(2), "", PeerUID(2), 0, shortID)
peers.addByShortID(b)
assertInvalidateShortIDs(false)
// An addition which does change the mapping
c := NewPeer(PeerName(0), "", PeerUID(0), 0, shortID)
peers.addByShortID(c)
assertInvalidateShortIDs(true)
// A deletion which does not change the mapping
peers.deleteByShortID(b)
assertInvalidateShortIDs(false)
// A deletion which does change the mapping
peers.deleteByShortID(c)
assertInvalidateShortIDs(true)
// Deleting the last peer with a short id does not cause invalidation
peers.deleteByShortID(a)
assertInvalidateShortIDs(false)
// But subsequent reuse that a short id does cause invalidation
peers.addByShortID(a)
assertInvalidateShortIDs(true)
}
func TestShortIDPropagation(t *testing.T) {
_, peers1 := newNode(PeerName(0))
_, peers2 := newNode(PeerName(1))
peers1.AddTestConnection(peers2.ourself.Peer)
peers1.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
peers12, _ := peers1.Fetch(PeerName(1))
old := peers12.PeerSummary
peers2.reassignLocalShortID()
peers1.ApplyUpdate(peers2.EncodePeers(peers2.Names()))
if peers12.Version == old.Version || peers12.ShortID == old.ShortID {
t.Fatal()
}
}
// Test the case where all short ids are taken, but then some peers go
// away, so the local peer reassigns
func TestDeferredShortIDReassignment(t *testing.T) {
rng := rand.New(rand.NewSource(time.Now().UnixNano()))
_, us := newNode(PeerName(1 << PeerShortIDBits))
// Connect us to other peers occupying all short ids
others := make([]*Peers, 1<<PeerShortIDBits)
for i := range others {
_, others[i] = newNode(PeerName(i))
others[i].setLocalShortID(PeerShortID(i))
us.AddTestConnection(others[i].ourself.Peer)
}
// Check that, as expected, the local peer does not own its
// short id
if us.byShortID[us.ourself.ShortID].peer == us.ourself.Peer {
t.Fatal()
}
// Disconnect one peer, and we should now be able to claim its
// short id
other := others[rng.Intn(1<<PeerShortIDBits)]
us.DeleteTestConnection(other.ourself.Peer)
us.GarbageCollect()
if us.byShortID[us.ourself.ShortID].peer != us.ourself.Peer {
t.Fatal()
}
}
+1 -4
View File
@@ -9,10 +9,7 @@ type ProtocolTag byte
const (
ProtocolHeartbeat ProtocolTag = iota
ProtocolConnectionEstablished
ProtocolFragmentationReceived
ProtocolStartFragmentationTest
ProtocolPMTUVerified
ProtocolInterHostControlMsg
ProtocolGossip
ProtocolGossipUnicast
ProtocolGossipBroadcast
+77 -234
View File
@@ -2,13 +2,9 @@ package router
import (
"bytes"
"code.google.com/p/gopacket/layers"
"encoding/binary"
"fmt"
"io"
"log"
"net"
"syscall"
"time"
)
@@ -17,15 +13,15 @@ const macMaxAge = 10 * time.Minute // [1]
// [1] should be greater than typical ARP cache expiries, i.e. > 3/2 *
// /proc/sys/net/ipv4_neigh/*/base_reachable_time_ms on Linux
type LogFrameFunc func(string, []byte, *layers.Ethernet)
type LogFrameFunc func(string, []byte, *EthernetDecoder)
type RouterConfig struct {
Port int
Iface *net.Interface
Password []byte
ConnLimit int
BufSz int
LogFrame LogFrameFunc
IntraHost IntraHost
InterHost InterHost
}
type Router struct {
@@ -37,24 +33,21 @@ type Router struct {
ConnectionMaker *ConnectionMaker
GossipChannels map[uint32]*GossipChannel
TopologyGossip Gossip
UDPListener *net.UDPConn
}
type PacketSource interface {
ReadPacket() ([]byte, error)
}
type PacketSink interface {
WritePacket([]byte) error
}
type PacketSourceSink interface {
PacketSource
PacketSink
}
func NewRouter(config RouterConfig, name PeerName, nickName string) *Router {
router := &Router{RouterConfig: config, GossipChannels: make(map[uint32]*GossipChannel)}
// If the caller didn't set the intrahost, replace it with a
// null implementation
if router.IntraHost == nil {
router.IntraHost = NullIntraHost{}
}
if router.InterHost == nil {
router.InterHost = NullInterHost{}
}
onMacExpiry := func(mac net.HardwareAddr, peer *Peer) {
log.Println("Expired MAC", mac, "at", peer)
}
@@ -65,33 +58,18 @@ func NewRouter(config RouterConfig, name PeerName, nickName string) *Router {
router.Ourself = NewLocalPeer(name, nickName, router)
router.Macs = NewMacCache(macMaxAge, onMacExpiry)
router.Peers = NewPeers(router.Ourself, onPeerGC)
router.Peers.FetchWithDefault(router.Ourself.Peer)
router.Routes = NewRoutes(router.Ourself, router.Peers)
router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port)
router.TopologyGossip = router.NewGossip("topology", router)
return router
}
// Start listening for packets from containers, TCP connections and
// packets forwarded over UDP.
func (router *Router) Start() {
// we need two pcap handles since they aren't thread-safe
var pio PacketSourceSink
var po PacketSink
var err error
if router.Iface != nil {
pio, err = NewPcapIO(router.Iface.Name, router.BufSz)
checkFatal(err)
po, err = NewPcapO(router.Iface.Name)
checkFatal(err)
}
router.Ourself.Start()
router.Macs.Start()
router.Routes.Start()
router.ConnectionMaker.Start()
router.UDPListener = router.listenUDP(router.Port, po)
checkFatal(router.InterHost.ConsumeInterHostPackets(router.Ourself.Peer, router.Peers, router.handleForwardedPacket))
checkFatal(router.IntraHost.ConsumeIntraHostPackets(router))
router.listenTCP(router.Port)
if pio != nil {
router.sniff(pio)
}
}
func (router *Router) Stop() error {
@@ -106,7 +84,7 @@ func (router *Router) UsingPassword() bool {
func (router *Router) Status() string {
var buf bytes.Buffer
fmt.Fprintln(&buf, "Our name is", router.Ourself)
fmt.Fprintln(&buf, "Sniffing traffic on", router.Iface)
fmt.Fprintln(&buf, "Sniffing traffic on", router.IntraHost)
fmt.Fprintf(&buf, "MACs:\n%s", router.Macs)
fmt.Fprintf(&buf, "Peers:\n%s", router.Peers)
fmt.Fprintf(&buf, "Routes:\n%s", router.Routes)
@@ -114,74 +92,39 @@ func (router *Router) Status() string {
return buf.String()
}
func (router *Router) sniff(pio PacketSourceSink) {
log.Println("Sniffing traffic on", router.Iface)
func (router *Router) CapturedPacket(key PacketKey) FlowOp {
//router.LogFrame("Sniffed", frameData, nil)
srcMac := net.HardwareAddr(key.SrcMAC[:])
srcPeer := router.Macs.Lookup(srcMac)
dec := NewEthernetDecoder()
mac := router.Iface.HardwareAddr
if router.Macs.Enter(mac, router.Ourself.Peer) {
log.Println("Discovered our MAC", mac)
}
go func() {
for {
pkt, err := pio.ReadPacket()
checkFatal(err)
router.LogFrame("Sniffed", pkt, nil)
router.handleCapturedPacket(pkt, dec, pio)
}
}()
}
func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecoder, po PacketSink) {
dec.DecodeLayers(frameData)
decodedLen := len(dec.decoded)
if decodedLen == 0 {
return
}
srcMac := dec.eth.SrcMAC
srcPeer, found := router.Macs.Lookup(srcMac)
// We need to filter out frames we injected ourselves. For such
// frames, the srcMAC will have been recorded as associated with a
// different peer.
if found && srcPeer != router.Ourself.Peer {
return
if srcPeer != nil && srcPeer != router.Ourself.Peer {
return nil
}
if router.Macs.Enter(srcMac, router.Ourself.Peer) {
log.Println("Discovered local MAC", srcMac)
}
if dec.DropFrame() {
return
}
dstMac := dec.eth.DstMAC
dstPeer, found := router.Macs.Lookup(dstMac)
if found && dstPeer == router.Ourself.Peer {
return
}
df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0)
if df {
router.LogFrame("Forwarding DF", frameData, &dec.eth)
} else {
router.LogFrame("Forwarding", frameData, &dec.eth)
}
// at this point we are handing over the frame to forwarders, so
// we need to make a copy of it in order to prevent the next
// capture from overwriting the data
frameLen := len(frameData)
frameCopy := make([]byte, frameLen, frameLen)
copy(frameCopy, frameData)
// If we don't know which peer corresponds to the dest MAC,
// broadcast it.
if !found {
router.Ourself.Broadcast(df, frameCopy, dec)
return
// Discard STP broadcasts
if key.DstMAC == [...]byte{0x01, 0x80, 0xC2, 0x00, 0x00, 0x00} {
return nil
}
err := router.Ourself.Forward(dstPeer, df, frameCopy, dec)
if ftbe, ok := err.(FrameTooBigError); ok {
err = dec.sendICMPFragNeeded(ftbe.EPMTU, po.WritePacket)
dstMac := net.HardwareAddr(key.DstMAC[:])
switch dstPeer := router.Macs.Lookup(dstMac); dstPeer {
case router.Ourself.Peer:
return nil
case nil:
// If we don't know which peer corresponds to the dest
// MAC, broadcast it.
return router.Ourself.Broadcast(key)
default:
//router.LogFrame("Forwarding", frameData, dec)
return router.Ourself.Forward(dstPeer, key)
}
checkWarn(err)
}
func (router *Router) listenTCP(localPort int) {
@@ -209,152 +152,47 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) {
remoteAddrStr := tcpConn.RemoteAddr().String()
log.Printf("->[%s] connection accepted\n", remoteAddrStr)
connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false, false)
connLocal := NewLocalConnection(connRemote, tcpConn, nil, router)
connLocal.Start(true)
StartLocalConnection(connRemote, tcpConn, nil, router, true)
}
func (router *Router) listenUDP(localPort int, po PacketSink) *net.UDPConn {
localAddr, err := net.ResolveUDPAddr("udp4", fmt.Sprint(":", localPort))
checkFatal(err)
conn, err := net.ListenUDP("udp4", localAddr)
checkFatal(err)
f, err := conn.File()
defer f.Close()
checkFatal(err)
fd := int(f.Fd())
// This one makes sure all packets we send out do not have DF set on them.
err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DONT)
checkFatal(err)
go router.udpReader(conn, po)
return conn
}
func (router *Router) udpReader(conn *net.UDPConn, po PacketSink) {
defer conn.Close()
dec := NewEthernetDecoder()
buf := make([]byte, MaxUDPPacketSize)
for {
n, sender, err := conn.ReadFromUDP(buf)
if err == io.EOF {
return
} else if err != nil {
log.Println("ignoring UDP read error", err)
continue
} else if n < NameSize {
log.Println("ignoring too short UDP packet from", sender)
continue
}
name := PeerNameFromBin(buf[:NameSize])
packet := make([]byte, n-NameSize)
copy(packet, buf[NameSize:n])
peerConn, found := router.Ourself.ConnectionTo(name)
if !found {
continue
}
relayConn, ok := peerConn.(*LocalConnection)
if !ok {
continue
}
if err := relayConn.Decryptor.IterateFrames(packet, router.handleUDPPacketFunc(relayConn, dec, sender, po)); err != nil {
// Errors during UDP packet decoding / processing are
// non-fatal. One common cause is that we receive and
// attempt to decrypt a "stray" packet. This can actually
// happen quite easily if there is some connection churn
// between two peers. After all, UDP isn't a
// connection-oriented protocol, yet we pretend it is.
//
// If anything really is seriously, unrecoverably amiss
// with a connection, that will typically result in missed
// heartbeats and the connection getting shut down because
// of that.
relayConn.Log(err)
}
func (router *Router) handleForwardedPacket(key ForwardPacketKey) FlowOp {
if key.DstPeer != router.Ourself.Peer {
// it's not for us, we're just relaying it
//router.LogFrame("Relaying", frame, dec)
return router.Ourself.Relay(key)
}
}
func (router *Router) handleUDPPacketFunc(relayConn *LocalConnection, dec *EthernetDecoder, sender *net.UDPAddr, po PacketSink) FrameConsumer {
return func(srcNameByte, dstNameByte []byte, frame []byte) {
srcPeer, found := router.Peers.Fetch(PeerNameFromBin(srcNameByte))
if !found {
return
}
dstPeer, found := router.Peers.Fetch(PeerNameFromBin(dstNameByte))
if !found {
return
}
// At this point, it's either unicast to us, or a broadcast
// (because the DstPeer on a forwarded broadcast packet is
// always set to the peer being forwarded to)
dec.DecodeLayers(frame)
decodedLen := len(dec.decoded)
if decodedLen == 0 {
return
}
// Handle special frames produced internally (rather than
// captured/forwarded) by the remote router.
//
// We really shouldn't be decoding these above, since they are
// not genuine Ethernet frames. However, it is actually more
// efficient to do so, as we want to optimise for the common
// (i.e. non-special) frames. These always need decoding, and
// detecting special frames is cheaper post decoding than pre.
if decodedLen == 1 && dec.IsSpecial() {
if srcPeer == relayConn.Remote() && dstPeer == router.Ourself.Peer {
handleSpecialFrame(relayConn, sender, frame)
}
return
}
df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0)
if dstPeer != router.Ourself.Peer {
// it's not for us, we're just relaying it
if df {
router.LogFrame("Relaying DF", frame, &dec.eth)
} else {
router.LogFrame("Relaying", frame, &dec.eth)
}
err := router.Ourself.Relay(srcPeer, dstPeer, df, frame, dec)
if ftbe, ok := err.(FrameTooBigError); ok {
err = dec.sendICMPFragNeeded(ftbe.EPMTU, func(icmpFrame []byte) error {
return router.Ourself.Forward(srcPeer, false, icmpFrame, nil)
})
}
checkWarn(err)
return
}
srcMac := dec.eth.SrcMAC
dstMac := dec.eth.DstMAC
if router.Macs.Enter(srcMac, srcPeer) {
log.Println("Discovered remote MAC", srcMac, "at", srcPeer)
}
if po != nil {
router.LogFrame("Injecting", frame, &dec.eth)
checkWarn(po.WritePacket(frame))
}
dstPeer, found = router.Macs.Lookup(dstMac)
if !found || dstPeer != router.Ourself.Peer {
router.LogFrame("Relaying broadcast", frame, &dec.eth)
router.Ourself.RelayBroadcast(srcPeer, df, frame, dec)
}
srcMac := net.HardwareAddr(key.SrcMAC[:])
dstMac := net.HardwareAddr(key.DstMAC[:])
if router.Macs.Enter(srcMac, key.SrcPeer) {
log.Println("Discovered remote MAC", srcMac, "at", key.SrcPeer)
}
}
func handleSpecialFrame(relayConn *LocalConnection, sender *net.UDPAddr, frame []byte) {
frameLen := len(frame)
//router.LogFrame("Injecting", frame, dec)
injectFop := router.IntraHost.InjectPacket(key.PacketKey)
dstPeer := router.Macs.Lookup(dstMac)
if dstPeer == router.Ourself.Peer {
return injectFop
}
//router.LogFrame("Relaying broadcast", frame, dec)
relayFop := router.Ourself.RelayBroadcast(key.SrcPeer, key.PacketKey)
switch {
case frameLen == EthernetOverhead+8:
relayConn.ReceivedHeartbeat(sender, binary.BigEndian.Uint64(frame[EthernetOverhead:]))
case frameLen == FragTestSize && bytes.Equal(frame, FragTest):
relayConn.SendProtocolMsg(ProtocolMsg{ProtocolFragmentationReceived, nil})
case frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery):
case injectFop == nil:
return relayFop
case relayFop == nil:
return injectFop
default:
frameLenBytes := []byte{0, 0}
binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead))
relayConn.SendProtocolMsg(ProtocolMsg{ProtocolPMTUVerified, frameLenBytes})
mfop := NewMultiFlowOp(false)
mfop.Add(injectFop)
mfop.Add(relayFop)
return mfop
}
}
@@ -426,3 +264,8 @@ func (router *Router) applyTopologyUpdate(update []byte) (PeerNameSet, PeerNameS
}
return origUpdate, newUpdate, nil
}
func (router *Router) broadcastPeerUpdate(peers ...*Peer) {
router.Routes.Recalculate()
router.TopologyGossip.GossipBroadcast(NewTopologyGossipData(router.Peers, append(peers, router.Ourself.Peer)...))
}
+78 -20
View File
@@ -7,14 +7,17 @@ import (
"sync"
)
type unicastRoutes map[PeerName]PeerName
type broadcastRoutes map[PeerName][]PeerName
type Routes struct {
sync.RWMutex
ourself *LocalPeer
peers *Peers
unicast map[PeerName]PeerName
unicastAll map[PeerName]PeerName // [1]
broadcast map[PeerName][]PeerName
broadcastAll map[PeerName][]PeerName // [1]
unicast unicastRoutes
unicastAll unicastRoutes // [1]
broadcast broadcastRoutes
broadcastAll broadcastRoutes // [1]
recalculate chan<- *struct{}
wait chan<- chan struct{}
// [1] based on *all* connections, not just established &
@@ -22,26 +25,23 @@ type Routes struct {
}
func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes {
recalculate := make(chan *struct{}, 1)
wait := make(chan chan struct{})
routes := &Routes{
ourself: ourself,
peers: peers,
unicast: make(map[PeerName]PeerName),
unicastAll: make(map[PeerName]PeerName),
broadcast: make(map[PeerName][]PeerName),
broadcastAll: make(map[PeerName][]PeerName)}
unicast: make(unicastRoutes),
unicastAll: make(unicastRoutes),
broadcast: make(broadcastRoutes),
broadcastAll: make(broadcastRoutes),
recalculate: recalculate,
wait: wait}
routes.unicast[ourself.Name] = UnknownPeerName
routes.unicastAll[ourself.Name] = UnknownPeerName
routes.broadcast[ourself.Name] = []PeerName{}
routes.broadcastAll[ourself.Name] = []PeerName{}
return routes
}
func (routes *Routes) Start() {
recalculate := make(chan *struct{}, 1)
wait := make(chan chan struct{})
routes.recalculate = recalculate
routes.wait = wait
go routes.run(recalculate, wait)
return routes
}
func (routes *Routes) PeerNames() PeerNameSet {
@@ -169,6 +169,8 @@ func (routes *Routes) calculate() {
routes.peers.RLock()
routes.ourself.RLock()
var (
oldUnicast = routes.unicast
oldBroadcast = routes.broadcast
unicast = routes.calculateUnicast(true)
unicastAll = routes.calculateUnicast(false)
broadcast = routes.calculateBroadcast(true)
@@ -183,6 +185,10 @@ func (routes *Routes) calculate() {
routes.broadcast = broadcast
routes.broadcastAll = broadcastAll
routes.Unlock()
if !unicast.equals(oldUnicast) || !broadcast.equals(oldBroadcast) {
routes.ourself.router.InterHost.InvalidateRoutes()
}
}
// Calculate all the routes for the question: if *we* want to send a
@@ -194,7 +200,7 @@ func (routes *Routes) calculate() {
// any knowledge of the MAC address at all. Thus there's no need
// to exchange knowledge of MAC addresses, nor any constraints on
// the routes that we construct.
func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) map[PeerName]PeerName {
func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) unicastRoutes {
_, unicast := routes.ourself.Routes(nil, establishedAndSymmetric)
return unicast
}
@@ -217,9 +223,9 @@ func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) map[PeerNam
// Y =/= Z /\ X.Routes(Y) <= X.Routes(Z) =>
// X.Routes(Y) u [P | Y.HasSymmetricConnectionTo(P)] <= X.Routes(Z)
// where <= is the subset relationship on keys of the returned map.
func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerName][]PeerName {
broadcast := make(map[PeerName][]PeerName)
for _, peer := range routes.peers.table {
func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) broadcastRoutes {
broadcast := make(broadcastRoutes)
for _, peer := range routes.peers.byName {
hops := []PeerName{}
if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found {
routes.ourself.ForEachConnectedPeer(establishedAndSymmetric, reached,
@@ -229,3 +235,55 @@ func (routes *Routes) calculateBroadcast(establishedAndSymmetric bool) map[PeerN
}
return broadcast
}
func (a unicastRoutes) equals(b unicastRoutes) bool {
for key, aval := range a {
if bval, ok := b[key]; !ok || bval != aval {
return false
}
}
for key := range b {
if _, ok := a[key]; !ok {
return false
}
}
return true
}
func (a broadcastRoutes) equals(b broadcastRoutes) bool {
set := make(map[PeerName]struct{})
for key, aval := range a {
bval, ok := b[key]
if !ok {
return false
}
for _, peer := range aval {
set[peer] = struct{}{}
}
for _, peer := range bval {
if _, ok := set[peer]; !ok {
return false
}
delete(set, peer)
}
if len(set) != 0 {
return false
}
}
for key := range b {
if _, ok := a[key]; !ok {
return false
}
}
return true
}
+1098
View File
File diff suppressed because it is too large. Load diff
-138
View File
@@ -1,138 +0,0 @@
package router
import (
"code.google.com/p/gopacket"
"code.google.com/p/gopacket/layers"
"log"
"net"
"syscall"
)
type UDPSender interface {
Send([]byte) error
Shutdown() error
}
type SimpleUDPSender struct {
conn *LocalConnection
udpConn *net.UDPConn
}
type RawUDPSender struct {
ipBuf gopacket.SerializeBuffer
opts gopacket.SerializeOptions
udpHeader *layers.UDP
socket *net.IPConn
conn *LocalConnection
}
type MsgTooBigError struct {
PMTU int // actual pmtu, i.e. what the kernel told us
}
func NewSimpleUDPSender(conn *LocalConnection) *SimpleUDPSender {
return &SimpleUDPSender{udpConn: conn.Router.UDPListener, conn: conn}
}
func (sender *SimpleUDPSender) Send(msg []byte) error {
_, err := sender.udpConn.WriteToUDP(msg, sender.conn.RemoteUDPAddr())
return err
}
func (sender *SimpleUDPSender) Shutdown() error {
return nil
}
func NewRawUDPSender(conn *LocalConnection) (*RawUDPSender, error) {
ipSocket, err := dialIP(conn)
if err != nil {
return nil, err
}
udpHeader := &layers.UDP{SrcPort: layers.UDPPort(conn.Router.Port)}
ipBuf := gopacket.NewSerializeBuffer()
opts := gopacket.SerializeOptions{
FixLengths: true,
// UDP header is calculated with a phantom IP
// header. Yes, it's totally nuts. Thankfully, for UDP
// over IPv4, the checksum is optional. It's not
// optional for IPv6, but we'll ignore that for
// now. TODO
ComputeChecksums: false}
return &RawUDPSender{
ipBuf: ipBuf,
opts: opts,
udpHeader: udpHeader,
socket: ipSocket,
conn: conn}, nil
}
func (sender *RawUDPSender) Send(msg []byte) error {
payload := gopacket.Payload(msg)
sender.udpHeader.DstPort = layers.UDPPort(sender.conn.RemoteUDPAddr().Port)
err := gopacket.SerializeLayers(sender.ipBuf, sender.opts, sender.udpHeader, &payload)
if err != nil {
return err
}
packet := sender.ipBuf.Bytes()
_, err = sender.socket.Write(packet)
if err == nil || PosixError(err) != syscall.EMSGSIZE {
return err
}
f, err := sender.socket.File()
if err != nil {
return err
}
defer f.Close()
fd := int(f.Fd())
log.Println("EMSGSIZE on send, expecting PMTU update (IP packet was",
len(packet), "bytes, payload was", len(msg), "bytes)")
pmtu, err := syscall.GetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU)
if err != nil {
return err
}
return MsgTooBigError{PMTU: pmtu}
}
func (sender *RawUDPSender) Shutdown() error {
defer func() { sender.socket = nil }()
return sender.socket.Close()
}
func dialIP(conn *LocalConnection) (*net.IPConn, error) {
ipLocalAddr, err := ipAddr(conn.TCPConn.LocalAddr())
if err != nil {
return nil, err
}
ipRemoteAddr, err := ipAddr(conn.TCPConn.RemoteAddr())
if err != nil {
return nil, err
}
ipSocket, err := net.DialIP("ip4:UDP", ipLocalAddr, ipRemoteAddr)
if err != nil {
return nil, err
}
f, err := ipSocket.File()
if err != nil {
return nil, err
}
defer f.Close()
fd := int(f.Fd())
// This Makes sure all packets we send out have DF set on them.
err = syscall.SetsockoptInt(fd, syscall.IPPROTO_IP, syscall.IP_MTU_DISCOVER, syscall.IP_PMTUDISC_DO)
if err != nil {
return nil, err
}
return ipSocket, nil
}
func ipAddr(addr net.Addr) (*net.IPAddr, error) {
host, _, err := net.SplitHostPort(addr.String())
if err != nil {
return nil, err
}
return &net.IPAddr{
IP: net.ParseIP(host),
Zone: ""}, nil
}
-8
View File
@@ -35,14 +35,6 @@ func PosixError(err error) error {
return operr.Err
}
func (mtbe MsgTooBigError) Error() string {
return fmt.Sprint("Msg too big error. PMTU is ", mtbe.PMTU)
}
func (ftbe FrameTooBigError) Error() string {
return fmt.Sprint("Frame too big error. Effective PMTU is ", ftbe.EPMTU)
}
func (upe UnknownPeerError) Error() string {
return fmt.Sprint("Reference to unknown peer ", upe.Name)
}
+42 -24
View File
@@ -65,6 +65,7 @@ exec_remote() {
-e WEAVE_PASSWORD \
-e WEAVE_PORT \
-e WEAVE_CONTAINER_NAME \
-e WEAVE_MTU \
-e DOCKER_BRIDGE \
$WEAVEEXEC_DOCKER_ARGS $EXEC_IMAGE --local "$@"
}
@@ -164,7 +165,6 @@ CONTAINER_NAME=${WEAVE_CONTAINER_NAME:-weave}
DNS_CONTAINER_NAME=weavedns
BRIDGE=weave
CONTAINER_IFNAME=ethwe
MTU=65535
PORT=${WEAVE_PORT:-6783}
HTTP_PORT=6784
DNS_HTTP_PORT=6785
@@ -176,6 +176,13 @@ if [ "$DOCKER_HOST" != "${DOCKER_HOST#tcp://}" ] ; then
fi
# GCE has the lowest underlay network MTU we're likely to encounter on
# a local network, at 1460 bytes. To get the overlay MTU from that we
# subtract 20 bytes for the outer IPv4 header, 8 bytes for the outer
# UDP hear, 8 bytes for the vxlan header, and 14 bytes for the inner
# ethernet header.
MTU=${WEAVE_MTU:-1410}
######################################################################
# general helpers; independent of docker and weave
######################################################################
@@ -256,7 +263,8 @@ random_mac() {
create_bridge() {
[ ! -d /sys/class/net/$BRIDGE ] && {
ip link add name $BRIDGE type bridge
docker run --rm --privileged --net=host $IMAGE --createdp --dpname=$BRIDGE
#ip link add name $BRIDGE type bridge
# Set a random MAC address on the bridge. Current Linux
# kernels already do this when creating a bridge, but there
# are rumours it was not always so.
@@ -265,9 +273,10 @@ create_bridge() {
# fails. Bridges take the lowest MTU of their interfaces. So
# instead we create a temporary interface with the desired
# MTU, attach that to the bridge, and then remove it again.
ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy
ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE
ip link del dev v${CONTAINER_IFNAME}du
#ip link add name v${CONTAINER_IFNAME}du mtu $MTU type dummy
#ip link set dev v${CONTAINER_IFNAME}du master $BRIDGE
#ip link del dev v${CONTAINER_IFNAME}du
ip link set dev $BRIDGE mtu $MTU
# Drop traffic from Docker bridge to Weave; it can break subnet isolation
if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then
# Note using -I to insert ahead of Docker's bridge rules
@@ -293,7 +302,8 @@ create_bridge() {
destroy_bridge() {
if [ -d /sys/class/net/$BRIDGE ] ; then
ip link del dev $BRIDGE
docker run --rm --privileged --net=host $IMAGE --deletedp --dpname=$BRIDGE
#ip link del dev $BRIDGE
fi
if [ "$DOCKER_BRIDGE" != "$BRIDGE" ] ; then
run_iptables -t filter -D FORWARD -i $DOCKER_BRIDGE -o $BRIDGE -j DROP 2>/dev/null || true
@@ -360,16 +370,14 @@ netnsenter() {
}
connect_container_to_bridge() {
if [ -h "$PROCFS/$CONTAINER_PID/ns/net" -a -h "$PROCFS/$$/ns/net" -a "$(readlink $PROCFS/$CONTAINER_PID/ns/net)" = "$(readlink $PROCFS/$$/ns/net)" ] ; then
echo "Container is running in the host network namespace, and therefore cannot be" >&2
echo "connected to weave. Perhaps the container was started with --net=host." >&2
return 1
fi
#if [ -h "$PROCFS/$CONTAINER_PID/ns/net" -a -h "$PROCFS/$$/ns/net" -a "$(readlink $PROCFS/$CONTAINER_PID/ns/net)" = "$(readlink $PROCFS/$$/ns/net)" ] ; then
# echo "Container is running in the host network namespace, and therefore cannot be" >&2
# echo "connected to weave. Perhaps the container was started with --net=host." >&2
# return 1
#fi
ip link add name $LOCAL_IFNAME mtu $MTU type veth peer name $GUEST_IFNAME mtu $MTU || return 1
if ! ethtool -K $GUEST_IFNAME tx off >/dev/null ||
! ip link set $LOCAL_IFNAME master $BRIDGE ||
! ip link set $LOCAL_IFNAME up ||
! ip link set $GUEST_IFNAME netns $PROCFS/$CONTAINER_PID/ns/net ; then
# failed before we assigned the veth to the container's
# namespace
@@ -377,8 +385,18 @@ connect_container_to_bridge() {
return 1
fi
if ! netnsenter ip link set $GUEST_IFNAME name $CONTAINER_IFNAME ||
! configure_arp_cache $CONTAINER_IFNAME "netnsenter" ; then
# Versions of the 'ip' command differ in whether they create a
# veth pair in the 'up' or 'down' state. Furthermore, moving one
# end into a netns normally sets them to 'down'. But, moving an
# end into the default netns is a null operation, so they stay up,
# which leads to errors later on. So to get a consistent result,
# we have to be explicit in setting them 'up' or 'down' as
# necessary.
if ! netnsenter ip link set $GUEST_IFNAME down ||
! netnsenter ip link set $GUEST_IFNAME name $CONTAINER_IFNAME up ||
! ip link set $LOCAL_IFNAME up ||
! docker run --rm --privileged --net=host $IMAGE --adddpif=$LOCAL_IFNAME --dpname=$BRIDGE ||
! configure_arp_cache $CONTAINER_IFNAME "netnsenter" ; then
return 1
fi
}
@@ -397,12 +415,12 @@ ask_version() {
######################################################################
launch() {
if ! netnsenter ip link show eth0 >/dev/null ; then
echo "Perhaps you are running the docker daemon with container networking disabled (-b=none)." >&2
return 1
fi
#if ! netnsenter ip link show eth0 >/dev/null ; then
# echo "Perhaps you are running the docker daemon with container networking disabled (-b=none)." >&2
# return 1
#fi
connect_container_to_bridge &&
netnsenter ethtool -K eth0 tx off >/dev/null &&
#netnsenter ethtool -K eth0 tx off >/dev/null &&
netnsenter ip link set $CONTAINER_IFNAME up
}
@@ -509,8 +527,8 @@ container_ip() {
fi
case "$status" in
"true ")
echo "$1 container has no IP address; is Docker networking enabled?" >&2
return 1
#echo "$1 container has no IP address; is Docker networking enabled?" >&2
CONTAINER_IP=127.0.0.1
;;
true*)
CONTAINER_IP="${status#true }"
@@ -806,8 +824,8 @@ case "$COMMAND" in
# additional parameters, such as resource limits, to docker
# when launching the weave container.
CONTAINER=$(docker run --privileged -d --name=$CONTAINER_NAME \
-p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp -e WEAVE_PASSWORD -v /var/run/docker.sock:/var/run/docker.sock \
$WEAVE_DOCKER_ARGS $IMAGE -iface $CONTAINER_IFNAME -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" $IPRANGE "$@")
-p $PORT:$CONTAINER_PORT/tcp -p $PORT:$CONTAINER_PORT/udp -e WEAVE_PASSWORD -v /var/run/docker.sock:/var/run/docker.sock --net=host \
$WEAVE_DOCKER_ARGS $IMAGE -dpname $BRIDGE -port $CONTAINER_PORT -name "$PEERNAME" -nickname "$(hostname)" $IPRANGE "$@")
with_container_netns $CONTAINER launch >/dev/null
wait_for_status $CONTAINER_NAME $HTTP_PORT
+89 -47
View File
@@ -1,7 +1,6 @@
package main
import (
"code.google.com/p/gopacket/layers"
"crypto/sha256"
"flag"
"fmt"
@@ -23,7 +22,6 @@ import (
var version = "(unreleased version)"
func main() {
log.SetPrefix(weave.Protocol + " ")
log.SetFlags(log.Ldate | log.Ltime | log.Lmicroseconds)
@@ -36,25 +34,36 @@ func main() {
runtime.GOMAXPROCS(procs)
var (
config weave.RouterConfig
justVersion bool
ifaceName string
routerName string
nickName string
password string
wait int
debug bool
pktdebug bool
prof string
peers []string
bufSzMB int
httpAddr string
iprangeCIDR string
peerCount int
apiPath string
// flags that cause immediate exit
justVersion bool
createDatapath bool
deleteDatapath bool
addDatapathInterface string
config weave.RouterConfig
ifaceName string
routerName string
nickName string
password string
wait int
debug bool
pktdebug bool
prof string
peers []string
bufSzMB int
httpAddr string
iprangeCIDR string
peerCount int
apiPath string
iface *net.Interface
datapathName string
)
flag.BoolVar(&justVersion, "version", false, "print version and exit")
flag.BoolVar(&createDatapath, "createdp", false, "create ODP datapath and exit")
flag.BoolVar(&deleteDatapath, "deletedp", false, "delete ODP datapath and exit")
flag.StringVar(&addDatapathInterface, "adddpif", "", "add a network interface to the ODP datapath and exit")
flag.IntVar(&config.Port, "port", weave.Port, "router port")
flag.StringVar(&ifaceName, "iface", "", "name of interface to capture/inject from (disabled if blank)")
flag.StringVar(&routerName, "name", "", "name of router (defaults to MAC of interface)")
@@ -70,13 +79,25 @@ func main() {
flag.StringVar(&iprangeCIDR, "iprange", "", "IP address range to allocate within, in CIDR notation")
flag.IntVar(&peerCount, "initpeercount", 0, "number of peers in network (for IP address allocation)")
flag.StringVar(&apiPath, "api", "unix:///var/run/docker.sock", "Path to Docker API socket")
flag.StringVar(&datapathName, "dpname", "", "ODP datapath name")
flag.Parse()
peers = flag.Args()
InitDefaultLogging(debug)
if justVersion {
switch {
case justVersion:
fmt.Printf("weave router %s\n", version)
os.Exit(0)
case createDatapath:
checkFatal(weave.CreateDatapath(datapathName))
os.Exit(0)
case deleteDatapath:
checkFatal(weave.DeleteDatapath(datapathName))
os.Exit(0)
case addDatapathInterface != "":
checkFatal(weave.AddDatapathInterface(datapathName, addDatapathInterface))
os.Exit(0)
}
log.Println("Command line options:", options())
@@ -84,33 +105,47 @@ func main() {
var err error
if password == "" {
password = os.Getenv("WEAVE_PASSWORD")
}
if ifaceName != "" {
config.Iface, err = weavenet.EnsureInterface(ifaceName, wait)
if err != nil {
log.Fatal(err)
iface, err := weavenet.EnsureInterface(ifaceName, wait)
checkFatal(err)
// bufsz flag is in MB
config.IntraHost, err = weave.NewPcap(iface, bufSzMB*1024*1024)
checkFatal(err)
config.InterHost = weave.NewSleeveInterHost(config.Port)
}
if datapathName != "" {
if ifaceName != "" {
log.Fatal("Both -iface and -dpname supplied")
}
if password != "" {
log.Fatal("-password supplied; encryption is not suppoorted with fast datapath")
}
fastdp, err := weave.NewFastDatapath(datapathName, config.Port)
checkFatal(err)
config.IntraHost = fastdp
config.InterHost = fastdp
}
if routerName == "" {
if config.Iface == nil {
if iface == nil {
log.Fatal("Either an interface must be specified with -iface or a name with -name")
}
routerName = config.Iface.HardwareAddr.String()
routerName = iface.HardwareAddr.String()
}
name, err := weave.PeerNameFromUserInput(routerName)
if err != nil {
log.Fatal(err)
}
checkFatal(err)
if nickName == "" {
nickName, err = os.Hostname()
if err != nil {
log.Fatal(err)
}
}
if password == "" {
password = os.Getenv("WEAVE_PASSWORD")
checkFatal(err)
}
if password == "" {
@@ -127,7 +162,6 @@ func main() {
defer profile.Start(&p).Stop()
}
config.BufSz = bufSzMB * 1024 * 1024
config.LogFrame = logFrameFunc(pktdebug)
router := weave.NewRouter(config, name, nickName)
@@ -169,31 +203,33 @@ func options() map[string]string {
func logFrameFunc(debug bool) weave.LogFrameFunc {
if !debug {
return func(prefix string, frame []byte, eth *layers.Ethernet) {}
return func(prefix string, frame []byte, dec *weave.EthernetDecoder) {}
}
return func(prefix string, frame []byte, eth *layers.Ethernet) {
return func(prefix string, frame []byte, dec *weave.EthernetDecoder) {
h := fmt.Sprintf("%x", sha256.Sum256(frame))
if eth == nil {
log.Println(prefix, len(frame), "bytes (", h, ")")
} else {
log.Println(prefix, len(frame), "bytes (", h, "):", eth.SrcMAC, "->", eth.DstMAC)
parts := []interface{}{prefix, len(frame), "bytes (", h, ")"}
if dec != nil {
parts = append(parts, dec.Eth.SrcMAC, "->", dec.Eth.DstMAC)
if dec.DF() {
parts = append(parts, "(DF)")
}
}
log.Println(parts...)
}
}
func initiateConnections(router *weave.Router, peers []string) {
for _, peer := range peers {
if err := router.ConnectionMaker.InitiateConnection(peer); err != nil {
log.Fatal(err)
}
checkFatal(router.ConnectionMaker.InitiateConnection(peer))
}
}
func createAllocator(router *weave.Router, apiPath string, iprangeCIDR string, quorum uint) *ipam.Allocator {
allocator, err := ipam.NewAllocator(router.Ourself.Peer.Name, router.Ourself.Peer.UID, router.Ourself.Peer.NickName, iprangeCIDR, quorum)
if err != nil {
log.Fatal(err)
}
checkFatal(err)
allocator.SetInterfaces(router.NewGossip("IPallocation", allocator))
allocator.Start()
err = updater.Start(apiPath, allocator)
@@ -276,3 +312,9 @@ func handleHTTP(router *weave.Router, httpAddr string, allocator *ipam.Allocator
log.Fatal("Unable to create http server", err)
}
}
func checkFatal(e error) {
if e != nil {
log.Fatal(e)
}
}