mirror of
https://github.com/buger/goreplay.git
synced 2024-04-21 12:32:02 +00:00
Do not use buffer pool for messages
This buffers are get re-used and data gets corrupted
This commit is contained in:
+3
-119
@@ -9,108 +9,9 @@ import (
|
||||
"time"
|
||||
"unsafe"
|
||||
|
||||
"github.com/buger/goreplay/simpletime"
|
||||
"github.com/buger/goreplay/size"
|
||||
)
|
||||
|
||||
var bufferPool = NewBufferPool(1000, 1)
|
||||
|
||||
type buf struct {
|
||||
b []byte
|
||||
created time.Time
|
||||
gc bool
|
||||
}
|
||||
|
||||
type bufPool struct {
|
||||
buffers chan *buf
|
||||
ttl int
|
||||
}
|
||||
|
||||
func NewBufferPool(max int, ttl int) *bufPool {
|
||||
pool := &bufPool{
|
||||
buffers: make(chan *buf, max),
|
||||
ttl: ttl,
|
||||
}
|
||||
|
||||
// Ensure that memory released over time
|
||||
go func() {
|
||||
var released int
|
||||
// GC
|
||||
for {
|
||||
for i := 0; i < 100; i++ {
|
||||
select {
|
||||
case c := <-pool.buffers:
|
||||
if simpletime.Now.Sub(c.created) < time.Duration(ttl)*time.Second {
|
||||
select {
|
||||
case pool.buffers <- c:
|
||||
default:
|
||||
stats.Add("active_buffer_count", -1)
|
||||
c.b = nil
|
||||
c.gc = true
|
||||
released++
|
||||
}
|
||||
} else {
|
||||
stats.Add("active_buffer_count", -1)
|
||||
// Else GC
|
||||
c.b = nil
|
||||
c.gc = true
|
||||
released++
|
||||
}
|
||||
default:
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
bufPoolCount.Set(int64(len(pool.buffers)))
|
||||
releasedCount.Set(int64(released))
|
||||
|
||||
time.Sleep(1000 * time.Millisecond)
|
||||
}
|
||||
}()
|
||||
|
||||
return pool
|
||||
}
|
||||
|
||||
// Borrow a Client from the pool.
|
||||
func (p *bufPool) Get() *buf {
|
||||
var c *buf
|
||||
select {
|
||||
case c = <-p.buffers:
|
||||
default:
|
||||
stats.Add("total_alloc_buffer_count", 1)
|
||||
stats.Add("active_buffer_count", 1)
|
||||
|
||||
c = new(buf)
|
||||
c.b = make([]byte, 1024)
|
||||
c.created = simpletime.Now
|
||||
|
||||
// Use this technique to find if pool leaks, and objects get GCd
|
||||
//
|
||||
// runtime.SetFinalizer(c, func(p *buf) {
|
||||
// if !p.gc {
|
||||
// panic("Pool leak")
|
||||
// }
|
||||
// })
|
||||
}
|
||||
return c
|
||||
}
|
||||
|
||||
// Return returns a Client to the pool.
|
||||
func (p *bufPool) Put(c *buf) {
|
||||
select {
|
||||
case p.buffers <- c:
|
||||
default:
|
||||
stats.Add("active_buffers", -1)
|
||||
c.gc = true
|
||||
c.b = nil
|
||||
// if pool overloaded, let it go
|
||||
}
|
||||
}
|
||||
|
||||
func (p *bufPool) Len() int {
|
||||
return len(p.buffers)
|
||||
}
|
||||
|
||||
// Stats every message carry its own stats object
|
||||
type Stats struct {
|
||||
LostData int
|
||||
@@ -130,7 +31,6 @@ type Message struct {
|
||||
packets []*Packet
|
||||
parser *MessageParser
|
||||
feedback interface{}
|
||||
dataBuf *buf
|
||||
Stats
|
||||
}
|
||||
|
||||
@@ -164,8 +64,6 @@ func (m *Message) UUID() []byte {
|
||||
}
|
||||
|
||||
func (m *Message) add(packet *Packet) bool {
|
||||
// fmt.Println("SEQ:", packet.Seq, " - ", len(packet.Payload))
|
||||
|
||||
// Skip duplicates
|
||||
for _, p := range m.packets {
|
||||
if p.Seq == packet.Seq {
|
||||
@@ -228,20 +126,10 @@ func (m *Message) PacketData() [][]byte {
|
||||
|
||||
// Data returns data in this message
|
||||
func (m *Message) Data() []byte {
|
||||
m.dataBuf = bufferPool.Get()
|
||||
var tmp []byte
|
||||
tmp, _ = copySlice(tmp, m.PacketData()...)
|
||||
|
||||
// var totalLen int
|
||||
// for _, p := range m.packets {
|
||||
// totalLen += len(p.Payload)
|
||||
// }
|
||||
// tmp := make([]byte, totalLen)
|
||||
var n int
|
||||
if m.dataBuf == nil {
|
||||
panic("asdsd")
|
||||
}
|
||||
m.dataBuf.b, n = copySlice(m.dataBuf.b, m.PacketData()...)
|
||||
|
||||
return m.dataBuf.b[:n]
|
||||
return tmp
|
||||
}
|
||||
|
||||
// SetProtocolState set feedback/data that can be used later, e.g with End or Start hint
|
||||
@@ -264,10 +152,6 @@ func (m *Message) Finalize() {
|
||||
for _, p := range m.packets {
|
||||
packetPool.Put(p)
|
||||
}
|
||||
|
||||
if m.dataBuf != nil {
|
||||
bufferPool.Put(m.dataBuf)
|
||||
}
|
||||
}
|
||||
|
||||
// Emitter message handler
|
||||
|
||||
Reference in New Issue
Block a user