Generalise ticker so we can use it for more than paxos,

and call tryPendingOps() from it so we eventually re-try operations
This commit is contained in:
Bryan Boreham committed 2016-01-12 14:38:44 +00:00
1 parent a5ddba03f7
commit 4672e9ec1f
2 files changed
+15 -20

No files matched your search

+14 -19
View File
@@ -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())
+1 -1
View File
@@ -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)
}