diff --git a/.github/workflows/ci-pr.yml b/.github/workflows/ci-pr.yml new file mode 100644 index 0000000..ade532d --- /dev/null +++ b/.github/workflows/ci-pr.yml @@ -0,0 +1,51 @@ +name: CI Build / Test (Pull Request) + +on: + pull_request: + types: [opened, synchronize] + +jobs: + + bench: + name: Comparative Benchmark + runs-on: [self-hosted, linux, benchmark-only] + steps: + - name: Set up Go + uses: actions/setup-go@v4 + with: + go-version: ^1.21 + id: go + + - name: Check out code into the Go module directory + uses: actions/checkout@v4 + with: + ref: ${{ github.event.pull_request.head.ref }} + fetch-depth: 0 + show-progress: false + + - name: Run Comparative Benchmark + id: bench + env: + BENCH_TORUN: . + BENCH_COUNT: 15 + BENCH_TIME: 5s + run: | + printenv PATH > .path + /home/app/bench.sh + + - name: Add / Update Benchmark Result Comment + continue-on-error: true + uses: marocchino/sticky-pull-request-comment@v2 + with: + GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }} + header: bench + hide_and_recreate: true + hide_classify: "OUTDATED" + message: | + ### Benchmark Result +
Benchmark diff with base branch + + ``` + ${{ steps.bench.outputs.diff }} + ``` +
diff --git a/.github/workflows/go.yml b/.github/workflows/ci-push.yml similarity index 78% rename from .github/workflows/go.yml rename to .github/workflows/ci-push.yml index 9c427ba..9080676 100644 --- a/.github/workflows/go.yml +++ b/.github/workflows/ci-push.yml @@ -1,26 +1,22 @@ -# This workflow will build a golang project -# For more information see: https://docs.github.com/en/actions/automating-builds-and-tests/building-and-testing-go - -name: Go +name: CI Build / Test (Push) on: - push - - pull_request jobs: build-linux: - name: Build on Linux + name: Build / Test on Linux runs-on: ubuntu-latest steps: - name: Set up Go - uses: actions/setup-go@v3 + uses: actions/setup-go@v4 with: go-version: ^1.21 id: go - name: Check out code into the Go module directory - uses: actions/checkout@v3 + uses: actions/checkout@v4 - name: Build for 386 run: GOOS=linux GOARCH=386 go build -v -x ./... diff --git a/capture/afpacket/afpacket/afpacket_mock.go b/capture/afpacket/afpacket/afpacket_mock.go index 338565c..cdafc38 100644 --- a/capture/afpacket/afpacket/afpacket_mock.go +++ b/capture/afpacket/afpacket/afpacket_mock.go @@ -6,12 +6,14 @@ package afpacket import ( "errors" "io" + "runtime" "time" "github.com/fako1024/slimcap/capture" "github.com/fako1024/slimcap/capture/afpacket/socket" "github.com/fako1024/slimcap/event" "github.com/fako1024/slimcap/link" + "golang.org/x/sys/unix" ) const ( @@ -28,8 +30,10 @@ const ( type MockSource struct { *Source - mockPackets chan capture.Packet - MockFd *socket.MockFileDescriptor + mockPackets chan capture.Packet + MockFd *socket.MockFileDescriptor + + cpuSet *unix.CPUSet packetAddCallbackFn func(payload []byte, totalLen uint32, pktType, ipLayerOffset byte) } @@ -40,6 +44,12 @@ func (m *MockSource) PacketAddCallbackFn(fn func(payload []byte, totalLen uint32 return m } +// CPUSet defines an explicit set of CPU cores to run / pin any background tasks on / to +func (m *MockSource) CPUSet(cpuSet *unix.CPUSet) *MockSource { + m.cpuSet = cpuSet + return m +} + // AddPacket adds a new mock packet to the source // This can happen prior to calling run or continuously while consuming data func (m *MockSource) AddPacket(pkt capture.Packet) error { @@ -107,6 +117,17 @@ func (m *MockSource) Pipe(src capture.Source, doneReadingChan chan struct{}) cha errChan := make(chan error) go func(errs chan error, done chan struct{}) { + + // Minimize scheduler overhead by locking this goroutine to the current thread. + // In addition, pin the task to the CPU set (if provided) + runtime.LockOSThread() + if m.cpuSet != nil { + if err := unix.SchedSetaffinity(0, m.cpuSet); err != nil { + errs <- err + return + } + } + for { if err := m.AddPacketFromSource(src); err != nil { if errors.Is(err, io.EOF) || errors.Is(err, capture.ErrCaptureStopped) { @@ -145,6 +166,16 @@ func (m *MockSource) RunNoDrain() chan error { errChan := make(chan error) go func(errs chan error) { + // Minimize scheduler overhead by locking this goroutine to the current thread. + // In addition, pin the task to the CPU set (if provided) + runtime.LockOSThread() + if m.cpuSet != nil { + if err := unix.SchedSetaffinity(0, m.cpuSet); err != nil { + errs <- err + return + } + } + // Populate a slice with all packets from the channel for repeated consumption packets := make([]capture.Packet, 0, len(m.mockPackets)) for i := 0; i < len(m.mockPackets); i++ { @@ -190,9 +221,19 @@ func (m *MockSource) Close() error { ////////////////////////////////////////////////////////////////////////////////////////////////////// -func (m *MockSource) run(errChan chan error) { +func (m *MockSource) run(errs chan error) { + + defer close(errs) - defer close(errChan) + // Minimize scheduler overhead by locking this goroutine to the current thread. + // In addition, pin the task to the CPU set (if provided) + runtime.LockOSThread() + if m.cpuSet != nil { + if err := unix.SchedSetaffinity(0, m.cpuSet); err != nil { + errs <- err + return + } + } for pkt := range m.mockPackets { @@ -200,10 +241,10 @@ func (m *MockSource) run(errChan chan error) { // Queue / trigger an event equivalent to receiving a new packet via the PPOLL syscall if err := event.ToMockHandler(m.eventHandler).SignalAvailableData(); err != nil { - errChan <- err + errs <- err return } } - errChan <- nil + errs <- nil } diff --git a/capture/afpacket/afpacket/afpacket_mock_test.go b/capture/afpacket/afpacket/afpacket_mock_test.go index e64301b..c809394 100644 --- a/capture/afpacket/afpacket/afpacket_mock_test.go +++ b/capture/afpacket/afpacket/afpacket_mock_test.go @@ -6,6 +6,7 @@ package afpacket import ( "fmt" "net" + "runtime" "sync" "testing" "time" @@ -13,6 +14,7 @@ import ( "github.com/fako1024/slimcap/capture" "github.com/fako1024/slimcap/link" "github.com/stretchr/testify/require" + "golang.org/x/sys/unix" ) func TestOptions(t *testing.T) { @@ -228,6 +230,12 @@ func TestPipe(t *testing.T) { func BenchmarkCaptureMethods(b *testing.B) { + // TODO: Add some logic / checking for isolated cores (and use those) and + // validation that there are sufficient cores + var cpuMaskFG, cpuMaskBG unix.CPUSet + cpuMaskFG.Set(1) + cpuMaskBG.Set(2) + testPacket, err := capture.BuildPacket( net.ParseIP("1.2.3.4"), net.ParseIP("4.5.6.7"), @@ -242,6 +250,7 @@ func BenchmarkCaptureMethods(b *testing.B) { Promiscuous(false), ) require.Nil(b, err) + mockSrc.CPUSet(&cpuMaskBG) for mockSrc.CanAddPackets() { require.Nil(b, mockSrc.AddPacket(testPacket)) @@ -249,7 +258,9 @@ func BenchmarkCaptureMethods(b *testing.B) { mockSrc.RunNoDrain() b.Run("NextPacket", func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { p, _ := mockSrc.NextPacket(nil) _ = p @@ -258,6 +269,7 @@ func BenchmarkCaptureMethods(b *testing.B) { b.Run("NextPacketInPlace", func(b *testing.B) { var p capture.Packet = mockSrc.NewPacket() + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -279,6 +291,7 @@ func BenchmarkCaptureMethods(b *testing.B) { b.Run("NextIPPacketInPlace", func(b *testing.B) { pkt := mockSrc.NewPacket() var p capture.IPLayer = pkt.IPLayer() + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -290,7 +303,9 @@ func BenchmarkCaptureMethods(b *testing.B) { }) b.Run("NextPacketFn", func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { _ = mockSrc.NextPacketFn(func(payload []byte, totalLen uint32, pktType, ipLayerOffset byte) error { _ = payload @@ -350,3 +365,19 @@ func testCaptureMethods(t *testing.T, fn func(t *testing.T, src *MockSource, i, // Close the mock source require.Nil(t, mockSrc.Close()) } + +func pinToCPU(cpuSet *unix.CPUSet) error { + + // If no CPU set is provided, do nothing + if cpuSet == nil { + return nil + } + + // Set affinity and lock thread + if err := unix.SchedSetaffinity(0, cpuSet); err != nil { + return err + } + runtime.LockOSThread() + + return nil +} diff --git a/capture/afpacket/afring/afring_mock.go b/capture/afpacket/afring/afring_mock.go index 49e3e2b..79aae1a 100644 --- a/capture/afpacket/afring/afring_mock.go +++ b/capture/afpacket/afring/afring_mock.go @@ -6,6 +6,7 @@ package afring import ( "errors" "io" + "runtime" "sync/atomic" "time" "unsafe" @@ -49,6 +50,7 @@ type MockSource struct { MockFd *socket.MockFileDescriptor + cpuSet *unix.CPUSet packetAddCallbackFn func(payload []byte, totalLen uint32, pktType, ipLayerOffset byte) } @@ -98,6 +100,12 @@ func (m *MockSource) PacketAddCallbackFn(fn func(payload []byte, totalLen uint32 return m } +// CPUSet defines an explicit set of CPU cores to run / pin any background tasks on / to +func (m *MockSource) CPUSet(cpuSet *unix.CPUSet) *MockSource { + m.cpuSet = cpuSet + return m +} + // AddPacket adds a new mock packet to the source // This can happen prior to calling run or continuously while consuming data, mimicking the // function of an actual ring buffer. Consequently, if the ring buffer is full and elements not @@ -196,6 +204,17 @@ func (m *MockSource) Pipe(src capture.Source, doneReadingChan chan struct{}) (er errChan = make(chan error, 1) go func(errs chan error, done chan struct{}) { + + // Minimize scheduler overhead by locking this goroutine to the current thread. + // In addition, pin the task to the CPU set (if provided) + runtime.LockOSThread() + if m.cpuSet != nil { + if err := unix.SchedSetaffinity(0, m.cpuSet); err != nil { + errs <- err + return + } + } + for { if err := m.AddPacketFromSource(src); err != nil { if errors.Is(err, io.EOF) || errors.Is(err, capture.ErrCaptureStopped) { @@ -222,10 +241,10 @@ func (m *MockSource) Pipe(src capture.Source, doneReadingChan chan struct{}) (er // Run executes processing of packets in the background, mimicking the function of an actual kernel // packet ring buffer func (m *MockSource) Run() <-chan error { - errChan := make(chan error) - go m.run(errChan) + errs := make(chan error) + go m.run(errs) - return errChan + return errs } // Done notifies the mock source that no more mock packets will be added, causing the ring buffer @@ -248,8 +267,18 @@ func (m *MockSource) ForceBlockskUnavailable() { ////////////////////////////////////////////////////////////////////////////////////////////////////// -func (m *MockSource) run(errChan chan<- error) { - defer close(errChan) +func (m *MockSource) run(errs chan<- error) { + defer close(errs) + + // Minimize scheduler overhead by locking this goroutine to the current thread. + // In addition, pin the task to the CPU set (if provided) + runtime.LockOSThread() + if m.cpuSet != nil { + if err := unix.SchedSetaffinity(0, m.cpuSet); err != nil { + errs <- err + return + } + } for block := range m.mockBlocks { @@ -263,12 +292,12 @@ func (m *MockSource) run(errChan chan<- error) { // Queue / trigger an event equivalent to receiving a new block via the PPOLL syscall if err := event.ToMockHandler(m.eventHandler).SignalAvailableData(); err != nil { - errChan <- err + errs <- err return } } - errChan <- nil + errs <- nil } func (m *MockSource) getBlockStatus(n int) (status uint32) { diff --git a/capture/afpacket/afring/afring_mock_nodrain.go b/capture/afpacket/afring/afring_mock_nodrain.go index ae9007f..070dcbb 100644 --- a/capture/afpacket/afring/afring_mock_nodrain.go +++ b/capture/afpacket/afring/afring_mock_nodrain.go @@ -69,13 +69,21 @@ func (m *MockSourceNoDrain) Run(releaseInterval time.Duration) (<-chan error, er m.wgRunning.Add(1) go func(errs chan error) { - // Minimize scheduler overhead by locking this goroutine to the current thread - runtime.LockOSThread() defer func() { close(errs) m.wgRunning.Done() }() + // Minimize scheduler overhead by locking this goroutine to the current thread. + // In addition, pin the task to the CPU set (if provided) + runtime.LockOSThread() + if m.cpuSet != nil { + if err := unix.SchedSetaffinity(0, m.cpuSet); err != nil { + errs <- err + return + } + } + // Queue / trigger a single event equivalent to receiving a new block via the PPOLL syscall and // instruct the mock socket to not release the semaphore. That way data can be consumed immediately // at all times diff --git a/capture/afpacket/afring/afring_mock_test.go b/capture/afpacket/afring/afring_mock_test.go index b06aa15..4195d98 100644 --- a/capture/afpacket/afring/afring_mock_test.go +++ b/capture/afpacket/afring/afring_mock_test.go @@ -6,6 +6,8 @@ package afring import ( "fmt" "net" + "runtime" + "runtime/debug" "sync" "testing" "time" @@ -13,6 +15,7 @@ import ( "github.com/fako1024/slimcap/capture" "github.com/fako1024/slimcap/link" "github.com/stretchr/testify/require" + "golang.org/x/sys/unix" ) func TestOptions(t *testing.T) { @@ -377,6 +380,19 @@ func TestPipe(t *testing.T) { func BenchmarkCaptureMethods(b *testing.B) { + // TODO: Add some logic / checking for isolated cores (and use those) and + // validation that there are sufficient cores + var cpuMaskFG, cpuMaskBG unix.CPUSet + cpuMaskFG.Set(1) + cpuMaskBG.Set(2) + + oldGCPercent := debug.SetGCPercent(-1) + oldMemLimit := debug.SetMemoryLimit(512 * 1024 * 1024) + defer func() { + debug.SetGCPercent(oldGCPercent) + debug.SetMemoryLimit(oldMemLimit) + }() + testPacket, err := capture.BuildPacket( net.ParseIP("1.2.3.4"), net.ParseIP("4.5.6.7"), @@ -401,6 +417,7 @@ func BenchmarkCaptureMethods(b *testing.B) { Promiscuous(false), ) require.Nil(b, err) + mockSrc.CPUSet(&cpuMaskBG) for mockSrc.CanAddPackets() { require.Nil(b, mockSrc.AddPacket(testPacket)) @@ -409,7 +426,9 @@ func BenchmarkCaptureMethods(b *testing.B) { require.Nil(b, err) b.Run(fmt.Sprintf("NextPacket_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { p, _ := mockSrc.NextPacket(nil) _ = p @@ -418,6 +437,7 @@ func BenchmarkCaptureMethods(b *testing.B) { b.Run(fmt.Sprintf("NextPacketInPlace_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { var p capture.Packet = mockSrc.NewPacket() + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -427,7 +447,9 @@ func BenchmarkCaptureMethods(b *testing.B) { }) b.Run(fmt.Sprintf("NextPayload_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { p, pktType, totalLen, _ := mockSrc.NextPayload(nil) _ = p @@ -439,6 +461,7 @@ func BenchmarkCaptureMethods(b *testing.B) { b.Run(fmt.Sprintf("NextPayloadInPlace_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { pkt := mockSrc.NewPacket() var p []byte = pkt.Payload() + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -450,7 +473,9 @@ func BenchmarkCaptureMethods(b *testing.B) { }) b.Run(fmt.Sprintf("NextPayloadZeroCopy_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { p, pktType, totalLen, _ := mockSrc.NextPayloadZeroCopy() _ = p @@ -460,7 +485,9 @@ func BenchmarkCaptureMethods(b *testing.B) { }) b.Run(fmt.Sprintf("NextIPPacket_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { p, pktType, totalLen, _ := mockSrc.NextIPPacket(nil) _ = p @@ -472,6 +499,7 @@ func BenchmarkCaptureMethods(b *testing.B) { b.Run(fmt.Sprintf("NextIPPacketInPlace_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { pkt := mockSrc.NewPacket() var p capture.IPLayer = pkt.IPLayer() + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() b.ResetTimer() for i := 0; i < b.N; i++ { @@ -483,7 +511,9 @@ func BenchmarkCaptureMethods(b *testing.B) { }) b.Run(fmt.Sprintf("NextIPPacketZeroCopy_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { p, pktType, totalLen, _ := mockSrc.NextIPPacketZeroCopy() _ = p @@ -493,7 +523,9 @@ func BenchmarkCaptureMethods(b *testing.B) { }) b.Run(fmt.Sprintf("NextPacketFn_%dkiBx%d", benchConfig.blockSize/1000, benchConfig.nBlocks), func(b *testing.B) { + require.Nil(b, pinToCPU(&cpuMaskFG)) b.ReportAllocs() + b.ResetTimer() for i := 0; i < b.N; i++ { _ = mockSrc.NextPacketFn(func(payload []byte, totalLen uint32, pktType, ipLayerOffset byte) error { _ = payload @@ -561,3 +593,20 @@ func testCaptureMethods(t *testing.T, fn func(t *testing.T, _ *MockSource, _, _ // Close the mock source require.Nil(t, mockSrc.Close()) } + +// TODO: Move this soemwhere it can be shared (maybe even outside slimcap) +func pinToCPU(cpuSet *unix.CPUSet) error { + + // If no CPU set is provided, do nothing + if cpuSet == nil { + return nil + } + + // Set affinity and lock thread + if err := unix.SchedSetaffinity(0, cpuSet); err != nil { + return err + } + runtime.LockOSThread() + + return nil +}