From 0d3ac68d653cf5cf018775e96ccad69ffeb545ba Mon Sep 17 00:00:00 2001
From: slasher <shenjie1@oppo.com>
Date: Tue, 12 Nov 2024 19:44:21 +0800
Subject: [PATCH] feat(access): with other timeout in read only data shard

. #22782275

close #3717

Signed-off-by: slasher <shenjie1@oppo.com>
---
 blobstore/access/stream/stream.go           |  2 ++
 blobstore/access/stream/stream_get.go       | 16 +++++++---
 blobstore/access/stream/stream_get_test.go  | 33 +++++++++++++++++++++
 blobstore/access/stream/stream_mock_test.go |  1 +
 blobstore/cmd/access/access.conf.default    |  1 +
 5 files changed, 49 insertions(+), 4 deletions(-)

diff --git a/blobstore/access/stream/stream.go b/blobstore/access/stream/stream.go
index 9f860ecd9..b261ba3ae 100644
--- a/blobstore/access/stream/stream.go
+++ b/blobstore/access/stream/stream.go
@@ -121,6 +121,7 @@ type StreamConfig struct {
 	EncoderEnableVerify        bool   `json:"encoder_enableverify"`
 	EncoderConcurrency         int    `json:"encoder_concurrency"`
 	MinReadShardsX             int    `json:"min_read_shards_x"`
+	ReadDataOnlyTimeoutMS      int    `json:"read_data_only_timeout_ms"`
 	ShardCrcDisabled           bool   `json:"shard_crc_disabled"`
 
 	MemPoolSizeClasses map[int]int `json:"mem_pool_size_classes"`
@@ -203,6 +204,7 @@ func confCheck(cfg *StreamConfig) error {
 	}
 	defaulter.LessOrEqual(&cfg.EncoderConcurrency, defaultEncoderConcurrency)
 	defaulter.LessOrEqual(&cfg.MinReadShardsX, defaultMinReadShardsX)
+	defaulter.LessOrEqual(&cfg.ReadDataOnlyTimeoutMS, 3*1000)
 
 	defaulter.LessOrEqual(&cfg.ClusterConfig.CMClientConfig.Config.ClientTimeoutMs, defaultTimeoutClusterMgr)
 	defaulter.LessOrEqual(&cfg.BlobnodeConfig.ClientTimeoutMs, defaultTimeoutBlobnode)
diff --git a/blobstore/access/stream/stream_get.go b/blobstore/access/stream/stream_get.go
index 2debc1cf9..e6937192d 100644
--- a/blobstore/access/stream/stream_get.go
+++ b/blobstore/access/stream/stream_get.go
@@ -552,6 +552,10 @@ func (h *Handler) getDataShardOnly(ctx context.Context, getTime *timeReadWrite,
 	firstShardIdx := int(blob.Offset) / shardSize
 	shardOffset := int(blob.Offset) % shardSize
 
+	ctx, cancel := context.WithDeadline(ctx,
+		time.Now().Add(time.Millisecond*time.Duration(h.ReadDataOnlyTimeoutMS)))
+	defer cancel()
+
 	startRead := time.Now()
 	remainSize := blob.ReadSize
 	bufOffset := 0
@@ -636,10 +640,14 @@ func (h *Handler) getOneShardFromHost(ctx context.Context, serviceController con
 			}
 		}
 
-		// new child span to get from blobnode, we should finish it here.
-		spanChild, ctxChild := trace.StartSpanFromContextWithTraceID(
-			context.Background(), "GetFromBlobnode", span.TraceID())
-		defer spanChild.Finish()
+		ctxChild := ctx
+		if cancelChan != nil { // cancelChan == nil means reading data shard only
+			// new child span to get from blobnode, we should finish it here.
+			var spanChild trace.Span
+			spanChild, ctxChild = trace.StartSpanFromContextWithTraceID(
+				context.Background(), "GetFromBlobnode", span.TraceID())
+			defer spanChild.Finish()
+		}
 
 		body, _, err := h.blobnodeClient.RangeGetShard(ctxChild, host, &args)
 		if err == nil {
diff --git a/blobstore/access/stream/stream_get_test.go b/blobstore/access/stream/stream_get_test.go
index 9f5d30f32..945f96219 100644
--- a/blobstore/access/stream/stream_get_test.go
+++ b/blobstore/access/stream/stream_get_test.go
@@ -291,6 +291,39 @@ func TestAccessStreamGetShardBroken(t *testing.T) {
 	}
 }
 
+func TestAccessStreamGetShardOnlyTimeout(t *testing.T) {
+	ctx := ctxWithName("TestAccessStreamGetShardOnlyTimeout")
+	dataShards.clean()
+	oldMs := streamer.ReadDataOnlyTimeoutMS
+	streamer.ReadDataOnlyTimeoutMS = 100
+	defer func() {
+		streamer.ReadDataOnlyTimeoutMS = oldMs
+		dataShards.clean()
+	}()
+
+	size := 1
+	buff := make([]byte, size)
+	rand.Read(buff)
+	loc, err := streamer.Put(ctx(), bytes.NewReader(buff), int64(size), nil)
+	require.NoError(t, err)
+
+	// blocking the data shard, force to waiting ReadDataOnlyTimeoutMS
+	vuidController.Block(1001)
+	defer func() {
+		vuidController.Unblock(1001)
+	}()
+	{
+		startTime := time.Now()
+		transfer, err := streamer.Get(ctx(), bytes.NewBuffer(nil), *loc, uint64(size), 0)
+		require.NoError(t, err)
+		err = transfer()
+		require.NoError(t, err)
+
+		duration := time.Since(startTime)
+		require.GreaterOrEqual(t, duration, 100*time.Millisecond, "greater duration:", duration)
+	}
+}
+
 func TestAccessStreamGetLocalIDC(t *testing.T) {
 	ctx := ctxWithName("TestAccessStreamGetLocalIDC")
 	dataShards.clean()
diff --git a/blobstore/access/stream/stream_mock_test.go b/blobstore/access/stream/stream_mock_test.go
index f0ad77c61..89f670eb7 100644
--- a/blobstore/access/stream/stream_mock_test.go
+++ b/blobstore/access/stream/stream_mock_test.go
@@ -521,6 +521,7 @@ func init() {
 			AllocRetryTimes:        3,
 			AllocRetryIntervalMS:   3000,
 			MinReadShardsX:         minReadShardsX,
+			ReadDataOnlyTimeoutMS:  10000,
 		},
 		discardVidChan: make(chan discardVid, 8),
 		stopCh:         make(chan struct{}),
diff --git a/blobstore/cmd/access/access.conf.default b/blobstore/cmd/access/access.conf.default
index b44c5f01c..bde79e303 100644
--- a/blobstore/cmd/access/access.conf.default
+++ b/blobstore/cmd/access/access.conf.default
@@ -51,6 +51,7 @@
         "encoder_concurrency": 1000,
         "encoder_enableverify": true,
         "min_read_shards_x": 1,
+        "read_data_only_timeout_ms": 3000,
         "shard_crc_disabled": false,
         "cluster_config": {
             "region": "region",