diff --git a/emitter.go b/emitter.go index 3db337c..1c985f1 100644 --- a/emitter.go +++ b/emitter.go @@ -152,7 +152,7 @@ func CopyMulty(src PluginReader, writers ...PluginWriter) error { } } else { for _, dst := range writers { - if _, err := dst.PluginWrite(msg); err != nil { + if _, err := dst.PluginWrite(msg); err != nil && err != io.ErrClosedPipe { return err } } diff --git a/input_file.go b/input_file.go index 5447492..3fa6d11 100644 --- a/input_file.go +++ b/input_file.go @@ -4,6 +4,7 @@ import ( "bufio" "bytes" "compress/gzip" + "container/heap" "errors" "fmt" "io" @@ -20,27 +21,71 @@ import ( "github.com/aws/aws-sdk-go/service/s3" ) -type fileInputReader struct { - reader *bufio.Reader +type filePayload struct { data []byte - file io.ReadCloser timestamp int64 - closed int32 // Value of 0 indicates that the file is still open. - s3 bool } -func (f *fileInputReader) parseNext() error { +// An IntHeap is a min-heap of ints. +type payloadQueue struct { + sync.RWMutex + s []*filePayload +} + +func (h payloadQueue) Len() int { return len(h.s) } +func (h payloadQueue) Less(i, j int) bool { return h.s[i].timestamp < h.s[j].timestamp } +func (h payloadQueue) Swap(i, j int) { h.s[i], h.s[j] = h.s[j], h.s[i] } + +func (h *payloadQueue) Push(x interface{}) { + // Push and Pop use pointer receivers because they modify the slice's length, + // not just its contents. + h.s = append(h.s, x.(*filePayload)) +} + +func (h *payloadQueue) Pop() interface{} { + old := h.s + n := len(old) + x := old[n-1] + h.s = old[0 : n-1] + return x +} + +func (h payloadQueue) Idx(i int) *filePayload { + h.RLock() + defer h.RUnlock() + + return h.s[i] +} + +type fileInputReader struct { + reader *bufio.Reader + file io.ReadCloser + closed int32 // Value of 0 indicates that the file is still open. + s3 bool + queue payloadQueue + readDepth int +} + +func (f *fileInputReader) parse(init chan struct{}) error { payloadSeparatorAsBytes := []byte(payloadSeparator) var buffer bytes.Buffer + var initialized bool + for { line, err := f.reader.ReadBytes('\n') if err != nil { if err != io.EOF { Debug(1, err) - } else { - f.Close() } + + f.Close() + + if !initialized { + close(init) + initialized = true + } + return err } @@ -48,21 +93,51 @@ func (f *fileInputReader) parseNext() error { asBytes := buffer.Bytes() meta := payloadMeta(asBytes) - f.timestamp, _ = strconv.ParseInt(string(meta[2]), 10, 64) - f.data = asBytes[:len(asBytes)-1] + timestamp, _ := strconv.ParseInt(string(meta[2]), 10, 64) + data := asBytes[:len(asBytes)-1] - return nil + f.queue.Lock() + heap.Push(&f.queue, &filePayload{ + timestamp: timestamp, + data: data, + }) + f.queue.Unlock() + + for { + if f.queue.Len() < f.readDepth { + break + } + + if !initialized { + close(init) + initialized = true + } + + time.Sleep(100 * time.Millisecond) + } + + buffer = bytes.Buffer{} + continue } buffer.Write(line) } - } -func (f *fileInputReader) ReadPayload() []byte { - defer f.parseNext() +func (f *fileInputReader) wait() { + for { + if atomic.LoadInt32(&f.closed) == 1 { + return + } - return f.data + if f.queue.Len() > 0 { + return + } + + time.Sleep(100 * time.Millisecond) + } + + return } // Close closes this plugin @@ -75,7 +150,7 @@ func (f *fileInputReader) Close() error { return nil } -func newFileInputReader(path string) *fileInputReader { +func newFileInputReader(path string, readDepth int) *fileInputReader { var file io.ReadCloser var err error @@ -90,7 +165,7 @@ func newFileInputReader(path string) *fileInputReader { return nil } - r := &fileInputReader{file: file, closed: 0} + r := &fileInputReader{file: file, closed: 0, readDepth: readDepth} if strings.HasSuffix(path, ".gz") { gzReader, err := gzip.NewReader(file) if err != nil { @@ -102,7 +177,11 @@ func newFileInputReader(path string) *fileInputReader { r.reader = bufio.NewReader(file) } - r.parseNext() + heap.Init(&r.queue) + + init := make(chan struct{}) + go r.parse(init) + <-init return r } @@ -116,16 +195,18 @@ type FileInput struct { readers []*fileInputReader speedFactor float64 loop bool + readDepth int } // NewFileInput constructor for FileInput. Accepts file path as argument. -func NewFileInput(path string, loop bool) (i *FileInput) { +func NewFileInput(path string, loop bool, readDepth int) (i *FileInput) { i = new(FileInput) i.data = make(chan []byte, 1000) i.exit = make(chan bool) i.path = path i.speedFactor = 1 i.loop = loop + i.readDepth = readDepth if err := i.init(); err != nil { return @@ -176,7 +257,7 @@ func (i *FileInput) init() (err error) { i.readers = make([]*fileInputReader, len(matches)) for idx, p := range matches { - i.readers[idx] = newFileInputReader(p) + i.readers[idx] = newFileInputReader(p, i.readDepth) } return nil @@ -201,11 +282,17 @@ func (i *FileInput) String() string { // Find reader with smallest timestamp e.g next payload in row func (i *FileInput) nextReader() (next *fileInputReader) { for _, r := range i.readers { - if r == nil || atomic.LoadInt32(&r.closed) != 0 { + if r == nil { continue } - if next == nil || r.timestamp < next.timestamp { + r.wait() + + if r.queue.Len() == 0 { + continue + } + + if next == nil || r.queue.Idx(0).timestamp > next.queue.Idx(0).timestamp { next = r continue } @@ -236,19 +323,23 @@ func (i *FileInput) emit() { } } + reader.queue.RLock() + payload := heap.Pop(&reader.queue).(*filePayload) + reader.queue.RUnlock() + if lastTime != -1 { - diff := reader.timestamp - lastTime + diff := payload.timestamp - lastTime if i.speedFactor != 1 { diff = int64(float64(diff) / i.speedFactor) } if diff >= 0 { - lastTime = reader.timestamp + lastTime = payload.timestamp time.Sleep(time.Duration(diff)) } } else { - lastTime = reader.timestamp + lastTime = payload.timestamp } // Recheck if we have exited since last check. @@ -256,7 +347,7 @@ func (i *FileInput) emit() { case <-i.exit: return default: - i.data <- reader.ReadPayload() + i.data <- payload.data } } diff --git a/input_file_test.go b/input_file_test.go index 91b9bcc..c6dcd8d 100644 --- a/input_file_test.go +++ b/input_file_test.go @@ -104,7 +104,7 @@ func TestInputFileMultipleFilesWithRequestsOnly(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -130,7 +130,7 @@ func TestInputFileRequestsWithLatency(t *testing.T) { file.Write([]byte("1 3 250000000\nrequest3")) file.Write([]byte(payloadSeparator)) - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), false, 100) start := time.Now().UnixNano() for i := 0; i < 3; i++ { @@ -170,7 +170,7 @@ func TestInputFileMultipleFilesWithRequestsAndResponses(t *testing.T) { file2.Write([]byte(payloadSeparator)) file2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) for i := '1'; i <= '4'; i++ { msg, _ := input.PluginRead() @@ -198,7 +198,7 @@ func TestInputFileLoop(t *testing.T) { file.Write([]byte(payloadSeparator)) file.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true) + input := NewFileInput(fmt.Sprintf("/tmp/%d", rnd), true, 100) // Even if we have just 2 requests in file, it should indifinitly loop for i := 0; i < 1000; i++ { @@ -226,7 +226,7 @@ func TestInputFileCompressed(t *testing.T) { name2 := output2.file.Name() output2.Close() - input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false) + input := NewFileInput(fmt.Sprintf("/tmp/%d*", rnd), false, 100) for i := 0; i < 2000; i++ { input.PluginRead() } @@ -326,7 +326,7 @@ func CreateCaptureFile(requestGenerator *RequestGenerator) *CaptureFile { func ReadFromCaptureFile(captureFile *os.File, count int, callback writeCallback) (err error) { wg := new(sync.WaitGroup) - input := NewFileInput(captureFile.Name(), false) + input := NewFileInput(captureFile.Name(), false, 100) output := NewTestOutput(func(msg *Message) { callback(msg) wg.Done() diff --git a/output_dummy.go b/output_dummy.go index e28653f..3d67a59 100644 --- a/output_dummy.go +++ b/output_dummy.go @@ -24,6 +24,7 @@ func (i *DummyOutput) PluginWrite(msg *Message) (int, error) { n += nn nn, err = os.Stdout.Write(payloadSeparatorAsBytes) n += nn + return n, err } diff --git a/output_file_test.go b/output_file_test.go index d68c183..53794fb 100644 --- a/output_file_test.go +++ b/output_file_test.go @@ -39,7 +39,7 @@ func TestFileOutput(t *testing.T) { emitter.Close() var counter int64 - input2 := NewFileInput("/tmp/test_requests.gor", false) + input2 := NewFileInput("/tmp/test_requests.gor", false, 100) output2 := NewTestOutput(func(*Message) { atomic.AddInt64(&counter, 1) wg.Done() diff --git a/plugins.go b/plugins.go index b1cef48..1f1363c 100644 --- a/plugins.go +++ b/plugins.go @@ -83,7 +83,6 @@ func (plugins *InOutPlugins) registerPlugin(constructor interface{}, options ... plugins.Outputs = append(plugins.Outputs, w) } plugins.All = append(plugins.All, plugin) - } // NewPlugins specify and initialize all available plugins @@ -119,7 +118,7 @@ func NewPlugins() *InOutPlugins { } for _, options := range Settings.InputFile { - plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop) + plugins.registerPlugin(NewFileInput, options, Settings.InputFileLoop, Settings.InputFileReadDepth) } for _, path := range Settings.OutputFile { diff --git a/s3_test.go b/s3_test.go index fc48be5..f1f85ff 100644 --- a/s3_test.go +++ b/s3_test.go @@ -127,7 +127,7 @@ func TestInputFileFromS3(t *testing.T) { <-output.closeCh } - input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd), false) + input := NewFileInput(fmt.Sprintf("s3://test-gor-eu/%d", rnd, 100), false) buf := make([]byte, 1000) for i := 0; i <= 19999; i++ { diff --git a/settings.go b/settings.go index 954aa0a..2ecdfc7 100644 --- a/settings.go +++ b/settings.go @@ -45,10 +45,11 @@ type AppSettings struct { OutputTCPConfig TCPOutputConfig OutputTCPStats bool `json:"output-tcp-stats"` - InputFile MultiOption `json:"input-file"` - InputFileLoop bool `json:"input-file-loop"` - OutputFile MultiOption `json:"output-file"` - OutputFileConfig FileOutputConfig + InputFile MultiOption `json:"input-file"` + InputFileLoop bool `json:"input-file-loop"` + InputFileReadDepth int `json:"input-file-read-depth"` + OutputFile MultiOption `json:"output-file"` + OutputFileConfig FileOutputConfig InputRAW MultiOption `json:"input_raw"` RAWInputConfig @@ -113,6 +114,7 @@ func init() { flag.Var(&Settings.InputFile, "input-file", "Read requests from file: \n\tgor --input-file ./requests.gor --output-http staging.com") flag.BoolVar(&Settings.InputFileLoop, "input-file-loop", false, "Loop input files, useful for performance testing.") + flag.IntVar(&Settings.InputFileReadDepth, "input-file-read-depth", 100, "GoReplay tries to read and cache multiple records, in advance. In parallel it also perform sorting of requests, if they came out of order. Since it needs hold this buffer in memory, bigger values can cause worse performance") flag.Var(&Settings.OutputFile, "output-file", "Write incoming requests to file: \n\tgor --input-raw :80 --output-file ./requests.gor") flag.DurationVar(&Settings.OutputFileConfig.FlushInterval, "output-file-flush-interval", time.Second, "Interval for forcing buffer flush to the file, default: 1s.")