diff --git a/ipam/allocator.go b/ipam/allocator.go index 0d1fceeb..ef0e0752 100644 --- a/ipam/allocator.go +++ b/ipam/allocator.go @@ -21,7 +21,7 @@ const ( msgRingUpdate msgSpaceRequestDenied - paxosInterval = time.Second * 5 + tickInterval = time.Second * 5 MinSubnetSize = 4 // first and last addresses are excluded, so 2 would be too small ) @@ -53,7 +53,8 @@ type Allocator struct { pendingClaims []operation // held until we know who owns the space gossip mesh.Gossip // our link to the outside world for sending messages paxos *paxos.Node - paxosTicker *time.Ticker + paxosActive bool + ticker *time.Ticker shuttingDown bool // to avoid doing any requests while trying to shut down isKnownPeer func(mesh.PeerName) bool now func() time.Time @@ -77,12 +78,14 @@ func NewAllocator(ourName mesh.PeerName, ourUID mesh.PeerUID, ourNickname string func (alloc *Allocator) Start() { actionChan := make(chan func(), mesh.ChannelSize) alloc.actionChan = actionChan + alloc.ticker = time.NewTicker(tickInterval) go alloc.actorLoop(actionChan) } // Stop makes the actor routine exit, for test purposes ONLY because any // calls after this is processed will hang. Async. func (alloc *Allocator) Stop() { + alloc.ticker.Stop() alloc.actionChan <- nil } @@ -501,19 +504,17 @@ func (alloc *Allocator) SetInterfaces(gossip mesh.Gossip) { func (alloc *Allocator) actorLoop(actionChan <-chan func()) { for { - var tickChan <-chan time.Time - if alloc.paxosTicker != nil { - tickChan = alloc.paxosTicker.C - } - select { case action := <-actionChan: if action == nil { return } action() - case <-tickChan: - alloc.propose() + case <-alloc.ticker.C: + if alloc.paxosActive { + alloc.propose() + } + alloc.tryPendingOps() } alloc.assertInvariants() @@ -525,18 +526,16 @@ func (alloc *Allocator) actorLoop(actionChan <-chan func()) { // Ensure we are making progress towards an established ring func (alloc *Allocator) establishRing() { - if !alloc.ring.Empty() || alloc.paxosTicker != nil { + if !alloc.ring.Empty() || alloc.paxosActive { return } + alloc.paxosActive = true alloc.propose() if ok, cons := alloc.paxos.Consensus(); ok { // If the quorum was 1, then proposing immediately // leads to consensus alloc.createRing(cons.Value) - } else { - // re-try until we get consensus - alloc.paxosTicker = time.NewTicker(paxosInterval) } } @@ -549,13 +548,9 @@ func (alloc *Allocator) createRing(peers []mesh.PeerName) { func (alloc *Allocator) ringUpdated() { // When we have a ring, we don't need paxos any more - if alloc.paxos != nil { + if alloc.paxosActive { + alloc.paxosActive = false alloc.paxos = nil - - if alloc.paxosTicker != nil { - alloc.paxosTicker.Stop() - alloc.paxosTicker = nil - } } alloc.space.UpdateRanges(alloc.ring.OwnedRanges()) diff --git a/ipam/status.go b/ipam/status.go index e7a252ba..3c9aeb12 100644 --- a/ipam/status.go +++ b/ipam/status.go @@ -36,7 +36,7 @@ func NewStatus(allocator *Allocator, defaultSubnet address.CIDR) *Status { } var paxosStatus *paxos.Status - if allocator.paxosTicker != nil { + if allocator.paxosActive { paxosStatus = paxos.NewStatus(allocator.paxos) }