Update weaveworks/mesh to v0.3

2019-10-15 09:01:00 +00:00
1 parent cb4a8769a8
commit 04680fc52c
@@ -686,12 +686,12 @@
  revision = "6b0aa22550d9325eb8f43418185859e13dc0de1d"

 [[projects]]
-  digest = "1:277e090a58afaeb00fb9fd04e2b2f3466ff8fb327a331e0e531a3399a72697e3"
+  digest = "1:ef8d0a1f90a33a968984ba2159a02304c591bbe964ad13e6408bf361975b3a58"
  name = "github.com/weaveworks/mesh"
  packages = ["."]
  pruneopts = "UT"
-  revision = "38e8100dfbc7b28ae8ffde5a2bbeceec57ffa52a"
-  version = "v0.1"
+  revision = "8889a805f7f6fe19a40dd175575d7f5efb7eda79"
+  version = "v0.3"

 [[projects]]
  branch = "master"
@@ -77,3 +77,12 @@ All contributions should be made as pull requests that satisfy the guidelines, b
 In addition, several mechanical checks are enforced.
 See [the lint script](/lint) for details.

+## <a name="help"></a>Getting Help
+
+If you have any questions about, feedback for or problems with `mesh`:
+
+- Invite yourself to the <a href="https://slack.weave.works/" target="_blank">Weave Users Slack</a>.
+- Ask a question on the [#general](https://weave-community.slack.com/messages/general/) slack channel.
+- [File an issue](https://github.com/weaveworks/mesh/issues/new).
+
+Your feedback is always welcome!
@@ -71,7 +71,6 @@ type LocalConnection struct {
 	heartbeatTCP    *time.Ticker
 	router          *Router
 	uid             uint64
-	actionChan      chan<- connectionAction
 	errorChan       chan<- error
 	finished        <-chan struct{} // closed to signal that actorLoop has finished
 	senders         *gossipSenders
@@ -84,7 +83,6 @@ func startLocalConnection(connRemote *remoteConnection, tcpConn *net.TCPConn, ro
 	if connRemote.local != router.Ourself.Peer {
 		panic("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{
@@ -93,13 +91,12 @@ func startLocalConnection(connRemote *remoteConnection, tcpConn *net.TCPConn, ro
 		tcpConn:          tcpConn,
 		trustRemote:      router.trusts(connRemote),
 		uid:              randUint64(),
-		actionChan:       actionChan,
 		errorChan:        errorChan,
 		finished:         finished,
 		logger:           logger,
 	}
 	conn.senders = newGossipSenders(conn, finished)
-	go conn.run(actionChan, errorChan, finished, acceptNewPeer)
+	go conn.run(errorChan, finished, acceptNewPeer)
 }

 func (conn *LocalConnection) logf(format string, args ...interface{}) {
@@ -158,18 +155,9 @@ func (conn *LocalConnection) shutdown(err error) {
 	}
 }

-// 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) {
+func (conn *LocalConnection) run(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.teardown(err) }()
 	defer close(finished)
@@ -265,7 +253,7 @@ func (conn *LocalConnection) run(actionChan <-chan connectionAction, errorChan <
 	// 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)
+	err = conn.actorLoop(errorChan)
 }

 func (conn *LocalConnection) makeFeatures() map[string]string {
@@ -354,7 +342,7 @@ func (conn *LocalConnection) registerRemote(remote *Peer, acceptNewPeer bool) er
 	return nil
 }

-func (conn *LocalConnection) actorLoop(actionChan <-chan connectionAction, errorChan <-chan error) (err error) {
+func (conn *LocalConnection) actorLoop(errorChan <-chan error) (err error) {
 	fwdErrorChan := conn.OverlayConn.ErrorChannel()
 	fwdEstablishedChan := conn.OverlayConn.EstablishedChannel()

@@ -364,8 +352,6 @@ func (conn *LocalConnection) actorLoop(actionChan <-chan connectionAction, error
 		case err = <-fwdErrorChan:
 		default:
 			select {
-			case action := <-actionChan:
-				err = action()
 			case <-conn.heartbeatTCP.C:
 				err = conn.sendSimpleProtocolMsg(ProtocolHeartbeat)
 			case <-fwdEstablishedChan:
@@ -486,10 +472,6 @@ func (err *peerNameCollisionError) Error() string {
 	return fmt.Sprintf("local %q and remote %q peer names collision", err.local, err.remote)
 }

-// The actor closure used by LocalConnection. If an action returns an error,
-// it will terminate the actor loop, which terminates the connection in turn.
-type connectionAction func() error
-
 func mustHave(features map[string]string, keys []string) error {
 	for _, key := range keys {
 		if _, ok := features[key]; !ok {
@@ -93,7 +93,7 @@ func (cm *connectionMaker) InitiateConnections(peers []string, replace bool) []e
 		}
 		if host == "" || !isAlnum(port) {
 			errors = append(errors, fmt.Errorf("invalid peer name %q, should be host[:port]", peer))
-		} else if addr, err := net.ResolveTCPAddr("tcp4", fmt.Sprintf("%s:%s", host, port)); err != nil {
+		} else if addr, err := net.ResolveTCPAddr("tcp", fmt.Sprintf("%s:%s", host, port)); err != nil {
 			errors = append(errors, err)
 		} else {
 			addrs[peer] = addr
@@ -8,13 +8,20 @@ import (
 	"time"
 )

+const (
+	deferTopologyUpdateDuration = 1 * time.Second
+)
+
 // localPeer is the only "active" peer in the mesh. It extends Peer with
 // additional behaviors, mostly to retrieve and manage connection state.
 type localPeer struct {
 	sync.RWMutex
 	*Peer
-	router     *Router
-	actionChan chan<- localPeerAction
+	router                *Router
+	actionChan            chan<- localPeerAction
+	topologyUpdates       peerNameSet
+	timer                 *time.Timer
+	pendingTopologyUpdate bool
 }

 // The actor closure used by localPeer.
@@ -23,11 +30,15 @@ type localPeerAction func()
 // newLocalPeer returns a usable LocalPeer.
 func newLocalPeer(name PeerName, nickName string, router *Router) *localPeer {
 	actionChan := make(chan localPeerAction, ChannelSize)
+	topologyUpdates := make(peerNameSet)
 	peer := &localPeer{
-		Peer:       newPeer(name, nickName, randomPeerUID(), 0, randomPeerShortID()),
-		router:     router,
-		actionChan: actionChan,
+		Peer:            newPeer(name, nickName, randomPeerUID(), 0, randomPeerShortID()),
+		router:          router,
+		actionChan:      actionChan,
+		topologyUpdates: topologyUpdates,
+		timer:           time.NewTimer(deferTopologyUpdateDuration),
 	}
+	peer.timer.Stop()
 	go peer.actorLoop(actionChan)
 	return peer
 }
@@ -82,15 +93,15 @@ func (peer *localPeer) createConnection(localAddr string, peerAddr string, accep
 	if err := peer.checkConnectionLimit(); err != nil {
 		return err
 	}
-	localTCPAddr, err := net.ResolveTCPAddr("tcp4", localAddr)
+	localTCPAddr, err := net.ResolveTCPAddr("tcp", localAddr)
 	if err != nil {
 		return err
 	}
-	remoteTCPAddr, err := net.ResolveTCPAddr("tcp4", peerAddr)
+	remoteTCPAddr, err := net.ResolveTCPAddr("tcp", peerAddr)
 	if err != nil {
 		return err
 	}
-	tcpConn, err := net.DialTCP("tcp4", localTCPAddr, remoteTCPAddr)
+	tcpConn, err := net.DialTCP("tcp", localTCPAddr, remoteTCPAddr)
 	if err != nil {
 		return err
 	}
@@ -136,6 +147,10 @@ func (peer *localPeer) encode(enc *gob.Encoder) {
 // ACTOR server

 func (peer *localPeer) actorLoop(actionChan <-chan localPeerAction) {
+	gossipInterval := defaultGossipInterval
+	if peer.router != nil {
+		gossipInterval = peer.router.gossipInterval()
+	}
 	gossipTimer := time.Tick(gossipInterval)
 	for {
 		select {
@@ -143,10 +158,22 @@ func (peer *localPeer) actorLoop(actionChan <-chan localPeerAction) {
 			action()
 		case <-gossipTimer:
 			peer.router.sendAllGossip()
+		case <-peer.timer.C:
+			peer.broadcastPendingTopologyUpdates()
 		}
 	}
 }

+func (peer *localPeer) broadcastPendingTopologyUpdates() {
+	peer.Lock()
+	gossipData := peer.topologyUpdates
+	peer.topologyUpdates = make(peerNameSet)
+	peer.pendingTopologyUpdate = false
+	peer.Unlock()
+	gossipData[peer.Peer.Name] = struct{}{}
+	peer.router.broadcastTopologyUpdate(gossipData)
+}
+
 func (peer *localPeer) handleAddConnection(conn ourConnection, isRestartedPeer bool) error {
 	if peer.Peer != conn.getLocal() {
 		panic("Attempt made to add connection to peer where peer is not the source of connection")
@@ -190,7 +217,6 @@ func (peer *localPeer) handleAddConnection(conn ourConnection, isRestartedPeer b
 		conn.logf("connection added (new peer)")
 		peer.router.sendAllGossipDown(conn)
 	}
-
 	peer.router.Routes.recalculate()
 	peer.broadcastPeerUpdate(conn.Remote())

@@ -240,7 +266,15 @@ func (peer *localPeer) broadcastPeerUpdate(peers ...*Peer) {
 	// context of a test, but that will involve significant
 	// reworking of tests.
 	if peer.router != nil {
-		peer.router.broadcastTopologyUpdate(append(peers, peer.Peer))
+		peer.Lock()
+		defer peer.Unlock()
+		if !peer.pendingTopologyUpdate {
+			peer.timer.Reset(deferTopologyUpdateDuration)
+			peer.pendingTopologyUpdate = true
+		}
+		for _, p := range peers {
+			peer.topologyUpdates[p.Name] = struct{}{}
+		}
 	}
 }

@@ -108,7 +108,11 @@ func (NullOverlay) Diagnostics() interface{} { return nil }
 func (NullOverlay) Confirm() {}

 // EstablishedChannel implements OverlayConnection.
-func (NullOverlay) EstablishedChannel() <-chan struct{} { return nil }
+func (NullOverlay) EstablishedChannel() <-chan struct{} {
+	c := make(chan struct{})
+	close(c)
+	return c
+}

 // ErrorChannel implements OverlayConnection.
 func (NullOverlay) ErrorChannel() <-chan error { return nil }
@@ -6,6 +6,11 @@ import (
 	"io"
 	"math/rand"
 	"sync"
+	"time"
+)
+
+const (
+	gcInterval = 1 * time.Second
 )

 // Peers collects all of the known peers in the mesh, including ourself.
@@ -18,6 +23,8 @@ type Peers struct {

 	// Called when the mapping from short IDs to peers changes
 	onInvalidateShortIDs []func()
+	timer                *time.Timer
+	pendingGC            bool
 }

 type shortIDPeers struct {
@@ -60,8 +67,11 @@ func newPeers(ourself *localPeer) *Peers {
 		ourself:   ourself,
 		byName:    make(map[PeerName]*Peer),
 		byShortID: make(map[PeerShortID]shortIDPeers),
+		timer:     time.NewTimer(gcInterval),
 	}
 	peers.fetchWithDefault(ourself.Peer)
+	peers.timer.Stop()
+	go peers.actorLoop()
 	return peers
 }

@@ -339,6 +349,18 @@ func (peers *Peers) forEach(fun func(*Peer)) {
 	}
 }

+func (peers *Peers) actorLoop() {
+	for {
+		select {
+		case <-peers.timer.C:
+			peers.GarbageCollect()
+			peers.Lock()
+			peers.pendingGC = false
+			peers.Unlock()
+		}
+	}
+}
+
 // Merge an incoming update with our own topology.
 //
 // We add peers hitherto unknown to us, and update peers for which the
@@ -363,7 +385,6 @@ func (peers *Peers) applyUpdate(update []byte) (peerNameSet, peerNameSet, error)

 	// Now apply the updates
 	newUpdate := peers.applyDecodedUpdate(decodedUpdate, decodedConns, &pending)
-	peers.garbageCollect(&pending)
 	for _, peerRemoved := range pending.removed {
 		delete(newUpdate, peerRemoved.Name)
 	}
@@ -373,6 +394,13 @@ func (peers *Peers) applyUpdate(update []byte) (peerNameSet, peerNameSet, error)
 		updateNames[peer.Name] = struct{}{}
 	}

+	if !peers.pendingGC {
+		// schedule a GarbageCollect() to run after gcInterval time period
+		// corresponding to all topology updates received during the period
+		peers.timer.Reset(gcInterval)
+		peers.pendingGC = true
+	}
+
 	return updateNames, newUpdate, nil
 }

@@ -54,10 +54,10 @@ type protocolIntroConn interface {
 type protocolIntroParams struct {
 	MinVersion byte
 	MaxVersion byte
+	Outbound   bool
 	Features   map[string]string
 	Conn       protocolIntroConn
 	Password   []byte
-	Outbound   bool
 }

 // The results from a successful protocol intro.
@@ -17,11 +17,12 @@ var (
 	// ChannelSize is the buffer size used by so-called actor goroutines
 	// throughout mesh.
 	ChannelSize = 16
+
+	defaultGossipInterval = 30 * time.Second
 )

 const (
 	tcpHeartbeat     = 30 * time.Second
-	gossipInterval   = 30 * time.Second
 	maxDuration      = time.Duration(math.MaxInt64)
 	acceptMaxTokens  = 100
 	acceptTokenDelay = 100 * time.Millisecond // [2]
@@ -32,11 +33,12 @@ const (
 type Config struct {
 	Host               string
 	Port               int
-	ProtocolMinVersion byte
 	Password           []byte
 	ConnLimit          int
+	ProtocolMinVersion byte
 	PeerDiscovery      bool
 	TrustedSubnets     []*net.IPNet
+	GossipInterval     *time.Duration
 }

 // Router manages communication between this peer and the rest of the mesh.
@@ -99,11 +101,11 @@ func (router *Router) usingPassword() bool {
 }

 func (router *Router) listenTCP() {
-	localAddr, err := net.ResolveTCPAddr("tcp4", net.JoinHostPort(router.Host, fmt.Sprint(router.Port)))
+	localAddr, err := net.ResolveTCPAddr("tcp", net.JoinHostPort(router.Host, fmt.Sprint(router.Port)))
 	if err != nil {
 		panic(err)
 	}
-	ln, err := net.ListenTCP("tcp4", localAddr)
+	ln, err := net.ListenTCP("tcp", localAddr)
 	if err != nil {
 		panic(err)
 	}
@@ -154,7 +156,7 @@ func (router *Router) gossipChannel(channelName string) *gossipChannel {
 	if channel, found = router.gossipChannels[channelName]; found {
 		return channel
 	}
-	channel = newGossipChannel(channelName, router.Ourself, router.Routes, &surrogateGossiper{}, router.logger)
+	channel = newGossipChannel(channelName, router.Ourself, router.Routes, &surrogateGossiper{router: router}, router.logger)
 	channel.logf("created surrogate channel")
 	router.gossipChannels[channelName] = channel
 	return channel
@@ -170,6 +172,14 @@ func (router *Router) gossipChannelSet() map[*gossipChannel]struct{} {
 	return channels
 }

+func (router *Router) gossipInterval() time.Duration {
+	if router.Config.GossipInterval != nil {
+		return *router.Config.GossipInterval
+	} else {
+		return defaultGossipInterval
+	}
+}
+
 func (router *Router) handleGossip(tag protocolTag, payload []byte) error {
 	decoder := gob.NewDecoder(bytes.NewReader(payload))
 	var channelName string
@@ -221,12 +231,9 @@ func (router *Router) sendPendingGossip() bool {

 // BroadcastTopologyUpdate is invoked whenever there is a change to the mesh
 // topology, and broadcasts the new set of peers to the mesh.
-func (router *Router) broadcastTopologyUpdate(update []*Peer) {
-	names := make(peerNameSet)
-	for _, p := range update {
-		names[p.Name] = struct{}{}
-	}
-	router.topologyGossip.GossipBroadcast(&topologyGossipData{peers: router.Peers, update: names})
+func (router *Router) broadcastTopologyUpdate(update peerNameSet) {
+	gossipData := &topologyGossipData{peers: router.Peers, update: update}
+	router.topologyGossip.GossipBroadcast(gossipData)
 }

 // OnGossipUnicast implements Gossiper, but always returns an error, as a
@@ -274,7 +281,7 @@ func (router *Router) applyTopologyUpdate(update []byte) (peerNameSet, peerNameS
 }

 func (router *Router) trusts(remote *remoteConnection) bool {
-	if tcpAddr, err := net.ResolveTCPAddr("tcp4", remote.remoteTCPAddr); err == nil {
+	if tcpAddr, err := net.ResolveTCPAddr("tcp", remote.remoteTCPAddr); err == nil {
 		for _, trustedSubnet := range router.TrustedSubnets {
 			if trustedSubnet.Contains(tcpAddr.IP) {
 				return true
@@ -3,6 +3,7 @@ package mesh
 import (
 	"math"
 	"sync"
+	"time"
 )

 type unicastRoutes map[PeerName]PeerName
@@ -11,23 +12,29 @@ type broadcastRoutes map[PeerName][]PeerName
 // routes aggregates unicast and broadcast routes for our peer.
 type routes struct {
 	sync.RWMutex
-	ourself      *localPeer
-	peers        *Peers
-	onChange     []func()
-	unicast      unicastRoutes
-	unicastAll   unicastRoutes // [1]
-	broadcast    broadcastRoutes
-	broadcastAll broadcastRoutes // [1]
-	recalc       chan<- *struct{}
-	wait         chan<- chan struct{}
-	action       chan<- func()
+	ourself       *localPeer
+	peers         *Peers
+	onChange      []func()
+	unicast       unicastRoutes
+	unicastAll    unicastRoutes // [1]
+	broadcast     broadcastRoutes
+	broadcastAll  broadcastRoutes // [1]
+	recalcTimer   *time.Timer
+	pendingRecalc bool
+	wait          chan chan struct{}
+	action        chan<- func()
 	// [1] based on *all* connections, not just established &
 	// symmetric ones
 }

+const (
+	// We defer recalculation requests by up to 100ms, in order to
+	// coalesce multiple recalcs together.
+	recalcDeferTime = 100 * time.Millisecond
+)
+
 // newRoutes returns a usable Routes based on the LocalPeer and existing Peers.
 func newRoutes(ourself *localPeer, peers *Peers) *routes {
-	recalculate := make(chan *struct{}, 1)
 	wait := make(chan chan struct{})
 	action := make(chan func())
 	r := &routes{
@@ -37,11 +44,12 @@ func newRoutes(ourself *localPeer, peers *Peers) *routes {
 		unicastAll:   unicastRoutes{ourself.Name: UnknownPeerName},
 		broadcast:    broadcastRoutes{ourself.Name: []PeerName{}},
 		broadcastAll: broadcastRoutes{ourself.Name: []PeerName{}},
-		recalc:       recalculate,
+		recalcTimer:  time.NewTimer(time.Hour),
 		wait:         wait,
 		action:       action,
 	}
-	go r.run(recalculate, wait, action)
+	r.recalcTimer.Stop()
+	go r.run(wait, action)
 	return r
 }

@@ -156,32 +164,47 @@ func (r *routes) randomNeighbours(except PeerName) []PeerName {
 // can effectively be made synchronous with a subsequent call to
 // EnsureRecalculated.
 func (r *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 r.recalc <- nil:
-	default:
+	r.Lock()
+	if !r.pendingRecalc {
+		r.recalcTimer.Reset(recalcDeferTime)
+		r.pendingRecalc = true
 	}
+	r.Unlock()
+}
+
+func (r *routes) clearPendingRecalcFlag() {
+	r.Lock()
+	r.pendingRecalc = false
+	r.Unlock()
 }

 // EnsureRecalculated waits for any preceding Recalculate requests to finish.
 func (r *routes) ensureRecalculated() {
-	done := make(chan struct{})
+	var done chan struct{}
+	// If another call is already waiting, wait on the same chan, otherwise make a new one
+	select {
+	case done = <-r.wait:
+	default:
+		done = make(chan struct{})
+	}
 	r.wait <- done
 	<-done
 }

-func (r *routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}, action <-chan func()) {
+func (r *routes) run(wait <-chan chan struct{}, action <-chan func()) {
 	for {
 		select {
-		case <-recalculate:
+		case <-r.recalcTimer.C:
+			r.clearPendingRecalcFlag()
 			r.calculate()
 		case done := <-wait:
-			select {
-			case <-recalculate:
+			r.Lock()
+			pending := r.pendingRecalc
+			r.Unlock()
+			if pending {
+				<-r.recalcTimer.C
+				r.clearPendingRecalcFlag()
 				r.calculate()
-			default:
 			}
 			close(done)
 		case f := <-action:
@@ -190,6 +213,8 @@ func (r *routes) run(recalculate <-chan *struct{}, wait <-chan chan struct{}, ac
 	}
 }

+// Calculate unicast and broadcast routes from r.ourself, and reset
+// the broadcast route cache.
 func (r *routes) calculate() {
 	r.peers.RLock()
 	r.ourself.RLock()
@@ -30,7 +30,7 @@ type Status struct {
 func NewStatus(router *Router) *Status {
 	return &Status{
 		Protocol:           Protocol,
-		ProtocolMinVersion: ProtocolMinVersion,
+		ProtocolMinVersion: int(router.ProtocolMinVersion),
 		ProtocolMaxVersion: ProtocolMaxVersion,
 		Encryption:         router.usingPassword(),
 		PeerDiscovery:      router.PeerDiscovery,
@@ -11,6 +11,7 @@ import (
 type surrogateGossiper struct {
 	sync.Mutex
 	prevUpdates []prevUpdate
+	router      *Router
 }

 type prevUpdate struct {
@@ -56,6 +57,10 @@ func (s *surrogateGossiper) OnGossip(update []byte) (GossipData, error) {
 	// (this time limit is arbitrary; surrogateGossiper should pass on new gossip immediately
 	// so there should be no reason for a duplicate to show up after a long time)
 	updateTime := now()
+	gossipInterval := defaultGossipInterval
+	if s.router != nil {
+		gossipInterval = s.router.gossipInterval()
+	}
 	deleteBefore := updateTime.Add(-gossipInterval)
 	keepFrom := len(s.prevUpdates)
 	for i, p := range s.prevUpdates {