Merge lp:~pedronis/ubuntu-push/little-bits-of-info-n-tweaks into lp:ubuntu-push

[Samuele Pedroni]

some more info reported out of session, some tweaks

various bits and bobs but there's logic to the madness:

* increase ping interval/timeout in some test relying in actual timeouts to be less aggressive

* tell the exchange whether it's fully done in Acked

* have the session report what was the effective elapsed ping interval

* remove a SetDeadline too much, exchange timeout is meant to cover pairs read-write, write-read

[John Lenton (chipaca)]

review: Approve

=== modified file 'server/broker/broker.go'
--- server/broker/broker.go	2014-01-23 20:13:22 +0000
+++ server/broker/broker.go	2014-01-31 17:25:49 +0000
@@ -40,8 +40,8 @@
 
 // Exchange leads the session through performing an exchange, typically delivery.
 type Exchange interface {
-	Prepare(BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)
-	Acked(BrokerSession) error
+	Prepare(sess BrokerSession) (outMessage protocol.SplittableMsg, inMessage interface{}, err error)
+	Acked(sess BrokerSession, done bool) error
 }
 
 // LevelsMap is the type for holding channel levels for session.
 
=== modified file 'server/broker/exchanges.go'
--- server/broker/exchanges.go	2014-01-23 18:39:10 +0000
+++ server/broker/exchanges.go	2014-01-31 17:25:49 +0000
@@ -39,6 +39,9 @@
 	NotificationPayloads []json.RawMessage
 }
 
+// check interface already here
+var _ Exchange = &BroadcastExchange{}
+
 func filterByLevel(clientLevel, topLevel int64, payloads []json.RawMessage) []json.RawMessage {
 	c := int64(len(payloads))
 	delta := topLevel - clientLevel
@@ -63,7 +66,7 @@
 }
 
 // Acked deals with an ACK for a BROADCAST.
-func (sbe *BroadcastExchange) Acked(sess BrokerSession) error {
+func (sbe *BroadcastExchange) Acked(sess BrokerSession, done bool) error {
 	scratchArea := sess.ExchangeScratchArea()
 	if scratchArea.ackMsg.Type != "ack" {
 		return &ErrAbort{"expected ACK message"}
 
=== modified file 'server/broker/exchanges_test.go'
--- server/broker/exchanges_test.go	2014-01-23 20:13:22 +0000
+++ server/broker/exchanges_test.go	2014-01-31 17:25:49 +0000
@@ -52,7 +52,7 @@
 	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"x"},{"a":"y"}]}`)
 	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
 	c.Assert(err, IsNil)
-	err = exchg.Acked(sess)
+	err = exchg.Acked(sess, true)
 	c.Assert(err, IsNil)
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(3))
 }
@@ -76,7 +76,7 @@
 	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
 	err = json.Unmarshal([]byte(`{}`), outMsg)
 	c.Assert(err, IsNil)
-	err = exchg.Acked(sess)
+	err = exchg.Acked(sess, true)
 	c.Assert(err, Not(IsNil))
 	c.Check(sess.LevelsMap[store.SystemInternalChannelId], Equals, int64(0))
 }
@@ -103,6 +103,6 @@
 	c.Check(string(marshalled), Equals, `{"T":"broadcast","ChanId":"0","TopLevel":3,"Payloads":[{"a":"y"}]}`)
 	err = json.Unmarshal([]byte(`{"T":"ack"}`), outMsg)
 	c.Assert(err, IsNil)
-	err = exchg.Acked(sess)
+	err = exchg.Acked(sess, true)
 	c.Assert(err, IsNil)
 }
 
=== modified file 'server/session/session.go'
--- server/session/session.go	2014-01-15 15:54:20 +0000
+++ server/session/session.go	2014-01-31 17:25:49 +0000
@@ -44,7 +44,6 @@
 	if connMsg.Type != "connect" {
 		return nil, &broker.ErrAbort{"expected CONNECT message"}
 	}
-	proto.SetDeadline(time.Now().Add(cfg.ExchangeTimeout()))
 	err = proto.WriteMessage(&protocol.ConnAckMsg{
 		Type:   "connack",
 		Params: protocol.ConnAckParams{PingInterval: cfg.PingInterval().String()},
@@ -70,14 +69,16 @@
 }
 
 // sessionLoop manages the exchanges of the protocol session.
-func sessionLoop(proto protocol.Protocol, sess broker.BrokerSession, cfg SessionConfig) error {
+func sessionLoop(proto protocol.Protocol, sess broker.BrokerSession, cfg SessionConfig, track SessionTracker) error {
 	pingInterval := cfg.PingInterval()
 	exchangeTimeout := cfg.ExchangeTimeout()
 	pingTimer := time.NewTimer(pingInterval)
+	intervalStart := time.Now()
 	ch := sess.SessionChannel()
 	for {
 		select {
 		case <-pingTimer.C:
+			track.EffectivePingInterval(time.Since(intervalStart))
 			pingMsg := &protocol.PingPongMsg{"ping"}
 			var pongMsg protocol.PingPongMsg
 			err := exchange(proto, pingMsg, &pongMsg, exchangeTimeout)
@@ -103,8 +104,9 @@
 				}
 				if done {
 					pingTimer.Reset(pingInterval)
+					intervalStart = time.Now()
 				}
-				err = exchg.Acked(sess)
+				err = exchg.Acked(sess, done)
 				if err != nil {
 					return err
 				}
@@ -134,5 +136,5 @@
 	}
 	track.Registered(sess)
 	defer brkr.Unregister(sess)
-	return track.End(sessionLoop(proto, sess, cfg))
+	return track.End(sessionLoop(proto, sess, cfg, track))
 }
 
=== modified file 'server/session/session_test.go'
--- server/session/session_test.go	2014-01-23 20:13:22 +0000
+++ server/session/session_test.go	2014-01-31 17:25:49 +0000
@@ -147,7 +147,6 @@
 	}()
 	c.Check(takeNext(down), Equals, "deadline 5ms")
 	up <- protocol.ConnectMsg{Type: "connect", ClientVer: "1", DeviceId: "dev-1"}
-	c.Check(takeNext(down), Equals, "deadline 5ms")
 	c.Check(takeNext(down), Equals, protocol.ConnAckMsg{
 		Type:   "connack",
 		Params: protocol.ConnAckParams{(10 * time.Millisecond).String()},
@@ -199,7 +198,6 @@
 	c.Check(err, Equals, io.ErrUnexpectedEOF)
 	// sanity
 	c.Check(takeNext(down), Matches, "deadline.*")
-	c.Check(takeNext(down), Matches, "deadline.*")
 	c.Check(takeNext(down), FitsTypeOf, protocol.ConnAckMsg{})
 }
 
@@ -218,13 +216,14 @@
 }
 
 func (s *sessionSuite) TestSessionLoop(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	sess := &testing.TestBrokerSession{}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, protocol.PingPongMsg{Type: "ping"})
@@ -239,13 +238,14 @@
 }
 
 func (s *sessionSuite) TestSessionLoopWriteError(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	sess := &testing.TestBrokerSession{}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), FitsTypeOf, protocol.PingPongMsg{})
@@ -255,13 +255,14 @@
 }
 
 func (s *sessionSuite) TestSessionLoopMismatch(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	sess := &testing.TestBrokerSession{}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, protocol.PingPongMsg{Type: "ping"})
@@ -294,18 +295,29 @@
 	finErr   error
 	finSleep time.Duration
 	nParts   int
+	done     chan interface{}
 }
 
 func (exchg *testExchange) Prepare(sess broker.BrokerSession) (outMsg protocol.SplittableMsg, inMsg interface{}, err error) {
 	return &testMsg{Type: "msg", nParts: exchg.nParts}, &exchg.inMsg, exchg.prepErr
 }
 
-func (exchg *testExchange) Acked(sess broker.BrokerSession) error {
+func (exchg *testExchange) Acked(sess broker.BrokerSession, done bool) error {
 	time.Sleep(exchg.finSleep)
+	if exchg.done != nil {
+		var doneStr string
+		if done {
+			doneStr = "y"
+		} else {
+			doneStr = "n"
+		}
+		exchg.done <- doneStr
+	}
 	return exchg.finErr
 }
 
 func (s *sessionSuite) TestSessionLoopExchange(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
@@ -314,7 +326,7 @@
 	exchanges <- &testExchange{}
 	sess := &testing.TestBrokerSession{Exchanges: exchanges}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, testMsg{Type: "msg"})
@@ -329,24 +341,28 @@
 }
 
 func (s *sessionSuite) TestSessionLoopExchangeSplit(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	exchanges := make(chan broker.Exchange, 1)
-	exchanges <- &testExchange{nParts: 2}
+	exchange := &testExchange{nParts: 2, done: make(chan interface{}, 2)}
+	exchanges <- exchange
 	sess := &testing.TestBrokerSession{Exchanges: exchanges}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, testMsg{Type: "msg", Part: 1, nParts: 2})
 	up <- nil // no write error
 	up <- testMsg{Type: "ack"}
+	c.Check(takeNext(exchange.done), Equals, "n")
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, testMsg{Type: "msg", Part: 2, nParts: 2})
 	up <- nil // no write error
 	up <- testMsg{Type: "ack"}
+	c.Check(takeNext(exchange.done), Equals, "y")
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, protocol.PingPongMsg{Type: "ping"})
 	up <- nil // no write error
@@ -356,6 +372,7 @@
 }
 
 func (s *sessionSuite) TestSessionLoopExchangePrepareError(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
@@ -365,13 +382,14 @@
 	exchanges <- &testExchange{prepErr: prepErr}
 	sess := &testing.TestBrokerSession{Exchanges: exchanges}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	err := <-errCh
 	c.Check(err, Equals, prepErr)
 }
 
 func (s *sessionSuite) TestSessionLoopExchangeAckedError(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
@@ -381,7 +399,7 @@
 	exchanges <- &testExchange{finErr: finErr}
 	sess := &testing.TestBrokerSession{Exchanges: exchanges}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), DeepEquals, testMsg{Type: "msg"})
@@ -392,6 +410,7 @@
 }
 
 func (s *sessionSuite) TestSessionLoopExchangeWriteError(c *C) {
+	nopTrack := NewTracker(nopLogger)
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
@@ -400,7 +419,7 @@
 	exchanges <- &testExchange{}
 	sess := &testing.TestBrokerSession{Exchanges: exchanges}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout, nopTrack)
 	}()
 	c.Check(takeNext(down), Equals, "deadline 2ms")
 	c.Check(takeNext(down), FitsTypeOf, testMsg{})
@@ -409,25 +428,38 @@
 	c.Check(err, Equals, io.ErrUnexpectedEOF)
 }
 
+type testTracker struct {
+	SessionTracker
+	interval chan interface{}
+}
+
+func (trk *testTracker) EffectivePingInterval(interval time.Duration) {
+	trk.interval <- interval
+}
+
 func (s *sessionSuite) TestSessionLoopExchangeNextPing(c *C) {
+	track := &testTracker{NewTracker(nopLogger), make(chan interface{}, 1)}
 	errCh := make(chan error, 1)
 	up := make(chan interface{}, 5)
 	down := make(chan interface{}, 5)
 	tp := &testProtocol{up, down}
 	exchanges := make(chan broker.Exchange, 1)
-	exchanges <- &testExchange{finSleep: 3 * time.Millisecond}
+	exchanges <- &testExchange{finSleep: 6 * time.Millisecond}
 	sess := &testing.TestBrokerSession{Exchanges: exchanges}
 	go func() {
-		errCh <- sessionLoop(tp, sess, cfg5msPingInterval2msExchangeTout)
+		errCh <- sessionLoop(tp, sess, cfg10msPingInterval5msExchangeTout, track)
 	}()
-	c.Check(takeNext(down), Equals, "deadline 2ms")
+	c.Check(takeNext(down), Equals, "deadline 5ms")
 	c.Check(takeNext(down), DeepEquals, testMsg{Type: "msg"})
 	up <- nil // no write error
 	up <- testMsg{Type: "ack"}
-	tack := time.Now() // next ping interval starts around here
-	c.Check(takeNext(down), Equals, "deadline 2ms")
+	// next ping interval starts around here
+	interval := takeNext(track.interval).(time.Duration)
+	c.Check(takeNext(down), Equals, "deadline 5ms")
 	c.Check(takeNext(down), DeepEquals, protocol.PingPongMsg{Type: "ping"})
-	c.Check(time.Since(tack) < (3+5)*time.Millisecond, Equals, true)
+	effectiveOfPing := float64(interval) / float64(10*time.Millisecond)
+	c.Check(effectiveOfPing > 0.95, Equals, true)
+	c.Check(effectiveOfPing < 1.15, Equals, true)
 	up <- nil // no write error
 	up <- io.EOF
 	err := <-errCh
@@ -470,9 +502,9 @@
 	return c.Conn.SetDeadline(t)
 }
 
-var cfg5msPingInterval = &testSessionConfig{
-	pingInterval:    5 * time.Millisecond,
-	exchangeTimeout: 5 * time.Millisecond,
+var cfg50msPingInterval = &testSessionConfig{
+	pingInterval:    50 * time.Millisecond,
+	exchangeTimeout: 10 * time.Millisecond,
 }
 
 var nopLogger = logger.NewSimpleLogger(ioutil.Discard, "error")
@@ -486,7 +518,7 @@
 	remSrv := &rememberDeadlineConn{srv, make([]string, 0, 2)}
 	brkr := newTestBroker()
 	go func() {
-		errCh <- Session(remSrv, brkr, cfg5msPingInterval, track)
+		errCh <- Session(remSrv, brkr, cfg50msPingInterval, track)
 	}()
 	io.WriteString(cli, "\x00")
 	io.WriteString(cli, "\x00\x20{\"T\":\"connect\",\"DeviceId\":\"DEV\"}")
@@ -494,7 +526,7 @@
 	downStream := bufio.NewReader(cli)
 	msg, err := downStream.ReadBytes(byte('}'))
 	c.Check(err, IsNil)
-	c.Check(msg, DeepEquals, []byte("\x00\x2f{\"T\":\"connack\",\"Params\":{\"PingInterval\":\"5ms\"}"))
+	c.Check(msg, DeepEquals, []byte("\x00\x30{\"T\":\"connack\",\"Params\":{\"PingInterval\":\"50ms\"}"))
 	// eat the last }
 	rbr, err := downStream.ReadByte()
 	c.Check(err, IsNil)
@@ -507,7 +539,7 @@
 	c.Check(len(brkr.registration), Equals, 0) // not yet unregistered
 	cli.Close()
 	err = <-errCh
-	c.Check(remSrv.deadlineKind, DeepEquals, []string{"read", "both", "both", "both"})
+	c.Check(remSrv.deadlineKind, DeepEquals, []string{"read", "both", "both"})
 	c.Check(err, Equals, io.EOF)
 	c.Check(takeNext(brkr.registration), Equals, "unregister DEV")
 	// tracking
@@ -538,7 +570,7 @@
 	defer lst.Close()
 	brkr := newTestBroker()
 	go func() {
-		errCh <- Session(srv, brkr, cfg5msPingInterval, track)
+		errCh <- Session(srv, brkr, cfg50msPingInterval, track)
 	}()
 	io.WriteString(cli, "\x10")
 	err := <-errCh
@@ -557,7 +589,7 @@
 	defer lst.Close()
 	brkr := newTestBroker()
 	go func() {
-		errCh <- Session(srv, brkr, cfg5msPingInterval, track)
+		errCh <- Session(srv, brkr, cfg50msPingInterval, track)
 	}()
 	cli.Close()
 	err := <-errCh
@@ -575,7 +607,7 @@
 	defer lst.Close()
 	brkr := newTestBroker()
 	go func() {
-		errCh <- Session(srv, brkr, cfg5msPingInterval, track)
+		errCh <- Session(srv, brkr, cfg50msPingInterval, track)
 	}()
 	io.WriteString(cli, "\x00")
 	io.WriteString(cli, "\x00")
 
=== modified file 'server/session/tracker.go'
--- server/session/tracker.go	2014-01-23 20:13:22 +0000
+++ server/session/tracker.go	2014-01-31 17:25:49 +0000
@@ -30,6 +30,8 @@
 	Start(WithRemoteAddr)
 	// Session got registered with broker as sess BrokerSession.
 	Registered(sess broker.BrokerSession)
+	// Report effective elapsed ping interval.
+	EffectivePingInterval(time.Duration)
 	// Session got ended with error err (can be nil).
 	End(error) error
 }
@@ -60,6 +62,9 @@
 	trk.Infof("session(%x) registered %v", trk.sessionId, sess.DeviceIdentifier())
 }
 
+func (trk *tracker) EffectivePingInterval(time.Duration) {
+}
+
 func (trk *tracker) End(err error) error {
 	trk.Debugf("session(%x) ended with: %v", trk.sessionId, err)
 	return err