object/bos: reduce memory allocation in bos

pkg/object/bos.go

@@ -20,6 +20,7 @@
 package object
 
 import (
+	"bytes"
 	"fmt"
 	"io"
 	"net/http"
@@ -32,6 +33,7 @@ import (
 	"github.com/baidubce/bce-sdk-go/bce"
 	"github.com/baidubce/bce-sdk-go/services/bos"
 	"github.com/baidubce/bce-sdk-go/services/bos/api"
+	"github.com/juicedata/juicefs/pkg/utils"
 )
 
 type bosclient struct {
@@ -114,7 +116,18 @@ func (q *bosclient) Put(key string, in io.Reader, getters ...AttrGetter) error {
 	if err != nil {
 		return err
 	}
-	body, err := bce.NewBodyFromSizedReader(b, vlen)
+	var data []byte
+	if bf, ok := b.(*bytes.Buffer); ok {
+		data = bf.Bytes()
+	} else {
+		data = utils.DynAlloc(int(vlen))
+		defer utils.DynFree(data)
+		_, err = io.ReadFull(b, data)
+		if err != nil {
+			return err
+		}
+	}
+	body, err := bce.NewBodyFromBytes(data)
 	if err != nil {
 		return err
 	}

pkg/object/qingstor.go

@@ -135,7 +135,7 @@ func findLen(in io.Reader) (io.Reader, int64, error) {
 			return nil, 0, err
 		}
 		vlen = int64(len(d))
-		in = bytes.NewReader(d)
+		in = bytes.NewBuffer(d)
 	}
 	return in, vlen, nil
 }

pkg/sync/sync.go

@@ -552,43 +552,14 @@ func (w *withProgress) Read(b []byte) (int, error) {
 	return n, err
 }
 
-func dynAlloc(size int) []byte {
-	zeros := utils.PowerOf2(size)
-	b := *dynPools[zeros].Get().(*[]byte)
-	if cap(b) < size {
-		panic(fmt.Sprintf("%d < %d", cap(b), size))
-	}
-	return b[:size]
-}
-
-func dynFree(b []byte) {
-	dynPools[utils.PowerOf2(cap(b))].Put(&b)
-}
-
-var dynPools []*sync.Pool
-
-func init() {
-	dynPools = make([]*sync.Pool, 33) // 1 - 8G
-	for i := 0; i < 33; i++ {
-		func(bits int) {
-			dynPools[i] = &sync.Pool{
-				New: func() interface{} {
-					b := make([]byte, 1<<bits)
-					return &b
-				},
-			}
-		}(i)
-	}
-}
-
 func doUploadPart(src, dst object.ObjectStorage, srckey string, off, size int64, key, uploadID string, num int, calChksum bool) (*object.Part, uint32, error) {
 	if limiter != nil {
 		limiter.Wait(size)
 	}
 	start := time.Now()
 	sz := size
-	data := dynAlloc(int(size))
-	defer dynFree(data)
+	data := utils.DynAlloc(int(size))
+	defer utils.DynFree(data)
 	var part *object.Part
 	var chksum uint32
 	err := try(3, func() error {

pkg/utils/alloc.go

@@ -28,7 +28,7 @@ var used int64
 
 // Alloc returns size bytes memory from Go heap.
 func Alloc(size int) []byte {
-	zeros := PowerOf2(size)
+	zeros := powerOf2(size)
 	b := *pools[zeros].Get().(*[]byte)
 	if cap(b) < size {
 		panic(fmt.Sprintf("%d < %d", cap(b), size))
@@ -41,7 +41,7 @@ func Alloc(size int) []byte {
 func Free(b []byte) {
 	// buf could be zero length
 	atomic.AddInt64(&used, -int64(cap(b)))
-	pools[PowerOf2(cap(b))].Put(&b)
+	pools[powerOf2(cap(b))].Put(&b)
 }
 
 // AllocMemory returns the allocated memory
@@ -51,7 +51,7 @@ func AllocMemory() int64 {
 
 var pools []*sync.Pool
 
-func PowerOf2(s int) int {
+func powerOf2(s int) int {
 	var bits int
 	var p int = 1
 	for p < s {

pkg/utils/dynamic_alloc.go

@@ -0,0 +1,62 @@
+/*
+ * JuiceFS, Copyright 2025 Juicedata, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package utils
+
+import (
+	"fmt"
+	"sync"
+)
+
+func DynAlloc(size int) []byte {
+	zeros := powerOf2(size)
+	b := *dynPools[zeros].Get().(*[]byte)
+	if cap(b) < size {
+		panic(fmt.Sprintf("%d < %d", cap(b), size))
+	}
+	return b[:size]
+}
+
+// DynFree b may be longer than the original, in the actual size into the pool
+func DynFree(b []byte) {
+	dynPools[FloorPowerOf2(cap(b))].Put(&b)
+}
+
+func FloorPowerOf2(s int) int {
+	var bits int
+	var p = 1
+	for p <= s {
+		bits++
+		p *= 2
+	}
+	return bits - 1
+}
+
+var dynPools []*sync.Pool
+
+func init() {
+	dynPools = make([]*sync.Pool, 34) // 1 - 8G
+	for i := 0; i < 34; i++ {
+		func(bits int) {
+			dynPools[i] = &sync.Pool{
+				New: func() interface{} {
+					b := make([]byte, 1<<bits)
+					return &b
+				},
+			}
+		}(i)
+	}
+}