github/goreplay
4
0
Fork 0
mirror of https://github.com/buger/goreplay.git synced 2026-10-21 05:22:12 +00:00
Code Issues Releases Wiki Activity

Merge pull request #194 from buger/fix-memory-leaks

Browse Source 
Fix memory leaks
This commit is contained in:
Leonid Bugaev
2015-09-01 22:21:13 +03:00
parent 67505d99d8 07a3c191d7
commit 323f85bcfb
12 changed files with 69 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+4
View File
@@ -8,3 +8,7 @@
*.gz
*.class
*.test
gor
Dockerfile
+2
View File
@@ -20,5 +20,7 @@ ADD . /gopath/src/github.com/buger/gor/
RUN javac -cp /tmp/commons-io-2.4/commons-io-2.4.jar ./examples/middleware/echo.java
RUN apt-get install graphviz -y
RUN go get -u github.com/golang/lint/golint
RUN go get
Makefile
+6 -2
View File
@@ -13,6 +13,10 @@ release-x86:
dbuild:
docker build -t gor .
profile:
go build && ./gor --output-http="http://localhost:9000" --input-dummy 0 --input-raw :9000 --input-http :9000 --memprofile=./mem.out --cpuprofile=./cpu.out --stats --output-http-stats --output-http-timeout 100ms
dlint:
docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor golint $(PKG)
@@ -20,7 +24,7 @@ drace:
docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test ./... $(ARGS) -v -race -timeout 15s
dtest:
docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go test ./... $(ARGS) -v -timeout 60s
docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go test ./ -timeout 60s $(ARGS) -v
dcover:
docker run -v `pwd`:$(SOURCE_PATH) -t -i --env GORACE="halt_on_error=1" gor go test $(ARGS) -race -v -timeout 15s -coverprofile=coverage.out
@@ -40,7 +44,7 @@ drun:
docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-http="http://localhost:9000" --input-raw :9000 --input-http :9000 --verbose --debug --middleware "./examples/middleware/echo.sh"
drun-2:
docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-file="./fixtures/requests.gor" --output-dummy=0 --verbose --debug --middleware "java -cp ./examples/middleware echo"
docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-http :9001 --output-dummy=0
drecord:
docker run -v `pwd`:$(SOURCE_PATH) -t -i gor go run $(SOURCE) --input-dummy=0 --output-file=requests.gor --verbose --debug
emitter.go
+2 -2
View File
@@ -48,7 +48,7 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
nr, er := src.Read(buf)
if nr > 0 && len(buf) > nr {
payload := buf[0:nr]
payload := buf[:nr]
_maxN := nr
if nr > 500 {
@@ -56,7 +56,7 @@ func CopyMulty(src io.Reader, writers ...io.Writer) (err error) {
}
if Settings.debug {
Debug("[EMITTER] input:", string(payload[0:_maxN]))
Debug("[EMITTER] input:", string(payload[0:_maxN]), nr, "from:", src)
}
if modifier != nil && isRequestPayload(payload) {
gor.go
+3 -3
View File
@@ -60,10 +60,10 @@ func profileCPU(cpuprofile string) {
}
pprof.StartCPUProfile(f)
time.AfterFunc(60*time.Second, func() {
time.AfterFunc(30*time.Second, func() {
pprof.StopCPUProfile()
f.Close()
log.Println("Stop profiling after 60 seconds")
log.Println("Stop profiling after 30 seconds")
})
}
}
@@ -74,7 +74,7 @@ func profileMEM(memprofile string) {
if err != nil {
log.Fatal(err)
}
time.AfterFunc(60*time.Second, func() {
time.AfterFunc(30*time.Second, func() {
pprof.WriteHeapProfile(f)
f.Close()
})
gor_stat.go
+3 -2
View File
@@ -4,6 +4,7 @@ import (
"log"
"strconv"
"time"
"runtime"
)
const (
@@ -27,7 +28,7 @@ func NewGorStat(statName string) (s *GorStat) {
s.count = 0
if Settings.stats {
log.Println(s.statName + ":latest,mean,max,count,count/second")
log.Println(s.statName + ":latest,mean,max,count,count/second,gcount")
go s.reportStats()
}
return
@@ -54,7 +55,7 @@ func (s *GorStat) Reset() {
}
func (s *GorStat) String() string {
return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count/rate)
return s.statName + ":" + strconv.Itoa(s.latest) + "," + strconv.Itoa(s.mean) + "," + strconv.Itoa(s.max) + "," + strconv.Itoa(s.count) + "," + strconv.Itoa(s.count/rate) + "," + strconv.Itoa(runtime.NumGoroutine())
}
func (s *GorStat) reportStats() {
input_dummy.go
+2 -2
View File
@@ -35,10 +35,10 @@ func (i *DummyInput) emit() {
case <-ticker.C:
uuid := uuid()
reqh := payloadHeader(RequestPayload, uuid, time.Now().UnixNano())
i.data <- append(reqh, []byte("POST /pub/WWW/å HTTP/1.1\nHost: www.w3.org\r\nContent-Length: 7\r\n\r\na=1&b=2")...)
i.data <- append(reqh, []byte("POST /pub/WWW/å HTTP/1.1\r\nHost: www.w3.org\r\nUser-Agent: Go 1.1 package http\r\nAccept-Encoding: gzip\r\nContent-Length: 7\r\n\r\na=1&b=2")...)
resh := payloadHeader(ResponsePayload, uuid, 1)
i.data <- append(resh, []byte("HTTP/1.1 200 OK\r\n\r\n")...)
i.data <- append(resh, []byte("HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n")...)
}
}
}
input_http.go
+10 -4
View File
@@ -18,7 +18,7 @@ type HTTPInput struct {
// NewHTTPInput constructor for HTTPInput. Accepts address with port which he will listen on.
func NewHTTPInput(address string) (i *HTTPInput) {
i = new(HTTPInput)
i.data = make(chan []byte)
i.data = make(chan []byte, 10000)
i.address = address
i.listen(address)
@@ -38,11 +38,17 @@ func (i *HTTPInput) Read(data []byte) (int, error) {
}
func (i *HTTPInput) handler(w http.ResponseWriter, r *http.Request) {
buf, _ := httputil.DumpRequest(r, true)
i.data <- buf
r.URL.Scheme = "http"
r.URL.Host = i.listener.Addr().String()
buf, _ := httputil.DumpRequestOut(r, true)
http.Error(w, http.StatusText(200), 200)
select {
case i.data <- buf:
default:
Debug("[INPUT-HTTP] Dropping requests because output can't process them fast enough")
}
}
func (i *HTTPInput) listen(address string) {
middleware_test.go
+7 -6
View File
@@ -131,17 +131,18 @@ func TestEchoMiddleware(t *testing.T) {
time.Sleep(time.Millisecond)
// Should receive 2 requests from original + 2 from replayed
wg.Add(4)
client := NewHTTPClient(from.URL, &HTTPClientConfig{Debug: false})
// Request should be echoed
client.Get("/a")
client.Get("/b")
for i:=0; i<10; i++ {
wg.Add(4)
// Request should be echoed
client.Get("/a")
client.Get("/b")
}
wg.Wait()
close(quit)
time.Sleep(100 * time.Millisecond)
time.Sleep(200 * time.Millisecond)
Settings.middleware = ""
}
output_http.go
+3 -4
View File
@@ -3,7 +3,6 @@ package main
import (
"github.com/buger/gor/proto"
"io"
"log"
"sync/atomic"
"time"
)
@@ -69,8 +68,8 @@ func NewHTTPOutput(address string, config *HTTPOutputConfig) io.Writer {
o.queueStats = NewGorStat("output_http")
}
o.queue = make(chan []byte, 100)
o.responses = make(chan response, 100)
o.queue = make(chan []byte, 1000)
o.responses = make(chan response, 1000)
o.needWorker = make(chan int, 1)
// Initial workers count
@@ -197,7 +196,7 @@ func (o *HTTPOutput) sendRequest(client *HTTPClient, request []byte) {
stop := time.Now()
if err != nil {
log.Println("Request error:", err)
Debug("Request error:", err)
}
if o.config.TrackResponses {
raw_socket_listener/listener.go
+12 -12
View File
@@ -124,8 +124,9 @@ func (t *Listener) readRAWSocket() {
defer t.conn.Close()
buf := make([]byte, 64*1024) // 64kb
for {
buf := make([]byte, 64*1024) // 64kb
// Note: ReadFrom receive messages without IP header
n, addr, err := t.conn.ReadFrom(buf)
@@ -139,14 +140,15 @@ func (t *Listener) readRAWSocket() {
}
if n > 0 {
go t.parsePacket(addr, buf[:n])
}
}
}
if t.isValidPacket(buf[:n]) {
newBuf := make([]byte, n)
copy(newBuf, buf[:n])
func (t *Listener) parsePacket(addr net.Addr, buf []byte) {
if t.isValidPacket(buf) {
t.packetsChan <- ParseTCPPacket(addr, buf)
go func(newBuf []byte){
t.packetsChan <- ParseTCPPacket(addr, newBuf)
}(newBuf)
}
}
}
}
@@ -232,12 +234,10 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
}
if t.captureResponse && isIncoming {
message.mu.Lock()
// If message have multiple packets, delete previous alias
if len(message.packets) > 0 {
delete(t.respAliases, message.ResponseAck)
}
message.mu.Unlock()
responseAck := packet.Seq + uint32(len(packet.Data))
t.respAliases[responseAck] = &request{message.Start, message.Ack}
@@ -245,7 +245,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) {
}
// Adding packet to message
message.packetsChan <- packet
message.AddPacket(packet)
}
// Receive TCP messages from the listener channel
@@ -257,4 +257,4 @@ func (t *Listener) Close() {
close(t.quit)
t.conn.Close()
return
}
}
raw_socket_listener/tcp_message.go
+15 -41
View File
@@ -31,8 +31,6 @@ type TCPMessage struct {
timer *time.Timer // Used for expire check
packetsChan chan *TCPPacket
delChan chan *TCPMessage
expire *time.Duration
@@ -44,48 +42,23 @@ type TCPMessage struct {
func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32, expire *time.Duration, IsIncoming bool) (msg *TCPMessage) {
msg = &TCPMessage{ID: ID, Ack: Ack, expire: expire, IsIncoming: IsIncoming}
msg.Start = time.Now().UnixNano()
msg.packetsChan = make(chan *TCPPacket)
msg.delChan = delChan // used for notifying that message completed or expired
msg.timer = time.NewTimer(0)
go msg.listen()
return
}
func (t *TCPMessage) listen() {
for {
select {
case packet, more := <-t.packetsChan:
if more {
t.AddPacket(packet)
} else {
// Stop loop if channel closed
return
}
}
}
}
// Timeout notifies message to stop listening, close channel and message ready to be sent
func (t *TCPMessage) Timeout() {
select {
// In some cases Timeout can be called multiple times (do not know how yet)
// Ensure that we did not close channel 2 times
case packet, ok := <-t.packetsChan:
if ok {
t.AddPacket(packet)
t.Timeout()
} else {
return
}
default:
close(t.packetsChan)
// Notify RAWListener that message is ready to be send to replay server
// Responses without requests gets discarded
if t.IsIncoming || t.RequestStart != 0 {
t.delChan <- t
}
t.mu.Lock()
if t.timer != nil {
t.timer.Stop()
}
t.mu.Unlock()
// Notify RAWListener that message is ready to be send to replay server
// Responses without requests gets discarded
if t.IsIncoming || t.RequestStart != 0 {
t.delChan <- t
}
}
@@ -110,9 +83,6 @@ func (t *TCPMessage) Size() (size int) {
// AddPacket to the message and ensure packet uniqueness
// TCP allows that packet can be re-send multiple times
func (t *TCPMessage) AddPacket(packet *TCPPacket) {
t.mu.Lock()
defer t.mu.Unlock()
packetFound := false
for _, pkt := range t.packets {
@@ -138,14 +108,18 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) {
if !t.isMultipart() {
t.Timeout()
} else {
t.mu.Lock()
// If more then 1 packet, wait for more, and set expiration
if len(t.packets) == 1 {
// Every time we receive packet we reset this timer
t.timer = time.AfterFunc(*t.expire, t.Timeout)
} else {
// Reset message timeout timer
t.timer.Reset(*t.expire)
if t.timer != nil {
t.timer.Reset(*t.expire)
}
}
t.mu.Unlock()
}
}