reweave/weave
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1149 from weaveworks/1118-respond-late-paxos

Browse Source 
Respond to paxos messages when we already have a ring

Fixes #1118.
This commit is contained in:
Matthias Radestock committed 2015-07-15 13:24:40 +01:00
2 parents 7e34ec61f6 b3979c9505
commit 52d3bcfe8b
10 files changed
+45 -38
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

ipam/allocator.go
+20 -14
View File
@@ -359,18 +359,18 @@ func (alloc *Allocator) OnGossipUnicast(sender router.PeerName, msg []byte) erro
alloc.donateSpace(r, sender)
resultChan <- nil
case msgRingUpdate:
resultChan <- alloc.update(msg[1:])
resultChan <- alloc.update(sender, msg[1:])
}
}
return <-resultChan
}
// OnGossipBroadcast (Sync)
func (alloc *Allocator) OnGossipBroadcast(msg []byte) (router.GossipData, error) {
alloc.debugln("OnGossipBroadcast:", len(msg), "bytes")
func (alloc *Allocator) OnGossipBroadcast(sender router.PeerName, msg []byte) (router.GossipData, error) {
alloc.debugln("OnGossipBroadcast from", sender, ":", len(msg), "bytes")
resultChan := make(chan error)
alloc.actionChan <- func() {
resultChan <- alloc.update(msg)
resultChan <- alloc.update(sender, msg)
}
return alloc.Gossip(), <-resultChan
}
@@ -419,7 +419,7 @@ func (alloc *Allocator) OnGossip(msg []byte) (router.GossipData, error) {
alloc.debugln("Allocator.OnGossip:", len(msg), "bytes")
resultChan := make(chan error)
alloc.actionChan <- func() {
resultChan <- alloc.update(msg)
resultChan <- alloc.update(router.UnknownPeerName, msg)
}
return nil, <-resultChan // for now, we never propagate updates. TBD
}
@@ -589,7 +589,7 @@ func (alloc *Allocator) sendRingUpdate(dest router.PeerName) {
alloc.gossip.GossipUnicast(dest, msg)
}
func (alloc *Allocator) update(msg []byte) error {
func (alloc *Allocator) update(sender router.PeerName, msg []byte) error {
reader := bytes.NewReader(msg)
decoder := gob.NewDecoder(reader)
var data gossipState
@@ -621,16 +621,22 @@ func (alloc *Allocator) update(msg []byte) error {
return err
}
if data.Paxos != nil && alloc.ring.Empty() {
if alloc.paxos.Update(data.Paxos) {
if alloc.paxos.Think() {
// If something important changed, broadcast
alloc.gossip.GossipBroadcast(alloc.Gossip())
}
if data.Paxos != nil {
if alloc.ring.Empty() {
if alloc.paxos.Update(data.Paxos) {
if alloc.paxos.Think() {
// If something important changed, broadcast
alloc.gossip.GossipBroadcast(alloc.Gossip())
}
if ok, cons := alloc.paxos.Consensus(); ok {
alloc.createRing(cons.Value)
if ok, cons := alloc.paxos.Consensus(); ok {
alloc.createRing(cons.Value)
}
}
} else if sender != router.UnknownPeerName {
// Sender is trying to initialize a ring, but we have one
// already - send it straight back
alloc.sendRingUpdate(sender)
}
}
ipam/allocator_test.go
+9 -9
View File
@@ -86,7 +86,7 @@ func TestBootstrap(t *testing.T) {
alloc2, _ := makeAllocatorWithMockGossip(t, peerNameString, testStart1+"/22", 2)
defer alloc2.Stop()
alloc1.OnGossipBroadcast(alloc2.Encode())
alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode())
alloc1.actionChan <- func() { alloc1.tryPendingOps() }
@@ -108,19 +108,19 @@ func TestBootstrap(t *testing.T) {
// alloc2 receives paxos update and broadcasts its reply
ExpectBroadcastMessage(alloc2, nil)
alloc2.OnGossipBroadcast(alloc1.Encode())
alloc2.OnGossipBroadcast(alloc1.ourName, alloc1.Encode())
ExpectBroadcastMessage(alloc1, nil)
alloc1.OnGossipBroadcast(alloc2.Encode())
alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode())
// both nodes will get consensus now so initialize the ring
ExpectBroadcastMessage(alloc2, nil)
ExpectBroadcastMessage(alloc2, nil)
alloc2.OnGossipBroadcast(alloc1.Encode())
alloc2.OnGossipBroadcast(alloc1.ourName, alloc1.Encode())
CheckAllExpectedMessagesSent(alloc1, alloc2)
alloc1.OnGossipBroadcast(alloc2.Encode())
alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode())
// now alloc1 should have space
AssertSent(t, done)
@@ -181,7 +181,7 @@ func TestCancel(t *testing.T) {
alloc2.Start()
// tell peers about each other
alloc1.OnGossipBroadcast(alloc2.Encode())
alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode())
// Get some IPs, so each allocator has some space
res1, _ := alloc1.Allocate("foo", subnet, nil)
@@ -254,9 +254,9 @@ func TestTransfer(t *testing.T) {
_, err = alloc3.Allocate("bar", subnet, nil)
require.True(t, err == nil, "Failed to get address")
router.GossipBroadcast(alloc2.Gossip())
alloc2.gossip.GossipBroadcast(alloc2.Gossip())
router.Flush()
router.GossipBroadcast(alloc3.Gossip())
alloc2.gossip.GossipBroadcast(alloc3.Gossip())
router.Flush()
router.RemovePeer(alloc2.ourName)
router.RemovePeer(alloc3.ourName)
@@ -476,7 +476,7 @@ func TestGossipSkew(t *testing.T) {
alloc2.now = func() time.Time { return time.Now().Add(time.Hour * 2) }
defer alloc2.Stop()
if _, err := alloc1.OnGossipBroadcast(alloc2.Encode()); err == nil {
if _, err := alloc1.OnGossipBroadcast(alloc2.ourName, alloc2.Encode()); err == nil {
t.Fail()
}
}
ipam/testutils_test.go
+1 -1
View File
@@ -208,7 +208,7 @@ func makeNetworkOfAllocators(size int, cidr string) ([]*Allocator, *gossip.TestR
allocs[i] = alloc
}
gossipRouter.GossipBroadcast(allocs[size-1].Gossip())
allocs[size-1].gossip.GossipBroadcast(allocs[size-1].Gossip())
gossipRouter.Flush()
return allocs, gossipRouter, subnet
}
nameserver/nameserver.go
+1 -1
View File
@@ -238,7 +238,7 @@ func (n *Nameserver) OnGossip(msg []byte) (router.GossipData, error) {
// merge received data into state and return a representation of
// the received data, for further propagation
func (n *Nameserver) OnGossipBroadcast(msg []byte) (router.GossipData, error) {
func (n *Nameserver) OnGossipBroadcast(_ router.PeerName, msg []byte) (router.GossipData, error) {
_, entries, err := n.receiveGossip(msg)
return entries, err
}
router/gossip.go
+1 -1
View File
@@ -21,7 +21,7 @@ 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(update []byte) (GossipData, error)
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
router/gossip_channel.go
+1 -1
View File
@@ -79,7 +79,7 @@ func (c *GossipChannel) deliverBroadcast(srcName PeerName, _ []byte, dec *gob.De
if err := dec.Decode(&payload); err != nil {
return err
}
data, err := c.gossiper.OnGossipBroadcast(payload)
data, err := c.gossiper.OnGossipBroadcast(srcName, payload)
if err != nil || data == nil {
return err
}
router/gossip_test.go
+1 -1
View File
@@ -212,7 +212,7 @@ func (g *testGossiper) OnGossipUnicast(sender PeerName, msg []byte) error {
return nil
}
func (g *testGossiper) OnGossipBroadcast(update []byte) (GossipData, error) {
func (g *testGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) {
g.Lock()
defer g.Unlock()
for _, v := range update {
router/router.go
+1 -1
View File
@@ -389,7 +389,7 @@ func (router *Router) OnGossipUnicast(sender PeerName, msg []byte) error {
return fmt.Errorf("unexpected topology gossip unicast: %v", msg)
}
func (router *Router) OnGossipBroadcast(update []byte) (GossipData, error) {
func (router *Router) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) {
origUpdate, _, err := router.applyTopologyUpdate(update)
if err != nil || len(origUpdate) == 0 {
return nil, err
router/surrogate_gossiper.go
+1 -1
View File
@@ -24,7 +24,7 @@ func (*SurrogateGossiper) OnGossipUnicast(sender PeerName, msg []byte) error {
return nil
}
func (*SurrogateGossiper) OnGossipBroadcast(update []byte) (GossipData, error) {
func (*SurrogateGossiper) OnGossipBroadcast(_ PeerName, update []byte) (GossipData, error) {
return NewSurrogateGossipData(update), nil
}
testing/gossip/mocks.go
+9 -8
View File
@@ -15,7 +15,8 @@ type unicastMessage struct {
buf []byte
}
type broadcastMessage struct {
data router.GossipData
sender router.PeerName
data router.GossipData
}
type exitMessage struct {
exitChan chan struct{}
@@ -33,10 +34,10 @@ func NewTestRouter(loss float32) *TestRouter {
return &TestRouter{make(map[router.PeerName]chan interface{}, 100), loss}
}
func (grouter *TestRouter) GossipBroadcast(update router.GossipData) error {
func (grouter *TestRouter) gossipBroadcast(sender router.PeerName, update router.GossipData) error {
for _, gossipChan := range grouter.gossipChans {
select {
case gossipChan <- broadcastMessage{data: update}:
case gossipChan <- broadcastMessage{sender: sender, data: update}:
default: // drop the message if we cannot send it
common.Log.Errorf("Dropping message")
}
@@ -65,7 +66,7 @@ type TestRouterClient struct {
sender router.PeerName
}
func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interface{}) {
func (grouter *TestRouter) run(sender router.PeerName, gossiper router.Gossiper, gossipChan chan interface{}) {
gossipTimer := time.Tick(10 * time.Second)
for {
select {
@@ -91,13 +92,13 @@ func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interfa
continue
}
for _, msg := range message.data.Encode() {
if _, err := gossiper.OnGossipBroadcast(msg); err != nil {
if _, err := gossiper.OnGossipBroadcast(message.sender, msg); err != nil {
panic(fmt.Sprintf("Error doing gossip broadcast: %s", err))
}
}
}
case <-gossipTimer:
grouter.GossipBroadcast(gossiper.Gossip())
grouter.gossipBroadcast(sender, gossiper.Gossip())
}
}
}
@@ -105,7 +106,7 @@ func (grouter *TestRouter) run(gossiper router.Gossiper, gossipChan chan interfa
func (grouter *TestRouter) Connect(sender router.PeerName, gossiper router.Gossiper) router.Gossip {
gossipChan := make(chan interface{}, 100)
go grouter.run(gossiper, gossipChan)
go grouter.run(sender, gossiper, gossipChan)
grouter.gossipChans[sender] = gossipChan
return TestRouterClient{grouter, sender}
@@ -121,5 +122,5 @@ func (client TestRouterClient) GossipUnicast(dstPeerName router.PeerName, buf []
}
func (client TestRouterClient) GossipBroadcast(update router.GossipData) error {
return client.router.GossipBroadcast(update)
return client.router.gossipBroadcast(client.sender, update)
}