Pool (#16)

* add pool

* elastic workerpool

* del buf

* modify usage

* modify readme

broker/broker.go

@@ -12,49 +12,59 @@ import (
 	"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"
-
-	"github.com/fhmq/hmq/logger"
 )
 
 var (
-	log = logger.Get().Named("Broker")
+	brokerLog *zap.Logger
 )
 
+type Message struct {
+	client *client
+	packet packets.ControlPacket
+}
+
 type Broker struct {
-	id        string
-	cid       uint64
-	mu        sync.Mutex
-	config    *Config
-	tlsConfig *tls.Config
-	AclConfig *acl.ACLConfig
-	clients   sync.Map
-	routes    sync.Map
-	remotes   sync.Map
-	nodes     map[string]interface{}
-	sl        *Sublist
-	rl        *RetainList
-	queues    map[string]int
+	id          string
+	cid         uint64
+	mu          sync.Mutex
+	config      *Config
+	tlsConfig   *tls.Config
+	AclConfig   *acl.ACLConfig
+	wpool       *pool.WorkerPool
+	clients     sync.Map
+	routes      sync.Map
+	remotes     sync.Map
+	nodes       map[string]interface{}
+	clusterPool chan *Message
+	messagePool chan *Message
+	sl          *Sublist
+	rl          *RetainList
+	queues      map[string]int
 }
 
 func NewBroker(config *Config) (*Broker, error) {
 	b := &Broker{
-		id:     GenUniqueId(),
-		config: config,
-		sl:     NewSublist(),
-		rl:     NewRetainList(),
-		nodes:  make(map[string]interface{}),
-		queues: make(map[string]int),
+		id:          GenUniqueId(),
+		config:      config,
+		wpool:       pool.New(config.Worker),
+		sl:          NewSublist(),
+		rl:          NewRetainList(),
+		nodes:       make(map[string]interface{}),
+		queues:      make(map[string]int),
+		clusterPool: make(chan *Message),
+		messagePool: make(chan *Message),
 	}
 	if b.config.TlsPort != "" {
 		tlsconfig, err := NewTLSConfig(b.config.TlsInfo)
 		if err != nil {
-			log.Error("new tlsConfig error", zap.Error(err))
+			brokerLog.Error("new tlsConfig error", zap.Error(err))
 			return nil, err
 		}
 		b.tlsConfig = tlsconfig
@@ -62,7 +72,7 @@ func NewBroker(config *Config) (*Broker, error) {
 	if b.config.Acl {
 		aclconfig, err := acl.AclConfigLoad(b.config.AclConf)
 		if err != nil {
-			log.Error("Load acl conf error", zap.Error(err))
+			brokerLog.Error("Load acl conf error", zap.Error(err))
 			return nil, err
 		}
 		b.AclConfig = aclconfig
@@ -71,12 +81,26 @@ func NewBroker(config *Config) (*Broker, error) {
 	return b, nil
 }
 
+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)
+		})
+	}
+
+}
+
 func (b *Broker) Start() {
 	if b == nil {
-		log.Error("broker is null")
+		brokerLog.Error("broker is null")
 		return
 	}
-	StartDispatcher()
+	go b.StartDispatcher()
 
 	//listen clinet over tcp
 	if b.config.Port != "" {
@@ -100,6 +124,7 @@ func (b *Broker) Start() {
 
 	//connect on other node in cluster
 	if b.config.Router != "" {
+		go b.processClusterInfo()
 		b.ConnectToDiscovery()
 	}
 
@@ -124,7 +149,7 @@ func StateMonitor() {
 func (b *Broker) StartWebsocketListening() {
 	path := b.config.WsPath
 	hp := ":" + b.config.WsPort
-	log.Info("Start Websocket Listening on ", zap.String("hp", hp), zap.String("path", path))
+	brokerLog.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 {
@@ -133,7 +158,7 @@ func (b *Broker) StartWebsocketListening() {
 		err = http.ListenAndServe(hp, nil)
 	}
 	if err != nil {
-		log.Error("ListenAndServe: " + err.Error())
+		brokerLog.Error("ListenAndServe:" + err.Error())
 		return
 	}
 }
@@ -152,14 +177,14 @@ func (b *Broker) StartClientListening(Tls bool) {
 	if Tls {
 		hp = b.config.TlsHost + ":" + b.config.TlsPort
 		l, err = tls.Listen("tcp", hp, b.tlsConfig)
-		log.Info("Start TLS Listening client on ", zap.String("hp", hp))
+		brokerLog.Info("Start TLS Listening client on ", zap.String("hp", hp))
 	} else {
 		hp := b.config.Host + ":" + b.config.Port
 		l, err = net.Listen("tcp", hp)
-		log.Info("Start Listening client on ", zap.String("hp", hp))
+		brokerLog.Info("Start Listening client on ", zap.String("hp", hp))
 	}
 	if err != nil {
-		log.Error("Error listening on ", zap.Error(err))
+		brokerLog.Error("Error listening on ", zap.Error(err))
 		return
 	}
 	tmpDelay := 10 * ACCEPT_MIN_SLEEP
@@ -167,7 +192,7 @@ func (b *Broker) StartClientListening(Tls bool) {
 		conn, err := l.Accept()
 		if err != nil {
 			if ne, ok := err.(net.Error); ok && ne.Temporary() {
-				log.Error("Temporary Client Accept Error(%v), sleeping %dms",
+				brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
 					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
@@ -175,7 +200,7 @@ func (b *Broker) StartClientListening(Tls bool) {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				log.Error("Accept error: %v", zap.Error(err))
+				brokerLog.Error("Accept error: %v", zap.Error(err))
 			}
 			continue
 		}
@@ -194,7 +219,7 @@ func (b *Broker) Handshake(conn net.Conn) bool {
 
 	// Force handshake
 	if err := nc.Handshake(); err != nil {
-		log.Error("TLS handshake error, ", zap.Error(err))
+		brokerLog.Error("TLS handshake error, ", zap.Error(err))
 		return false
 	}
 	nc.SetReadDeadline(time.Time{})
@@ -210,18 +235,18 @@ func TlsTimeout(conn *tls.Conn) {
 	}
 	cs := nc.ConnectionState()
 	if !cs.HandshakeComplete {
-		log.Error("TLS handshake timeout")
+		brokerLog.Error("TLS handshake timeout")
 		nc.Close()
 	}
 }
 
 func (b *Broker) StartClusterListening() {
 	var hp string = b.config.Cluster.Host + ":" + b.config.Cluster.Port
-	log.Info("Start Listening cluster on ", zap.String("hp", hp))
+	brokerLog.Info("Start Listening cluster on ", zap.String("hp", hp))
 
 	l, e := net.Listen("tcp", hp)
 	if e != nil {
-		log.Error("Error listening on ", zap.Error(e))
+		brokerLog.Error("Error listening on ", zap.Error(e))
 		return
 	}
 
@@ -231,7 +256,7 @@ func (b *Broker) StartClusterListening() {
 		conn, err := l.Accept()
 		if err != nil {
 			if ne, ok := err.(net.Error); ok && ne.Temporary() {
-				log.Error("Temporary Client Accept Error(%v), sleeping %dms",
+				brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
 					zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
 				time.Sleep(tmpDelay)
 				tmpDelay *= 2
@@ -239,7 +264,7 @@ func (b *Broker) StartClusterListening() {
 					tmpDelay = ACCEPT_MAX_SLEEP
 				}
 			} else {
-				log.Error("Accept error: %v", zap.Error(err))
+				brokerLog.Error("Accept error: %v", zap.Error(err))
 			}
 			continue
 		}
@@ -253,16 +278,16 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	//process connect packet
 	packet, err := packets.ReadPacket(conn)
 	if err != nil {
-		log.Error("read connect packet error: ", zap.Error(err))
+		brokerLog.Error("read connect packet error: ", zap.Error(err))
 		return
 	}
 	if packet == nil {
-		log.Error("received nil packet")
+		brokerLog.Error("received nil packet")
 		return
 	}
 	msg, ok := packet.(*packets.ConnectPacket)
 	if !ok {
-		log.Error("received msg that was not Connect")
+		brokerLog.Error("received msg that was not Connect")
 		return
 	}
 	connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
@@ -270,7 +295,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 	connack.SessionPresent = msg.CleanSession
 	err = connack.Write(conn)
 	if err != nil {
-		log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
+		brokerLog.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
 		return
 	}
 
@@ -303,40 +328,33 @@ func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
 
 	cid := c.info.clientID
 
-	var msgPool *MessagePool
 	var exist bool
 	var old interface{}
 
 	switch typ {
 	case CLIENT:
-		msgPool = MSGPool[idx%MessagePoolNum].GetPool()
-		c.mp = msgPool
 		old, exist = b.clients.Load(cid)
 		if exist {
-			log.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
+			brokerLog.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
 			ol, ok := old.(*client)
 			if ok {
-				msg := &Message{client: c, packet: DisconnectdPacket}
-				ol.mp.queue <- msg
+				ol.Close()
 			}
 		}
 		b.clients.Store(cid, c)
 	case ROUTER:
-		msgPool = MSGPool[(MessagePoolNum + idx)].GetPool()
-		c.mp = msgPool
 		old, exist = b.routes.Load(cid)
 		if exist {
-			log.Warn("router exist, close old...")
+			brokerLog.Warn("router exist, close old...")
 			ol, ok := old.(*client)
 			if ok {
-				msg := &Message{client: c, packet: DisconnectdPacket}
-				ol.mp.queue <- msg
+				ol.Close()
 			}
 		}
 		b.routes.Store(cid, c)
 	}
 
-	c.readLoop()
+	c.readLoop(b.messagePool)
 }
 
 func (b *Broker) ConnectToDiscovery() {
@@ -346,8 +364,8 @@ func (b *Broker) ConnectToDiscovery() {
 	for {
 		conn, err = net.Dial("tcp", b.config.Router)
 		if err != nil {
-			log.Error("Error trying to connect to route: ", zap.Error(err))
-			log.Debug("Connect to route timeout ,retry...")
+			brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
+			brokerLog.Debug("Connect to route timeout ,retry...")
 
 			if 0 == tempDelay {
 				tempDelay = 1 * time.Second
@@ -363,7 +381,7 @@ func (b *Broker) ConnectToDiscovery() {
 		}
 		break
 	}
-	log.Debug("connect to router success :", zap.String("Router", b.config.Router))
+	brokerLog.Debug("connect to router success :", zap.String("Router", b.config.Router))
 
 	cid := b.id
 	info := info{
@@ -383,11 +401,22 @@ func (b *Broker) ConnectToDiscovery() {
 	c.SendConnect()
 	c.SendInfo()
 
-	c.mp = &MSGPool[(MessagePoolNum + 2)]
-	go c.readLoop()
+	go c.readLoop(b.clusterPool)
 	go c.StartPing()
 }
 
+func (b *Broker) processClusterInfo() {
+	for {
+		msg, ok := <-b.clusterPool
+		if !ok {
+			brokerLog.Error("read message from cluster channel error")
+			return
+		}
+		ProcessMessage(msg)
+	}
+
+}
+
 func (b *Broker) connectRouter(id, addr string) {
 	var conn net.Conn
 	var err error
@@ -402,13 +431,13 @@ func (b *Broker) connectRouter(id, addr string) {
 
 		conn, err = net.Dial("tcp", addr)
 		if err != nil {
-			log.Error("Error trying to connect to route: ", zap.Error(err))
+			brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
 
 			if retryTimes > 50 {
 				return
 			}
 
-			log.Debug("Connect to route timeout ,retry...")
+			brokerLog.Debug("Connect to route timeout ,retry...")
 
 			if 0 == timeDelay {
 				timeDelay = 1 * time.Second
@@ -446,12 +475,9 @@ func (b *Broker) connectRouter(id, addr string) {
 	c.init()
 	b.remotes.Store(cid, c)
 
-	c.mp = MSGPool[(MessagePoolNum + 1)].GetPool()
-
 	c.SendConnect()
-	// c.SendInfo()
 
-	go c.readLoop()
+	go c.readLoop(b.messagePool)
 	go c.StartPing()
 
 }
@@ -510,7 +536,7 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
 	if len(subInfo.Topics) > 0 {
 		err := c.WriterPacket(subInfo)
 		if err != nil {
-			log.Error("Send localsubs To Router error :", zap.Error(err))
+			brokerLog.Error("Send localsubs To Router error :", zap.Error(err))
 		}
 	}
 }
@@ -527,7 +553,7 @@ func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacke
 		return true
 
 	})
-	// log.Info("BroadcastInfoMessage success ")
+	// brokerLog.Info("BroadcastInfoMessage success ")
 }
 
 func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
@@ -539,7 +565,7 @@ func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
 		}
 		return true
 	})
-	// log.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
+	// brokerLog.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
 }
 
 func (b *Broker) removeClient(c *client) {
@@ -553,7 +579,7 @@ func (b *Broker) removeClient(c *client) {
 	case REMOTE:
 		b.remotes.Delete(clientId)
 	}
-	// log.Info("delete client ,", clientId)
+	// brokerLog.Info("delete client ,", clientId)
 }
 
 func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
@@ -567,7 +593,7 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
 		if sub != nil {
 			err := sub.client.WriterPacket(packet)
 			if err != nil {
-				log.Error("process message for psub error,  ", zap.Error(err))
+				brokerLog.Error("process message for psub error,  ", zap.Error(err))
 			}
 		}
 	}