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 duplicate packets raw_sockets

Browse Source
This commit is contained in:
Urban Ishimwe
2020-08-24 18:36:22 +02:00
parent dcef7a6eb7
commit 83174c0fd4
3 changed files with 49 additions and 95 deletions
Download Patch File Download Diff File Expand all files Collapse all files
capture/capture.go
+7 -2
View File
@@ -44,6 +44,7 @@ type Listener struct {
Activate func() error // function is used to activate the engine. it must be called before reading packets
Handles map[string]gopacket.PacketDataSource
Interfaces []NetInterface
loopIndex int
Reading chan bool // this channel is closed when the listener has started reading packets
PcapOptions
Engine EngineType
@@ -73,7 +74,7 @@ func (eng *EngineType) Set(v string) error {
*eng = EnginePcap
case "pcap_file":
*eng = EnginePcapFile
case "sock_raw", "af_packet":
case "raw_socket", "af_packet":
*eng = EngineRawSocket
default:
return fmt.Errorf("invalid engine %s", v)
@@ -88,7 +89,7 @@ func (eng *EngineType) String() (e string) {
case EnginePcap:
e = "libpcap"
case EngineRawSocket:
e = "sock_raw"
e = "raw_socket"
default:
e = ""
}
@@ -308,6 +309,7 @@ func (l *Listener) SocketHandle(ifi NetInterface) (handle *SockRaw, err error) {
handle.Close()
return nil, fmt.Errorf("BPF filter error: %q%s, interface: %q", err, l.BPFFilter, ifi.Name)
}
handle.SetLoopbackIndex(int32(l.loopIndex))
return
}
@@ -426,6 +428,9 @@ func (l *Listener) setInterfaces() (err error) {
}
for i := 0; i < len(ifis); i++ {
if ifis[i].Flags&net.FlagLoopback != 0 {
l.loopIndex = ifis[i].Index
}
if ifis[i].Flags&net.FlagUp == 0 {
continue
}
capture/capture_test.go
+16 -85
View File
@@ -11,6 +11,7 @@ import (
"github.com/google/gopacket"
"github.com/google/gopacket/layers"
"github.com/google/gopacket/pcap"
)
var LoopBack = func() net.Interface {
@@ -164,21 +165,7 @@ func TestPcapHandler(t *testing.T) {
t.Errorf("expected error to be nil, got %v", err)
return
}
quit := make(chan bool, 1)
pckts := 0
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
errCh := l.ListenBackground(ctx, func(packet gopacket.Packet) {
pckts++
if pckts == 10 {
quit <- true
}
})
select {
case err = <-errCh:
t.Error(err)
case <-l.Reading:
}
defer l.Handles[LoopBack.Name].(*pcap.Handle).Close()
if err != nil {
t.Errorf("expected error to be nil, got %v", err)
return
@@ -186,10 +173,9 @@ func TestPcapHandler(t *testing.T) {
for i := 0; i < 5; i++ {
_, _ = net.Dial("tcp", "127.0.0.1:8000")
}
select {
case <-time.After(time.Second * 2):
t.Error("failed to parse packets in time")
case <-quit:
sts, _ := l.Handles[LoopBack.Name].(*pcap.Handle).Stats()
if sts.PacketsReceived < 5 {
t.Errorf("expected >=5 packets got %d", sts.PacketsReceived)
}
}
@@ -204,21 +190,7 @@ func TestSocketHandler(t *testing.T) {
t.Errorf("expected error to be nil, got %v", err)
return
}
quit := make(chan bool, 1)
pckts := 0
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
errCh := l.ListenBackground(ctx, func(packet gopacket.Packet) {
pckts++
if pckts == 10 {
quit <- true
}
})
select {
case err = <-errCh:
t.Error(err)
case <-l.Reading:
}
defer l.Handles[LoopBack.Name].(*SockRaw).Close()
if err != nil {
t.Errorf("expected error to be nil, got %v", err)
return
@@ -226,10 +198,9 @@ func TestSocketHandler(t *testing.T) {
for i := 0; i < 5; i++ {
_, _ = net.Dial("tcp", "127.0.0.1:8000")
}
select {
case <-time.After(time.Second * 2):
t.Error("failed to parse packets in time")
case <-quit:
sts, _ := l.Handles[LoopBack.Name].(*SockRaw).Stats()
if sts.Packets < 5 {
t.Errorf("expected >=5 packets got %d", sts.Packets)
}
}
@@ -305,36 +276,19 @@ func BenchmarkPcap(b *testing.B) {
b.Errorf("expected error to be nil, got %v", err)
return
}
quit := make(chan bool, 1)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
pckts := 0
errCh := l.ListenBackground(ctx, func(_ gopacket.Packet) {
pckts++
if pckts == b.N*2 {
quit <- true
}
})
select {
case err = <-errCh:
b.Error(err)
case <-l.Reading:
}
defer l.Handles[LoopBack.Name].(*pcap.Handle).Close()
for i := 0; i < b.N; i++ {
_, _ = net.Dial("tcp", "127.0.0.1:8000")
}
select {
case <-time.After(time.Second):
case <-quit:
}
b.Logf("%d/%d packets in %s", pckts, b.N*2, time.Since(now))
sts, _ := l.Handles[LoopBack.Name].(*pcap.Handle).Stats()
b.Logf("%d packets in %s", sts.PacketsReceived, time.Since(now))
}
func BenchmarkRawSocket(b *testing.B) {
now := time.Now()
var err error
l, err := NewListener(LoopBack.Name, 8000, "", EngineRawSocket, true)
l, err := NewListener(LoopBack.Name, 0, "", EngineRawSocket, true)
if err != nil {
b.Errorf("expected error to be nil, got %v", err)
return
@@ -344,33 +298,10 @@ func BenchmarkRawSocket(b *testing.B) {
b.Errorf("expected error to be nil, got %v", err)
return
}
sock := l.Handles[LoopBack.Name].(*SockRaw)
quit := make(chan bool, 1)
ctx, cancel := context.WithCancel(context.Background())
defer cancel()
pckts := 0
errCh := l.ListenBackground(ctx, func(_ gopacket.Packet) {
pckts++
if pckts == b.N*2 {
quit <- true
}
})
select {
case err = <-errCh:
b.Error(err)
case <-l.Reading:
}
defer l.Handles[LoopBack.Name].(*SockRaw).Close()
for i := 0; i < b.N; i++ {
buf := generateHeaders(1, 1<<10)
err = sock.WritePacketData(buf[:])
if err != nil {
b.Error(err)
}
}
select {
case <-time.After(time.Second):
case <-quit:
_, _ = net.Dial("tcp", "127.0.0.1:8000")
}
sts, _ := l.Handles[LoopBack.Name].(*SockRaw).Stats()
b.Logf("%d/%d packets in %s", sts.Packets-sts.Drops, sts.Packets, time.Since(now))
b.Logf("%d packets in %s", sts.Packets, time.Since(now))
}
capture/sock.go
+26 -8
View File
@@ -37,9 +37,10 @@ type SockRaw struct {
fd int
ifindex int
snaplen int
pollTimeout int
frame int
pollTimeout uintptr
frame uint32 // current frame
buf []byte // points to the memory space of the ring buffer shared with the kernel.
loopIndex int32 // this field must filled to avoid reading packet twice on a loopback device
}
// NewSockRaw returns new M'maped sock_raw on packet version 2.
@@ -53,7 +54,7 @@ func NewSockRaw(ifi net.Interface) (*SockRaw, error) {
fd: fd,
ifindex: ifi.Index,
snaplen: unix.IP_MAXPACKET,
pollTimeout: -1,
pollTimeout: ^uintptr(0),
}
// set packet version
@@ -115,23 +116,32 @@ func (sock *SockRaw) ReadPacketData() (buf []byte, ci gopacket.CaptureInfo, err
Fd: int32(sock.fd),
Events: unix.POLLIN,
}
i := sock.frame * FRAMESIZE
var i int
read:
i = int(sock.frame * FRAMESIZE)
tpHdr = (*unix.Tpacket2Hdr)(unsafe.Pointer(&sock.buf[i]))
for tpHdr.Status&unix.TP_STATUS_USER == 0 {
_, _, e := unix.Syscall(unix.SYS_POLL, uintptr(unsafe.Pointer(poll)), 1, uintptr(sock.pollTimeout))
_, _, e := unix.Syscall(unix.SYS_POLL, uintptr(unsafe.Pointer(poll)), 1, sock.pollTimeout)
if e != 0 && e != unix.EINTR {
return buf, ci, e
}
}
sock.frame = (sock.frame + 1) % FRAMENR
tpHdr.Status = unix.TP_STATUS_KERNEL
sockAddr := (*unix.RawSockaddrLinklayer)(unsafe.Pointer(&sock.buf[i+tpacket2hdrlen]))
// parse out repeating packets on loopback, 4 frames will be wasted obviously!
if sockAddr.Ifindex == sock.loopIndex && sock.frame%2 != 0 {
goto read
}
ci.Length = int(tpHdr.Len)
ci.Timestamp = time.Unix(int64(tpHdr.Sec), int64(tpHdr.Nsec))
ci.InterfaceIndex = int(sockAddr.Ifindex)
buf = make([]byte, tpHdr.Snaplen)
ci.CaptureLength = copy(buf, sock.buf[i+int(tpHdr.Mac):])
sock.frame = (sock.frame + 1) % FRAMENR
return
}
@@ -169,7 +179,7 @@ func (sock *SockRaw) SetSnapLen(snap int) error {
func (sock *SockRaw) SetTimeout(t time.Duration) error {
sock.mu.Lock()
defer sock.mu.Unlock()
sock.pollTimeout = int(t)
sock.pollTimeout = uintptr(t)
return nil
}
@@ -204,7 +214,8 @@ func (sock *SockRaw) SetBPFFilter(expr string) error {
return unix.SetsockoptSockFprog(sock.fd, unix.SOL_SOCKET, unix.SO_ATTACH_FILTER, fprog)
}
// SetPromiscuous sets promiscous mode to the required value. If it is enabled, traffic not destined for the interface will also be captured.
// SetPromiscuous sets promiscous mode to the required value. for better result capture on all interfaces instead.
// If it is enabled, traffic not destined for the interface will also be captured.
func (sock *SockRaw) SetPromiscuous(b bool) error {
sock.mu.Lock()
defer sock.mu.Unlock()
@@ -228,6 +239,13 @@ func (sock *SockRaw) Stats() (*unix.TpacketStats, error) {
return unix.GetsockoptTpacketStats(sock.fd, unix.SOL_PACKET, unix.PACKET_STATISTICS)
}
// SetLoopbackIndex necessary to avoid reading packet twice on a loopback device
func (sock *SockRaw) SetLoopbackIndex(i int32) {
sock.mu.Lock()
defer sock.mu.Unlock()
sock.loopIndex = i
}
// WritePacketData transmits a raw packet.
func (sock *SockRaw) WritePacketData(pkt []byte) error {
_, err := unix.Write(sock.fd, pkt)