7x31
/
sniffer-agent
mirror of https://github.com/zr-hebo/sniffer-agent.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

修复设置抓包率引起的查询时间异常的问题

This commit is contained in:
hebo 2019-11-14 21:52:19 +08:00
parent 16722c878c
commit 85bef28022
2 changed files with 20 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
session-dealer/mysql/cover_range.go
View File

@ -6,17 +6,17 @@ type coverageNode struct {
end int64 end int64
next *coverageNode next *coverageNode
crp *coveragePool crp *coveragePool
} }
func newCoverage(begin, end int64) (*coverageNode) { func newCoverage(begin, end int64) (*coverageNode) {
return &coverageNode{ return &coverageNode{
begin: begin, begin: begin,
end: end, end: end,
} }
} }
func (crn *coverageNode) Recovery() { func (crn *coverageNode) Recovery() {
crn.crp.Enqueue(crn) crn.crp.Enqueue(crn)
} }
@ -28,6 +28,7 @@ func NewCoverRanges() *coverRanges {
return &coverRanges{ return &coverRanges{
head: &coverageNode{ head: &coverageNode{
begin: -1, begin: -1,
end: -1,
}, },
} }
} }
@ -42,7 +43,7 @@ func (crs *coverRanges) clear() {
crs.head.next = nil crs.head.next = nil
} }
func (crs *coverRanges) addRange(node *coverageNode) { func (crs *coverRanges) addRange(node *coverageNode) {
// insert range in asc order // insert range in asc order
var currRange = crs.head; var currRange = crs.head;
for currRange != nil && currRange.next != nil { for currRange != nil && currRange.next != nil {
@ -52,10 +53,9 @@ func (crs *coverRanges) addRange(node *coverageNode) {
node.next = checkRange node.next = checkRange
node = nil node = nil
break break
} else {
currRange = checkRange
} }
currRange = checkRange
} }
if node != nil && currRange != nil { if node != nil && currRange != nil {
@ -65,12 +65,12 @@ func (crs *coverRanges) addRange(node *coverageNode) {
crs.mergeRanges() crs.mergeRanges()
} }
func (crs *coverRanges) mergeRanges() { func (crs *coverRanges) mergeRanges() {
// merge ranges // merge ranges
currRange := crs.head.next currRange := crs.head.next
for currRange != nil && currRange.next != nil { for currRange != nil && currRange.next != nil {
checkRange := currRange.next checkRange := currRange.next
if currRange.end >= checkRange.begin && currRange.end < checkRange.end { if currRange.begin <= checkRange.begin && currRange.end >= checkRange.begin && currRange.end < checkRange.end {
currRange.end = checkRange.end currRange.end = checkRange.end
currRange.next = checkRange.next currRange.next = checkRange.next
checkRange.Recovery() checkRange.Recovery()
@ -81,19 +81,17 @@ func (crs *coverRanges) mergeRanges() {
} }
} }
type coveragePool struct { type coveragePool struct {
queue chan *coverageNode queue chan *coverageNode
} }
func NewCoveragePool() (cp *coveragePool) { func NewCoveragePool() (cp *coveragePool) {
return &coveragePool{ return &coveragePool{
queue: make(chan *coverageNode, 256), queue: make(chan *coverageNode, 256),
} }
} }
func (crp *coveragePool) NewCoverage(begin, end int64)(cn *coverageNode) { func (crp *coveragePool) NewCoverage(begin, end int64) (cn *coverageNode) {
cn = crp.Dequeue() cn = crp.Dequeue()
cn.begin = begin cn.begin = begin
cn.end = end cn.end = end
@ -115,7 +113,7 @@ func (crp *coveragePool) Enqueue(cn *coverageNode) {
} }
} }
func (crp *coveragePool) Dequeue() (cn *coverageNode) { func (crp *coveragePool) Dequeue() (cn *coverageNode) {
// log.Debugf("coveragePool dequeue: %d", len(crp.queue)) // log.Debugf("coveragePool dequeue: %d", len(crp.queue))
defer func() { defer func() {
@ -126,7 +124,7 @@ func (crp *coveragePool) Dequeue() (cn *coverageNode) {
}() }()
select { select {
case cn = <- crp.queue: case cn = <-crp.queue:
return return
default: default:

8
session-dealer/mysql/session.go
View File

@ -87,7 +87,7 @@ func (ms *MysqlSession) ReceiveTCPPacket(newPkt *model.TCPPacket) {
ms.readFromClient(newPkt.Seq, newPkt.Payload) ms.readFromClient(newPkt.Seq, newPkt.Payload)
} else { } else {
ms.readFromServer(newPkt.Payload) ms.readFromServer(newPkt.Seq, newPkt.Payload)
qp := ms.GenerateQueryPiece() qp := ms.GenerateQueryPiece()
if qp != nil { if qp != nil {
ms.queryPieceReceiver <- qp ms.queryPieceReceiver <- qp
@ -99,7 +99,7 @@ func (ms *MysqlSession) resetBeginTime() {
ms.stmtBeginTime = time.Now().UnixNano() / millSecondUnit ms.stmtBeginTime = time.Now().UnixNano() / millSecondUnit
} }
func (ms *MysqlSession) readFromServer(bytes []byte) { func (ms *MysqlSession) readFromServer(respSeq int64, bytes []byte) {
if ms.expectSendSize < 1 && len(bytes) > 4 { if ms.expectSendSize < 1 && len(bytes) > 4 {
ms.expectSendSize = extractMysqlPayloadSize(bytes[:4]) ms.expectSendSize = extractMysqlPayloadSize(bytes[:4])
contents := bytes[4:] contents := bytes[4:]
@ -107,6 +107,10 @@ func (ms *MysqlSession) readFromServer(bytes []byte) {
ms.prepareInfo.prepareStmtID = bytesToInt(contents[1:5]) ms.prepareInfo.prepareStmtID = bytesToInt(contents[1:5])
} }
} }
if ms.coverRanges.head.next == nil || ms.coverRanges.head.next.end != respSeq {
ms.clear()
}
} }
func (ms *MysqlSession) checkFinish() bool { func (ms *MysqlSession) checkFinish() bool {