feat(meshlib): support borrowed unary frame buffers

connection_impl.go

@@ -601,3 +601,21 @@ func (c *connection) setState(newState connState) {
 func (c *connection) changeState(from, to connState) bool {
 	return atomic.CompareAndSwapInt32(&c.state, from, to)
 }
+
+// MarkRead and ConsumedSlices retain the normal Reader lifetime: the caller
+// must finish borrowing before Release. Neither method reads ahead or waits.
+func (c *connection) MarkRead() ReadMark { return c.inputBuffer.MarkRead() }
+
+func (c *connection) ConsumedLen(mark ReadMark) int { return c.inputBuffer.ConsumedLen(mark) }
+
+func (c *connection) ConsumedSlices(mark ReadMark, dst [][]byte) [][]byte {
+	return c.inputBuffer.ConsumedSlices(mark, dst)
+}
+
+func (c *connection) WaitReadable(n int) error { return c.waitRead(n) }
+func (c *connection) PeekSlices(n int, dst [][]byte) ([][]byte, error) {
+	if err := c.waitRead(n); err != nil {
+		return dst, err
+	}
+	return c.inputBuffer.PeekSlices(n, dst), nil
+}

go.mod

@@ -3,7 +3,9 @@ module github.com/cloudwego/netpoll
 go 1.20
 
 require (
-	github.com/bytedance/gopkg v0.1.1
+	github.com/bytedance/gopkg v0.1.4
 	github.com/cloudwego/gopkg v0.1.4
 	golang.org/x/sys v0.30.0
 )
+
+replace github.com/cloudwego/gopkg => github.com/junliurs/cloudwego-gopkg v0.0.0-20261105192813-e9461b4bc999

go.sum

@@ -1,10 +1,6 @@
-github.com/bytedance/gopkg v0.1.1 h1:3azzgSkiaw79u24a+w9arfH8OfnQQ4MHUt9lJFREEaE=
-github.com/bytedance/gopkg v0.1.1/go.mod h1:576VvJ+eJgyCzdjS+c4+77QF3p7ubbtiKARP3TxducM=
-github.com/cloudwego/gopkg v0.1.4 h1:EoQiCG4sTonTPHxOGE0VlQs+sQR+Hsi2uN0qqwu8O50=
-github.com/cloudwego/gopkg v0.1.4/go.mod h1:FQuXsRWRsSqJLsMVd5SYzp8/Z1y5gXKnVvRrWUOsCMI=
-github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg=
+github.com/bytedance/gopkg v0.1.4 h1:oZnQwnX82KAIWb7033bEwtxvTqXcYMxDBaQxo5JJHWM=
+github.com/bytedance/gopkg v0.1.4/go.mod h1:v1zWfPm21Fb+OsyXN2VAHdL6TBb2L88anLQgdyje6R4=
+github.com/junliurs/cloudwego-gopkg v0.0.0-20261105192813-e9461b4bc999 h1:mSf8zc6KfnIYTJ7GyrHI0URNRBy8+A0cZG2VxB5/kh8=
+github.com/junliurs/cloudwego-gopkg v0.0.0-20261105192813-e9461b4bc999/go.mod h1:yK8wKouWJR6rbJ12Qb9VLTwbzcrg7c2rMAXHCfZT6h4=
 golang.org/x/sys v0.30.0 h1:QjkSwP/36a20jFYWkSue1YwXzLmsV5Gfq7Eiy72C1uc=
 golang.org/x/sys v0.30.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA=
-gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

meshlib_borrow_test.go

@@ -0,0 +1,91 @@
+package netpoll
+
+import (
+	"bytes"
+	"testing"
+)
+
+func TestMeshlibBorrowConsumedMessagePreservesNodesAndBoundary(t *testing.T) {
+	b := NewLinkBuffer()
+	defer b.Close()
+	first := []byte("first-message")
+	second := []byte("next-message")
+	_, _ = b.WriteBinary(first)
+	_ = b.Flush()
+	mark := b.MarkRead()
+	if err := b.Skip(len(first)); err != nil {
+		t.Fatal(err)
+	}
+	_, _ = b.WriteBinary(second)
+	_ = b.Flush()
+	var scratch [4][]byte
+	parts := b.ConsumedSlices(mark, scratch[:0])
+	if got := b.ConsumedLen(mark); got != len(first) {
+		t.Fatalf("consumed %d bytes, want %d", got, len(first))
+	}
+	if len(parts) != 1 || !bytes.Equal(parts[0], first) {
+		t.Fatal(parts)
+	}
+	firstView := parts[0]
+	again := b.ConsumedSlices(mark, scratch[:0])
+	if &again[0][0] != &firstView[0] {
+		t.Fatal("copied payload")
+	}
+	remaining, err := b.Next(len(second))
+	if err != nil || !bytes.Equal(remaining, second) {
+		t.Fatal("consumed next message", err)
+	}
+	if got := b.ConsumedLen(mark); got != len(first)+len(second) {
+		t.Fatalf("consumed %d bytes across nodes", got)
+	}
+}
+
+func TestMeshlibReadMarksSurviveCopyReadsUntilRelease(t *testing.T) {
+	b := NewLinkBuffer()
+	defer b.Close()
+	for _, text := range []string{"first-block", "middle-block", "last-block"} {
+		_, _ = b.WriteBinary([]byte(text))
+		_ = b.Flush()
+	}
+	first := b.MarkRead()
+	if n := b.readCopy(make([]byte, len("first-block"))); n != len("first-block") {
+		t.Fatal(n)
+	}
+	second := b.MarkRead()
+	if n := b.readCopy(make([]byte, len("middle-block")+1)); n != len("middle-block")+1 {
+		t.Fatal(n)
+	}
+	for _, test := range []struct {
+		mark ReadMark
+		want string
+	}{{first, "first-blockmiddle-blockl"}, {second, "middle-blockl"}} {
+		if got := b.ConsumedLen(test.mark); got != len(test.want) {
+			t.Fatalf("consumed=%d want=%d", got, len(test.want))
+		}
+		parts := b.ConsumedSlices(test.mark, nil)
+		if got := string(bytes.Join(parts, nil)); got != test.want {
+			t.Fatalf("borrowed=%q want=%q", got, test.want)
+		}
+	}
+	if err := b.Release(); err != nil {
+		t.Fatal(err)
+	}
+	if b.readMarked || b.head != b.read {
+		t.Fatal("Release retained a mark or consumed nodes")
+	}
+}
+
+func TestMeshlibCopyReadsStillReleaseNodesWithoutMarks(t *testing.T) {
+	b := NewLinkBuffer()
+	defer b.Close()
+	for _, text := range []string{"first-block", "last-block"} {
+		_, _ = b.WriteBinary([]byte(text))
+		_ = b.Flush()
+	}
+	if n := b.readCopy(make([]byte, len("first-block")+1)); n != len("first-block")+1 {
+		t.Fatal(n)
+	}
+	if b.head != b.read {
+		t.Fatal("unmarked copy read retained consumed nodes")
+	}
+}

nocopy_linkbuffer.go

@@ -56,10 +56,11 @@ type UnsafeLinkBuffer struct {
 	length     int64
 	mallocSize int
 
-	head  *linkBufferNode // release head
-	read  *linkBufferNode // read head
-	flush *linkBufferNode // malloc head
-	write *linkBufferNode // malloc tail
+	head       *linkBufferNode // release head
+	read       *linkBufferNode // read head
+	flush      *linkBufferNode // malloc head
+	write      *linkBufferNode // malloc tail
+	readMarked bool            // retain consumed nodes until Release while a ReadMark is live
 
 	// buf allocated by Next when cross-package, which should be freed when release
 	caches [][]byte
@@ -117,6 +118,9 @@ func (b *UnsafeLinkBuffer) readCopy(p []byte) (n int) {
 	for b.read != b.flush && b.read.Len() == 0 {
 		b.read = b.read.next
 	}
+	if b.readMarked {
+		return n
+	}
 	// release consumed nodes that are not readExposed.
 	// exposed nodes stay in the chain so Release() can free them later.
 	//
@@ -272,6 +276,7 @@ func (b *UnsafeLinkBuffer) Skip(n int) (err error) {
 // Release the node that has been read.
 // b.flush == nil indicates that this LinkBuffer is created by LinkBuffer.Slice
 func (b *UnsafeLinkBuffer) Release() (err error) {
+	b.readMarked = false
 	for b.read != b.flush && b.read.Len() == 0 {
 		b.read = b.read.next
 	}
@@ -961,3 +966,76 @@ func (node *linkBufferNode) reusable() bool {
 func (node *linkBufferNode) readExposed() bool {
 	return node.mode&flagReadExposed > 0
 }
+
+// ReadMark identifies a retained read position. It is valid only until Release.
+// Taking a mark does not allocate, walk nodes, or inspect message contents.
+type ReadMark struct {
+	node   *linkBufferNode
+	offset int
+}
+
+func (b *UnsafeLinkBuffer) MarkRead() ReadMark {
+	b.readMarked = true
+	return ReadMark{b.read, b.read.off}
+}
+
+// ConsumedLen counts bytes consumed since mark, including Skip. Like the mark,
+// it is valid only before Release. It neither allocates nor inspects payloads.
+func (b *UnsafeLinkBuffer) ConsumedLen(mark ReadMark) int {
+	start, size := mark.offset, 0
+	for node := mark.node; node != nil; node = node.next {
+		// The reactor may extend the current write node. For the current
+		// reader node only off is needed; do not read its mutable slice header.
+		end := node.off
+		if node != b.read {
+			end = len(node.buf)
+		}
+		if end > start {
+			size += end - start
+		}
+		if node == b.read {
+			break
+		}
+		start = 0
+	}
+	return size
+}
+
+// ConsumedSlices borrows only bytes consumed since mark, without coalescing.
+func (b *UnsafeLinkBuffer) ConsumedSlices(mark ReadMark, dst [][]byte) [][]byte {
+	start := mark.offset
+	for node := mark.node; node != nil; node = node.next {
+		// The reactor may extend the current write node. For the current
+		// reader node only off is needed; do not read its mutable slice header.
+		end := node.off
+		if node != b.read {
+			end = len(node.buf)
+		}
+		if end > start {
+			dst = append(dst, node.buf[start:end])
+		}
+		if node == b.read {
+			break
+		}
+		start = 0
+	}
+	return dst
+}
+
+// PeekSlices borrows readable nodes without creating a contiguous payload.
+func (b *UnsafeLinkBuffer) PeekSlices(n int, dst [][]byte) [][]byte {
+	for node := b.read; n > 0 && node != nil; node = node.next {
+		v := node.buf[node.off:]
+		if len(v) > n {
+			v = v[:n]
+		}
+		if len(v) > 0 {
+			dst = append(dst, v)
+			n -= len(v)
+		}
+		if node == b.flush {
+			break
+		}
+	}
+	return dst
+}

nocopy_linkbuffer_race.go

@@ -235,3 +235,27 @@ func (b *SafeLinkBuffer) indexByte(c byte, skip int) int {
 	defer b.Unlock()
 	return b.UnsafeLinkBuffer.indexByte(c, skip)
 }
+
+func (b *SafeLinkBuffer) MarkRead() ReadMark {
+	b.Lock()
+	defer b.Unlock()
+	return b.UnsafeLinkBuffer.MarkRead()
+}
+
+func (b *SafeLinkBuffer) ConsumedLen(mark ReadMark) int {
+	b.Lock()
+	defer b.Unlock()
+	return b.UnsafeLinkBuffer.ConsumedLen(mark)
+}
+
+func (b *SafeLinkBuffer) ConsumedSlices(mark ReadMark, dst [][]byte) [][]byte {
+	b.Lock()
+	defer b.Unlock()
+	return b.UnsafeLinkBuffer.ConsumedSlices(mark, dst)
+}
+
+func (b *SafeLinkBuffer) PeekSlices(n int, dst [][]byte) [][]byte {
+	b.Lock()
+	defer b.Unlock()
+	return b.UnsafeLinkBuffer.PeekSlices(n, dst)
+}