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)