github/goreplay
4
0
Fork 0
mirror of https://github.com/buger/goreplay.git synced 2024-04-21 12:32:02 +00:00
Code Issues Releases Wiki Activity

Fix grouting spawn loop and memory leaks

Browse Source
This commit is contained in:
Leonid Bugaev
2015-09-01 20:27:26 +03:00
parent 8daa7fdcab
commit 661dd2026f
11 changed files with 52 additions and 65 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
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 ./... -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
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
+7 -3
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,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
raw_socket_listener/tcp_message.go
+5 -37
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,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 {