Add: 0-copy reader

common/buf/buffer.go

@@ -31,6 +31,22 @@ func New() *Buffer {
 	}
 }
 
+func NewExisted(b []byte) *Buffer {
+	if cap(b) < Size {
+		panic("Invalid buffer")
+	}
+
+	oLen := len(b)
+	if oLen < Size {
+		b = append(b, make([]byte, Size-oLen)...)
+	}
+
+	return &Buffer{
+		v:   b,
+		end: int32(oLen),
+	}
+}
+
 // StackNew creates a new Buffer object on stack.
 // This method is for buffers that is released in the same function.
 func StackNew() Buffer {

transport/internet/grpc/encoding/conn.go

@@ -42,18 +42,13 @@ func NewHunkConn(hc HunkConn, cancel context.CancelFunc) net.Conn {
 }
 
 func (h *HunkReaderWriter) forceFetch() error {
-	// clean up buffer, safety first!
-	h.buf = h.buf[:0]
-	h.index = 0
-
-	hunk := new(Hunk)
-	hunk.Data = h.buf
-	err := h.hc.RecvMsg(hunk)
+	hunk, err := h.hc.Recv()
 	if err != nil {
 		return newError("failed to fetch hunk from gRPC tunnel").Base(err)
 	}
 
 	h.buf = hunk.Data
+	h.index = 0
 
 	return nil
 }
@@ -70,6 +65,27 @@ func (h *HunkReaderWriter) Read(buf []byte) (int, error) {
 	return n, nil
 }
 
+func (h *HunkReaderWriter) ReadMultiBuffer() (buf.MultiBuffer, error) {
+	if h.index >= len(h.buf) {
+		if err := h.forceFetch(); err != nil {
+			return nil, err
+		}
+	}
+
+	if cap(h.buf) == buf.Size {
+		b := h.buf
+		h.index = len(h.buf)
+		return buf.MultiBuffer{buf.NewExisted(b)}, nil
+	}
+
+	b := buf.New()
+	_, err := b.ReadFrom(h)
+	if err != nil {
+		return nil, err
+	}
+	return buf.MultiBuffer{b}, nil
+}
+
 func (h *HunkReaderWriter) Write(buf []byte) (int, error) {
 	err := h.hc.Send(&Hunk{Data: buf[:]})
 	if err != nil {