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

Optimise packet capture thread

Browse Source 
Ensure that thread which capture packets as fast as possible.
Packet parsing logic moved to different threads.
Additionally using os.LockOsThread to reduce CPU context switching
This commit is contained in:
Leonid Bugaev
2021-07-26 21:01:59 +03:00
parent 79ff8823a3
commit 214edb45f6
6 changed files with 212 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
capture/capture.go
+59 -26
View File
@@ -15,6 +15,7 @@ import (
"syscall"
"time"
"github.com/buger/goreplay/proto"
"github.com/buger/goreplay/size"
"github.com/buger/goreplay/tcp"
@@ -59,9 +60,13 @@ type Listener struct {
loopIndex int
Reading chan bool // this channel is closed when the listener has started reading packets
PcapOptions
Engine EngineType
ports []uint16 // src or/and dst ports
trackResponse bool
Engine EngineType
ports []uint16 // src or/and dst ports
trackResponse bool
expiry time.Duration
allowIncomplete bool
messages chan *tcp.Message
protocol tcp.TCPProtocol
host string // pcap file name or interface (name, hardware addr, index or ip address)
@@ -121,7 +126,7 @@ func (eng *EngineType) String() (e string) {
// NewListener creates and initialize a new Listener. if transport or/and engine are invalid/unsupported
// is "tcp" and "pcap", are assumed. l.Engine and l.Transport can help to get the values used.
// if there is an error it will be associated with getting network interfaces
func NewListener(host string, ports []uint16, transport string, engine EngineType, trackResponse bool) (l *Listener, err error) {
func NewListener(host string, ports []uint16, transport string, engine EngineType, protocol tcp.TCPProtocol, trackResponse bool, expiry time.Duration, allowIncomplete bool) (l *Listener, err error) {
l = &Listener{}
l.host = host
@@ -139,6 +144,11 @@ func NewListener(host string, ports []uint16, transport string, engine EngineTyp
l.closeDone = make(chan struct{})
l.quit = make(chan struct{})
l.Reading = make(chan bool)
l.expiry = expiry
l.allowIncomplete = allowIncomplete
l.protocol = protocol
l.messages = make(chan *tcp.Message, 10000)
switch engine {
default:
l.Engine = EnginePcap
@@ -171,8 +181,8 @@ func (l *Listener) SetPcapOptions(opts PcapOptions) {
// Listen listens for packets from the handles, and call handler on every packet received
// until the context done signal is sent or there is unrecoverable error on all handles.
// this function must be called after activating pcap handles
func (l *Listener) Listen(ctx context.Context, handler PacketHandler) (err error) {
l.read(handler)
func (l *Listener) Listen(ctx context.Context) (err error) {
l.read()
done := ctx.Done()
select {
case <-done:
@@ -185,11 +195,11 @@ func (l *Listener) Listen(ctx context.Context, handler PacketHandler) (err error
}
// ListenBackground is like listen but can run concurrently and signal error through channel
func (l *Listener) ListenBackground(ctx context.Context, handler PacketHandler) chan error {
func (l *Listener) ListenBackground(ctx context.Context) chan error {
err := make(chan error, 1)
go func() {
defer close(err)
if e := l.Listen(ctx, handler); err != nil {
if e := l.Listen(ctx); err != nil {
err <- e
}
}()
@@ -332,11 +342,34 @@ func (l *Listener) SocketHandle(ifi pcap.Interface) (handle Socket, err error) {
return
}
func (l *Listener) read(handler PacketHandler) {
func http1StartHint(pckt *tcp.Packet) (isRequest, isResponse bool) {
if proto.HasRequestTitle(pckt.Payload) {
return true, false
}
if proto.HasResponseTitle(pckt.Payload) {
return false, true
}
// No request or response detected
return false, false
}
func http1EndHint(m *tcp.Message) bool {
if m.MissingChunk() {
return false
}
return proto.HasFullPayload(m, m.PacketData()...)
}
func (l *Listener) read() {
l.Lock()
defer l.Unlock()
for key, handle := range l.Handles {
go func(key string, hndl packetHandle) {
runtime.LockOSThread()
defer l.closeHandles(key)
linkSize := 14
linkType := int(layers.LinkTypeEthernet)
@@ -351,6 +384,13 @@ func (l *Listener) read(handler PacketHandler) {
}
}
messageParser := tcp.NewMessageParser(l.messages, l.ports, hndl.ips, l.expiry, l.allowIncomplete)
if l.protocol == tcp.ProtocolHTTP {
messageParser.Start = http1StartHint
messageParser.End = http1EndHint
}
timer := time.NewTicker(1 * time.Second)
for {
@@ -371,23 +411,12 @@ func (l *Listener) read(handler PacketHandler) {
ci.Timestamp = time.Now()
}
pckt, err := tcp.ParsePacket(data, linkType, linkSize, &ci, false)
if err == nil {
for _, p := range l.ports {
if pckt.DstPort == p {
for _, ip := range hndl.ips {
if pckt.DstIP.Equal(ip) {
pckt.Direction = tcp.DirIncoming
break
}
}
break
}
}
handler(pckt)
}
messageParser.PacketHandler(&tcp.PcapPacket{
Data: data,
LType: linkType,
LTypeLen: linkSize,
Ci: &ci,
})
continue
}
if enext, ok := err.(pcap.NextError); ok && enext == pcap.NextErrorTimeoutExpired {
@@ -413,6 +442,10 @@ func (l *Listener) read(handler PacketHandler) {
close(l.Reading)
}
func (l *Listener) Messages() chan *tcp.Message {
return l.messages
}
func (l *Listener) closeHandles(key string) {
l.Lock()
defer l.Unlock()
input_raw.go
+4 -65
View File
@@ -16,40 +16,6 @@ import (
"github.com/buger/goreplay/tcp"
)
// TCPProtocol is a number to indicate type of protocol
type TCPProtocol uint8
const (
// ProtocolHTTP ...
ProtocolHTTP TCPProtocol = iota
// ProtocolBinary ...
ProtocolBinary
)
// Set is here so that TCPProtocol can implement flag.Var
func (protocol *TCPProtocol) Set(v string) error {
switch v {
case "", "http":
*protocol = ProtocolHTTP
case "binary":
*protocol = ProtocolBinary
default:
return fmt.Errorf("unsupported protocol %s", v)
}
return nil
}
func (protocol *TCPProtocol) String() string {
switch *protocol {
case ProtocolBinary:
return "binary"
case ProtocolHTTP:
return "http"
default:
return ""
}
}
// RAWInputConfig represents configuration that can be applied on raw input
type RAWInputConfig struct {
capture.PcapOptions
@@ -57,7 +23,7 @@ type RAWInputConfig struct {
CopyBufferSize size.Size `json:"copy-buffer-size"`
Engine capture.EngineType `json:"input-raw-engine"`
TrackResponse bool `json:"input-raw-track-response"`
Protocol TCPProtocol `json:"input-raw-protocol"`
Protocol tcp.TCPProtocol `json:"input-raw-protocol"`
RealIPHeader string `json:"input-raw-realip-header"`
Stats bool `json:"input-raw-stats"`
AllowIncomplete bool `json:"input-raw-allow-incomplete"`
@@ -117,7 +83,7 @@ func (i *RAWInput) PluginRead() (*Message, error) {
select {
case <-i.quit:
return nil, ErrorStopped
case msgTCP = <-i.messageParser.Messages():
case msgTCP = <-i.listener.Messages():
msg.Data = msgTCP.Data()
}
@@ -142,14 +108,13 @@ func (i *RAWInput) PluginRead() (*Message, error) {
stat := msgTCP.Stats
go i.addStats(stat)
}
msgTCP.Finalize()
msgTCP = nil
return &msg, nil
}
func (i *RAWInput) listen(address string) {
var err error
i.listener, err = capture.NewListener(i.host, i.ports, "", i.Engine, i.TrackResponse)
i.listener, err = capture.NewListener(i.host, i.ports, "", i.Engine, i.Protocol, i.TrackResponse, i.Expire, i.AllowIncomplete)
if err != nil {
log.Fatal(err)
}
@@ -158,15 +123,10 @@ func (i *RAWInput) listen(address string) {
if err != nil {
log.Fatal(err)
}
i.messageParser = tcp.NewMessageParser(i.CopyBufferSize, i.Expire, i.AllowIncomplete, Debug)
if i.Protocol == ProtocolHTTP {
i.messageParser.Start = http1StartHint
i.messageParser.End = http1EndHint
}
var ctx context.Context
ctx, i.cancelListener = context.WithCancel(context.Background())
errCh := i.listener.ListenBackground(ctx, i.messageParser.PacketHandler)
errCh := i.listener.ListenBackground(ctx)
<-i.listener.Reading
Debug(1, i)
go func() {
@@ -210,24 +170,3 @@ func (i *RAWInput) addStats(mStats tcp.Stats) {
i.messageStats = append(i.messageStats, mStats)
i.Unlock()
}
func http1StartHint(pckt *tcp.Packet) (isRequest, isResponse bool) {
if proto.HasRequestTitle(pckt.Payload) {
return true, false
}
if proto.HasResponseTitle(pckt.Payload) {
return false, true
}
// No request or response detected
return false, false
}
func http1EndHint(m *tcp.Message) bool {
if m.MissingChunk() {
return false
}
return proto.HasFullPayload(m, m.PacketData()...)
}
input_raw_test.go
+6 -5
View File
@@ -16,6 +16,7 @@ import (
"github.com/buger/goreplay/capture"
"github.com/buger/goreplay/proto"
"github.com/buger/goreplay/tcp"
)
const testRawExpire = time.Millisecond * 200
@@ -43,7 +44,7 @@ func TestRAWInputIPv4(t *testing.T) {
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: 0,
Protocol: ProtocolHTTP,
Protocol: tcp.ProtocolHTTP,
TrackResponse: true,
RealIPHeader: "X-Real-IP",
}
@@ -113,7 +114,7 @@ func TestRAWInputNoKeepAlive(t *testing.T) {
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: testRawExpire,
Protocol: ProtocolHTTP,
Protocol: tcp.ProtocolHTTP,
TrackResponse: true,
}
input := NewRAWInput(":"+port, conf)
@@ -178,7 +179,7 @@ func TestRAWInputIPv6(t *testing.T) {
var respCounter, reqCounter int64
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Protocol: ProtocolHTTP,
Protocol: tcp.ProtocolHTTP,
TrackResponse: true,
}
input := NewRAWInput(originAddr, conf)
@@ -235,7 +236,7 @@ func TestInputRAWChunkedEncoding(t *testing.T) {
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: time.Second,
Protocol: ProtocolHTTP,
Protocol: tcp.ProtocolHTTP,
TrackResponse: true,
AllowIncomplete: true,
}
@@ -315,7 +316,7 @@ func BenchmarkRAWInputWithReplay(b *testing.B) {
conf := RAWInputConfig{
Engine: capture.EnginePcap,
Expire: testRawExpire,
Protocol: ProtocolHTTP,
Protocol: tcp.ProtocolHTTP,
TrackResponse: true,
}
input := NewRAWInput(originAddr, conf)
tcp/tcp_message.go
+94 -64
View File
@@ -3,15 +3,48 @@ package tcp
import (
"encoding/binary"
"encoding/hex"
_ "fmt"
"fmt"
"net"
"reflect"
"sort"
"time"
"unsafe"
"github.com/buger/goreplay/size"
)
// TCPProtocol is a number to indicate type of protocol
type TCPProtocol uint8
const (
// ProtocolHTTP ...
ProtocolHTTP TCPProtocol = iota
// ProtocolBinary ...
ProtocolBinary
)
// Set is here so that TCPProtocol can implement flag.Var
func (protocol *TCPProtocol) Set(v string) error {
switch v {
case "", "http":
*protocol = ProtocolHTTP
case "binary":
*protocol = ProtocolBinary
default:
return fmt.Errorf("unsupported protocol %s", v)
}
return nil
}
func (protocol *TCPProtocol) String() string {
switch *protocol {
case ProtocolBinary:
return "binary"
case ProtocolHTTP:
return "http"
default:
return ""
}
}
// Stats every message carry its own stats object
type Stats struct {
LostData int
@@ -151,16 +184,9 @@ func (m *Message) Sort() {
sort.SliceStable(m.packets, func(i, j int) bool { return m.packets[i].Seq < m.packets[j].Seq })
}
func (m *Message) Finalize() {
}
// Emitter message handler
type Emitter func(*Message)
// Debugger is the debugger function. first params is the indicator of the issue's priority
// the higher the number, the lower the priority. it can be 4 <= level <= 6.
type Debugger func(int, ...interface{})
// HintEnd hints the parser to stop the session, see MessageParser.End
// when set, it will be executed before checking FIN or RST flag
type HintEnd func(*Message) bool
@@ -172,9 +198,7 @@ type HintStart func(*Packet) (IsRequest, IsOutgoing bool)
// MessageParser holds data of all tcp messages in progress(still receiving/sending packets).
// message is identified by its source port and dst port, and last 4bytes of src IP.
type MessageParser struct {
debug Debugger
maxSize size.Size // maximum message size, default 5mb
m map[uint64]*Message
m map[uint64]*Message
messageExpire time.Duration // the maximum time to wait for the final packet, minimum is 100ms
allowIncompete bool
@@ -182,14 +206,15 @@ type MessageParser struct {
Start HintStart
ticker *time.Ticker
messages chan *Message
packets chan *Packet
packets chan *PcapPacket
close chan struct{} // to signal that we are able to close
ports []uint16
ips []net.IP
}
// NewMessageParser returns a new instance of message parser
func NewMessageParser(maxSize size.Size, messageExpire time.Duration, allowIncompete bool, debugger Debugger) (parser *MessageParser) {
func NewMessageParser(messages chan *Message, ports []uint16, ips []net.IP, messageExpire time.Duration, allowIncompete bool) (parser *MessageParser) {
parser = new(MessageParser)
parser.debug = debugger
parser.messageExpire = messageExpire
if parser.messageExpire == 0 {
@@ -197,17 +222,21 @@ func NewMessageParser(maxSize size.Size, messageExpire time.Duration, allowIncom
}
parser.allowIncompete = allowIncompete
parser.maxSize = maxSize
if parser.maxSize < 1 {
parser.maxSize = 5 << 20
}
parser.packets = make(chan *Packet, 10000)
parser.messages = make(chan *Message, 10000)
parser.packets = make(chan *PcapPacket, 10000)
if messages == nil {
messages = make(chan *Message, 1000)
}
parser.messages = messages
parser.m = make(map[uint64]*Message)
parser.ticker = time.NewTicker(time.Millisecond * 100)
parser.close = make(chan struct{}, 1)
parser.ports = ports
parser.ips = ips
go parser.wait()
return parser
}
@@ -215,7 +244,7 @@ func NewMessageParser(maxSize size.Size, messageExpire time.Duration, allowIncom
var packetLen int
// Packet returns packet handler
func (parser *MessageParser) PacketHandler(packet *Packet) {
func (parser *MessageParser) PacketHandler(packet *PcapPacket) {
packetLen++
parser.packets <- packet
}
@@ -227,7 +256,7 @@ func (parser *MessageParser) wait() {
for {
select {
case pckt := <-parser.packets:
parser.processPacket(pckt)
parser.processPacket(parser.parsePacket(pckt))
case now = <-parser.ticker.C:
parser.timer(now)
case <-parser.close:
@@ -240,30 +269,51 @@ func (parser *MessageParser) wait() {
}
}
func (parser *MessageParser) parsePacket(pcapPkt *PcapPacket) *Packet {
pckt, err := ParsePacket(pcapPkt.Data, pcapPkt.LType, pcapPkt.LTypeLen, pcapPkt.Ci, false)
if err != nil {
stats.Add("packet_error", 1)
return nil
}
for _, p := range parser.ports {
if pckt.DstPort == p {
for _, ip := range parser.ips {
if pckt.DstIP.Equal(ip) {
pckt.Direction = DirIncoming
break
}
}
break
}
}
return pckt
}
func (parser *MessageParser) processPacket(pckt *Packet) {
if pckt == nil {
return
}
// Trying to build unique hash, but there is small chance of collision
// No matter if it is request or response, all packets in the same message have same
m, ok := parser.m[pckt.MessageID()]
switch {
case ok:
if m.Direction == DirUnknown {
if in, out := parser.Start(pckt); in || out {
if in {
m.Direction = DirIncoming
} else {
m.Direction = DirOutcoming
}
}
}
parser.addPacket(m, pckt)
return
case pckt.Direction == DirUnknown && parser.Start != nil:
if in, out := parser.Start(pckt); !(in || out) {
// Packet can be received out of order, so give it another chance
if pckt.Retry < 2 && len(pckt.Payload) > 0 {
// Requeue not known packets
pckt.Retry++
select {
case parser.packets <- pckt:
return
default:
}
}
return
} else {
if in, out := parser.Start(pckt); in || out {
if in {
pckt.Direction = DirIncoming
} else {
@@ -281,20 +331,10 @@ func (parser *MessageParser) processPacket(pckt *Packet) {
}
func (parser *MessageParser) addPacket(m *Message, pckt *Packet) bool {
trunc := m.Length + len(pckt.Payload) - int(parser.maxSize)
if trunc > 0 {
m.Truncated = true
stats.Add("message_timeout_count", 1)
pckt.Payload = pckt.Payload[:int(parser.maxSize)-m.Length]
}
if !m.add(pckt) {
return false
}
if trunc > 0 {
return false
}
// If we are using protocol parsing, like HTTP, depend on its parsing func.
// For the binary procols wait for message to expire
if parser.End != nil {
@@ -311,10 +351,6 @@ func (parser *MessageParser) Read() *Message {
return m
}
func (parser *MessageParser) Messages() chan *Message {
return parser.messages
}
func (parser *MessageParser) Emit(m *Message) {
stats.Add("message_count", 1)
@@ -332,6 +368,9 @@ var failMsg int
func (parser *MessageParser) timer(now time.Time) {
packetLen = 0
packetQueueLen.Set(int64(len(parser.packets)))
messageQueueLen.Set(int64(len(parser.m)))
for _, m := range parser.m {
if now.Sub(m.End) > parser.messageExpire {
m.TimedOut = true
@@ -339,19 +378,10 @@ func (parser *MessageParser) timer(now time.Time) {
failMsg++
if parser.End == nil || parser.allowIncompete {
parser.Emit(m)
} else {
// Just remove
delete(parser.m, m.packets[0].MessageID())
m.Finalize()
}
}
}
}
// this function should not block other parser operations
func (parser *MessageParser) Debug(level int, args ...interface{}) {
if parser.debug != nil {
parser.debug(level, args...)
delete(parser.m, m.packets[0].MessageID())
}
}
}
tcp/tcp_packet.go
+14 -8
View File
@@ -30,17 +30,16 @@ func copySlice(to []byte, skip int, from ...[]byte) ([]byte, int) {
}
var stats *expvar.Map
var bufPoolCount *expvar.Int
var releasedCount *expvar.Int
var packetQueueLen, messageQueueLen *expvar.Int
func init() {
bufPoolCount = new(expvar.Int)
releasedCount = new(expvar.Int)
packetQueueLen = new(expvar.Int)
messageQueueLen = new(expvar.Int)
stats = expvar.NewMap("tcp")
stats.Init()
stats.Set("buffer_pool_count", bufPoolCount)
stats.Set("buffer_released", releasedCount)
stats.Set("packet_queue", packetQueueLen)
stats.Set("message_queue", messageQueueLen)
}
type Dir int
@@ -76,10 +75,17 @@ type Packet struct {
gc bool
}
type PcapPacket struct {
Data []byte
LType int
LTypeLen int
Ci *gopacket.CaptureInfo
}
// ParsePacket parse raw packets
func ParsePacket(data []byte, lType, lTypeLen int, cp *gopacket.CaptureInfo, allowEmpty bool) (pckt *Packet, err error) {
func ParsePacket(data []byte, lType, lTypeLen int, ci *gopacket.CaptureInfo, allowEmpty bool) (pckt *Packet, err error) {
pckt = new(Packet)
if err := pckt.parse(data, lType, lTypeLen, cp, allowEmpty); err != nil {
if err := pckt.parse(data, lType, lTypeLen, ci, allowEmpty); err != nil {
return nil, err
}
tcp/tcp_test.go
+35 -56
View File
@@ -79,7 +79,7 @@ func TestRequestResponseMapping(t *testing.T) {
{SrcPort: 80, DstPort: 60000, Ack: 71, Seq: 56, Direction: DirOutcoming, Timestamp: time.Unix(8, 0), Payload: []byte("Content-Length: 0\r\n\r\n")},
}
parser := NewMessageParser(1<<20, time.Second, false, nil)
parser := NewMessageParser(nil, nil, nil, time.Second, false)
parser.Start = func(pckt *Packet) (bool, bool) {
return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
}
@@ -88,7 +88,7 @@ func TestRequestResponseMapping(t *testing.T) {
}
for _, packet := range packets {
parser.PacketHandler(packet)
parser.processPacket(packet)
}
messages := []*Message{}
@@ -109,7 +109,7 @@ func TestRequestResponseMapping(t *testing.T) {
}
func TestMessageParserWithHint(t *testing.T) {
parser := NewMessageParser(1<<20, time.Second, false, nil)
parser := NewMessageParser(nil, nil, nil, time.Second, false)
parser.Start = func(pckt *Packet) (bool, bool) {
return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
}
@@ -131,7 +131,7 @@ func TestMessageParserWithHint(t *testing.T) {
}
for _, p := range packets {
parser.PacketHandler(p)
parser.processPacket(p)
}
messages := []*Message{}
@@ -154,26 +154,28 @@ func TestMessageParserWithHint(t *testing.T) {
}
func TestMessageParserWrongOrder(t *testing.T) {
parser := NewMessageParser(1<<20, time.Second, true, nil)
parser := NewMessageParser(nil, nil, nil, time.Second, false)
parser.Start = func(pckt *Packet) (bool, bool) {
return proto.HasRequestTitle(pckt.Payload), proto.HasResponseTitle(pckt.Payload)
}
parser.End = func(m *Message) bool {
return proto.HasFullPayload(m, m.PacketData()...)
}
packets := GetPackets(true, 1, 30, nil)
packets[6] = GetPackets(false, 4, 1, []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7"))[0]
packets[5] = GetPackets(false, 5, 1, []byte("\r\nMozilla\r\n9\r\nDeveloper\r"))[0]
packets[4] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
// Duplicate with same seq
packets[7] = GetPackets(false, 6, 1, []byte("\n7\r\nNetwork\r\n0\r\n\r\n"))[0]
packets := []*Packet{
// Seq of first response packet match Ack of first request packet
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 66, Direction: DirIncoming, Timestamp: time.Unix(5, 0), Payload: []byte("MozillaDeveloper")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(1, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nTransfer-Encoding: chunked\r\n\r\n7\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 42, Direction: DirOutcoming, Timestamp: time.Unix(3, 0), Payload: []byte("\n7\r\nNetwork\r\n0\r\n\r\n")},
packets[16] = GetPackets(true, 14, 1, []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n"))[0]
packets[15] = GetPackets(true, 15, 1, []byte("MozillaDeveloper"))[0]
packets[14] = GetPackets(true, 16, 1, []byte("Network"))[0]
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 1, Direction: DirIncoming, Timestamp: time.Unix(4, 0), Payload: []byte("POST / HTTP/1.1\r\nContent-Type: text/plain\r\nContent-Length: 23\r\n\r\n")},
{SrcPort: 80, DstPort: 60000, Ack: 1, Seq: 18, Direction: DirOutcoming, Timestamp: time.Unix(2, 0), Payload: []byte("\r\nMozilla\r\n9\r\nDeveloper\r")},
for i := 0; i < 30; i++ {
parser.PacketHandler(packets[i])
{SrcPort: 80, DstPort: 60000, Ack: 89, Seq: 1, Direction: DirOutcoming, Timestamp: time.Unix(7, 0), Payload: []byte("HTTP/1.1 200 OK\r\nContent-Type: text/plain\r\nContent-Length: 0\r\n\r\n")},
{SrcPort: 60000, DstPort: 80, Ack: 60, Seq: 82, Direction: DirIncoming, Timestamp: time.Unix(6, 0), Payload: []byte("Network")},
}
for _, p := range packets {
parser.processPacket(p)
}
m := parser.Read()
@@ -184,6 +186,12 @@ func TestMessageParserWrongOrder(t *testing.T) {
m = parser.Read()
if !bytes.HasSuffix(m.Data(), []byte("Content-Length: 0\r\n\r\n")) {
t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Content-Length: 0\r\n\r\n"))
}
m = parser.Read()
if !bytes.HasSuffix(m.Data(), []byte("Network")) {
t.Errorf("expected to %q to have suffix %q", m.Data(), []byte("Network"))
}
@@ -193,9 +201,9 @@ func TestMessageParserWithoutHint(t *testing.T) {
var data [63 << 10]byte
packets := GetPackets(true, 1, 10, data[:])
p := NewMessageParser(63<<10*10, time.Second, false, nil)
p := NewMessageParser(nil, nil, nil, time.Second, false)
for _, v := range packets {
p.PacketHandler(v)
p.processPacket(v)
}
m := p.Read()
@@ -204,42 +212,13 @@ func TestMessageParserWithoutHint(t *testing.T) {
}
}
func TestMessageMaxSizeReached(t *testing.T) {
var data [63 << 10]byte
packets := GetPackets(true, 1, 2, data[:])
packets = append(packets, GetPackets(false, 1, 1, make([]byte, 63<<10+10))...)
p := NewMessageParser(63<<10+10, time.Millisecond, false, nil)
for _, v := range packets {
p.PacketHandler(v)
}
time.Sleep(10 * time.Millisecond)
m := p.Read()
if m.Length != 63<<10+10 {
t.Errorf("expected %d to equal %d", m.Length, 63<<10+10)
}
if !m.Truncated {
t.Error("expected message to be truncated")
}
m = p.Read()
if m.Length != 63<<10+10 {
t.Errorf("expected %d to equal %d", m.Length, 63<<10+10)
}
if m.Truncated {
t.Error("expected message to not be truncated")
}
}
func TestMessageTimeoutReached(t *testing.T) {
var data [63 << 10]byte
packets := GetPackets(true, 1, 2, data[:])
p := NewMessageParser(1<<20, 10*time.Millisecond, true, nil)
p.PacketHandler(packets[0])
p := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true)
p.processPacket(packets[0])
time.Sleep(time.Millisecond * 100)
p.PacketHandler(packets[1])
p.processPacket(packets[1])
m := p.Read()
if m.Length != 63<<10 {
t.Errorf("expected %d to equal %d", m.Length, 63<<10)
@@ -253,9 +232,9 @@ func BenchmarkMessageUUID(b *testing.B) {
packets := GetPackets(true, 1, 5, nil)
var uuid []byte
parser := NewMessageParser(0, 0, false, nil)
parser := NewMessageParser(nil, nil, nil, 10*time.Millisecond, true)
for _, p := range packets {
parser.PacketHandler(p)
parser.processPacket(p)
}
msg := parser.Read()
@@ -282,12 +261,12 @@ func BenchmarkPacketParseAndSort(b *testing.B) {
func BenchmarkMessageParserWithoutHint(b *testing.B) {
var chunk = []byte("111111111111111111111111111111")
packets := GetPackets(true, 1, 1000, chunk)
p := NewMessageParser(1<<20, time.Second*2, false, nil)
p := NewMessageParser(nil, nil, nil, 2*time.Second, false)
b.ResetTimer()
b.ReportMetric(float64(1000), "packets/op")
for i := 0; i < b.N; i++ {
for _, v := range packets {
p.PacketHandler(v)
p.processPacket(v)
}
p.Read()
}
@@ -306,7 +285,7 @@ func BenchmarkMessageParserWithHint(b *testing.B) {
packets[i] = GetPackets(false, 1, 1, buf[i])[0]
}
parser := NewMessageParser(1<<30, time.Second*10, false, nil)
parser := NewMessageParser(nil, nil, nil, 2*time.Second, false)
parser.Start = func(pckt *Packet) (bool, bool) {
return false, proto.HasResponseTitle(pckt.Payload)
}
@@ -318,7 +297,7 @@ func BenchmarkMessageParserWithHint(b *testing.B) {
b.ReportMetric(float64(1000), "chunks/op")
for i := 0; i < b.N; i++ {
for j := range packets {
parser.PacketHandler(packets[j])
parser.processPacket(packets[j])
}
parser.Read()
}