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.
This commit is contained in:
Matthias Radestock committed 2015-11-09 15:45:06 +00:00
1 parent 97622b961c
commit d78f6614cc
7 files changed
+56 -49

No files matched your search

+18 -19
View File
@@ -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:
+1 -1
View File
@@ -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.
+28 -21
View File
@@ -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
}
+5 -4
View File
@@ -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)
+2 -2
View File
@@ -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()}))
+1 -1
View File
@@ -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 {
+1 -1
View File
@@ -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 {