From d78f6614cc9b3c22372f326be7cf4f6b44e199d3 Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Fri, 6 Nov 2015 23:05:33 +0000 Subject: [PATCH] refactor: separate forwarder from overlay connection ...since the main connections only need to know about the former. We make LocalConnection.Overlay (formerly '.forwarder') public since the router needs access to it, and it shouldn't be doing that via a private field. --- router/connection.go | 37 +++++++++++++++--------------- router/fastdp.go | 2 +- router/overlay.go | 49 +++++++++++++++++++++++----------------- router/overlay_switch.go | 9 ++++---- router/router.go | 4 ++-- router/sleeve.go | 2 +- router/status.go | 2 +- 7 files changed, 56 insertions(+), 49 deletions(-) diff --git a/router/connection.go b/router/connection.go index 7a1f878f..a1877cfd 100644 --- a/router/connection.go +++ b/router/connection.go @@ -49,7 +49,7 @@ type LocalConnection struct { uid uint64 actionChan chan<- ConnectionAction finished <-chan struct{} // closed to signal that actorLoop has finished - forwarder OverlayForwarder + OverlayConn OverlayConnection } type ConnectionAction func() error @@ -190,7 +190,7 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch conn.Log("connection ready; using protocol version", conn.version) - params := ForwarderParams{ + params := OverlayConnectionParams{ RemotePeer: conn.remote, LocalAddr: conn.TCPConn.LocalAddr().(*net.TCPAddr), RemoteAddr: conn.TCPConn.RemoteAddr().(*net.TCPAddr), @@ -200,28 +200,27 @@ func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, finished ch SendControlMessage: conn.sendOverlayControlMessage, Features: intro.Features, } - if conn.forwarder, err = conn.Router.Overlay.MakeForwarder(params); err != nil { + if conn.OverlayConn, err = conn.Router.Overlay.PrepareConnection(params); err != nil { return } // As soon as we do AddConnection, the new connection becomes // visible to the packet routing logic. So AddConnection must - // come after MakeForwarder + // come after PrepareConnection if err = conn.Router.Ourself.AddConnection(conn); err != nil { return } conn.Router.ConnectionMaker.ConnectionCreated(conn) - // Forwarder confirmation comes after AddConnection, because - // only after that completes do we know the connection is - // valid: in particular that it is not a duplicate connection - // to the same peer. Sending heartbeats on a duplicate - // connection can trip up crypto at the other end, since the - // associated UDP packets may get decoded by the other - // connection. It is also generally wasteful to engage in any - // interaction with the remote on a connection that turns out - // to be invalid. - conn.forwarder.Confirm() + // OverlayConnection confirmation comes after AddConnection, + // because only after that completes do we know the connection is + // valid: in particular that it is not a duplicate connection to + // the same peer. Overlay communication on a duplicate connection + // can cause problems such as tripping up overlay crypto at the + // other end due to data being decoded by the other connection. It + // is also generally wasteful to engage in any interaction with + // the remote on a connection that turns out to be invalid. + conn.OverlayConn.Confirm() // receiveTCP must follow also AddConnection. In the absence // of any indirect connectivity to the remote peer, the first @@ -341,8 +340,8 @@ func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) er } func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction) (err error) { - fwdErrorChan := conn.forwarder.ErrorChannel() - fwdEstablishedChan := conn.forwarder.EstablishedChannel() + fwdErrorChan := conn.OverlayConn.ErrorChannel() + fwdEstablishedChan := conn.OverlayConn.EstablishedChannel() for err == nil { select { @@ -383,8 +382,8 @@ func (conn *LocalConnection) shutdown(err error) { conn.heartbeatTCP.Stop() } - if conn.forwarder != nil { - conn.forwarder.Stop() + if conn.OverlayConn != nil { + conn.OverlayConn.Stop() } conn.Router.ConnectionMaker.ConnectionTerminated(conn, err) @@ -428,7 +427,7 @@ func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) switch tag { case ProtocolHeartbeat: case ProtocolConnectionEstablished, ProtocolFragmentationReceived, ProtocolPMTUVerified, ProtocolOverlayControlMsg: - conn.forwarder.ControlMessage(byte(tag), payload) + conn.OverlayConn.ControlMessage(byte(tag), payload) case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip: return conn.Router.handleGossip(tag, payload) default: diff --git a/router/fastdp.go b/router/fastdp.go index 440a93b8..0c92bf9c 100644 --- a/router/fastdp.go +++ b/router/fastdp.go @@ -506,7 +506,7 @@ type fastDatapathForwarder struct { errorChan chan error } -func (fastdp fastDatapathOverlay) MakeForwarder(params ForwarderParams) (OverlayForwarder, error) { +func (fastdp fastDatapathOverlay) PrepareConnection(params OverlayConnectionParams) (OverlayConnection, error) { if params.SessionKey != nil { // No encryption suport in fastdp. The weaver main.go // is responsible for ensuring this doesn't happen. diff --git a/router/overlay.go b/router/overlay.go index e0e95364..a219e631 100644 --- a/router/overlay.go +++ b/router/overlay.go @@ -9,8 +9,9 @@ type Overlay interface { // Start consuming forwarded packets. StartConsumingPackets(*Peer, *Peers, OverlayConsumer) error - // Form a packet-forwarding connection. - MakeForwarder(ForwarderParams) (OverlayForwarder, error) + // Prepare on overlay connection. The connection should remain + // passive until it has been Confirm()ed. + PrepareConnection(OverlayConnectionParams) (OverlayConnection, error) // The routes have changed, so any cached information should // be discarded. @@ -26,12 +27,12 @@ type Overlay interface { Diagnostics() interface{} } -type ForwarderParams struct { +type OverlayConnectionParams struct { RemotePeer *Peer // The local address of the corresponding TCP connection. Used to // derive the local IP address for sending. May differ for - // different forwarders. + // different overlay connections. LocalAddr *net.TCPAddr // The remote address of the corresponding TCP connection. Used to @@ -50,7 +51,7 @@ type ForwarderParams struct { SessionKey *[32]byte // Function to send a control message to the counterpart - // forwarder. + // overlay connection. SendControlMessage func(tag byte, msg []byte) error // Features passed at connection initiation @@ -61,26 +62,22 @@ type ForwarderParams struct { // to decode the frame. type OverlayConsumer func(ForwardPacketKey) FlowOp -// All of the machinery to forward packets to a particular peer -type OverlayForwarder interface { - // Forward a packet across the connection. May be called as - // soon as the forwarder is created, in particular before - // Confirm(). The return value nil means the key could not be - // handled by this forwarder. - Forward(ForwardPacketKey) FlowOp - +// All of the machinery to manage overlay connectivity to a particular +// peer +type OverlayConnection interface { // Confirm that the connection is really wanted, and so the - // Overlay should begin heartbeats etc. to verify the - // operation of the forwarder. + // Overlay should begin heartbeats etc. to verify the operation of + // the overlay connection. Confirm() - // A channel indicating that the forwarder is established, - // i.e. its operation has been confirmed. + // A channel indicating that the overlay connection is + // established, i.e. its operation has been confirmed. EstablishedChannel() <-chan struct{} - // A channel indicating an error from the forwarder. The - // forwarder is not expected to be operational after the first - // error, so the channel only needs to buffer a single error. + // A channel indicating an error from the overlay connection. The + // overlay connection is not expected to be operational after the + // first error, so the channel only needs to buffer a single + // error. ErrorChannel() <-chan error Stop() @@ -94,13 +91,23 @@ type OverlayForwarder interface { DisplayName() string } +// All of the machinery to forward packets to a particular peer +type OverlayForwarder interface { + OverlayConnection + // Forward a packet across the connection. May be called as soon + // as the overlay connection is created, in particular before + // Confirm(). The return value nil means the key could not be + // handled by this forwarder. + Forward(ForwardPacketKey) FlowOp +} + type NullOverlay struct{} func (NullOverlay) StartConsumingPackets(*Peer, *Peers, OverlayConsumer) error { return nil } -func (NullOverlay) MakeForwarder(ForwarderParams) (OverlayForwarder, error) { +func (NullOverlay) PrepareConnection(OverlayConnectionParams) (OverlayConnection, error) { return NullOverlay{}, nil } diff --git a/router/overlay_switch.go b/router/overlay_switch.go index aa18d174..8cf67922 100644 --- a/router/overlay_switch.go +++ b/router/overlay_switch.go @@ -76,7 +76,7 @@ type namedOverlay struct { // Find the common set of overlays supported by both sides, with the // ordering being the same on both sides too. -func (osw *OverlaySwitch) commonOverlays(params ForwarderParams) ([]namedOverlay, error) { +func (osw *OverlaySwitch) commonOverlays(params OverlayConnectionParams) ([]namedOverlay, error) { var peerOverlays []string if overlaysFeature, present := params.Features["Overlays"]; present { peerOverlays = strings.Split(overlaysFeature, " ") @@ -160,9 +160,9 @@ type subForwarderEvent struct { err error } -func (osw *OverlaySwitch) MakeForwarder(params ForwarderParams) (OverlayForwarder, error) { +func (osw *OverlaySwitch) PrepareConnection(params OverlayConnectionParams) (OverlayConnection, error) { if _, present := params.Features["Overlays"]; !present && osw.compatOverlay != nil { - return osw.compatOverlay.MakeForwarder(params) + return osw.compatOverlay.PrepareConnection(params) } overlays, err := osw.commonOverlays(params) @@ -200,11 +200,12 @@ func (osw *OverlaySwitch) MakeForwarder(params ForwarderParams) (OverlayForwarde return origSendControlMessage(ProtocolOverlayControlMsg, xmsg) } - subFwd, err := overlay.MakeForwarder(params) + subConn, err := overlay.PrepareConnection(params) if err != nil { fwd.stopFrom(0) return nil, err } + subFwd := subConn.(OverlayForwarder) subStopChan := make(chan struct{}) go monitorForwarder(i, eventsChan, subStopChan, subFwd) diff --git a/router/router.go b/router/router.go index 6d70eb92..014a19b4 100644 --- a/router/router.go +++ b/router/router.go @@ -263,7 +263,7 @@ func (router *Router) relay(key ForwardPacketKey) FlowOp { return DiscardingFlowOp{} } - return conn.(*LocalConnection).forwarder.Forward(key) + return conn.(*LocalConnection).OverlayConn.(OverlayForwarder).Forward(key) } func (router *Router) relayBroadcast(srcPeer *Peer, key PacketKey) FlowOp { @@ -275,7 +275,7 @@ func (router *Router) relayBroadcast(srcPeer *Peer, key PacketKey) FlowOp { op := NewMultiFlowOp(true) for _, conn := range router.Ourself.ConnectionsTo(nextHops) { - op.Add(conn.(*LocalConnection).forwarder.Forward(ForwardPacketKey{ + op.Add(conn.(*LocalConnection).OverlayConn.(OverlayForwarder).Forward(ForwardPacketKey{ PacketKey: key, SrcPeer: srcPeer, DstPeer: conn.Remote()})) diff --git a/router/sleeve.go b/router/sleeve.go index cc6694ec..bf4102f9 100644 --- a/router/sleeve.go +++ b/router/sleeve.go @@ -349,7 +349,7 @@ type controlMessage struct { msg []byte } -func (sleeve *SleeveOverlay) MakeForwarder(params ForwarderParams) (OverlayForwarder, error) { +func (sleeve *SleeveOverlay) PrepareConnection(params OverlayConnectionParams) (OverlayConnection, error) { name := sleeve.localPeer.NameByte var crypto sleeveCrypto if params.SessionKey != nil { diff --git a/router/status.go b/router/status.go index 469f9329..64cbd6e0 100644 --- a/router/status.go +++ b/router/status.go @@ -176,7 +176,7 @@ func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus state = "established" } lc, _ := conn.(*LocalConnection) - info := fmt.Sprintf("%-6v %v", lc.forwarder.DisplayName(), conn.Remote()) + info := fmt.Sprintf("%-6v %v", lc.OverlayConn.DisplayName(), conn.Remote()) slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, info}) } for address, target := range cm.targets {