Merge lp:~chipaca/ubuntu-push/no-left-over-goroutines-p1 into lp:ubuntu-push/automatic

[John Lenton]

pedronis' branch, with a test fixed.

[John Lenton (chipaca)]

Ignore the prereq :)

[Samuele Pedroni (pedronis)]

review: Approve

[Ubuntu One Auto Pilot (otto-pilot)]

No proposals found for merge of https://code.launchpad.net/~chipaca/ubuntu-push/ubuntu-push into https://code.launchpad.net/~ubuntu-push-hackers/ubuntu-push/automatic.

=== modified file 'bus/testing/testing_endpoint.go'
--- bus/testing/testing_endpoint.go	2014-07-04 23:00:42 +0000
+++ bus/testing/testing_endpoint.go	2015-02-03 11:03:23 +0000
@@ -89,7 +89,11 @@
 				ticker := tc.watchTicker
 				tc.watchLck.RUnlock()
 				if ticker != nil {
-					<-ticker
+					_, ok := <-ticker
+					if !ok {
+						// bail out
+						return
+					}
 				} else {
 					time.Sleep(10 * time.Millisecond)
 				}

=== modified file 'client/client_test.go'
--- client/client_test.go	2015-01-22 09:52:07 +0000
+++ client/client_test.go	2015-02-03 11:03:23 +0000
@@ -27,6 +27,7 @@
 	"os"
 	"path/filepath"
 	"reflect"
+	//"runtime"
 	"testing"
 	"time"
 
@@ -203,6 +204,15 @@
 	cs.writeTestConfig(nil)
 }
 
+func (cs *clientSuite) TearDownTest(c *C) {
+	//fmt.Println("GOROUTINE# ", runtime.NumGoroutine())
+	/*
+	   var x [16*1024]byte
+	   sz := runtime.Stack(x[:], true)
+	   fmt.Println(string(x[:sz]))
+	*/
+}
+
 type sqlientSuite struct{ clientSuite }
 
 func (s *sqlientSuite) SetUpSuite(c *C) {
@@ -651,7 +661,9 @@
 	)
 	siCond := condition.Fail2Work(2)
 	siEndp := testibus.NewMultiValuedTestingEndpoint(siCond, condition.Work(true), []interface{}{int32(101), "mako", "daily", "Unknown", map[string]string{}})
-	testibus.SetWatchTicker(cEndp, make(chan bool))
+	tickerCh := make(chan bool)
+	testibus.SetWatchTicker(cEndp, tickerCh)
+	defer close(tickerCh)
 	// ok, create the thing
 	cli := NewPushClient(cs.configPath, cs.leveldbPath)
 	cli.log = cs.log
@@ -700,6 +712,7 @@
 	c.Assert(cli.initSessionAndPoller(), IsNil)
 	cs.log.ResetCapture()
 	cli.hasConnectivity = true
+	defer cli.session.Close()
 	cli.handleErr(errors.New("bananas"))
 	c.Check(cs.log.Captured(), Matches, ".*session exited.*bananas\n")
 }
@@ -712,6 +725,7 @@
 	cli := NewPushClient(cs.configPath, "")
 	ln, err := cli.seenStateFactory()
 	c.Assert(err, IsNil)
+	defer ln.Close()
 	c.Check(fmt.Sprintf("%T", ln), Equals, "*seenstate.memSeenState")
 }
 
@@ -719,6 +733,7 @@
 	cli := NewPushClient(cs.configPath, ":memory:")
 	ln, err := cli.seenStateFactory()
 	c.Assert(err, IsNil)
+	defer ln.Close()
 	c.Check(fmt.Sprintf("%T", ln), Equals, "*seenstate.sqliteSeenState")
 }
 
@@ -733,6 +748,7 @@
 	c.Assert(cli.initSessionAndPoller(), IsNil)
 
 	c.Assert(cli.hasConnectivity, Equals, false)
+	defer cli.session.Close()
 	cli.handleConnState(true)
 	c.Check(cli.hasConnectivity, Equals, true)
 	c.Assert(cli.session, NotNil)

=== modified file 'client/session/seenstate/seenstate.go'
--- client/session/seenstate/seenstate.go	2014-05-14 17:42:24 +0000
+++ client/session/seenstate/seenstate.go	2015-02-03 11:03:23 +0000
@@ -28,8 +28,10 @@
 	// GetAll() returns a "simple" map of the current levels.
 	GetAllLevels() (map[string]int64, error)
 	// FilterBySeen filters notifications already seen, keep track
-	// of them as well
+	// of them as well.
 	FilterBySeen([]protocol.Notification) ([]protocol.Notification, error)
+	// Close closes state.
+	Close()
 }
 
 type memSeenState struct {
@@ -58,6 +60,9 @@
 	return acc, nil
 }
 
+func (m *memSeenState) Close() {
+}
+
 var _ SeenState = (*memSeenState)(nil)
 
 // NewSeenState returns an implementation of SeenState that is memory-based and

=== modified file 'client/session/seenstate/sqlseenstate.go'
--- client/session/seenstate/sqlseenstate.go	2014-05-14 17:42:24 +0000
+++ client/session/seenstate/sqlseenstate.go	2015-02-03 11:03:23 +0000
@@ -48,6 +48,11 @@
 	return &sqliteSeenState{db}, nil
 }
 
+// Closes closes the underlying db.
+func (ps *sqliteSeenState) Close() {
+	ps.db.Close()
+}
+
 func (ps *sqliteSeenState) SetLevel(level string, top int64) error {
 	_, err := ps.db.Exec("REPLACE INTO level_map (level, top) VALUES (?, ?)", level, top)
 	if err != nil {

=== modified file 'client/session/seenstate/sqlseenstate_test.go'
--- client/session/seenstate/sqlseenstate_test.go	2014-05-14 17:42:24 +0000
+++ client/session/seenstate/sqlseenstate_test.go	2015-02-03 11:03:23 +0000
@@ -112,6 +112,15 @@
 	c.Check(err, ErrorMatches, "cannot insert .*")
 }
 
+func (s *sqlsSuite) TestClose(c *C) {
+	dir := c.MkDir()
+	filename := dir + "test.db"
+	sqls, err := NewSqliteSeenState(filename)
+	c.Check(err, IsNil)
+	c.Assert(sqls, NotNil)
+	sqls.Close()
+}
+
 func (s *sqlsSuite) TestDropPrevThan(c *C) {
 	dir := c.MkDir()
 	filename := dir + "test.db"

=== modified file 'client/session/session_test.go'
--- client/session/session_test.go	2014-11-25 17:36:27 +0000
+++ client/session/session_test.go	2015-02-03 11:03:23 +0000
@@ -162,6 +162,7 @@
 
 func (*brokenSeenState) SetLevel(string, int64) error            { return errors.New("broken.") }
 func (*brokenSeenState) GetAllLevels() (map[string]int64, error) { return nil, errors.New("broken.") }
+func (*brokenSeenState) Close()                                  {}
 func (*brokenSeenState) FilterBySeen([]protocol.Notification) ([]protocol.Notification, error) {
 	return nil, errors.New("broken.")
 }

=== modified file 'util/redialer.go'
--- util/redialer.go	2015-01-21 17:55:35 +0000
+++ util/redialer.go	2015-02-03 11:03:23 +0000
@@ -19,7 +19,6 @@
 
 import (
 	"sync"
-	"sync/atomic"
 	"time"
 )
 
@@ -79,26 +78,53 @@
 }
 
 type autoRedialer struct {
-	stateP *redialerState
-	dial   func() error
-	jitter func(time.Duration) time.Duration
+	stateLock     sync.RWMutex
+	stateValue    redialerState
+	stopping      chan struct{}
+	reallyStopped chan struct{}
+	dial          func() error
+	jitter        func(time.Duration) time.Duration
 }
 
 func (ar *autoRedialer) state() redialerState {
-	return redialerState(atomic.LoadUint32((*uint32)(ar.stateP)))
+	ar.stateLock.RLock()
+	defer ar.stateLock.RUnlock()
+	return ar.stateValue
 }
 
 func (ar *autoRedialer) setState(s redialerState) {
-	atomic.StoreUint32((*uint32)(ar.stateP), uint32(s))
+	ar.stateLock.Lock()
+	defer ar.stateLock.Unlock()
+	ar.stateValue = s
 }
 
 func (ar *autoRedialer) setStateIfEqual(oldState, newState redialerState) bool {
-	return atomic.CompareAndSwapUint32((*uint32)(ar.stateP), uint32(oldState), uint32(newState))
+	ar.stateLock.Lock()
+	defer ar.stateLock.Unlock()
+	if ar.stateValue != oldState {
+		return false
+	}
+	ar.stateValue = newState
+	return true
+}
+
+func (ar *autoRedialer) setStateStopped() {
+	ar.stateLock.Lock()
+	defer ar.stateLock.Unlock()
+	switch ar.stateValue {
+	case Stopped:
+		return
+	case Unconfigured:
+		close(ar.reallyStopped)
+	}
+	ar.stateValue = Stopped
+	close(ar.stopping)
 }
 
 func (ar *autoRedialer) Stop() {
 	if ar != nil {
-		ar.setState(Stopped)
+		ar.setStateStopped()
+		<-ar.reallyStopped
 	}
 }
 
@@ -113,6 +139,7 @@
 		// XXX log this
 		return 0
 	}
+	defer close(ar.reallyStopped)
 
 	var timeout time.Duration
 	var dialAttempts uint32 = 0 // unsigned so it can wrap safely ...
@@ -134,15 +161,22 @@
 			timeout += ar.jitter(timeout)
 		}
 		dialAttempts++
-		time.Sleep(timeout)
+		select {
+		case <-ar.stopping:
+		case <-time.After(timeout):
+		}
 	}
 }
 
 // Returns a stoppable AutoRedialer using the provided Dialer. If the Dialer
 // is also a Jitterer, the backoff will be jittered.
 func NewAutoRedialer(dialer Dialer) AutoRedialer {
-	state := Unconfigured
-	ar := &autoRedialer{stateP: &state, dial: dialer.Dial}
+	ar := &autoRedialer{
+		stateValue:    Unconfigured,
+		dial:          dialer.Dial,
+		reallyStopped: make(chan struct{}),
+		stopping:      make(chan struct{}),
+	}
 	jitterer, ok := dialer.(Jitterer)
 	if ok {
 		ar.jitter = jitterer.Jitter