diff --git a/router/connection.go b/router/connection.go index eb340726..7107e272 100644 --- a/router/connection.go +++ b/router/connection.go @@ -123,7 +123,7 @@ func (conn *LocalConnection) log(args ...interface{}) { const ( CSendTCP = iota CSetEstablished - CSetRemoteUDPAddr + CReceivedHeartbeat CShutdown ) @@ -135,12 +135,12 @@ func (conn *LocalConnection) Shutdown(err error) { } // Async -func (conn *LocalConnection) SetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) { +func (conn *LocalConnection) ReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) { if remoteUDPAddr == nil { return } conn.queryChan <- &ConnectionInteraction{ - Interaction: Interaction{code: CSetRemoteUDPAddr}, + Interaction: Interaction{code: CReceivedHeartbeat}, payload: remoteUDPAddr} } @@ -176,10 +176,7 @@ func (conn *LocalConnection) run(queryChan <-chan *ConnectionInteraction, accept conn.Router.Ourself.AddConnection(conn) if conn.remoteUDPAddr != nil { - if err := conn.ensureForwarders(); err == nil { - conn.heartbeat = time.NewTicker(FastHeartbeat) - conn.forwardHeartbeatFrame() // avoid initial wait - } else { + if err := conn.sendFastHeartbeats(); err != nil { conn.log("connection shutting down due to error:", err) return } @@ -207,10 +204,10 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) case CShutdown: err = query.payload.(error) terminate = true + case CReceivedHeartbeat: + err = conn.handleReceivedHeartbeat(query.payload.(*net.UDPAddr)) case CSetEstablished: err = conn.handleSetEstablished() - case CSetRemoteUDPAddr: - err = conn.handleSetRemoteUDPAddr(query.payload.(*net.UDPAddr)) case CSendTCP: err = conn.handleSendTCP(query.payload.([]byte)) } @@ -226,25 +223,36 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction) return } -func (conn *LocalConnection) handleSetRemoteUDPAddr(remoteUDPAddr *net.UDPAddr) error { +// Handlers +// +// NB: The conn.* fields are only written by the connection actor +// process, which is the caller of the handlers. Hence we do not need +// locks for reading, and only need write locks for fields read by +// other processes. + +func (conn *LocalConnection) handleReceivedHeartbeat(remoteUDPAddr *net.UDPAddr) error { + oldRemoteUDPAddr := conn.remoteUDPAddr + old := conn.receivedHeartbeat conn.Lock() - old := conn.remoteUDPAddr conn.remoteUDPAddr = remoteUDPAddr + conn.receivedHeartbeat = true conn.Unlock() - if old == nil { + if !old { if err := conn.handleSendTCP(ProtocolConnectionEstablishedByte); err != nil { return err } - return conn.handleSetEstablished() - } else if old.String() != remoteUDPAddr.String() { + } + if oldRemoteUDPAddr == nil { + return conn.sendFastHeartbeats() + } else if oldRemoteUDPAddr.String() != remoteUDPAddr.String() { log.Println("Peer", conn.remote.Name, "moved from", old, "to", remoteUDPAddr) } return nil } func (conn *LocalConnection) handleSetEstablished() error { - conn.Lock() old := conn.established + conn.Lock() conn.established = true conn.Unlock() if old { @@ -275,14 +283,6 @@ func (conn *LocalConnection) handleSendTCP(msg []byte) error { } func (conn *LocalConnection) handleShutdown() { - // Whilst some of these elements may have been written to whilst - // holding locks, they were only written to by the connection - // actor process. handleShutdown is only called by the connection - // actor process. So there is no need to take locks to read these - // (or write elements which are only read by the same - // process). Taking locks is only done for elements which are read - // by other processes. - if conn.TCPConn != nil { checkWarn(conn.TCPConn.Close()) } @@ -303,8 +303,10 @@ func (conn *LocalConnection) handleShutdown() { conn.Router.ConnectionMaker.ConnectionTerminated(conn.remoteTCPAddr) } +// Helpers + func (conn *LocalConnection) handshake(enc *gob.Encoder, dec *gob.Decoder, acceptNewPeer bool) error { - // We do not need to worry about locking in here as at this point, + // We do not need to worry about locking in here as at this point // the connection is not reachable by any go-routine other than // ourself. Only when we add this connection to the conn.local // peer will it be visible from multiple go-routines. @@ -526,6 +528,16 @@ func (conn *LocalConnection) extendReadDeadline() { conn.TCPConn.SetReadDeadline(time.Now().Add(ReadTimeout)) } + +func (conn *LocalConnection) sendFastHeartbeats() error { + err := conn.ensureForwarders() + if err == nil { + conn.heartbeat = time.NewTicker(FastHeartbeat) + conn.forwardHeartbeatFrame() // avoid initial wait + } + return err +} + // Heartbeating serves two purposes: a) keeping NAT paths alive, and // b) updating a remote peer's knowledge of our address, in the event // it changes (e.g. because NAT paths expired). diff --git a/router/consts.go b/router/consts.go index 1b94a73f..5bdd14f8 100644 --- a/router/consts.go +++ b/router/consts.go @@ -7,7 +7,7 @@ import ( const ( Protocol = "weave" - ProtocolVersion = 8 + ProtocolVersion = 9 EthernetOverhead = 14 UDPOverhead = 28 // 20 bytes for IPv4, 8 bytes for UDP Port = 6783 diff --git a/router/router.go b/router/router.go index a8b833d3..7e05ec04 100644 --- a/router/router.go +++ b/router/router.go @@ -265,7 +265,7 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F if relayConn.Remote().Name == srcPeer.Name { if frameLen == 0 { - relayConn.SetRemoteUDPAddr(sender) + relayConn.ReceivedHeartbeat(sender) return nil } else if frameLen == FragTestSize && bytes.Equal(frame, FragTest) { relayConn.SendTCP(ProtocolFragmentationReceivedByte) diff --git a/router/types.go b/router/types.go index 850767ac..d32ccaf5 100644 --- a/router/types.go +++ b/router/types.go @@ -42,25 +42,26 @@ type RemoteConnection struct { type LocalConnection struct { sync.RWMutex RemoteConnection - TCPConn *net.TCPConn - tcpSender TCPSender - remoteUDPAddr *net.UDPAddr - established bool - stackFrag bool - effectivePMTU int - SessionKey *[32]byte - heartbeat *time.Ticker - fetchAll *time.Ticker - fragTest *time.Ticker - forwardChan chan<- *ForwardedFrame - forwardChanDF chan<- *ForwardedFrame - stopForward chan<- interface{} - stopForwardDF chan<- interface{} - verifyPMTU chan<- int - Decryptor Decryptor - Router *Router - UID uint64 - queryChan chan<- *ConnectionInteraction + TCPConn *net.TCPConn + tcpSender TCPSender + remoteUDPAddr *net.UDPAddr + established bool + receivedHeartbeat bool + stackFrag bool + effectivePMTU int + SessionKey *[32]byte + heartbeat *time.Ticker + fetchAll *time.Ticker + fragTest *time.Ticker + forwardChan chan<- *ForwardedFrame + forwardChanDF chan<- *ForwardedFrame + stopForward chan<- interface{} + stopForwardDF chan<- interface{} + verifyPMTU chan<- int + Decryptor Decryptor + Router *Router + UID uint64 + queryChan chan<- *ConnectionInteraction } type ConnectionInteraction struct {