Optimizing internal rpc

Author: dobyte

internal/transporter/internal/client/client.go

@@ -8,12 +8,13 @@ import (
 
 	"github.com/dobyte/due/v2/core/buffer"
 	"github.com/dobyte/due/v2/errors"
+	"github.com/dobyte/due/v2/log"
 	"golang.org/x/sync/errgroup"
 )
 
 const (
 	defaultTimeout = 3 * time.Second // 调用超时时间
-	defaultConnNum = 20              // 默认连接数
+	defaultConnNum = 10              // 默认连接数
 )
 
 type chWrite struct {
@@ -57,6 +58,7 @@ func (c *Client) Establish() error {
 			conn := newConn(c, c.disorderlyQueue)
 
 			if err := conn.dial(); err != nil {
+				log.Warnf("conn dial failed: %v", err)
 				return err
 			}
 

internal/transporter/internal/client/conn.go

@@ -33,7 +33,7 @@ type Conn struct {
 func newConn(cli *Client, queue chan *chWrite) *Conn {
 	c := &Conn{}
 	c.cli = cli
-	c.state.Store(def.ConnClosed)
+	c.state.Store(def.ConnHanged)
 	c.pending = newPending()
 	c.orderlyQueue = make(chan *chWrite, 4096)
 	c.disorderlyQueue = queue
@@ -97,30 +97,26 @@ func (c *Conn) send(ch *chWrite, isOrderly ...bool) error {
 }
 
 // 处理连接
-func (c *Conn) process(conn net.Conn) (err error) {
+func (c *Conn) process(conn net.Conn) error {
 	c.ctx, c.cancel = context.WithCancel(context.Background())
 	c.state.Store(def.ConnOpened)
 	c.lastHeartbeatTime.Store(xtime.Now().Unix())
 
 	go c.read(conn)
 
-	defer func() {
-		if err != nil {
-			c.close()
-		}
-	}()
-
-	if err = c.handshake(conn); err != nil {
-		return
-	}
+	if err := c.handshake(conn); err != nil {
+		c.close()
 
-	go c.write(conn)
+		return err
+	} else {
+		go c.write(conn)
 
-	return
+		return nil
+	}
 }
 
 // 握手
-func (c *Conn) handshake(conn net.Conn) (err error) {
+func (c *Conn) handshake(conn net.Conn) error {
 	var (
 		seq  = uint64(1)
 		call = make(chan []byte)
@@ -131,15 +127,22 @@ func (c *Conn) handshake(conn net.Conn) (err error) {
 
 	c.pending.store(seq, call)
 
-	if _, err = conn.Write(buf.Bytes()); err != nil {
+	defer close(call)
+
+	if _, err := conn.Write(buf.Bytes()); err != nil {
 		c.pending.delete(seq)
-	} else {
-		<-call
+		return err
 	}
 
-	close(call)
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
+	defer cancel()
 
-	return
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-call:
+		return nil
+	}
 }
 
 // 读取数据
@@ -188,6 +191,7 @@ func (c *Conn) write(conn net.Conn) {
 			deadline := t.Add(-2 * def.HeartbeatInterval).Unix()
 
 			if c.lastHeartbeatTime.Load() < deadline {
+				log.Warn("connection heartbeat timeout")
 				c.retry(conn)
 				return
 			} else {
@@ -246,7 +250,7 @@ func (c *Conn) retry(conn net.Conn) {
 		return
 	}
 
-	conn.Close()
+	_ = conn.Close()
 
 	if c.cancel != nil {
 		c.cancel()
@@ -259,7 +263,9 @@ func (c *Conn) retry(conn net.Conn) {
 
 // 关闭连接
 func (c *Conn) close() {
-	c.state.Store(def.ConnClosed)
+	if c.state.Swap(def.ConnClosed) == def.ConnClosed {
+		return
+	}
 
 	c.cli.done()
 

internal/transporter/internal/server/conn.go

@@ -128,7 +128,9 @@ func (c *Conn) write() {
 			return
 		case <-ticker.C:
 			deadline := xtime.Now().Add(-2 * def.HeartbeatInterval).Unix()
+
 			if atomic.LoadInt64(&c.lastHeartbeatTime) < deadline {
+				log.Warn("connection heartbeat timeout")
 				_ = c.close(true)
 				return
 			}

internal/transporter/internal/server/def.go

@@ -15,6 +15,8 @@ import (
 
 const scheme = "drpc"
 
+type RouteHandler func(conn *Conn, data []byte) error
+
 type Server struct {
 	listener    net.Listener           // 监听器
 	listenAddr  string                 // 监听地址