reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: split out LocalPeer

Browse Source 
This has been a source of confusion for a long time.
This commit is contained in:
Matthias Radestock committed 2014-12-25 12:57:45 +00:00
1 parent 059360416a
commit 465915d7d4
6 files changed
+64 -62
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

router/connection.go
+5 -5
View File
@@ -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:]))
router/connection_maker.go
+2 -2
View File
@@ -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,
router/peer.go
+39 -40
View File
@@ -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)
}
router/peers.go
+2 -2
View File
@@ -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)
}
router/router.go
+10 -11
View File
@@ -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)
}
router/types.go
+6 -2
View File
@@ -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 {