feat(access): add nopdata in access

. #1000117047

Signed-off-by: slasher <shenjie1@oppo.com>

blobstore/access/stream/stream_get.go

@@ -906,7 +906,7 @@ func emptyDataShardIndexes(sizes ec.BufferSizes) map[int]struct{} {
 	firstEmptyIdx := (sizes.DataSize + sizes.ShardSize - 1) / sizes.ShardSize
 	n := sizes.ECDataSize / sizes.ShardSize
 	if firstEmptyIdx >= n {
-		return make(map[int]struct{})
+		return nil
 	}
 
 	set := make(map[int]struct{}, n-firstEmptyIdx)

blobstore/access/stream/stream_mock_test.go

@@ -43,6 +43,7 @@ import (
 	"github.com/cubefs/cubefs/blobstore/common/trace"
 	"github.com/cubefs/cubefs/blobstore/testing/mocks"
 	_ "github.com/cubefs/cubefs/blobstore/testing/nolog"
+	"github.com/cubefs/cubefs/blobstore/util/bytespool"
 )
 
 var (
@@ -301,9 +302,13 @@ var storageAPIPutShard = func(ctx context.Context, host string, args *blobnode.P
 	defer memPool.Put(buffer)
 
 	buffer = buffer[:int(args.Size)]
-	_, err = io.ReadFull(args.Body, buffer)
-	if err != nil {
-		return
+	if args.NopData {
+		bytespool.Zero(buffer)
+	} else {
+		_, err = io.ReadFull(args.Body, buffer)
+		if err != nil {
+			return
+		}
 	}
 
 	crc = crc32.ChecksumIEEE(buffer)

blobstore/access/stream/stream_put.go

@@ -37,9 +37,6 @@ import (
 	"github.com/cubefs/cubefs/blobstore/util/retry"
 )
 
-// TODO: To Be Continue
-//  put empty shard to blobnode if file has been aligned.
-
 // Put put one object
 //
 //	required: size, file size
@@ -125,6 +122,7 @@ func (h *Handler) Put(ctx context.Context,
 		if err != nil {
 			return nil, err
 		}
+		empties := emptyDataShardIndexes(buffer.BufferSizes)
 
 		readBuff := buffer.DataBuf[:bsize]
 		shards, err := encoder.Split(buffer.ECDataBuf)
@@ -157,7 +155,7 @@ func (h *Handler) Put(ctx context.Context,
 		buffer = nil
 		<-ready
 		startWrite := time.Now()
-		err = h.writeToBlobnodesWithHystrix(ctx, blobident, shards, func() {
+		err = h.writeToBlobnodesWithHystrix(ctx, blobident, shards, empties, func() {
 			takeoverBuffer.Release()
 			ready <- struct{}{}
 		})
@@ -172,12 +170,12 @@ func (h *Handler) Put(ctx context.Context,
 }
 
 func (h *Handler) writeToBlobnodesWithHystrix(ctx context.Context,
-	blob blobIdent, shards [][]byte, callback func(),
+	blob blobIdent, shards [][]byte, empties map[int]struct{}, callback func(),
 ) error {
 	safe := make(chan struct{}, 1)
 	err := hystrix.Do(rwCommand, func() error {
 		safe <- struct{}{}
-		return h.writeToBlobnodes(ctx, blob, shards, callback)
+		return h.writeToBlobnodes(ctx, blob, shards, empties, callback)
 	}, nil)
 
 	select {
@@ -197,7 +195,7 @@ type shardPutStatus struct {
 // takeover ec buffer release by callback.
 // return if had quorum successful shards, then wait all shards in background.
 func (h *Handler) writeToBlobnodes(ctx context.Context,
-	blob blobIdent, shards [][]byte, callback func(),
+	blob blobIdent, shards [][]byte, empties map[int]struct{}, callback func(),
 ) (err error) {
 	span := trace.SpanFromContextSafe(ctx)
 	clusterID, vid, bid := blob.cid, blob.vid, blob.bid
@@ -247,6 +245,8 @@ func (h *Handler) writeToBlobnodes(ctx context.Context,
 				wg.Done()
 			}()
 
+			_, empty := empties[index]
+
 			diskID := unit.DiskID
 			args := &blobnode.PutShardArgs{
 				DiskID: diskID,
@@ -254,6 +254,8 @@ func (h *Handler) writeToBlobnodes(ctx context.Context,
 				Bid:    bid,
 				Size:   int64(len(shards[index])),
 				Type:   blobnode.NormalIO,
+
+				NopData: empty,
 			}
 
 			crcDisable := h.ShardCrcWriteDisable
@@ -287,7 +289,9 @@ func (h *Handler) writeToBlobnodes(ctx context.Context,
 				crc       uint32
 			)
 			writeErr = retry.ExponentialBackoff(3, 200).RuptOn(func() (bool, error) {
-				args.Body = bytes.NewReader(shards[index])
+				if !args.NopData {
+					args.Body = bytes.NewReader(shards[index])
+				}
 
 				crc, err = h.blobnodeClient.PutShard(ctxChild, host, args)
 				if err == nil {

blobstore/access/stream/stream_putat.go

@@ -56,6 +56,7 @@ func (h *Handler) PutAt(ctx context.Context, rc io.Reader,
 	if err != nil {
 		return err
 	}
+	empties := emptyDataShardIndexes(buffer.BufferSizes)
 
 	putTime := new(timeReadWrite)
 	putTime.IncA(time.Since(st))
@@ -92,7 +93,7 @@ func (h *Handler) PutAt(ctx context.Context, rc io.Reader,
 	takeoverBuffer := buffer
 	buffer = nil
 	startWrite := time.Now()
-	err = h.writeToBlobnodesWithHystrix(ctx, blobident, shards, func() {
+	err = h.writeToBlobnodesWithHystrix(ctx, blobident, shards, empties, func() {
 		takeoverBuffer.Release()
 	})
 	putTime.IncW(time.Since(startWrite))

blobstore/api/access/proto.go

@@ -220,7 +220,7 @@ func (args *PutArgs) IsValid() bool {
 // PutResp put response result
 type PutResp struct {
 	Location   proto.Location `json:"location"`
-	HashSumMap HashSumMap     `json:"hashsum"`
+	HashSumMap HashSumMap     `json:"hashsum,omitempty"`
 }
 
 // PutAtArgs for service /putat