diff --git a/weed/command/s3.go b/weed/command/s3.go index 972b83bafd..f74bcda38b 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -739,6 +739,7 @@ type vectorDiscovery struct { conn *grpc.ClientConn client vector_pb.VectorServiceClient nextResolve time.Time + resolving chan struct{} // closed when the in-flight resolve lands } // reset drops the cached worker so the next call rediscovers — invoked on @@ -763,12 +764,23 @@ func isTransportError(err error) bool { return status.Code(err) == codes.Unavailable } -func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) { +// resolve runs without d.mu held so a slow master cannot stall cached-client callers. +func (d *vectorDiscovery) resolve() { + workers := d.discover(d.masters, d.grpcDialOption) d.mu.Lock() defer d.mu.Unlock() - workers := d.discover(d.masters, d.grpcDialOption) + // Measured from lookup end so a slow lookup cannot leave the next one already due. + defer func() { + interval := vectorRecheckInterval + if d.client != nil { + interval = vectorResolveInterval + } + d.nextResolve = time.Now().Add(interval) + close(d.resolving) + d.resolving = nil + }() if len(workers) == 0 { - return nil, errNoVectorWorker + return } worker := workers[0] for _, w := range workers { @@ -778,20 +790,17 @@ func (d *vectorDiscovery) resolve() (vector_pb.VectorServiceClient, error) { } } if worker != d.worker { + conn, err := pb.GrpcDial(context.Background(), worker, false, d.grpcDialOption) + if err != nil { + glog.Warningf("S3 Vectors: dial weed-vector %s: %v", worker, err) + return + } if d.conn != nil { d.conn.Close() } - conn, err := pb.GrpcDial(context.Background(), worker, false, d.grpcDialOption) - if err != nil { - return nil, err - } d.worker, d.conn = worker, conn d.client = vector_pb.NewVectorServiceClient(conn) - // cli() scheduled this resolve on the short recheck window while no - // client existed; a live client only refreshes every resolve interval. - d.nextResolve = time.Now().Add(vectorResolveInterval) } - return d.client, nil } // A live client refreshes membership at most every vectorResolveInterval. @@ -807,20 +816,25 @@ var errNoVectorWorker = status.Error(codes.Unavailable, "no weed-vector register func (d *vectorDiscovery) cli() (vector_pb.VectorServiceClient, error) { d.mu.Lock() - cli := d.client - if !time.Now().Before(d.nextResolve) { - interval := vectorRecheckInterval - if cli != nil { - interval = vectorResolveInterval - } - d.nextResolve = time.Now().Add(interval) - d.mu.Unlock() - return d.resolve() + if d.resolving == nil && !time.Now().Before(d.nextResolve) { + d.resolving = make(chan struct{}) + go d.resolve() } + cli, resolving := d.client, d.resolving d.mu.Unlock() if cli != nil { return cli, nil } + // No cached client: wait for the in-flight lookup instead of failing. + if resolving != nil { + <-resolving + d.mu.Lock() + cli = d.client + d.mu.Unlock() + if cli != nil { + return cli, nil + } + } return nil, errNoVectorWorker } diff --git a/weed/command/s3_vector_discovery_test.go b/weed/command/s3_vector_discovery_test.go index 3c6f561995..2a516e3599 100644 --- a/weed/command/s3_vector_discovery_test.go +++ b/weed/command/s3_vector_discovery_test.go @@ -1,6 +1,7 @@ package command import ( + "fmt" "sync/atomic" "testing" "time" @@ -8,6 +9,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "google.golang.org/grpc" "google.golang.org/grpc/codes" + "google.golang.org/grpc/connectivity" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" ) @@ -59,3 +61,185 @@ func TestVectorDiscoveryRechecksRegistration(t *testing.T) { t.Fatalf("live client rediscovered inside the refresh window: %d lookups", got) } } + +// A lookup slower than the throttle interval (an unreachable first master +// costs the full dial timeout) must not leave the next lookup already due, +// or every request queues behind back-to-back lookups. +func TestVectorDiscoveryThrottlesFromLookupEnd(t *testing.T) { + const lookup = 600 * time.Millisecond + var calls int32 + d := &vectorDiscovery{ + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + atomic.AddInt32(&calls, 1) + time.Sleep(lookup) + return nil + }, + } + + if _, err := d.cli(); status.Code(err) != codes.Unavailable { + t.Fatalf("no registered worker: expected Unavailable, got %v", err) + } + if _, err := d.cli(); status.Code(err) != codes.Unavailable { + t.Fatalf("no registered worker: expected Unavailable, got %v", err) + } + if got := atomic.LoadInt32(&calls); got != 1 { + t.Fatalf("slow lookup was repeated right away: %d lookups", got) + } + + d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + time.Sleep(lookup) + return []string{"127.0.0.1:9301"} + } + for i := 0; i < 2; i++ { + d.mu.Lock() + d.nextResolve = time.Time{} + d.mu.Unlock() + if _, err := d.cli(); err != nil { + t.Fatalf("worker registered: expected a client, got %v", err) + } + waitResolved(d) + } + d.mu.Lock() + left := time.Until(d.nextResolve) + d.mu.Unlock() + if left < vectorResolveInterval-lookup/2 { + t.Fatalf("refresh of a live worker is due in %v, want ~%v", left, vectorResolveInterval) + } +} + +func waitResolved(d *vectorDiscovery) { + d.mu.Lock() + resolving := d.resolving + d.mu.Unlock() + if resolving != nil { + <-resolving + } +} + +// A lookup stuck on an unreachable master must not hold up requests that +// already have a live worker, and concurrent callers start only one lookup. +func TestVectorDiscoveryLookupDoesNotBlockCachedClient(t *testing.T) { + d := &vectorDiscovery{ + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + return []string{"127.0.0.1:9301"} + }, + } + cached, err := d.cli() + if err != nil { + t.Fatalf("worker registered: expected a client, got %v", err) + } + + var calls int32 + entered, release := make(chan struct{}), make(chan struct{}) + d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + if atomic.AddInt32(&calls, 1) == 1 { + close(entered) + } + <-release + return []string{"127.0.0.1:9301"} + } + d.mu.Lock() + d.nextResolve = time.Time{} + d.mu.Unlock() + + const callers = 8 + got := make(chan error, callers+1) + call := func() { + c, err := d.cli() + if err == nil && c != cached { + err = fmt.Errorf("got a different client") + } + got <- err + } + go call() + <-entered + for i := 0; i < callers; i++ { + go call() + } + timeout := time.After(2 * time.Second) + for i := 0; i <= callers; i++ { + select { + case err := <-got: + if err != nil { + close(release) + t.Fatalf("cached client: %v", err) + } + case <-timeout: + close(release) + t.Fatalf("cached client blocked behind an in-flight lookup") + } + } + close(release) + waitResolved(d) + if n := atomic.LoadInt32(&calls); n != 1 { + t.Fatalf("concurrent callers ran %d lookups, want 1", n) + } +} + +// Callers with no worker wait for the in-flight lookup and share its result +// rather than failing while it runs. +func TestVectorDiscoveryUncachedCallersShareLookup(t *testing.T) { + var calls int32 + release := make(chan struct{}) + d := &vectorDiscovery{ + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + atomic.AddInt32(&calls, 1) + <-release + return []string{"127.0.0.1:9301"} + }, + } + const callers = 8 + got := make(chan error, callers) + for i := 0; i < callers; i++ { + go func() { + _, err := d.cli() + got <- err + }() + } + time.Sleep(50 * time.Millisecond) + close(release) + for i := 0; i < callers; i++ { + if err := <-got; err != nil { + t.Fatalf("caller waiting on the lookup: %v", err) + } + } + if n := atomic.LoadInt32(&calls); n != 1 { + t.Fatalf("concurrent callers ran %d lookups, want 1", n) + } +} + +// A failed dial to a new worker must leave the cached connection open. +func TestVectorDiscoveryKeepsConnOnDialError(t *testing.T) { + d := &vectorDiscovery{ + grpcDialOption: grpc.WithTransportCredentials(insecure.NewCredentials()), + discover: func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + return []string{"127.0.0.1:9301"} + }, + } + cached, err := d.cli() + if err != nil { + t.Fatalf("worker registered: expected a client, got %v", err) + } + + // Without transport credentials the dial fails synchronously. + d.grpcDialOption = nil + d.discover = func(masters []pb.ServerAddress, opt grpc.DialOption) []string { + return []string{"127.0.0.1:9302"} + } + d.mu.Lock() + d.nextResolve = time.Time{} + d.mu.Unlock() + if c, err := d.cli(); err != nil || c != cached { + t.Fatalf("cached client: got %v, %v", c, err) + } + waitResolved(d) + + d.mu.Lock() + defer d.mu.Unlock() + if d.client != cached || d.conn.GetState() == connectivity.Shutdown { + t.Fatalf("failed dial dropped the cached connection") + } +}