mirrors/tendermint
1
0
Fork 0
mirror of https://github.com/tendermint/tendermint.git synced 2026-08-16 12:16:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

Improve reconnections + fixing unit tests

Browse Source
This commit is contained in:
Juan Leni
2019-04-03 14:15:50 +02:00
parent 37cc8db430
commit ab033fbc81
7 changed files with 56 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
privval/errors.go
+2
View File
@@ -17,6 +17,8 @@ var (
ErrListenerTimeout = ListenerTimeoutError{}
ErrListenerNoConnection = fmt.Errorf("signer listening endpoint is not connected")
ErrDialerTimeout = fmt.Errorf("signer dialer endpoint timed out")
ErrDialerReadTimeout = fmt.Errorf("signer dialer endpoint read timed out")
ErrDialerWriteTimeout = fmt.Errorf("signer dialer endpoint write timed out")
)
// RemoteSignerError allows (remote) validators to include meaningful error descriptions in their reply.
privval/signer_client.go
+1
View File
@@ -54,6 +54,7 @@ func (sc *SignerClient) WaitForConnection(maxWait time.Duration) error {
func (sc *SignerClient) GetPubKey() crypto.PubKey {
response, err := sc.endpoint.SendRequest(&PubKeyRequest{})
if err != nil {
sc.endpoint.Logger.Error("error sending request", "err", err)
return nil
}
privval/signer_dialer_endpoint.go
+16 -13
View File
@@ -13,27 +13,24 @@ import (
)
const (
defaultMaxDialRetries = 10
defaultMaxDialRetries = 10
defaultRetryWaitMilliseconds = 100
)
// SignerServiceEndpointOption sets an optional parameter on the SignerDialerEndpoint.
type SignerServiceEndpointOption func(*SignerDialerEndpoint)
// SignerServiceEndpointTimeoutReadWrite sets the read and write timeout for connections
// SignerDialerEndpointTimeoutReadWrite sets the read and write timeout for connections
// from external signing processes.
func SignerServiceEndpointTimeoutReadWrite(timeout time.Duration) SignerServiceEndpointOption {
func SignerDialerEndpointTimeoutReadWrite(timeout time.Duration) SignerServiceEndpointOption {
return func(ss *SignerDialerEndpoint) { ss.timeoutReadWrite = timeout }
}
// SignerServiceEndpointConnRetries sets the amount of attempted retries to AcceptNewConnection.
func SignerServiceEndpointConnRetries(retries int) SignerServiceEndpointOption {
// SignerDialerEndpointConnRetries sets the amount of attempted retries to AcceptNewConnection.
func SignerDialerEndpointConnRetries(retries int) SignerServiceEndpointOption {
return func(ss *SignerDialerEndpoint) { ss.maxConnRetries = retries }
}
// TODO(jleni): Create a common type for a signerEndpoint (common for both listener/dialer)
// read
// write
// SignerDialerEndpoint dials using its dialer and responds to any
// signature requests using its privVal.
type SignerDialerEndpoint struct {
@@ -44,6 +41,7 @@ type SignerDialerEndpoint struct {
conn net.Conn
timeoutReadWrite time.Duration
retryWait time.Duration
maxConnRetries int
chainID string
@@ -66,6 +64,7 @@ func NewSignerDialerEndpoint(
se := &SignerDialerEndpoint{
dialer: dialer,
timeoutReadWrite: defaultTimeoutReadWriteSeconds * time.Second,
retryWait: defaultRetryWaitMilliseconds * time.Millisecond,
maxConnRetries: defaultMaxDialRetries,
chainID: chainID,
@@ -91,11 +90,12 @@ func (ss *SignerDialerEndpoint) OnStart() error {
// OnStop implements cmn.Service.
func (ss *SignerDialerEndpoint) OnStop() {
ss.Logger.Debug("SignerDialerEndpoint: OnStop calling Close")
_ = ss.Close()
// Stop service loop
close(ss.stopCh)
<-ss.stoppedCh
_ = ss.Close()
}
// Close closes the underlying net.Conn.
@@ -142,7 +142,7 @@ func (ss *SignerDialerEndpoint) readMessage() (msg RemoteSignerMsg, err error) {
const maxRemoteSignerMsgSize = 1024 * 10
_, err = cdc.UnmarshalBinaryLengthPrefixedReader(ss.conn, &msg, maxRemoteSignerMsgSize)
if _, ok := err.(timeoutError); ok {
err = cmn.ErrorWrap(ErrDialerTimeout, err.Error())
err = cmn.ErrorWrap(ErrDialerReadTimeout, err.Error())
}
return
@@ -165,7 +165,7 @@ func (ss *SignerDialerEndpoint) writeMessage(msg RemoteSignerMsg) (err error) {
_, err = cdc.MarshalBinaryLengthPrefixedWriter(ss.conn, msg)
if _, ok := err.(timeoutError); ok {
err = cmn.ErrorWrap(ErrDialerTimeout, err.Error())
err = cmn.ErrorWrap(ErrDialerWriteTimeout, err.Error())
}
return
@@ -224,10 +224,13 @@ func (ss *SignerDialerEndpoint) serviceLoop() {
if ss.conn == nil {
ss.conn, err = ss.dialer()
if err != nil {
ss.Logger.Info("Try connect", "err", err)
ss.conn = nil // Explicitly set to nil because dialer returns an interface (https://golang.org/doc/faq#nil_error)
retries++
// Wait between retries
time.Sleep(ss.retryWait)
continue
}
}
privval/signer_listener_endpoint.go
+21 -18
View File
@@ -18,7 +18,7 @@ type SignerValidatorEndpointOption func(*SignerListenerEndpoint)
type SignerListenerEndpoint struct {
cmn.BaseService
extMtx sync.Mutex
mtx sync.Mutex
listener net.Listener
conn net.Conn
@@ -71,43 +71,39 @@ func (ve *SignerListenerEndpoint) OnStop() {
}
}
ve.Logger.Debug("SignerListenerEndpoint: OnStop close stopCh")
// Stop service loop
ve.stopCh <- struct{}{}
<-ve.stoppedCh
}
// Close closes the underlying net.Conn.
func (ve *SignerListenerEndpoint) Close() error {
ve.extMtx.Lock()
defer ve.extMtx.Unlock()
ve.mtx.Lock()
defer ve.mtx.Unlock()
ve.Logger.Debug("SignerListenerEndpoint: Close")
ve.dropConnection()
ve.Logger.Debug("SignerListenerEndpoint: Closed")
return nil
}
// IsConnected indicates if there is an active connection
func (ve *SignerListenerEndpoint) IsConnected() bool {
ve.extMtx.Lock()
defer ve.extMtx.Unlock()
ve.mtx.Lock()
defer ve.mtx.Unlock()
return ve.isConnected()
}
// WaitForConnection waits maxWait for a connection or returns a timeout error
func (ve *SignerListenerEndpoint) WaitForConnection(maxWait time.Duration) error {
ve.extMtx.Lock()
defer ve.extMtx.Unlock()
ve.mtx.Lock()
defer ve.mtx.Unlock()
return ve.ensureConnection(maxWait)
}
// SendRequest sends a request and waits for a response
func (ve *SignerListenerEndpoint) SendRequest(request RemoteSignerMsg) (RemoteSignerMsg, error) {
ve.extMtx.Lock()
defer ve.extMtx.Unlock()
ve.mtx.Lock()
defer ve.mtx.Unlock()
// TODO: Add retries.. that include dropping the connection and
@@ -237,19 +233,16 @@ func (ve *SignerListenerEndpoint) ensureConnection(maxWait time.Duration) error
// dropConnection closes the current connection but does not touch the listening socket
func (ve *SignerListenerEndpoint) dropConnection() {
ve.Logger.Debug("SignerListenerEndpoint: dropConnection")
if ve.conn != nil {
if err := ve.conn.Close(); err != nil {
ve.Logger.Error("SignerListenerEndpoint::dropConnection", "err", err)
}
ve.conn = nil
}
ve.Logger.Debug("SignerListenerEndpoint: dropConnection DONE")
}
func (ve *SignerListenerEndpoint) serviceLoop() {
defer close(ve.stoppedCh)
defer ve.Logger.Debug("SignerListenerEndpoint::serviceLoop EXIT")
ve.Logger.Debug("SignerListenerEndpoint::serviceLoop")
for {
@@ -261,8 +254,18 @@ func (ve *SignerListenerEndpoint) serviceLoop() {
conn, err := ve.acceptNewConnection()
if err == nil {
ve.Logger.Info("Connected")
ve.connectedCh <- conn
break
select {
case ve.connectedCh <- conn:
{
ve.Logger.Debug("SignerListenerEndpoint: connection relayed")
}
case <-ve.stopCh:
{
ve.Logger.Debug("SignerListenerEndpoint: stopping")
return
}
}
}
}
}
privval/signer_listener_endpoint_test.go
+7 -7
View File
@@ -66,8 +66,8 @@ func TestSignerRemoteRetryTCPOnly(t *testing.T) {
)
defer serviceEndpoint.Stop()
SignerServiceEndpointTimeoutReadWrite(time.Millisecond)(serviceEndpoint)
SignerServiceEndpointConnRetries(retries)(serviceEndpoint)
SignerDialerEndpointTimeoutReadWrite(time.Millisecond)(serviceEndpoint)
SignerDialerEndpointConnRetries(retries)(serviceEndpoint)
err = serviceEndpoint.Start()
assert.NoError(t, err)
@@ -75,7 +75,7 @@ func TestSignerRemoteRetryTCPOnly(t *testing.T) {
select {
case attempts := <-attemptCh:
assert.Equal(t, retries, attempts)
case <-time.After(100 * time.Millisecond):
case <-time.After(1500 * time.Millisecond):
t.Error("expected remote to observe connection attempts")
}
}
@@ -98,8 +98,8 @@ func TestRetryConnToRemoteSigner(t *testing.T) {
validatorEndpoint = newSignerValidatorEndpoint(logger, tc.addr, thisConnTimeout)
)
SignerServiceEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint)
SignerServiceEndpointConnRetries(10)(serviceEndpoint)
SignerDialerEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint)
SignerDialerEndpointConnRetries(10)(serviceEndpoint)
getStartEndpoint(t, readyCh, validatorEndpoint)
defer validatorEndpoint.Stop()
@@ -188,8 +188,8 @@ func getMockEndpoints(
validatorEndpoint = newSignerValidatorEndpoint(logger, addr, testTimeoutReadWrite)
)
SignerServiceEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint)
SignerServiceEndpointConnRetries(1e6)(serviceEndpoint)
SignerDialerEndpointTimeoutReadWrite(testTimeoutReadWrite)(serviceEndpoint)
SignerDialerEndpointConnRetries(1e6)(serviceEndpoint)
getStartEndpoint(t, readyCh, validatorEndpoint)