Decouple peer name from URI

Keep the existing `Set` method as-is, but make it a wrapper around the
code.

Also, fix the new peerpicker.set() so it doesn't leak connections if any
Close() call fails.

galaxycache.go

@@ -96,15 +96,15 @@ type Universe struct {
 // NewUniverse is the main constructor for the Universe object. It is passed a
 // FetchProtocol (to specify fetching via GRPC or HTTP) and its own URL along
 // with options.
-func NewUniverse(protocol FetchProtocol, selfURL string, opts ...UniverseOpt) *Universe {
+func NewUniverse(protocol FetchProtocol, selfID string, opts ...UniverseOpt) *Universe {
 	options := &universeOpts{}
 	for _, opt := range opts {
 		opt(options)
 	}
 
 	c := &Universe{
 		galaxies:   make(map[string]*Galaxy),
-		peerPicker: newPeerPicker(protocol, selfURL, options.hashOpts),
+		peerPicker: newPeerPicker(protocol, selfID, options.hashOpts),
 		recorder:   options.recorder,
 	}
 
@@ -114,8 +114,8 @@ func NewUniverse(protocol FetchProtocol, selfURL string, opts ...UniverseOpt) *U
 // NewUniverseWithOpts is a deprecated constructor for the Universe object that
 // defines a non-default hash function and number of replicas.  Please use
 // `NewUniverse` with the `WithHashOpts` option instead.
-func NewUniverseWithOpts(protocol FetchProtocol, selfURL string, options *HashOptions) *Universe {
-	return NewUniverse(protocol, selfURL, WithHashOpts(options))
+func NewUniverseWithOpts(protocol FetchProtocol, selfID string, options *HashOptions) *Universe {
+	return NewUniverse(protocol, selfID, WithHashOpts(options))
 }
 
 // NewGalaxy creates a coordinated galaxy-aware BackendGetter from a
@@ -197,7 +197,7 @@ func (universe *Universe) GetGalaxy(name string) *Galaxy {
 // Each PeerURL value should be a valid base URL,
 // for example "example.net:8000".
 func (universe *Universe) Set(peerURLs ...string) error {
-	return universe.peerPicker.set(peerURLs...)
+	return universe.peerPicker.setURLs(peerURLs...)
 }
 
 // Shutdown closes all open fetcher connections

galaxycache_test.go

@@ -45,7 +45,7 @@ const (
 )
 
 func initSetup() (*Universe, context.Context, chan string) {
-	return NewUniverse(&TestProtocol{}, "test"), context.TODO(), make(chan string)
+	return NewUniverse(&TestProtocol{TestFetchers: map[string]*TestFetcher{}}, "test"), context.TODO(), make(chan string)
 }
 
 func setupStringGalaxyTest(cacheFills *AtomicInt) (*Galaxy, context.Context, chan string) {
@@ -222,13 +222,17 @@ func TestPeers(t *testing.T) {
 			numGets:      200,
 			expectedHits: map[string]int{"fetcher0": 50, "fetcher1": 50, "fetcher2": 50, "fetcher3": 50},
 			cacheSize:    1 << 20,
+			initFunc: func(g *Galaxy, fetchers map[string]*TestFetcher) {
+				fetchers["fetcher0"] = &TestFetcher{}
+			},
 		},
 		{
 			testName:     "cached_base",
 			numGets:      200,
 			expectedHits: map[string]int{"fetcher0": 0, "fetcher1": 48, "fetcher2": 47, "fetcher3": 48},
 			cacheSize:    1 << 20,
-			initFunc: func(g *Galaxy, _ map[string]*TestFetcher) {
+			initFunc: func(g *Galaxy, fetchers map[string]*TestFetcher) {
+				fetchers["fetcher0"] = &TestFetcher{}
 				for i := 0; i < 200; i++ {
 					key := fmt.Sprintf("%d", i)
 					var got StringCodec
@@ -246,6 +250,7 @@ func TestPeers(t *testing.T) {
 			expectedHits: map[string]int{"fetcher0": 100, "fetcher1": 50, "fetcher2": 0, "fetcher3": 50},
 			cacheSize:    1 << 20,
 			initFunc: func(g *Galaxy, fetchers map[string]*TestFetcher) {
+				fetchers["fetcher0"] = &TestFetcher{}
 				fetchers["fetcher2"].fail = true
 			},
 		},
@@ -266,7 +271,9 @@ func TestPeers(t *testing.T) {
 			universe := NewUniverseWithOpts(testproto, "fetcher0", hashOpts)
 			dummyCtx := context.TODO()
 
-			universe.Set("fetcher0", "fetcher1", "fetcher2", "fetcher3")
+			if setErr := universe.Set("fetcher0", "fetcher1", "fetcher2", "fetcher3"); setErr != nil {
+				t.Fatalf("failed to set peers on universe: %s", setErr)
+			}
 
 			getter := func(_ context.Context, key string, dest Codec) error {
 				// these are local hits
@@ -688,7 +695,7 @@ func TestRecorder(t *testing.T) {
 	getter := func(_ context.Context, key string, dest Codec) error {
 		return dest.UnmarshalBinary([]byte("got:" + key))
 	}
-	u := NewUniverse(&TestProtocol{}, "test-universe", WithRecorder(meter))
+	u := NewUniverse(&TestProtocol{TestFetchers: map[string]*TestFetcher{}}, "test-universe", WithRecorder(meter))
 	g := u.NewGalaxy("test", 1024, GetterFunc(getter))
 	var s StringCodec
 	err := g.Get(context.Background(), "foo", &s)

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/golang/protobuf v1.4.3
 	github.com/vimeo/go-clocks v1.1.2
 	go.opencensus.io v0.22.5
+	golang.org/x/sync v0.1.0
 	google.golang.org/grpc v1.35.0
 )
 

go.sum

@@ -57,6 +57,8 @@ golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJ
 golang.org/x/sync v0.0.0-20181108010431-42b317875d0f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190227155943-e225da77a7e6/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.1.0 h1:wsuoTGHzEhffawBOhz5CYhcrV4IdKZbEyZjBMuTp12o=
+golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190502145724-3ef323f4f1fd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=