Repository: dragonflyoss/Dragonfly2
Pull request #1539 (2022-08-08 07:29:56)

feat: schedulers create main peer record
Signed-off-by: Gaius <gaius.qi@gmail.com>

<!--- Provide a general summary of your changes in the Title above -->

## Description
- Schedulers create main peer record.
<!--- Describe your changes in detail -->

## Related Issue

<!--- This project only accepts pull requests related to open issues -->
<!--- If suggesting a new feature or change, please discuss it in an issue first -->
<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
<!--- Please link to the issue here: -->

## Motivation and Context

<!--- Why is this change required? What problem does it solve? -->

## Screenshots (if appropriate)

## Types of changes

<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->

- [ ] Bug fix (non-breaking change which fixes an issue)
- [x] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing functionality to change)
- [ ] Documentation Update (if none of the other choices apply)

## Checklist

<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->

- [ ] My change requires a change to the documentation.
- [ ] I have updated the documentation accordingly.
- [x] I have read the **CONTRIBUTING** document.
- [ ] I have added tests to cover my changes.


codecov-commenter: # [Codecov](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) Report
> Merging [#1539](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (518b8fe) into [main](https://codecov.io/gh/dragonflyoss/Dragonfly2/commit/3e68e912ee3358676058ebbfaa57d55ad0261575?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) (3e68e91) will **decrease** coverage by `0.31%`.
> The diff coverage is `46.66%`.

[![Impacted file tree graph](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539/graphs/tree.svg?width=650&height=150&src=pr&token=PUICJYSAT7&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss)

```diff
@@            Coverage Diff             @@
##             main    #1539      +/-   ##
==========================================
- Coverage   43.21%   42.90%   -0.32%     
==========================================
  Files         116      116              
  Lines       10824    10875      +51     
==========================================
- Hits         4678     4666      -12     
- Misses       5659     5726      +67     
+ Partials      487      483       -4     
```

| Flag | Coverage Δ | |
|---|---|---|
| Object-compatibility-e2etests | `?` | |
| e2etests | `?` | |
| unittests | `42.90% <46.66%> (-0.32%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss) | Coverage Δ | |
|---|---|---|
| [scheduler/resource/seed\_peer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3Jlc291cmNlL3NlZWRfcGVlci5nbw==) | `19.11% <0.00%> (-2.92%)` | :arrow_down: |
| [scheduler/service/service.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3NlcnZpY2Uvc2VydmljZS5nbw==) | `77.35% <33.33%> (-5.74%)` | :arrow_down: |
| [scheduler/resource/peer.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3Jlc291cmNlL3BlZXIuZ28=) | `65.98% <100.00%> (+4.15%)` | :arrow_up: |
| [scheduler/scheduler/evaluator/evaluator\_base.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-c2NoZWR1bGVyL3NjaGVkdWxlci9ldmFsdWF0b3IvZXZhbHVhdG9yX2Jhc2UuZ28=) | `90.76% <100.00%> (ø)` | |
| [client/daemon/peer/peertask\_piecetask\_poller.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX3BpZWNldGFza19wb2xsZXIuZ28=) | `30.43% <0.00%> (-16.67%)` | :arrow_down: |
| [client/daemon/peer/peertask\_conductor.go](https://codecov.io/gh/dragonflyoss/Dragonfly2/pull/1539/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=dragonflyoss#diff-Y2xpZW50L2RhZW1vbi9wZWVyL3BlZXJ0YXNrX2NvbmR1Y3Rvci5nbw==) | `56.32% <0.00%> (+0.19%)` | :arrow_up: |


Patch:
diff --git a/scheduler/resource/peer.go b/scheduler/resource/peer.go
index 07828601..d42e6264 100644
--- a/scheduler/resource/peer.go
+++ b/scheduler/resource/peer.go
@@ -18,6 +18,7 @@ package resource
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
 	"net/http"
@@ -116,8 +117,11 @@ type Peer struct {
 	// Tag is peer tag.
 	Tag string
 
-	// Pieces is piece bitset.
-	Pieces *bitset.BitSet
+	// Pieces is finished piece set.
+	Pieces set.SafeSet[*schedulerv1.PieceResult]
+
+	// Pieces is finished pieces bitset.
+	FinishedPieces *bitset.BitSet
 
 	// pieceCosts is piece downloaded time.
 	pieceCosts []int64
@@ -166,7 +170,8 @@ func NewPeer(id string, task *Task, host *Host, options ...PeerOption) *Peer {
 	p := &Peer{
 		ID:               id,
 		Tag:              DefaultTag,
-		Pieces:           &bitset.BitSet{},
+		Pieces:           set.NewSafeSet[*schedulerv1.PieceResult](),
+		FinishedPieces:   &bitset.BitSet{},
 		pieceCosts:       []int64{},
 		Stream:           &atomic.Value{},
 		Task:             task,
@@ -328,6 +333,37 @@ func (p *Peer) Parents() []*Peer {
 	return parents
 }
 
+// MainParent returns the parent whose
+// peer has downloaded the most pieces.
+func (p *Peer) MainParent() (*Peer, error) {
+	var (
+		parents       = map[string]int{}
+		maxPieceCount = 0
+		mainParentID  string
+	)
+
+	for _, piece := range p.Pieces.Values() {
+		if piece.DstPid != "" {
+			parents[piece.DstPid]++
+		}
+
+		if parents[piece.DstPid] > maxPieceCount {
+			mainParentID = piece.DstPid
+		}
+	}
+
+	if mainParentID == "" {
+		return nil, errors.New("can not found main parent")
+	}
+
+	parent, err := p.Task.DAG.GetVertex(mainParentID)
+	if err != nil {
+		return nil, err
+	}
+
+	return parent.Value, nil
+}
+
 // Children returns children of peer.
 func (p *Peer) Children() []*Peer {
 	vertex, err := p.Task.DAG.GetVertex(p.ID)
diff --git a/scheduler/resource/seed_peer.go b/scheduler/resource/seed_peer.go
index 1ae9a5ed..8040ccc9 100644
--- a/scheduler/resource/seed_peer.go
+++ b/scheduler/resource/seed_peer.go
@@ -121,7 +121,16 @@ func (s *seedPeer) TriggerTask(ctx context.Context, task *Task) (*Peer, *schedul
 
 		// Handle piece download successfully.
 		peer.Log.Infof("receive piece from seed peer: %#v %#v", piece, piece.PieceInfo)
-		peer.Pieces.Set(uint(piece.PieceInfo.PieceNum))
+		peer.Pieces.Add(&schedulerv1.PieceResult{
+			TaskId:          task.ID,
+			SrcPid:          peer.ID,
+			BeginTime:       piece.BeginTime,
+			EndTime:         piece.EndTime,
+			Success:         true,
+			PieceInfo:       piece.PieceInfo,
+			ExtendAttribute: piece.ExtendAttribute,
+		})
+		peer.FinishedPieces.Set(uint(piece.PieceInfo.PieceNum))
 		peer.AppendPieceCost(pkgtime.SubNano(int64(piece.EndTime), int64(piece.BeginTime)).Milliseconds())
 		task.StorePiece(piece.PieceInfo)
 
diff --git a/scheduler/scheduler/evaluator/evaluator_base.go b/scheduler/scheduler/evaluator/evaluator_base.go
index ed08364e..a6cd1068 100644
--- a/scheduler/scheduler/evaluator/evaluator_base.go
+++ b/scheduler/scheduler/evaluator/evaluator_base.go
@@ -97,14 +97,14 @@ func calculatePieceScore(parent *resource.Peer, child *resource.Peer, totalPiece
 	// If the total piece is determined, normalize the number of
 	// pieces downloaded by the parent node.
 	if totalPieceCount > 0 {
-		finishedPieceCount := parent.Pieces.Count()
+		finishedPieceCount := parent.FinishedPieces.Count()
 		return float64(finishedPieceCount) / float64(totalPieceCount)
 	}
 
 	// Use the difference between the parent node and the child node to
 	// download the piece to roughly represent the piece score.
-	parentFinishedPieceCount := parent.Pieces.Count()
-	childFinishedPieceCount := child.Pieces.Count()
+	parentFinishedPieceCount := parent.FinishedPieces.Count()
+	childFinishedPieceCount := child.FinishedPieces.Count()
 	return float64(parentFinishedPieceCount) - float64(childFinishedPieceCount)
 }
 
diff --git a/scheduler/service/service.go b/scheduler/service/service.go
index b421930e..9e74d9ad 100644
--- a/scheduler/service/service.go
+++ b/scheduler/service/service.go
@@ -442,7 +442,15 @@ func (s *Service) AnnounceTask(ctx context.Context, req *schedulerv1.AnnounceTas
 
 		// Load downloaded piece infos.
 		for _, pieceInfo := range req.PiecePacket.PieceInfos {
-			peer.Pieces.Set(uint(pieceInfo.PieceNum))
+			peer.Pieces.Add(&schedulerv1.PieceResult{
+				TaskId:          taskID,
+				SrcPid:          peerID,
+				DstPid:          req.PiecePacket.DstPid,
+				Success:         true,
+				PieceInfo:       pieceInfo,
+				ExtendAttribute: req.PiecePacket.ExtendAttribute,
+			})
+			peer.FinishedPieces.Set(uint(pieceInfo.PieceNum))
 			peer.AppendPieceCost(int64(pieceInfo.DownloadCost) * int64(time.Millisecond))
 			task.StorePiece(pieceInfo)
 		}
@@ -638,8 +646,9 @@ func (s *Service) handleEndOfPiece(ctx context.Context, peer *resource.Peer) {}
 
 // handlePieceSuccess handles successful piece.
 func (s *Service) handlePieceSuccess(ctx context.Context, peer *resource.Peer, piece *schedulerv1.PieceResult) {
-	// Update peer piece info
-	peer.Pieces.Set(uint(piece.PieceInfo.PieceNum))
+	// Update peer piece info.
+	peer.Pieces.Add(piece)
+	peer.FinishedPieces.Set(uint(piece.PieceInfo.PieceNum))
 	peer.AppendPieceCost(pkgtime.SubNano(int64(piece.EndTime), int64(piece.BeginTime)).Milliseconds())
 
 	// When the peer downloads back-to-source,
@@ -843,24 +852,43 @@ func (s *Service) handleTaskFail(ctx context.Context, task *resource.Task, backT
 
 // createRecord stores peer download records.
 func (s *Service) createRecord(peer *resource.Peer, peerState int, req *schedulerv1.PeerResult) {
+	parent, err := peer.MainParent()
+	if err != nil {
+		peer.Log.Error(err)
+		return
+	}
+
 	record := storage.Record{
-		ID:              peer.ID,
-		IP:              peer.Host.IP,
-		Hostname:        peer.Host.Hostname,
-		Tag:             peer.Tag,
-		Cost:            req.Cost,
-		PieceCount:      int32(peer.Pieces.Count()),
-		TotalPieceCount: peer.Task.TotalPieceCount.Load(),
-		ContentLength:   peer.Task.ContentLength.Load(),
-		SecurityDomain:  peer.Host.SecurityDomain,
-		IDC:             peer.Host.IDC,
-		NetTopology:     peer.Host.NetTopology,
-		Location:        peer.Host.Location,
-		FreeUploadLoad:  peer.Host.FreeUploadLoad(),
-		State:           peerState,
-		HostType:        int(peer.Host.Type),
-		CreateAt:        peer.CreateAt.Load().UnixNano(),
-		UpdateAt:        peer.UpdateAt.Load().UnixNano(),
+		ID:                   peer.ID,
+		IP:                   peer.Host.IP,
+		Hostname:             peer.Host.Hostname,
+		Tag:                  peer.Tag,
+		Cost:                 req.Cost,
+		PieceCount:           int32(peer.FinishedPieces.Count()),
+		TotalPieceCount:      peer.Task.TotalPieceCount.Load(),
+		ContentLength:        peer.Task.ContentLength.Load(),
+		SecurityDomain:       peer.Host.SecurityDomain,
+		IDC:                  peer.Host.IDC,
+		NetTopology:          peer.Host.NetTopology,
+		Location:             peer.Host.Location,
+		FreeUploadLoad:       peer.Host.FreeUploadLoad(),
+		State:                peerState,
+		HostType:             int(peer.Host.Type),
+		CreateAt:             peer.CreateAt.Load().UnixNano(),
+		UpdateAt:             peer.UpdateAt.Load().UnixNano(),
+		ParentID:             parent.ID,
+		ParentIP:             parent.Host.IP,
+		ParentHostname:       parent.Host.Hostname,
+		ParentTag:            parent.Tag,
+		ParentPieceCount:     int32(parent.FinishedPieces.Count()),
+		ParentSecurityDomain: parent.Host.SecurityDomain,
+		ParentIDC:            parent.Host.IDC,
+		ParentNetTopology:    parent.Host.NetTopology,
+		ParentLocation:       parent.Host.Location,
+		ParentFreeUploadLoad: parent.Host.FreeUploadLoad(),
+		ParentHostType:       int(parent.Host.Type),
+		ParentCreateAt:       parent.CreateAt.Load().UnixNano(),
+		ParentUpdateAt:       parent.UpdateAt.Load().UnixNano(),
 	}
 
 	if err := s.storage.Create(record); err != nil {


Test patch:
diff --git a/scheduler/resource/peer_test.go b/scheduler/resource/peer_test.go
index fd1ae57e..e3687a55 100644
--- a/scheduler/resource/peer_test.go
+++ b/scheduler/resource/peer_test.go
@@ -55,7 +55,8 @@ func TestPeer_NewPeer(t *testing.T) {
 			expect: func(t *testing.T, peer *Peer, mockTask *Task, mockHost *Host) {
 				assert := assert.New(t)
 				assert.Equal(peer.ID, mockPeerID)
-				assert.Empty(peer.Pieces)
+				assert.Equal(peer.Pieces.Len(), uint(0))
+				assert.Empty(peer.FinishedPieces)
 				assert.Equal(len(peer.PieceCosts()), 0)
 				assert.Empty(peer.Stream)
 				assert.Equal(peer.FSM.Current(), PeerStatePending)
@@ -74,7 +75,8 @@ func TestPeer_NewPeer(t *testing.T) {
 				assert := assert.New(t)
 				assert.Equal(peer.ID, mockPeerID)
 				assert.Equal(peer.Tag, "foo")
-				assert.Empty(peer.Pieces)
+				assert.Equal(peer.Pieces.Len(), uint(0))
+				assert.Empty(peer.FinishedPieces)
 				assert.Equal(len(peer.PieceCosts()), 0)
 				assert.Empty(peer.Stream)
 				assert.Equal(peer.FSM.Current(), PeerStatePending)
@@ -311,6 +313,73 @@ func TestPeer_Parents(t *testing.T) {
 	}
 }
 
+func TestPeer_MainParent(t *testing.T) {
+	tests := []struct {
+		name   string
+		expect func(t *testing.T, peer *Peer, seedPeer *Peer, stream schedulerv1.Scheduler_ReportPieceResultServer)
+	}{
+		{
+			name: "get main parent",
+			expect: func(t *testing.T, peer *Peer, seedPeer *Peer, stream schedulerv1.Scheduler_ReportPieceResultServer) {
+				assert := assert.New(t)
+				peer.Pieces.Add(&schedulerv1.PieceResult{
+					DstPid: peer.ID,
+				})
+				peer.Pieces.Add(&schedulerv1.PieceResult{
+					DstPid: seedPeer.ID,
+				})
+				peer.Pieces.Add(&schedulerv1.PieceResult{
+					DstPid: seedPeer.ID,
+				})
+				peer.Task.StorePeer(peer)
+				peer.Task.StorePeer(seedPeer)
+				parent, err := peer.MainParent()
+				assert.NoError(err)
+				assert.Equal(parent.ID, seedPeer.ID)
+			},
+		},
+		{
+			name: "can not found main parent",
+			expect: func(t *testing.T, peer *Peer, seedPeer *Peer, stream schedulerv1.Scheduler_ReportPieceResultServer) {
+				assert := assert.New(t)
+				peer.Pieces.Add(&schedulerv1.PieceResult{})
+				peer.Task.StorePeer(peer)
+				peer.Task.StorePeer(seedPeer)
+				_, err := peer.MainParent()
+				assert.EqualError(err, "can not found main parent")
+			},
+		},
+		{
+			name: "get main parent failed",
+			expect: func(t *testing.T, peer *Peer, seedPeer *Peer, stream schedulerv1.Scheduler_ReportPieceResultServer) {
+				assert := assert.New(t)
+				peer.Pieces.Add(&schedulerv1.PieceResult{
+					DstPid: "foo",
+				})
+				peer.Pieces.Add(&schedulerv1.PieceResult{})
+				peer.Task.StorePeer(peer)
+				peer.Task.StorePeer(seedPeer)
+				_, err := peer.MainParent()
+				assert.EqualError(err, "vertex not found")
+			},
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			ctl := gomock.NewController(t)
+			defer ctl.Finish()
+			stream := mocks.NewMockScheduler_ReportPieceResultServer(ctl)
+
+			mockHost := NewHost(mockRawHost)
+			mockTask := NewTask(mockTaskID, mockTaskURL, commonv1.TaskType_Normal, mockTaskURLMeta, WithBackToSourceLimit(mockTaskBackToSourceLimit))
+			peer := NewPeer(mockPeerID, mockTask, mockHost)
+			seedPeer := NewPeer(mockSeedPeerID, mockTask, mockHost)
+			tc.expect(t, peer, seedPeer, stream)
+		})
+	}
+}
+
 func TestPeer_Children(t *testing.T) {
 	tests := []struct {
 		name   string
diff --git a/scheduler/scheduler/evaluator/evaluator_base_test.go b/scheduler/scheduler/evaluator/evaluator_base_test.go
index f1a18de2..ecd5a836 100644
--- a/scheduler/scheduler/evaluator/evaluator_base_test.go
+++ b/scheduler/scheduler/evaluator/evaluator_base_test.go
@@ -113,7 +113,7 @@ func TestEvaluatorBase_Evaluate(t *testing.T) {
 			mock: func(parent *resource.Peer, child *resource.Peer) {
 				parent.Host.SecurityDomain = "bac"
 				child.Host.SecurityDomain = "bac"
-				parent.Pieces.Set(0)
+				parent.FinishedPieces.Set(0)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -128,7 +128,7 @@ func TestEvaluatorBase_Evaluate(t *testing.T) {
 			mock: func(parent *resource.Peer, child *resource.Peer) {
 				parent.Host.SecurityDomain = ""
 				child.Host.SecurityDomain = "baz"
-				parent.Pieces.Set(0)
+				parent.FinishedPieces.Set(0)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -143,7 +143,7 @@ func TestEvaluatorBase_Evaluate(t *testing.T) {
 			mock: func(parent *resource.Peer, child *resource.Peer) {
 				parent.Host.SecurityDomain = "baz"
 				child.Host.SecurityDomain = ""
-				parent.Pieces.Set(0)
+				parent.FinishedPieces.Set(0)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -179,7 +179,7 @@ func TestEvaluatorBase_calculatePieceScore(t *testing.T) {
 			child:           resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost),
 			totalPieceCount: 0,
 			mock: func(parent *resource.Peer, child *resource.Peer) {
-				parent.Pieces.Set(0)
+				parent.FinishedPieces.Set(0)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -192,7 +192,7 @@ func TestEvaluatorBase_calculatePieceScore(t *testing.T) {
 			child:           resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost),
 			totalPieceCount: 0,
 			mock: func(parent *resource.Peer, child *resource.Peer) {
-				child.Pieces.Set(0)
+				child.FinishedPieces.Set(0)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -205,9 +205,9 @@ func TestEvaluatorBase_calculatePieceScore(t *testing.T) {
 			child:           resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost),
 			totalPieceCount: 0,
 			mock: func(parent *resource.Peer, child *resource.Peer) {
-				parent.Pieces.Set(0)
-				child.Pieces.Set(0)
-				child.Pieces.Set(1)
+				parent.FinishedPieces.Set(0)
+				child.FinishedPieces.Set(0)
+				child.FinishedPieces.Set(1)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -220,8 +220,8 @@ func TestEvaluatorBase_calculatePieceScore(t *testing.T) {
 			child:           resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost),
 			totalPieceCount: 0,
 			mock: func(parent *resource.Peer, child *resource.Peer) {
-				parent.Pieces.Set(0)
-				child.Pieces.Set(0)
+				parent.FinishedPieces.Set(0)
+				child.FinishedPieces.Set(0)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -234,9 +234,9 @@ func TestEvaluatorBase_calculatePieceScore(t *testing.T) {
 			child:           resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost),
 			totalPieceCount: 0,
 			mock: func(parent *resource.Peer, child *resource.Peer) {
-				parent.Pieces.Set(0)
-				parent.Pieces.Set(1)
-				child.Pieces.Set(0)
+				parent.FinishedPieces.Set(0)
+				parent.FinishedPieces.Set(1)
+				child.FinishedPieces.Set(0)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
@@ -260,8 +260,8 @@ func TestEvaluatorBase_calculatePieceScore(t *testing.T) {
 			child:           resource.NewPeer(idgen.PeerID("127.0.0.1"), mockTask, mockHost),
 			totalPieceCount: 10,
 			mock: func(parent *resource.Peer, child *resource.Peer) {
-				parent.Pieces.Set(0)
-				parent.Pieces.Set(1)
+				parent.FinishedPieces.Set(0)
+				parent.FinishedPieces.Set(1)
 			},
 			expect: func(t *testing.T, score float64) {
 				assert := assert.New(t)
diff --git a/scheduler/scheduler/scheduler_test.go b/scheduler/scheduler/scheduler_test.go
index 151e9eef..121f47a9 100644
--- a/scheduler/scheduler/scheduler_test.go
+++ b/scheduler/scheduler/scheduler_test.go
@@ -512,7 +512,7 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 				mockPeer.FSM.SetState(resource.PeerStateRunning)
 				peer.Task.StorePeer(peer)
 				peer.Task.StorePeer(mockPeer)
-				mockPeer.Pieces.Set(0)
+				mockPeer.FinishedPieces.Set(0)
 
 				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
 			},
@@ -530,7 +530,7 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 				mockPeer.IsBackToSource.Store(true)
 				peer.Task.StorePeer(peer)
 				peer.Task.StorePeer(mockPeer)
-				mockPeer.Pieces.Set(0)
+				mockPeer.FinishedPieces.Set(0)
 				peer.StoreStream(stream)
 				gomock.InOrder(
 					md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1),
@@ -569,7 +569,7 @@ func TestScheduler_NotifyAndFindParent(t *testing.T) {
 				peer.Task.BackToSourcePeers.Add(candidatePeer.ID)
 				mockPeer.IsBackToSource.Store(true)
 				candidatePeer.IsBackToSource.Store(true)
-				mockPeer.Pieces.Set(0)
+				mockPeer.FinishedPieces.Set(0)
 				peer.StoreStream(stream)
 				gomock.InOrder(
 					md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1),
@@ -725,10 +725,10 @@ func TestScheduler_FindParent(t *testing.T) {
 				peer.Task.BackToSourcePeers.Add(mockPeers[1].ID)
 				mockPeers[0].IsBackToSource.Store(true)
 				mockPeers[1].IsBackToSource.Store(true)
-				mockPeers[0].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(1)
-				mockPeers[1].Pieces.Set(2)
+				mockPeers[0].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(1)
+				mockPeers[1].FinishedPieces.Set(2)
 
 				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
 			},
@@ -750,10 +750,10 @@ func TestScheduler_FindParent(t *testing.T) {
 				peer.Task.StorePeer(mockPeers[2])
 				mockPeers[0].Host.Type = resource.HostTypeSuperSeed
 				mockPeers[1].Host.Type = resource.HostTypeSuperSeed
-				mockPeers[0].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(1)
-				mockPeers[1].Pieces.Set(2)
+				mockPeers[0].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(1)
+				mockPeers[1].FinishedPieces.Set(2)
 
 				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
 			},
@@ -772,10 +772,10 @@ func TestScheduler_FindParent(t *testing.T) {
 				peer.Task.StorePeer(peer)
 				peer.Task.StorePeer(mockPeers[0])
 				peer.Task.StorePeer(mockPeers[1])
-				mockPeers[0].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(1)
-				mockPeers[1].Pieces.Set(2)
+				mockPeers[0].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(1)
+				mockPeers[1].FinishedPieces.Set(2)
 
 				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
 			},
@@ -803,10 +803,10 @@ func TestScheduler_FindParent(t *testing.T) {
 					t.Fatal(err)
 				}
 
-				mockPeers[0].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(1)
-				mockPeers[1].Pieces.Set(2)
+				mockPeers[0].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(1)
+				mockPeers[1].FinishedPieces.Set(2)
 
 				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{}, false).Times(1)
 			},
@@ -848,10 +848,10 @@ func TestScheduler_FindParent(t *testing.T) {
 				peer.Task.BackToSourcePeers.Add(mockPeers[1].ID)
 				mockPeers[0].IsBackToSource.Store(true)
 				mockPeers[1].IsBackToSource.Store(true)
-				mockPeers[0].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(0)
-				mockPeers[1].Pieces.Set(1)
-				mockPeers[1].Pieces.Set(2)
+				mockPeers[0].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(0)
+				mockPeers[1].FinishedPieces.Set(1)
+				mockPeers[1].FinishedPieces.Set(2)
 
 				md.GetSchedulerClusterConfig().Return(types.SchedulerClusterConfig{
 					FilterParentLimit: 3,
diff --git a/scheduler/service/service_test.go b/scheduler/service/service_test.go
index 0bbced9a..891a8fbb 100644
--- a/scheduler/service/service_test.go
+++ b/scheduler/service/service_test.go
@@ -1059,7 +1059,6 @@ func TestService_ReportPeerResult(t *testing.T) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
-					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -1082,7 +1081,6 @@ func TestService_ReportPeerResult(t *testing.T) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
-					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -1105,7 +1103,6 @@ func TestService_ReportPeerResult(t *testing.T) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
-					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -1128,7 +1125,6 @@ func TestService_ReportPeerResult(t *testing.T) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
-					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -1151,7 +1147,6 @@ func TestService_ReportPeerResult(t *testing.T) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
-					ms.Create(gomock.Any()).Return(nil).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -1174,7 +1169,6 @@ func TestService_ReportPeerResult(t *testing.T) {
 				gomock.InOrder(
 					mr.PeerManager().Return(peerManager).Times(1),
 					mp.Load(gomock.Eq(mockPeerID)).Return(mockPeer, true).Times(1),
-					ms.Create(gomock.Any()).Return(errors.New("foo")).Times(1),
 				)
 			},
 			expect: func(t *testing.T, peer *resource.Peer, err error) {
@@ -1349,7 +1343,7 @@ func TestService_AnnounceTask(t *testing.T) {
 				assert.True(ok)
 				assert.EqualValues(piece, &commonv1.PieceInfo{PieceNum: 1, DownloadCost: 1})
 
-				assert.Equal(mockPeer.Pieces.Count(), uint(1))
+				assert.Equal(mockPeer.FinishedPieces.Count(), uint(1))
 				assert.Equal(mockPeer.PieceCosts()[0], int64(1*time.Millisecond))
 				assert.Equal(mockPeer.FSM.Current(), resource.PeerStateSucceeded)
 			},
@@ -1392,7 +1386,7 @@ func TestService_AnnounceTask(t *testing.T) {
 				assert.True(ok)
 
 				assert.EqualValues(piece, &commonv1.PieceInfo{PieceNum: 1, DownloadCost: 1})
-				assert.Equal(mockPeer.Pieces.Count(), uint(1))
+				assert.Equal(mockPeer.FinishedPieces.Count(), uint(1))
 				assert.Equal(mockPeer.PieceCosts()[0], int64(1*time.Millisecond))
 				assert.Equal(mockPeer.FSM.Current(), resource.PeerStateSucceeded)
 			},
@@ -1435,7 +1429,7 @@ func TestService_AnnounceTask(t *testing.T) {
 				assert.True(ok)
 
 				assert.EqualValues(piece, &commonv1.PieceInfo{PieceNum: 1, DownloadCost: 1})
-				assert.Equal(mockPeer.Pieces.Count(), uint(1))
+				assert.Equal(mockPeer.FinishedPieces.Count(), uint(1))
 				assert.Equal(mockPeer.PieceCosts()[0], int64(1*time.Millisecond))
 				assert.Equal(mockPeer.FSM.Current(), resource.PeerStateSucceeded)
 			},
@@ -1478,7 +1472,7 @@ func TestService_AnnounceTask(t *testing.T) {
 				assert.True(ok)
 
 				assert.EqualValues(piece, &commonv1.PieceInfo{PieceNum: 1, DownloadCost: 1})
-				assert.Equal(mockPeer.Pieces.Count(), uint(1))
+				assert.Equal(mockPeer.FinishedPieces.Count(), uint(1))
 				assert.Equal(mockPeer.PieceCosts()[0], int64(1*time.Millisecond))
 				assert.Equal(mockPeer.FSM.Current(), resource.PeerStateSucceeded)
 			},
@@ -2427,7 +2421,8 @@ func TestService_handlePieceSuccess(t *testing.T) {
 			},
 			expect: func(t *testing.T, peer *resource.Peer) {
 				assert := assert.New(t)
-				assert.Equal(peer.Pieces.Count(), uint(1))
+				assert.Equal(peer.Pieces.Len(), uint(1))
+				assert.Equal(peer.FinishedPieces.Count(), uint(1))
 				assert.Equal(peer.PieceCosts(), []int64{1})
 			},
 		},
@@ -2447,7 +2442,8 @@ func TestService_handlePieceSuccess(t *testing.T) {
 			},
 			expect: func(t *testing.T, peer *resource.Peer) {
 				assert := assert.New(t)
-				assert.Equal(peer.Pieces.Count(), uint(1))
+				assert.Equal(peer.Pieces.Len(), uint(1))
+				assert.Equal(peer.FinishedPieces.Count(), uint(1))
 				assert.Equal(peer.PieceCosts(), []int64{1})
 				piece, ok := peer.Task.LoadPiece(0)
 				assert.True(ok)