mirror of
https://github.com/etcd-io/etcd.git
synced 2024-09-27 06:25:44 +00:00
clientv3: check if KV.Client is closed
For https://github.com/coreos/etcd/issues/5495.
This commit is contained in:
parent
d116c116fe
commit
7b5657cf1a
@ -384,5 +384,5 @@ func dialEndpointList(c *Client) (*grpc.ClientConn, error) {
|
||||
// progress can be made, even after reconnecting.
|
||||
func isHaltErr(ctx context.Context, err error) bool {
|
||||
isRPCError := strings.HasPrefix(grpc.ErrorDesc(err), "etcdserver: ")
|
||||
return isRPCError || ctx.Err() != nil
|
||||
return isRPCError || ctx.Err() != nil || err == rpctypes.ErrConnClosed
|
||||
}
|
||||
|
@ -279,6 +279,42 @@ func TestKVRange(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestKVGetErrConnClosed(t *testing.T) {
|
||||
defer testutil.AfterTest(t)
|
||||
|
||||
clus := integration.NewClusterV3(t, &integration.ClusterConfig{Size: 1})
|
||||
defer clus.Terminate(t)
|
||||
|
||||
cli := clus.Client(0)
|
||||
kv := clientv3.NewKV(cli)
|
||||
|
||||
closed, donec := make(chan struct{}), make(chan struct{})
|
||||
go func() {
|
||||
select {
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("cli.Close took too long")
|
||||
case <-closed:
|
||||
}
|
||||
|
||||
if _, err := kv.Get(context.TODO(), "foo"); err != rpctypes.ErrConnClosed {
|
||||
t.Fatalf("expected %v, got %v", rpctypes.ErrConnClosed, err)
|
||||
}
|
||||
close(donec)
|
||||
}()
|
||||
|
||||
if err := cli.Close(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
clus.TakeClient(0)
|
||||
close(closed)
|
||||
|
||||
select {
|
||||
case <-time.After(3 * time.Second):
|
||||
t.Fatal("kv.Get took too long")
|
||||
case <-donec:
|
||||
}
|
||||
}
|
||||
|
||||
func TestKVDeleteRange(t *testing.T) {
|
||||
defer testutil.AfterTest(t)
|
||||
|
||||
|
@ -17,6 +17,8 @@ package clientv3
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"github.com/coreos/etcd/etcdserver/api/v3rpc/rpctypes"
|
||||
|
||||
"golang.org/x/net/context"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
@ -81,10 +83,14 @@ func (r *remoteClient) tryUpdate() bool {
|
||||
func (r *remoteClient) acquire(ctx context.Context) error {
|
||||
for {
|
||||
r.client.mu.RLock()
|
||||
closed := r.client.cancel == nil
|
||||
c := r.client.conn
|
||||
r.mu.Lock()
|
||||
match := r.conn == c
|
||||
r.mu.Unlock()
|
||||
if closed {
|
||||
return rpctypes.ErrConnClosed
|
||||
}
|
||||
if match {
|
||||
return nil
|
||||
}
|
||||
|
Loading…
x
Reference in New Issue
Block a user