Merge branch 'weave-fix-unittest' of https://github.com/inercia/weave into inercia-weave-fix-unittest

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

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

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)

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

@@ -33,7 +33,6 @@ func main() {
 		udpbuf          int
 		fallback        string
 		refreshInterval int
-		refreshWorkers  int
 		relevantTime    int
 		maxAnswers      int
 		cacheLen        int
@@ -58,7 +57,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")
@@ -122,7 +120,6 @@ func main() {
 		Iface:           iface,
 		LocalTTL:        ttl,
 		RefreshInterval: refreshInterval,
-		RefreshWorkers:  refreshWorkers,
 		RelevantTime:    relevantTime,
 	}
 	zone, err := weavedns.NewZoneDb(zoneConfig)