[connect-ip] further enhance packet splice and add high performance connection.Pipe implementation (#45)

Pipe is good to like 40gbit/s on my crummy old XPS laptop.

Author: dpeckett

go.mod

@@ -222,6 +222,7 @@ require (
 	github.com/hashicorp/mdns v1.0.1 // indirect
 	github.com/hashicorp/memberlist v0.5.0 // indirect
 	github.com/hashicorp/vic v1.5.1-0.20190403131502-bbfe86ec9443 // indirect
+	github.com/hedzr/go-ringbuf/v2 v2.2.1 // indirect
 	github.com/iancoleman/strcase v0.3.0 // indirect
 	github.com/imdario/mergo v0.3.16 // indirect
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect

go.sum

@@ -612,6 +612,8 @@ github.com/hashicorp/memberlist v0.5.0 h1:EtYPN8DpAURiapus508I4n9CzHs2W+8NZGbmmR
 github.com/hashicorp/memberlist v0.5.0/go.mod h1:yvyXLpo0QaGE59Y7hDTsTzDD25JYBZ4mHgHUZ8lrOI0=
 github.com/hashicorp/vic v1.5.1-0.20190403131502-bbfe86ec9443 h1:O/pT5C1Q3mVXMyuqg7yuAWUg/jMZR1/0QTzTRdNR6Uw=
 github.com/hashicorp/vic v1.5.1-0.20190403131502-bbfe86ec9443/go.mod h1:bEpDU35nTu0ey1EXjwNwPjI9xErAsoOCmcMb9GKvyxo=
+github.com/hedzr/go-ringbuf/v2 v2.2.1 h1:bnIRxSCWYt4vs5UCDCOYf+r1C8cQC7tkcOdjOTaVzNk=
+github.com/hedzr/go-ringbuf/v2 v2.2.1/go.mod h1:N3HsRpbHvPkX9GsykpkPoR2vD6WRR6GbU7tx/9GLE4M=
 github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU=
 github.com/iancoleman/strcase v0.3.0 h1:nTXanmYxhfFAMjZL34Ov6gkzEsSJZ5DbhxWjvSASxEI=
 github.com/iancoleman/strcase v0.3.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=

pkg/tunnel/connection/pipe.go

@@ -0,0 +1,129 @@
+package connection
+
+import (
+	"context"
+	"errors"
+	"runtime"
+	"sync"
+
+	"github.com/hedzr/go-ringbuf/v2"
+	"github.com/hedzr/go-ringbuf/v2/mpmc"
+)
+
+var _ Connection = (*Pipe)(nil)
+
+type Pipe struct {
+	readRing  mpmc.RingBuffer[[]byte]
+	writeRing mpmc.RingBuffer[[]byte]
+	ctx       context.Context
+	cancel    context.CancelFunc
+}
+
+var bufPool = sync.Pool{
+	New: func() interface{} {
+		b := make([]byte, 2048) // Adjust size if needed
+		return &b
+	},
+}
+
+// NewPipe creates a pair of connected Pipe instances for bidirectional communication.
+func NewPipe(ctx context.Context) (*Pipe, *Pipe) {
+	ringAtoB := ringbuf.New[[]byte](1024)
+	ringBtoA := ringbuf.New[[]byte](1024)
+
+	ctx, cancel := context.WithCancel(ctx)
+
+	pipeA := &Pipe{
+		readRing:  ringBtoA,
+		writeRing: ringAtoB,
+		ctx:       ctx,
+		cancel:    cancel,
+	}
+	pipeB := &Pipe{
+		readRing:  ringAtoB,
+		writeRing: ringBtoA,
+		ctx:       ctx,
+		cancel:    cancel,
+	}
+
+	return pipeA, pipeB
+}
+
+// ReadPacket reads a packet into the provided buffer.
+func (p *Pipe) ReadPacket(buf []byte) (int, error) {
+	select {
+	case <-p.ctx.Done():
+		return 0, errors.New("pipe closed")
+	default:
+		var item []byte
+		var err error
+		for {
+			item, err = p.readRing.Dequeue()
+			if err != nil {
+				if errors.Is(err, mpmc.ErrQueueEmpty) {
+					runtime.Gosched()
+
+					// Has the context been cancelled?
+					select {
+					case <-p.ctx.Done():
+						bufPool.Put(&item)
+						return 0, errors.New("pipe closed")
+					default:
+						// Continue to try to dequeue
+						continue
+					}
+				}
+				return 0, err
+			}
+			break
+		}
+		n := copy(buf, item)
+		bufPool.Put(&item)
+		return n, nil
+	}
+}
+
+// WritePacket writes a packet from the provided buffer.
+func (p *Pipe) WritePacket(b []byte) ([]byte, error) {
+	select {
+	case <-p.ctx.Done():
+		return nil, errors.New("pipe closed")
+	default:
+		bufPtr := bufPool.Get().(*[]byte)
+		buf := *bufPtr
+		if cap(buf) < len(b) {
+			buf = make([]byte, len(b))
+		}
+		buf = buf[:len(b)]
+		copy(buf, b)
+
+		for {
+			err := p.writeRing.Enqueue(buf)
+			if err != nil {
+				if errors.Is(err, mpmc.ErrQueueFull) {
+					runtime.Gosched()
+
+					// Has the context been cancelled?
+					select {
+					case <-p.ctx.Done():
+						bufPool.Put(&buf)
+						return nil, errors.New("pipe closed")
+					default:
+						// Continue to try to enqueue
+						continue
+					}
+				}
+				return nil, err
+			}
+			break
+		}
+
+		return nil, nil
+	}
+}
+
+// Close terminates the pipe.
+func (p *Pipe) Close() error {
+	p.cancel()
+	return nil
+}

pkg/tunnel/connection/pipe_test.go

@@ -0,0 +1,47 @@
+package connection_test
+
+import (
+	"bytes"
+	"testing"
+	"time"
+
+	"github.com/apoxy-dev/apoxy-cli/pkg/tunnel/connection"
+)
+
+func TestPipeThroughput(t *testing.T) {
+	const (
+		packetSize = 1024      // 1 KB per packet
+		numPackets = 1_000_000 // Total packets to send
+	)
+
+	p1, p2 := connection.NewPipe(t.Context())
+
+	payload := bytes.Repeat([]byte("X"), packetSize)
+	buf := make([]byte, packetSize)
+
+	done := make(chan struct{})
+
+	go func() {
+		for i := 0; i < numPackets; i++ {
+			if _, err := p2.ReadPacket(buf); err != nil {
+				t.Fatal(err)
+			}
+		}
+		close(done)
+	}()
+
+	start := time.Now()
+
+	for i := 0; i < numPackets; i++ {
+		if _, err := p1.WritePacket(payload); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	<-done
+	duration := time.Since(start)
+
+	throughputGbps := (float64(packetSize*numPackets*8) / 1e9) / duration.Seconds()
+	t.Logf("Sent %d packets of %d bytes in %s", numPackets, packetSize, duration)
+	t.Logf("Throughput: %.2f Gbps", throughputGbps)
+}