use cache reduce malloc

.gitignore

@@ -5,3 +5,4 @@ one_key.sh
 sniffer-agent
 # vendor/github.com
 profile*
+*.svg

capture/network.go

@@ -117,7 +117,7 @@ func (nc *networkCard) listenNormal() {
 			}
 
 			// throw packets according to a certain probability
-			throwPacketRate := communicator.GetThrowPacketRate()
+			throwPacketRate := communicator.GetTCPThrowPacketRate()
 			if throwPacketRate >= 1.0 {
 				time.Sleep(time.Second*3)
 				continue
@@ -255,7 +255,11 @@ func readFromServerPackage(
 
 	if tcpPkt.FIN {
 		sessionKey := spliceSessionKey(srcIP, srcPort)
+		session := sessionPool[*sessionKey]
+		if session != nil {
+			session.Close()
 			delete(sessionPool, *sessionKey)
+		}
 		return
 	}
 
@@ -285,7 +289,11 @@ func readToServerPackage(
 	// when client try close connection remove session from session pool
 	if tcpPkt.FIN {
 		sessionKey := spliceSessionKey(srcIP, srcPort)
+		session := sessionPool[*sessionKey]
+		if session != nil {
+			session.Close()
 			delete(sessionPool, *sessionKey)
+		}
 		log.Debugf("close connection from %s", *sessionKey)
 		return
 	}

communicator/controller.go

@@ -82,6 +82,10 @@ func outletSetConfig(resp http.ResponseWriter, req *http.Request) {
 	mp.Err = SetConfig(ep.ConfigName, ep.Value)
 }
 
-func GetThrowPacketRate() float64 {
+func GetTCPThrowPacketRate() float64 {
-	return throwPacketRate.value
+	return throwPacketRate.tcpTPR
+}
+
+func GetMysqlThrowPacketRate() float64 {
+	return throwPacketRate.mysqlTPR
 }

communicator/model.go

@@ -1,6 +1,9 @@
 package communicator
 
-import "fmt"
+import (
+	"fmt"
+	"math"
+)
 
 type configItem interface {
 	setVal (interface{}) error
@@ -9,13 +12,15 @@ type configItem interface {
 
 type throwPacketRateConfig struct {
 	name string
-	value float64
+	tcpTPR float64
+	mysqlTPR float64
 }
 
 func newThrowPacketRateConfig() (tpr *throwPacketRateConfig) {
 	tpr = &throwPacketRateConfig{
 		name: THROW_PACKET_RATE,
-		value: 0.0,
+		tcpTPR: 0.0,
+		mysqlTPR: 0.0,
 	}
 	return
 }
@@ -27,14 +32,11 @@ func (tc *throwPacketRateConfig) setVal (val interface{}) (err error){
 		return
 	}
 
-	tc.value = realVal
+	tc.mysqlTPR = realVal
+	tc.tcpTPR = math.Sqrt(realVal)
 	return
 }
 
 func (tc *throwPacketRateConfig) getVal () (val interface{}){
-	return tc.value
+	return tc.mysqlTPR
-}
-
-func (tc *throwPacketRateConfig) GetValFloat64 () (val float64){
-	return tc.value
 }

model/cache_pool.go

@@ -0,0 +1,66 @@
+package model
+
+import (
+	"fmt"
+	"reflect"
+	"unsafe"
+
+	log "github.com/sirupsen/logrus"
+)
+
+
+type sliceBufferPool struct {
+	queue chan []byte
+	bufferSize int
+	name string
+}
+
+func NewSliceBufferPool(name string, bufferSize int) (sbp *sliceBufferPool) {
+	return &sliceBufferPool{
+		queue: make(chan []byte, 512),
+		bufferSize: bufferSize,
+		name: name,
+	}
+}
+
+func (sbp *sliceBufferPool) Enqueue(buffer []byte)  {
+	defer func() {
+		log.Debugf("after enqueue from %s, there is %d elements", sbp.name, len(sbp.queue))
+	}()
+
+	if cap(buffer) < 1 {
+		return
+	}
+
+	sbp.queue <- buffer
+}
+
+func (sbp *sliceBufferPool) DequeueWithInit(initSize int) (buffer []byte)  {
+	if initSize >= sbp.bufferSize {
+		panic(fmt.Sprintf("package size bigger than max buffer size need deal:%d",
+			sbp.bufferSize))
+	}
+
+	defer func() {
+		// reset cache byte
+		pbytes := (*reflect.SliceHeader)(unsafe.Pointer(&buffer))
+		pbytes.Len = initSize
+	}()
+
+	buffer = sbp.Dequeue()
+	return
+}
+
+func (sbp *sliceBufferPool) Dequeue() (buffer []byte)  {
+	defer func() {
+		log.Debugf("after dequeue from %s, there is %d elements", sbp.name, len(sbp.queue))
+	}()
+
+	select {
+	case buffer = <- sbp.queue:
+		return
+	default:
+		buffer = make([]byte, 0, sbp.bufferSize)
+		return
+	}
+}

model/query_piece.go

@@ -1,14 +1,16 @@
 package model
 
 import (
-	// "encoding/json"
+	"bytes"
-	"github.com/json-iterator/go"
+
+	"encoding/json"
+	// "github.com/json-iterator/go"
 	"time"
 
 	"github.com/pingcap/tidb/util/hack"
 )
 
-var json = jsoniter.ConfigCompatibleWithStandardLibrary
+// var json = jsoniter.ConfigCompatibleWithStandardLibrary
 
 type QueryPiece interface {
 	String() *string
@@ -32,20 +34,23 @@ type MysqlQueryPiece struct {
 	ThrowPacketRate    float64  `json:"tpr"`
 	BeginTime    int64  `json:"bt"`
 	CostTimeInMS int64   `json:"cms"`
+
+	jsonContent     []byte    `json:"-"`
 }
 
 type PooledMysqlQueryPiece struct {
 	MysqlQueryPiece
 	recoverPool *mysqlQueryPiecePool
+	sliceBufferPool *sliceBufferPool
 }
 
 const (
-	datetimeFormat = "2006-01-02 15:04:05"
 	millSecondUnit = int64(time.Millisecond)
 )
 
 var (
 	mqpp = NewMysqlQueryPiecePool()
+	localSliceBufferPool = NewSliceBufferPool("json cache", 2*1024*1024)
 )
 
 func NewPooledMysqlQueryPiece(
@@ -53,11 +58,6 @@ func NewPooledMysqlQueryPiece(
 	clientPort, serverPort int, throwPacketRate float64, stmtBeginTime int64) (
 	mqp *PooledMysqlQueryPiece) {
 	mqp = mqpp.Dequeue()
-	if mqp == nil {
-		mqp = &PooledMysqlQueryPiece{
-			MysqlQueryPiece: MysqlQueryPiece{},
-		}
-	}
 
 	nowInMS := time.Now().UnixNano() / millSecondUnit
 	mqp.SessionID = sessionID
@@ -73,6 +73,7 @@ func NewPooledMysqlQueryPiece(
 	mqp.BeginTime = stmtBeginTime
 	mqp.CostTimeInMS = nowInMS - stmtBeginTime
 	mqp.recoverPool = mqpp
+	mqp.sliceBufferPool = localSliceBufferPool
 
 	return
 }
@@ -83,13 +84,23 @@ func (mqp *MysqlQueryPiece) String() (*string) {
 	return &contentStr
 }
 
-func (mqp *MysqlQueryPiece) Bytes() (bytes []byte) {
+func (mqp *MysqlQueryPiece) Bytes() (content []byte) {
-	content, err := json.Marshal(mqp)
+	// content, err := json.Marshal(mqp)
-	if err != nil {
+	if len(mqp.jsonContent) > 0 {
-		return []byte(err.Error())
+		return mqp.jsonContent
 	}
 
-	return content
+	var cacheBuffer = localSliceBufferPool.Dequeue()
+	buffer := bytes.NewBuffer(cacheBuffer)
+	err := json.NewEncoder(buffer).Encode(mqp)
+	if err != nil {
+		mqp.jsonContent = []byte(err.Error())
+
+	} else {
+		mqp.jsonContent = buffer.Bytes()
+	}
+
+	return mqp.jsonContent
 }
 
 func (mqp *MysqlQueryPiece) GetSQL() (str *string) {
@@ -106,4 +117,6 @@ func (mqp *MysqlQueryPiece) SetNeedSyncSend(syncSend bool) {
 
 func (pmqp *PooledMysqlQueryPiece) Recovery() {
 	pmqp.recoverPool.Enqueue(pmqp)
+	pmqp.sliceBufferPool.Enqueue(pmqp.jsonContent[:0])
+	pmqp.jsonContent = nil
 }

model/query_piece_pool.go

@@ -5,13 +5,13 @@ import (
 )
 
 type mysqlQueryPiecePool struct {
-	queue  []*PooledMysqlQueryPiece
+	queue  chan *PooledMysqlQueryPiece
 	lock sync.Mutex
 }
 
 func NewMysqlQueryPiecePool() (mqpp *mysqlQueryPiecePool) {
 	return &mysqlQueryPiecePool{
-		queue: make([]*PooledMysqlQueryPiece, 0, 5000),
+		queue: make(chan *PooledMysqlQueryPiece, 1024),
 	}
 }
 
@@ -19,18 +19,21 @@ func (mqpp *mysqlQueryPiecePool) Enqueue(pmqp *PooledMysqlQueryPiece)  {
 	mqpp.lock.Lock()
 	defer mqpp.lock.Unlock()
 
-	mqpp.queue = append(mqpp.queue, pmqp)
+
+	mqpp.queue <- pmqp
 }
 
 func (mqpp *mysqlQueryPiecePool) Dequeue() (pmqp *PooledMysqlQueryPiece)  {
 	mqpp.lock.Lock()
 	defer mqpp.lock.Unlock()
 
-	if len(mqpp.queue) < 1 {
+	select {
-		return nil
+	case pmqp = <- mqpp.queue:
-	}
-
-	pmqp = mqpp.queue[0]
-	mqpp.queue = mqpp.queue[1:]
 		return
+	default:
+		pmqp = &PooledMysqlQueryPiece{
+			MysqlQueryPiece: MysqlQueryPiece{},
+		}
+		return
+	}
 }

session-dealer/model.go

@@ -4,4 +4,5 @@ import "github.com/zr-hebo/sniffer-agent/model"
 
 type ConnSession interface {
 	ReceiveTCPPacket(*model.TCPPacket)
+	Close()
 }

session-dealer/mysql/config.go

@@ -3,6 +3,7 @@ package mysql
 import (
 	"flag"
 	"fmt"
+	"github.com/zr-hebo/sniffer-agent/model"
 	"regexp"
 )
 
@@ -17,6 +18,7 @@ var (
 	adminPasswd string
 
 	coverRangePool = NewCoveragePool()
+	localStmtCache = model.NewSliceBufferPool("statement cache", MaxMysqlPacketLen)
 )
 
 func init() {

session-dealer/mysql/cover_range.go

@@ -0,0 +1,133 @@
+package mysql
+
+import (
+	log "github.com/sirupsen/logrus"
+)
+
+// coverageNode record tcp package begin and end seq id
+type coverageNode struct {
+	begin int64
+	end   int64
+
+	next *coverageNode
+	crp *coveragePool
+}
+
+func newCoverage(begin, end int64) (*coverageNode) {
+	return &coverageNode{
+		begin: begin,
+		end: end,
+	}
+}
+
+func (crn *coverageNode) Recovery()  {
+	crn.crp.Enqueue(crn)
+}
+
+type coverRanges struct {
+	head *coverageNode
+}
+
+func NewCoverRanges() *coverRanges {
+	return &coverRanges{
+		head: &coverageNode{
+			begin: -1,
+		},
+	}
+}
+
+func (crs *coverRanges) clear() {
+	currRange := crs.head.next;
+	for currRange != nil {
+		node := currRange
+		currRange = currRange.next
+		node.Recovery()
+	}
+	crs.head.next = nil
+}
+
+func (crs *coverRanges) addRange(node *coverageNode)  {
+	// insert range in asc order
+	var currRange = crs.head;
+	for currRange != nil && currRange.next != nil {
+		checkRange := currRange.next
+		if checkRange != nil && checkRange.begin >= node.begin {
+			currRange.next = node
+			node.next = checkRange
+			node = nil
+			break
+
+		} else {
+			currRange = checkRange
+		}
+	}
+
+	if node != nil && currRange != nil {
+		currRange.next = node
+	}
+
+	crs.mergeRanges()
+}
+
+func (crs *coverRanges) mergeRanges()  {
+	// merge ranges
+	currRange := crs.head.next
+	for currRange != nil && currRange.next != nil {
+		checkRange := currRange.next
+		if currRange.end >= checkRange.begin && currRange.end < checkRange.end {
+			currRange.end = checkRange.end
+			currRange.next = checkRange.next
+			checkRange.Recovery()
+
+		} else {
+			currRange = currRange.next
+		}
+	}
+}
+
+
+type coveragePool struct {
+	queue  chan *coverageNode
+}
+
+
+func NewCoveragePool() (cp *coveragePool) {
+	return &coveragePool{
+		queue: make(chan *coverageNode, 256),
+	}
+}
+
+func (crp *coveragePool) NewCoverage(begin, end int64)(cn *coverageNode)  {
+	cn = crp.Dequeue()
+	cn.begin = begin
+	cn.end = end
+	return
+}
+
+func (crp *coveragePool) Enqueue(cn *coverageNode)  {
+	log.Debugf("coveragePool enqueue: %d", len(crp.queue))
+	if cn == nil {
+		return
+	}
+
+	crp.queue <- cn
+}
+
+func (crp *coveragePool) Dequeue() (cn *coverageNode)  {
+	log.Debugf("coveragePool dequeue: %d", len(crp.queue))
+
+	defer func() {
+		cn.begin = -1
+		cn.end = -1
+		cn.next = nil
+		cn.crp = crp
+	}()
+
+	select {
+	case cn = <- crp.queue:
+		return
+	default:
+		cn = &coverageNode{}
+		return
+	}
+}

session-dealer/mysql/model.go

@@ -7,128 +7,3 @@ type handshakeResponse41 struct {
 	DBName     string
 	Auth       []byte
 }
-
-// coverageNode record tcp package begin and end seq id
-type coverageNode struct {
-	begin int64
-	end   int64
-
-	next *coverageNode
-	crp *coveragePool
-}
-
-func newCoverage(begin, end int64) (*coverageNode) {
-	return &coverageNode{
-		begin: begin,
-		end: end,
-	}
-}
-
-func (crn *coverageNode) Recovery()  {
-	crn.crp.Enqueue(crn)
-}
-
-type coveragePool struct {
-	queue  []*coverageNode
-}
-
-
-func NewCoveragePool() (cp *coveragePool) {
-	return &coveragePool{
-		queue: make([]*coverageNode, 0, 256),
-	}
-}
-
-func (crp *coveragePool) Enqueue(cn *coverageNode)  {
-	if cn == nil {
-		return
-	}
-
-	crp.queue = append(crp.queue, cn)
-}
-
-func (crp *coveragePool) NewCoverage(begin, end int64)(cn *coverageNode)  {
-	cn = crp.Dequeue()
-	cn.begin = begin
-	cn.end = end
-	return
-}
-
-func (crp *coveragePool) Dequeue() (cn *coverageNode)  {
-	defer func() {
-		cn.begin = -1
-		cn.end = -1
-		cn.next = nil
-		cn.crp = crp
-	}()
-
-	if len(crp.queue) < 1 {
-		cn = &coverageNode{}
-		return
-	}
-
-	cn = crp.queue[0]
-	crp.queue = crp.queue[1:]
-	return
-}
-
-type coverRanges struct {
-	head *coverageNode
-}
-
-func NewCoverRanges() *coverRanges {
-	return &coverRanges{
-		head: &coverageNode{
-			begin: -1,
-		},
-	}
-}
-
-func (crs *coverRanges) clear() {
-	currRange := crs.head.next;
-	if currRange != nil {
-		node := currRange
-		currRange = currRange.next
-		node.Recovery()
-	}
-	crs.head.next = nil
-}
-
-func (crs *coverRanges) addRange(node *coverageNode)  {
-	// insert range in asc order
-	var currRange = crs.head;
-	for currRange != nil && currRange.next != nil {
-		checkRange := currRange.next
-		if checkRange != nil && checkRange.begin >= node.begin {
-			currRange.next = node
-			node.next = checkRange
-			node = nil
-			break
-
-		} else {
-			currRange = checkRange
-		}
-	}
-
-	if node != nil {
-		currRange.next = node
-	}
-
-	crs.mergeRanges()
-}
-
-func (crs *coverRanges) mergeRanges()  {
-	// merge ranges
-	currRange := crs.head.next
-	for currRange != nil && currRange.next != nil {
-		checkRange := currRange.next
-		if currRange.end >= checkRange.begin && currRange.end < checkRange.end {
-			currRange.end = checkRange.end
-			currRange.next = checkRange.next
-			checkRange.Recovery()
-
-		} else {
-			currRange = currRange.next
-		}
-	}
-}

session-dealer/mysql/session.go

@@ -122,7 +122,12 @@ func (ms *MysqlSession) checkFinish() bool {
 	return false
 }
 
+func (ms *MysqlSession) Close() {
+	ms.clear()
+}
+
 func (ms *MysqlSession) clear() {
+	localStmtCache.Enqueue(ms.cachedStmtBytes)
 	ms.cachedStmtBytes = nil
 	ms.expectReceiveSize = -1
 	ms.expectSendSize = -1
@@ -155,15 +160,13 @@ func (ms *MysqlSession) readFromClient(seqID int64, bytes []byte) {
 		ms.beginSeqID = seqID
 		ms.endSeqID = seqID
 
-		// if len(ms.cachedStmtBytes) > 0 {
+
-		// 	copy(newCache[:len(ms.cachedStmtBytes)], ms.cachedStmtBytes)
-		// }
 		if int64(ms.expectReceiveSize) < int64(len(contents)) {
 			log.Warnf("receive invalid mysql packet")
 			return
 		}
 
-		newCache := make([]byte, ms.expectReceiveSize)
+		newCache := localStmtCache.DequeueWithInit(ms.expectReceiveSize)
 		copy(newCache[:len(contents)], contents)
 		ms.cachedStmtBytes = newCache
 
@@ -310,5 +313,5 @@ func filterQueryPieceBySQL(mqp *model.PooledMysqlQueryPiece, querySQL []byte) (m
 func (ms *MysqlSession) composeQueryPiece() (mqp *model.PooledMysqlQueryPiece) {
 	return model.NewPooledMysqlQueryPiece(
 		ms.connectionID, ms.clientHost, ms.visitUser, ms.visitDB, ms.clientHost, ms.serverIP,
-		ms.clientPort, ms.serverPort, communicator.GetThrowPacketRate(), ms.stmtBeginTime)
+		ms.clientPort, ms.serverPort, communicator.GetMysqlThrowPacketRate(), ms.stmtBeginTime)
 }