From dc275d70097eb69d70040a9d28dbf465c9296a1f Mon Sep 17 00:00:00 2001 From: David Wragg Date: Mon, 21 Sep 2015 08:33:11 +0100 Subject: [PATCH] Restore sleeve compatibility with 1.1 Addresses #1396 --- router/connection.go | 8 ++--- router/overlay.go | 10 +++--- router/protocol.go | 5 ++- router/sleeve.go | 79 +++++++++++++++++++++----------------------- 4 files changed, 51 insertions(+), 51 deletions(-) diff --git a/router/connection.go b/router/connection.go index e3967ec6..ed1b94df 100644 --- a/router/connection.go +++ b/router/connection.go @@ -376,8 +376,8 @@ func (conn *LocalConnection) forwarderCrypto() *OverlayCrypto { } } -func (conn *LocalConnection) sendOverlayControlMessage(msg []byte) error { - return conn.sendProtocolMsg(ProtocolMsg{ProtocolOverlayControlMsg, msg}) +func (conn *LocalConnection) sendOverlayControlMessage(tag ProtocolTag, msg []byte) error { + return conn.sendProtocolMsg(ProtocolMsg{tag, msg}) } type ConnectionAsForwarderListener struct{ conn *LocalConnection } @@ -432,8 +432,8 @@ func (conn *LocalConnection) receiveTCP(receiver TCPReceiver) { func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) error { switch tag { case ProtocolHeartbeat: - case ProtocolOverlayControlMsg: - conn.forwarder.ControlMessage(payload) + case ProtocolConnectionEstablished, ProtocolFragmentationReceived, ProtocolPMTUVerified, ProtocolOverlayControlMsg: + conn.forwarder.ControlMessage(tag, payload) case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip: return conn.Router.handleGossip(tag, payload) default: diff --git a/router/overlay.go b/router/overlay.go index 0be39ae9..b84db7d6 100644 --- a/router/overlay.go +++ b/router/overlay.go @@ -38,7 +38,7 @@ type ForwarderParams struct { // Function to send a control message to the counterpart // forwarder. - SendControlMessage func([]byte) error + SendControlMessage func(tag ProtocolTag, msg []byte) error } // When a consumer is called, the decoder will already have been used @@ -66,8 +66,10 @@ type OverlayForwarder interface { Stop() - // Handle a message from the peer - ControlMessage([]byte) + // Handle a message from the peer. 'tag' exists for + // compatibility, and should always be + // ProtocolOverlayControlMessage for non-sleeve overlays. + ControlMessage(tag ProtocolTag, msg []byte) } type OverlayForwarderListener interface { @@ -98,5 +100,5 @@ func (NullOverlay) Forward(ForwardPacketKey) FlowOp { func (NullOverlay) Stop() { } -func (NullOverlay) ControlMessage([]byte) { +func (NullOverlay) ControlMessage(ProtocolTag, []byte) { } diff --git a/router/protocol.go b/router/protocol.go index cc7a6e81..f1959373 100644 --- a/router/protocol.go +++ b/router/protocol.go @@ -329,10 +329,13 @@ type ProtocolTag byte const ( ProtocolHeartbeat ProtocolTag = iota - ProtocolOverlayControlMsg + ProtocolConnectionEstablished + ProtocolFragmentationReceived + ProtocolPMTUVerified ProtocolGossip ProtocolGossipUnicast ProtocolGossipBroadcast + ProtocolOverlayControlMsg ) type ProtocolMsg struct { diff --git a/router/sleeve.go b/router/sleeve.go index 8ab2b119..36eb9cab 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -277,13 +277,14 @@ type sleeveForwarder struct { sleeve *SleeveOverlay remotePeer *Peer remotePeerBin []byte - sendControlMsg func([]byte) error + sendControlMsg func(ProtocolTag, []byte) error connUID uint64 // Channels to communicate with the aggregator goroutine aggregatorChan chan<- aggregatorFrame aggregatorDFChan chan<- aggregatorFrame specialChan chan<- specialFrame + controlMsgChan chan<- controlMessage confirmedChan chan<- struct{} finishedChan <-chan struct{} @@ -326,13 +327,18 @@ type aggregatorFrame struct { frame []byte } -// A "special" message over UDP, or a control message. The sender is -// nil for control messages. +// A "special" frame over UDP type specialFrame struct { sender *net.UDPAddr frame []byte } +// A control message +type controlMessage struct { + tag ProtocolTag + msg []byte +} + func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwarder, error) { var crypto OverlayCrypto if params.Crypto != nil { @@ -349,6 +355,7 @@ func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwa aggChan := make(chan aggregatorFrame, ChannelSize) aggDFChan := make(chan aggregatorFrame, ChannelSize) specialChan := make(chan specialFrame, 1) + controlMsgChan := make(chan controlMessage, 1) confirmedChan := make(chan struct{}) finishedChan := make(chan struct{}) @@ -361,6 +368,7 @@ func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwa aggregatorChan: aggChan, aggregatorDFChan: aggDFChan, specialChan: specialChan, + controlMsgChan: controlMsgChan, confirmedChan: confirmedChan, finishedChan: finishedChan, remoteAddr: params.RemoteAddr, @@ -372,8 +380,8 @@ func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwa senderDF: newUDPSenderDF(params.LocalIP, sleeve.localPort), } - go fwd.run(aggChan, aggDFChan, specialChan, confirmedChan, - finishedChan) + go fwd.run(aggChan, aggDFChan, specialChan, controlMsgChan, + confirmedChan, finishedChan) return fwd, nil } @@ -569,9 +577,9 @@ func frameTooBig(frame []byte, mtu int) bool { return len(frame) > mtu+EthernetOverhead } -func (fwd *sleeveForwarder) ControlMessage(msg []byte) { +func (fwd *sleeveForwarder) ControlMessage(tag ProtocolTag, msg []byte) { select { - case fwd.specialChan <- specialFrame{nil, msg}: + case fwd.controlMsgChan <- controlMessage{tag, msg}: case <-fwd.finishedChan: } } @@ -588,6 +596,7 @@ func (fwd *sleeveForwarder) Stop() { func (fwd *sleeveForwarder) run(aggChan <-chan aggregatorFrame, aggDFChan <-chan aggregatorFrame, specialChan <-chan specialFrame, + controlMsgChan <-chan controlMessage, confirmedChan <-chan struct{}, finishedChan chan<- struct{}) { defer close(finishedChan) @@ -605,14 +614,11 @@ loop: err = fwd.aggregateAndSend(frame, aggDFChan, fwd.crypto.EncDF, fwd.senderDF, fwd.maxPayload) - case special := <-specialChan: - if special.sender == nil { - // Control messages are sent on specialChan, - // with a nil sender - err = fwd.handleControlMsg(special.frame) - } else { - err = fwd.handleSpecialFrame(special) - } + case sf := <-specialChan: + err = fwd.handleSpecialFrame(sf) + + case cm := <-controlMsgChan: + err = fwd.handleControlMessage(cm) case _, ok := <-confirmedChan: if !ok { @@ -741,32 +747,20 @@ func (fwd *sleeveForwarder) handleSpecialFrame(special specialFrame) error { } } -const ( - HeartbeatAck = iota - FragTestAck - MTUTestAck -) - -func (fwd *sleeveForwarder) handleControlMsg(msg []byte) error { - if len(msg) == 0 { - log.Print(fwd.logPrefix(), - "Received zero-length control message") - return nil - } - - switch msg[0] { - case HeartbeatAck: +func (fwd *sleeveForwarder) handleControlMessage(cm controlMessage) error { + switch cm.tag { + case ProtocolConnectionEstablished: return fwd.handleHeartbeatAck() - case FragTestAck: + case ProtocolFragmentationReceived: return fwd.handleFragTestAck() - case MTUTestAck: - return fwd.handleMTUTestAck(msg) + case ProtocolPMTUVerified: + return fwd.handleMTUTestAck(cm.msg) default: log.Print(fwd.logPrefix(), - "Ignoring unknown control message: ", msg[0]) + "Ignoring unknown control message tag: ", cm.tag) return nil } } @@ -832,7 +826,7 @@ func (fwd *sleeveForwarder) handleHeartbeat(special specialFrame) error { if !fwd.ackedHeartbeat { fwd.ackedHeartbeat = true - if err := fwd.sendControlMsg([]byte{HeartbeatAck}); err != nil { + if err := fwd.sendControlMsg(ProtocolConnectionEstablished, nil); err != nil { return err } } @@ -857,6 +851,8 @@ func (fwd *sleeveForwarder) setRemoteAddr(addr *net.UDPAddr) { } func (fwd *sleeveForwarder) handleHeartbeatAck() error { + log.Debug(fwd.logPrefix(), "handleHeartbeatAck") + // The connection is now regarded as established fwd.notifyEstablished() @@ -899,7 +895,7 @@ func (fwd *sleeveForwarder) handleFragTest(frame []byte) error { return nil } - return fwd.sendControlMsg([]byte{FragTestAck}) + return fwd.sendControlMsg(ProtocolFragmentationReceived, nil) } func (fwd *sleeveForwarder) handleFragTestAck() error { @@ -943,19 +939,18 @@ func (fwd *sleeveForwarder) sendMTUTest() error { } func (fwd *sleeveForwarder) handleMTUTest(frame []byte) error { - buf := make([]byte, 3) - buf[0] = MTUTestAck - binary.BigEndian.PutUint16(buf[1:], uint16(len(frame)-EthernetOverhead)) - return fwd.sendControlMsg(buf) + buf := make([]byte, 2) + binary.BigEndian.PutUint16(buf, uint16(len(frame)-EthernetOverhead)) + return fwd.sendControlMsg(ProtocolPMTUVerified, buf) } func (fwd *sleeveForwarder) handleMTUTestAck(msg []byte) error { - if len(msg) < 3 { + if len(msg) < 2 { log.Print(fwd.logPrefix(), "Received truncated MTUTestAck") return nil } - mtu := int(binary.BigEndian.Uint16(msg[1:])) + mtu := int(binary.BigEndian.Uint16(msg)) log.Debug(fwd.logPrefix(), "handleMTUTestAck: for mtu candidate ", mtu) if mtu != fwd.mtuCandidate {