Use the scheduled queue for refreshing names

Some minor time tweaks...

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

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
 }

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")
 }

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)