From d6a9fc57cca86a2975706bb06de39cdf52e47534 Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Wed, 3 Jun 2015 11:27:08 +0200 Subject: [PATCH 1/2] New scheduled queue, for scheduling calls to callables --- common/sched_queue.go | 145 +++++++++++++++++++++++++++++++++++++ common/sched_queue_test.go | 87 ++++++++++++++++++++++ nameserver/mocks_test.go | 3 + 3 files changed, 235 insertions(+) create mode 100644 common/sched_queue.go create mode 100644 common/sched_queue_test.go diff --git a/common/sched_queue.go b/common/sched_queue.go new file mode 100644 index 00000000..df6a74ad --- /dev/null +++ b/common/sched_queue.go @@ -0,0 +1,145 @@ +package common + +import ( + "container/heap" + "math" + "time" + + "github.com/benbjohnson/clock" + "runtime" + "sync/atomic" +) + +// A callable is a function that is called and return an (optional) next scheduled time +type callable func() time.Time + +type schedCall struct { + t time.Time + c callable +} + +func (sc *schedCall) FromNow(now time.Time) time.Duration { + if sc.t.After(now) { + return time.Duration(sc.t.Sub(now).Nanoseconds()) + } + return time.Duration(0) +} + +//////////////////////////////////////////////////////////////////////////////// + +// A min-heap of scheduled callables +type schedCallsHeap []*schedCall + +func (ch *schedCallsHeap) Len() int { return len(*ch) } +func (ch *schedCallsHeap) Less(i, j int) bool { return (*ch)[i].t.Before((*ch)[j].t) } +func (ch *schedCallsHeap) Swap(i, j int) { + (*ch)[i], (*ch)[j] = (*ch)[j], (*ch)[i] +} + +func (ch *schedCallsHeap) Push(x interface{}) { + // Push and Pop use pointer receivers because they modify the slice's length, + // not just its contents. + entry := x.(*schedCall) + *ch = append(*ch, entry) +} + +func (ch *schedCallsHeap) Pop() interface{} { + old := *ch + n := len(old) + item := old[n-1] + *ch = old[0 : n-1] + return item +} + +//////////////////////////////////////////////////////////////////////////////// + +// SchedQueue is queue of scheduled callables +type SchedQueue struct { + clock clock.Clock + callablesH schedCallsHeap + schedChan chan *schedCall + closeChan chan bool + counter uint64 // number of calls invoked so far (used for stats). Note: it will wrap. +} + +// NewSchedQueue creates a new scheduled queue +func NewSchedQueue(clock clock.Clock) *SchedQueue { + cq := SchedQueue{ + clock: clock, + schedChan: make(chan *schedCall), + closeChan: make(chan bool), + } + heap.Init(&cq.callablesH) + return &cq +} + +// Start starts the scheduled queue +func (cq *SchedQueue) Start() { + go func() { + defer func() { Debug.Printf("[sched-q] Quitting (%d calls pending)", len(cq.callablesH)) }() + + var now time.Time + timer := cq.clock.Timer(time.Duration(math.MaxInt64)) + + for { + // Wait until an insertion, until the next callable or until we are Stop()ed + select { + case ce := <-cq.schedChan: + heap.Push(&cq.callablesH, ce) + now = cq.clock.Now() + Debug.Printf("[sched-q] Call scheduled [len:%d]", len(cq.callablesH)) + case <-cq.closeChan: + return + case now = <-timer.C: + } + + timer.Stop() + durationUntilNext := cq.durationUntilNext(now) + for durationUntilNext == 0 { + sched := heap.Pop(&cq.callablesH).(*schedCall) + + atomic.AddUint64(&cq.counter, 1) + schedNextTime := sched.c() + + if !schedNextTime.IsZero() { + Debug.Printf("[sched-q] Re-scheduled at %s", schedNextTime) + sched.t = schedNextTime + heap.Push(&cq.callablesH, sched) // TODO: use a Fix() instead of Pop()&Push() + } + + now = cq.clock.Now() + durationUntilNext = cq.durationUntilNext(now) + } + + timer = cq.clock.Timer(durationUntilNext) + } + }() +} + +// Stop stops the scheduled queue +func (cq *SchedQueue) Stop() { + Debug.Printf("[sched-q] Stopping...") + cq.closeChan <- true +} + +// Add schedules a call. +// The callable should not modify the scheduled queue in any way. +func (cq *SchedQueue) Add(c callable, t time.Time) { + Debug.Printf("[sched-q] Adding call at %s", t) + ce := schedCall{c: c, t: t} + cq.schedChan <- &ce +} + +// Counter returns the number of calls invoked in the queued +// Note: the result will wrap over time. +func (cq *SchedQueue) Count() uint64 { + return atomic.LoadUint64(&cq.counter) +} + +func (cq *SchedQueue) durationUntilNext(now time.Time) time.Duration { + if len(cq.callablesH) > 0 { + firstSched := cq.callablesH[0] + return firstSched.FromNow(now) + } + return time.Duration(math.MaxInt64) +} diff --git a/common/sched_queue_test.go b/common/sched_queue_test.go new file mode 100644 index 00000000..81375a79 --- /dev/null +++ b/common/sched_queue_test.go @@ -0,0 +1,87 @@ +package common + +import ( + "testing" + "time" + + "github.com/benbjohnson/clock" + wt "github.com/weaveworks/weave/testing" + "log" +) + +// Ensure we can add new calls while forwarding the clock +func TestSchedCallsBasic(t *testing.T) { + InitDefaultLogging(testing.Verbose()) + Info.Println("TestSchedCallsBasic starting") + + const testSecs = 1000 + clk := clock.NewMock() + schedQueue := NewSchedQueue(clk) + schedQueue.Start() + defer schedQueue.Stop() + + c := func() time.Time { + return clk.Now().Add(time.Second) + } + + schedQueue.Add(c, clk.Now().Add(time.Second)) + for i := 0; i < testSecs; i++ { + clk.Add(time.Second) + } + + t.Logf("Now: %s - calls: %d", clk.Now(), schedQueue.Count()) + wt.AssertEqualInt(t, (int)(schedQueue.Count()), testSecs, "Number of calls") +} + +// Ensure we can create a 100 seconds gap in the middle of the time travel +func TestSchedCallsGap(t *testing.T) { + InitDefaultLogging(testing.Verbose()) + Info.Println("TestSchedCallsGap starting") + + const testSecs = 1000 + clk := clock.NewMock() + schedQueue := NewSchedQueue(clk) + schedQueue.Start() + defer schedQueue.Stop() + + c2 := func() time.Time { + if schedQueue.Count() == testSecs/2 { + return clk.Now().Add(time.Duration(100) * time.Second) + } + return clk.Now().Add(time.Second) + } + + schedQueue.Add(c2, clk.Now().Add(time.Second)) + for i := 0; i < testSecs; i++ { + clk.Add(time.Second) + } + + t.Logf("Now: %s - calls: %d", clk.Now(), schedQueue.Count()) + wt.AssertEqualInt(t, (int)(schedQueue.Count()), testSecs-100+1, "Number of calls") +} + +func TestSchedCallsStop(t *testing.T) { + InitDefaultLogging(testing.Verbose()) + Info.Println("TestSchedCallsStop starting") + + const testSecs = 1000 + clk := clock.NewMock() + schedQueue := NewSchedQueue(clk) + schedQueue.Start() + defer schedQueue.Stop() + + c2 := func() time.Time { + if schedQueue.Count() == testSecs/2 { + return time.Time{} + } + return clk.Now().Add(time.Second) + } + + schedQueue.Add(c2, clk.Now().Add(time.Second)) + for i := 0; i < testSecs; i++ { + clk.Add(time.Second) + } + + t.Logf("Now: %s - calls: %d", clk.Now(), schedQueue.Count()) + wt.AssertEqualInt(t, (int)(schedQueue.Count()), testSecs/2, "Number of calls") +} diff --git a/nameserver/mocks_test.go b/nameserver/mocks_test.go index 1bec091d..d2cb9810 100644 --- a/nameserver/mocks_test.go +++ b/nameserver/mocks_test.go @@ -432,6 +432,9 @@ func newMockedClock() *mockedClock { return &mockedClock{clock.NewMock()} } +// Note: moving the clock forward does not take into account the data that could +// be waiting in channels. So your code should not depend on the channels +// for creating new timers. Otherwise, time travelling will not be reliable... func (clk *mockedClock) Forward(secs int) { Debug.Printf(">>>>>>> Moving clock forward %d seconds - Time traveling >>>>>>>", secs) clk.Add(time.Duration(secs) * time.Second) From 2eb39d962af820ba21072752e5995c1b3c9d8208 Mon Sep 17 00:00:00 2001 From: Alvaro Saurin Date: Wed, 3 Jun 2015 11:30:45 +0200 Subject: [PATCH 2/2] Use the scheduled queue for refreshing names Some minor time tweaks... --- nameserver/zone.go | 53 +++----------- nameserver/zone_lookup.go | 127 ++++++++++----------------------- nameserver/zone_lookup_test.go | 11 +-- prog/weavedns/main.go | 3 - 4 files changed, 53 insertions(+), 141 deletions(-) diff --git a/nameserver/zone.go b/nameserver/zone.go index 99e35c19..fc75e612 100644 --- a/nameserver/zone.go +++ b/nameserver/zone.go @@ -20,10 +20,8 @@ const ( DefaultLocalDomain = "weave.local." // The default name used for the local domain DefaultRefreshInterval = int(DefaultLocalTTL) // Period for background updates with mDNS DefaultRelevantTime = 60 // When to forget info about remote info if nobody asks... - DefaultNumUpdaters = 4 // Default number of background updaters - defaultRefreshMailbox = 1000 // Number of on-the-fly background queries - defaultRemoteIdent = "weave:remote" // Ident used for info obtained from mDNS + defaultRemoteIdent = "weave:remote" // Ident used for info obtained from mDNS ) // +1 to also exclude a dot @@ -422,12 +420,8 @@ type ZoneConfig struct { Iface *net.Interface // Refresh interval for local names in the zone database (disabled by default) (in seconds) RefreshInterval int - // Number of background updaters for names - RefreshWorkers int // Forget about remote info if nobody asks for it in this time (in seconds) RelevantTime int - // Max pending refresh requests - MaxRefreshRequests int // TTL returnined in local asnwers (with mDNS) LocalTTL int // Force a specific mDNS client @@ -453,21 +447,13 @@ type ZoneDb struct { relevantLimit time.Duration // if no one asks for something in this time, it is not relevant a anymore... clock clock.Clock - refreshChan chan refreshRequest // channel for enqueing requests for updating names - refreshSchedChan chan refreshRequest // channel for enqueing requests for updating names + refreshScheds *SchedQueue refreshCloseChan chan bool - refreshWg sync.WaitGroup refreshInterval time.Duration - refreshWorkers int } // Create a new zone database func NewZoneDb(config ZoneConfig) (zone *ZoneDb, err error) { - maxRefreshRequests := defaultRefreshMailbox - if config.MaxRefreshRequests > 0 { - maxRefreshRequests = config.MaxRefreshRequests - } - zone = &ZoneDb{ domain: config.Domain, idents: make(identRecordSet), @@ -476,10 +462,7 @@ func NewZoneDb(config ZoneConfig) (zone *ZoneDb, err error) { iface: config.Iface, clock: config.Clock, relevantLimit: time.Duration(DefaultRelevantTime) * time.Second, - refreshChan: make(chan refreshRequest, maxRefreshRequests), - refreshSchedChan: make(chan refreshRequest, maxRefreshRequests), refreshCloseChan: make(chan bool), - refreshWorkers: DefaultNumUpdaters, } // fix the default configuration parameters @@ -495,8 +478,9 @@ func NewZoneDb(config ZoneConfig) (zone *ZoneDb, err error) { if config.RelevantTime > 0 { zone.relevantLimit = time.Duration(config.RelevantTime) * time.Second } - if config.RefreshWorkers > 0 { - zone.refreshWorkers = config.RefreshWorkers + + if zone.refreshInterval > 0 { + zone.refreshScheds = NewSchedQueue(zone.clock) } // Create the mDNS client and server @@ -533,21 +517,8 @@ func (zone *ZoneDb) Start() (err error) { if err = zone.mdnsSrv.Start(zone.iface); err != nil { return } - - // Start some name refreshers... if zone.refreshInterval > 0 { - Info.Printf("[zonedb] Starting %d background name updaters", zone.refreshWorkers) - for i := 0; i < zone.refreshWorkers; i++ { - zone.refreshWg.Add(1) - go zone.updater(i) - } - - zone.refreshWg.Add(1) - go zone.periodicUpdater() - } else { - // TODO: the refresher also garbage collects irrelevant remote info... we should have an alternative mechanism - // when we don't use refreshers. - Warning.Printf("[zonedb] Remote info will not be garbage collected") + zone.refreshScheds.Start() } return @@ -557,8 +528,7 @@ func (zone *ZoneDb) Start() (err error) { func (zone *ZoneDb) Stop() error { if zone.refreshInterval > 0 { Debug.Printf("[zonedb] Closing background updaters...") - close(zone.refreshCloseChan) - zone.refreshWg.Wait() + zone.refreshScheds.Stop() } Debug.Printf("[zonedb] Exiting mDNS client and server...") @@ -680,14 +650,13 @@ func (zone *ZoneDb) IsNameRelevant(n string) bool { lastAccess := time.Time{} for _, nameset := range zone.idents { - if t := nameset.getNameLastAccess(n); !t.IsZero() { - if lastAccess.IsZero() || t.After(lastAccess) { - lastAccess = t - } + t := nameset.getNameLastAccess(n) + if !t.IsZero() && (lastAccess.IsZero() || t.After(lastAccess)) { + lastAccess = t } } - return !lastAccess.IsZero() && zone.clock.Now().Sub(lastAccess) < zone.relevantLimit + return !lastAccess.IsZero() && zone.clock.Now().Sub(lastAccess) <= zone.relevantLimit } // Returns true if the info we have for a remote name has expired. Local names are never expired diff --git a/nameserver/zone_lookup.go b/nameserver/zone_lookup.go index fa3cdc78..ba4603bc 100644 --- a/nameserver/zone_lookup.go +++ b/nameserver/zone_lookup.go @@ -1,11 +1,11 @@ package nameserver import ( + "net" "time" "github.com/miekg/dns" . "github.com/weaveworks/weave/common" - "net" ) type uniqZoneRecordKey struct { @@ -231,14 +231,6 @@ func (zone *ZoneDb) DomainLookupInaddr(inaddr string) (res []ZoneRecord, err err // disappeared from the network) // -// TODO: for the sake of simplicity, we implement this mechanism with two channels: one for immediate -// updates and another one for scheduled requests. We could use a heap of "next-time"s and a timer, -// but that would require timers cancellations/updates/etc on insertions/removals/etc, and -// it is probably not worth the trouble as we always use the same refresh period. -// It could be useful if we move to a solution where we set update times from the responses TTLs, -// but we currently use a fixed TTL (30secs), the same as the refresh period... -// Anyway, maybe we will move to a gossip-based solution instead of doing this polling... - // Check if we must start updating a name and, in that case, trigger a immediate update func (zone *ZoneDb) startUpdatingName(name string) { if zone.refreshInterval > 0 { @@ -252,93 +244,46 @@ func (zone *ZoneDb) startUpdatingName(name string) { n.lastRefreshTime = now Debug.Printf("[zonedb] Creating new immediate refresh request for '%s'", name) - zone.refreshChan <- refreshRequest{name: name, time: now} + zone.refreshScheds.Add(func() time.Time { return zone.updater(name) }, now) } } } -// A worker for updating the list of IPs we have for a name -func (zone *ZoneDb) updater(num int) { - defer zone.refreshWg.Done() - - Debug.Printf("[zonedb] Starting background updater #%d...", num) - for { - select { - case <-zone.refreshCloseChan: - Debug.Printf("[zonedb] Background updater #%d: interrupted while waiting for requests: exiting", num) - return - - case request := <-zone.refreshChan: - // if nobody has asked for this name for long time, just forget about it... - // this will eventually garbage collect the `refreshChan` and all remote info in absence of activity - if !zone.IsNameRelevant(request.name) || zone.IsNameExpired(request.name) { - Debug.Printf("[zonedb] '%s' seem to be irrelevant now: removing any remote information", request.name) - zone.mx.Lock() - zone.getNameSet(defaultRemoteIdent).deleteNameIP(request.name, net.IP{}) - zone.mx.Unlock() - continue - } - - // perform the refresh for this name - name := dns.Fqdn(request.name) - Debug.Printf("[zonedb] Refreshing name '%s' with mDNS...", name) - res, _ := zone.mdnsCli.InsistentLookupName(request.name) - if res != nil && len(res) > 0 { - numIps := len(res) - zone.mx.Lock() - now := zone.clock.Now() - added, removed := zone.getNameSet(defaultRemoteIdent).getName(name, true).updateIPs(res, now) - zone.mx.Unlock() - Debug.Printf("[zonedb] Obtained %d IPs for name '%s' with mDNS: %d added, %d removed", - numIps, name, added, removed) - - // once the name has been updated, we insert the request (back) in the periodic requests channel - now = zone.clock.Now() - request.time = now.Add(zone.refreshInterval) - Debug.Printf("[zonedb] Rescheduling update for '%s' in %.2f secs", - request.name, zone.refreshInterval.Seconds()) - zone.refreshSchedChan <- request - } else { - Debug.Printf("[zonedb] nobody knows about '%s'... removing", name) - zone.mx.Lock() - zone.getNameSet(defaultRemoteIdent).deleteNameIP(request.name, net.IP{}) - zone.mx.Unlock() - } - } +// Update the IPs we have for a name +func (zone *ZoneDb) updater(name string) (nextTime time.Time) { + deleteRemoteInfo := func() { + zone.mx.Lock() + zone.getNameSet(defaultRemoteIdent).deleteNameIP(name, net.IP{}) + zone.mx.Unlock() } -} -// The periodic updater -// Consume requests from the `refreshSchedChan`, where requests with increasing scheduling time are enqueued -// for refreshing names... -func (zone *ZoneDb) periodicUpdater() { - defer zone.refreshWg.Done() - for { - select { - case <-zone.refreshCloseChan: - Debug.Printf("[zonedb] Periodic updater: interrupted while waiting for requests: exiting") - return - - case request := <-zone.refreshSchedChan: - now := zone.clock.Now() - // we can sleep until the update time has arrived, as requests are sorted by scheduled time (new request - // in this channel will always be scheduled later than the last item in the channel), so we can - // safely suspend this goroutine until then... - if request.time.After(now) { - ddiff := time.Duration(request.time.Sub(now).Nanoseconds()) - timer := zone.clock.Timer(ddiff) - Debug.Printf("[zonedb] Periodic updater: new request for %s: sleeping for %.2f secs...", - request.name, ddiff.Seconds()) - select { - case <-zone.refreshCloseChan: - Debug.Printf("[zonedb] Periodic updater: interrupted while sleeping: exiting") - return - case <-timer.C: - } - } - - // once the time has arrived, we insert the request in the immediate refresh channel - zone.refreshChan <- request - } + // if nobody has asked for this name for long time, just forget about it... + if !zone.IsNameRelevant(name) || zone.IsNameExpired(name) { + Debug.Printf("[zonedb] '%s' seem to be irrelevant now: removing any remote information", name) + deleteRemoteInfo() + return } + + // perform the refresh for this name + fullName := dns.Fqdn(name) + startTime := zone.clock.Now() + Debug.Printf("[zonedb] Refreshing name '%s' with mDNS...", fullName) + res, _ := zone.mdnsCli.InsistentLookupName(fullName) + if res != nil && len(res) > 0 { + numIps := len(res) + zone.mx.Lock() + now := zone.clock.Now() + added, removed := zone.getNameSet(defaultRemoteIdent).getName(name, true).updateIPs(res, now) + zone.mx.Unlock() + Debug.Printf("[zonedb] Obtained %d IPs for name '%s' with mDNS: %d added, %d removed", + numIps, name, added, removed) + + // once the name has been updated, we re-schedule the update + nextTime = startTime.Add(zone.refreshInterval) + Debug.Printf("[zonedb] Rescheduling update for '%s' in %s", name, nextTime.Sub(zone.clock.Now())) + } else { + Debug.Printf("[zonedb] nobody knows about '%s'... removing", name) + deleteRemoteInfo() + } + return } diff --git a/nameserver/zone_lookup_test.go b/nameserver/zone_lookup_test.go index fde104bb..43007cb9 100644 --- a/nameserver/zone_lookup_test.go +++ b/nameserver/zone_lookup_test.go @@ -72,7 +72,6 @@ func TestZoneRefresh(t *testing.T) { Debug.Printf("Wait for a while, until a refresh is performed...") clk.Forward(refreshInterval + 1) - Debug.Printf("A refresh should have been scheduled now: we should have 3 IPs:") Debug.Printf("the first (local) IP and the others obtained from zone2 with a mDNS query") Debug.Printf("Asking for '%s' again... we should have 3 IPs now", name) @@ -82,7 +81,7 @@ func TestZoneRefresh(t *testing.T) { wt.AssertEqualInt(t, len(res), 3, "lookup result length") Debug.Printf("We will not ask for `name` for a while, so it will become irrelevant and will be removed...") - clk.Forward(relevantTime + refreshInterval + 1) + clk.Forward(refreshInterval + relevantTime + 1) // the name should be irrelevant now, and all remote info should have been // removed from the zone database @@ -93,17 +92,19 @@ func TestZoneRefresh(t *testing.T) { Debug.Printf("There is no remote info about this name at zone 1: a new IP appears remotely meanwhile...") clk.Forward(1) + Debug.Printf("Adding '%s' to Db #2", name) dbs[1].Zone.AddRecord("someident", name, net.ParseIP(addr4)) Debug.Printf("When we ask about this name again, we get 4 IPs (1 local, 3 remote)") - clk.Forward(1) Debug.Printf("Asking for '%s' again... the first lookup will return only the local results", name) res, err = dbs[0].Zone.DomainLookupName(name) wt.AssertEqualInt(t, len(res), 1, "lookup result length") Debug.Printf("... but a second lookup should return all the results in the network") - clk.Forward(1) + + clk.Forward(refreshInterval + 1) + res, err = dbs[0].Zone.DomainLookupName(name) Debug.Printf("Got: %s", res) - wt.AssertEqualInt(t, len(res), 4, "lookup result length") // TODO: this fails 1% of the runs... !!?? + wt.AssertEqualInt(t, len(res), 4, "lookup result length") } diff --git a/prog/weavedns/main.go b/prog/weavedns/main.go index 9b6ab270..887523f2 100644 --- a/prog/weavedns/main.go +++ b/prog/weavedns/main.go @@ -30,7 +30,6 @@ func main() { udpbuf int fallback string refreshInterval int - refreshWorkers int relevantTime int maxAnswers int cacheLen int @@ -54,7 +53,6 @@ func main() { // advanced options flag.IntVar(&negTTL, "neg-ttl", weavedns.DefaultCacheNegLocalTTL, "negative TTL (in secs) for unanswered queries for local names") flag.IntVar(&refreshInterval, "refresh", weavedns.DefaultRefreshInterval, "refresh interval (in secs) for local names (0=disable)") - flag.IntVar(&refreshWorkers, "refresh-workers", weavedns.DefaultNumUpdaters, "default number of background updaters") flag.IntVar(&maxAnswers, "max-answers", weavedns.DefaultMaxAnswers, "maximum number of answers returned to clients (0=unlimited)") flag.IntVar(&relevantTime, "relevant", weavedns.DefaultRelevantTime, "life time for info in the absence of queries (in secs)") flag.IntVar(&udpbuf, "udpbuf", weavedns.DefaultUDPBuflen, "UDP buffer length") @@ -88,7 +86,6 @@ func main() { Iface: iface, LocalTTL: ttl, RefreshInterval: refreshInterval, - RefreshWorkers: refreshWorkers, RelevantTime: relevantTime, } zone, err := weavedns.NewZoneDb(zoneConfig)