fix: replace allocating download pages by using chunk.NewOffPage to s…

…ync.Pool.Get

pkg/sync/download.go

@@ -21,7 +21,6 @@ import (
 	"io"
 	"sync"
 
-	"github.com/juicedata/juicefs/pkg/chunk"
 	"github.com/juicedata/juicefs/pkg/object"
 )
 
@@ -33,7 +32,7 @@ type parallelDownloader struct {
 	fsize      int64
 	blockSize  int64
 	concurrent chan int
-	buffers    map[int64]*chunk.Page
+	buffers    map[int64]*[]byte
 	off        int64
 	err        error
 }
@@ -50,6 +49,15 @@ func (r *parallelDownloader) setErr(err error) {
 	r.err = err
 }
 
+const downloadBufSize = 10 << 20
+
+var downloadBufPool = sync.Pool{
+	New: func() interface{} {
+		buf := make([]byte, downloadBufSize)
+		return &buf
+	},
+}
+
 func (r *parallelDownloader) download() {
 	for off := int64(0); off < r.fsize; off += r.blockSize {
 		r.concurrent <- 1
@@ -73,18 +81,18 @@ func (r *parallelDownloader) download() {
 					r.setErr(e)
 				} else { //nolint:typecheck
 					defer in.Close()
-					p := chunk.NewOffPage(int(size))
-					_, e = io.ReadFull(in, p.Data)
+					p := downloadBufPool.Get().(*[]byte)
+					_, e = io.ReadFull(in, *p)
 					if e != nil {
 						r.setErr(e)
-						p.Release()
+						downloadBufPool.Put(p)
 					} else {
 						r.Lock()
 						if r.buffers != nil {
 							r.buffers[off] = p
 							saved = true
 						} else {
-							p.Release()
+							downloadBufPool.Put(p)
 						}
 						r.Unlock()
 					}
@@ -115,10 +123,10 @@ func (r *parallelDownloader) Read(b []byte) (int, error) {
 	if p == nil {
 		return 0, r.err
 	}
-	n := copy(b, p.Data[r.off-off:])
+	n := copy(b, (*p)[r.off-off:])
 	r.off += int64(n)
-	if r.off == off+int64(len(p.Data)) {
-		p.Release()
+	if r.off == off+int64(len(*p)) {
+		downloadBufPool.Put(&p)
 		r.Lock()
 		delete(r.buffers, off)
 		r.Unlock()
@@ -134,7 +142,7 @@ func (r *parallelDownloader) Close() {
 	r.Lock()
 	defer r.Unlock()
 	for _, p := range r.buffers {
-		p.Release()
+		downloadBufPool.Put(&p)
 	}
 	r.buffers = nil
 	if r.err == nil {
@@ -152,7 +160,7 @@ func newParallelDownloader(store object.ObjectStorage, key string, size int64, b
 		fsize:      size,
 		blockSize:  bSize,
 		concurrent: concurrent,
-		buffers:    make(map[int64]*chunk.Page),
+		buffers:    make(map[int64]*[]byte),
 	}
 	down.notify = sync.NewCond(down)
 	go down.download()

pkg/sync/sync.go

@@ -359,7 +359,7 @@ func doCopySingle(src, dst object.ObjectStorage, key string, size int64) error {
 	if size > maxBlock && !inMap(dst, readInMem) && !inMap(src, fastStreamRead) {
 		var err error
 		var in io.Reader
-		downer := newParallelDownloader(src, key, size, 10<<20, concurrent)
+		downer := newParallelDownloader(src, key, size, downloadBufSize, concurrent)
 		defer downer.Close()
 		if inMap(dst, streamWrite) {
 			in = downer