chore: adapt new ReadWait interfaces

common/net/deadline/conn.go

@@ -0,0 +1,149 @@
+package deadline
+
+import (
+	"net"
+	"os"
+	"time"
+
+	"github.com/metacubex/mihomo/common/atomic"
+
+	"github.com/sagernet/sing/common/buf"
+	"github.com/sagernet/sing/common/bufio"
+	"github.com/sagernet/sing/common/network"
+)
+
+type connReadResult struct {
+	buffer []byte
+	err    error
+}
+
+type Conn struct {
+	network.ExtendedConn
+	deadline     atomic.TypedValue[time.Time]
+	pipeDeadline pipeDeadline
+	disablePipe  atomic.Bool
+	inRead       atomic.Bool
+	resultCh     chan *connReadResult
+}
+
+func NewConn(conn net.Conn) *Conn {
+	c := &Conn{
+		ExtendedConn: bufio.NewExtendedConn(conn),
+		pipeDeadline: makePipeDeadline(),
+		resultCh:     make(chan *connReadResult, 1),
+	}
+	c.resultCh <- nil
+	return c
+}
+
+func (c *Conn) Read(p []byte) (n int, err error) {
+	select {
+	case result := <-c.resultCh:
+		if result != nil {
+			n = copy(p, result.buffer)
+			err = result.err
+			if n >= len(result.buffer) {
+				c.resultCh <- nil // finish cache read
+			} else {
+				result.buffer = result.buffer[n:]
+				c.resultCh <- result // push back for next call
+			}
+			return
+		} else {
+			c.resultCh <- nil
+			break
+		}
+	case <-c.pipeDeadline.wait():
+		return 0, os.ErrDeadlineExceeded
+	}
+
+	if c.disablePipe.Load() {
+		return c.ExtendedConn.Read(p)
+	} else if c.deadline.Load().IsZero() {
+		c.inRead.Store(true)
+		defer c.inRead.Store(false)
+		return c.ExtendedConn.Read(p)
+	}
+
+	<-c.resultCh
+	go c.pipeRead(len(p))
+
+	return c.Read(p)
+}
+
+func (c *Conn) pipeRead(size int) {
+	buffer := make([]byte, size)
+	n, err := c.ExtendedConn.Read(buffer)
+	buffer = buffer[:n]
+	c.resultCh <- &connReadResult{
+		buffer: buffer,
+		err:    err,
+	}
+}
+
+func (c *Conn) ReadBuffer(buffer *buf.Buffer) (err error) {
+	select {
+	case result := <-c.resultCh:
+		if result != nil {
+			n, _ := buffer.Write(result.buffer)
+			err = result.err
+
+			if n >= len(result.buffer) {
+				c.resultCh <- nil // finish cache read
+			} else {
+				result.buffer = result.buffer[n:]
+				c.resultCh <- result // push back for next call
+			}
+			return
+		} else {
+			c.resultCh <- nil
+			break
+		}
+	case <-c.pipeDeadline.wait():
+		return os.ErrDeadlineExceeded
+	}
+
+	if c.disablePipe.Load() {
+		return c.ExtendedConn.ReadBuffer(buffer)
+	} else if c.deadline.Load().IsZero() {
+		c.inRead.Store(true)
+		defer c.inRead.Store(false)
+		return c.ExtendedConn.ReadBuffer(buffer)
+	}
+
+	<-c.resultCh
+	go c.pipeRead(buffer.FreeLen())
+
+	return c.ReadBuffer(buffer)
+}
+
+func (c *Conn) SetReadDeadline(t time.Time) error {
+	if c.disablePipe.Load() {
+		return c.ExtendedConn.SetReadDeadline(t)
+	} else if c.inRead.Load() {
+		c.disablePipe.Store(true)
+		return c.ExtendedConn.SetReadDeadline(t)
+	}
+	c.deadline.Store(t)
+	c.pipeDeadline.set(t)
+	return nil
+}
+
+func (c *Conn) ReaderReplaceable() bool {
+	select {
+	case result := <-c.resultCh:
+		c.resultCh <- result
+		if result != nil {
+			return false // cache reading
+		} else {
+			break
+		}
+	default:
+		return false // pipe reading
+	}
+	return c.disablePipe.Load() || c.deadline.Load().IsZero()
+}
+
+func (c *Conn) Upstream() any {
+	return c.ExtendedConn
+}

common/net/deadline/packet_sing.go

@@ -5,6 +5,7 @@ import (
 	"runtime"
 
 	"github.com/metacubex/mihomo/common/net/packet"
+
 	"github.com/sagernet/sing/common/buf"
 	"github.com/sagernet/sing/common/bufio"
 	M "github.com/sagernet/sing/common/metadata"
@@ -121,17 +122,18 @@ type singPacketReadWaiter struct {
 
 type singWaitReadResult singReadResult
 
-func (c *singPacketReadWaiter) InitializeReadWaiter(newBuffer func() *buf.Buffer) {
-	c.packetReadWaiter.InitializeReadWaiter(newBuffer)
+func (c *singPacketReadWaiter) InitializeReadWaiter(options N.ReadWaitOptions) (needCopy bool) {
+	return c.packetReadWaiter.InitializeReadWaiter(options)
 }
 
-func (c *singPacketReadWaiter) WaitReadPacket() (destination M.Socksaddr, err error) {
+func (c *singPacketReadWaiter) WaitReadPacket() (buffer *buf.Buffer, destination M.Socksaddr, err error) {
 FOR:
 	for {
 		select {
 		case result := <-c.netPacketConn.resultCh:
 			if result != nil {
 				if result, ok := result.(*singWaitReadResult); ok {
+					buffer = result.buffer
 					destination = result.destination
 					err = result.err
 					c.netPacketConn.resultCh <- nil // finish cache read
@@ -145,7 +147,7 @@ FOR:
 				break FOR
 			}
 		case <-c.netPacketConn.pipeDeadline.wait():
-			return M.Socksaddr{}, os.ErrDeadlineExceeded
+			return nil, M.Socksaddr{}, os.ErrDeadlineExceeded
 		}
 	}
 
@@ -154,8 +156,7 @@ FOR:
 	} else if c.netPacketConn.deadline.Load().IsZero() {
 		c.netPacketConn.inRead.Store(true)
 		defer c.netPacketConn.inRead.Store(false)
-		destination, err = c.packetReadWaiter.WaitReadPacket()
-		return
+		return c.packetReadWaiter.WaitReadPacket()
 	}
 
 	<-c.netPacketConn.resultCh
@@ -165,8 +166,9 @@ FOR:
 }
 
 func (c *singPacketReadWaiter) pipeWaitReadPacket() {
-	destination, err := c.packetReadWaiter.WaitReadPacket()
+	buffer, destination, err := c.packetReadWaiter.WaitReadPacket()
 	result := &singWaitReadResult{}
+	result.buffer = buffer
 	result.destination = destination
 	result.err = err
 	c.netPacketConn.resultCh <- result

common/net/packet/packet_sing.go

@@ -24,16 +24,16 @@ type enhanceSingPacketConn struct {
 func (c *enhanceSingPacketConn) WaitReadFrom() (data []byte, put func(), addr net.Addr, err error) {
 	var buff *buf.Buffer
 	var dest M.Socksaddr
-	newBuffer := func() *buf.Buffer {
-		buff = buf.NewPacket() // do not use stack buffer
-		return buff
-	}
+	rwOptions := N.ReadWaitOptions{}
 	if c.packetReadWaiter != nil {
-		c.packetReadWaiter.InitializeReadWaiter(newBuffer)
-		defer c.packetReadWaiter.InitializeReadWaiter(nil)
-		dest, err = c.packetReadWaiter.WaitReadPacket()
+		c.packetReadWaiter.InitializeReadWaiter(rwOptions)
+		buff, dest, err = c.packetReadWaiter.WaitReadPacket()
 	} else {
-		dest, err = c.SingPacketConn.ReadPacket(newBuffer())
+		buff = rwOptions.NewPacketBuffer()
+		dest, err = c.SingPacketConn.ReadPacket(buff)
+		if buff != nil {
+			rwOptions.PostReturn(buff)
+		}
 	}
 	if dest.IsFqdn() {
 		addr = dest
@@ -41,9 +41,7 @@ func (c *enhanceSingPacketConn) WaitReadFrom() (data []byte, put func(), addr ne
 		addr = dest.UDPAddr()
 	}
 	if err != nil {
-		if buff != nil {
-			buff.Release()
-		}
+		buff.Release()
 		return
 	}
 	if buff == nil {

common/net/sing.go

@@ -5,9 +5,10 @@ import (
 	"net"
 	"runtime"
 
+	"github.com/metacubex/mihomo/common/net/deadline"
+
 	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/bufio"
-	"github.com/sagernet/sing/common/bufio/deadline"
 	"github.com/sagernet/sing/common/network"
 )
 
@@ -19,8 +20,10 @@ type ExtendedConn = network.ExtendedConn
 type ExtendedWriter = network.ExtendedWriter
 type ExtendedReader = network.ExtendedReader
 
+var WriteBuffer = bufio.WriteBuffer
+
 func NewDeadlineConn(conn net.Conn) ExtendedConn {
-	return deadline.NewFallbackConn(conn)
+	return deadline.NewConn(conn)
 }
 
 func NeedHandshake(conn any) bool {