feat: add ProxyStream with more performant data forwarding

auth.go

@@ -166,7 +166,7 @@ func (a UserPassAuthenticator) Authenticate(reader io.Reader, writer io.Writer)
 // authenticate is used to handle connection authentication
 func (s *Server) authenticate(conn io.Writer, bufConn io.Reader) (*AuthContext, error) {
 	// Get the methods
-	methods, err := readMethods(bufConn)
+	methods, err := ReadMethods(bufConn)
 	if err != nil {
 		return nil, fmt.Errorf("failed to get auth methods: %v", err)
 	}
@@ -190,9 +190,9 @@ func noAcceptableAuth(conn io.Writer) error {
 	return ErrNoSupportedAuth
 }
 
-// readMethods is used to read the number of methods
+// ReadMethods is used to read the number of methods
 // and proceeding auth methods
-func readMethods(r io.Reader) ([]byte, error) {
+func ReadMethods(r io.Reader) ([]byte, error) {
 	header := []byte{0}
 	if _, err := r.Read(header); err != nil {
 		return nil, err

go.mod

@@ -1,3 +1,3 @@
 module github.com/haxii/socks5
 
-go 1.12
+go 1.22

request.go

@@ -2,6 +2,7 @@ package socks5
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net"
@@ -422,12 +423,103 @@ type closeWriter interface {
 	CloseWrite() error
 }
 
-// proxy is used to shuffle data from src to destination, and sends errors
-// down a dedicated channel
+type closeReader interface {
+	CloseRead() error
+}
+
 func proxy(dst io.Writer, src io.Reader, errCh chan error) {
-	_, err := io.Copy(dst, src)
-	if tcpConn, ok := dst.(closeWriter); ok {
-		tcpConn.CloseWrite()
-	}
+	err := ProxyStream(src, dst)
 	errCh <- err
 }
+
+// ProxyStream forwards data from src to dst, similar to io.Copy, but with improved performance.
+// Unlike io.Copy’s sequential read/write model, it allows reads to continue while writes are in progress,
+// avoiding read-side underutilization when writes are slow.
+// ProxyStream closes both the read and write sides when the transfer completes.
+func ProxyStream(src io.Reader, dst io.Writer) error {
+	const (
+		bufSize    = 64 << 10 // 64 KiB
+		numBuffers = 16
+	)
+
+	type bufSlot struct {
+		buf []byte
+		n   int
+	}
+
+	bufPool := make(chan *bufSlot, numBuffers)
+	for i := 0; i < numBuffers; i++ {
+		bufPool <- &bufSlot{buf: make([]byte, bufSize)}
+	}
+
+	writeDataCh := make(chan *bufSlot, numBuffers)
+	// Start small, it will grow if necessary to a maximum of bufSize * bufSize
+	writeBuf := make([]byte, 0, bufSize)
+
+	// Writer goroutine
+	writeErrCh := make(chan error, 1)
+	go func() {
+		var writeErr error
+		for buf := range writeDataCh {
+			writeBuf = append(writeBuf, buf.buf[:buf.n]...)
+			bufPool <- buf
+
+		batchLoop:
+			for i := 0; i < numBuffers; i++ {
+				select {
+				case buf2, ok := <-writeDataCh:
+					if !ok {
+						break batchLoop
+					}
+					writeBuf = append(writeBuf, buf2.buf[:buf2.n]...)
+					bufPool <- buf2
+				default:
+					break batchLoop
+				}
+			}
+			_, writeErr = dst.Write(writeBuf)
+			writeBuf = writeBuf[:0]
+
+			if writeErr != nil {
+				break
+			}
+		}
+		if writeErr != nil {
+			writeErr = fmt.Errorf("write error: %v", writeErr)
+		}
+		writeErrCh <- writeErr
+
+		// Close read side to unblock reader loop
+		if conn, ok := src.(closeReader); ok {
+			_ = conn.CloseRead()
+		}
+	}()
+
+	// Reader loop
+	var readErr error
+	for readErr == nil {
+		buf := <-bufPool
+		buf.n, readErr = src.Read(buf.buf)
+		if buf.n > 0 {
+			writeDataCh <- buf
+		} else {
+			bufPool <- buf
+		}
+	}
+
+	if errors.Is(readErr, io.EOF) {
+		readErr = nil
+	} else {
+		readErr = fmt.Errorf("read error: %v", readErr)
+	}
+
+	close(writeDataCh)
+	writeErr := <-writeErrCh
+
+	// Close write side
+	if conn, ok := dst.(closeWriter); ok {
+		_ = conn.CloseWrite()
+	}
+
+	return errors.Join(writeErr, readErr)
+}

socks5.go

@@ -188,3 +188,32 @@ func (s *Server) ServeConn(conn net.Conn) error {
 
 	return nil
 }
+
+// ServeConnNoAuth is used to serve a single connection where the auth
+// negotiation has already been handled by the caller (e.g. locally on the client side).
+// It skips the version byte read and authentication, going directly to request handling.
+func (s *Server) ServeConnNoAuth(conn net.Conn) error {
+	defer conn.Close()
+
+	request, err := NewRequest(conn)
+	if err != nil {
+		if err == errUnrecognizedAddrType {
+			if err := sendReply(conn, ReplyAddrTypeNotSupported, nil); err != nil {
+				return fmt.Errorf("failed to send reply: %v", err)
+			}
+		}
+		return fmt.Errorf("failed to read destination address: %v", err)
+	}
+	request.AuthContext = &AuthContext{Method: AuthMethodNoAuth}
+	if client, ok := conn.RemoteAddr().(*net.TCPAddr); ok {
+		request.RemoteAddr = &AddrSpec{IP: client.IP, Port: client.Port}
+	}
+
+	if err := s.handleRequest(request, conn); err != nil {
+		err = fmt.Errorf("failed to handle request: %v", err)
+		s.config.Logger.Printf("socks: %v", err)
+		return err
+	}
+
+	return nil
+}