[#301] rpc: Make client stream initialization get cancelled by dial timeout
All checks were successful
All checks were successful
* `c.conn` may be already invalidated but the rpc client can't detect this. `NewStream` may hang trying to open a stream with invalidated connection. Using timer with `dialTimeout` for `NewStream` fixes this problem. Signed-off-by: Airat Arifullin <a.arifullin@yadro.com>
This commit is contained in:
parent
2c79f770e4
commit
8ea650ac76
1 changed files with 47 additions and 8 deletions
|
@ -3,6 +3,7 @@ package client
|
||||||
import (
|
import (
|
||||||
"context"
|
"context"
|
||||||
"io"
|
"io"
|
||||||
|
"time"
|
||||||
|
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/common"
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/common"
|
||||||
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/message"
|
"git.frostfs.info/TrueCloudLab/frostfs-api-go/v2/rpc/message"
|
||||||
|
@ -51,18 +52,56 @@ func (c *Client) Init(info common.CallMethodInfo, opts ...CallOption) (MessageRe
|
||||||
}
|
}
|
||||||
|
|
||||||
ctx, cancel := context.WithCancel(prm.ctx)
|
ctx, cancel := context.WithCancel(prm.ctx)
|
||||||
stream, err := c.conn.NewStream(ctx, &grpc.StreamDesc{
|
|
||||||
StreamName: info.Name,
|
// `conn.NewStream` doesn't check if `conn` may turn up invalidated right before this invocation.
|
||||||
ServerStreams: info.ServerStream(),
|
// In such cases, the operation can hang indefinitely, with the context timeout being the only
|
||||||
ClientStreams: info.ClientStream(),
|
// mechanism to cancel it.
|
||||||
}, toMethodName(info))
|
//
|
||||||
if err != nil {
|
// We use a separate timer instead of context timeout because the latter
|
||||||
|
// would propagate to all subsequent read/write operations on the opened stream,
|
||||||
|
// which is not desired for the stream's lifecycle management.
|
||||||
|
dialTimeoutTimer := time.NewTimer(c.dialTimeout)
|
||||||
|
defer dialTimeoutTimer.Stop()
|
||||||
|
|
||||||
|
type newStreamRes struct {
|
||||||
|
stream grpc.ClientStream
|
||||||
|
err error
|
||||||
|
}
|
||||||
|
newStreamCh := make(chan newStreamRes)
|
||||||
|
|
||||||
|
go func() {
|
||||||
|
stream, err := c.conn.NewStream(ctx, &grpc.StreamDesc{
|
||||||
|
StreamName: info.Name,
|
||||||
|
ServerStreams: info.ServerStream(),
|
||||||
|
ClientStreams: info.ClientStream(),
|
||||||
|
}, toMethodName(info))
|
||||||
|
|
||||||
|
newStreamCh <- newStreamRes{
|
||||||
|
stream: stream,
|
||||||
|
err: err,
|
||||||
|
}
|
||||||
|
}()
|
||||||
|
|
||||||
|
var res newStreamRes
|
||||||
|
|
||||||
|
select {
|
||||||
|
case <-dialTimeoutTimer.C:
|
||||||
cancel()
|
cancel()
|
||||||
return nil, err
|
res = <-newStreamCh
|
||||||
|
if res.stream != nil && res.err == nil {
|
||||||
|
_ = res.stream.CloseSend()
|
||||||
|
}
|
||||||
|
return nil, context.Canceled
|
||||||
|
case res = <-newStreamCh:
|
||||||
|
}
|
||||||
|
|
||||||
|
if res.err != nil {
|
||||||
|
cancel()
|
||||||
|
return nil, res.err
|
||||||
}
|
}
|
||||||
|
|
||||||
return &streamWrapper{
|
return &streamWrapper{
|
||||||
ClientStream: stream,
|
ClientStream: res.stream,
|
||||||
cancel: cancel,
|
cancel: cancel,
|
||||||
timeout: c.rwTimeout,
|
timeout: c.rwTimeout,
|
||||||
}, nil
|
}, nil
|
||||||
|
|
Loading…
Reference in a new issue