luxolus
/
open-consul
2
0
Fork 0
Code Issues 1 Pull Requests Packages Releases Wiki Activity

agent: refactor sync loop to linear flow of control

This commit is contained in:
Frank Schroeder 2017-08-28 14:17:10 +02:00 committed by Frank Schröder
parent 5302479ad5
commit 98e5dc86fb
1 changed files with 49 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
agent/ae/ae.go
View File

@ -93,12 +93,10 @@ func (s *StateSyncer) Run() {
Sync: Sync:
for { for {
// update the sync status switch err := s.State.UpdateSyncState(); {
err := s.State.UpdateSyncState()
if err == nil {
break
}
// update sync status failed
case err != nil:
s.Logger.Printf("[ERR] agent: failed to sync remote state: %v", err) s.Logger.Printf("[ERR] agent: failed to sync remote state: %v", err)
// retry updating sync status after some time or when a consul // retry updating sync status after some time or when a consul
@ -106,7 +104,7 @@ Sync:
select { select {
// consul server added to cluster. // consul server added to cluster.
// retry sooner than retryFailIntv to converge cluster quicker // retry sooner than retryFailIntv to converge cluster sooner
// but stagger delay to avoid thundering herd // but stagger delay to avoid thundering herd
case <-s.ServerUpCh: case <-s.ServerUpCh:
select { select {
@ -122,12 +120,13 @@ Sync:
case <-s.ShutdownCh: case <-s.ShutdownCh:
return return
} }
}
// Force-trigger sync to pickup any changes // update sync status OK
default:
// force-trigger sync to pickup any changes
s.triggerSync() s.triggerSync()
// Wait for sync events // do partial syncs until it is time for a full sync again
for { for {
select { select {
// todo(fs): why don't we honor the ServerUpCh here as well? // todo(fs): why don't we honor the ServerUpCh here as well?
@ -141,7 +140,7 @@ Sync:
// } // }
case <-time.After(s.Interval + stagger(s.Interval)): case <-time.After(s.Interval + stagger(s.Interval)):
goto Sync continue Sync
case <-s.TriggerCh: case <-s.TriggerCh:
if s.Paused() { if s.Paused() {
@ -156,6 +155,8 @@ Sync:
} }
} }
} }
}
}
// Pause temporarily disables sync runs. // Pause temporarily disables sync runs.
func (s *StateSyncer) Pause() { func (s *StateSyncer) Pause() {