From db32671c52731d45523aee9a33d997a5aaa30d6a Mon Sep 17 00:00:00 2001
From: Leonid Bugaev <leonsbox@gmail.com>
Date: Sun, 27 Jun 2021 21:05:10 +0000
Subject: [PATCH] Memory utilization improvements

Do not use sync.Pool, use channel based impementation instead. Sync pool does not work for long lived objects, and trigger frequent GC.
Because of MUCH more efficient memory utilization, GOGC now set to 500, which significantly improve CPU as wel.
---
 gor.go             |  6 +++++-
 tcp/tcp_message.go | 40 ++++++++++++++--------------------------
 tcp/tcp_packet.go  | 39 +++++++++++++++++++++++++++++++++------
 3 files changed, 52 insertions(+), 33 deletions(-)

diff --git a/gor.go b/gor.go
index abad0e0..da0fd1f 100644
--- a/gor.go
+++ b/gor.go
@@ -11,7 +11,7 @@ import (
 	"os"
 	"os/signal"
 	"runtime"
-	_ "runtime/debug"
+	"runtime/debug"
 	"runtime/pprof"
 	"syscall"
 	"time"
@@ -40,6 +40,10 @@ func main() {
 		runtime.GOMAXPROCS(runtime.NumCPU() * 2)
 	}
 
+	if os.Getenv("GOGC") == "" {
+		debug.SetGCPercent(500)
+	}
+
 	args := os.Args[1:]
 	var plugins *InOutPlugins
 	if len(args) > 0 && args[0] == "file-server" {
diff --git a/tcp/tcp_message.go b/tcp/tcp_message.go
index d6d1746..4db342d 100644
--- a/tcp/tcp_message.go
+++ b/tcp/tcp_message.go
@@ -9,7 +9,6 @@ import (
 	"time"
 	"unsafe"
 
-	"github.com/buger/goreplay/ring"
 	"github.com/buger/goreplay/size"
 )
 
@@ -189,8 +188,8 @@ type MessageParser struct {
 	End           HintEnd
 	Start         HintStart
 	ticker        *time.Ticker
-	messages      *ring.RingBuffer
-	packets       *ring.RingBuffer
+	messages      chan *Message
+	packets       chan *Packet
 	close         chan struct{} // to signal that we are able to close
 }
 
@@ -207,8 +206,8 @@ func NewMessageParser(maxSize size.Size, messageExpire time.Duration, debugger D
 		parser.maxSize = 5 << 20
 	}
 
-	parser.packets = ring.NewRingBuffer(10000)
-	parser.messages = ring.NewRingBuffer(10000)
+	parser.packets = make(chan *Packet, 10000)
+	parser.messages = make(chan *Message, 10000)
 
 	parser.m = make(map[uint64]*Message)
 	parser.ticker = time.NewTicker(time.Millisecond * 50)
@@ -222,7 +221,7 @@ var packetLen int
 // Packet returns packet handler
 func (parser *MessageParser) PacketHandler(packet *Packet) {
 	packetLen++
-	parser.packets.Offer(packet)
+	parser.packets <- packet
 }
 
 var processedPackets int
@@ -232,15 +231,9 @@ func (parser *MessageParser) wait() {
 		now time.Time
 	)
 	for {
-		pckt, err := parser.packets.Poll(-1)
-		if err == nil {
-			processedPackets++
-			parser.processPacket(pckt.(*Packet))
-		} else {
-			time.Sleep(50 * time.Millisecond)
-		}
-
 		select {
+		case pckt := <-parser.packets:
+			parser.processPacket(pckt)
 		case now = <-parser.ticker.C:
 			parser.timer(now)
 		case <-parser.close:
@@ -248,7 +241,7 @@ func (parser *MessageParser) wait() {
 			// parser.Close should wait for this function to return
 			parser.close <- struct{}{}
 			return
-		default:
+			// default:
 		}
 	}
 }
@@ -270,9 +263,9 @@ func (parser *MessageParser) processPacket(pckt *Packet) {
 				// Requeue not known packets
 				pckt.Retry++
 
-				if ok, _ := parser.packets.Offer(pckt); !ok {
-					// Drop packet if it does not fit to ring buffer
-					pckt.Payload = pckt.Payload[:]
+				select {
+				case parser.packets <- pckt:
+				default:
 					packetPool.Put(pckt)
 				}
 			}
@@ -308,19 +301,14 @@ func (parser *MessageParser) addPacket(m *Message, pckt *Packet) {
 }
 
 func (parser *MessageParser) Read() *Message {
-	for {
-		if m, err := parser.messages.Poll(-1); err != nil {
-			time.Sleep(50 * time.Millisecond)
-		} else {
-			return m.(*Message)
-		}
-	}
+	m := <-parser.messages
+	return m
 }
 
 func (parser *MessageParser) Emit(m *Message) {
 	delete(parser.m, m.packets[0].MessageID())
 
-	parser.messages.Offer(m)
+	parser.messages <- m
 }
 
 func GetUnexportedField(field reflect.Value) interface{} {
diff --git a/tcp/tcp_packet.go b/tcp/tcp_packet.go
index 049f29e..7786c69 100644
--- a/tcp/tcp_packet.go
+++ b/tcp/tcp_packet.go
@@ -4,7 +4,6 @@ import (
 	"encoding/binary"
 	"fmt"
 	"net"
-	"sync"
 	"time"
 
 	"github.com/google/gopacket"
@@ -18,10 +17,38 @@ func copySlice(b, a []byte) []byte {
 	return b[:len(a)]
 }
 
-var packetPool = &sync.Pool{
-	New: func() interface{} {
-		return new(Packet)
-	},
+var packetPool = NewPool(10000)
+
+// Pool holds Clients.
+type Pool struct {
+	pool chan *Packet
+}
+
+// NewPool creates a new pool of Clients.
+func NewPool(max int) *Pool {
+	return &Pool{
+		pool: make(chan *Packet, max),
+	}
+}
+
+// Borrow a Client from the pool.
+func (p *Pool) Get() *Packet {
+	var c *Packet
+	select {
+	case c = <-p.pool:
+	default:
+		c = new(Packet)
+	}
+	return c
+}
+
+// Return returns a Client to the pool.
+func (p *Pool) Put(c *Packet) {
+	select {
+	case p.pool <- c:
+	default:
+		// let it go, let it go...
+	}
 }
 
 /*
@@ -46,7 +73,7 @@ type Packet struct {
 
 // ParsePacket parse raw packets
 func ParsePacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo) (pckt *Packet, err error) {
-	pckt = packetPool.Get().(*Packet)
+	pckt = packetPool.Get()
 	if err := pckt.parse(data, lType, lTypeLen, cp); err != nil {
 		packetPool.Put(pckt)
 		return nil, err