diff --git a/docs/architecture.txt b/docs/architecture.txt index 1461192a..4dba5c45 100644 --- a/docs/architecture.txt +++ b/docs/architecture.txt @@ -83,13 +83,17 @@ Router: local peer. It will tell the local connection (communicating to the actor thread) about the UDP address of the remote peer. The local connection will then start its forwarder threads as - described in 6a, and start sending slow heartbeats. The local connection - is marked as established which means it is included in network updates - broadcast to our peers. We also send to the remote peer via TCP a - ConnectionEstablished message. The remote peer receives this (on - the TCP receiver process), tells the connection actor process, - which then replaces the fast heartbeater with a slow heartbeater - and similarly marks the connection as established. + described in 6a, and start sending fast heartbeats. We send to the + remote peer via TCP a ConnectionEstablished message. The remote + peer receives this (on the TCP receiver process), tells the + connection actor process, which then replaces the fast heartbeater + with a slow heartbeater and marks the connection as established + (which means it is included in network updates broadcast to our + peers). + 6c. When the connection initiator receives the fast heartbeat from + the remote peer it sends to the remote peer via a TCP a + ConnectionEstablished message. This is handled by the remote peer + as described in 6b. 7. Whenever a connection is established or terminated, the local peer's version is incremented. Whenever this happens, the peer generates a network update message which is broadcast to its 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/peers.go b/router/peers.go index fd5d5782..071fb5e0 100644 --- a/router/peers.go +++ b/router/peers.go @@ -112,7 +112,11 @@ func (peers *Peers) String() string { peers.ForEach(func(name PeerName, peer *Peer) { buf.WriteString(fmt.Sprint(peer, "\n")) peer.ForEachConnection(func(remoteName PeerName, conn Connection) { - buf.WriteString(fmt.Sprintf(" -> %v [%v]\n", remoteName, conn.RemoteTCPAddr())) + established := "" + if !conn.Established() { + established = " (unestablished)" + } + buf.WriteString(fmt.Sprintf(" -> %v [%v%s]\n", remoteName, conn.RemoteTCPAddr(), established)) }) }) return buf.String() 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 {