Reviewers: mp+213569_code.launchpad.net,

Message:
Please take a look.

Description:
state/api: fixes to parallel api.Open

With these changes, api.Open will now
do the right thing if one of the dialers
succeeds before all are started. We also
introduce a short delay between starting
dialers, to avoid unnecessary dial
attempts.

https://code.launchpad.net/~axwalk/juju-core/apiclient-open-parallel-fixes/+merge/213569

(do not edit description out of merge proposal)


Please review this at https://codereview.appspot.com/82900045/

Affected files (+49, -12 lines):
   A [revision details]
   M state/api/apiclient.go
   M state/api/apiclient_test.go
   M state/api/export_test.go


Index: [revision details]
=== added file '[revision details]'
--- [revision details]	2012-01-01 00:00:00 +0000
+++ [revision details]	2012-01-01 00:00:00 +0000
@@ -0,0 +1,2 @@
+Old revision: tarmac-20140401002201-jdbrb1jd9gomhhci
+New revision: <email address hidden>

Index: state/api/apiclient.go
=== modified file 'state/api/apiclient.go'
--- state/api/apiclient.go	2014-03-31 14:43:24 +0000
+++ state/api/apiclient.go	2014-04-01 03:13:49 +0000
@@ -6,6 +6,7 @@
  import (
  	"crypto/tls"
  	"crypto/x509"
+	"fmt"
  	"io"
  	"time"

@@ -84,6 +85,10 @@
  // DialOpts holds configuration parameters that control the
  // Dialing behavior when connecting to a state server.
  type DialOpts struct {
+	// DialAddressInterval is the amount of time to wait
+	// before starting to dial another address.
+	DialAddressInterval time.Duration
+
  	// Timeout is the amount of time to wait contacting
  	// a state server.
  	Timeout time.Duration
@@ -97,8 +102,9 @@
  // parameters for contacting a state server.
  func DefaultDialOpts() DialOpts {
  	return DialOpts{
-		Timeout:    10 * time.Minute,
-		RetryDelay: 2 * time.Second,
+		DialAddressInterval: 50 * time.Millisecond,
+		Timeout:             10 * time.Minute,
+		RetryDelay:          2 * time.Second,
  	}
  }

@@ -114,9 +120,16 @@
  	try := parallel.NewTry(maxParallelDial, nil)
  	defer try.Kill()
  	for _, addr := range info.Addrs {
-		if err := dialWebsocket(addr, opts, pool, try); err != nil {
+		err := dialWebsocket(addr, opts, pool, try)
+		if err == parallel.ErrStopped {
+			break
+		} else if err != nil {
  			return nil, err
  		}
+		select {
+		case <-time.After(opts.DialAddressInterval):
+		case <-try.Dead():
+		}
  	}
  	try.Close()
  	result, err := try.Result()
@@ -160,28 +173,36 @@
  		RootCAs:    rootCAs,
  		ServerName: "anything",
  	}
+	return try.Start(newWebsocketDialer(cfg, opts))
+}
+
+// new WebsocketDialler returns a function that
+// can be passed to utils/parallel.Try.Start.
+func newWebsocketDialer(cfg *websocket.Config, opts DialOpts) func(<-chan  
struct{}) (io.Closer, error) {
  	openAttempt := utils.AttemptStrategy{
  		Total: opts.Timeout,
  		Delay: opts.RetryDelay,
  	}
-	return try.Start(func(stop <-chan struct{}) (io.Closer, error) {
-		err := parallel.ErrStopped
+	return func(stop <-chan struct{}) (io.Closer, error) {
  		for a := openAttempt.Start(); a.Next(); {
  			select {
  			case <-stop:
-				break
+				return nil, parallel.ErrStopped
  			default:
  			}
  			logger.Infof("dialing %q", cfg.Location)
-			var conn *websocket.Conn
-			conn, err = websocket.DialConfig(cfg)
+			conn, err := websocket.DialConfig(cfg)
  			if err == nil {
  				return conn, nil
  			}
-			logger.Debugf("error dialing API server, will retry: %v", err)
+			if a.HasNext() {
+				logger.Debugf("error dialing API server, will retry: %v", err)
+			} else {
+				return nil, fmt.Errorf("timed out connecting to %q", cfg.Location)
+			}
  		}
-		return nil, err
-	})
+		panic("unreachable")
+	}
  }

  func (s *State) heartbeatMonitor() {


Index: state/api/apiclient_test.go
=== modified file 'state/api/apiclient_test.go'
--- state/api/apiclient_test.go	2014-03-31 13:03:59 +0000
+++ state/api/apiclient_test.go	2014-04-01 03:13:49 +0000
@@ -11,6 +11,7 @@

  	jujutesting "launchpad.net/juju-core/juju/testing"
  	"launchpad.net/juju-core/state/api"
+	"launchpad.net/juju-core/utils/parallel"
  )

  type apiclientSuite struct {
@@ -75,5 +76,14 @@
  	addr := listener.Addr().String()
  	info.Addrs = []string{addr, addr, addr}
  	_, err = api.Open(info, api.DialOpts{})
-	c.Assert(err, gc.ErrorMatches, "websocket.Dial .*: read tcp .*:  
connection reset by peer")
+	c.Assert(err, gc.ErrorMatches, `timed out connecting to "wss://.*/"`)
+}
+
+func (s *apiclientSuite) TestDialWebsocketStopped(c *gc.C) {
+	stopped := make(chan struct{})
+	f := api.NewWebsocketDialer(nil, api.DialOpts{})
+	close(stopped)
+	result, err := f(stopped)
+	c.Assert(err, gc.Equals, parallel.ErrStopped)
+	c.Assert(result, gc.IsNil)
  }


Index: state/api/export_test.go
=== modified file 'state/api/export_test.go'
--- state/api/export_test.go	2013-12-17 15:30:13 +0000
+++ state/api/export_test.go	2014-04-01 03:13:49 +0000
@@ -3,6 +3,10 @@

  package api

+var (
+	NewWebsocketDialer = newWebsocketDialer
+)
+
  // SetServerRoot allows changing the URL to the internal API server
  // that AddLocalCharm uses in order to test NotImplementedError.
  func SetServerRoot(c *Client, root string) {