diff --git a/weed/cluster/lock_client.go b/weed/cluster/lock_client.go index 1da611cf25..5b3c76cb6f 100644 --- a/weed/cluster/lock_client.go +++ b/weed/cluster/lock_client.go @@ -435,11 +435,12 @@ func (lock *LiveLock) doLock(lockDuration, attemptTimeout time.Duration) (errorM previousHostFiler := lock.hostFiler previousOwner := lock.owner - err = pb.WithFilerClient(false, 0, lock.hostFiler, lock.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { - // Bound the attempt so an unreachable owner fails within the retry - // window. - ctx, cancel := context.WithTimeout(context.Background(), attemptTimeout) - defer cancel() + // Bound the whole attempt — connection acquisition included — so an + // unreachable owner fails within the retry window. + ctx, cancel := context.WithTimeout(context.Background(), attemptTimeout) + defer cancel() + err = pb.WithGrpcClient(ctx, false, 0, func(grpcConnection *grpc.ClientConn) error { + client := filer_pb.NewSeaweedFilerClient(grpcConnection) resp, err := client.DistributedLock(ctx, &filer_pb.LockRequest{ Name: lock.key, SecondsToLock: int64(lockDuration.Seconds()), @@ -484,7 +485,7 @@ func (lock *LiveLock) doLock(lockDuration, attemptTimeout time.Duration) (errorM } } return err - }) + }, lock.hostFiler.ToGrpcAddress(), false, lock.grpcDialOption) if err != nil && !lock.hostFiler.Equals(lock.lc.seedFiler) { lock.consecutiveFailures++ diff --git a/weed/pb/grpc_client_server.go b/weed/pb/grpc_client_server.go index 2a119c075e..f8f5837d8a 100644 --- a/weed/pb/grpc_client_server.go +++ b/weed/pb/grpc_client_server.go @@ -272,7 +272,7 @@ func GrpcDial(ctx context.Context, address string, waitForReady bool, opts ...gr return grpc.DialContext(ctx, address, options...) } -func getOrCreateConnection(address string, waitForReady bool, opts ...grpc.DialOption) (*versionedGrpcClient, error) { +func getOrCreateConnection(ctx context.Context, address string, waitForReady bool, opts ...grpc.DialOption) (*versionedGrpcClient, error) { grpcClientsLock.Lock() defer grpcClientsLock.Unlock() @@ -284,7 +284,6 @@ func getOrCreateConnection(address string, waitForReady bool, opts ...grpc.DialO } glog.V(2).Infof("Creating new gRPC connection to %s", address) - ctx := context.Background() grpcConnection, err := GrpcDial(ctx, address, waitForReady, opts...) if err != nil { return nil, fmt.Errorf("fail to dial %s: %v", address, err) @@ -468,15 +467,16 @@ func shouldInvalidateConnection(ctx context.Context, err error) bool { } // WithGrpcClient In streamingMode, always use a fresh connection. Otherwise, try to reuse an existing connection. -// ctx must be the context that bounds the RPC fn issues, so that a -// Canceled/DeadlineExceeded fired by ctx itself does not tear down the shared -// cached ClientConn out from under other concurrent callers. Pass -// context.Background() when fn picks its own context: a Canceled/DeadlineExceeded -// then never invalidates, since nothing here can attribute it to the channel. +// ctx must be the context that bounds this call's connection acquisition and +// the RPC fn issues, so that a Canceled/DeadlineExceeded fired by ctx itself +// does not tear down the shared cached ClientConn out from under other +// concurrent callers. Pass context.Background() when fn picks its own +// context: a Canceled/DeadlineExceeded then never invalidates, since nothing +// here can attribute it to the channel. func WithGrpcClient(ctx context.Context, streamingMode bool, signature int32, fn func(*grpc.ClientConn) error, address string, waitForReady bool, opts ...grpc.DialOption) error { if !streamingMode { - vgc, err := getOrCreateConnection(address, waitForReady, opts...) + vgc, err := getOrCreateConnection(ctx, address, waitForReady, opts...) if err != nil { return fmt.Errorf("getOrCreateConnection %s: %v", address, err) } @@ -498,7 +498,7 @@ func WithGrpcClient(ctx context.Context, streamingMode bool, signature int32, fn } // Streaming mode: dedicate a fresh ClientConn to this call. - dialCtx := context.Background() + dialCtx := ctx if signature != 0 { // Optimize: Use AppendToOutgoingContext instead of creating new map dialCtx = metadata.AppendToOutgoingContext(dialCtx, "sw-client-id", fmt.Sprintf("%d", signature))