diff --git a/Gopkg.lock b/Gopkg.lock index 171cecbc..6815ec24 100644 --- a/Gopkg.lock +++ b/Gopkg.lock @@ -686,12 +686,12 @@ revision = "6b0aa22550d9325eb8f43418185859e13dc0de1d" [[projects]] - digest = "1:ef8d0a1f90a33a968984ba2159a02304c591bbe964ad13e6408bf361975b3a58" + digest = "1:dfa8458da99855b0f0e13f580e6107b46311db15dc7f76cb80c131b937f7f8aa" name = "github.com/weaveworks/mesh" packages = ["."] pruneopts = "UT" - revision = "8889a805f7f6fe19a40dd175575d7f5efb7eda79" - version = "v0.3" + revision = "58dbcc3e8e63eed7dd77f3a6286855ae8c8876c1" + version = "v0.4" [[projects]] branch = "master" @@ -1108,7 +1108,6 @@ "github.com/weaveworks/mesh", "golang.org/x/crypto/hkdf", "golang.org/x/crypto/nacl/secretbox", - "golang.org/x/net/context", "golang.org/x/sys/unix", "golang.org/x/tools/cover", "k8s.io/api/core/v1", diff --git a/ipam/testutils_test.go b/ipam/testutils_test.go index ccd6efe1..8761d793 100644 --- a/ipam/testutils_test.go +++ b/ipam/testutils_test.go @@ -65,6 +65,9 @@ func (m *mockGossipComms) GossipBroadcast(update mesh.GossipData) { } } +func (m *mockGossipComms) GossipNeighbourSubset(update mesh.GossipData) { + m.GossipBroadcast(update) +} func equalByteBuffer(a, b []byte) bool { if len(a) != len(b) { return false diff --git a/testing/gossip/mocks.go b/testing/gossip/mocks.go index 8cafd89d..2824e83a 100644 --- a/testing/gossip/mocks.go +++ b/testing/gossip/mocks.go @@ -199,3 +199,7 @@ func (client TestRouterClient) GossipUnicast(dstPeerName mesh.PeerName, buf []by func (client TestRouterClient) GossipBroadcast(update mesh.GossipData) { client.router.gossipBroadcast(client.sender, update) } + +func (client TestRouterClient) GossipNeighbourSubset(update mesh.GossipData) { + client.router.gossip(client.sender, update) +} diff --git a/vendor/github.com/weaveworks/mesh/circle.yml b/vendor/github.com/weaveworks/mesh/circle.yml deleted file mode 100644 index 6f4df805..00000000 --- a/vendor/github.com/weaveworks/mesh/circle.yml +++ /dev/null @@ -1,4 +0,0 @@ -test: - pre: - - ./lint - diff --git a/vendor/github.com/weaveworks/mesh/gossip.go b/vendor/github.com/weaveworks/mesh/gossip.go index fe90d133..44bd6ede 100644 --- a/vendor/github.com/weaveworks/mesh/gossip.go +++ b/vendor/github.com/weaveworks/mesh/gossip.go @@ -22,6 +22,9 @@ type Gossip interface { // // TODO(pb): rename to Broadcast? GossipBroadcast(update GossipData) + + // GossipNeighbourSubset emits a message to subset of neighbour peers in the mesh. + GossipNeighbourSubset(update GossipData) } // Gossiper is the receiving interface. diff --git a/vendor/github.com/weaveworks/mesh/gossip_channel.go b/vendor/github.com/weaveworks/mesh/gossip_channel.go index 2ff512e7..d95688b8 100644 --- a/vendor/github.com/weaveworks/mesh/gossip_channel.go +++ b/vendor/github.com/weaveworks/mesh/gossip_channel.go @@ -83,6 +83,12 @@ func (c *gossipChannel) GossipBroadcast(update GossipData) { c.relayBroadcast(c.ourself.Name, update) } +// GossipNeighbourSubset implements Gossip, relaying update to subset of members of the +// channel. +func (c *gossipChannel) GossipNeighbourSubset(update GossipData) { + c.relay(c.ourself.Name, update) +} + // Send relays data into the channel topology via random neighbours. func (c *gossipChannel) Send(data GossipData) { c.relay(c.ourself.Name, data) diff --git a/vendor/github.com/weaveworks/mesh/lint b/vendor/github.com/weaveworks/mesh/lint index ec403c59..5bbb9e84 100755 --- a/vendor/github.com/weaveworks/mesh/lint +++ b/vendor/github.com/weaveworks/mesh/lint @@ -4,19 +4,10 @@ set -o errexit set -o nounset set -o pipefail -if [ ! $(command -v gometalinter) ] +if [ ! $(command -v golangci-lint) ] then - go get github.com/alecthomas/gometalinter - gometalinter --install --vendor + curl -sfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s -- -b $(go env GOPATH)/bin v1.20.0 + golangci-lint --version fi -gometalinter \ - --exclude='error return value not checked.*(Close|Log|Print).*\(errcheck\)$' \ - --exclude='.*_test\.go:.*error return value not checked.*\(errcheck\)$' \ - --exclude='duplicate of.*_test.go.*\(dupl\)$' \ - --disable=aligncheck \ - --disable=gotype \ - --disable=gas \ - --cyclo-over=20 \ - --tests \ - --deadline=20s +golangci-lint run diff --git a/vendor/github.com/weaveworks/mesh/peers.go b/vendor/github.com/weaveworks/mesh/peers.go index 6b174e9a..c10b0318 100644 --- a/vendor/github.com/weaveworks/mesh/peers.go +++ b/vendor/github.com/weaveworks/mesh/peers.go @@ -350,14 +350,11 @@ 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() - } + for range peers.timer.C { + peers.GarbageCollect() + peers.Lock() + peers.pendingGC = false + peers.Unlock() } } diff --git a/vendor/github.com/weaveworks/mesh/router.go b/vendor/github.com/weaveworks/mesh/router.go index 736be318..9af54181 100644 --- a/vendor/github.com/weaveworks/mesh/router.go +++ b/vendor/github.com/weaveworks/mesh/router.go @@ -24,8 +24,8 @@ var ( const ( tcpHeartbeat = 30 * time.Second maxDuration = time.Duration(math.MaxInt64) - acceptMaxTokens = 100 - acceptTokenDelay = 100 * time.Millisecond // [2] + acceptMaxTokens = 20 + acceptTokenDelay = 50 * time.Millisecond ) // Config defines dimensions of configuration for the router. @@ -233,7 +233,7 @@ func (router *Router) sendPendingGossip() bool { // topology, and broadcasts the new set of peers to the mesh. func (router *Router) broadcastTopologyUpdate(update peerNameSet) { gossipData := &topologyGossipData{peers: router.Peers, update: update} - router.topologyGossip.GossipBroadcast(gossipData) + router.topologyGossip.GossipNeighbourSubset(gossipData) } // OnGossipUnicast implements Gossiper, but always returns an error, as a diff --git a/vendor/github.com/weaveworks/mesh/routes.go b/vendor/github.com/weaveworks/mesh/routes.go index 858da783..0e72d0f3 100644 --- a/vendor/github.com/weaveworks/mesh/routes.go +++ b/vendor/github.com/weaveworks/mesh/routes.go @@ -2,6 +2,7 @@ package mesh import ( "math" + "math/rand" "sync" "time" ) @@ -127,7 +128,7 @@ func (r *routes) lookupOrCalculate(name PeerName, broadcast *broadcastRoutes, es return <-res } -// RandomNeighbours chooses min(log2(n_peers), n_neighbouring_peers) +// RandomNeighbours chooses min(2 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 @@ -135,29 +136,39 @@ func (r *routes) lookupOrCalculate(name PeerName, broadcast *broadcastRoutes, es // 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 +// Note that we choose 2log2(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 (r *routes) randomNeighbours(except PeerName) []PeerName { - destinations := make(peerNameSet) r.RLock() defer r.RUnlock() - count := int(math.Log2(float64(len(r.unicastAll)))) - // depends on go's random map iteration + var total int64 = 0 + weights := make(map[PeerName]int64) + // First iterate the whole set, counting how often each neighbour appears for _, dst := range r.unicastAll { if dst != UnknownPeerName && dst != except { - destinations[dst] = struct{}{} - if len(destinations) >= count { - break - } + total++ + weights[dst]++ } } - res := make([]PeerName, 0, len(destinations)) - for dst := range destinations { - res = append(res, dst) + needed := int(math.Min(2*math.Log2(float64(len(r.unicastAll))), float64(len(weights)))) + destinations := make([]PeerName, 0, needed) + for len(destinations) < needed { + // Pick a random point on the distribution and linear search for it + rnd := rand.Int63n(total) + for dst, count := range weights { + if rnd < count { + destinations = append(destinations, dst) + // Remove the one we selected from consideration + delete(weights, dst) + total -= count + break + } + rnd -= count + } } - return res + return destinations } // Recalculate requests recalculation of the routing table. This is async but