Update buffer ref usage

common/buf/buffer.go

@@ -13,7 +13,7 @@ import (
 
 const (
 	ReversedHeader = 1024
-	BufferSize     = 20 * 1024
+	BufferSize     = 32 * 1024
 	UDPBufferSize  = 16 * 1024
 )
 
@@ -130,9 +130,6 @@ func (b *Buffer) Write(data []byte) (n int, err error) {
 	if b.IsFull() {
 		return 0, io.ErrShortBuffer
 	}
-	if b.end+len(data) > b.Cap() {
-		panic("buffer overflow: cap " + strconv.Itoa(len(b.data)) + ",end " + strconv.Itoa(b.end) + ", need " + strconv.Itoa(len(data)))
-	}
 	n = copy(b.data[b.end:], data)
 	b.end += n
 	return
@@ -338,9 +335,7 @@ func (b *Buffer) IncRef() {
 }
 
 func (b *Buffer) DecRef() {
-	if atomic.AddInt32(&b.refs, -1) == 0 {
-		b.Release()
-	}
+	atomic.AddInt32(&b.refs, -1)
 }
 
 func (b *Buffer) Release() {

common/bufio/buffer.go

@@ -3,25 +3,27 @@ package bufio
 import (
 	"io"
 
+	"github.com/sagernet/sing/common"
 	"github.com/sagernet/sing/common/buf"
+	N "github.com/sagernet/sing/common/network"
 )
 
 type BufferedReader struct {
-	upstream io.Reader
+	upstream N.ExtendedReader
 	buffer   *buf.Buffer
 }
 
 func NewBufferedReader(upstream io.Reader, buffer *buf.Buffer) *BufferedReader {
 	return &BufferedReader{
-		upstream: upstream,
+		upstream: NewExtendedReader(upstream),
 		buffer:   buffer,
 	}
 }
 
 func (r *BufferedReader) Read(p []byte) (n int, err error) {
 	if r.buffer.IsEmpty() {
-		r.buffer.FullReset()
-		_, err = r.buffer.ReadFrom(r.upstream)
+		r.buffer.Reset()
+		err = r.upstream.ReadBuffer(r.buffer)
 		if err != nil {
 			return
 		}
@@ -29,6 +31,21 @@ func (r *BufferedReader) Read(p []byte) (n int, err error) {
 	return r.buffer.Read(p)
 }
 
+func (r *BufferedReader) ReadBuffer(buffer *buf.Buffer) error {
+	if r.buffer.IsEmpty() {
+		r.buffer.Reset()
+		err := r.upstream.ReadBuffer(r.buffer)
+		if err != nil {
+			return err
+		}
+	}
+	if r.buffer.Len() > buffer.FreeLen() {
+		return common.Error(buffer.ReadFullFrom(r.buffer, buffer.FreeLen()))
+	} else {
+		return common.Error(buffer.ReadFullFrom(r.buffer, r.buffer.Len()))
+	}
+}
+
 func (r *BufferedReader) WriteTo(w io.Writer) (n int64, err error) {
 	return CopyExtendedBuffer(NewExtendedWriter(w), NewExtendedReader(r.upstream), r.buffer)
 }
@@ -36,3 +53,35 @@ func (r *BufferedReader) WriteTo(w io.Writer) (n int64, err error) {
 func (w *BufferedReader) Upstream() any {
 	return w.upstream
 }
+
+type BufferedWriter struct {
+	upstream io.Writer
+	buffer   *buf.Buffer
+}
+
+func NewBufferedWriter(upstream io.Writer, buffer *buf.Buffer) *BufferedWriter {
+	return &BufferedWriter{
+		upstream: upstream,
+		buffer:   buffer,
+	}
+}
+
+func (w *BufferedWriter) Write(p []byte) (n int, err error) {
+	for {
+		var writeN int
+		writeN, err = w.buffer.Write(p[n:])
+		n += writeN
+		if n == len(p) {
+			return
+		}
+		_, err = w.upstream.Write(w.buffer.Bytes())
+		if err != nil {
+			return
+		}
+		w.buffer.FullReset()
+	}
+}
+
+func (w *BufferedWriter) ReadFrom(r io.Reader) (n int64, err error) {
+	return CopyExtendedBuffer(NewExtendedWriter(w), NewExtendedReader(r), w.buffer)
+}

common/bufio/conn.go

@@ -59,11 +59,11 @@ func CopyExtended(dst N.ExtendedWriter, src N.ExtendedReader) (n int64, err erro
 	} else if dstUnsafe {
 		return CopyExtendedWithPool(dst, src)
 	}
-
 	_buffer := buf.StackNew()
 	defer common.KeepAlive(_buffer)
-
-	return CopyExtendedBuffer(dst, src, common.Dup(_buffer))
+	buffer := common.Dup(_buffer)
+	defer buffer.Release()
+	return CopyExtendedBuffer(dst, src, buffer)
 }
 
 func CopyExtendedBuffer(dst N.ExtendedWriter, src N.ExtendedReader, buffer *buf.Buffer) (n int64, err error) {
@@ -145,6 +145,7 @@ func CopyPacket(dst N.PacketWriter, src N.PacketReader) (n int64, err error) {
 	_buffer := buf.StackNewPacket()
 	defer common.KeepAlive(_buffer)
 	buffer := common.Dup(_buffer)
+	defer buffer.Release()
 	buffer.IncRef()
 	defer buffer.DecRef()
 	var destination M.Socksaddr
@@ -175,6 +176,7 @@ func CopyPacketTimeout(dst N.PacketWriter, src N.TimeoutPacketReader, timeout ti
 	_buffer := buf.StackNewPacket()
 	defer common.KeepAlive(_buffer)
 	buffer := common.Dup(_buffer)
+	defer buffer.Release()
 	buffer.IncRef()
 	defer buffer.DecRef()
 	var destination M.Socksaddr

transport/udp/listener.go

@@ -161,6 +161,7 @@ func (l *Listener) loop() {
 	_buffer := buf.StackNewPacket()
 	defer common.KeepAlive(_buffer)
 	buffer := common.Dup(_buffer)
+	defer buffer.Release()
 	buffer.IncRef()
 	defer buffer.DecRef()
 	if !l.tproxy {