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

Merge pull request #355 from rade/protocol_sender

Browse Source 
LGTM.
This commit is contained in:
Bryan Boreham committed 2015-01-21 11:48:54 +00:00
2 parents 9daf753d8a c6f5e7ec94
commit bdb8b4625b
6 files changed
+65 -52
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

router/connection.go
+21 -17
View File
@@ -167,7 +167,7 @@ func (conn *LocalConnection) log(args ...interface{}) {
// ACTOR client API
const (
CSendTCP = iota
CSendProtocolMsg = iota
CSetEstablished
CReceivedHeartbeat
CShutdown
@@ -200,10 +200,10 @@ func (conn *LocalConnection) SetEstablished() {
}
// Async
func (conn *LocalConnection) SendTCP(msg []byte) {
func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) {
conn.queryChan <- &ConnectionInteraction{
Interaction: Interaction{code: CSendTCP},
payload: msg}
Interaction: Interaction{code: CSendProtocolMsg},
payload: m}
}
// ACTOR server
@@ -265,16 +265,16 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction)
err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr))
case CSetEstablished:
err = conn.handleSetEstablished()
case CSendTCP:
err = conn.handleSendTCP(query.payload.([]byte))
case CSendProtocolMsg:
err = conn.handleSendProtocolMsg(query.payload.(ProtocolMsg))
}
case <-tickerChan(conn.heartbeat):
conn.Forward(true, conn.heartbeatFrame, nil)
case <-tickerChan(conn.fetchAll):
err = conn.handleSendTCP(ProtocolFetchAllByte)
err = conn.handleSendSimpleProtocolMsg(ProtocolFetchAll)
case <-tickerChan(conn.fragTest):
conn.setStackFrag(false)
err = conn.handleSendTCP(ProtocolStartFragmentationTestByte)
err = conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest)
}
}
return
@@ -295,7 +295,7 @@ func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr)
conn.receivedHeartbeat = true
conn.Unlock()
if !old {
if err := conn.handleSendTCP(ProtocolConnectionEstablishedByte); err != nil {
if err := conn.handleSendSimpleProtocolMsg(ProtocolConnectionEstablished); err != nil {
return err
}
}
@@ -332,18 +332,22 @@ func (conn *LocalConnection) handleSetEstablished() error {
conn.fragTest = time.NewTicker(FragTestInterval)
// avoid initial waits for timers to fire
conn.Forward(true, conn.heartbeatFrame, nil)
if err := conn.handleSendTCP(ProtocolFetchAllByte); err != nil {
if err := conn.handleSendSimpleProtocolMsg(ProtocolFetchAll); err != nil {
return err
}
conn.setStackFrag(false)
if err := conn.handleSendTCP(ProtocolStartFragmentationTestByte); err != nil {
if err := conn.handleSendSimpleProtocolMsg(ProtocolStartFragmentationTest); err != nil {
return err
}
return nil
}
func (conn *LocalConnection) handleSendTCP(msg []byte) error {
return conn.tcpSender.Send(msg)
func (conn *LocalConnection) handleSendSimpleProtocolMsg(tag ProtocolTag) error {
return conn.handleSendProtocolMsg(ProtocolMsg{tag: tag})
}
func (conn *LocalConnection) handleSendProtocolMsg(m ProtocolMsg) error {
return conn.tcpSender.Send(Concat([]byte{byte(m.tag)}, m.msg))
}
func (conn *LocalConnection) handleShutdown() {
@@ -530,7 +534,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
continue
}
payload := msg[1:]
switch msg[0] {
switch ProtocolTag(msg[0]) {
case ProtocolConnectionEstablished:
// We sent fast heartbeats to the remote peer, which has
// now received at least one of them and told us via this
@@ -564,13 +568,13 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
// 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.SendTCP(Concat(ProtocolUpdateByte, conn.Router.Peers.EncodeAllPeers()))
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.SendTCP(ProtocolFetchAllByte)
conn.SendProtocolMsg(ProtocolMsg{ProtocolFetchAll, nil})
continue
}
if conn.CheckFatal(err) != nil {
@@ -579,7 +583,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder) {
if len(newUpdate) != 0 {
conn.Router.ConnectionMaker.Refresh()
conn.Router.Routes.Recalculate()
conn.Router.Ourself.BroadcastTCP(Concat(ProtocolUpdateByte, newUpdate))
conn.Router.Ourself.SendProtocolMsg(ProtocolMsg{ProtocolUpdate, newUpdate})
}
case ProtocolPMTUVerified:
conn.verifyPMTU <- int(binary.BigEndian.Uint16(payload))
router/consts.go
+2 -21
View File
@@ -6,8 +6,6 @@ import (
)
const (
Protocol = "weave"
ProtocolVersion = 10
EthernetOverhead = 14
UDPOverhead = 28 // 20 bytes for IPv4, 8 bytes for UDP
Port = 6783
@@ -28,24 +26,7 @@ const (
MaxDuration = time.Duration(math.MaxInt64)
)
const (
ProtocolConnectionEstablished = iota
ProtocolFragmentationReceived
ProtocolStartFragmentationTest
ProtocolNonce
ProtocolFetchAll
ProtocolUpdate
ProtocolPMTUVerified
)
var (
FragTest = make([]byte, FragTestSize)
PMTUDiscovery = make([]byte, PMTUDiscoverySize)
ProtocolConnectionEstablishedByte = []byte{ProtocolConnectionEstablished}
ProtocolFragmentationReceivedByte = []byte{ProtocolFragmentationReceived}
ProtocolStartFragmentationTestByte = []byte{ProtocolStartFragmentationTest}
ProtocolNonceByte = []byte{ProtocolNonce}
ProtocolFetchAllByte = []byte{ProtocolFetchAll}
ProtocolUpdateByte = []byte{ProtocolUpdate}
ProtocolPMTUVerifiedByte = []byte{ProtocolPMTUVerified}
FragTest = make([]byte, FragTestSize)
PMTUDiscovery = make([]byte, PMTUDiscoverySize)
)
router/crypto.go
+4 -3
View File
@@ -85,7 +85,8 @@ func EncodeNonce(df bool) (*[24]byte, []byte, error) {
flags = flags | 1
}
SetNonceLow15Bits(&nonce, flags)
return &nonce, Concat(ProtocolNonceByte, nonce[:]), nil
// NB: need to make a copy since callers may modify the array
return &nonce, Concat(nonce[:]), nil
}
func DecodeNonce(msg []byte) (bool, *[24]byte) {
@@ -206,7 +207,7 @@ func (ne *NaClEncryptor) Bytes() []byte {
if err = ne.conn.CheckFatal(err); err != nil {
return []byte{}
}
ne.conn.SendTCP(encodedNonce)
ne.conn.SendProtocolMsg(ProtocolMsg{ProtocolNonce, encodedNonce})
ne.nonce = freshNonce
nonce = freshNonce
}
@@ -225,7 +226,7 @@ func (ne *NaClEncryptor) Bytes() []byte {
return []byte{}
}
ne.nonceChan <- nonce
ne.conn.SendTCP(encodedNonce)
ne.conn.SendProtocolMsg(ProtocolMsg{ProtocolNonce, encodedNonce})
}
ne.offset = offset
router/local_peer.go
+9 -9
View File
@@ -117,9 +117,9 @@ func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) err
const (
PAddConnection = iota
PBroadcastTCP
PDeleteConnection
PConnectionEstablished
PSendProtocolMsg
)
// Async: rely on the peer to shut us down if we shouldn't be adding
@@ -147,10 +147,10 @@ func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) {
}
// Async.
func (peer *LocalPeer) BroadcastTCP(msg []byte) {
func (peer *LocalPeer) SendProtocolMsg(m ProtocolMsg) {
peer.queryChan <- &PeerInteraction{
Interaction: Interaction{code: PBroadcastTCP},
payload: msg}
Interaction: Interaction{code: PSendProtocolMsg},
payload: m}
}
// ACTOR server
@@ -169,8 +169,8 @@ func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) {
query.resultChan <- nil
case PConnectionEstablished:
peer.handleConnectionEstablished(query.payload.(*LocalConnection))
case PBroadcastTCP:
peer.handleBroadcastTCP(query.payload.([]byte))
case PSendProtocolMsg:
peer.handleSendProtocolMsg(query.payload.(ProtocolMsg))
}
}
}
@@ -248,9 +248,9 @@ func (peer *LocalPeer) handleConnectionEstablished(conn *LocalConnection) {
peer.broadcastPeerUpdate(conn.Remote())
}
func (peer *LocalPeer) handleBroadcastTCP(msg []byte) {
func (peer *LocalPeer) handleSendProtocolMsg(m ProtocolMsg) {
peer.ForEachConnection(func(_ PeerName, conn Connection) {
conn.(*LocalConnection).SendTCP(msg)
conn.(ProtocolSender).SendProtocolMsg(m)
})
}
@@ -278,7 +278,7 @@ func (peer *LocalPeer) connectionEstablished(conn Connection) {
func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) {
peer.Router.Routes.Recalculate()
peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer.Peer)...)))
peer.handleSendProtocolMsg(ProtocolMsg{ProtocolUpdate, EncodePeers(append(peers, peer.Peer)...)})
}
func (peer *LocalPeer) checkConnectionLimit() error {
router/protocol.go
+27
View File
@@ -0,0 +1,27 @@
package router
const (
Protocol = "weave"
ProtocolVersion = 10
)
type ProtocolTag byte
const (
ProtocolConnectionEstablished ProtocolTag = iota
ProtocolFragmentationReceived
ProtocolStartFragmentationTest
ProtocolNonce
ProtocolFetchAll
ProtocolUpdate
ProtocolPMTUVerified
)
type ProtocolMsg struct {
tag ProtocolTag
msg []byte
}
type ProtocolSender interface {
SendProtocolMsg(m ProtocolMsg)
}
router/router.go
+2 -2
View File
@@ -308,12 +308,12 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F
case frameLen == EthernetOverhead+8:
relayConn.ReceivedHeartbeat(sender, binary.BigEndian.Uint64(frame[EthernetOverhead:]))
case frameLen == FragTestSize && bytes.Equal(frame, FragTest):
relayConn.SendTCP(ProtocolFragmentationReceivedByte)
relayConn.SendProtocolMsg(ProtocolMsg{ProtocolFragmentationReceived, nil})
case frameLen == PMTUDiscoverySize && bytes.Equal(frame, PMTUDiscovery):
default:
frameLenBytes := []byte{0, 0}
binary.BigEndian.PutUint16(frameLenBytes, uint16(frameLen-EthernetOverhead))
relayConn.SendTCP(Concat(ProtocolPMTUVerifiedByte, frameLenBytes))
relayConn.SendProtocolMsg(ProtocolMsg{ProtocolPMTUVerified, frameLenBytes})
}
return nil
}