mirror of https://github.com/status-im/consul.git
update github.com/hashicorp/{serf,memberlist,go-sockaddr} (#5189)
This activates large-cluster improvements in the gossip layer from https://github.com/hashicorp/memberlist/pull/167
This commit is contained in:
R.B. Boyer
GitHub
parent
d4f7a830a8
commit
b96391ecff
|
|
@ -16,7 +16,7 @@ var (
|
|||
// Centralize all regexps and regexp.Copy() where necessary.
|
||||
signRE *regexp.Regexp = regexp.MustCompile(`^[\s]*[+-]`)
|
||||
whitespaceRE *regexp.Regexp = regexp.MustCompile(`[\s]+`)
|
||||
ifNameRE *regexp.Regexp = regexp.MustCompile(`^Ethernet adapter ([^\s:]+):`)
|
||||
ifNameRE *regexp.Regexp = regexp.MustCompile(`^(?:Ethernet|Wireless LAN) adapter ([^:]+):`)
|
||||
ipAddrRE *regexp.Regexp = regexp.MustCompile(`^ IPv[46] Address\. \. \. \. \. \. \. \. \. \. \. : ([^\s]+)`)
|
||||
)
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
DEPS = $(go list -f '{{range .Imports}}{{.}} {{end}}' ./...)
|
||||
DEPS := $(shell go list -f '{{range .Imports}}{{.}} {{end}}' ./...)
|
||||
|
||||
test: subnet
|
||||
go test ./...
|
||||
|
||||
|
|
@ -13,7 +14,7 @@ cov:
|
|||
open /tmp/coverage.html
|
||||
|
||||
deps:
|
||||
go get -d -v ./...
|
||||
go get -t -d -v ./...
|
||||
echo $(DEPS) | xargs -n1 go get -d
|
||||
|
||||
.PHONY: test cov integ
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
# memberlist [](https://godoc.org/github.com/hashicorp/memberlist)
|
||||
# memberlist [](https://godoc.org/github.com/hashicorp/memberlist) [](https://travis-ci.org/hashicorp/memberlist)
|
||||
|
||||
memberlist is a [Go](http://www.golang.org) library that manages cluster
|
||||
membership and member failure detection using a gossip based protocol.
|
||||
|
|
|
|||
|
|
@ -29,6 +29,11 @@ func (b *memberlistBroadcast) Invalidates(other Broadcast) bool {
|
|||
return b.node == mb.node
|
||||
}
|
||||
|
||||
// memberlist.NamedBroadcast optional interface
|
||||
func (b *memberlistBroadcast) Name() string {
|
||||
return b.node
|
||||
}
|
||||
|
||||
func (b *memberlistBroadcast) Message() []byte {
|
||||
return b.msg
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,24 @@
|
|||
module github.com/hashicorp/memberlist
|
||||
|
||||
require (
|
||||
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da
|
||||
github.com/davecgh/go-spew v1.1.1 // indirect
|
||||
github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c
|
||||
github.com/hashicorp/go-immutable-radix v1.0.0 // indirect
|
||||
github.com/hashicorp/go-msgpack v0.0.0-20150518234257-fa3f63826f7c
|
||||
github.com/hashicorp/go-multierror v1.0.0
|
||||
github.com/hashicorp/go-sockaddr v0.0.0-20190103214136-e92cdb5343bb
|
||||
github.com/kr/pretty v0.1.0 // indirect
|
||||
github.com/miekg/dns v1.0.14
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c // indirect
|
||||
github.com/pmezard/go-difflib v1.0.0 // indirect
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529
|
||||
github.com/stretchr/testify v1.2.2
|
||||
golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3 // indirect
|
||||
golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519 // indirect
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 // indirect
|
||||
golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5 // indirect
|
||||
google.golang.org/appengine v1.4.0 // indirect
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
|
||||
gopkg.in/vmihailenco/msgpack.v2 v2.9.1 // indirect
|
||||
)
|
||||
|
|
@ -0,0 +1,53 @@
|
|||
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da h1:8GUt8eRujhVEGZFFEjBj46YV4rDjvGrNxb0KMWYkL2I=
|
||||
github.com/armon/go-metrics v0.0.0-20180917152333-f0300d1749da/go.mod h1:Q73ZrmVTwzkszR9V5SSuryQ31EELlFMUz1kKyl939pY=
|
||||
github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
|
||||
github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
|
||||
github.com/golang/protobuf v1.2.0 h1:P3YflyNX/ehuJFLhxviNdFxQPkGK5cDcApsge1SqnvM=
|
||||
github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
|
||||
github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c h1:964Od4U6p2jUkFxvCydnIczKteheJEzHRToSGK3Bnlw=
|
||||
github.com/google/btree v0.0.0-20180813153112-4030bb1f1f0c/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ5JPQ=
|
||||
github.com/hashicorp/errwrap v1.0.0 h1:hLrqtEDnRye3+sgx6z4qVLNuviH3MR5aQ0ykNJa/UYA=
|
||||
github.com/hashicorp/errwrap v1.0.0/go.mod h1:YH+1FKiLXxHSkmPseP+kNlulaMuP3n2brvKWEqk/Jc4=
|
||||
github.com/hashicorp/go-immutable-radix v1.0.0 h1:AKDB1HM5PWEA7i4nhcpwOrO2byshxBjXVn/J/3+z5/0=
|
||||
github.com/hashicorp/go-immutable-radix v1.0.0/go.mod h1:0y9vanUI8NX6FsYoO3zeMjhV/C5i9g4Q3DwcSNZ4P60=
|
||||
github.com/hashicorp/go-msgpack v0.0.0-20150518234257-fa3f63826f7c h1:BTAbnbegUIMB6xmQCwWE8yRzbA4XSpnZY5hvRJC188I=
|
||||
github.com/hashicorp/go-msgpack v0.0.0-20150518234257-fa3f63826f7c/go.mod h1:ahLV/dePpqEmjfWmKiqvPkv/twdG7iPBM1vqhUKIvfM=
|
||||
github.com/hashicorp/go-multierror v1.0.0 h1:iVjPR7a6H0tWELX5NxNe7bYopibicUzc7uPribsnS6o=
|
||||
github.com/hashicorp/go-multierror v1.0.0/go.mod h1:dHtQlpGsu+cZNNAkkCN/P3hoUDHhCYQXV3UM06sGGrk=
|
||||
github.com/hashicorp/go-sockaddr v0.0.0-20190103214136-e92cdb5343bb h1:YrwA8w5SBkUIH5BzN2pMYhno+txUCOD5+PVXwLS6ddI=
|
||||
github.com/hashicorp/go-sockaddr v0.0.0-20190103214136-e92cdb5343bb/go.mod h1:7Xibr9yA9JjQq1JpNB2Vw7kxv8xerXegt+ozgdvDeDU=
|
||||
github.com/hashicorp/go-uuid v1.0.0 h1:RS8zrF7PhGwyNPOtxSClXXj9HA8feRnJzgnI1RJCSnM=
|
||||
github.com/hashicorp/go-uuid v1.0.0/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
|
||||
github.com/hashicorp/golang-lru v0.5.0 h1:CL2msUPvZTLb5O648aiLNJw3hnBxN2+1Jq8rCOH9wdo=
|
||||
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
|
||||
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/miekg/dns v1.0.14 h1:9jZdLNd/P4+SfEJ0TNyxYpsK8N4GtfylBLqtbYN1sbA=
|
||||
github.com/miekg/dns v1.0.14/go.mod h1:W1PPwlIAgtquWBMBEV9nkV9Cazfe8ScdGz/Lj7v3Nrg=
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c h1:Lgl0gzECD8GnQ5QCWA8o6BtfL6mDH5rQgM4/fX3avOs=
|
||||
github.com/pascaldekloe/goe v0.0.0-20180627143212-57f6aae5913c/go.mod h1:lzWF7FIEvWOWxwDKqyGYQf6ZUaNfKdP144TG7ZOy1lc=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 h1:nn5Wsu0esKSJiIVhscUtVbo7ada43DJhG55ua/hjS5I=
|
||||
github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=
|
||||
github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
|
||||
github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
|
||||
golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3 h1:KYQXGkl6vs02hK7pK4eIbw0NpNPedieTSTEiJ//bwGs=
|
||||
golang.org/x/crypto v0.0.0-20181029021203-45a5f77698d3/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
|
||||
golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519 h1:x6rhz8Y9CjbgQkccRGmELH6K+LJj7tOoh3XWeC1yaQM=
|
||||
golang.org/x/net v0.0.0-20181023162649-9b4f9f5ad519/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4 h1:YUO/7uOKsKeq9UokNS62b8FYywz3ker1l1vDZRCRefw=
|
||||
golang.org/x/sync v0.0.0-20181221193216-37e7f081c4d4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
|
||||
golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5 h1:x6r4Jo0KNzOOzYd8lbcRsqjuqEASK6ob3auvWYM4/8U=
|
||||
golang.org/x/sys v0.0.0-20181026203630-95b1ffbd15a5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
|
||||
google.golang.org/appengine v1.4.0 h1:/wp5JvzpHIxhs/dumFmF7BXTf3Z+dd4uXta4kVyO508=
|
||||
google.golang.org/appengine v1.4.0/go.mod h1:xpcJRLb0r/rnEns0DIKYYv+WjYCduHsrkT7/EB5XEv4=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
|
||||
gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
|
||||
gopkg.in/vmihailenco/msgpack.v2 v2.9.1 h1:kb0VV7NuIojvRfzwslQeP3yArBqJHW9tOl4t38VS1jM=
|
||||
gopkg.in/vmihailenco/msgpack.v2 v2.9.1/go.mod h1:/3Dn1Npt9+MYyLpYYXjInO/5jvMLamn+AEGwNEOatn8=
|
||||
|
|
@ -665,3 +665,27 @@ func (m *Memberlist) hasShutdown() bool {
|
|||
func (m *Memberlist) hasLeft() bool {
|
||||
return atomic.LoadInt32(&m.leave) == 1
|
||||
}
|
||||
|
||||
func (m *Memberlist) getNodeState(addr string) nodeStateType {
|
||||
m.nodeLock.RLock()
|
||||
defer m.nodeLock.RUnlock()
|
||||
|
||||
n := m.nodeMap[addr]
|
||||
return n.State
|
||||
}
|
||||
|
||||
func (m *Memberlist) getNodeStateChange(addr string) time.Time {
|
||||
m.nodeLock.RLock()
|
||||
defer m.nodeLock.RUnlock()
|
||||
|
||||
n := m.nodeMap[addr]
|
||||
return n.StateChange
|
||||
}
|
||||
|
||||
func (m *Memberlist) changeNode(addr string, f func(*nodeState)) {
|
||||
m.nodeLock.Lock()
|
||||
defer m.nodeLock.Unlock()
|
||||
|
||||
n := m.nodeMap[addr]
|
||||
f(n)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -221,6 +221,16 @@ func (t *NetTransport) Shutdown() error {
|
|||
// and hands them off to the stream channel.
|
||||
func (t *NetTransport) tcpListen(tcpLn *net.TCPListener) {
|
||||
defer t.wg.Done()
|
||||
|
||||
// baseDelay is the initial delay after an AcceptTCP() error before attempting again
|
||||
const baseDelay = 5 * time.Millisecond
|
||||
|
||||
// maxDelay is the maximum delay after an AcceptTCP() error before attempting again.
|
||||
// In the case that tcpListen() is error-looping, it will delay the shutdown check.
|
||||
// Therefore, changes to maxDelay may have an effect on the latency of shutdown.
|
||||
const maxDelay = 1 * time.Second
|
||||
|
||||
var loopDelay time.Duration
|
||||
for {
|
||||
conn, err := tcpLn.AcceptTCP()
|
||||
if err != nil {
|
||||
|
|
@ -228,9 +238,22 @@ func (t *NetTransport) tcpListen(tcpLn *net.TCPListener) {
|
|||
break
|
||||
}
|
||||
|
||||
if loopDelay == 0 {
|
||||
loopDelay = baseDelay
|
||||
} else {
|
||||
loopDelay *= 2
|
||||
}
|
||||
|
||||
if loopDelay > maxDelay {
|
||||
loopDelay = maxDelay
|
||||
}
|
||||
|
||||
t.logger.Printf("[ERR] memberlist: Error accepting TCP connection: %v", err)
|
||||
time.Sleep(loopDelay)
|
||||
continue
|
||||
}
|
||||
// No error, reset loop delay
|
||||
loopDelay = 0
|
||||
|
||||
t.streamCh <- conn
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,8 +1,10 @@
|
|||
package memberlist
|
||||
|
||||
import (
|
||||
"sort"
|
||||
"math"
|
||||
"sync"
|
||||
|
||||
"github.com/google/btree"
|
||||
)
|
||||
|
||||
// TransmitLimitedQueue is used to queue messages to broadcast to
|
||||
|
|
@ -19,15 +21,93 @@ type TransmitLimitedQueue struct {
|
|||
// number of retransmissions attempted.
|
||||
RetransmitMult int
|
||||
|
||||
sync.Mutex
|
||||
bcQueue limitedBroadcasts
|
||||
mu sync.Mutex
|
||||
tq *btree.BTree // stores *limitedBroadcast as btree.Item
|
||||
tm map[string]*limitedBroadcast
|
||||
idGen int64
|
||||
}
|
||||
|
||||
type limitedBroadcast struct {
|
||||
transmits int // Number of transmissions attempted.
|
||||
transmits int // btree-key[0]: Number of transmissions attempted.
|
||||
msgLen int64 // btree-key[1]: copied from len(b.Message())
|
||||
id int64 // btree-key[2]: unique incrementing id stamped at submission time
|
||||
b Broadcast
|
||||
|
||||
name string // set if Broadcast is a NamedBroadcast
|
||||
}
|
||||
|
||||
// Less tests whether the current item is less than the given argument.
|
||||
//
|
||||
// This must provide a strict weak ordering.
|
||||
// If !a.Less(b) && !b.Less(a), we treat this to mean a == b (i.e. we can only
|
||||
// hold one of either a or b in the tree).
|
||||
//
|
||||
// default ordering is
|
||||
// - [transmits=0, ..., transmits=inf]
|
||||
// - [transmits=0:len=999, ..., transmits=0:len=2, ...]
|
||||
// - [transmits=0:len=999,id=999, ..., transmits=0:len=999:id=1, ...]
|
||||
func (b *limitedBroadcast) Less(than btree.Item) bool {
|
||||
o := than.(*limitedBroadcast)
|
||||
if b.transmits < o.transmits {
|
||||
return true
|
||||
} else if b.transmits > o.transmits {
|
||||
return false
|
||||
}
|
||||
if b.msgLen > o.msgLen {
|
||||
return true
|
||||
} else if b.msgLen < o.msgLen {
|
||||
return false
|
||||
}
|
||||
return b.id > o.id
|
||||
}
|
||||
|
||||
// for testing; emits in transmit order if reverse=false
|
||||
func (q *TransmitLimitedQueue) orderedView(reverse bool) []*limitedBroadcast {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
|
||||
out := make([]*limitedBroadcast, 0, q.lenLocked())
|
||||
q.walkReadOnlyLocked(reverse, func(cur *limitedBroadcast) bool {
|
||||
out = append(out, cur)
|
||||
return true
|
||||
})
|
||||
|
||||
return out
|
||||
}
|
||||
|
||||
// walkReadOnlyLocked calls f for each item in the queue traversing it in
|
||||
// natural order (by Less) when reverse=false and the opposite when true. You
|
||||
// must hold the mutex.
|
||||
//
|
||||
// This method panics if you attempt to mutate the item during traversal. The
|
||||
// underlying btree should also not be mutated during traversal.
|
||||
func (q *TransmitLimitedQueue) walkReadOnlyLocked(reverse bool, f func(*limitedBroadcast) bool) {
|
||||
if q.lenLocked() == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
iter := func(item btree.Item) bool {
|
||||
cur := item.(*limitedBroadcast)
|
||||
|
||||
prevTransmits := cur.transmits
|
||||
prevMsgLen := cur.msgLen
|
||||
prevID := cur.id
|
||||
|
||||
keepGoing := f(cur)
|
||||
|
||||
if prevTransmits != cur.transmits || prevMsgLen != cur.msgLen || prevID != cur.id {
|
||||
panic("edited queue while walking read only")
|
||||
}
|
||||
|
||||
return keepGoing
|
||||
}
|
||||
|
||||
if reverse {
|
||||
q.tq.Descend(iter) // end with transmit 0
|
||||
} else {
|
||||
q.tq.Ascend(iter) // start with transmit 0
|
||||
}
|
||||
}
|
||||
type limitedBroadcasts []*limitedBroadcast
|
||||
|
||||
// Broadcast is something that can be broadcasted via gossip to
|
||||
// the memberlist cluster.
|
||||
|
|
@ -45,123 +125,298 @@ type Broadcast interface {
|
|||
Finished()
|
||||
}
|
||||
|
||||
// NamedBroadcast is an optional extension of the Broadcast interface that
|
||||
// gives each message a unique string name, and that is used to optimize
|
||||
//
|
||||
// You shoud ensure that Invalidates() checks the same uniqueness as the
|
||||
// example below:
|
||||
//
|
||||
// func (b *foo) Invalidates(other Broadcast) bool {
|
||||
// nb, ok := other.(NamedBroadcast)
|
||||
// if !ok {
|
||||
// return false
|
||||
// }
|
||||
// return b.Name() == nb.Name()
|
||||
// }
|
||||
//
|
||||
// Invalidates() isn't currently used for NamedBroadcasts, but that may change
|
||||
// in the future.
|
||||
type NamedBroadcast interface {
|
||||
Broadcast
|
||||
// The unique identity of this broadcast message.
|
||||
Name() string
|
||||
}
|
||||
|
||||
// UniqueBroadcast is an optional interface that indicates that each message is
|
||||
// intrinsically unique and there is no need to scan the broadcast queue for
|
||||
// duplicates.
|
||||
//
|
||||
// You should ensure that Invalidates() always returns false if implementing
|
||||
// this interface. Invalidates() isn't currently used for UniqueBroadcasts, but
|
||||
// that may change in the future.
|
||||
type UniqueBroadcast interface {
|
||||
Broadcast
|
||||
// UniqueBroadcast is just a marker method for this interface.
|
||||
UniqueBroadcast()
|
||||
}
|
||||
|
||||
// QueueBroadcast is used to enqueue a broadcast
|
||||
func (q *TransmitLimitedQueue) QueueBroadcast(b Broadcast) {
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
q.queueBroadcast(b, 0)
|
||||
}
|
||||
|
||||
// Check if this message invalidates another
|
||||
n := len(q.bcQueue)
|
||||
for i := 0; i < n; i++ {
|
||||
if b.Invalidates(q.bcQueue[i].b) {
|
||||
q.bcQueue[i].b.Finished()
|
||||
copy(q.bcQueue[i:], q.bcQueue[i+1:])
|
||||
q.bcQueue[n-1] = nil
|
||||
q.bcQueue = q.bcQueue[:n-1]
|
||||
n--
|
||||
// lazyInit initializes internal data structures the first time they are
|
||||
// needed. You must already hold the mutex.
|
||||
func (q *TransmitLimitedQueue) lazyInit() {
|
||||
if q.tq == nil {
|
||||
q.tq = btree.New(32)
|
||||
}
|
||||
if q.tm == nil {
|
||||
q.tm = make(map[string]*limitedBroadcast)
|
||||
}
|
||||
}
|
||||
|
||||
// queueBroadcast is like QueueBroadcast but you can use a nonzero value for
|
||||
// the initial transmit tier assigned to the message. This is meant to be used
|
||||
// for unit testing.
|
||||
func (q *TransmitLimitedQueue) queueBroadcast(b Broadcast, initialTransmits int) {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
|
||||
q.lazyInit()
|
||||
|
||||
if q.idGen == math.MaxInt64 {
|
||||
// it's super duper unlikely to wrap around within the retransmit limit
|
||||
q.idGen = 1
|
||||
} else {
|
||||
q.idGen++
|
||||
}
|
||||
id := q.idGen
|
||||
|
||||
lb := &limitedBroadcast{
|
||||
transmits: initialTransmits,
|
||||
msgLen: int64(len(b.Message())),
|
||||
id: id,
|
||||
b: b,
|
||||
}
|
||||
unique := false
|
||||
if nb, ok := b.(NamedBroadcast); ok {
|
||||
lb.name = nb.Name()
|
||||
} else if _, ok := b.(UniqueBroadcast); ok {
|
||||
unique = true
|
||||
}
|
||||
|
||||
// Check if this message invalidates another.
|
||||
if lb.name != "" {
|
||||
if old, ok := q.tm[lb.name]; ok {
|
||||
old.b.Finished()
|
||||
q.deleteItem(old)
|
||||
}
|
||||
} else if !unique {
|
||||
// Slow path, hopefully nothing hot hits this.
|
||||
var remove []*limitedBroadcast
|
||||
q.tq.Ascend(func(item btree.Item) bool {
|
||||
cur := item.(*limitedBroadcast)
|
||||
|
||||
// Special Broadcasts can only invalidate each other.
|
||||
switch cur.b.(type) {
|
||||
case NamedBroadcast:
|
||||
// noop
|
||||
case UniqueBroadcast:
|
||||
// noop
|
||||
default:
|
||||
if b.Invalidates(cur.b) {
|
||||
cur.b.Finished()
|
||||
remove = append(remove, cur)
|
||||
}
|
||||
}
|
||||
return true
|
||||
})
|
||||
for _, cur := range remove {
|
||||
q.deleteItem(cur)
|
||||
}
|
||||
}
|
||||
|
||||
// Append to the queue
|
||||
q.bcQueue = append(q.bcQueue, &limitedBroadcast{0, b})
|
||||
// Append to the relevant queue.
|
||||
q.addItem(lb)
|
||||
}
|
||||
|
||||
// deleteItem removes the given item from the overall datastructure. You
|
||||
// must already hold the mutex.
|
||||
func (q *TransmitLimitedQueue) deleteItem(cur *limitedBroadcast) {
|
||||
_ = q.tq.Delete(cur)
|
||||
if cur.name != "" {
|
||||
delete(q.tm, cur.name)
|
||||
}
|
||||
|
||||
if q.tq.Len() == 0 {
|
||||
// At idle there's no reason to let the id generator keep going
|
||||
// indefinitely.
|
||||
q.idGen = 0
|
||||
}
|
||||
}
|
||||
|
||||
// addItem adds the given item into the overall datastructure. You must already
|
||||
// hold the mutex.
|
||||
func (q *TransmitLimitedQueue) addItem(cur *limitedBroadcast) {
|
||||
_ = q.tq.ReplaceOrInsert(cur)
|
||||
if cur.name != "" {
|
||||
q.tm[cur.name] = cur
|
||||
}
|
||||
}
|
||||
|
||||
// getTransmitRange returns a pair of min/max values for transmit values
|
||||
// represented by the current queue contents. Both values represent actual
|
||||
// transmit values on the interval [0, len). You must already hold the mutex.
|
||||
func (q *TransmitLimitedQueue) getTransmitRange() (minTransmit, maxTransmit int) {
|
||||
if q.lenLocked() == 0 {
|
||||
return 0, 0
|
||||
}
|
||||
minItem, maxItem := q.tq.Min(), q.tq.Max()
|
||||
if minItem == nil || maxItem == nil {
|
||||
return 0, 0
|
||||
}
|
||||
|
||||
min := minItem.(*limitedBroadcast).transmits
|
||||
max := maxItem.(*limitedBroadcast).transmits
|
||||
|
||||
return min, max
|
||||
}
|
||||
|
||||
// GetBroadcasts is used to get a number of broadcasts, up to a byte limit
|
||||
// and applying a per-message overhead as provided.
|
||||
func (q *TransmitLimitedQueue) GetBroadcasts(overhead, limit int) [][]byte {
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
|
||||
// Fast path the default case
|
||||
if len(q.bcQueue) == 0 {
|
||||
if q.lenLocked() == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
transmitLimit := retransmitLimit(q.RetransmitMult, q.NumNodes())
|
||||
bytesUsed := 0
|
||||
var toSend [][]byte
|
||||
|
||||
for i := len(q.bcQueue) - 1; i >= 0; i-- {
|
||||
// Check if this is within our limits
|
||||
b := q.bcQueue[i]
|
||||
msg := b.b.Message()
|
||||
if bytesUsed+overhead+len(msg) > limit {
|
||||
var (
|
||||
bytesUsed int
|
||||
toSend [][]byte
|
||||
reinsert []*limitedBroadcast
|
||||
)
|
||||
|
||||
// Visit fresher items first, but only look at stuff that will fit.
|
||||
// We'll go tier by tier, grabbing the largest items first.
|
||||
minTr, maxTr := q.getTransmitRange()
|
||||
for transmits := minTr; transmits <= maxTr; /*do not advance automatically*/ {
|
||||
free := int64(limit - bytesUsed - overhead)
|
||||
if free <= 0 {
|
||||
break // bail out early
|
||||
}
|
||||
|
||||
// Search for the least element on a given tier (by transmit count) as
|
||||
// defined in the limitedBroadcast.Less function that will fit into our
|
||||
// remaining space.
|
||||
greaterOrEqual := &limitedBroadcast{
|
||||
transmits: transmits,
|
||||
msgLen: free,
|
||||
id: math.MaxInt64,
|
||||
}
|
||||
lessThan := &limitedBroadcast{
|
||||
transmits: transmits + 1,
|
||||
msgLen: math.MaxInt64,
|
||||
id: math.MaxInt64,
|
||||
}
|
||||
var keep *limitedBroadcast
|
||||
q.tq.AscendRange(greaterOrEqual, lessThan, func(item btree.Item) bool {
|
||||
cur := item.(*limitedBroadcast)
|
||||
// Check if this is within our limits
|
||||
if int64(len(cur.b.Message())) > free {
|
||||
// If this happens it's a bug in the datastructure or
|
||||
// surrounding use doing something like having len(Message())
|
||||
// change over time. There's enough going on here that it's
|
||||
// probably sane to just skip it and move on for now.
|
||||
return true
|
||||
}
|
||||
keep = cur
|
||||
return false
|
||||
})
|
||||
if keep == nil {
|
||||
// No more items of an appropriate size in the tier.
|
||||
transmits++
|
||||
continue
|
||||
}
|
||||
|
||||
msg := keep.b.Message()
|
||||
|
||||
// Add to slice to send
|
||||
bytesUsed += overhead + len(msg)
|
||||
toSend = append(toSend, msg)
|
||||
|
||||
// Check if we should stop transmission
|
||||
b.transmits++
|
||||
if b.transmits >= transmitLimit {
|
||||
b.b.Finished()
|
||||
n := len(q.bcQueue)
|
||||
q.bcQueue[i], q.bcQueue[n-1] = q.bcQueue[n-1], nil
|
||||
q.bcQueue = q.bcQueue[:n-1]
|
||||
q.deleteItem(keep)
|
||||
if keep.transmits+1 >= transmitLimit {
|
||||
keep.b.Finished()
|
||||
} else {
|
||||
// We need to bump this item down to another transmit tier, but
|
||||
// because it would be in the same direction that we're walking the
|
||||
// tiers, we will have to delay the reinsertion until we are
|
||||
// finished our search. Otherwise we'll possibly re-add the message
|
||||
// when we ascend to the next tier.
|
||||
keep.transmits++
|
||||
reinsert = append(reinsert, keep)
|
||||
}
|
||||
}
|
||||
|
||||
// If we are sending anything, we need to re-sort to deal
|
||||
// with adjusted transmit counts
|
||||
if len(toSend) > 0 {
|
||||
q.bcQueue.Sort()
|
||||
for _, cur := range reinsert {
|
||||
q.addItem(cur)
|
||||
}
|
||||
|
||||
return toSend
|
||||
}
|
||||
|
||||
// NumQueued returns the number of queued messages
|
||||
func (q *TransmitLimitedQueue) NumQueued() int {
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
return len(q.bcQueue)
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
return q.lenLocked()
|
||||
}
|
||||
|
||||
// Reset clears all the queued messages
|
||||
func (q *TransmitLimitedQueue) Reset() {
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
for _, b := range q.bcQueue {
|
||||
b.b.Finished()
|
||||
// lenLocked returns the length of the overall queue datastructure. You must
|
||||
// hold the mutex.
|
||||
func (q *TransmitLimitedQueue) lenLocked() int {
|
||||
if q.tq == nil {
|
||||
return 0
|
||||
}
|
||||
q.bcQueue = nil
|
||||
return q.tq.Len()
|
||||
}
|
||||
|
||||
// Reset clears all the queued messages. Should only be used for tests.
|
||||
func (q *TransmitLimitedQueue) Reset() {
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
|
||||
q.walkReadOnlyLocked(false, func(cur *limitedBroadcast) bool {
|
||||
cur.b.Finished()
|
||||
return true
|
||||
})
|
||||
|
||||
q.tq = nil
|
||||
q.tm = nil
|
||||
q.idGen = 0
|
||||
}
|
||||
|
||||
// Prune will retain the maxRetain latest messages, and the rest
|
||||
// will be discarded. This can be used to prevent unbounded queue sizes
|
||||
func (q *TransmitLimitedQueue) Prune(maxRetain int) {
|
||||
q.Lock()
|
||||
defer q.Unlock()
|
||||
q.mu.Lock()
|
||||
defer q.mu.Unlock()
|
||||
|
||||
// Do nothing if queue size is less than the limit
|
||||
n := len(q.bcQueue)
|
||||
if n < maxRetain {
|
||||
return
|
||||
for q.tq.Len() > maxRetain {
|
||||
item := q.tq.Max()
|
||||
if item == nil {
|
||||
break
|
||||
}
|
||||
cur := item.(*limitedBroadcast)
|
||||
cur.b.Finished()
|
||||
q.deleteItem(cur)
|
||||
}
|
||||
|
||||
// Invalidate the messages we will be removing
|
||||
for i := 0; i < n-maxRetain; i++ {
|
||||
q.bcQueue[i].b.Finished()
|
||||
}
|
||||
|
||||
// Move the messages, and retain only the last maxRetain
|
||||
copy(q.bcQueue[0:], q.bcQueue[n-maxRetain:])
|
||||
q.bcQueue = q.bcQueue[:maxRetain]
|
||||
}
|
||||
|
||||
func (b limitedBroadcasts) Len() int {
|
||||
return len(b)
|
||||
}
|
||||
|
||||
func (b limitedBroadcasts) Less(i, j int) bool {
|
||||
return b[i].transmits < b[j].transmits
|
||||
}
|
||||
|
||||
func (b limitedBroadcasts) Swap(i, j int) {
|
||||
b[i], b[j] = b[j], b[i]
|
||||
}
|
||||
|
||||
func (b limitedBroadcasts) Sort() {
|
||||
sort.Sort(sort.Reverse(b))
|
||||
}
|
||||
|
|
|
|||
|
|
@ -233,6 +233,15 @@ START:
|
|||
m.probeNode(&node)
|
||||
}
|
||||
|
||||
// probeNodeByAddr just safely calls probeNode given only the address of the node (for tests)
|
||||
func (m *Memberlist) probeNodeByAddr(addr string) {
|
||||
m.nodeLock.RLock()
|
||||
n := m.nodeMap[addr]
|
||||
m.nodeLock.RUnlock()
|
||||
|
||||
m.probeNode(n)
|
||||
}
|
||||
|
||||
// probeNode handles a single round of failure checking on a node.
|
||||
func (m *Memberlist) probeNode(node *nodeState) {
|
||||
defer metrics.MeasureSince([]string{"memberlist", "probeNode"}, time.Now())
|
||||
|
|
|
|||
|
|
@ -78,10 +78,9 @@ func retransmitLimit(retransmitMult, n int) int {
|
|||
// shuffleNodes randomly shuffles the input nodes using the Fisher-Yates shuffle
|
||||
func shuffleNodes(nodes []*nodeState) {
|
||||
n := len(nodes)
|
||||
for i := n - 1; i > 0; i-- {
|
||||
j := rand.Intn(i + 1)
|
||||
rand.Shuffle(n, func(i, j int) {
|
||||
nodes[i], nodes[j] = nodes[j], nodes[i]
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// pushPushScale is used to scale the time interval at which push/pull
|
||||
|
|
|
|||
|
|
@ -16,6 +16,9 @@ func (b *broadcast) Invalidates(other memberlist.Broadcast) bool {
|
|||
return false
|
||||
}
|
||||
|
||||
// implements memberlist.UniqueBroadcast
|
||||
func (b *broadcast) UniqueBroadcast() {}
|
||||
|
||||
func (b *broadcast) Message() []byte {
|
||||
return b.msg
|
||||
}
|
||||
|
|
|
|||
|
|
@ -126,7 +126,7 @@
|
|||
{"path":"github.com/hashicorp/go-plugin","checksumSHA1":"lbG9uwM7qJlTIBg+8mjCC88sCPc=","revision":"e8d22c780116115ae5624720c9af0c97afe4f551","revisionTime":"2018-03-31T00:25:53Z"},
|
||||
{"path":"github.com/hashicorp/go-retryablehttp","checksumSHA1":"s6mKR1dSKP04+A3kwSsFr/PvsOU=","revision":"3b087ef2d313afe6c55b2f511d20db04ca767075","revisionTime":"2018-05-31T21:13:21Z"},
|
||||
{"path":"github.com/hashicorp/go-rootcerts","checksumSHA1":"A1PcINvF3UiwHRKn8UcgARgvGRs=","revision":"6bb64b370b90e7ef1fa532be9e591a81c3493e00","revisionTime":"2016-05-03T14:34:40Z"},
|
||||
{"path":"github.com/hashicorp/go-sockaddr","checksumSHA1":"eCWvhgknHMj5K19ePPjIA3l401Q=","revision":"9b4c5fa5b10a683339a270d664474b9f4aee62fc","revisionTime":"2017-10-30T10:43:12Z"},
|
||||
{"path":"github.com/hashicorp/go-sockaddr","checksumSHA1":"J47ySO1q0gcnmoMnir1q1loKzCk=","revision":"e92cdb5343bbaf42b0a596937ae0f382270d6759","revisionTime":"2019-01-03T21:41:36Z"},
|
||||
{"path":"github.com/hashicorp/go-sockaddr/template","checksumSHA1":"PDp9DVLvf3KWxhs4G4DpIwauMSU=","revision":"9b4c5fa5b10a683339a270d664474b9f4aee62fc","revisionTime":"2017-10-30T10:43:12Z"},
|
||||
{"path":"github.com/hashicorp/go-syslog","checksumSHA1":"xZ7Ban1x//6uUIU1xtrTbCYNHBc=","revision":"42a2b573b664dbf281bd48c3cc12c086b17a39ba","revisionTime":"2015-02-18T18:19:46Z"},
|
||||
{"path":"github.com/hashicorp/go-uuid","checksumSHA1":"mAkPa/RLuIwN53GbwIEMATexams=","revision":"64130c7a86d732268a38cb04cfbaf0cc987fda98","revisionTime":"2016-07-17T02:21:40Z"},
|
||||
|
|
@ -146,12 +146,12 @@
|
|||
{"path":"github.com/hashicorp/hil","checksumSHA1":"kqCMCHy2b+RBMKC+ER+OPqp8C3E=","revision":"1e86c6b523c55d1fa6c6e930ce80b548664c95c2","revisionTime":"2016-07-11T23:18:37Z"},
|
||||
{"path":"github.com/hashicorp/hil/ast","checksumSHA1":"UICubs001+Q4MsUf9zl2vcMzWQQ=","revision":"1e86c6b523c55d1fa6c6e930ce80b548664c95c2","revisionTime":"2016-07-11T23:18:37Z"},
|
||||
{"path":"github.com/hashicorp/logutils","checksumSHA1":"vt+P9D2yWDO3gdvdgCzwqunlhxU=","revision":"0dc08b1671f34c4250ce212759ebd880f743d883","revisionTime":"2015-06-09T07:04:31Z"},
|
||||
{"path":"github.com/hashicorp/memberlist","checksumSHA1":"q6yTL5vSGnWxUtcocVU3YIG/HNc=","revision":"b195c8e4fcc6284fff1583fd6ab09e68ca207551","revisionTime":"2018-08-09T14:04:54Z"},
|
||||
{"path":"github.com/hashicorp/memberlist","checksumSHA1":"yAu2gPVXIh28yJ2If5gZPrf04kU=","revision":"1a62499c21db33d57691001d5e08a71ec857b18f","revisionTime":"2019-01-03T22:22:36Z"},
|
||||
{"path":"github.com/hashicorp/net-rpc-msgpackrpc","checksumSHA1":"qnlqWJYV81ENr61SZk9c65R1mDo=","revision":"a14192a58a694c123d8fe5481d4a4727d6ae82f3","revisionTime":"2015-11-16T02:03:38Z"},
|
||||
{"path":"github.com/hashicorp/raft","checksumSHA1":"3U9bQLEMikE47n4TZP6uOdgXIyQ=","revision":"da92cfe76e0c1c9b94bbc9d884ec4b2b3b90b699","revisionTime":"2018-08-17T18:12:11Z","version":"master","versionExact":"master"},
|
||||
{"path":"github.com/hashicorp/raft-boltdb","checksumSHA1":"QAxukkv54/iIvLfsUP6IK4R0m/A=","revision":"d1e82c1ec3f15ee991f7cc7ffd5b67ff6f5bbaee","revisionTime":"2015-02-01T20:08:39Z"},
|
||||
{"path":"github.com/hashicorp/serf/coordinate","checksumSHA1":"0PeWsO2aI+2PgVYlYlDPKfzCLEQ=","revision":"19bbd39e421bdf3559d5025fb2c760f5ffa56233","revisionTime":"2018-08-09T14:17:58Z"},
|
||||
{"path":"github.com/hashicorp/serf/serf","checksumSHA1":"axdQxCEwvUr1AygfYIMMxPkS1pY=","revision":"19bbd39e421bdf3559d5025fb2c760f5ffa56233","revisionTime":"2018-08-09T14:17:58Z"},
|
||||
{"path":"github.com/hashicorp/serf/coordinate","checksumSHA1":"0PeWsO2aI+2PgVYlYlDPKfzCLEQ=","revision":"c7f3bc96b40972e67dfbe007c1fa825cf59ac8c2","revisionTime":"2019-01-04T15:39:47Z"},
|
||||
{"path":"github.com/hashicorp/serf/serf","checksumSHA1":"siLn7zwVHQk070rpd99BTktGfTs=","revision":"c7f3bc96b40972e67dfbe007c1fa825cf59ac8c2","revisionTime":"2019-01-04T15:39:47Z"},
|
||||
{"path":"github.com/hashicorp/vault/api","checksumSHA1":"LYQZ+o7zJCda/6LibdN0spFco34=","revision":"533003e27840d9646cb4e7d23b3a113895da1dd0","revisionTime":"2018-06-20T14:55:40Z","version":"v0.10.3","versionExact":"v0.10.3"},
|
||||
{"path":"github.com/hashicorp/vault/audit","checksumSHA1":"2JOC+Ur0S3U8Gqv2cfNB3zxgSBk=","revision":"c737968235c8673b872350f0a047877bee396342","revisionTime":"2018-06-20T16:45:32Z"},
|
||||
{"path":"github.com/hashicorp/vault/builtin/logical/database/dbplugin","checksumSHA1":"RCwWixWwKG6j2vF9iVoxbCzo6p4=","revision":"c737968235c8673b872350f0a047877bee396342","revisionTime":"2018-06-20T16:45:32Z"},
|
||||
|
|
|
|||
Loading…
Reference in New Issue