From 465915d7d4c3245808413d9f48ea9311953987ed Mon Sep 17 00:00:00 2001 From: Matthias Radestock Date: Thu, 25 Dec 2014 12:57:45 +0000 Subject: [PATCH] refactor: split out LocalPeer This has been a source of confusion for a long time. --- router/connection.go | 10 ++--- router/connection_maker.go | 4 +- router/peer.go | 79 +++++++++++++++++++------------------- router/peers.go | 4 +- router/router.go | 21 +++++----- router/types.go | 8 +++- 6 files changed, 64 insertions(+), 62 deletions(-) diff --git a/router/connection.go b/router/connection.go index f03d3b9c..56a048e4 100644 --- a/router/connection.go +++ b/router/connection.go @@ -52,7 +52,7 @@ func (conn *RemoteConnection) String() string { // Async. Does not return anything. If the connection is successful, // it will end up in the local peer's connections map. func NewLocalConnection(connRemote *RemoteConnection, acceptNewPeer bool, tcpConn *net.TCPConn, udpAddr *net.UDPAddr, router *Router) { - if connRemote.local != router.Ourself { + if connRemote.local != router.Ourself.Peer { log.Fatal("Attempt to create local connection from a peer which is not ourself") } @@ -165,7 +165,7 @@ func (conn *LocalConnection) queryLoop(queryChan <-chan *ConnectionInteraction, conn.handleShutdown() return } - conn.local.AddConnection(conn) + conn.Router.Ourself.AddConnection(conn) if conn.remoteUDPAddr != nil { if err = conn.ensureForwarders(); err == nil { conn.heartbeat = time.NewTicker(FastHeartbeat) @@ -231,7 +231,7 @@ func (conn *LocalConnection) handleSetEstablished() error { conn.established = true conn.Unlock() if !old { - conn.local.ConnectionEstablished(conn) + conn.Router.Ourself.ConnectionEstablished(conn) if err := conn.ensureForwarders(); err != nil { return err } @@ -280,7 +280,7 @@ func (conn *LocalConnection) handleShutdown() { if conn.remote != nil { conn.remote.DecrementLocalRefCount() - conn.local.DeleteConnection(conn) + conn.Router.Ourself.DeleteConnection(conn) } stopTicker(conn.heartbeat) @@ -507,7 +507,7 @@ func (conn *LocalConnection) receiveTCP(decoder *gob.Decoder, usingPassword bool } if len(newUpdate) != 0 { conn.Router.Routes.Recalculate() - conn.local.BroadcastTCP(Concat(ProtocolUpdateByte, newUpdate)) + conn.Router.Ourself.BroadcastTCP(Concat(ProtocolUpdateByte, newUpdate)) } } else if msg[0] == ProtocolPMTUVerified { conn.verifyPMTU <- int(binary.BigEndian.Uint16(msg[1:])) diff --git a/router/connection_maker.go b/router/connection_maker.go index 6733f95d..6767676f 100644 --- a/router/connection_maker.go +++ b/router/connection_maker.go @@ -22,7 +22,7 @@ const ( ) type ConnectionMaker struct { - ourself *Peer + ourself *LocalPeer peers *Peers targets map[string]*Target cmdLineAddress map[string]bool @@ -41,7 +41,7 @@ type ConnectionMakerInteraction struct { address string } -func StartConnectionMaker(ourself *Peer, peers *Peers) *ConnectionMaker { +func StartConnectionMaker(ourself *LocalPeer, peers *Peers) *ConnectionMaker { queryChan := make(chan *ConnectionMakerInteraction, ChannelSize) state := &ConnectionMaker{ ourself: ourself, diff --git a/router/peer.go b/router/peer.go index 079d76c7..3fe0ec66 100644 --- a/router/peer.go +++ b/router/peer.go @@ -20,6 +20,12 @@ func NewPeer(name PeerName, uid uint64, version uint64, router *Router) *Peer { Router: router} } +func (peer *Peer) String() string { + peer.RLock() + defer peer.RUnlock() + return fmt.Sprint("Peer ", peer.Name, " (v", peer.version, ") (UID ", peer.UID, ")") +} + func (peer *Peer) Version() uint64 { peer.RLock() defer peer.RUnlock() @@ -133,7 +139,7 @@ func (peer *Peer) Routes(stopAt *Peer, symmetric bool) (bool, map[PeerName]PeerN } func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) { - if peer == peer.Router.Ourself { + if peer == peer.Router.Ourself.Peer { log.Fatal("Attempt to set version and connections on the local peer", peer.Name) } peer.Lock() @@ -142,15 +148,23 @@ func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerN peer.connections = connections } -func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { - return peer.Relay(peer, dstPeer, df, frame, dec) +func StartLocalPeer(name PeerName, router *Router) *LocalPeer { + peer := &LocalPeer{Peer: NewPeer(name, 0, 0, router)} + queryChan := make(chan *PeerInteraction, ChannelSize) + peer.queryChan = queryChan + go peer.queryLoop(queryChan) + return peer } -func (peer *Peer) Broadcast(df bool, frame []byte, dec *EthernetDecoder) error { - return peer.RelayBroadcast(peer, df, frame, dec) +func (peer *LocalPeer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { + return peer.Relay(peer.Peer, dstPeer, df, frame, dec) } -func (peer *Peer) Relay(srcPeer, dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { +func (peer *LocalPeer) Broadcast(df bool, frame []byte, dec *EthernetDecoder) error { + return peer.RelayBroadcast(peer.Peer, df, frame, dec) +} + +func (peer *LocalPeer) Relay(srcPeer, dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { relayPeerName, found := peer.Router.Routes.Unicast(dstPeer.Name) if !found { // Not necessarily an error as there could be a race with the @@ -171,7 +185,7 @@ func (peer *Peer) Relay(srcPeer, dstPeer *Peer, df bool, frame []byte, dec *Ethe dec) } -func (peer *Peer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { +func (peer *LocalPeer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { nextHops := peer.Router.Routes.Broadcast(srcPeer.Name) if len(nextHops) == 0 { return nil @@ -203,22 +217,7 @@ func (peer *Peer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec *Ethe return nil } -func (peer *Peer) String() string { - peer.RLock() - defer peer.RUnlock() - return fmt.Sprint("Peer ", peer.Name, " (v", peer.version, ") (UID ", peer.UID, ")") -} - -func (peer *Peer) StartLocalPeer() { - if peer.Router.Ourself != peer { - log.Fatal("Attempt to start peer which is not the local peer") - } - queryChan := make(chan *PeerInteraction, ChannelSize) - peer.queryChan = queryChan - go peer.queryLoop(queryChan) -} - -func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error { +func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) error { if err := peer.checkConnectionLimit(); err != nil { return err } @@ -237,7 +236,7 @@ func (peer *Peer) CreateConnection(peerAddr string, acceptNewPeer bool) error { if err != nil { return err } - connRemote := NewRemoteConnection(peer, nil, tcpConn.RemoteAddr().String()) + connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String()) NewLocalConnection(connRemote, acceptNewPeer, tcpConn, udpAddr, peer.Router) return nil } @@ -253,14 +252,14 @@ const ( // Async: rely on the peer to shut us down if we shouldn't be adding // ourselves, so therefore this can be async -func (peer *Peer) AddConnection(conn *LocalConnection) { +func (peer *LocalPeer) AddConnection(conn *LocalConnection) { peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PAddConnection}, payload: conn} } // Sync. -func (peer *Peer) DeleteConnection(conn *LocalConnection) { +func (peer *LocalPeer) DeleteConnection(conn *LocalConnection) { resultChan := make(chan interface{}) peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PDeleteConnection, resultChan: resultChan}, @@ -269,14 +268,14 @@ func (peer *Peer) DeleteConnection(conn *LocalConnection) { } // Async. -func (peer *Peer) ConnectionEstablished(conn *LocalConnection) { +func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) { peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PConnectionEstablished}, payload: conn} } // Async. -func (peer *Peer) BroadcastTCP(msg []byte) { +func (peer *LocalPeer) BroadcastTCP(msg []byte) { peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PBroadcastTCP}, payload: msg} @@ -284,7 +283,7 @@ func (peer *Peer) BroadcastTCP(msg []byte) { // ACTOR server -func (peer *Peer) queryLoop(queryChan <-chan *PeerInteraction) { +func (peer *LocalPeer) queryLoop(queryChan <-chan *PeerInteraction) { for { query, ok := <-queryChan if !ok { @@ -304,8 +303,8 @@ func (peer *Peer) queryLoop(queryChan <-chan *PeerInteraction) { } } -func (peer *Peer) handleAddConnection(conn *LocalConnection) { - if peer != conn.Local() { +func (peer *LocalPeer) handleAddConnection(conn *LocalConnection) { + if peer.Peer != conn.Local() { log.Fatal("Attempt made to add connection to peer where peer is not the source of connection") } if conn.Remote() == nil { @@ -344,8 +343,8 @@ func (peer *Peer) handleAddConnection(conn *LocalConnection) { peer.Unlock() } -func (peer *Peer) handleDeleteConnection(conn *LocalConnection) { - if peer != conn.Local() { +func (peer *LocalPeer) handleDeleteConnection(conn *LocalConnection) { + if peer.Peer != conn.Local() { log.Fatal("Attempt made to delete connection from peer where peer is not the source of connection") } if conn.Remote() == nil { @@ -367,14 +366,14 @@ func (peer *Peer) handleDeleteConnection(conn *LocalConnection) { } // Must do garbage collection first to ensure we don't send out an // update with unreachable peers (can cause looping) - peer.Router.Peers.GarbageCollect(peer.Router.Ourself, peer.Router.Macs) + peer.Router.Peers.GarbageCollect(peer.Router.Ourself.Peer, peer.Router.Macs) if broadcast { peer.broadcastPeerUpdate() } } -func (peer *Peer) handleConnectionEstablished(conn *LocalConnection) { - if peer != conn.Local() { +func (peer *LocalPeer) handleConnectionEstablished(conn *LocalConnection) { + if peer.Peer != conn.Local() { log.Fatal("Peer informed of active connection where peer is not the source of connection") } if dupConn, found := peer.connections[conn.Remote().Name]; !found || conn != dupConn { @@ -388,18 +387,18 @@ func (peer *Peer) handleConnectionEstablished(conn *LocalConnection) { peer.broadcastPeerUpdate(conn.Remote()) } -func (peer *Peer) handleBroadcastTCP(msg []byte) { +func (peer *LocalPeer) handleBroadcastTCP(msg []byte) { peer.ForEachConnection(func(_ PeerName, conn Connection) { conn.(*LocalConnection).SendTCP(msg) }) } -func (peer *Peer) broadcastPeerUpdate(peers ...*Peer) { +func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { peer.Router.Routes.Recalculate() - peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer)...))) + peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer.Peer)...))) } -func (peer *Peer) checkConnectionLimit() error { +func (peer *LocalPeer) checkConnectionLimit() error { if 0 != peer.Router.ConnLimit && peer.ConnectionCount() >= peer.Router.ConnLimit { return fmt.Errorf("Connection limit reached (%v)", peer.Router.ConnLimit) } diff --git a/router/peers.go b/router/peers.go index 12c889a3..706c5164 100644 --- a/router/peers.go +++ b/router/peers.go @@ -68,9 +68,9 @@ func (peers *Peers) ApplyUpdate(update []byte, router *Router) ([]byte, error) { } // Now apply the updates - newUpdate := peers.applyUpdate(decodedUpdate, decodedConns, router.Ourself) + newUpdate := peers.applyUpdate(decodedUpdate, decodedConns, router.Ourself.Peer) - for _, peerRemoved := range peers.garbageCollect(router.Ourself, router.Macs) { + for _, peerRemoved := range peers.garbageCollect(router.Ourself.Peer, router.Macs) { delete(newUpdate, peerRemoved.Name) } diff --git a/router/router.go b/router/router.go index c968c270..19c395d6 100644 --- a/router/router.go +++ b/router/router.go @@ -34,9 +34,8 @@ func NewRouter(iface *net.Interface, name PeerName, password []byte, connLimit i if len(password) > 0 { router.Password = &password } - ourself := NewPeer(name, 0, 0, router) - router.Ourself = router.Peers.FetchWithDefault(ourself) - router.Ourself.StartLocalPeer() + router.Ourself = StartLocalPeer(name, router) + router.Peers.FetchWithDefault(router.Ourself.Peer) log.Println("Our name is", router.Ourself.Name) return router @@ -53,7 +52,7 @@ func (router *Router) Start() { po, err := NewPcapO(router.Iface.Name) checkFatal(err) router.ConnectionMaker = StartConnectionMaker(router.Ourself, router.Peers) - router.Routes = StartRoutes(router.Ourself, router.Peers) + router.Routes = StartRoutes(router.Ourself.Peer, router.Peers) router.UDPListener = router.listenUDP(Port, po) router.listenTCP(Port) router.sniff(pio) @@ -77,7 +76,7 @@ func (router *Router) sniff(pio PacketSourceSink) { injectFrame := func(frame []byte) error { return pio.WritePacket(frame) } checkFrameTooBig := func(err error) error { return dec.CheckFrameTooBig(err, injectFrame) } mac := router.Iface.HardwareAddr - if router.Macs.Enter(mac, router.Ourself) { + if router.Macs.Enter(mac, router.Ourself.Peer) { log.Println("Discovered our MAC", mac) } go func() { @@ -101,10 +100,10 @@ func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecode // We need to filter out frames we injected ourselves. For such // frames, the srcMAC will have been recorded as associated with a // different peer. - if found && srcPeer != router.Ourself { + if found && srcPeer != router.Ourself.Peer { return nil } - if router.Macs.Enter(srcMac, router.Ourself) { + if router.Macs.Enter(srcMac, router.Ourself.Peer) { log.Println("Discovered local MAC", srcMac) } if dec.DropFrame() { @@ -112,7 +111,7 @@ func (router *Router) handleCapturedPacket(frameData []byte, dec *EthernetDecode } dstMac := dec.eth.DstMAC dstPeer, found := router.Macs.Lookup(dstMac) - if found && dstPeer == router.Ourself { + if found && dstPeer == router.Ourself.Peer { return nil } df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0) @@ -157,7 +156,7 @@ func (router *Router) acceptTCP(tcpConn *net.TCPConn) { // someone else is dialing us, so our udp sender is the conn // on Port and we wait for them to send us something on UDP to // start. - connRemote := NewRemoteConnection(router.Ourself, nil, tcpConn.RemoteAddr().String()) + connRemote := NewRemoteConnection(router.Ourself.Peer, nil, tcpConn.RemoteAddr().String()) NewLocalConnection(connRemote, true, tcpConn, nil, router) } @@ -245,7 +244,7 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F df := decodedLen == 2 && (dec.ip.Flags&layers.IPv4DontFragment != 0) srcMac := dec.eth.SrcMAC - if dstPeer != router.Ourself { + if dstPeer != router.Ourself.Peer { // it's not for us, we're just relaying it if decodedLen == 0 { return nil @@ -288,7 +287,7 @@ func (router *Router) handleUDPPacketFunc(dec *EthernetDecoder, po PacketSink) F checkWarn(po.WritePacket(frame)) dstPeer, found = router.Macs.Lookup(dec.eth.DstMAC) - if !found || dstPeer != router.Ourself { + if !found || dstPeer != router.Ourself.Peer { return checkFrameTooBig(router.Ourself.RelayBroadcast(srcPeer, df, frame, dec), srcPeer) } diff --git a/router/types.go b/router/types.go index adc46358..5f7fbde2 100644 --- a/router/types.go +++ b/router/types.go @@ -12,7 +12,7 @@ import ( ) type Router struct { - Ourself *Peer + Ourself *LocalPeer Iface *net.Interface Macs *MacCache Peers *Peers @@ -34,7 +34,11 @@ type Peer struct { UID uint64 Router *Router localRefCount uint64 - queryChan chan<- *PeerInteraction +} + +type LocalPeer struct { + *Peer + queryChan chan<- *PeerInteraction } type PeerInteraction struct {