reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Replace previous topology communication with gossip-based mechanism

Browse Source
This commit is contained in:
Bryan Boreham committed 2015-01-27 12:45:40 +00:00
1 parent 18df01a4bf
commit 5dc9b085a0
6 files changed
+76 -69
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

router/connection.go
-37
View File
@@ -48,7 +48,6 @@ type LocalConnection struct {
SessionKey *[32]byte
heartbeatFrame *ForwardedFrame
heartbeat *time.Ticker
fetchAll *time.Ticker
fragTest *time.Ticker
forwardChan chan<- *ForwardedFrame
forwardChanDF chan<- *ForwardedFrame
@@ -285,8 +284,6 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction)
}
case <-tickerChan(conn.heartbeat):
conn.Forward(true, conn.heartbeatFrame, nil)
case <-tickerChan(conn.fetchAll):
err = conn.handleSendSimpleProtocolMsg(ProtocolFetchAll)
case <-tickerChan(conn.fragTest):
conn.setStackFrag(false)
err = conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest)
@@ -343,13 +340,9 @@ func (conn *LocalConnection) handleSetEstablished() error {
frame: PMTUDiscovery},
nil)
conn.heartbeat = time.NewTicker(SlowHeartbeat)
conn.fetchAll = time.NewTicker(FetchAllInterval)
conn.fragTest = time.NewTicker(FragTestInterval)
// avoid initial waits for timers to fire
conn.Forward(true, conn.heartbeatFrame, nil)
if err := conn.handleSendSimpleProtocolMsg(ProtocolFetchAll); err != nil {
return err
}
conn.setStackFrag(false)
if err := conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
return err
@@ -376,7 +369,6 @@ func (conn *LocalConnection) handleShutdown() {
}
stopTicker(conn.heartbeat)
stopTicker(conn.fetchAll)
stopTicker(conn.fragTest)
// blank out the forwardChan so that the router processes don't
@@ -572,35 +564,6 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte)
return fmt.Errorf("unexpected nonce on unencrypted connection")
}
conn.Decryptor.ReceiveNonce(payload)
case ProtocolFetchAll:
// There are exactly two messages that relate to topology
// updates.
//
// 1. FetchAll. This carries no payload. The receiver responds
// with the entire topology model as the receiver has it.
//
// 2. Update. This carries a topology payload. The receiver
// merges it with its own topology model. If the payload is a
// subset of the receiver's topology, no further action is
// taken. Otherwise, the receiver sends out to all its
// connections an "improved" update.
conn.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, conn.Router.Peers.EncodeAllPeers()})
case ProtocolUpdate:
newUpdate, err := conn.Router.Peers.ApplyUpdate(payload)
if _, ok := err.(UnknownPeersError); err != nil && ok {
// That update contained a peer we didn't know about;
// request full update
conn.SendProtocolMsg(ProtocolMsg{ProtocolFetchAll, nil})
break
}
if err != nil {
return err
}
if len(newUpdate) != 0 {
conn.Router.ConnectionMaker.Refresh()
conn.Router.Routes.Recalculate()
conn.Router.Ourself.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, newUpdate})
}
case ProtocolPMTUVerified:
conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload))
case ProtocolGossipUnicast:
router/consts.go
+1 -2
View File
@@ -18,9 +18,8 @@ const (
PMTUDiscoverySize = 60000
FastHeartbeat = 500 * time.Millisecond
SlowHeartbeat = 10 * time.Second
FetchAllInterval = 30 * time.Second
FragTestInterval = 5 * time.Minute
ReadTimeout = 2 * FetchAllInterval
ReadTimeout = 1 * time.Minute
PMTUVerifyAttempts = 8
PMTUVerifyTimeout = 10 * time.Millisecond // gets doubled with every attempt
MaxDuration = time.Duration(math.MaxInt64)
router/local_peer.go
+8 -1
View File
@@ -270,7 +270,14 @@ func (peer *LocalPeer) handleSendProtocolMsg(m ProtocolMsg) {
func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
peer.Router.Routes.Recalculate()
peer.handleSendProtocolMsg(ProtocolMsg{ProtocolUpdate, EncodePeers(append(peers, peer.Peer)...)})
// TODO We should just be invoking TopologyGossip.GossipBroadcast
// here, but route calculation is asynchronous and in this
// particular case would likely result in the broadcast not
// reaching all peers. So instead we slightly break the Gossip
// abstraction (hence the cast) and send a regular update. This is
// less efficient though since it will almost certainly reach
// peers more than once.
peer.Router.TopologyGossip.(*GossipChannel).SendGossipMsg(EncodePeers(append(peers, peer.Peer)...))
}
func (peer *LocalPeer) checkConnectionLimit() error {
router/protocol.go
-2
View File
@@ -12,8 +12,6 @@ const (
ProtocolFragmentationReceived
ProtocolStartFragmentationTest
ProtocolNonce
ProtocolFetchAll
ProtocolUpdate
ProtocolPMTUVerified
ProtocolGossip
ProtocolGossipUnicast
router/router.go
+38
View File
@@ -25,6 +25,7 @@ type Router struct {
Routes *Routes
ConnectionMaker *ConnectionMaker
GossipChannels map[uint32]*GossipChannel
TopologyGossip Gossip
UDPListener *net.UDPConn
Password *[]byte
ConnLimit int
@@ -68,6 +69,7 @@ func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit i
router.Peers.FetchWithDefault(router.Ourself.Peer)
router.Routes = NewRoutes(router.Ourself.Peer, router.Peers)
router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers)
router.TopologyGossip = router.NewGossip("topology", router)
return router
}
@@ -350,3 +352,39 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F
return nil
}
}
// Gossiper methods - the Router is the topology Gossiper
func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error {
return fmt.Errorf("unexpected topology gossip unicast: %v", msg)
}
func (router *Router) OnGossipBroadcast(msg []byte) error {
return fmt.Errorf("unexpected topology gossip broadcast: %v", msg)
}
// Return state of everything we know; intended to be called periodically
func (router *Router) Gossip() []byte {
return router.Peers.EncodeAllPeers()
}
// merge in state and return "everything new I've just learnt",
// or nil if nothing in the received message was new
func (router *Router) OnGossip(buf []byte) ([]byte, error) {
newUpdate, err := router.Peers.ApplyUpdate(buf)
if _, ok := err.(UnknownPeersError); err != nil && ok {
// That update contained a peer we didn't know about; we
// ignore this; eventually we should receive an update
// containing a complete topology.
return nil, nil
}
if err != nil {
return nil, err
}
if len(newUpdate) == 0 {
return nil, nil
}
router.ConnectionMaker.Refresh()
router.Routes.Recalculate()
return newUpdate, nil
}
site/how-it-works.md
+29 -27
View File
@@ -105,38 +105,40 @@ other peers. Weave peers communicate their knowledge of the topology
information on to their own neighbours, and so on, until the entire
network knows about any change.
Topology is communicated over the TCP links between peers. There are
two message types:
Topology is communicated over the TCP links between peers, using a
Gossip mechanism. Topology messages are sent by a peer...
##### FetchAll
This carries no payload. The receiver responds with the entire
topology model as the receiver has it. FetchAll is sent:
- when a connection has been established; the entire topology is sent
to the remote peer, and an incremental update, containing
information on just the two peers at the ends of the connection, is
sent to all neighbours
- when a connection has been torn down; an update containing just
information about the local peer is sent to all neighbours
- periodically, on a timer, in case someone has missed an update.
* when an update is received that references a peer that the receiver
does not know
* periodically, on a timer, in case any updates have been missed
##### Update
This carries a topology payload. This is sent:
* upon receipt of a FetchAll message, as above,
* when a connection is established - the update contains the two
peers that have just connected
* when a connection is deleted - the update contains just the peer
that lost the connection
The receiver merges the update with its own topology model, adding
peers hitherto unknown to it, and updating peers for which the update
contains a more recent version than known to it. If there were any
such new/updated peers, then an improved update containing them is
sent out on all connections.
The receiver of a topology update merges that update with its own
topology model, adding peers hitherto unknown to it, and updating
peers for which the update contains a more recent version than known
to it. If there were any such new/updated peers, then an improved
update containing them is sent out on all connections.
If the update mentions a peer that the receiver does not know, then
the entire update is rejected and the receiver sends a FetchAll
message back to the sender.
the entire update is ignored.
#### Message details
A topology update message is laid out like this:
Every gossip message is structured as follows:
+-----------------------------------+
| 1-byte message type - Gossip |
+-----------------------------------+
| 4-byte Gossip channel - Topology |
+-----------------------------------+
| Peer Name of source |
+-----------------------------------+
| Gossip payload (topology update) |
+-----------------------------------+
The topology update payload is laid out like this:
+-----------------------------------+
| Peer 1: Name |
@@ -183,7 +185,7 @@ no longer has any connections within the network, it drops all
knowledge of that second peer.
#### Out-of-date topology
The peer-to-peer passing of updates is not instantaneous, so it is
The peer-to-peer gossiping of updates is not instantaneous, so it is
very possible for a node elsewhere in the network to have an
out-of-date view.