modify_message_pool

broker/broker.go

@@ -4,25 +4,24 @@ package broker
 
 import (
 	"crypto/tls"
+	"github.com/eclipse/paho.mqtt.golang/packets"
+	"github.com/fhmq/hmq/lib/acl"
+	"github.com/fhmq/hmq/pool"
+	"github.com/segmentio/fasthash/fnv1a"
+	"github.com/shirou/gopsutil/mem"
+	"go.uber.org/zap"
+	"golang.org/x/net/websocket"
 	"net"
 	"net/http"
 	"runtime/debug"
 	"sync"
 	"sync/atomic"
 	"time"
-
-	"github.com/fhmq/hmq/lib/acl"
-	"github.com/fhmq/hmq/pool"
-
-	"github.com/eclipse/paho.mqtt.golang/packets"
-	"github.com/shirou/gopsutil/mem"
-	"go.uber.org/zap"
-
-	"golang.org/x/net/websocket"
 )
 
-var (
-	brokerLog *zap.Logger
+const (
+	MessagePoolNum        = 1024
+	MessagePoolMessageNum = 1024
 )
 
 type Message struct {
@@ -43,12 +42,21 @@ type Broker struct {
 	remotes     sync.Map
 	nodes       map[string]interface{}
 	clusterPool chan *Message
-	messagePool chan *Message
+	messagePool []chan *Message
 	sl          *Sublist
 	rl          *RetainList
 	queues      map[string]int
 }
 
+func newMessagePool() []chan *Message {
+	pool := make([]chan *Message, 0)
+	for i := 0; i < MessagePoolNum; i++ {
+		ch := make(chan *Message, MessagePoolMessageNum)
+		pool = append(pool, ch)
+	}
+	return pool
+}
+
 func NewBroker(config *Config) (*Broker, error) {
 	b := &Broker{
 		id:          GenUniqueId(),
@@ -59,12 +67,12 @@ func NewBroker(config *Config) (*Broker, error) {
 		nodes:       make(map[string]interface{}),
 		queues:      make(map[string]int),
 		clusterPool: make(chan *Message),
-		messagePool: make(chan *Message),
+		messagePool: newMessagePool(),
 	}
 	if b.config.TlsPort != "" {
 		tlsconfig, err := NewTLSConfig(b.config.TlsInfo)
 		if err != nil {
-			brokerLog.Error("new tlsConfig error", zap.Error(err))
+			log.Error("new tlsConfig error", zap.Error(err))
 			return nil, err
 		}
 		b.tlsConfig = tlsconfig
@@ -72,7 +80,7 @@ func NewBroker(config *Config) (*Broker, error) {
 	if b.config.Acl {
 		aclconfig, err := acl.AclConfigLoad(b.config.AclConf)
 		if err != nil {
-			brokerLog.Error("Load acl conf error", zap.Error(err))
+			log.Error("Load acl conf error", zap.Error(err))
 			return nil, err
 		}
 		b.AclConfig = aclconfig
@@ -82,22 +90,26 @@ func NewBroker(config *Config) (*Broker, error) {
 }
 
 func (b *Broker) StartDispatcher() {
-	for {
-		msg, ok := <-b.messagePool
-		if !ok {
-			brokerLog.Error("read message from client channel error")
-			return
-		}
-		b.wpool.Submit(func() {
-			ProcessMessage(msg)
-		})
+	for _, mpool := range b.messagePool {
+		go func(ch chan *Message) {
+			for {
+				msg, ok := <-ch
+				if !ok {
+					log.Error("read message from client channel error")
+					return
+				}
+				b.wpool.Submit(func() {
+					ProcessMessage(msg)
+				})
+			}
+		}(mpool)
 	}
 
 }
 
 func (b *Broker) Start() {
 	if b == nil {
-		brokerLog.Error("broker is null")
+		log.Error("broker is null")
 		return
 	}
 	go b.StartDispatcher()
@@ -149,7 +161,7 @@ func StateMonitor() {
 func (b *Broker) StartWebsocketListening() {
 	path := b.config.WsPath
 	hp := ":" + b.config.WsPort
-	brokerLog.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path))
+	log.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path))
 	http.Handle(path, websocket.Handler(b.wsHandler))
 	var err error
 	if b.config.WsTLS {
@@ -158,7 +170,7 @@ func (b *Broker) StartWebsocketListening() {
 		err = http.ListenAndServe(hp, nil)
 	}
 	if err != nil {
-		brokerLog.Error("ListenAndServe:" + err.Error())
+		log.Error("ListenAndServe:" + err.Error())
 		return
 	}
 }
@@ -167,7 +179,7 @@ func (b *Broker) wsHandler(ws *websocket.Conn) {
 	// io.Copy(ws, ws)
 	atomic.AddUint64(&b.cid, 1)
 	ws.PayloadType = websocket.BinaryFrame
-	b.handleConnection(CLIENT, ws, b.cid)
+	b.handleConnection(CLIENT, ws)
 }
 
 func (b *Broker) StartClientListening(Tls bool) {
@@ -177,14 +189,14 @@ func (b *Broker) StartClientListening(Tls bool) {
 	if Tls {
 		hp = b.config.TlsHost + ":" + b.config.TlsPort
 		l, err = tls.Listen("tcp", hp, b.tlsConfig)
-		brokerLog.Info("Start TLS Listening client on ", zap.String("hp", hp))
+		log.Info("Start TLS Listening client on ", zap.String("hp", hp))
 	} else {
 		hp := b.config.Host + ":" + b.config.Port
 		l, err = net.Listen("tcp", hp)
-		brokerLog.Info("Start Listening client on ", zap.String("hp", hp))
+		log.Info("Start Listening client on ", zap.String("hp", hp))
 	}
 	if err != nil {
-		brokerLog.Error("Error listening on ", zap.Error(err))
+		log.Error("Error listening on ", zap.Error(err))
 		return
 	}
 	tmpDelay := 10 * ACCEPT_MIN_SLEEP
@@ -192,7 +204,7 @@ func (b *Broker) StartClientListening(Tls bool) {
 		conn, err := l.Accept()
 		if err != nil {
 			if ne, ok := err.(net.Error); ok && ne.Temporary() {
-				brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
+				log.Error("Temporary Client Accept Error(%v), sleeping %dms",
 					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
@@ -200,13 +212,13 @@ func (b *Broker) StartClientListening(Tls bool) {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				brokerLog.Error("Accept error: %v", zap.Error(err))
+				log.Error("Accept error: %v", zap.Error(err))
 			}
 			continue
 		}
 		tmpDelay = ACCEPT_MIN_SLEEP
 		atomic.AddUint64(&b.cid, 1)
-		go b.handleConnection(CLIENT, conn, b.cid)
+		go b.handleConnection(CLIENT, conn)
 
 	}
 }
@@ -219,7 +231,7 @@ func (b *Broker) Handshake(conn net.Conn) bool {
 
 	// Force handshake
 	if err := nc.Handshake(); err != nil {
-		brokerLog.Error("TLS handshake error, ", zap.Error(err))
+		log.Error("TLS handshake error, ", zap.Error(err))
 		return false
 	}
 	nc.SetReadDeadline(time.Time{})
@@ -235,28 +247,27 @@ func TlsTimeout(conn *tls.Conn) {
 	}
 	cs := nc.ConnectionState()
 	if !cs.HandshakeComplete {
-		brokerLog.Error("TLS handshake timeout")
+		log.Error("TLS handshake timeout")
 		nc.Close()
 	}
 }
 
 func (b *Broker) StartClusterListening() {
 	var hp string = b.config.Cluster.Host + ":" + b.config.Cluster.Port
-	brokerLog.Info("Start Listening cluster on ", zap.String("hp", hp))
+	log.Info("Start Listening cluster on ", zap.String("hp", hp))
 
 	l, e := net.Listen("tcp", hp)
 	if e != nil {
-		brokerLog.Error("Error listening on ", zap.Error(e))
+		log.Error("Error listening on ", zap.Error(e))
 		return
 	}
 
-	var idx uint64 = 0
 	tmpDelay := 10 * ACCEPT_MIN_SLEEP
 	for {
 		conn, err := l.Accept()
 		if err != nil {
 			if ne, ok := err.(net.Error); ok && ne.Temporary() {
-				brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
+				log.Error("Temporary Client Accept Error(%v), sleeping %dms",
 					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
@@ -264,30 +275,30 @@ func (b *Broker) StartClusterListening() {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				brokerLog.Error("Accept error: %v", zap.Error(err))
+				log.Error("Accept error: %v", zap.Error(err))
 			}
 			continue
 		}
 		tmpDelay = ACCEPT_MIN_SLEEP
 
-		go b.handleConnection(ROUTER, conn, idx)
+		go b.handleConnection(ROUTER, conn)
 	}
 }
 
-func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
+func (b *Broker) handleConnection(typ int, conn net.Conn) {
 	//process connect packet
 	packet, err := packets.ReadPacket(conn)
 	if err != nil {
-		brokerLog.Error("read connect packet error: ", zap.Error(err))
+		log.Error("read connect packet error: ", zap.Error(err))
 		return
 	}
 	if packet == nil {
-		brokerLog.Error("received nil packet")
+		log.Error("received nil packet")
 		return
 	}
 	msg, ok := packet.(*packets.ConnectPacket)
 	if !ok {
-		brokerLog.Error("received msg that was not Connect")
+		log.Error("received msg that was not Connect")
 		return
 	}
 	connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
@@ -295,7 +306,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	connack.SessionPresent = msg.CleanSession
 	err = connack.Write(conn)
 	if err != nil {
-		brokerLog.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
+		log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
 		return
 	}
 
@@ -335,7 +346,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	case CLIENT:
 		old, exist = b.clients.Load(cid)
 		if exist {
-			brokerLog.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
+			log.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
 			ol, ok := old.(*client)
 			if ok {
 				ol.Close()
@@ -345,7 +356,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	case ROUTER:
 		old, exist = b.routes.Load(cid)
 		if exist {
-			brokerLog.Warn("router exist, close old...")
+			log.Warn("router exist, close old...")
 			ol, ok := old.(*client)
 			if ok {
 				ol.Close()
@@ -354,7 +365,9 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 		b.routes.Store(cid, c)
 	}
 
-	c.readLoop(b.messagePool)
+	mpool := b.messagePool[fnv1a.HashString64(cid)%MessagePoolNum]
+
+	c.readLoop(mpool)
 }
 
 func (b *Broker) ConnectToDiscovery() {
@@ -364,8 +377,8 @@ func (b *Broker) ConnectToDiscovery() {
 	for {
 		conn, err = net.Dial("tcp", b.config.Router)
 		if err != nil {
-			brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
-			brokerLog.Debug("Connect to route timeout ,retry...")
+			log.Error("Error trying to connect to route: ", zap.Error(err))
+			log.Debug("Connect to route timeout ,retry...")
 
 			if 0 == tempDelay {
 				tempDelay = 1 * time.Second
@@ -381,7 +394,7 @@ func (b *Broker) ConnectToDiscovery() {
 		}
 		break
 	}
-	brokerLog.Debug("connect to router success :", zap.String("Router", b.config.Router))
+	log.Debug("connect to router success :", zap.String("Router", b.config.Router))
 
 	cid := b.id
 	info := info{
@@ -409,7 +422,7 @@ func (b *Broker) processClusterInfo() {
 	for {
 		msg, ok := <-b.clusterPool
 		if !ok {
-			brokerLog.Error("read message from cluster channel error")
+			log.Error("read message from cluster channel error")
 			return
 		}
 		ProcessMessage(msg)
@@ -431,13 +444,13 @@ func (b *Broker) connectRouter(id, addr string) {
 
 		conn, err = net.Dial("tcp", addr)
 		if err != nil {
-			brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
+			log.Error("Error trying to connect to route: ", zap.Error(err))
 
 			if retryTimes > 50 {
 				return
 			}
 
-			brokerLog.Debug("Connect to route timeout ,retry...")
+			log.Debug("Connect to route timeout ,retry...")
 
 			if 0 == timeDelay {
 				timeDelay = 1 * time.Second
@@ -477,7 +490,8 @@ func (b *Broker) connectRouter(id, addr string) {
 
 	c.SendConnect()
 
-	go c.readLoop(b.messagePool)
+	mpool := b.messagePool[fnv1a.HashString64(cid)%MessagePoolNum]
+	go c.readLoop(mpool)
 	go c.StartPing()
 
 }
@@ -536,7 +550,7 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
 	if len(subInfo.Topics) > 0 {
 		err := c.WriterPacket(subInfo)
 		if err != nil {
-			brokerLog.Error("Send localsubs To Router error :", zap.Error(err))
+			log.Error("Send localsubs To Router error :", zap.Error(err))
 		}
 	}
 }
@@ -553,7 +567,7 @@ func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacke
 		return true
 
 	})
-	// brokerLog.Info("BroadcastInfoMessage success ")
+	// log.Info("BroadcastInfoMessage success ")
 }
 
 func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
@@ -565,7 +579,7 @@ func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
 		}
 		return true
 	})
-	// brokerLog.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
+	// log.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
 }
 
 func (b *Broker) removeClient(c *client) {
@@ -579,7 +593,7 @@ func (b *Broker) removeClient(c *client) {
 	case REMOTE:
 		b.remotes.Delete(clientId)
 	}
-	// brokerLog.Info("delete client ,", clientId)
+	// log.Info("delete client ,", clientId)
 }
 
 func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
@@ -593,7 +607,7 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
 		if sub != nil {
 			err := sub.client.WriterPacket(packet)
 			if err != nil {
-				brokerLog.Error("process message for psub error,  ", zap.Error(err))
+				log.Error("process message for psub error,  ", zap.Error(err))
 			}
 		}
 	}