fix: client connection error

This commit is contained in:
finley
2022-07-03 17:56:30 +08:00
parent 380d936ed5
commit 45edbdd7c7
2 changed files with 72 additions and 33 deletions

View File

@@ -1,6 +1,7 @@
package client package client
import ( import (
"errors"
"github.com/hdt3213/godis/interface/redis" "github.com/hdt3213/godis/interface/redis"
"github.com/hdt3213/godis/lib/logger" "github.com/hdt3213/godis/lib/logger"
"github.com/hdt3213/godis/lib/sync/wait" "github.com/hdt3213/godis/lib/sync/wait"
@@ -8,6 +9,7 @@ import (
"github.com/hdt3213/godis/redis/protocol" "github.com/hdt3213/godis/redis/protocol"
"net" "net"
"runtime/debug" "runtime/debug"
"strings"
"sync" "sync"
"time" "time"
) )
@@ -57,12 +59,7 @@ func MakeClient(addr string) (*Client, error) {
func (client *Client) Start() { func (client *Client) Start() {
client.ticker = time.NewTicker(10 * time.Second) client.ticker = time.NewTicker(10 * time.Second)
go client.handleWrite() go client.handleWrite()
go func() { go client.handleRead()
err := client.handleRead()
if err != nil {
logger.Error(err)
}
}()
go client.heartbeat() go client.heartbeat()
} }
@@ -80,27 +77,35 @@ func (client *Client) Close() {
close(client.waitingReqs) close(client.waitingReqs)
} }
func (client *Client) handleConnectionError(err error) error { func (client *Client) reconnect() {
err1 := client.conn.Close() _ = client.conn.Close() // ignore possible errors from repeated closes
if err1 != nil {
if opErr, ok := err1.(*net.OpError); ok { var conn net.Conn
if opErr.Err.Error() != "use of closed network connection" { for i := 0; i < 3; i++ {
return err1 var err error
} conn, err = net.Dial("tcp", client.addr)
if err != nil {
logger.Error("reconnect error: " + err.Error())
time.Sleep(time.Second)
continue
} else { } else {
return err1 break
} }
} }
conn, err1 := net.Dial("tcp", client.addr) if conn == nil { // reach max retry, abort
if err1 != nil { client.Close()
logger.Error(err1) return
return err1
} }
client.conn = conn client.conn = conn
go func() { //
_ = client.handleRead() close(client.waitingReqs)
}() for req := range client.waitingReqs {
return nil req.err = errors.New("connection closed")
req.waiting.Done()
}
client.waitingReqs = make(chan *request, chanSize)
// restart handle read
go client.handleRead()
} }
func (client *Client) heartbeat() { func (client *Client) heartbeat() {
@@ -155,14 +160,14 @@ func (client *Client) doRequest(req *request) {
} }
re := protocol.MakeMultiBulkReply(req.args) re := protocol.MakeMultiBulkReply(req.args)
bytes := re.ToBytes() bytes := re.ToBytes()
_, err := client.conn.Write(bytes) var err error
i := 0 for i := 0; i < 3; i++ { // only retry, waiting for handleRead
for err != nil && i < 3 { _, err = client.conn.Write(bytes)
err = client.handleConnectionError(err) if err == nil ||
if err == nil { (!strings.Contains(err.Error(), "timeout") && // only retry timeout
_, err = client.conn.Write(bytes) !strings.Contains(err.Error(), "deadline exceeded")) {
break
} }
i++
} }
if err == nil { if err == nil {
client.waitingReqs <- req client.waitingReqs <- req
@@ -189,14 +194,13 @@ func (client *Client) finishRequest(reply redis.Reply) {
} }
} }
func (client *Client) handleRead() error { func (client *Client) handleRead() {
ch := parser.ParseStream(client.conn) ch := parser.ParseStream(client.conn)
for payload := range ch { for payload := range ch {
if payload.Err != nil { if payload.Err != nil {
client.finishRequest(protocol.MakeErrReply(payload.Err.Error())) client.reconnect()
continue return
} }
client.finishRequest(payload.Data) client.finishRequest(payload.Data)
} }
return nil
} }

View File

@@ -1,10 +1,12 @@
package client package client
import ( import (
"bytes"
"github.com/hdt3213/godis/lib/logger" "github.com/hdt3213/godis/lib/logger"
"github.com/hdt3213/godis/redis/protocol" "github.com/hdt3213/godis/redis/protocol"
"strconv" "strconv"
"testing" "testing"
"time"
) )
func TestClient(t *testing.T) { func TestClient(t *testing.T) {
@@ -104,3 +106,36 @@ func TestClient(t *testing.T) {
client.Close() client.Close()
} }
func TestReconnect(t *testing.T) {
logger.Setup(&logger.Settings{
Path: "logs",
Name: "godis",
Ext: ".log",
TimeFormat: "2006-01-02",
})
client, err := MakeClient("localhost:6379")
if err != nil {
t.Error(err)
}
client.Start()
_ = client.conn.Close()
time.Sleep(time.Second) // wait for reconnecting
success := false
for i := 0; i < 3; i++ {
result := client.Send([][]byte{
[]byte("PING"),
})
if bytes.Equal(result.ToBytes(), []byte("+PONG\r\n")) {
success = true
break
}
}
if !success {
t.Error("reconnect error")
}
//var wg sync.WaitGroup
//wg.Add(1)
//wg.Wait()
}