Fix chunked streaming (#1015)

* http: refactor out crlf reading function

Make it a bit simpler and make it reusable.

* streaming: fix chunked stream test

This test is supposed to check for stream unchunking, but it's not really
effective with that because chunks created by createChunkedBody() get wrapped
into another chunk by writeBodyStream(), so we end up with chunkedBody in
request handler although what we really want is plain body.

Deduplicate test and benchmark also.

* streaming: fix Read interface

It wasn't actually compatible with io.Reader as io.Reader _never_ returns n >
len(p) while this function easily did that for chunked payloads confusing its
users:

panic: runtime error: slice bounds out of range [:528] with capacity 512

goroutine 562 [running]:
io.ReadAll(0x9f4380, 0xc0003be1a0, 0xc0004fcd80, 0x0, 0x0, 0xc00086bc30, 0x46f99b)
        /usr/lib64/go/1.16/src/io/io.go:634 +0x205
io/ioutil.ReadAll(...)
        /usr/lib64/go/1.16/src/io/ioutil/ioutil.go:27
github.com/valyala/fasthttp.getChunkedTestEnv.func1(0xc001fdc680)
        /home/rik/dev/fasthttp/streaming_test.go:108 +0x6c
github.com/valyala/fasthttp.(*Server).serveConn(0xc000416d80, 0xa034e8, 0xc0004da880, 0x0, 0x0)
        /home/rik/dev/fasthttp/server.go:2219 +0x12ee
github.com/valyala/fasthttp.(*workerPool).workerFunc(0xc000148960, 0xc0003be160)
        /home/rik/dev/fasthttp/workerpool.go:223 +0xba
github.com/valyala/fasthttp.(*workerPool).getCh.func1(0xc000148960, 0xc0003be160, 0x8b4ec0, 0xc0003be160)
        /home/rik/dev/fasthttp/workerpool.go:195 +0x35
created by github.com/valyala/fasthttp.(*workerPool).getCh
        /home/rik/dev/fasthttp/workerpool.go:194 +0x11f

It also returned len(p) in some cases where it read less than that.

1 files changed, 30 insertions, 19 deletions

diff --git a/streaming.go b/streaming.go
index 39000a2..1a3d748 100644
--- a/streaming.go
+++ b/streaming.go
@@ -3,7 +3,6 @@ package fasthttp
 import (
 	"bufio"
 	"bytes"
-	"fmt"
 	"io"
 	"sync"
 
@@ -15,36 +14,47 @@ type requestStream struct {
 	reader          *bufio.Reader
 	totalBytesRead  int
 	contentLength   int
+	chunkLeft       int
 }
 
 func (rs *requestStream) Read(p []byte) (int, error) {
+	var (
+		n   int
+		err error
+	)
 	if rs.contentLength == -1 {
-		p = p[:0]
-		strCRLFLen := len(strCRLF)
-		chunkSize, err := parseChunkSize(rs.reader)
-		if err != nil {
-			return len(p), err
+		if rs.chunkLeft == 0 {
+			chunkSize, err := parseChunkSize(rs.reader)
+			if err != nil {
+				return 0, err
+			}
+			if chunkSize == 0 {
+				err = readCrLf(rs.reader)
+				if err == nil {
+					err = io.EOF
+				}
+				return 0, err
+			}
+			rs.chunkLeft = chunkSize
 		}
-		p, err = appendBodyFixedSize(rs.reader, p, chunkSize+strCRLFLen)
-		if err != nil {
-			return len(p), err
+		bytesToRead := len(p)
+		if rs.chunkLeft < len(p) {
+			bytesToRead = rs.chunkLeft
 		}
-		if !bytes.Equal(p[len(p)-strCRLFLen:], strCRLF) {
-			return len(p), ErrBrokenChunk{
-				error: fmt.Errorf("cannot find crlf at the end of chunk"),
-			}
+		n, err = rs.reader.Read(p[:bytesToRead])
+		rs.totalBytesRead += n
+		rs.chunkLeft -= n
+		if err == io.EOF {
+			err = io.ErrUnexpectedEOF
 		}
-		p = p[:len(p)-strCRLFLen]
-		if chunkSize == 0 {
-			return len(p), io.EOF
+		if err == nil && rs.chunkLeft == 0 {
+			err = readCrLf(rs.reader)
 		}
-		return len(p), nil
+		return n, err
 	}
 	if rs.totalBytesRead == rs.contentLength {
 		return 0, io.EOF
 	}
-	var n int
-	var err error
 	prefetchedSize := int(rs.prefetchedBytes.Size())
 	if prefetchedSize > rs.totalBytesRead {
 		left := prefetchedSize - rs.totalBytesRead
@@ -87,6 +97,7 @@ func acquireRequestStream(b *bytebufferpool.ByteBuffer, r *bufio.Reader, content
 func releaseRequestStream(rs *requestStream) {
 	rs.prefetchedBytes = nil
 	rs.totalBytesRead = 0
+	rs.chunkLeft = 0
 	rs.reader = nil
 	requestStreamPool.Put(rs)
 }