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

Benchmarking, packaging, and fix issues, tests and perfomance (#797)

Browse Source 
### performance
- handling of the very big packet(any size that can be buffered)
- speeding up TCP sessions by using message hints: Added **proto.HasFullPayload** that helps to validate the entire HTTP request, it supports `Chunked` encoding too! Added **proto.HasRequestTitle** and **proto.HasResponseTitle** for validating the beginning of HTTP request. Those methods are used `input_raw.go` with `TCP`.
- supports Keep-Alive: the above functions helps to support keep-alive

### Packaging
- **capture:** engines(capture/doc.go)
- **tcp:** tcp message parser (tcp/doc.go)

### benchmarking
- **capture.BenchmarkPcapDump:** the benchmarks regarding dumping packets in a pcap file
- **capture.BenchmarkPcapFile:** the benchmarks of reading packets from a pcap file
- **capture.BenchmarkPcap:** the benchmarks of parsing packets from the loopback interface with pcap handles
- **proto.BenchmarkHasFullPayload:**: benchmarking this function which validates the HTTP payload
- **tcp.BenchmarkPacketParseAndSort:** benchmarks of parsing and sorting packets
- **tcp.BenchmarkMessageParserWithoutHint:** benchmarks of message reasembling by using `SYN` and `FIN` flag
- **tcp.BenchmarkMessageParserWithHint:** benchmarks of message reasembling by using `proto.HasRequestTitle` and `proto.HasFullPayload` flag

### issues
see linked issues

###  tests
- fixed input raw and engine tests

**Most of the changed of the files, was about using functionalities of** `tcp` **and** `capture` **in existing functionalities**
This commit is contained in:
Urban Ishimwe
2020-08-11 12:44:53 +03:00
committed by GitHub
parent 2f81fba370
commit fdc8b094f0
60 changed files with 3106 additions and 3825 deletions
Download Patch File Download Diff File Expand all files Collapse all files
emitter.go
+32 -41
View File
@@ -10,8 +10,10 @@ import (
)
type emitter struct {
sync.Mutex
sync.WaitGroup
quit chan int
quit chan int
plugins *InOutPlugins
}
// NewEmitter creates and initializes new `emitter` object.
@@ -23,8 +25,11 @@ func NewEmitter(quit chan int) *emitter {
// Start initialize loop for sending data from inputs to outputs
func (e *emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
e.Add(1)
defer e.Done()
defer e.Wait()
if Settings.CopyBufferSize < 1 {
Settings.CopyBufferSize = 5 << 20
}
e.plugins = plugins
if middlewareCmd != "" {
middleware := NewMiddleware(middlewareCmd)
@@ -43,8 +48,8 @@ func (e *emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
go func() {
defer e.Done()
if err := CopyMulty(e.quit, middleware, plugins.Outputs...); err != nil {
log.Println("Error during copy: ", err)
e.close()
Debug(2, "Error during copy: ", err)
e.Close()
}
}()
go func() {
@@ -62,8 +67,8 @@ func (e *emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
go func(in io.Reader) {
defer e.Done()
if err := CopyMulty(e.quit, in, plugins.Outputs...); err != nil {
log.Println("Error during copy: ", err)
e.close()
Debug(2, "Error during copy: ", err)
e.Close()
}
}(in)
}
@@ -74,22 +79,13 @@ func (e *emitter) Start(plugins *InOutPlugins, middlewareCmd string) {
go func(r io.Reader) {
defer e.Done()
if err := CopyMulty(e.quit, r, plugins.Outputs...); err != nil {
log.Println("Error during copy: ", err)
e.close()
Debug(2, "Error during copy: ", err)
e.Close()
}
}(r)
}
}
}
for {
select {
case <-e.quit:
finalize(plugins)
return
case <-time.After(100 * time.Millisecond):
}
}
}
func (e *emitter) close() {
@@ -103,12 +99,22 @@ func (e *emitter) close() {
// Close closes all the goroutine and waits for it to finish.
func (e *emitter) Close() {
e.close()
e.Wait()
for _, p := range e.plugins.Inputs {
if cp, ok := p.(io.Closer); ok {
cp.Close()
}
}
for _, p := range e.plugins.Outputs {
if cp, ok := p.(io.Closer); ok {
cp.Close()
}
}
e.plugins = nil // avoid further accidental usage
}
// CopyMulty copies from 1 reader to multiple writers
func CopyMulty(stop chan int, src io.Reader, writers ...io.Writer) error {
buf := make([]byte, Settings.copyBufferSize)
buf := make([]byte, Settings.CopyBufferSize)
wIndex := 0
modifier := NewHTTPModifier(&Settings.ModifierConfig)
filteredRequests := make(map[string]time.Time)
@@ -124,10 +130,6 @@ func CopyMulty(stop chan int, src io.Reader, writers ...io.Writer) error {
return nil
default:
}
if err == io.EOF || err == ErrorStopped {
return nil
}
if err != nil {
return err
}
@@ -136,24 +138,16 @@ func CopyMulty(stop chan int, src io.Reader, writers ...io.Writer) error {
if nr > 500 {
_maxN = 500
}
if nr > 0 && len(buf) > nr {
if nr > 0 {
payload := buf[:nr]
meta := payloadMeta(payload)
if len(meta) < 3 {
if Settings.Debug {
Debug("[EMITTER] Found malformed record", string(payload[0:_maxN]), nr, "from:", src)
}
Debug(2, "[EMITTER] Found malformed record", string(payload[0:_maxN]), nr, "from:", src)
continue
}
requestID := string(meta[1])
if nr >= 5*1024*1024 {
log.Println("INFO: Large packet... We received ", len(payload), " bytes from ", src)
}
if Settings.Debug {
Debug("[EMITTER] input:", string(payload[0:_maxN]), nr, "from:", src)
}
Debug(3, "[EMITTER] input:", string(payload[0:_maxN]), nr, "from:", src)
if modifier != nil {
if isRequestPayload(payload) {
@@ -172,9 +166,8 @@ func CopyMulty(stop chan int, src io.Reader, writers ...io.Writer) error {
payload = append(payload[:headSize], body...)
}
if Settings.Debug {
Debug("[EMITTER] Rewritten input:", len(payload), "First 500 bytes:", string(payload[0:_maxN]))
}
Debug(3, "[EMITTER] Rewritten input:", len(payload), "First %d bytes:", _maxN, string(payload[0:_maxN]))
} else {
if _, ok := filteredRequests[requestID]; ok {
delete(filteredRequests, requestID)
@@ -198,7 +191,7 @@ func CopyMulty(stop chan int, src io.Reader, writers ...io.Writer) error {
hasher := fnv.New32a()
// First 20 bytes contain tcp session
id := payloadID(payload)
hasher.Write(id[:20])
hasher.Write(id)
wIndex = int(hasher.Sum32()) % len(writers)
writers[wIndex].Write(payload)
@@ -221,8 +214,6 @@ func CopyMulty(stop chan int, src io.Reader, writers ...io.Writer) error {
}
}
}
} else if nr > 0 {
log.Println("WARN: Packet", nr, "bytes is too large to process. Consider increasing --copy-buffer-size")
}
// Run GC on each 1000 request