From 8daa7fdcab953d4741d82c74162ada4261c9e6b5 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Wed, 26 Aug 2015 22:25:58 +0300 Subject: [PATCH 1/4] Fix memory leaks --- Dockerfile | 2 ++ Makefile | 2 +- raw_socket_listener/listener.go | 2 +- raw_socket_listener/tcp_message.go | 11 ++++++++--- 4 files changed, 12 insertions(+), 5 deletions(-) diff --git a/Dockerfile b/Dockerfile index bf86e03..04779c5 100644 --- a/Dockerfile +++ b/Dockerfile @@ -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 \ No newline at end of file diff --git a/Makefile b/Makefile index 339df03..86b207f 100644 --- a/Makefile +++ b/Makefile @@ -20,7 +20,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 5s $(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 diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index 4fe232d..a29e722 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -257,4 +257,4 @@ func (t *Listener) Close() { close(t.quit) t.conn.Close() return -} +} \ No newline at end of file diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index cabac1d..d5dacee 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -46,7 +46,6 @@ func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32, expire *time 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() @@ -69,6 +68,10 @@ func (t *TCPMessage) listen() { // Timeout notifies message to stop listening, close channel and message ready to be sent func (t *TCPMessage) Timeout() { + if t.timer != nil { + t.timer.Stop() + } + select { // In some cases Timeout can be called multiple times (do not know how yet) // Ensure that we did not close channel 2 times @@ -139,12 +142,14 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { t.Timeout() } else { // If more then 1 packet, wait for more, and set expiration - if len(t.packets) == 1 { + 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) + } } } } From 661dd2026f482a66f9842d9e81387369adba5a21 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Tue, 1 Sep 2015 20:27:26 +0300 Subject: [PATCH 2/4] Fix grouting spawn loop and memory leaks --- .gitignore | 4 +++ Makefile | 8 ++++-- emitter.go | 4 +-- gor.go | 6 ++--- gor_stat.go | 5 ++-- input_dummy.go | 4 +-- input_http.go | 14 +++++++--- middleware_test.go | 13 ++++----- output_http.go | 7 +++-- raw_socket_listener/listener.go | 10 ++++--- raw_socket_listener/tcp_message.go | 42 ++++-------------------------- 11 files changed, 52 insertions(+), 65 deletions(-) diff --git a/.gitignore b/.gitignore index 0acc83e..21ac4c1 100644 --- a/.gitignore +++ b/.gitignore @@ -8,3 +8,7 @@ *.gz *.class + +*.test + +gor diff --git a/Makefile b/Makefile index 86b207f..d44977d 100644 --- a/Makefile +++ b/Makefile @@ -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 ./... -timeout 5s $(ARGS) -v + 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 diff --git a/emitter.go b/emitter.go index d767d45..d39320c 100644 --- a/emitter.go +++ b/emitter.go @@ -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) { diff --git a/gor.go b/gor.go index 4cd8596..61abfb3 100644 --- a/gor.go +++ b/gor.go @@ -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() }) diff --git a/gor_stat.go b/gor_stat.go index 4932cee..d7e14e4 100644 --- a/gor_stat.go +++ b/gor_stat.go @@ -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() { diff --git a/input_dummy.go b/input_dummy.go index 17907d3..632fb2f 100644 --- a/input_dummy.go +++ b/input_dummy.go @@ -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")...) } } } diff --git a/input_http.go b/input_http.go index f7717c5..d573d61 100644 --- a/input_http.go +++ b/input_http.go @@ -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) { diff --git a/middleware_test.go b/middleware_test.go index 8474eeb..e96b42b 100644 --- a/middleware_test.go +++ b/middleware_test.go @@ -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 = "" } diff --git a/output_http.go b/output_http.go index 9f82d37..0da99b1 100644 --- a/output_http.go +++ b/output_http.go @@ -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 { diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index a29e722..fb22bbd 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -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,7 +140,10 @@ func (t *Listener) readRAWSocket() { } if n > 0 { - go t.parsePacket(addr, buf[:n]) + newBuf := make([]byte, n) + copy(newBuf, buf[:n]) + + go t.parsePacket(addr, newBuf) } } } @@ -245,7 +249,7 @@ func (t *Listener) processTCPPacket(packet *TCPPacket) { } // Adding packet to message - message.packetsChan <- packet + message.AddPacket(packet) } // Receive TCP messages from the listener channel diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index d5dacee..d1bf98b 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -31,8 +31,6 @@ type TCPMessage struct { timer *time.Timer // Used for expire check - packetsChan chan *TCPPacket - delChan chan *TCPMessage expire *time.Duration @@ -44,51 +42,21 @@ 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 - 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() { if t.timer != nil { t.timer.Stop() } - 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 - } + // 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 } } @@ -142,7 +110,7 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { t.Timeout() } else { // If more then 1 packet, wait for more, and set expiration - if len(t.packets) > 1 { + if len(t.packets) == 1 { // Every time we receive packet we reset this timer t.timer = time.AfterFunc(*t.expire, t.Timeout) } else { From 4f3354507cabca5727e4b5fcf6b894993b62ef1c Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Tue, 1 Sep 2015 22:10:06 +0300 Subject: [PATCH 3/4] Fix race --- raw_socket_listener/listener.go | 2 -- raw_socket_listener/tcp_message.go | 7 ++++--- 2 files changed, 4 insertions(+), 5 deletions(-) diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index fb22bbd..a74480a 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -236,12 +236,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} diff --git a/raw_socket_listener/tcp_message.go b/raw_socket_listener/tcp_message.go index d1bf98b..d649f1c 100644 --- a/raw_socket_listener/tcp_message.go +++ b/raw_socket_listener/tcp_message.go @@ -49,9 +49,11 @@ func NewTCPMessage(ID string, delChan chan *TCPMessage, Ack uint32, expire *time // Timeout notifies message to stop listening, close channel and message ready to be sent func (t *TCPMessage) Timeout() { + 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 @@ -81,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 { @@ -109,6 +108,7 @@ 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 @@ -119,6 +119,7 @@ func (t *TCPMessage) AddPacket(packet *TCPPacket) { t.timer.Reset(*t.expire) } } + t.mu.Unlock() } } From 07a3c191d72ebaaa0b87af554dd77ceb57c071b6 Mon Sep 17 00:00:00 2001 From: Leonid Bugaev Date: Tue, 1 Sep 2015 22:16:29 +0300 Subject: [PATCH 4/4] Speedup input-raw --- raw_socket_listener/listener.go | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/raw_socket_listener/listener.go b/raw_socket_listener/listener.go index a74480a..8eb538b 100644 --- a/raw_socket_listener/listener.go +++ b/raw_socket_listener/listener.go @@ -140,20 +140,18 @@ func (t *Listener) readRAWSocket() { } if n > 0 { - newBuf := make([]byte, n) - copy(newBuf, buf[:n]) + if t.isValidPacket(buf[:n]) { + newBuf := make([]byte, n) + copy(newBuf, buf[:n]) - go t.parsePacket(addr, newBuf) + go func(newBuf []byte){ + t.packetsChan <- ParseTCPPacket(addr, newBuf) + }(newBuf) + } } } } -func (t *Listener) parsePacket(addr net.Addr, buf []byte) { - if t.isValidPacket(buf) { - t.packetsChan <- ParseTCPPacket(addr, buf) - } -} - func (t *Listener) isValidPacket(buf []byte) bool { // To avoid full packet parsing every time, we manually parsing values needed for packet filtering // http://en.wikipedia.org/wiki/Transmission_Control_Protocol