From 8602132ab6283ae6bb7d947fc362b9c3dd17b725 Mon Sep 17 00:00:00 2001 From: Peter Bourgon Date: Thu, 24 Sep 2015 16:56:37 +0200 Subject: [PATCH] Move concurrency from resolve to HTTP GET - Process DNS resolution serially - Process up to 10 HTTP GET (for app ID) concurrently More than 10 concurrent GET requests will block on the semaphore. This will cause the staticResolver.resolve method to block, which is probably fine: it will just delay the next resolve loop, currently at 1m intervals. To make this a little bit more robust, I've also added a fastClient for app ID resolution, with a timeout (total, including connect, request, and response) of 5s. --- probe/resolver.go | 24 ++-------------- probe/resolver_test.go | 34 ----------------------- xfer/http_publisher.go | 7 ++++- xfer/multi_publisher.go | 36 ++++++++++++++++++++---- xfer/multi_publisher_internal_test.go | 40 +++++++++++++++++++++++++++ 5 files changed, 80 insertions(+), 61 deletions(-) create mode 100644 xfer/multi_publisher_internal_test.go diff --git a/probe/resolver.go b/probe/resolver.go index 1d99e41c0..58119497a 100644 --- a/probe/resolver.go +++ b/probe/resolver.go @@ -15,12 +15,9 @@ var ( lookupIP = net.LookupIP ) -const maxConcurrentLookup = 10 - type staticResolver struct { set func(string, []string) targets []target - sema semaphore quit chan struct{} } @@ -35,7 +32,6 @@ func newStaticResolver(targets []string, set func(target string, endpoints []str r := staticResolver{ targets: prepare(targets), set: set, - sema: newSemaphore(maxConcurrentLookup), quit: make(chan struct{}), } go r.loop() @@ -79,17 +75,15 @@ func prepare(strs []string) []target { } func (r staticResolver) resolve() { - for t, endpoints := range resolveMany(r.sema, r.targets) { + for t, endpoints := range resolveMany(r.targets) { r.set(t.String(), endpoints) } } -func resolveMany(s semaphore, targets []target) map[target][]string { +func resolveMany(targets []target) map[target][]string { result := map[target][]string{} for _, t := range targets { - c := make(chan []string) - go func(t target) { s.p(); defer s.v(); c <- resolveOne(t) }(t) - result[t] = <-c + result[t] = resolveOne(t) } return result } @@ -115,15 +109,3 @@ func resolveOne(t target) []string { } return endpoints } - -type semaphore chan struct{} - -func newSemaphore(n int) semaphore { - c := make(chan struct{}, n) - for i := 0; i < n; i++ { - c <- struct{}{} - } - return semaphore(c) -} -func (s semaphore) p() { <-s } -func (s semaphore) v() { s <- struct{}{} } diff --git a/probe/resolver_test.go b/probe/resolver_test.go index 88e5807aa..4463e8f22 100644 --- a/probe/resolver_test.go +++ b/probe/resolver_test.go @@ -96,40 +96,6 @@ func TestResolver(t *testing.T) { } } -func TestSemaphore(t *testing.T) { - n := 3 - s := newSemaphore(n) - - // First n should be fine - for i := 0; i < n; i++ { - ok := make(chan struct{}) - go func() { s.p(); close(ok) }() - select { - case <-ok: - case <-time.After(10 * time.Millisecond): - t.Errorf("p (%d) failed", i+1) - } - } - - // This should block - ok := make(chan struct{}) - go func() { s.p(); close(ok) }() - select { - case <-ok: - t.Errorf("%dth p OK, but should block", n+1) - case <-time.After(10 * time.Millisecond): - //t.Logf("%dth p blocks, as expected", n+1) - } - - s.v() - - select { - case <-ok: - case <-time.After(10 * time.Millisecond): - t.Errorf("%dth p didn't resolve in time", n+1) - } -} - func makeIPs(addrs ...string) []net.IP { var ips []net.IP for _, addr := range addrs { diff --git a/xfer/http_publisher.go b/xfer/http_publisher.go index bb9aa3f57..8ff6e2f13 100644 --- a/xfer/http_publisher.go +++ b/xfer/http_publisher.go @@ -5,6 +5,7 @@ import ( "fmt" "io" "net/http" + "time" "github.com/weaveworks/scope/common/sanitize" ) @@ -16,10 +17,14 @@ type HTTPPublisher struct { probeID string } +var fastClient = http.Client{ + Timeout: 5 * time.Second, +} + // NewHTTPPublisher returns an HTTPPublisher ready for use. func NewHTTPPublisher(target, token, probeID string) (string, *HTTPPublisher, error) { targetAPI := sanitize.URL("http://", 0, "/api")(target) - resp, err := http.Get(targetAPI) + resp, err := fastClient.Get(targetAPI) if err != nil { return "", nil, err } diff --git a/xfer/multi_publisher.go b/xfer/multi_publisher.go index 88d1455aa..532525600 100644 --- a/xfer/multi_publisher.go +++ b/xfer/multi_publisher.go @@ -15,6 +15,7 @@ import ( type MultiPublisher struct { mtx sync.Mutex factory func(endpoint string) (string, Publisher, error) + sema semaphore list []tuple } @@ -22,6 +23,7 @@ type MultiPublisher struct { func NewMultiPublisher(factory func(endpoint string) (string, Publisher, error)) *MultiPublisher { return &MultiPublisher{ factory: factory, + sema: newSemaphore(maxConcurrentGET), } } @@ -30,8 +32,11 @@ type tuple struct { target string // DNS name endpoint string // IP addr id string // unique ID from app + err error // if factory failed } +const maxConcurrentGET = 10 + // Set declares that the target (DNS name) resolves to the provided endpoints // (IPs), and that we want to publish to each of those endpoints. Set replaces // any existing publishers to the given target. Set invokes the factory method @@ -39,14 +44,23 @@ type tuple struct { // unique ID. func (p *MultiPublisher) Set(target string, endpoints []string) { // Convert endpoints to publishers. - list := make([]tuple, 0, len(p.list)+len(endpoints)) + c := make(chan tuple, len(endpoints)) for _, endpoint := range endpoints { - id, publisher, err := p.factory(endpoint) - if err != nil { - log.Printf("multi-publisher set: %s (%s): %v", target, endpoint, err) + go func(endpoint string) { + p.sema.p() + defer p.sema.v() + id, publisher, err := p.factory(endpoint) + c <- tuple{publisher, target, endpoint, id, err} + }(endpoint) + } + list := make([]tuple, 0, len(p.list)+len(endpoints)) + for i := 0; i < cap(c); i++ { + t := <-c + if t.err != nil { + log.Printf("multi-publisher set: %s (%s): %v", t.target, t.endpoint, t.err) continue } - list = append(list, tuple{publisher, target, endpoint, id}) + list = append(list, t) } // Copy all other tuples over to the new list. @@ -115,3 +129,15 @@ func (p *MultiPublisher) appendFilter(list []tuple, f func(tuple) bool) []tuple } return list } + +type semaphore chan struct{} + +func newSemaphore(n int) semaphore { + c := make(chan struct{}, n) + for i := 0; i < n; i++ { + c <- struct{}{} + } + return semaphore(c) +} +func (s semaphore) p() { <-s } +func (s semaphore) v() { s <- struct{}{} } diff --git a/xfer/multi_publisher_internal_test.go b/xfer/multi_publisher_internal_test.go new file mode 100644 index 000000000..4400b1b2e --- /dev/null +++ b/xfer/multi_publisher_internal_test.go @@ -0,0 +1,40 @@ +package xfer + +import ( + "testing" + "time" +) + +func TestSemaphore(t *testing.T) { + n := 3 + s := newSemaphore(n) + + // First n should be fine + for i := 0; i < n; i++ { + ok := make(chan struct{}) + go func() { s.p(); close(ok) }() + select { + case <-ok: + case <-time.After(10 * time.Millisecond): + t.Errorf("p (%d) failed", i+1) + } + } + + // This should block + ok := make(chan struct{}) + go func() { s.p(); close(ok) }() + select { + case <-ok: + t.Errorf("%dth p OK, but should block", n+1) + case <-time.After(10 * time.Millisecond): + //t.Logf("%dth p blocks, as expected", n+1) + } + + s.v() + + select { + case <-ok: + case <-time.After(10 * time.Millisecond): + t.Errorf("%dth p didn't resolve in time", n+1) + } +}