transport_sender.go

package raft

import (
	"runtime"
	"sync"
	"time"

	"master-server/raft/logger"
	"master-server/raft/proto"
	"master-server/raft/util"
)

type unreachableReporter func(uint64)

type transportSender struct {
	nodeId      uint64
	concurrency uint64
	senderType  SocketType
	resolver    SocketResolver
	inputc      []chan *proto.Message
	send        func(msg *proto.Message)
	mu          sync.Mutex
	stopc       chan struct{}
}

func newTransportSender(nodeId, concurrency uint64, buffSize int, senderType SocketType, resolver SocketResolver) *transportSender {
	sender := &transportSender{
		nodeId:      nodeId,
		concurrency: concurrency,
		senderType:  senderType,
		resolver:    resolver,
		inputc:      make([]chan *proto.Message, concurrency),
		stopc:       make(chan struct{}),
	}
	for i := uint64(0); i < concurrency; i++ {
		sender.inputc[i] = make(chan *proto.Message, buffSize)
		sender.loopSend(sender.inputc[i])
	}

	if (concurrency & (concurrency - 1)) == 0 {
		sender.send = func(msg *proto.Message) {
			idx := 0
			if concurrency > 1 {
				idx = int(msg.ID&concurrency - 1)
			}
			sender.inputc[idx] <- msg
		}
	} else {
		sender.send = func(msg *proto.Message) {
			idx := 0
			if concurrency > 1 {
				idx = int(msg.ID % concurrency)
			}
			sender.inputc[idx] <- msg
		}
	}
	return sender
}

func (s *transportSender) stop() {
	s.mu.Lock()
	defer s.mu.Unlock()

	select {
	case <-s.stopc:
		return
	default:
		close(s.stopc)
	}
}

func (s *transportSender) loopSend(recvc chan *proto.Message) {
	util.RunWorkerUtilStop(func() {
		conn := getConn(s.nodeId, s.senderType, s.resolver, 0, 2*time.Second)
		bufWr := util.NewBufferWriter(conn, 16*KB)

		defer func() {
			if conn != nil {
				conn.Close()
			}
		}()

		loopCount := 0
		var err error
		for {
			loopCount = loopCount + 1
			if loopCount > 8 {
				loopCount = 0
				runtime.Gosched()
			}

			select {
			case <-s.stopc:
				return

			case msg := <-recvc:
				if conn == nil {
					conn = getConn(s.nodeId, s.senderType, s.resolver, 0, 2*time.Second)
					if conn == nil {
						time.Sleep(50 * time.Millisecond)
						continue
					}
					bufWr.Reset(conn)
				}
				err = msg.Encode(bufWr)
				proto.ReturnMessage(msg)
				if err != nil {
					goto flush
				}
				// group send message
				flag := false
				for i := 0; i < 16; i++ {
					select {
					case msg := <-recvc:
						err = msg.Encode(bufWr)
						proto.ReturnMessage(msg)
						if err != nil {
							goto flush
						}
					default:
						flag = true
					}
					if flag {
						break
					}
				}
			}

		flush:
			// flush write
			if err == nil {
				err = bufWr.Flush()
			}
			if err != nil {
				logger.Error("[Transport]send message[%s] to %v[%s] error:[%v].", s.senderType, s.nodeId, conn.RemoteAddr(), err)
				conn.Close()
				conn = nil
			}
		}
	}, s.stopc)
}

func getConn(nodeId uint64, socketType SocketType, resolver SocketResolver, rdTime, wrTime time.Duration) (conn *util.ConnTimeout) {
	var (
		addr string
		err  error
	)
	if addr, err = resolver.NodeAddress(nodeId, socketType); err == nil {
		if conn, err = util.DialTimeout(addr, 2*time.Second); err == nil {
			conn.SetReadTimeout(rdTime)
			conn.SetWriteTimeout(wrTime)
		}
	}

	if err != nil {
		conn = nil
		if logger.IsEnableDebug() {
			logger.Debug("[Transport] get connection[%s] to %v[%s] failed,error is: %s", socketType, nodeId, addr, err)
		}
	}
	return
}