package weave import ( "fmt" "log" "net" "sort" ) func NewPeer(name PeerName, uid uint64, version uint64, router *Router) *Peer { if uid == 0 { uid = randUint64() } return &Peer{ Name: name, NameByte: name.Bin(), connections: make(map[PeerName]Connection), version: version, UID: uid, Router: router} } func (peer *Peer) Version() uint64 { peer.RLock() defer peer.RUnlock() return peer.version } func (peer *Peer) IncrementLocalRefCount() { peer.Lock() defer peer.Unlock() peer.localRefCount += 1 } func (peer *Peer) DecrementLocalRefCount() { peer.Lock() defer peer.Unlock() peer.localRefCount -= 1 } func (peer *Peer) IsLocallyReferenced() bool { peer.RLock() defer peer.RUnlock() return peer.localRefCount != 0 } func (peer *Peer) ConnectionCount() int { peer.RLock() defer peer.RUnlock() return len(peer.connections) } func (peer *Peer) ConnectionTo(name PeerName) (Connection, bool) { peer.RLock() defer peer.RUnlock() conn, found := peer.connections[name] return conn, found // yes, you really can't inline that. FFS. } func (peer *Peer) ForEachConnection(fun func(PeerName, Connection)) { peer.RLock() defer peer.RUnlock() for name, conn := range peer.connections { fun(name, conn) } } func (peer *Peer) HasPathTo(target *Peer, symmetric bool) (bool, map[PeerName]bool) { peer.RLock() defer peer.RUnlock() reached := make(map[PeerName]bool) nextWorklist := []*Peer{peer} reached[peer.Name] = true for len(nextWorklist) > 0 { worklist := nextWorklist sort.Sort(ListOfPeers(worklist)) nextWorklist = []*Peer{} for _, curPeer := range worklist { if curPeer == target { return true, reached } for remoteName, conn := range curPeer.connections { if _, found := reached[remoteName]; found { continue } // By taking this lock, we ensure no changes can occur // to any peer we have reached. As a result, if/when // we reach the target, we know the entire path from // peer to target is valid - no changes can have been // made to that path during our calculation. remote := conn.Remote() remote.RLock() defer remote.RUnlock() if _, found := remote.connections[curPeer.Name]; !symmetric || found { nextWorklist = append(nextWorklist, remote) reached[remoteName] = true } } } } return false, reached } func (peer *Peer) SetVersionAndConnections(version uint64, connections map[PeerName]Connection) { if peer == peer.Router.Ourself { log.Fatal("Attempt to set version and connections on the local peer", peer.Name) } peer.Lock() defer peer.Unlock() peer.version = version peer.connections = connections for remoteName, conn := range connections { peer.Router.ConnectionMaker.EnsureConnection(remoteName, conn.RemoteTCPAddr()) } } func (peer *Peer) Forward(dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { return peer.Relay(peer, dstPeer, df, frame, dec) } func (peer *Peer) Broadcast(df bool, frame []byte, dec *EthernetDecoder) error { return peer.RelayBroadcast(peer, df, frame, dec) } func (peer *Peer) Relay(srcPeer, dstPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { relayPeerName, found := peer.Router.Topology.Unicast(dstPeer.Name) if !found { // Not necessarily an error as there could be a race with the // dst disappearing whilst the frame is in flight log.Println("Received packet for unknown destination:", dstPeer.Name) return nil } conn, found := peer.ConnectionTo(relayPeerName) if !found { // Again, could just be a race, not necessarily an error log.Println("Unable to find connection to relay peer", relayPeerName) return nil } return conn.(*LocalConnection).Forward(df, &ForwardedFrame{ srcPeer: srcPeer, dstPeer: dstPeer, frame: frame}, dec) } func (peer *Peer) RelayBroadcast(srcPeer *Peer, df bool, frame []byte, dec *EthernetDecoder) error { nextHops := peer.Router.Topology.Broadcast(srcPeer.Name) if len(nextHops) == 0 { return nil } // We must not hold a read lock on peer during the conn.Forward // below, since that is a potentially blocking operation (e.g. if // the channel is full). nextConns := make([]*LocalConnection, 0, len(nextHops)) peer.RLock() for _, hopName := range nextHops { conn, found := peer.connections[hopName] // Again, !found could just be due to a race. if found { nextConns = append(nextConns, conn.(*LocalConnection)) } } peer.RUnlock() var err error for _, conn := range nextConns { err = conn.Forward(df, &ForwardedFrame{ srcPeer: srcPeer, dstPeer: conn.Remote(), frame: frame}, dec) if err != nil { return err } } 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, expectedName PeerName) error { if err := peer.checkConnectionLimit(); err != nil { return err } // We're dialing the remote so that means connections will come from random ports _, _, err := net.SplitHostPort(peerAddr) var addrStr string if err == nil { addrStr = peerAddr } else { addrStr = fmt.Sprintf("%s:%d", peerAddr, Port) } tcpAddr, tcpErr := net.ResolveTCPAddr("tcp4", addrStr) udpAddr, udpErr := net.ResolveUDPAddr("udp4", addrStr) if tcpErr != nil || udpErr != nil { // they really should have the same value, but just in case... if tcpErr == nil { return udpErr } return tcpErr } tcpConn, err := net.DialTCP("tcp4", nil, tcpAddr) if err != nil { return err } connRemote := NewRemoteConnection(peer, nil, tcpConn.RemoteAddr().String()) NewLocalConnection(connRemote, expectedName, tcpConn, udpAddr, peer.Router) return nil } // ACTOR client API const ( PAddConnection = iota PBroadcastTCP = iota PDeleteConnection = iota PConnectionEstablished = iota ) // 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) { peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PAddConnection}, payload: conn} } // Sync. func (peer *Peer) DeleteConnection(conn *LocalConnection) { resultChan := make(chan interface{}) peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PDeleteConnection, resultChan: resultChan}, payload: conn} <-resultChan } // Async. func (peer *Peer) ConnectionEstablished(conn *LocalConnection) { peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PConnectionEstablished}, payload: conn} } // Async. func (peer *Peer) BroadcastTCP(msg []byte) { peer.queryChan <- &PeerInteraction{ Interaction: Interaction{code: PBroadcastTCP}, payload: msg} } // ACTOR server func (peer *Peer) queryLoop(queryChan <-chan *PeerInteraction) { for { query, ok := <-queryChan if !ok { return } switch { case query.code == PAddConnection: peer.handleAddConnection(query.payload.(*LocalConnection)) case query.code == PDeleteConnection: peer.handleDeleteConnection(query.payload.(*LocalConnection)) query.resultChan <- nil case query.code == PConnectionEstablished: peer.handleConnectionEstablished(query.payload.(*LocalConnection)) case query.code == PBroadcastTCP: peer.handleBroadcastTCP(query.payload.([]byte)) } } } func (peer *Peer) handleAddConnection(conn *LocalConnection) { if peer != conn.Local() { log.Fatal("Attempt made to add connection to peer where peer is not the source of connection") } if conn.Remote() == nil { log.Fatal("Attempt made to add connection to peer with unknown remote peer") } toName := conn.Remote().Name dupErr := fmt.Errorf("Multiple connections to %s added to %s", toName, peer.Name) // deliberately non symmetrical if dupConn, found := peer.connections[toName]; found { if dupConn == conn { return } else { // conn.UID is used as the tie breaker here, in the // knowledge that both sides will make the same decision. dupConnLocal := dupConn.(*LocalConnection) if conn.UID == dupConnLocal.UID { // oh good grief. Sod it, just kill both of them. conn.CheckFatal(dupErr) dupConnLocal.CheckFatal(dupErr) peer.handleDeleteConnection(dupConnLocal) return } else if conn.UID < dupConnLocal.UID { dupConnLocal.CheckFatal(dupErr) peer.handleDeleteConnection(dupConnLocal) } else { conn.CheckFatal(dupErr) return } } } if err := peer.checkConnectionLimit(); err != nil { conn.CheckFatal(err) return } peer.Lock() peer.connections[toName] = conn peer.Unlock() } func (peer *Peer) handleDeleteConnection(conn *LocalConnection) { if peer != conn.Local() { log.Fatal("Attempt made to delete connection from peer where peer is not the source of connection") } if conn.Remote() == nil { log.Fatal("Attempt made to delete connection to peer with unknown remote peer") } toName := conn.Remote().Name if connFound, found := peer.connections[toName]; !found || connFound != conn { return } peer.Lock() delete(peer.connections, toName) peer.Unlock() broadcast := false if conn.Established() { peer.Lock() peer.version += 1 peer.Unlock() broadcast = true } // 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) if broadcast { peer.broadcastPeerUpdate() } } func (peer *Peer) handleConnectionEstablished(conn *LocalConnection) { if 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 { conn.CheckFatal(fmt.Errorf("Cannot set unknown connection active")) return } peer.Lock() peer.version += 1 peer.Unlock() log.Println("Peer", peer.Name, "established active connection to remote peer", conn.Remote().Name, "at", conn.RemoteTCPAddr()) peer.broadcastPeerUpdate(conn.Remote()) } func (peer *Peer) handleBroadcastTCP(msg []byte) { peer.Router.Topology.RebuildRoutes() peer.ForEachConnection(func(_ PeerName, conn Connection) { conn.(*LocalConnection).SendTCP(msg) }) } func (peer *Peer) broadcastPeerUpdate(peers ...*Peer) { peer.handleBroadcastTCP(Concat(ProtocolUpdateByte, EncodePeers(append(peers, peer)...))) } func (peer *Peer) checkConnectionLimit() error { if 0 != peer.Router.ConnLimit && peer.ConnectionCount() >= peer.Router.ConnLimit { return fmt.Errorf("Connection limit reached (%v)", peer.Router.ConnLimit) } return nil }