diff --git a/mesh/connection.go b/mesh/connection.go deleted file mode 100644 index 0adfc793..00000000 --- a/mesh/connection.go +++ /dev/null @@ -1,473 +0,0 @@ -package mesh - -import ( - "fmt" - "net" - "strconv" - "sync" - "time" -) - -type Connection interface { - Local() *Peer - Remote() *Peer - RemoteTCPAddr() string - Outbound() bool - Established() bool - BreakTie(Connection) ConnectionTieBreak - Shutdown(error) - Log(args ...interface{}) -} - -type ConnectionTieBreak int - -const ( - TieBreakWon ConnectionTieBreak = iota - TieBreakLost - TieBreakTied -) - -var ErrConnectToSelf = fmt.Errorf("Cannot connect to ourself") - -type RemoteConnection struct { - local *Peer - remote *Peer - remoteTCPAddr string - outbound bool - established bool -} - -type LocalConnection struct { - sync.RWMutex - RemoteConnection - TCPConn *net.TCPConn - TrustRemote bool // is remote on a trusted subnet? - TrustedByRemote bool // does remote trust us? - version byte - tcpSender TCPSender - SessionKey *[32]byte - heartbeatTCP *time.Ticker - Router *Router - uid uint64 - actionChan chan<- ConnectionAction - errorChan chan<- error - finished <-chan struct{} // closed to signal that actorLoop has finished - OverlayConn OverlayConnection - gossipSenders *GossipSenders -} - -type GossipConnection interface { - GossipSenders() *GossipSenders -} - -type ConnectionAction func() error - -func NewRemoteConnection(from, to *Peer, tcpAddr string, outbound bool, established bool) *RemoteConnection { - return &RemoteConnection{ - local: from, - remote: to, - remoteTCPAddr: tcpAddr, - outbound: outbound, - established: established, - } -} - -func (conn *RemoteConnection) Local() *Peer { return conn.local } -func (conn *RemoteConnection) Remote() *Peer { return conn.remote } -func (conn *RemoteConnection) RemoteTCPAddr() string { return conn.remoteTCPAddr } -func (conn *RemoteConnection) Outbound() bool { return conn.outbound } -func (conn *RemoteConnection) Established() bool { return conn.established } -func (conn *RemoteConnection) BreakTie(Connection) ConnectionTieBreak { return TieBreakTied } -func (conn *RemoteConnection) Shutdown(error) {} - -func (conn *RemoteConnection) Log(args ...interface{}) { - log.Println(append(append([]interface{}{}, fmt.Sprintf("->[%s|%s]:", conn.remoteTCPAddr, conn.remote)), args...)...) -} - -func (conn *RemoteConnection) ErrorLog(args ...interface{}) { - log.Errorln(append(append([]interface{}{}, fmt.Sprintf("->[%s|%s]:", conn.remoteTCPAddr, conn.remote)), args...)...) -} - -// Does not return anything. If the connection is successful, it will -// end up in the local peer's connections map. -func StartLocalConnection(connRemote *RemoteConnection, tcpConn *net.TCPConn, router *Router, acceptNewPeer bool) { - if connRemote.local != router.Ourself.Peer { - log.Fatal("Attempt to create local connection from a peer which is not ourself") - } - actionChan := make(chan ConnectionAction, ChannelSize) - errorChan := make(chan error, 1) - finished := make(chan struct{}) - conn := &LocalConnection{ - RemoteConnection: *connRemote, // NB, we're taking a copy of connRemote here. - Router: router, - TCPConn: tcpConn, - TrustRemote: router.Trusts(connRemote), - uid: randUint64(), - actionChan: actionChan, - errorChan: errorChan, - finished: finished} - conn.gossipSenders = NewGossipSenders(conn, finished) - go conn.run(actionChan, errorChan, finished, acceptNewPeer) -} - -func (conn *LocalConnection) BreakTie(dupConn Connection) ConnectionTieBreak { - dupConnLocal := dupConn.(*LocalConnection) - // conn.uid is used as the tie breaker here, in the knowledge that - // both sides will make the same decision. - if conn.uid < dupConnLocal.uid { - return TieBreakWon - } else if dupConnLocal.uid < conn.uid { - return TieBreakLost - } else { - return TieBreakTied - } -} - -func (conn *LocalConnection) Established() bool { - return conn.established -} - -func (conn *LocalConnection) SendProtocolMsg(m ProtocolMsg) error { - if err := conn.sendProtocolMsg(m); err != nil { - conn.Shutdown(err) - return err - } - return nil -} - -func (conn *LocalConnection) GossipSenders() *GossipSenders { - return conn.gossipSenders -} - -// ACTOR methods - -// NB: The conn.* fields are only written by the connection actor -// process, which is the caller of the ConnectionAction funs. Hence we -// do not need locks for reading, and only need write locks for fields -// read by other processes. - -// Non-blocking. -func (conn *LocalConnection) Shutdown(err error) { - // err should always be a real error, even if only io.EOF - if err == nil { - panic("nil error") - } - - select { - case conn.errorChan <- err: - default: - } -} - -// Send an actor request to the actorLoop, but don't block if -// actorLoop has exited - see http://blog.golang.org/pipelines for -// pattern -func (conn *LocalConnection) sendAction(action ConnectionAction) { - select { - case conn.actionChan <- action: - case <-conn.finished: - } -} - -// ACTOR server - -func (conn *LocalConnection) run(actionChan <-chan ConnectionAction, errorChan <-chan error, finished chan<- struct{}, acceptNewPeer bool) { - var err error // important to use this var and not create another one with 'err :=' - defer func() { conn.shutdown(err) }() - defer close(finished) - - conn.TCPConn.SetLinger(0) - intro, err := ProtocolIntroParams{ - MinVersion: conn.Router.ProtocolMinVersion, - MaxVersion: ProtocolMaxVersion, - Features: conn.makeFeatures(), - Conn: conn.TCPConn, - Password: conn.Router.Password, - Outbound: conn.outbound, - }.DoIntro() - if err != nil { - return - } - - conn.SessionKey = intro.SessionKey - conn.tcpSender = intro.Sender - conn.version = intro.Version - - remote, err := conn.parseFeatures(intro.Features) - if err != nil { - return - } - - if err = conn.registerRemote(remote, acceptNewPeer); err != nil { - return - } - - conn.Log("connection ready; using protocol version", conn.version) - - // only use negotiated session key for untrusted connections - var sessionKey *[32]byte - if conn.Untrusted() { - sessionKey = conn.SessionKey - } - - params := OverlayConnectionParams{ - RemotePeer: conn.remote, - LocalAddr: conn.TCPConn.LocalAddr().(*net.TCPAddr), - RemoteAddr: conn.TCPConn.RemoteAddr().(*net.TCPAddr), - Outbound: conn.outbound, - ConnUID: conn.uid, - SessionKey: sessionKey, - SendControlMessage: conn.sendOverlayControlMessage, - Features: intro.Features, - } - if conn.OverlayConn, err = conn.Router.Overlay.PrepareConnection(params); err != nil { - return - } - - // As soon as we do AddConnection, the new connection becomes - // visible to the packet routing logic. So AddConnection must - // come after PrepareConnection - if err = conn.Router.Ourself.AddConnection(conn); err != nil { - return - } - conn.Router.ConnectionMaker.ConnectionCreated(conn) - - // OverlayConnection confirmation comes after AddConnection, - // because only after that completes do we know the connection is - // valid: in particular that it is not a duplicate connection to - // the same peer. Overlay communication on a duplicate connection - // can cause problems such as tripping up overlay crypto at the - // other end due to data being decoded by the other connection. It - // is also generally wasteful to engage in any interaction with - // the remote on a connection that turns out to be invalid. - conn.OverlayConn.Confirm() - - // receiveTCP must follow also AddConnection. In the absence - // of any indirect connectivity to the remote peer, the first - // we hear about it (and any peers reachable from it) is - // through topology gossip it sends us on the connection. We - // must ensure that the connection has been added to Ourself - // prior to processing any such gossip, otherwise we risk - // immediately gc'ing part of that newly received portion of - // the topology (though not the remote peer itself, since that - // will have a positive ref count), leaving behind dangling - // references to peers. Hence we must invoke AddConnection, - // which is *synchronous*, first. - conn.heartbeatTCP = time.NewTicker(TCPHeartbeat) - go conn.receiveTCP(intro.Receiver) - - // AddConnection must precede actorLoop. More precisely, it - // must precede shutdown, since that invokes DeleteConnection - // and is invoked on termination of this entire - // function. Essentially this boils down to a prohibition on - // running AddConnection in a separate goroutine, at least not - // without some synchronisation. Which in turn requires the - // launching of the receiveTCP goroutine to precede actorLoop. - err = conn.actorLoop(actionChan, errorChan) -} - -func (conn *LocalConnection) makeFeatures() map[string]string { - features := map[string]string{ - "PeerNameFlavour": PeerNameFlavour, - "Name": conn.local.Name.String(), - "NickName": conn.local.NickName, - "ShortID": fmt.Sprint(conn.local.ShortID), - "UID": fmt.Sprint(conn.local.UID), - "ConnID": fmt.Sprint(conn.uid), - "Trusted": fmt.Sprint(conn.TrustRemote), - } - conn.Router.Overlay.AddFeaturesTo(features) - return features -} - -type features map[string]string - -func (features features) MustHave(keys []string) error { - for _, key := range keys { - if _, ok := features[key]; !ok { - return fmt.Errorf("Field %s is missing", key) - } - } - return nil -} - -func (features features) Get(key string) string { - return features[key] -} - -func (conn *LocalConnection) parseFeatures(features features) (*Peer, error) { - if err := features.MustHave([]string{"PeerNameFlavour", "Name", "NickName", "UID", "ConnID"}); err != nil { - return nil, err - } - - remotePeerNameFlavour := features.Get("PeerNameFlavour") - if remotePeerNameFlavour != PeerNameFlavour { - return nil, fmt.Errorf("Peer name flavour mismatch (ours: '%s', theirs: '%s')", PeerNameFlavour, remotePeerNameFlavour) - } - - name, err := PeerNameFromString(features.Get("Name")) - if err != nil { - return nil, err - } - - nickName := features.Get("NickName") - - var shortID uint64 - var hasShortID bool - if shortIDStr, present := features["ShortID"]; present { - hasShortID = true - shortID, err = strconv.ParseUint(shortIDStr, 10, PeerShortIDBits) - if err != nil { - return nil, err - } - } - - var trusted bool - if trustedStr, present := features["Trusted"]; present { - trusted, err = strconv.ParseBool(trustedStr) - if err != nil { - return nil, err - } - } - conn.TrustedByRemote = trusted - - uid, err := ParsePeerUID(features.Get("UID")) - if err != nil { - return nil, err - } - - remoteConnID, err := strconv.ParseUint(features.Get("ConnID"), 10, 64) - if err != nil { - return nil, err - } - - conn.uid ^= remoteConnID - peer := NewPeer(name, nickName, uid, 0, PeerShortID(shortID)) - peer.HasShortID = hasShortID - return peer, nil -} - -func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) error { - if acceptNewPeer { - conn.remote = conn.Router.Peers.FetchWithDefault(remote) - } else { - conn.remote = conn.Router.Peers.FetchAndAddRef(remote.Name) - if conn.remote == nil { - return fmt.Errorf("Found unknown remote name: %s at %s", remote.Name, conn.remoteTCPAddr) - } - } - - if conn.remote == conn.local { - return ErrConnectToSelf - } - - return nil -} - -func (conn *LocalConnection) actorLoop(actionChan <-chan ConnectionAction, errorChan <-chan error) (err error) { - fwdErrorChan := conn.OverlayConn.ErrorChannel() - fwdEstablishedChan := conn.OverlayConn.EstablishedChannel() - - for err == nil { - select { - case err = <-errorChan: - case err = <-fwdErrorChan: - default: - select { - case action := <-actionChan: - err = action() - case <-conn.heartbeatTCP.C: - err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat) - case <-fwdEstablishedChan: - conn.established = true - fwdEstablishedChan = nil - conn.Router.Ourself.ConnectionEstablished(conn) - case err = <-errorChan: - case err = <-fwdErrorChan: - } - } - } - return -} - -func (conn *LocalConnection) shutdown(err error) { - if conn.remote == nil { - log.Errorf("->[%s] connection shutting down due to error during handshake: %v", conn.remoteTCPAddr, err) - } else { - conn.ErrorLog("connection shutting down due to error:", err) - } - - if conn.TCPConn != nil { - checkWarn(conn.TCPConn.Close()) - } - - if conn.remote != nil { - conn.Router.Peers.Dereference(conn.remote) - conn.Router.Ourself.DeleteConnection(conn) - } - - if conn.heartbeatTCP != nil { - conn.heartbeatTCP.Stop() - } - - if conn.OverlayConn != nil { - conn.OverlayConn.Stop() - } - - conn.Router.ConnectionMaker.ConnectionTerminated(conn, err) -} - -func (conn *LocalConnection) sendOverlayControlMessage(tag byte, msg []byte) error { - return conn.sendProtocolMsg(ProtocolMsg{ProtocolTag(tag), msg}) -} - -// Helpers - -func (conn *LocalConnection) sendSimpleProtocolMsg(tag ProtocolTag) error { - return conn.sendProtocolMsg(ProtocolMsg{tag: tag}) -} - -func (conn *LocalConnection) sendProtocolMsg(m ProtocolMsg) error { - return conn.tcpSender.Send(append([]byte{byte(m.tag)}, m.msg...)) -} - -func (conn *LocalConnection) receiveTCP(receiver TCPReceiver) { - var err error - for { - conn.extendReadDeadline() - - var msg []byte - if msg, err = receiver.Receive(); err != nil { - break - } - if len(msg) < 1 { - conn.Log("ignoring blank msg") - continue - } - if err = conn.handleProtocolMsg(ProtocolTag(msg[0]), msg[1:]); err != nil { - break - } - } - conn.Shutdown(err) -} - -func (conn *LocalConnection) handleProtocolMsg(tag ProtocolTag, payload []byte) error { - switch tag { - case ProtocolHeartbeat: - case ProtocolReserved1, ProtocolReserved2, ProtocolReserved3, ProtocolOverlayControlMsg: - conn.OverlayConn.ControlMessage(byte(tag), payload) - case ProtocolGossipUnicast, ProtocolGossipBroadcast, ProtocolGossip: - return conn.Router.handleGossip(tag, payload) - default: - conn.Log("ignoring unknown protocol tag:", tag) - } - return nil -} - -func (conn *LocalConnection) extendReadDeadline() { - conn.TCPConn.SetReadDeadline(time.Now().Add(TCPHeartbeat * 2)) -} - -func (conn *LocalConnection) Untrusted() bool { - return !conn.TrustRemote || !conn.TrustedByRemote -} diff --git a/mesh/connection_maker.go b/mesh/connection_maker.go deleted file mode 100644 index d38f94c4..00000000 --- a/mesh/connection_maker.go +++ /dev/null @@ -1,317 +0,0 @@ -package mesh - -import ( - "fmt" - "math/rand" - "net" - "time" -) - -const ( - InitialInterval = 2 * time.Second - MaxInterval = 6 * time.Minute - ResetAfter = 1 * time.Minute -) - -type peerAddrs map[string]*net.TCPAddr - -type ConnectionMaker struct { - ourself *LocalPeer - peers *Peers - port int - discovery bool - targets map[string]*Target - connections map[Connection]struct{} - directPeers peerAddrs - actionChan chan<- ConnectionMakerAction -} - -type TargetState int - -const ( - TargetWaiting TargetState = iota // we are waiting to connect there - TargetAttempting // we are attempting to connect there - TargetConnected // we are connected to there -) - -// Information about an address where we may find a peer -type Target struct { - state TargetState - lastError error // reason for disconnection last time - tryAfter time.Time // next time to try this address - tryInterval time.Duration // retry delay on next failure -} - -type ConnectionMakerAction func() bool - -func NewConnectionMaker(ourself *LocalPeer, peers *Peers, port int, discovery bool) *ConnectionMaker { - actionChan := make(chan ConnectionMakerAction, ChannelSize) - cm := &ConnectionMaker{ - ourself: ourself, - peers: peers, - port: port, - discovery: discovery, - directPeers: peerAddrs{}, - targets: make(map[string]*Target), - connections: make(map[Connection]struct{}), - actionChan: actionChan} - go cm.queryLoop(actionChan) - return cm -} - -func (cm *ConnectionMaker) InitiateConnections(peers []string, replace bool) []error { - errors := []error{} - addrs := peerAddrs{} - for _, peer := range peers { - host, port, err := net.SplitHostPort(peer) - if err != nil { - host = peer - port = "0" // we use that as an indication that "no port was supplied" - } - if addr, err := net.ResolveTCPAddr("tcp4", fmt.Sprintf("%s:%s", host, port)); err != nil { - errors = append(errors, err) - } else { - addrs[peer] = addr - } - } - cm.actionChan <- func() bool { - if replace { - cm.directPeers = peerAddrs{} - } - for peer, addr := range addrs { - cm.directPeers[peer] = addr - // curtail any existing reconnect interval - if target, found := cm.targets[cm.completeAddr(*addr)]; found { - target.nextTryNow() - } - } - return true - } - return errors -} - -func (cm *ConnectionMaker) ForgetConnections(peers []string) { - cm.actionChan <- func() bool { - for _, peer := range peers { - delete(cm.directPeers, peer) - } - return false - } -} - -func (cm *ConnectionMaker) ConnectionAborted(address string, err error) { - cm.actionChan <- func() bool { - target := cm.targets[address] - target.state = TargetWaiting - target.lastError = err - target.nextTryLater() - return true - } -} - -func (cm *ConnectionMaker) ConnectionCreated(conn Connection) { - cm.actionChan <- func() bool { - cm.connections[conn] = void - if conn.Outbound() { - target := cm.targets[conn.RemoteTCPAddr()] - target.state = TargetConnected - } - return false - } -} - -func (cm *ConnectionMaker) ConnectionTerminated(conn Connection, err error) { - cm.actionChan <- func() bool { - delete(cm.connections, conn) - if conn.Outbound() { - target := cm.targets[conn.RemoteTCPAddr()] - target.state = TargetWaiting - target.lastError = err - switch { - case err == ErrConnectToSelf: - target.nextTryNever() - case time.Now().After(target.tryAfter.Add(ResetAfter)): - target.nextTryNow() - default: - target.nextTryLater() - } - } - return true - } -} - -func (cm *ConnectionMaker) Refresh() { - cm.actionChan <- func() bool { return true } -} - -func (cm *ConnectionMaker) queryLoop(actionChan <-chan ConnectionMakerAction) { - timer := time.NewTimer(MaxDuration) - run := func() { timer.Reset(cm.checkStateAndAttemptConnections()) } - for { - select { - case action := <-actionChan: - if action() { - run() - } - case <-timer.C: - run() - } - } -} - -func (cm *ConnectionMaker) completeAddr(addr net.TCPAddr) string { - if addr.Port == 0 { - addr.Port = cm.port - } - return addr.String() -} - -func (cm *ConnectionMaker) checkStateAndAttemptConnections() time.Duration { - var ( - validTarget = make(map[string]struct{}) - directTarget = make(map[string]struct{}) - ) - ourConnectedPeers, ourConnectedTargets, ourInboundIPs := cm.ourConnections() - - addTarget := func(address string) { - if _, connected := ourConnectedTargets[address]; connected { - return - } - validTarget[address] = void - if _, found := cm.targets[address]; found { - return - } - target := &Target{state: TargetWaiting} - target.nextTryNow() - cm.targets[address] = target - } - - // Add direct targets that are not connected - for _, addr := range cm.directPeers { - attempt := true - if addr.Port == 0 { - // If a peer was specified w/o a port, then we do not - // attempt to connect to it if we have any inbound - // connections from that IP. - if _, connected := ourInboundIPs[addr.IP.String()]; connected { - attempt = false - } - } - address := cm.completeAddr(*addr) - directTarget[address] = void - if attempt { - addTarget(address) - } - } - - // Add targets for peers that someone else is connected to, but we - // aren't - if cm.discovery { - cm.addPeerTargets(ourConnectedPeers, addTarget) - } - - return cm.connectToTargets(validTarget, directTarget) -} - -func (cm *ConnectionMaker) ourConnections() (PeerNameSet, map[string]struct{}, map[string]struct{}) { - var ( - ourConnectedPeers = make(PeerNameSet) - ourConnectedTargets = make(map[string]struct{}) - ourInboundIPs = make(map[string]struct{}) - ) - for conn := range cm.connections { - address := conn.RemoteTCPAddr() - ourConnectedPeers[conn.Remote().Name] = void - ourConnectedTargets[address] = void - if conn.Outbound() { - continue - } - if ip, _, err := net.SplitHostPort(address); err == nil { // should always succeed - ourInboundIPs[ip] = void - } - } - return ourConnectedPeers, ourConnectedTargets, ourInboundIPs -} - -func (cm *ConnectionMaker) addPeerTargets(ourConnectedPeers PeerNameSet, addTarget func(string)) { - cm.peers.ForEach(func(peer *Peer) { - if peer == cm.ourself.Peer { - return - } - // Modifying peer.connections requires a write lock on Peers, - // and since we are holding a read lock (due to the ForEach), - // access without locking the peer is safe. - for otherPeer, conn := range peer.connections { - if otherPeer == cm.ourself.Name { - continue - } - if _, connected := ourConnectedPeers[otherPeer]; connected { - continue - } - address := conn.RemoteTCPAddr() - if conn.Outbound() { - addTarget(address) - } else if ip, _, err := net.SplitHostPort(address); err == nil { - // There is no point connecting to the (likely - // ephemeral) remote port of an inbound connection - // that some peer has. Let's try to connect on the - // weave port instead. - addTarget(fmt.Sprintf("%s:%d", ip, cm.port)) - } - } - }) -} - -func (cm *ConnectionMaker) connectToTargets(validTarget map[string]struct{}, directTarget map[string]struct{}) time.Duration { - now := time.Now() // make sure we catch items just added - after := MaxDuration - for address, target := range cm.targets { - if target.state != TargetWaiting { - continue - } - if _, valid := validTarget[address]; !valid { - delete(cm.targets, address) - continue - } - if target.tryAfter.IsZero() { - continue - } - switch duration := target.tryAfter.Sub(now); { - case duration <= 0: - target.state = TargetAttempting - _, isCmdLineTarget := directTarget[address] - go cm.attemptConnection(address, isCmdLineTarget) - case duration < after: - after = duration - } - } - return after -} - -func (cm *ConnectionMaker) attemptConnection(address string, acceptNewPeer bool) { - log.Printf("->[%s] attempting connection", address) - if err := cm.ourself.CreateConnection(address, acceptNewPeer); err != nil { - log.Errorf("->[%s] error during connection attempt: %v", address, err) - cm.ConnectionAborted(address, err) - } -} - -func (t *Target) nextTryNever() { - t.tryAfter = time.Time{} - t.tryInterval = MaxInterval -} - -func (t *Target) nextTryNow() { - t.tryAfter = time.Now() - t.tryInterval = InitialInterval -} - -// The delay at the nth retry is a random value in the range -// [i-i/2,i+i/2], where i = InitialInterval * 1.5^(n-1). -func (t *Target) nextTryLater() { - t.tryAfter = time.Now().Add(t.tryInterval/2 + time.Duration(rand.Int63n(int64(t.tryInterval)))) - t.tryInterval = t.tryInterval * 3 / 2 - if t.tryInterval > MaxInterval { - t.tryInterval = MaxInterval - } -} diff --git a/mesh/gossip.go b/mesh/gossip.go deleted file mode 100644 index d757e996..00000000 --- a/mesh/gossip.go +++ /dev/null @@ -1,275 +0,0 @@ -package mesh - -import ( - "fmt" - "sync" -) - -type GossipData interface { - Encode() [][]byte - Merge(GossipData) GossipData -} - -type Gossip interface { - // specific message from one peer to another - // intermediate peers relay it using unicast topology. - GossipUnicast(dstPeerName PeerName, msg []byte) error - // send gossip to every peer, relayed using broadcast topology. - GossipBroadcast(update GossipData) error -} - -type Gossiper interface { - OnGossipUnicast(sender PeerName, msg []byte) error - // merge received data into state and return a representation of - // the received data, for further propagation - OnGossipBroadcast(sender PeerName, update []byte) (GossipData, error) - // return state of everything we know; gets called periodically - Gossip() GossipData - // merge received data into state and return "everything new I've - // just learnt", or nil if nothing in the received data was new - OnGossip(update []byte) (GossipData, error) -} - -// Accumulates GossipData that needs to be sent to one destination, -// and sends it when possible. -type GossipSender struct { - sync.Mutex - makeMsg func(msg []byte) ProtocolMsg - makeBroadcastMsg func(srcName PeerName, msg []byte) ProtocolMsg - sender ProtocolSender - gossip GossipData - broadcasts map[PeerName]GossipData - more chan<- struct{} - flush chan<- chan<- bool // for testing -} - -func NewGossipSender(makeMsg func(msg []byte) ProtocolMsg, makeBroadcastMsg func(srcName PeerName, msg []byte) ProtocolMsg, sender ProtocolSender, stop <-chan struct{}) *GossipSender { - more := make(chan struct{}, 1) - flush := make(chan chan<- bool) - s := &GossipSender{ - makeMsg: makeMsg, - makeBroadcastMsg: makeBroadcastMsg, - sender: sender, - broadcasts: make(map[PeerName]GossipData), - more: more, - flush: flush} - go s.run(stop, more, flush) - return s -} - -func (s *GossipSender) run(stop <-chan struct{}, more <-chan struct{}, flush <-chan chan<- bool) { - sent := false - for { - select { - case <-stop: - return - case <-more: - sentSomething, err := s.deliver(stop) - if err != nil { - return - } - sent = sent || sentSomething - case ch := <-flush: // for testing - // send anything pending, then reply back whether we sent - // anything since previous flush - select { - case <-more: - sentSomething, err := s.deliver(stop) - if err != nil { - return - } - sent = sent || sentSomething - default: - } - ch <- sent - sent = false - } - } -} - -func (s *GossipSender) deliver(stop <-chan struct{}) (bool, error) { - sent := false - // We must not hold our lock when sending, since that would block - // the callers of Send/Broadcast while we are stuck waiting for - // network congestion to clear. So we pick and send one piece of - // data at a time, only holding the lock during the picking. - for { - select { - case <-stop: - return sent, nil - default: - } - data, makeProtocolMsg := s.pick() - if data == nil { - return sent, nil - } - for _, msg := range data.Encode() { - if err := s.sender.SendProtocolMsg(makeProtocolMsg(msg)); err != nil { - return sent, err - } - } - sent = true - } -} - -func (s *GossipSender) pick() (data GossipData, makeProtocolMsg func(msg []byte) ProtocolMsg) { - s.Lock() - defer s.Unlock() - switch { - case s.gossip != nil: // usually more important than broadcasts - data = s.gossip - makeProtocolMsg = s.makeMsg - s.gossip = nil - case len(s.broadcasts) > 0: - for srcName, d := range s.broadcasts { - data = d - makeProtocolMsg = func(msg []byte) ProtocolMsg { return s.makeBroadcastMsg(srcName, msg) } - delete(s.broadcasts, srcName) - break - } - } - return -} - -func (s *GossipSender) Send(data GossipData) { - s.Lock() - defer s.Unlock() - if s.empty() { - defer s.prod() - } - if s.gossip == nil { - s.gossip = data - } else { - s.gossip = s.gossip.Merge(data) - } -} - -func (s *GossipSender) Broadcast(srcName PeerName, data GossipData) { - s.Lock() - defer s.Unlock() - if s.empty() { - defer s.prod() - } - d, found := s.broadcasts[srcName] - if !found { - s.broadcasts[srcName] = data - } else { - s.broadcasts[srcName] = d.Merge(data) - } -} - -func (s *GossipSender) empty() bool { return s.gossip == nil && len(s.broadcasts) == 0 } - -func (s *GossipSender) prod() { - select { - case s.more <- void: - default: - } -} - -// for testing -func (s *GossipSender) Flush() bool { - ch := make(chan bool) - s.flush <- ch - return <-ch -} - -type GossipSenders struct { - sync.Mutex - sender ProtocolSender - stop <-chan struct{} - senders map[string]*GossipSender -} - -func NewGossipSenders(sender ProtocolSender, stop <-chan struct{}) *GossipSenders { - return &GossipSenders{sender: sender, stop: stop, senders: make(map[string]*GossipSender)} -} - -func (gs *GossipSenders) Sender(channelName string, makeGossipSender func(sender ProtocolSender, stop <-chan struct{}) *GossipSender) *GossipSender { - gs.Lock() - defer gs.Unlock() - s, found := gs.senders[channelName] - if !found { - s = makeGossipSender(gs.sender, gs.stop) - gs.senders[channelName] = s - } - return s -} - -// for testing -func (gs *GossipSenders) Flush() bool { - sent := false - gs.Lock() - defer gs.Unlock() - for _, sender := range gs.senders { - sent = sender.Flush() || sent - } - return sent -} - -type GossipChannels map[string]*GossipChannel - -func (router *Router) NewGossip(channelName string, g Gossiper) Gossip { - channel := NewGossipChannel(channelName, router.Ourself, router.Routes, g) - router.gossipLock.Lock() - defer router.gossipLock.Unlock() - if _, found := router.gossipChannels[channelName]; found { - checkFatal(fmt.Errorf("[gossip] duplicate channel %s", channelName)) - } - router.gossipChannels[channelName] = channel - return channel -} - -func (router *Router) gossipChannel(channelName string) *GossipChannel { - router.gossipLock.RLock() - channel, found := router.gossipChannels[channelName] - router.gossipLock.RUnlock() - if found { - return channel - } - router.gossipLock.Lock() - defer router.gossipLock.Unlock() - if channel, found = router.gossipChannels[channelName]; found { - return channel - } - channel = NewGossipChannel(channelName, router.Ourself, router.Routes, &surrogateGossiper) - channel.log("created surrogate channel") - router.gossipChannels[channelName] = channel - return channel -} - -func (router *Router) gossipChannelSet() map[*GossipChannel]struct{} { - channels := make(map[*GossipChannel]struct{}) - router.gossipLock.RLock() - defer router.gossipLock.RUnlock() - for _, channel := range router.gossipChannels { - channels[channel] = void - } - return channels -} - -func (router *Router) SendAllGossip() { - for channel := range router.gossipChannelSet() { - if gossip := channel.gossiper.Gossip(); gossip != nil { - channel.Send(gossip) - } - } -} - -func (router *Router) SendAllGossipDown(conn Connection) { - for channel := range router.gossipChannelSet() { - if gossip := channel.gossiper.Gossip(); gossip != nil { - channel.SendDown(conn, gossip) - } - } -} - -// for testing - -func (router *Router) sendPendingGossip() bool { - sentSomething := false - for conn := range router.Ourself.Connections() { - sentSomething = conn.(GossipConnection).GossipSenders().Flush() || sentSomething - } - return sentSomething -} diff --git a/mesh/gossip_channel.go b/mesh/gossip_channel.go deleted file mode 100644 index 6a0c01f6..00000000 --- a/mesh/gossip_channel.go +++ /dev/null @@ -1,162 +0,0 @@ -package mesh - -import ( - "bytes" - "encoding/gob" - "fmt" -) - -type GossipChannel struct { - name string - ourself *LocalPeer - routes *Routes - gossiper Gossiper -} - -func NewGossipChannel(channelName string, ourself *LocalPeer, routes *Routes, g Gossiper) *GossipChannel { - return &GossipChannel{ - name: channelName, - ourself: ourself, - routes: routes, - gossiper: g, - } -} - -func (router *Router) handleGossip(tag ProtocolTag, payload []byte) error { - decoder := gob.NewDecoder(bytes.NewReader(payload)) - var channelName string - if err := decoder.Decode(&channelName); err != nil { - return err - } - channel := router.gossipChannel(channelName) - var srcName PeerName - if err := decoder.Decode(&srcName); err != nil { - return err - } - switch tag { - case ProtocolGossipUnicast: - return channel.deliverUnicast(srcName, payload, decoder) - case ProtocolGossipBroadcast: - return channel.deliverBroadcast(srcName, payload, decoder) - case ProtocolGossip: - return channel.deliver(srcName, payload, decoder) - } - return nil -} - -func (c *GossipChannel) deliverUnicast(srcName PeerName, origPayload []byte, dec *gob.Decoder) error { - var destName PeerName - if err := dec.Decode(&destName); err != nil { - return err - } - if c.ourself.Name == destName { - var payload []byte - if err := dec.Decode(&payload); err != nil { - return err - } - return c.gossiper.OnGossipUnicast(srcName, payload) - } - if err := c.relayUnicast(destName, origPayload); err != nil { - c.log(err) - } - return nil -} - -func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.Decoder) error { - var payload []byte - if err := dec.Decode(&payload); err != nil { - return err - } - data, err := c.gossiper.OnGossipBroadcast(srcName, payload) - if err != nil || data == nil { - return err - } - if err := c.relayBroadcast(srcName, data); err != nil { - c.log(err) - } - return nil -} - -func (c *GossipChannel) deliver(srcName PeerName, _ []byte, dec *gob.Decoder) error { - var payload []byte - if err := dec.Decode(&payload); err != nil { - return err - } - update, err := c.gossiper.OnGossip(payload) - if err != nil || update == nil { - return err - } - c.relay(srcName, update) - return nil -} - -func (c *GossipChannel) GossipUnicast(dstPeerName PeerName, msg []byte) error { - return c.relayUnicast(dstPeerName, GobEncode(c.name, c.ourself.Name, dstPeerName, msg)) -} - -func (c *GossipChannel) GossipBroadcast(update GossipData) error { - return c.relayBroadcast(c.ourself.Name, update) -} - -func (c *GossipChannel) Send(data GossipData) { - c.relay(c.ourself.Name, data) -} - -func (c *GossipChannel) SendDown(conn Connection, data GossipData) { - c.senderFor(conn).Send(data) -} - -func (c *GossipChannel) relayUnicast(dstPeerName PeerName, buf []byte) (err error) { - if relayPeerName, found := c.routes.UnicastAll(dstPeerName); !found { - err = fmt.Errorf("unknown relay destination: %s", dstPeerName) - } else if conn, found := c.ourself.ConnectionTo(relayPeerName); !found { - err = fmt.Errorf("unable to find connection to relay peer %s", relayPeerName) - } else { - conn.(ProtocolSender).SendProtocolMsg(ProtocolMsg{ProtocolGossipUnicast, buf}) - } - return err -} - -func (c *GossipChannel) relayBroadcast(srcName PeerName, update GossipData) error { - c.routes.EnsureRecalculated() - for _, conn := range c.ourself.ConnectionsTo(c.routes.BroadcastAll(srcName)) { - c.senderFor(conn).Broadcast(srcName, update) - } - return nil -} - -func (c *GossipChannel) relay(srcName PeerName, data GossipData) { - c.routes.EnsureRecalculated() - for _, conn := range c.ourself.ConnectionsTo(c.routes.RandomNeighbours(srcName)) { - c.senderFor(conn).Send(data) - } -} - -func (c *GossipChannel) senderFor(conn Connection) *GossipSender { - return conn.(GossipConnection).GossipSenders().Sender(c.name, c.makeGossipSender) -} - -func (c *GossipChannel) makeGossipSender(sender ProtocolSender, stop <-chan struct{}) *GossipSender { - return NewGossipSender(c.makeMsg, c.makeBroadcastMsg, sender, stop) -} - -func (c *GossipChannel) makeMsg(msg []byte) ProtocolMsg { - return ProtocolMsg{ProtocolGossip, GobEncode(c.name, c.ourself.Name, msg)} -} - -func (c *GossipChannel) makeBroadcastMsg(srcName PeerName, msg []byte) ProtocolMsg { - return ProtocolMsg{ProtocolGossipBroadcast, GobEncode(c.name, srcName, msg)} -} - -func (c *GossipChannel) log(args ...interface{}) { - log.Println(append(append([]interface{}{}, "[gossip "+c.name+"]:"), args...)...) -} - -func GobEncode(items ...interface{}) []byte { - buf := new(bytes.Buffer) - enc := gob.NewEncoder(buf) - for _, i := range items { - checkFatal(enc.Encode(i)) - } - return buf.Bytes() -} diff --git a/mesh/gossip_test.go b/mesh/gossip_test.go deleted file mode 100644 index aea73453..00000000 --- a/mesh/gossip_test.go +++ /dev/null @@ -1,237 +0,0 @@ -package mesh - -import ( - "fmt" - "sync" - "testing" - - "github.com/stretchr/testify/require" -) - -// TODO test gossip unicast; atm we only test topology gossip and -// surrogates, neither of which employ unicast. - -type MockGossipConnection struct { - RemoteConnection - dest *Router - gossipSenders *GossipSenders - start chan struct{} -} - -func NewTestRouter(name string) *Router { - peerName, _ := PeerNameFromString(name) - router := NewRouter(Config{}, peerName, "nick", nil) - router.Start() - return router -} - -func (conn *MockGossipConnection) SendProtocolMsg(protocolMsg ProtocolMsg) error { - <-conn.start - return conn.dest.handleGossip(protocolMsg.tag, protocolMsg.msg) -} - -func (conn *MockGossipConnection) GossipSenders() *GossipSenders { - return conn.gossipSenders -} - -func (conn *MockGossipConnection) Start() { - close(conn.start) -} - -func sendPendingGossip(routers ...*Router) { - // Loop until all routers report they didn't send anything - for sentSomething := true; sentSomething; { - sentSomething = false - for _, router := range routers { - sentSomething = router.sendPendingGossip() || sentSomething - } - } -} - -func AddTestGossipConnection(r1, r2 *Router) { - c1 := r1.NewTestGossipConnection(r2) - c2 := r2.NewTestGossipConnection(r1) - c1.Start() - c2.Start() -} - -func (router *Router) NewTestGossipConnection(r *Router) *MockGossipConnection { - to := r.Ourself.Peer - toPeer := NewPeer(to.Name, to.NickName, to.UID, 0, to.ShortID) - toPeer = router.Peers.FetchWithDefault(toPeer) // Has side-effect of incrementing refcount - - conn := &MockGossipConnection{ - RemoteConnection: RemoteConnection{router.Ourself.Peer, toPeer, "", false, true}, - dest: r, - start: make(chan struct{}), - } - conn.gossipSenders = NewGossipSenders(conn, make(chan struct{})) - router.Ourself.handleAddConnection(conn) - router.Ourself.handleConnectionEstablished(conn) - return conn -} - -func (router *Router) DeleteTestGossipConnection(r *Router) { - toName := r.Ourself.Peer.Name - conn, _ := router.Ourself.ConnectionTo(toName) - router.Peers.Dereference(conn.Remote()) - router.Ourself.handleDeleteConnection(conn) -} - -// Create a Peer representing the receiver router, with connections to -// the routers supplied as arguments, carrying across all UID and -// version information. -func (router *Router) tp(routers ...*Router) *Peer { - peer := NewPeerFrom(router.Ourself.Peer) - connections := make(map[PeerName]Connection) - for _, r := range routers { - p := NewPeerFrom(r.Ourself.Peer) - connections[r.Ourself.Peer.Name] = newMockConnection(peer, p) - } - peer.Version = router.Ourself.Peer.Version - peer.connections = connections - return peer -} - -// Check that the topology of router matches the peers and all of their connections -func checkTopology(t *testing.T, router *Router, wantedPeers ...*Peer) { - router.Peers.RLock() - checkTopologyPeers(t, true, router.Peers.allPeers(), wantedPeers...) - router.Peers.RUnlock() -} - -func flushAndCheckTopology(t *testing.T, routers []*Router, wantedPeers ...*Peer) { - sendPendingGossip(routers...) - for _, r := range routers { - checkTopology(t, r, wantedPeers...) - } -} - -func TestGossipTopology(t *testing.T) { - // Create some peers that will talk to each other - r1 := NewTestRouter("01:00:00:01:00:00") - r2 := NewTestRouter("02:00:00:02:00:00") - r3 := NewTestRouter("03:00:00:03:00:00") - routers := []*Router{r1, r2, r3} - // Check state when they have no connections - checkTopology(t, r1, r1.tp()) - checkTopology(t, r2, r2.tp()) - - // Now try adding some connections - AddTestGossipConnection(r1, r2) - sendPendingGossip(r1, r2) - checkTopology(t, r1, r1.tp(r2), r2.tp(r1)) - checkTopology(t, r2, r1.tp(r2), r2.tp(r1)) - - AddTestGossipConnection(r2, r3) - flushAndCheckTopology(t, routers, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2)) - - AddTestGossipConnection(r3, r1) - flushAndCheckTopology(t, routers, r1.tp(r2, r3), r2.tp(r1, r3), r3.tp(r1, r2)) - - // Drop the connection from 2 to 3 - r2.DeleteTestGossipConnection(r3) - flushAndCheckTopology(t, routers, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1, r2)) - - // Drop the connection from 1 to 3 - r1.DeleteTestGossipConnection(r3) - sendPendingGossip(r1, r2, r3) - checkTopology(t, r1, r1.tp(r2), r2.tp(r1)) - checkTopology(t, r2, r1.tp(r2), r2.tp(r1)) - // r3 still thinks r1 has a connection to it - checkTopology(t, r3, r1.tp(r2, r3), r2.tp(r1), r3.tp(r1, r2)) -} - -func TestGossipSurrogate(t *testing.T) { - // create the topology r1 <-> r2 <-> r3 - r1 := NewTestRouter("01:00:00:01:00:00") - r2 := NewTestRouter("02:00:00:02:00:00") - r3 := NewTestRouter("03:00:00:03:00:00") - routers := []*Router{r1, r2, r3} - AddTestGossipConnection(r1, r2) - AddTestGossipConnection(r3, r2) - flushAndCheckTopology(t, routers, r1.tp(r2), r2.tp(r1, r3), r3.tp(r2)) - - // create a gossiper at either end, but not the middle - g1 := newTestGossiper() - g3 := newTestGossiper() - s1 := r1.NewGossip("Test", g1) - s3 := r3.NewGossip("Test", g3) - - // broadcast a message from each end, check it reaches the other - broadcast(s1, 1) - broadcast(s3, 2) - sendPendingGossip(r1, r2, r3) - g1.checkHas(t, 2) - g3.checkHas(t, 1) - - // check that each end gets their message back through periodic - // gossip - r1.SendAllGossip() - r3.SendAllGossip() - sendPendingGossip(r1, r2, r3) - g1.checkHas(t, 1, 2) - g3.checkHas(t, 1, 2) -} - -type testGossiper struct { - sync.RWMutex - state map[byte]struct{} -} - -func newTestGossiper() *testGossiper { - return &testGossiper{state: make(map[byte]struct{})} -} - -func (g *testGossiper) OnGossipUnicast(sender PeerName, msg []byte) error { - return nil -} - -func (g *testGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) { - g.Lock() - defer g.Unlock() - for _, v := range update { - g.state[v] = void - } - return NewSurrogateGossipData(update), nil -} - -func (g *testGossiper) Gossip() GossipData { - g.RLock() - defer g.RUnlock() - state := make([]byte, len(g.state)) - for v := range g.state { - state = append(state, v) - } - return NewSurrogateGossipData(state) -} - -func (g *testGossiper) OnGossip(update []byte) (GossipData, error) { - g.Lock() - defer g.Unlock() - var delta []byte - for _, v := range update { - if _, found := g.state[v]; !found { - delta = append(delta, v) - g.state[v] = void - } - } - if len(delta) == 0 { - return nil, nil - } - return NewSurrogateGossipData(delta), nil -} - -func (g *testGossiper) checkHas(t *testing.T, vs ...byte) { - g.RLock() - defer g.RUnlock() - for _, v := range vs { - if _, found := g.state[v]; !found { - require.FailNow(t, fmt.Sprintf("%d is missing", v)) - } - } -} - -func broadcast(s Gossip, v byte) { - s.GossipBroadcast(NewSurrogateGossipData([]byte{v})) -} diff --git a/mesh/local_peer.go b/mesh/local_peer.go deleted file mode 100644 index 7592c81b..00000000 --- a/mesh/local_peer.go +++ /dev/null @@ -1,264 +0,0 @@ -package mesh - -import ( - "fmt" - "net" - "sync" - "time" -) - -type LocalPeer struct { - sync.RWMutex - *Peer - router *Router - actionChan chan<- LocalPeerAction -} - -type LocalPeerAction func() - -func NewLocalPeer(name PeerName, nickName string, router *Router) *LocalPeer { - actionChan := make(chan LocalPeerAction, ChannelSize) - peer := &LocalPeer{ - Peer: NewPeer(name, nickName, randomPeerUID(), 0, randomPeerShortID()), - router: router, - actionChan: actionChan, - } - go peer.actorLoop(actionChan) - return peer -} - -func (peer *LocalPeer) Connections() ConnectionSet { - connections := make(ConnectionSet) - peer.RLock() - defer peer.RUnlock() - for _, conn := range peer.connections { - connections[conn] = void - } - return connections -} - -func (peer *LocalPeer) 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 *LocalPeer) ConnectionsTo(names []PeerName) []Connection { - if len(names) == 0 { - return nil - } - conns := make([]Connection, 0, len(names)) - peer.RLock() - defer peer.RUnlock() - for _, name := range names { - conn, found := peer.connections[name] - // Again, !found could just be due to a race. - if found { - conns = append(conns, conn) - } - } - return conns -} - -func (peer *LocalPeer) CreateConnection(peerAddr string, acceptNewPeer bool) error { - if err := peer.checkConnectionLimit(); err != nil { - return err - } - tcpAddr, err := net.ResolveTCPAddr("tcp4", peerAddr) - if err != nil { - return err - } - tcpConn, err := net.DialTCP("tcp4", nil, tcpAddr) - if err != nil { - return err - } - connRemote := NewRemoteConnection(peer.Peer, nil, tcpConn.RemoteAddr().String(), true, false) - StartLocalConnection(connRemote, tcpConn, peer.router, acceptNewPeer) - return nil -} - -// ACTOR client API - -// Sync. -func (peer *LocalPeer) AddConnection(conn *LocalConnection) error { - resultChan := make(chan error) - peer.actionChan <- func() { - resultChan <- peer.handleAddConnection(conn) - } - return <-resultChan -} - -// Async. -func (peer *LocalPeer) ConnectionEstablished(conn *LocalConnection) { - peer.actionChan <- func() { - peer.handleConnectionEstablished(conn) - } -} - -// Sync. -func (peer *LocalPeer) DeleteConnection(conn *LocalConnection) { - resultChan := make(chan interface{}) - peer.actionChan <- func() { - peer.handleDeleteConnection(conn) - resultChan <- nil - } - <-resultChan -} - -// ACTOR server - -func (peer *LocalPeer) actorLoop(actionChan <-chan LocalPeerAction) { - gossipTimer := time.Tick(GossipInterval) - for { - select { - case action := <-actionChan: - action() - case <-gossipTimer: - peer.router.SendAllGossip() - } - } -} - -func (peer *LocalPeer) handleAddConnection(conn Connection) error { - 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 { - 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", conn.Remote(), peer.String()) - // deliberately non symmetrical - if dupConn, found := peer.connections[toName]; found { - if dupConn == conn { - return nil - } - switch conn.BreakTie(dupConn) { - case TieBreakWon: - dupConn.Shutdown(dupErr) - peer.handleDeleteConnection(dupConn) - case TieBreakLost: - return dupErr - case TieBreakTied: - // oh good grief. Sod it, just kill both of them. - dupConn.Shutdown(dupErr) - peer.handleDeleteConnection(dupConn) - return dupErr - } - } - if err := peer.checkConnectionLimit(); err != nil { - return err - } - _, isConnectedPeer := peer.router.Routes.Unicast(toName) - peer.addConnection(conn) - if isConnectedPeer { - conn.Log("connection added") - } else { - conn.Log("connection added (new peer)") - peer.router.SendAllGossipDown(conn) - } - - peer.router.Routes.Recalculate() - peer.broadcastPeerUpdate(conn.Remote()) - - return nil -} - -func (peer *LocalPeer) handleConnectionEstablished(conn Connection) { - 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 { - conn.Shutdown(fmt.Errorf("Cannot set unknown connection active")) - return - } - peer.connectionEstablished(conn) - conn.Log("connection fully established") - - peer.router.Routes.Recalculate() - peer.broadcastPeerUpdate() -} - -func (peer *LocalPeer) handleDeleteConnection(conn Connection) { - 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 { - 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.deleteConnection(conn) - conn.Log("connection deleted") - // 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.Routes.Recalculate() - peer.broadcastPeerUpdate() -} - -// helpers - -func (peer *LocalPeer) broadcastPeerUpdate(peers ...*Peer) { - // Some tests run without a router. This should be fixed so - // that the relevant part of Router can be easily run in the - // context of a test, but that will involve significant - // reworking of tests. - if peer.router != nil { - peer.router.BroadcastTopologyUpdate(append(peers, peer.Peer)) - } -} - -func (peer *LocalPeer) checkConnectionLimit() error { - limit := peer.router.ConnLimit - if 0 != limit && peer.connectionCount() >= limit { - return fmt.Errorf("Connection limit reached (%v)", limit) - } - return nil -} - -func (peer *LocalPeer) addConnection(conn Connection) { - peer.Lock() - defer peer.Unlock() - peer.connections[conn.Remote().Name] = conn - peer.Version++ -} - -func (peer *LocalPeer) deleteConnection(conn Connection) { - peer.Lock() - defer peer.Unlock() - delete(peer.connections, conn.Remote().Name) - peer.Version++ -} - -func (peer *LocalPeer) connectionEstablished(conn Connection) { - peer.Lock() - defer peer.Unlock() - peer.Version++ -} - -func (peer *LocalPeer) connectionCount() int { - peer.RLock() - defer peer.RUnlock() - return len(peer.connections) -} - -func (peer *LocalPeer) setShortID(shortID PeerShortID) { - peer.Lock() - defer peer.Unlock() - peer.ShortID = shortID - peer.Version++ -} - -func (peer *LocalPeer) setVersionBeyond(version uint64) bool { - peer.Lock() - defer peer.Unlock() - if version >= peer.Version { - peer.Version = version + 1 - return true - } - return false -} diff --git a/mesh/mocks_test.go b/mesh/mocks_test.go deleted file mode 100644 index 12ae0d74..00000000 --- a/mesh/mocks_test.go +++ /dev/null @@ -1,115 +0,0 @@ -// No mocks are tested by this file. -// -// It supplies some mock implementations to other unit tests, and is -// named "...test.go" so it is only compiled under `go test`. - -package mesh - -import ( - "fmt" - "testing" - - "github.com/stretchr/testify/require" -) - -// Add to peers a connection from peers.ourself to p -func (peers *Peers) AddTestConnection(p *Peer) { - summary := p.PeerSummary - summary.Version = 0 - toPeer := NewPeerFromSummary(summary) - toPeer = peers.FetchWithDefault(toPeer) // Has side-effect of incrementing refcount - conn := newMockConnection(peers.ourself.Peer, toPeer) - peers.ourself.addConnection(conn) - peers.ourself.connectionEstablished(conn) -} - -// Add to peers a connection from p1 to p2 -func (peers *Peers) AddTestRemoteConnection(p1, p2 *Peer) { - fromPeer := NewPeerFrom(p1) - fromPeer = peers.FetchWithDefault(fromPeer) - toPeer := NewPeerFrom(p2) - toPeer = peers.FetchWithDefault(toPeer) - peers.ourself.addConnection(&RemoteConnection{fromPeer, toPeer, "", false, false}) -} - -func (peers *Peers) DeleteTestConnection(p *Peer) { - toName := p.Name - toPeer := peers.Fetch(toName) - peers.Dereference(toPeer) - conn, _ := peers.ourself.ConnectionTo(toName) - peers.ourself.deleteConnection(conn) -} - -// mockConnection used in testing is very similar to a -// RemoteConnection, without the RemoteTCPAddr(). We are making it a -// separate type in order to distinguish what is created by the test -// from what is created by the real code. -func newMockConnection(from, to *Peer) Connection { - type mockConnection struct{ RemoteConnection } - return &mockConnection{RemoteConnection{from, to, "", false, false}} -} - -func checkEqualConns(t *testing.T, ourName PeerName, got, wanted map[PeerName]Connection) { - checkConns := make(PeerNameSet) - for _, conn := range wanted { - checkConns[conn.Remote().Name] = void - } - for _, conn := range got { - remoteName := conn.Remote().Name - if _, found := checkConns[remoteName]; found { - delete(checkConns, remoteName) - } else { - require.FailNow(t, fmt.Sprintf("Unexpected connection from %s to %s", ourName, remoteName)) - } - } - if len(checkConns) > 0 { - require.FailNow(t, fmt.Sprintf("Expected connections not found: from %s to %v", ourName, checkConns)) - } -} - -// Get all the peers from a Peers in a slice -func (peers *Peers) allPeers() []*Peer { - var res []*Peer - for _, peer := range peers.byName { - res = append(res, peer) - } - return res -} - -func (peers *Peers) allPeersExcept(excludeName PeerName) []*Peer { - res := peers.allPeers() - for i, peer := range res { - if peer.Name == excludeName { - return append(res[:i], res[i+1:]...) - } - } - return res -} - -// Check that the peers slice matches the wanted peers -func checkPeerArray(t *testing.T, peers []*Peer, wantedPeers ...*Peer) { - checkTopologyPeers(t, false, peers, wantedPeers...) -} - -// Check that the peers slice matches the wanted peers and optionally -// all of their connections -func checkTopologyPeers(t *testing.T, checkConns bool, peers []*Peer, wantedPeers ...*Peer) { - check := make(map[PeerName]*Peer) - for _, peer := range wantedPeers { - check[peer.Name] = peer - } - for _, peer := range peers { - name := peer.Name - if wantedPeer, found := check[name]; found { - if checkConns { - checkEqualConns(t, name, peer.connections, wantedPeer.connections) - } - delete(check, name) - } else { - require.FailNow(t, fmt.Sprintf("Unexpected peer: %s", name)) - } - } - if len(check) > 0 { - require.FailNow(t, fmt.Sprintf("Expected peers not found: %v", check)) - } -} diff --git a/mesh/overlay.go b/mesh/overlay.go deleted file mode 100644 index 5c48c927..00000000 --- a/mesh/overlay.go +++ /dev/null @@ -1,116 +0,0 @@ -package mesh - -import ( - "net" -) - -type Overlay interface { - // Enhance a features map with overlay-related features - AddFeaturesTo(map[string]string) - - // Prepare on overlay connection. The connection should remain - // passive until it has been Confirm()ed. - PrepareConnection(OverlayConnectionParams) (OverlayConnection, error) - - // Obtain diagnostic information specific to the overlay - Diagnostics() interface{} -} - -type OverlayConnectionParams struct { - RemotePeer *Peer - - // The local address of the corresponding TCP connection. Used to - // derive the local IP address for sending. May differ for - // different overlay connections. - LocalAddr *net.TCPAddr - - // The remote address of the corresponding TCP connection. Used to - // determine the address to send to, but only if the TCP - // connection is outbound. Otherwise the Overlay needs to discover - // it (e.g. from incoming datagrams). - RemoteAddr *net.TCPAddr - - // Is the corresponding TCP connection outbound? - Outbound bool - - // Unique identifier for this connection - ConnUID uint64 - - // Session key, if connection is encrypted; nil otherwise. - // - // NB: overlay connections must take care not to use nonces which - // may collide with those of the main connection. These nonces are - // 192 bits, with the top most bit unspecified, the next bit set - // to 1, followed by 126 zero bits, and a message sequence number - // in the lowest 64 bits. - SessionKey *[32]byte - - // Function to send a control message to the counterpart - // overlay connection. - SendControlMessage func(tag byte, msg []byte) error - - // Features passed at connection initiation - Features map[string]string -} - -// All of the machinery to manage overlay connectivity to a particular -// peer -type OverlayConnection interface { - // Confirm that the connection is really wanted, and so the - // Overlay should begin heartbeats etc. to verify the operation of - // the overlay connection. - Confirm() - - // A channel indicating that the overlay connection is - // established, i.e. its operation has been confirmed. - EstablishedChannel() <-chan struct{} - - // A channel indicating an error from the overlay connection. The - // overlay connection is not expected to be operational after the - // first error, so the channel only needs to buffer a single - // error. - ErrorChannel() <-chan error - - Stop() - - // Handle a message from the peer. 'tag' exists for - // compatibility, and should always be - // ProtocolOverlayControlMessage for non-sleeve overlays. - ControlMessage(tag byte, msg []byte) - - // User facing overlay name - DisplayName() string -} - -type NullOverlay struct{} - -func (NullOverlay) AddFeaturesTo(map[string]string) { -} - -func (NullOverlay) PrepareConnection(OverlayConnectionParams) (OverlayConnection, error) { - return NullOverlay{}, nil -} - -func (NullOverlay) Diagnostics() interface{} { - return nil -} -func (NullOverlay) Confirm() { -} - -func (NullOverlay) EstablishedChannel() <-chan struct{} { - return nil -} - -func (NullOverlay) ErrorChannel() <-chan error { - return nil -} - -func (NullOverlay) Stop() { -} - -func (NullOverlay) ControlMessage(byte, []byte) { -} - -func (NullOverlay) DisplayName() string { - return "null" -} diff --git a/mesh/peer.go b/mesh/peer.go deleted file mode 100644 index 44d14893..00000000 --- a/mesh/peer.go +++ /dev/null @@ -1,182 +0,0 @@ -package mesh - -import ( - "crypto/rand" - "encoding/binary" - "fmt" - "sort" - "strconv" -) - -func randBytes(n int) []byte { - buf := make([]byte, n) - _, err := rand.Read(buf) - checkFatal(err) - return buf -} - -func randUint64() (r uint64) { - return binary.LittleEndian.Uint64(randBytes(8)) -} - -func randUint16() (r uint16) { - return binary.LittleEndian.Uint16(randBytes(2)) -} - -type PeerUID uint64 - -func randomPeerUID() PeerUID { - for { - uid := randUint64() - if uid != 0 { // uid 0 is reserved for peer placeholder - return PeerUID(uid) - } - } -} - -func ParsePeerUID(s string) (PeerUID, error) { - uid, err := strconv.ParseUint(s, 10, 64) - return PeerUID(uid), err -} - -// Short IDs exist for the sake of fast datapath. They are 12 bits, -// randomly assigned, but we detect and recover from collisions. This -// does limit us to 4096 peers, but that should be sufficient for a -// while. -type PeerShortID uint16 - -const PeerShortIDBits = 12 - -func randomPeerShortID() PeerShortID { - return PeerShortID(randUint16() & (1< PeerNameY, -// which says "in order to send a message to X, the peer should send -// the message to its neighbour Y". -// -// Because currently we do not have weightings on the connections -// between peers, there is no need to use a minimum spanning tree -// algorithm. Instead we employ the simpler and cheaper breadth-first -// widening. The computation is deterministic, which ensures that when -// it is performed on the same data by different peers, they get the -// same result. This is important since otherwise we risk message loss -// or routing cycles. -// -// When the 'establishedAndSymmetric' flag is set, only connections -// that are marked as 'established' and are symmetric (i.e. where both -// sides indicate they have a connection to the other) are considered. -// -// When a non-nil stopAt peer is supplied, the widening stops when it -// reaches that peer. The boolean return indicates whether that has -// happened. -// -// NB: This function should generally be invoked while holding a read -// lock on Peers and LocalPeer. -func (peer *Peer) Routes(stopAt *Peer, establishedAndSymmetric bool) (bool, map[PeerName]PeerName) { - routes := make(unicastRoutes) - routes[peer.Name] = UnknownPeerName - nextWorklist := []*Peer{peer} - for len(nextWorklist) > 0 { - worklist := nextWorklist - sort.Sort(ListOfPeers(worklist)) - nextWorklist = []*Peer{} - for _, curPeer := range worklist { - if curPeer == stopAt { - return true, routes - } - curPeer.ForEachConnectedPeer(establishedAndSymmetric, routes, - func(remotePeer *Peer) { - nextWorklist = append(nextWorklist, remotePeer) - remoteName := remotePeer.Name - // We now know how to get to remoteName: the same - // way we get to curPeer. Except, if curPeer is - // the starting peer in which case we know we can - // reach remoteName directly. - if curPeer == peer { - routes[remoteName] = remoteName - } else { - routes[remoteName] = routes[curPeer.Name] - } - }) - } - } - return false, routes -} - -func (peer *Peer) ForEachConnectedPeer(establishedAndSymmetric bool, exclude map[PeerName]PeerName, f func(*Peer)) { - for remoteName, conn := range peer.connections { - if establishedAndSymmetric && !conn.Established() { - continue - } - if _, found := exclude[remoteName]; found { - continue - } - remotePeer := conn.Remote() - if remoteConn, found := remotePeer.connections[peer.Name]; !establishedAndSymmetric || (found && remoteConn.Established()) { - f(remotePeer) - } - } -} diff --git a/mesh/peer_name_hash.go b/mesh/peer_name_hash.go deleted file mode 100644 index ac76103e..00000000 --- a/mesh/peer_name_hash.go +++ /dev/null @@ -1,46 +0,0 @@ -// +build peer_name_hash - -// Let peer names be sha256 hashes...of anything (as long as it's -// unique). - -package mesh - -import ( - "crypto/sha256" - "encoding/hex" -) - -type PeerName string - -const ( - PeerNameFlavour = "hash" - NameSize = sha256.Size >> 1 - UnknownPeerName = PeerName("") -) - -func PeerNameFromUserInput(userInput string) (PeerName, error) { - // fixed-length identity - nameByteAry := sha256.Sum256([]byte(userInput)) - return PeerNameFromBin(nameByteAry[:NameSize]), nil -} - -func PeerNameFromString(nameStr string) (PeerName, error) { - if _, err := hex.DecodeString(nameStr); err != nil { - return UnknownPeerName, err - } - return PeerName(nameStr), nil -} - -func PeerNameFromBin(nameByte []byte) PeerName { - return PeerName(hex.EncodeToString(nameByte)) -} - -func (name PeerName) Bin() []byte { - res, err := hex.DecodeString(string(name)) - checkFatal(err) - return res -} - -func (name PeerName) String() string { - return string(name) -} diff --git a/mesh/peer_name_mac.go b/mesh/peer_name_mac.go deleted file mode 100644 index c2b6cdbe..00000000 --- a/mesh/peer_name_mac.go +++ /dev/null @@ -1,75 +0,0 @@ -// The !peer_name_alternative effectively makes this the default, -// i.e. to choose an alternative, run "go build -tags -// 'peer_name_alternative peer_name_hash'" - -// +build peer_name_mac !peer_name_alternative - -// Let peer names be MACs... -// -// MACs need to be unique across our network, or bad things will -// happen anyway. So they make pretty good candidates for peer -// names. And doing so is pretty efficient both computationally and -// network overhead wise. -// -// Note that we do not mandate *what* MAC should be used as the peer -// name. In particular it doesn't actually have to be the MAC of, say, -// the network interface the peer is sniffing on. - -package mesh - -import ( - "net" -) - -// PeerName is used as a map key. Since net.HardwareAddr isn't -// suitable for that - it's a slice, and slices can't be map keys - we -// convert that to/from uint64. -type PeerName uint64 - -const ( - PeerNameFlavour = "mac" - NameSize = 6 - UnknownPeerName = PeerName(0) -) - -func PeerNameFromUserInput(userInput string) (PeerName, error) { - res, err := PeerNameFromString(userInput) - return res, err -} - -func PeerNameFromString(nameStr string) (PeerName, error) { - mac, err := net.ParseMAC(nameStr) - if err != nil { - return UnknownPeerName, err - } - return PeerName(macint(mac)), nil -} - -func PeerNameFromBin(nameByte []byte) PeerName { - return PeerName(macint(net.HardwareAddr(nameByte))) -} - -func (name PeerName) Bin() []byte { - return intmac(uint64(name)) -} - -func (name PeerName) String() string { - return intmac(uint64(name)).String() -} - -func macint(mac net.HardwareAddr) (r uint64) { - for _, b := range mac { - r <<= 8 - r |= uint64(b) - } - return -} - -func intmac(key uint64) (r net.HardwareAddr) { - r = make([]byte, 6) - for i := 5; i >= 0; i-- { - r[i] = byte(key) - key >>= 8 - } - return -} diff --git a/mesh/peers.go b/mesh/peers.go deleted file mode 100644 index f4573665..00000000 --- a/mesh/peers.go +++ /dev/null @@ -1,530 +0,0 @@ -package mesh - -import ( - "bytes" - "encoding/gob" - "io" - "math/rand" - "sync" -) - -var void = struct{}{} - -type Peers struct { - sync.RWMutex - ourself *LocalPeer - byName map[PeerName]*Peer - byShortID map[PeerShortID]ShortIDPeers - onGC []func(*Peer) - - // Called when the mapping from short ids to peers changes - onInvalidateShortIDs []func() -} - -type ShortIDPeers struct { - // If we know about a single peer with the short id, this is - // that peer. If there is a collision, this is the peer with - // the lowest Name. - peer *Peer - - // In case of a collision, this holds the other peers. - others []*Peer -} - -type PeerNameSet map[PeerName]struct{} - -type ConnectionSummary struct { - NameByte []byte - RemoteTCPAddr string - Outbound bool - Established bool -} - -// Pending notifications due to changes to Peers that need to be sent -// out once the Peers is unlocked. -type PeersPendingNotifications struct { - // Peers that have been GCed - removed []*Peer - - // The mapping from shorts ids to peers changed - invalidateShortIDs bool - - // The local short ID needs reassigning due to a collision - reassignLocalShortID bool - - // The local peer was modified - localPeerModified bool -} - -func NewPeers(ourself *LocalPeer) *Peers { - peers := &Peers{ - ourself: ourself, - byName: make(map[PeerName]*Peer), - byShortID: make(map[PeerShortID]ShortIDPeers), - } - peers.FetchWithDefault(ourself.Peer) - return peers -} - -func (peers *Peers) OnGC(callback func(*Peer)) { - peers.Lock() - defer peers.Unlock() - - // Although the array underlying peers.onGC might be accessed - // without holding the lock in unlockAndNotify, we don't - // support removing callbacks, so a simple append here is - // safe. - peers.onGC = append(peers.onGC, callback) -} - -func (peers *Peers) OnInvalidateShortIDs(callback func()) { - peers.Lock() - defer peers.Unlock() - - // Safe, as in OnGC - peers.onInvalidateShortIDs = append(peers.onInvalidateShortIDs, callback) -} - -func (peers *Peers) unlockAndNotify(pending *PeersPendingNotifications) { - broadcastLocalPeer := (pending.reassignLocalShortID && peers.reassignLocalShortID(pending)) || - pending.localPeerModified - onGC := peers.onGC - onInvalidateShortIDs := peers.onInvalidateShortIDs - peers.Unlock() - - if pending.removed != nil { - for _, callback := range onGC { - for _, peer := range pending.removed { - callback(peer) - } - } - } - - if pending.invalidateShortIDs { - for _, callback := range onInvalidateShortIDs { - callback() - } - } - - if broadcastLocalPeer { - peers.ourself.broadcastPeerUpdate() - } -} - -func (peers *Peers) addByShortID(peer *Peer, pending *PeersPendingNotifications) { - if !peer.HasShortID { - return - } - - entry, ok := peers.byShortID[peer.ShortID] - if !ok { - entry = ShortIDPeers{peer: peer} - } else if entry.peer == nil { - // This short ID is free, but was used in the past. - // Because we are reusing it, it's an invalidation - // event. - entry.peer = peer - pending.invalidateShortIDs = true - } else if peer.Name < entry.peer.Name { - // Short ID collision, this peer becomes the principal - // peer for the short ID, bumping the previous one - // into others. - - if entry.peer == peers.ourself.Peer { - // The bumped peer is peers.ourself, so we - // need to look for a new short id - pending.reassignLocalShortID = true - } - - entry.others = append(entry.others, entry.peer) - entry.peer = peer - pending.invalidateShortIDs = true - } else { - // Short ID collision, this peer is secondary - entry.others = append(entry.others, peer) - } - - peers.byShortID[peer.ShortID] = entry -} - -func (peers *Peers) deleteByShortID(peer *Peer, pending *PeersPendingNotifications) { - if !peer.HasShortID { - return - } - - entry := peers.byShortID[peer.ShortID] - var otherIndex int - - if peer != entry.peer { - // peer is secondary, find its index in others - otherIndex = -1 - - for i, other := range entry.others { - if peer == other { - otherIndex = i - break - } - } - - if otherIndex < 0 { - return - } - } else if len(entry.others) != 0 { - // need to find the peer with the lowest name to - // become the new principal one - otherIndex = 0 - minName := entry.others[0].Name - - for i := 1; i < len(entry.others); i++ { - otherName := entry.others[i].Name - if otherName < minName { - minName = otherName - otherIndex = i - } - } - - entry.peer = entry.others[otherIndex] - pending.invalidateShortIDs = true - } else { - // This is the last peer with the short id. We clear - // the entry, don't delete it, in order to detect when - // it gets re-used. - peers.byShortID[peer.ShortID] = ShortIDPeers{} - return - } - - entry.others[otherIndex] = entry.others[len(entry.others)-1] - entry.others = entry.others[:len(entry.others)-1] - peers.byShortID[peer.ShortID] = entry -} - -func (peers *Peers) reassignLocalShortID(pending *PeersPendingNotifications) bool { - newShortID, ok := peers.chooseShortID() - if ok { - peers.setLocalShortID(newShortID, pending) - return true - } - - // Otherwise we'll try again later on in garbageColleect - return false -} - -func (peers *Peers) setLocalShortID(newShortID PeerShortID, pending *PeersPendingNotifications) { - peers.deleteByShortID(peers.ourself.Peer, pending) - peers.ourself.setShortID(newShortID) - peers.addByShortID(peers.ourself.Peer, pending) -} - -// Choose an available short id at random -func (peers *Peers) chooseShortID() (PeerShortID, bool) { - rng := rand.New(rand.NewSource(int64(randUint64()))) - - // First, just try picking some short ids at random, and - // seeing if they are available: - for i := 0; i < 10; i++ { - shortID := PeerShortID(rng.Intn(1 << PeerShortIDBits)) - if peers.byShortID[shortID].peer == nil { - return shortID, true - } - } - - // Looks like most short ids are used. So count the number of - // unused ones, and pick one at random. - available := int(1 << PeerShortIDBits) - for _, entry := range peers.byShortID { - if entry.peer != nil { - available-- - } - } - - if available == 0 { - // All short ids are used. - return 0, false - } - - n := rng.Intn(available) - var i PeerShortID - for { - if peers.byShortID[i].peer == nil { - if n == 0 { - return PeerShortID(i), true - } - - n-- - } - - i++ - } -} - -func (peers *Peers) FetchWithDefault(peer *Peer) *Peer { - peers.Lock() - var pending PeersPendingNotifications - defer peers.unlockAndNotify(&pending) - - if existingPeer, found := peers.byName[peer.Name]; found { - existingPeer.localRefCount++ - return existingPeer - } - - peers.byName[peer.Name] = peer - peers.addByShortID(peer, &pending) - peer.localRefCount++ - return peer -} - -func (peers *Peers) Fetch(name PeerName) *Peer { - peers.RLock() - defer peers.RUnlock() - return peers.byName[name] -} - -func (peers *Peers) FetchAndAddRef(name PeerName) *Peer { - peers.Lock() - defer peers.Unlock() - peer := peers.byName[name] - if peer != nil { - peer.localRefCount++ - } - return peer -} - -func (peers *Peers) FetchByShortID(shortID PeerShortID) *Peer { - peers.RLock() - defer peers.RUnlock() - return peers.byShortID[shortID].peer -} - -func (peers *Peers) Dereference(peer *Peer) { - peers.Lock() - defer peers.Unlock() - peer.localRefCount-- -} - -func (peers *Peers) ForEach(fun func(*Peer)) { - peers.RLock() - defer peers.RUnlock() - for _, peer := range peers.byName { - fun(peer) - } -} - -// Merge an incoming update with our own topology. -// -// We add peers hitherto unknown to us, and update peers for which the -// update contains a more recent version than known to us. The return -// value is a) a representation of the received update, and b) an -// "improved" update containing just these new/updated elements. -func (peers *Peers) ApplyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) { - peers.Lock() - var pending PeersPendingNotifications - defer peers.unlockAndNotify(&pending) - - newPeers, decodedUpdate, decodedConns, err := peers.decodeUpdate(update) - if err != nil { - return nil, nil, err - } - - // Add new peers - for name, newPeer := range newPeers { - peers.byName[name] = newPeer - peers.addByShortID(newPeer, &pending) - } - - // Now apply the updates - newUpdate := peers.applyUpdate(decodedUpdate, decodedConns, &pending) - peers.garbageCollect(&pending) - for _, peerRemoved := range pending.removed { - delete(newUpdate, peerRemoved.Name) - } - - updateNames := make(PeerNameSet) - for _, peer := range decodedUpdate { - updateNames[peer.Name] = void - } - - return updateNames, newUpdate, nil -} - -func (peers *Peers) Names() PeerNameSet { - peers.RLock() - defer peers.RUnlock() - - names := make(PeerNameSet) - for name := range peers.byName { - names[name] = void - } - return names -} - -func (peers *Peers) EncodePeers(names PeerNameSet) []byte { - buf := new(bytes.Buffer) - enc := gob.NewEncoder(buf) - peers.RLock() - defer peers.RUnlock() - for name := range names { - if peer, found := peers.byName[name]; found { - if peer == peers.ourself.Peer { - peers.ourself.Encode(enc) - } else { - peer.Encode(enc) - } - } - } - return buf.Bytes() -} - -func (peers *Peers) GarbageCollect() { - peers.Lock() - var pending PeersPendingNotifications - defer peers.unlockAndNotify(&pending) - - peers.garbageCollect(&pending) -} - -func (peers *Peers) garbageCollect(pending *PeersPendingNotifications) { - peers.ourself.RLock() - _, reached := peers.ourself.Routes(nil, false) - peers.ourself.RUnlock() - - for name, peer := range peers.byName { - if _, found := reached[peer.Name]; !found && peer.localRefCount == 0 { - delete(peers.byName, name) - peers.deleteByShortID(peer, pending) - pending.removed = append(pending.removed, peer) - } - } - - if len(pending.removed) > 0 && peers.byShortID[peers.ourself.ShortID].peer != peers.ourself.Peer { - // The local peer doesn't own its short id. Garbage - // collection might have freed some up, so try to - // reassign. - pending.reassignLocalShortID = true - } -} - -func (peers *Peers) decodeUpdate(update []byte) (newPeers map[PeerName]*Peer, decodedUpdate []*Peer, decodedConns [][]ConnectionSummary, err error) { - newPeers = make(map[PeerName]*Peer) - decodedUpdate = []*Peer{} - decodedConns = [][]ConnectionSummary{} - - decoder := gob.NewDecoder(bytes.NewReader(update)) - - for { - peerSummary, connSummaries, decErr := decodePeer(decoder) - if decErr == io.EOF { - break - } else if decErr != nil { - err = decErr - return - } - newPeer := NewPeerFromSummary(peerSummary) - decodedUpdate = append(decodedUpdate, newPeer) - decodedConns = append(decodedConns, connSummaries) - if _, found := peers.byName[newPeer.Name]; !found { - newPeers[newPeer.Name] = newPeer - } - } - - for _, connSummaries := range decodedConns { - for _, connSummary := range connSummaries { - remoteName := PeerNameFromBin(connSummary.NameByte) - if _, found := newPeers[remoteName]; found { - continue - } - if _, found := peers.byName[remoteName]; found { - continue - } - // Update refers to a peer which we have no knowledge of. - newPeers[remoteName] = NewPeerPlaceholder(remoteName) - } - } - return -} - -func (peers *Peers) applyUpdate(decodedUpdate []*Peer, decodedConns [][]ConnectionSummary, pending *PeersPendingNotifications) PeerNameSet { - newUpdate := make(PeerNameSet) - for idx, newPeer := range decodedUpdate { - connSummaries := decodedConns[idx] - name := newPeer.Name - // guaranteed to find peer in the peers.byName - switch peer := peers.byName[name]; peer { - case peers.ourself.Peer: - if newPeer.UID != peer.UID { - // The update contains information about an old - // incarnation of ourselves. We increase our version - // number beyond that which we received, so our - // information supersedes the old one when it is - // received by other peers. - pending.localPeerModified = peers.ourself.setVersionBeyond(newPeer.Version) - } - case newPeer: - peer.connections = makeConnsMap(peer, connSummaries, peers.byName) - newUpdate[name] = void - default: // existing peer - if newPeer.Version < peer.Version || - (newPeer.Version == peer.Version && - (newPeer.UID < peer.UID || - (newPeer.UID == peer.UID && - (!newPeer.HasShortID || peer.HasShortID)))) { - continue - } - peer.Version = newPeer.Version - peer.UID = newPeer.UID - peer.NickName = newPeer.NickName - peer.connections = makeConnsMap(peer, connSummaries, peers.byName) - - if newPeer.ShortID != peer.ShortID || newPeer.HasShortID != peer.HasShortID { - peers.deleteByShortID(peer, pending) - peer.ShortID = newPeer.ShortID - peer.HasShortID = newPeer.HasShortID - peers.addByShortID(peer, pending) - } - newUpdate[name] = void - } - } - return newUpdate -} - -func (peer *Peer) Encode(enc *gob.Encoder) { - checkFatal(enc.Encode(peer.PeerSummary)) - - connSummaries := []ConnectionSummary{} - for _, conn := range peer.connections { - connSummaries = append(connSummaries, ConnectionSummary{ - conn.Remote().NameByte, - conn.RemoteTCPAddr(), - conn.Outbound(), - conn.Established(), - }) - } - - checkFatal(enc.Encode(connSummaries)) -} - -func (peer *LocalPeer) Encode(enc *gob.Encoder) { - peer.RLock() - defer peer.RUnlock() - peer.Peer.Encode(enc) -} - -func decodePeer(dec *gob.Decoder) (peerSummary PeerSummary, connSummaries []ConnectionSummary, err error) { - if err = dec.Decode(&peerSummary); err != nil { - return - } - if err = dec.Decode(&connSummaries); err != nil { - return - } - return -} - -func makeConnsMap(peer *Peer, connSummaries []ConnectionSummary, byName map[PeerName]*Peer) map[PeerName]Connection { - conns := make(map[PeerName]Connection) - for _, connSummary := range connSummaries { - name := PeerNameFromBin(connSummary.NameByte) - remotePeer := byName[name] - conn := NewRemoteConnection(peer, remotePeer, connSummary.RemoteTCPAddr, connSummary.Outbound, connSummary.Established) - conns[name] = conn - } - return conns -} diff --git a/mesh/peers_test.go b/mesh/peers_test.go deleted file mode 100644 index 5c2b0094..00000000 --- a/mesh/peers_test.go +++ /dev/null @@ -1,376 +0,0 @@ -package mesh - -import ( - "fmt" - "math/rand" - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -// TODO we should also test: -// -// - applying an incremental update, including the case where that -// leads to an UnknownPeerError -// -// - the "improved update" calculation -// -// - non-gc of peers that are only referenced locally - -func newNode(name PeerName) (*Peer, *Peers) { - peer := NewLocalPeer(name, "", nil) - peers := NewPeers(peer) - return peer.Peer, peers -} - -// Check that ApplyUpdate copies the whole topology from peers -func checkApplyUpdate(t *testing.T, peers *Peers) { - dummyName, _ := PeerNameFromString("99:00:00:01:00:00") - // We need a new node outside of the network, with a connection - // into it. - _, testBedPeers := newNode(dummyName) - testBedPeers.AddTestConnection(peers.ourself.Peer) - testBedPeers.ApplyUpdate(peers.EncodePeers(peers.Names())) - - checkTopologyPeers(t, true, testBedPeers.allPeersExcept(dummyName), peers.allPeers()...) -} - -func TestPeersEncoding(t *testing.T) { - const numNodes = 20 - const numIters = 1000 - var peer [numNodes]*Peer - var ps [numNodes]*Peers - for i := 0; i < numNodes; i++ { - name, _ := PeerNameFromString(fmt.Sprintf("%02d:00:00:01:00:00", i)) - peer[i], ps[i] = newNode(name) - } - - var conns []struct{ from, to int } - for i := 0; i < numIters; i++ { - oper := rand.Intn(2) - switch oper { - case 0: - from, to := rand.Intn(numNodes), rand.Intn(numNodes) - if from != to { - if _, found := peer[from].connections[peer[to].Name]; !found { - ps[from].AddTestConnection(peer[to]) - conns = append(conns, struct{ from, to int }{from, to}) - checkApplyUpdate(t, ps[from]) - } - } - case 1: - if len(conns) > 0 { - n := rand.Intn(len(conns)) - c := conns[n] - ps[c.from].DeleteTestConnection(peer[c.to]) - ps[c.from].GarbageCollect() - checkApplyUpdate(t, ps[c.from]) - conns = append(conns[:n], conns[n+1:]...) - } - } - } -} - -func garbageCollect(peers *Peers) []*Peer { - var removed []*Peer - peers.OnGC(func(peer *Peer) { removed = append(removed, peer) }) - peers.GarbageCollect() - return removed -} - -func TestPeersGarbageCollection(t *testing.T) { - const ( - peer1NameString = "01:00:00:01:00:00" - peer2NameString = "02:00:00:02:00:00" - peer3NameString = "03:00:00:03:00:00" - ) - var ( - peer1Name, _ = PeerNameFromString(peer1NameString) - peer2Name, _ = PeerNameFromString(peer2NameString) - peer3Name, _ = PeerNameFromString(peer3NameString) - ) - - // Create some peers with some connections to each other - p1, ps1 := newNode(peer1Name) - p2, ps2 := newNode(peer2Name) - p3, ps3 := newNode(peer3Name) - ps1.AddTestConnection(p2) - ps2.AddTestRemoteConnection(p1, p2) - ps2.AddTestConnection(p1) - ps2.AddTestConnection(p3) - ps3.AddTestConnection(p1) - ps1.AddTestConnection(p3) - ps2.AddTestRemoteConnection(p1, p3) - ps2.AddTestRemoteConnection(p3, p1) - - // Every peer is referenced, so nothing should be dropped - require.Empty(t, garbageCollect(ps1), "peers removed") - require.Empty(t, garbageCollect(ps2), "peers removed") - require.Empty(t, garbageCollect(ps3), "peers removed") - - // Drop the connection from 2 to 3, and 3 isn't garbage-collected - // because 1 has a connection to 3 - ps2.DeleteTestConnection(p3) - require.Empty(t, garbageCollect(ps2), "peers removed") - - // Drop the connection from 1 to 3, and 3 will get removed by - // garbage-collection - ps1.DeleteTestConnection(p3) - checkPeerArray(t, garbageCollect(ps1), p3) -} - -func TestShortIDCollisions(t *testing.T) { - rng := rand.New(rand.NewSource(time.Now().UnixNano())) - _, peers := newNode(PeerName(1 << PeerShortIDBits)) - - // Make enough peers that short id collisions are - // overwhelmingly likely - ps := make([]*Peer, 1< entry.peer.Name) - } - } - - // Check that every peer was seen - for _, n := range counts { - require.Equal(t, 1, n) - } - - // Delete all the peers - shuffle() - for _, p := range ps { - peers.deleteByShortID(p, &pending) - } - - for _, entry := range peers.byShortID { - if entry.peer != peers.ourself.Peer { - require.Nil(t, entry.peer) - } - - require.Empty(t, entry.others) - } -} - -// Test the easy case of short id reassignment, when few short ids are taken -func TestShortIDReassignmentEasy(t *testing.T) { - rng := rand.New(rand.NewSource(time.Now().UnixNano())) - _, peers := newNode(PeerName(0)) - - for i := 1; i <= 10; i++ { - peers.FetchWithDefault(NewPeer(PeerName(i), "", PeerUID(i), 0, - PeerShortID(rng.Intn(1< minVersion { - minVersion = params.MinVersion - } - - theirMaxVersion := header[len(ProtocolBytes)+1] - maxVersion := theirMaxVersion - if maxVersion > params.MaxVersion { - maxVersion = params.MaxVersion - } - - if minVersion > maxVersion { - return 0, fmt.Errorf("remote version range [%d,%d] is incompatible with ours [%d,%d]", - theirMinVersion, theirMaxVersion, - params.MinVersion, params.MaxVersion) - } - - if err := <-writeDone; err != nil { - return 0, err - } - - return maxVersion, nil -} - -// The V1 procotol consists of the protocol identification/version -// header, followed by a stream of gobified values. The first value -// is the encoded features map (never encrypted). The subsequent -// values are the messages on the connection (encrypted for an -// encrypted connection). For an encrypted connection, the public key -// is passed in the "PublicKey" feature as a string of hex digits. -func (res *ProtocolIntroResults) doIntroV1(params ProtocolIntroParams, pubKey, privKey *[32]byte) error { - features := filterV1Features(params.Features) - if pubKey != nil { - features["PublicKey"] = hex.EncodeToString(pubKey[:]) - } - - enc := gob.NewEncoder(params.Conn) - dec := gob.NewDecoder(params.Conn) - - // Encode in a separate goroutine to avoid the possibility of - // deadlock. The result channel is of size 1 so that the - // goroutine does not linger even if we encounter an error on - // the read side. - encodeDone := make(chan error, 1) - go func() { - encodeDone <- enc.Encode(features) - }() - - if err := dec.Decode(&res.Features); err != nil { - return err - } - - if err := <-encodeDone; err != nil { - return err - } - - res.Sender = NewGobTCPSender(enc) - res.Receiver = NewGobTCPReceiver(dec) - - if pubKey == nil { - if _, present := res.Features["PublicKey"]; present { - return ErrExpectedNoCrypto - } - } else { - remotePubKeyStr, ok := res.Features["PublicKey"] - if !ok { - return ErrExpectedCrypto - } - - remotePubKey, err := hex.DecodeString(remotePubKeyStr) - if err != nil { - return err - } - - res.setupCrypto(params, remotePubKey, privKey) - } - - res.Features = filterV1Features(res.Features) - return nil -} - -// In the V1 protocol, the intro fields are sent unencrypted. So we -// restrict them to an established subset of fields that are assumed -// to be safe. -func filterV1Features(intro map[string]string) map[string]string { - safe := make(map[string]string) - for _, k := range ProtocolV1Features { - if val, ok := intro[k]; ok { - safe[k] = val - } - } - - return safe -} - -// The V2 procotol consists of the protocol identification/version -// header, followed by: -// -// - A single "encryption flag" byte: 0 for no encryption, 1 for -// encryption. -// -// - When the connection is encrypted, 32 bytes follow containing the -// public key. -// -// - Then a stream of length-prefixed messages, which are encrypted -// for an encrypted connection. -// -// The first message contains the encoded features map (so in contrast -// to V1, it will be encrypted on an encrypted connection). -func (res *ProtocolIntroResults) doIntroV2(params ProtocolIntroParams, pubKey, privKey *[32]byte) error { - // Public key exchange - var wbuf []byte - if pubKey == nil { - wbuf = []byte{0} - } else { - wbuf = make([]byte, 1+len(*pubKey)) - wbuf[0] = 1 - copy(wbuf[1:], (*pubKey)[:]) - } - - // Write in a separate goroutine to avoid the possibility of - // deadlock. The result channel is of size 1 so that the - // goroutine does not linger even if we encounter an error on - // the read side. - writeDone := make(chan error, 1) - go func() { - _, err := params.Conn.Write(wbuf) - writeDone <- err - }() - - rbuf := make([]byte, 1) - if _, err := io.ReadFull(params.Conn, rbuf); err != nil { - return err - } - - switch rbuf[0] { - case 0: - if pubKey != nil { - return ErrExpectedCrypto - } - - res.Sender = NewLengthPrefixTCPSender(params.Conn) - res.Receiver = NewLengthPrefixTCPReceiver(params.Conn) - - case 1: - if pubKey == nil { - return ErrExpectedNoCrypto - } - - rbuf = make([]byte, len(pubKey)) - if _, err := io.ReadFull(params.Conn, rbuf); err != nil { - return err - } - - res.Sender = NewLengthPrefixTCPSender(params.Conn) - res.Receiver = NewLengthPrefixTCPReceiver(params.Conn) - res.setupCrypto(params, rbuf, privKey) - - default: - return fmt.Errorf("Bad encryption flag %d", rbuf[0]) - } - - if err := <-writeDone; err != nil { - return err - } - - // Features exchange - go func() { - buf := new(bytes.Buffer) - if err := gob.NewEncoder(buf).Encode(¶ms.Features); err != nil { - writeDone <- err - return - } - - writeDone <- res.Sender.Send(buf.Bytes()) - }() - - rbuf, err := res.Receiver.Receive() - if err != nil { - return err - } - - if err := gob.NewDecoder(bytes.NewReader(rbuf)).Decode(&res.Features); err != nil { - return err - } - - if err := <-writeDone; err != nil { - return err - } - - return nil -} - -func (res *ProtocolIntroResults) setupCrypto(params ProtocolIntroParams, remotePubKey []byte, privKey *[32]byte) { - var remotePubKeyArr [32]byte - copy(remotePubKeyArr[:], remotePubKey) - res.SessionKey = FormSessionKey(&remotePubKeyArr, privKey, params.Password) - res.Sender = NewEncryptedTCPSender(res.Sender, res.SessionKey, params.Outbound) - res.Receiver = NewEncryptedTCPReceiver(res.Receiver, res.SessionKey, params.Outbound) -} - -type ProtocolTag byte - -const ( - ProtocolHeartbeat = iota - ProtocolReserved1 // legacy overlay control messages - ProtocolReserved2 // - ProtocolReserved3 // - ProtocolGossip - ProtocolGossipUnicast - ProtocolGossipBroadcast - ProtocolOverlayControlMsg -) - -type ProtocolMsg struct { - tag ProtocolTag - msg []byte -} - -type ProtocolSender interface { - SendProtocolMsg(m ProtocolMsg) error -} diff --git a/mesh/protocol_crypto.go b/mesh/protocol_crypto.go deleted file mode 100644 index 2adb6921..00000000 --- a/mesh/protocol_crypto.go +++ /dev/null @@ -1,180 +0,0 @@ -package mesh - -import ( - "crypto/rand" - "crypto/sha256" - "encoding/binary" - "encoding/gob" - "fmt" - "io" - "sync" - - "golang.org/x/crypto/nacl/box" - "golang.org/x/crypto/nacl/secretbox" -) - -const MaxTCPMsgSize = 10 * 1024 * 1024 - -func GenerateKeyPair() (publicKey, privateKey *[32]byte, err error) { - return box.GenerateKey(rand.Reader) -} - -func FormSessionKey(remotePublicKey, localPrivateKey *[32]byte, secretKey []byte) *[32]byte { - var sharedKey [32]byte - box.Precompute(&sharedKey, remotePublicKey, localPrivateKey) - sharedKeySlice := sharedKey[:] - sharedKeySlice = append(sharedKeySlice, secretKey...) - sessionKey := sha256.Sum256(sharedKeySlice) - return &sessionKey -} - -// TCP Senders/Receivers - -// The lowest 64 bits of the nonce contain the message sequence -// number. The top most bit indicates the connection polarity at the -// sender - '1' for outbound; the next indicates protocol type - '1' -// for TCP. The remaining 126 bits are zero. The polarity is needed so -// that the two ends of a connection do not use the same nonces; the -// protocol type so that the TCP connection nonces are distinct from -// nonces used by overlay connections, if they share the session key. -// This is a requirement of the NaCl Security Model; see -// http://nacl.cr.yp.to/box.html. -type TCPCryptoState struct { - sessionKey *[32]byte - nonce [24]byte - seqNo uint64 -} - -func NewTCPCryptoState(sessionKey *[32]byte, outbound bool) *TCPCryptoState { - s := &TCPCryptoState{sessionKey: sessionKey} - if outbound { - s.nonce[0] |= (1 << 7) - } - s.nonce[0] |= (1 << 6) - return s -} - -func (s *TCPCryptoState) advance() { - s.seqNo++ - binary.BigEndian.PutUint64(s.nonce[16:24], s.seqNo) -} - -type TCPSender interface { - Send([]byte) error -} - -type GobTCPSender struct { - encoder *gob.Encoder -} - -type LengthPrefixTCPSender struct { - writer io.Writer -} - -type EncryptedTCPSender struct { - sync.RWMutex - sender TCPSender - state *TCPCryptoState -} - -func NewGobTCPSender(encoder *gob.Encoder) *GobTCPSender { - return &GobTCPSender{encoder: encoder} -} - -func (sender *GobTCPSender) Send(msg []byte) error { - return sender.encoder.Encode(msg) -} - -func NewLengthPrefixTCPSender(writer io.Writer) *LengthPrefixTCPSender { - return &LengthPrefixTCPSender{writer: writer} -} - -func (sender *LengthPrefixTCPSender) Send(msg []byte) error { - l := len(msg) - if l > MaxTCPMsgSize { - return fmt.Errorf("outgoing message exceeds maximum size: %d > %d", l, MaxTCPMsgSize) - } - // We copy the message so we can send it in a single Write - // operation, thus making this thread-safe without locking. - prefixedMsg := make([]byte, 4+l) - binary.BigEndian.PutUint32(prefixedMsg, uint32(l)) - copy(prefixedMsg[4:], msg) - _, err := sender.writer.Write(prefixedMsg) - return err -} - -func NewEncryptedTCPSender(sender TCPSender, sessionKey *[32]byte, outbound bool) *EncryptedTCPSender { - return &EncryptedTCPSender{sender: sender, state: NewTCPCryptoState(sessionKey, outbound)} -} - -func (sender *EncryptedTCPSender) Send(msg []byte) error { - sender.Lock() - defer sender.Unlock() - encodedMsg := secretbox.Seal(nil, msg, &sender.state.nonce, sender.state.sessionKey) - sender.state.advance() - return sender.sender.Send(encodedMsg) -} - -type TCPReceiver interface { - Receive() ([]byte, error) -} - -type GobTCPReceiver struct { - decoder *gob.Decoder -} - -type LengthPrefixTCPReceiver struct { - reader io.Reader -} - -type EncryptedTCPReceiver struct { - receiver TCPReceiver - state *TCPCryptoState -} - -func NewGobTCPReceiver(decoder *gob.Decoder) *GobTCPReceiver { - return &GobTCPReceiver{decoder: decoder} -} - -func (receiver *GobTCPReceiver) Receive() ([]byte, error) { - var msg []byte - err := receiver.decoder.Decode(&msg) - return msg, err -} - -func NewLengthPrefixTCPReceiver(reader io.Reader) *LengthPrefixTCPReceiver { - return &LengthPrefixTCPReceiver{reader: reader} -} - -func (receiver *LengthPrefixTCPReceiver) Receive() ([]byte, error) { - lenPrefix := make([]byte, 4) - if _, err := io.ReadFull(receiver.reader, lenPrefix); err != nil { - return nil, err - } - l := binary.BigEndian.Uint32(lenPrefix) - if l > MaxTCPMsgSize { - return nil, fmt.Errorf("incoming message exceeds maximum size: %d > %d", l, MaxTCPMsgSize) - } - msg := make([]byte, l) - _, err := io.ReadFull(receiver.reader, msg) - return msg, err -} - -func NewEncryptedTCPReceiver(receiver TCPReceiver, sessionKey *[32]byte, outbound bool) *EncryptedTCPReceiver { - return &EncryptedTCPReceiver{receiver: receiver, state: NewTCPCryptoState(sessionKey, !outbound)} -} - -func (receiver *EncryptedTCPReceiver) Receive() ([]byte, error) { - msg, err := receiver.receiver.Receive() - if err != nil { - return nil, err - } - - decodedMsg, success := secretbox.Open(nil, msg, &receiver.state.nonce, receiver.state.sessionKey) - if !success { - return nil, fmt.Errorf("Unable to decrypt TCP msg") - } - - receiver.state.advance() - return decodedMsg, nil -} diff --git a/mesh/protocol_test.go b/mesh/protocol_test.go deleted file mode 100644 index 1ea2ec41..00000000 --- a/mesh/protocol_test.go +++ /dev/null @@ -1,96 +0,0 @@ -package mesh - -import ( - "io" - "testing" - "time" - - "github.com/stretchr/testify/require" -) - -type testConn struct { - io.Writer - io.Reader -} - -func (testConn) SetDeadline(t time.Time) error { - return nil -} - -func (testConn) SetReadDeadline(t time.Time) error { - return nil -} - -func (testConn) SetWriteDeadline(t time.Time) error { - return nil -} - -func connPair() (ProtocolIntroConn, ProtocolIntroConn) { - a := testConn{} - b := testConn{} - a.Reader, b.Writer = io.Pipe() - b.Reader, a.Writer = io.Pipe() - return &a, &b -} - -func doIntro(t *testing.T, params ProtocolIntroParams) <-chan ProtocolIntroResults { - ch := make(chan ProtocolIntroResults, 1) - go func() { - res, err := params.DoIntro() - require.Nil(t, err) - ch <- res - }() - return ch -} - -func doProtocolIntro(t *testing.T, aver, bver byte, password []byte) byte { - aconn, bconn := connPair() - aresch := doIntro(t, ProtocolIntroParams{ - MinVersion: ProtocolMinVersion, - MaxVersion: aver, - Features: map[string]string{"Name": "A"}, - Conn: aconn, - Outbound: true, - Password: password, - }) - bresch := doIntro(t, ProtocolIntroParams{ - MinVersion: ProtocolMinVersion, - MaxVersion: bver, - Features: map[string]string{"Name": "B"}, - Conn: bconn, - Outbound: false, - Password: password, - }) - ares := <-aresch - bres := <-bresch - - // Check that features were conveyed - require.Equal(t, "B", ares.Features["Name"]) - require.Equal(t, "A", bres.Features["Name"]) - - // Check that Senders and Receivers work - go func() { - require.Nil(t, ares.Sender.Send([]byte("Hello from A"))) - require.Nil(t, bres.Sender.Send([]byte("Hello from B"))) - }() - - data, err := bres.Receiver.Receive() - require.Nil(t, err) - require.Equal(t, "Hello from A", string(data)) - - data, err = ares.Receiver.Receive() - require.Nil(t, err) - require.Equal(t, "Hello from B", string(data)) - - require.Equal(t, ares.Version, bres.Version) - return ares.Version -} - -func TestProtocolIntro(t *testing.T) { - require.Equal(t, 2, int(doProtocolIntro(t, 2, 2, nil))) - require.Equal(t, 2, int(doProtocolIntro(t, 2, 2, []byte("sekr1t")))) - require.Equal(t, 1, int(doProtocolIntro(t, 1, 2, nil))) - require.Equal(t, 1, int(doProtocolIntro(t, 1, 2, []byte("pa55")))) - require.Equal(t, 1, int(doProtocolIntro(t, 2, 1, nil))) - require.Equal(t, 1, int(doProtocolIntro(t, 2, 1, []byte("w0rd")))) -} diff --git a/mesh/router.go b/mesh/router.go deleted file mode 100644 index 598e485d..00000000 --- a/mesh/router.go +++ /dev/null @@ -1,198 +0,0 @@ -package mesh - -import ( - "fmt" - "math" - "net" - "sync" - "time" - - "github.com/weaveworks/weave/common" -) - -const ( - Port = 6783 - ChannelSize = 16 - TCPHeartbeat = 30 * time.Second - GossipInterval = 30 * time.Second - MaxDuration = time.Duration(math.MaxInt64) - - acceptMaxTokens = 100 // [1] - acceptTokenDelay = 100 * time.Millisecond // [2] -) - -// [1] capacity of token bucket for rate limiting accepts - -// [2] control rate at which new tokens are added to the bucket - -var ( - log = common.Log - checkFatal = common.CheckFatal - checkWarn = common.CheckWarn -) - -type Config struct { - Port int - ProtocolMinVersion byte - Password []byte - ConnLimit int - PeerDiscovery bool - TrustedSubnets []*net.IPNet -} - -type Router struct { - Config - Overlay Overlay - Ourself *LocalPeer - Peers *Peers - Routes *Routes - ConnectionMaker *ConnectionMaker - gossipLock sync.RWMutex - gossipChannels GossipChannels - TopologyGossip Gossip - acceptLimiter *TokenBucket -} - -func NewRouter(config Config, name PeerName, nickName string, overlay Overlay) *Router { - router := &Router{Config: config, gossipChannels: make(GossipChannels)} - - if overlay == nil { - overlay = NullOverlay{} - } - - router.Overlay = overlay - router.Ourself = NewLocalPeer(name, nickName, router) - router.Peers = NewPeers(router.Ourself) - router.Peers.OnGC(func(peer *Peer) { - log.Println("Removed unreachable peer", peer) - }) - router.Routes = NewRoutes(router.Ourself, router.Peers) - router.ConnectionMaker = NewConnectionMaker(router.Ourself, router.Peers, router.Port, router.PeerDiscovery) - router.TopologyGossip = router.NewGossip("topology", router) - router.acceptLimiter = NewTokenBucket(acceptMaxTokens, acceptTokenDelay) - - return router -} - -// Start listening for TCP connections. This is separate from -// NewRouter so that gossipers can register before we start forming -// connections. -func (router *Router) Start() { - router.listenTCP(router.Port) -} - -func (router *Router) Stop() error { - // TODO: perform graceful shutdown... - return nil -} - -func (router *Router) UsingPassword() bool { - return router.Password != nil -} - -func (router *Router) listenTCP(localPort int) { - localAddr, err := net.ResolveTCPAddr("tcp4", fmt.Sprint(":", localPort)) - checkFatal(err) - ln, err := net.ListenTCP("tcp4", localAddr) - checkFatal(err) - go func() { - defer ln.Close() - for { - tcpConn, err := ln.AcceptTCP() - if err != nil { - log.Errorln(err) - continue - } - router.acceptTCP(tcpConn) - router.acceptLimiter.Wait() - } - }() -} - -func (router *Router) acceptTCP(tcpConn *net.TCPConn) { - remoteAddrStr := tcpConn.RemoteAddr().String() - log.Printf("->[%s] connection accepted", remoteAddrStr) - connRemote := NewRemoteConnection(router.Ourself.Peer, nil, remoteAddrStr, false, false) - StartLocalConnection(connRemote, tcpConn, router, true) -} - -// Gossiper methods - the Router is the topology Gossiper - -type TopologyGossipData struct { - peers *Peers - update PeerNameSet -} - -func (d *TopologyGossipData) Merge(other GossipData) GossipData { - names := make(PeerNameSet) - for name := range d.update { - names[name] = void - } - for name := range other.(*TopologyGossipData).update { - names[name] = void - } - return &TopologyGossipData{peers: d.peers, update: names} -} - -func (d *TopologyGossipData) Encode() [][]byte { - return [][]byte{d.peers.EncodePeers(d.update)} -} - -func (router *Router) BroadcastTopologyUpdate(update []*Peer) { - names := make(PeerNameSet) - for _, p := range update { - names[p.Name] = void - } - router.TopologyGossip.GossipBroadcast( - &TopologyGossipData{peers: router.Peers, update: names}) -} - -func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error { - return fmt.Errorf("unexpected topology gossip unicast: %v", msg) -} - -func (router *Router) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) { - origUpdate, _, err := router.applyTopologyUpdate(update) - if err != nil || len(origUpdate) == 0 { - return nil, err - } - return &TopologyGossipData{peers: router.Peers, update: origUpdate}, nil -} - -func (router *Router) Gossip() GossipData { - return &TopologyGossipData{peers: router.Peers, update: router.Peers.Names()} -} - -func (router *Router) OnGossip(update []byte) (GossipData, error) { - _, newUpdate, err := router.applyTopologyUpdate(update) - if err != nil || len(newUpdate) == 0 { - return nil, err - } - return &TopologyGossipData{peers: router.Peers, update: newUpdate}, nil -} - -func (router *Router) applyTopologyUpdate(update []byte) (PeerNameSet, PeerNameSet, error) { - origUpdate, newUpdate, err := router.Peers.ApplyUpdate(update) - if err != nil { - return nil, nil, err - } - if len(newUpdate) > 0 { - router.ConnectionMaker.Refresh() - router.Routes.Recalculate() - } - return origUpdate, newUpdate, nil -} - -func (router *Router) Trusts(remote *RemoteConnection) bool { - if tcpAddr, err := net.ResolveTCPAddr("tcp4", remote.remoteTCPAddr); err == nil { - for _, trustedSubnet := range router.TrustedSubnets { - if trustedSubnet.Contains(tcpAddr.IP) { - return true - } - } - } else { - // Should not happen as remoteTCPAddr was obtained from TCPConn - log.Errorf("Unable to parse remote TCP addr: %s", err) - } - return false -} diff --git a/mesh/routes.go b/mesh/routes.go deleted file mode 100644 index 3c5df239..00000000 --- a/mesh/routes.go +++ /dev/null @@ -1,252 +0,0 @@ -package mesh - -import ( - "math" - "sync" -) - -type unicastRoutes map[PeerName]PeerName -type broadcastRoutes map[PeerName][]PeerName - -type Routes struct { - sync.RWMutex - ourself *LocalPeer - peers *Peers - onChange []func() - unicast unicastRoutes - unicastAll unicastRoutes // [1] - broadcast broadcastRoutes - broadcastAll broadcastRoutes // [1] - recalculate chan<- *struct{} - wait chan<- chan struct{} - action chan<- func() - // [1] based on *all* connections, not just established & - // symmetric ones -} - -func NewRoutes(ourself *LocalPeer, peers *Peers) *Routes { - recalculate := make(chan *struct{}, 1) - wait := make(chan chan struct{}) - action := make(chan func()) - routes := &Routes{ - ourself: ourself, - peers: peers, - unicast: make(unicastRoutes), - unicastAll: make(unicastRoutes), - broadcast: make(broadcastRoutes), - broadcastAll: make(broadcastRoutes), - recalculate: recalculate, - wait: wait, - action: action} - routes.unicast[ourself.Name] = UnknownPeerName - routes.unicastAll[ourself.Name] = UnknownPeerName - routes.broadcast[ourself.Name] = []PeerName{} - routes.broadcastAll[ourself.Name] = []PeerName{} - go routes.run(recalculate, wait, action) - return routes -} - -func (routes *Routes) OnChange(callback func()) { - routes.Lock() - defer routes.Unlock() - routes.onChange = append(routes.onChange, callback) -} - -func (routes *Routes) PeerNames() PeerNameSet { - return routes.peers.Names() -} - -func (routes *Routes) Unicast(name PeerName) (PeerName, bool) { - routes.RLock() - defer routes.RUnlock() - hop, found := routes.unicast[name] - return hop, found -} - -func (routes *Routes) UnicastAll(name PeerName) (PeerName, bool) { - routes.RLock() - defer routes.RUnlock() - hop, found := routes.unicastAll[name] - return hop, found -} - -func (routes *Routes) Broadcast(name PeerName) []PeerName { - return routes.lookupOrCalculate(name, &routes.broadcast, true) -} - -func (routes *Routes) BroadcastAll(name PeerName) []PeerName { - return routes.lookupOrCalculate(name, &routes.broadcastAll, false) -} - -func (routes *Routes) lookupOrCalculate(name PeerName, broadcast *broadcastRoutes, establishedAndSymmetric bool) []PeerName { - routes.RLock() - hops, found := (*broadcast)[name] - routes.RUnlock() - if found { - return hops - } - res := make(chan []PeerName) - routes.action <- func() { - routes.RLock() - hops, found := (*broadcast)[name] - routes.RUnlock() - if found { - res <- hops - return - } - routes.peers.RLock() - routes.ourself.RLock() - hops = routes.calculateBroadcast(name, establishedAndSymmetric) - routes.ourself.RUnlock() - routes.peers.RUnlock() - res <- hops - routes.Lock() - (*broadcast)[name] = hops - routes.Unlock() - } - return <-res -} - -// Choose min(log2(n_peers), n_neighbouring_peers) neighbours, with a -// random distribution that is topology-sensitive, favouring -// neighbours at the end of "bottleneck links". We determine the -// latter based on the unicast routing table. If a neighbour appears -// as the value more frequently than others - meaning that we reach a -// higher proportion of peers via that neighbour than other neighbours -// - then it is chosen with a higher probability. -// -// Note that we choose log2(n_peers) *neighbours*, not -// peers. Consequently, on sparsely connected peers this function -// returns a higher proportion of neighbours than elsewhere. In -// extremis, on peers with fewer than log2(n_peers) neighbours, all -// neighbours are returned. -func (routes *Routes) RandomNeighbours(except PeerName) []PeerName { - destinations := make(PeerNameSet) - routes.RLock() - defer routes.RUnlock() - count := int(math.Log2(float64(len(routes.unicastAll)))) - // depends on go's random map iteration - for _, dst := range routes.unicastAll { - if dst != UnknownPeerName && dst != except { - destinations[dst] = void - if len(destinations) >= count { - break - } - } - } - res := make([]PeerName, 0, len(destinations)) - for dst := range destinations { - res = append(res, dst) - } - return res -} - -// Request recalculation of the routing table. This is async but can -// effectively be made synchronous with a subsequent call to -// EnsureRecalculated. -func (routes *Routes) Recalculate() { - // The use of a 1-capacity channel in combination with the - // non-blocking send is an optimisation that results in multiple - // requests being coalesced. - select { - case routes.recalculate <- nil: - default: - } -} - -// Wait for any preceding Recalculate requests to be processed. -func (routes *Routes) EnsureRecalculated() { - done := make(chan struct{}) - routes.wait <- done - <-done -} - -func (routes *Routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}, action <-chan func()) { - for { - select { - case <-recalculate: - routes.calculate() - case done := <-wait: - select { - case <-recalculate: - routes.calculate() - default: - } - close(done) - case f := <-action: - f() - } - } -} - -func (routes *Routes) calculate() { - routes.peers.RLock() - routes.ourself.RLock() - var ( - unicast = routes.calculateUnicast(true) - unicastAll = routes.calculateUnicast(false) - broadcast = make(broadcastRoutes) - broadcastAll = make(broadcastRoutes) - ) - broadcast[routes.ourself.Name] = routes.calculateBroadcast(routes.ourself.Name, true) - broadcastAll[routes.ourself.Name] = routes.calculateBroadcast(routes.ourself.Name, false) - routes.ourself.RUnlock() - routes.peers.RUnlock() - - routes.Lock() - routes.unicast = unicast - routes.unicastAll = unicastAll - routes.broadcast = broadcast - routes.broadcastAll = broadcastAll - onChange := routes.onChange - routes.Unlock() - - for _, callback := range onChange { - callback() - } -} - -// Calculate all the routes for the question: if *we* want to send a -// packet to Peer X, what is the next hop? -// -// When we sniff a packet, we determine the destination peer -// ourself. Consequently, we can relay the packet via any -// arbitrary peers - the intermediate peers do not have to have -// any knowledge of the MAC address at all. Thus there's no need -// to exchange knowledge of MAC addresses, nor any constraints on -// the routes that we construct. -func (routes *Routes) calculateUnicast(establishedAndSymmetric bool) unicastRoutes { - _, unicast := routes.ourself.Routes(nil, establishedAndSymmetric) - return unicast -} - -// Calculate the route to answer the question: if we receive a -// broadcast originally from Peer X, which peers should we pass the -// frames on to? -// -// When the topology is stable, and thus all peers perform route -// calculations based on the same data, the algorithm ensures that -// broadcasts reach every peer exactly once. -// -// This is largely due to properties of the Peer.Routes algorithm. In -// particular: -// -// ForAll X,Y,Z in Peers. -// X.Routes(Y) <= X.Routes(Z) \/ -// X.Routes(Z) <= X.Routes(Y) -// ForAll X,Y,Z in Peers. -// Y =/= Z /\ X.Routes(Y) <= X.Routes(Z) => -// X.Routes(Y) u [P | Y.HasSymmetricConnectionTo(P)] <= X.Routes(Z) -// where <= is the subset relationship on keys of the returned map. -func (routes *Routes) calculateBroadcast(name PeerName, establishedAndSymmetric bool) []PeerName { - hops := []PeerName{} - peer, found := routes.peers.byName[name] - if !found { - return hops - } - if found, reached := peer.Routes(routes.ourself.Peer, establishedAndSymmetric); found { - routes.ourself.ForEachConnectedPeer(establishedAndSymmetric, reached, - func(remotePeer *Peer) { hops = append(hops, remotePeer.Name) }) - } - return hops -} diff --git a/mesh/status.go b/mesh/status.go deleted file mode 100644 index 559dae66..00000000 --- a/mesh/status.go +++ /dev/null @@ -1,211 +0,0 @@ -package mesh - -import ( - "fmt" - "net" -) - -type Status struct { - Protocol string - ProtocolMinVersion int - ProtocolMaxVersion int - Encryption bool - PeerDiscovery bool - Name string - NickName string - Port int - Peers []PeerStatus - UnicastRoutes []UnicastRouteStatus - BroadcastRoutes []BroadcastRouteStatus - Connections []LocalConnectionStatus - Targets []string - OverlayDiagnostics interface{} - TrustedSubnets []string -} - -type PeerStatus struct { - Name string - NickName string - UID PeerUID - ShortID PeerShortID - Version uint64 - Connections []ConnectionStatus -} - -type ConnectionStatus struct { - Name string - NickName string - Address string - Outbound bool - Established bool -} - -type UnicastRouteStatus struct { - Dest, Via string -} - -type BroadcastRouteStatus struct { - Source string - Via []string -} - -type LocalConnectionStatus struct { - Address string - Outbound bool - State string - Info string -} - -func NewStatus(router *Router) *Status { - return &Status{ - Protocol, - ProtocolMinVersion, - ProtocolMaxVersion, - router.UsingPassword(), - router.PeerDiscovery, - router.Ourself.Name.String(), - router.Ourself.NickName, - router.Port, - NewPeerStatusSlice(router.Peers), - NewUnicastRouteStatusSlice(router.Routes), - NewBroadcastRouteStatusSlice(router.Routes), - NewLocalConnectionStatusSlice(router.ConnectionMaker), - NewTargetSlice(router.ConnectionMaker), - router.Overlay.Diagnostics(), - NewTrustedSubnetsSlice(router.TrustedSubnets)} -} - -func NewPeerStatusSlice(peers *Peers) []PeerStatus { - var slice []PeerStatus - - peers.ForEach(func(peer *Peer) { - var connections []ConnectionStatus - if peer == peers.ourself.Peer { - for conn := range peers.ourself.Connections() { - connections = append(connections, newConnectionStatus(conn)) - } - } else { - // Modifying peer.connections requires a write lock on - // Peers, and since we are holding a read lock (due to the - // ForEach), access without locking the peer is safe. - for _, conn := range peer.connections { - connections = append(connections, newConnectionStatus(conn)) - } - } - slice = append(slice, PeerStatus{ - peer.Name.String(), - peer.NickName, - peer.UID, - peer.ShortID, - peer.Version, - connections}) - }) - - return slice -} - -func newConnectionStatus(c Connection) ConnectionStatus { - return ConnectionStatus{ - c.Remote().Name.String(), - c.Remote().NickName, - c.RemoteTCPAddr(), - c.Outbound(), - c.Established()} -} - -func NewUnicastRouteStatusSlice(routes *Routes) []UnicastRouteStatus { - routes.RLock() - defer routes.RUnlock() - - var slice []UnicastRouteStatus - for dest, via := range routes.unicast { - slice = append(slice, UnicastRouteStatus{dest.String(), via.String()}) - } - return slice -} - -func NewBroadcastRouteStatusSlice(routes *Routes) []BroadcastRouteStatus { - routes.RLock() - defer routes.RUnlock() - - var slice []BroadcastRouteStatus - for source, via := range routes.broadcast { - var hops []string - for _, hop := range via { - hops = append(hops, hop.String()) - } - slice = append(slice, BroadcastRouteStatus{source.String(), hops}) - } - return slice -} - -func NewLocalConnectionStatusSlice(cm *ConnectionMaker) []LocalConnectionStatus { - resultChan := make(chan []LocalConnectionStatus, 0) - cm.actionChan <- func() bool { - var slice []LocalConnectionStatus - for conn := range cm.connections { - state := "pending" - if conn.Established() { - state = "established" - } - lc, _ := conn.(*LocalConnection) - info := fmt.Sprintf("%-6v %v", lc.OverlayConn.DisplayName(), conn.Remote()) - if lc.Router.UsingPassword() { - if lc.Untrusted() { - info = fmt.Sprintf("%-11v %v", "encrypted", info) - } else { - info = fmt.Sprintf("%-11v %v", "unencrypted", info) - } - } - slice = append(slice, LocalConnectionStatus{conn.RemoteTCPAddr(), conn.Outbound(), state, info}) - } - for address, target := range cm.targets { - add := func(state, info string) { - slice = append(slice, LocalConnectionStatus{address, true, state, info}) - } - switch target.state { - case TargetWaiting: - until := "never" - if !target.tryAfter.IsZero() { - until = target.tryAfter.String() - } - if target.lastError == nil { // shouldn't happen - add("waiting", "until: "+until) - } else { - add("failed", target.lastError.Error()+", retry: "+until) - } - case TargetAttempting: - if target.lastError == nil { - add("connecting", "") - } else { - add("retrying", target.lastError.Error()) - } - case TargetConnected: - } - } - resultChan <- slice - return false - } - return <-resultChan -} - -func NewTargetSlice(cm *ConnectionMaker) []string { - resultChan := make(chan []string, 0) - cm.actionChan <- func() bool { - var slice []string - for peer := range cm.directPeers { - slice = append(slice, peer) - } - resultChan <- slice - return false - } - return <-resultChan -} - -func NewTrustedSubnetsSlice(trustedSubnets []*net.IPNet) []string { - trustedSubnetStrs := []string{} - for _, trustedSubnet := range trustedSubnets { - trustedSubnetStrs = append(trustedSubnetStrs, trustedSubnet.String()) - } - return trustedSubnetStrs -} diff --git a/mesh/surrogate_gossiper.go b/mesh/surrogate_gossiper.go deleted file mode 100644 index 7204d4aa..00000000 --- a/mesh/surrogate_gossiper.go +++ /dev/null @@ -1,45 +0,0 @@ -package mesh - -type SurrogateGossipData struct { - messages [][]byte -} - -func NewSurrogateGossipData(msg []byte) *SurrogateGossipData { - return &SurrogateGossipData{messages: [][]byte{msg}} -} - -func (d *SurrogateGossipData) Encode() [][]byte { - return d.messages -} - -func (d *SurrogateGossipData) Merge(other GossipData) GossipData { - o := other.(*SurrogateGossipData) - messages := make([][]byte, 0, len(d.messages)+len(o.messages)) - messages = append(messages, d.messages...) - messages = append(messages, o.messages...) - return &SurrogateGossipData{messages: messages} -} - -// SurrogateGossiper ignores unicasts and relays broadcasts and gossips. - -type SurrogateGossiper struct{} - -func (*SurrogateGossiper) OnGossipUnicast(sender PeerName, msg []byte) error { - return nil -} - -func (*SurrogateGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) { - return NewSurrogateGossipData(update), nil -} - -func (*SurrogateGossiper) Gossip() GossipData { - return nil -} - -func (*SurrogateGossiper) OnGossip(update []byte) (GossipData, error) { - return NewSurrogateGossipData(update), nil -} - -var ( - surrogateGossiper SurrogateGossiper -) diff --git a/mesh/token_bucket.go b/mesh/token_bucket.go deleted file mode 100644 index e6daeb3b..00000000 --- a/mesh/token_bucket.go +++ /dev/null @@ -1,42 +0,0 @@ -package mesh - -import ( - "time" -) - -type TokenBucket struct { - capacity int64 // Maximum capacity of bucket - tokenInterval time.Duration // Token replenishment rate - refillDuration time.Duration // Time to refill from empty - earliestUnspentToken time.Time -} - -func NewTokenBucket(capacity int64, tokenInterval time.Duration) *TokenBucket { - tb := TokenBucket{ - capacity: capacity, - tokenInterval: tokenInterval, - refillDuration: tokenInterval * time.Duration(capacity)} - - tb.earliestUnspentToken = tb.capacityToken() - - return &tb -} - -func (tb *TokenBucket) Wait() { - // If earliest unspent token is in the future, sleep until then - time.Sleep(tb.earliestUnspentToken.Sub(time.Now())) - - // Alternatively, enforce bucket capacity if necessary - capacityToken := tb.capacityToken() - if tb.earliestUnspentToken.Before(capacityToken) { - tb.earliestUnspentToken = capacityToken - } - - // 'Remove' a token from the bucket - tb.earliestUnspentToken = tb.earliestUnspentToken.Add(tb.tokenInterval) -} - -// Determine the historic token timestamp representing a full bucket -func (tb *TokenBucket) capacityToken() time.Time { - return time.Now().Add(-tb.refillDuration).Truncate(tb.tokenInterval) -}