dragonflyoss
/
dragonfly
mirror of https://github.com/dragonflyoss/dragonfly.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: optimize parameters in seed peer DownloadTask (#2947) 

Signed-off-by: Gaius <gaius.qi@gmail.com>
This commit is contained in:
Gaius 2023-12-19 11:37:00 +08:00 committed by GitHub
parent a58fa9c32e
commit 003376c6d3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
8 changed files with 65 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -3,7 +3,7 @@ module d7y.io/dragonfly/v2
go 1.21
require (
d7y.io/api/v2 v2.0.62
d7y.io/api/v2 v2.0.64
github.com/MysteriousPotato/go-lockable v1.0.0
github.com/RichardKnop/machinery v1.10.6
github.com/Showmax/go-fqdn v1.0.0

4
go.sum
View File

@ -51,8 +51,8 @@ cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohl
cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
cloud.google.com/go/storage v1.14.0/go.mod h1:GrKmX003DSIwi9o29oFT7YDnHYwZoctc3fOKtUw0Xmo=
d7y.io/api/v2 v2.0.62 h1:q4/r24DxWT+4zsGGMe8HqbjC3cw+B/s2+gwI2oKC7Og=
d7y.io/api/v2 v2.0.62/go.mod h1:ve0R4ePgRYZVdnVyhWTOi2LdP3/gyf21ZwP2xij+3Io=
d7y.io/api/v2 v2.0.64 h1:2mdQ0maJZZgogfQHoCyzi1TBczyby1WeyFau13ywmDw=
d7y.io/api/v2 v2.0.64/go.mod h1:ve0R4ePgRYZVdnVyhWTOi2LdP3/gyf21ZwP2xij+3Io=
dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
dmitri.shuralyov.com/gpu/mtl v0.0.0-20201218220906-28db891af037/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
github.com/Azure/azure-sdk-for-go/sdk/azcore v1.0.0/go.mod h1:uGG2W01BaETf0Ozp+QxxKJdMBNRWPdstHG0Fmdwn1/U=

81
pkg/rpc/dfdaemon/client/client_v2.go
View File

@ -27,7 +27,6 @@ import (
grpc_prometheus "github.com/grpc-ecosystem/go-grpc-prometheus"
"google.golang.org/grpc"
commonv2 "d7y.io/api/v2/pkg/apis/common/v2"
dfdaemonv2 "d7y.io/api/v2/pkg/apis/dfdaemon/v2"
logger "d7y.io/dragonfly/v2/internal/dflog"
@ -61,7 +60,7 @@ func GetV2(ctx context.Context, target string, opts ...grpc.DialOption) (V2, err
}
return &v2{
DfdaemonClient: dfdaemonv2.NewDfdaemonClient(conn),
DfdaemonUploadClient: dfdaemonv2.NewDfdaemonUploadClient(conn),
ClientConn: conn,
}, nil
}
@ -69,88 +68,54 @@ func GetV2(ctx context.Context, target string, opts ...grpc.DialOption) (V2, err
// V2 is the interface for v2 version of the grpc client.
type V2 interface {
// SyncPieces syncs pieces from the other peers.
SyncPieces(context.Context, *dfdaemonv2.SyncPiecesRequest, ...grpc.CallOption) (dfdaemonv2.DfdaemonUpload_SyncPiecesClient, error)
// DownloadPiece downloads piece from the other peer.
DownloadPiece(context.Context, *dfdaemonv2.DownloadPieceRequest, ...grpc.CallOption) (*dfdaemonv2.DownloadPieceResponse, error)
// DownloadTask downloads task back-to-source.
// DownloadTask downloads task from the other peer.
DownloadTask(context.Context, *dfdaemonv2.DownloadTaskRequest, ...grpc.CallOption) error
// UploadTask uploads task to p2p network.
UploadTask(context.Context, *dfdaemonv2.UploadTaskRequest, ...grpc.CallOption) error
// StatTask stats task information.
StatTask(context.Context, *dfdaemonv2.StatTaskRequest, ...grpc.CallOption) (*commonv2.Task, error)
// DeleteTask deletes task from p2p network.
DeleteTask(context.Context, *dfdaemonv2.DeleteTaskRequest, ...grpc.CallOption) error
// Close tears down the ClientConn and all underlying connections.
Close() error
}
// v2 provides v2 version of the dfdaemon grpc function.
type v2 struct {
dfdaemonv2.DfdaemonClient
dfdaemonv2.DfdaemonUploadClient
*grpc.ClientConn
}
// Trigger client to download file.
func (v *v2) SyncPieces(ctx context.Context, req *dfdaemonv2.DownloadPieceRequest, opts ...grpc.CallOption) (*dfdaemonv2.DownloadPieceResponse, error) {
// SyncPieces syncs pieces from the other peers.
func (v *v2) SyncPieces(ctx context.Context, req *dfdaemonv2.SyncPiecesRequest, opts ...grpc.CallOption) (dfdaemonv2.DfdaemonUpload_SyncPiecesClient, error) {
ctx, cancel := context.WithTimeout(ctx, contextTimeout)
defer cancel()
return v.DfdaemonClient.DownloadPiece(
return v.DfdaemonUploadClient.SyncPieces(
ctx,
req,
opts...,
)
}
// DownloadTask downloads task back-to-source.
// DownloadPiece downloads piece from the other peer.
func (v *v2) DownloadPiece(ctx context.Context, req *dfdaemonv2.DownloadPieceRequest, opts ...grpc.CallOption) (*dfdaemonv2.DownloadPieceResponse, error) {
ctx, cancel := context.WithTimeout(ctx, contextTimeout)
defer cancel()
return v.DfdaemonUploadClient.DownloadPiece(
ctx,
req,
opts...,
)
}
// DownloadTask downloads task from the other peer.
func (v *v2) DownloadTask(ctx context.Context, req *dfdaemonv2.DownloadTaskRequest, opts ...grpc.CallOption) error {
ctx, cancel := context.WithTimeout(ctx, contextTimeout)
defer cancel()
_, err := v.DfdaemonClient.DownloadTask(
ctx,
req,
opts...,
)
return err
}
// UploadTask uploads task to p2p network.
func (v *v2) UploadTask(ctx context.Context, req *dfdaemonv2.UploadTaskRequest, opts ...grpc.CallOption) error {
ctx, cancel := context.WithTimeout(ctx, contextTimeout)
defer cancel()
_, err := v.DfdaemonClient.UploadTask(
ctx,
req,
opts...,
)
return err
}
// StatTask stats task information.
func (v *v2) StatTask(ctx context.Context, req *dfdaemonv2.StatTaskRequest, opts ...grpc.CallOption) (*commonv2.Task, error) {
ctx, cancel := context.WithTimeout(ctx, contextTimeout)
defer cancel()
return v.DfdaemonClient.StatTask(
ctx,
req,
opts...,
)
}
// DeleteTask deletes task from p2p network.
func (v *v2) DeleteTask(ctx context.Context, req *dfdaemonv2.DeleteTaskRequest, opts ...grpc.CallOption) error {
ctx, cancel := context.WithTimeout(ctx, contextTimeout)
defer cancel()
_, err := v.DfdaemonClient.DeleteTask(
_, err := v.DfdaemonUploadClient.DownloadTask(
ctx,
req,
opts...,

53
pkg/rpc/dfdaemon/client/mocks/client_v2_mock.go
View File

@ -12,7 +12,6 @@ import (
context "context"
reflect "reflect"
common "d7y.io/api/v2/pkg/apis/common/v2"
dfdaemon "d7y.io/api/v2/pkg/apis/dfdaemon/v2"
gomock "go.uber.org/mock/gomock"
grpc "google.golang.org/grpc"
@ -55,25 +54,6 @@ func (mr *MockV2MockRecorder) Close() *gomock.Call {
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Close", reflect.TypeOf((*MockV2)(nil).Close))
}
// DeleteTask mocks base method.
func (m *MockV2) DeleteTask(arg0 context.Context, arg1 *dfdaemon.DeleteTaskRequest, arg2 ...grpc.CallOption) error {
m.ctrl.T.Helper()
varargs := []any{arg0, arg1}
for _, a := range arg2 {
varargs = append(varargs, a)
}
ret := m.ctrl.Call(m, "DeleteTask", varargs...)
ret0, _ := ret[0].(error)
return ret0
}
// DeleteTask indicates an expected call of DeleteTask.
func (mr *MockV2MockRecorder) DeleteTask(arg0, arg1 any, arg2 ...any) *gomock.Call {
mr.mock.ctrl.T.Helper()
varargs := append([]any{arg0, arg1}, arg2...)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DeleteTask", reflect.TypeOf((*MockV2)(nil).DeleteTask), varargs...)
}
// DownloadPiece mocks base method.
func (m *MockV2) DownloadPiece(arg0 context.Context, arg1 *dfdaemon.DownloadPieceRequest, arg2 ...grpc.CallOption) (*dfdaemon.DownloadPieceResponse, error) {
m.ctrl.T.Helper()
@ -113,41 +93,22 @@ func (mr *MockV2MockRecorder) DownloadTask(arg0, arg1 any, arg2 ...any) *gomock.
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DownloadTask", reflect.TypeOf((*MockV2)(nil).DownloadTask), varargs...)
}
// StatTask mocks base method.
func (m *MockV2) StatTask(arg0 context.Context, arg1 *dfdaemon.StatTaskRequest, arg2 ...grpc.CallOption) (*common.Task, error) {
// SyncPieces mocks base method.
func (m *MockV2) SyncPieces(arg0 context.Context, arg1 *dfdaemon.SyncPiecesRequest, arg2 ...grpc.CallOption) (dfdaemon.DfdaemonUpload_SyncPiecesClient, error) {
m.ctrl.T.Helper()
varargs := []any{arg0, arg1}
for _, a := range arg2 {
varargs = append(varargs, a)
}
ret := m.ctrl.Call(m, "StatTask", varargs...)
ret0, _ := ret[0].(*common.Task)
ret := m.ctrl.Call(m, "SyncPieces", varargs...)
ret0, _ := ret[0].(dfdaemon.DfdaemonUpload_SyncPiecesClient)
ret1, _ := ret[1].(error)
return ret0, ret1
}
// StatTask indicates an expected call of StatTask.
func (mr *MockV2MockRecorder) StatTask(arg0, arg1 any, arg2 ...any) *gomock.Call {
// SyncPieces indicates an expected call of SyncPieces.
func (mr *MockV2MockRecorder) SyncPieces(arg0, arg1 any, arg2 ...any) *gomock.Call {
mr.mock.ctrl.T.Helper()
varargs := append([]any{arg0, arg1}, arg2...)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "StatTask", reflect.TypeOf((*MockV2)(nil).StatTask), varargs...)
}
// UploadTask mocks base method.
func (m *MockV2) UploadTask(arg0 context.Context, arg1 *dfdaemon.UploadTaskRequest, arg2 ...grpc.CallOption) error {
m.ctrl.T.Helper()
varargs := []any{arg0, arg1}
for _, a := range arg2 {
varargs = append(varargs, a)
}
ret := m.ctrl.Call(m, "UploadTask", varargs...)
ret0, _ := ret[0].(error)
return ret0
}
// UploadTask indicates an expected call of UploadTask.
func (mr *MockV2MockRecorder) UploadTask(arg0, arg1 any, arg2 ...any) *gomock.Call {
mr.mock.ctrl.T.Helper()
varargs := append([]any{arg0, arg1}, arg2...)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "UploadTask", reflect.TypeOf((*MockV2)(nil).UploadTask), varargs...)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "SyncPieces", reflect.TypeOf((*MockV2)(nil).SyncPieces), varargs...)
}

5
scheduler/resource/seed_peer.go
View File

@ -33,7 +33,6 @@ import (
"d7y.io/dragonfly/v2/pkg/idgen"
"d7y.io/dragonfly/v2/pkg/net/http"
"d7y.io/dragonfly/v2/pkg/rpc/common"
"d7y.io/dragonfly/v2/pkg/types"
"d7y.io/dragonfly/v2/scheduler/config"
"d7y.io/dragonfly/v2/scheduler/metrics"
)
@ -47,7 +46,7 @@ const (
type SeedPeer interface {
// DownloadTask downloads task back-to-source.
// Used only in v2 version of the grpc.
DownloadTask(context.Context, *Task, types.HostType) error
DownloadTask(context.Context, *commonv2.Download) error
// TriggerTask triggers the seed peer to download task.
// Used only in v1 version of the grpc.
@ -88,7 +87,7 @@ func newSeedPeer(cfg *config.ResourceConfig, client SeedPeerClient, peerManager
// TODO Implement DownloadTask
// DownloadTask downloads task back-to-source.
// Used only in v2 version of the grpc.
func (s *seedPeer) DownloadTask(ctx context.Context, task *Task, hostType types.HostType) error {
func (s *seedPeer) DownloadTask(ctx context.Context, download *commonv2.Download) error {
// ctx, cancel := context.WithCancel(trace.ContextWithSpan(context.Background(), trace.SpanFromContext(ctx)))
// defer cancel()

10
scheduler/resource/seed_peer_mock.go
View File

@ -12,9 +12,9 @@ import (
context "context"
reflect "reflect"
common "d7y.io/api/v2/pkg/apis/common/v2"
scheduler "d7y.io/api/v2/pkg/apis/scheduler/v1"
http "d7y.io/dragonfly/v2/pkg/net/http"
types "d7y.io/dragonfly/v2/pkg/types"
gomock "go.uber.org/mock/gomock"
)
@ -56,17 +56,17 @@ func (mr *MockSeedPeerMockRecorder) Client() *gomock.Call {
}
// DownloadTask mocks base method.
func (m *MockSeedPeer) DownloadTask(arg0 context.Context, arg1 *Task, arg2 types.HostType) error {
func (m *MockSeedPeer) DownloadTask(arg0 context.Context, arg1 *common.Download) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "DownloadTask", arg0, arg1, arg2)
ret := m.ctrl.Call(m, "DownloadTask", arg0, arg1)
ret0, _ := ret[0].(error)
return ret0
}
// DownloadTask indicates an expected call of DownloadTask.
func (mr *MockSeedPeerMockRecorder) DownloadTask(arg0, arg1, arg2 any) *gomock.Call {
func (mr *MockSeedPeerMockRecorder) DownloadTask(arg0, arg1 any) *gomock.Call {
mr.mock.ctrl.T.Helper()
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DownloadTask", reflect.TypeOf((*MockSeedPeer)(nil).DownloadTask), arg0, arg1, arg2)
return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "DownloadTask", reflect.TypeOf((*MockSeedPeer)(nil).DownloadTask), arg0, arg1)
}
// Stop mocks base method.

10
scheduler/service/service_v2.go
View File

@ -840,7 +840,7 @@ func (v *V2) handleRegisterPeerRequest(ctx context.Context, stream schedulerv2.S
blocklist := set.NewSafeSet[string]()
blocklist.Add(peer.ID)
if task.FSM.Is(resource.TaskStateFailed) || !task.HasAvailablePeer(blocklist) {
if err := v.downloadTaskBySeedPeer(ctx, peer); err != nil {
if err := v.downloadTaskBySeedPeer(ctx, req.GetDownload(), peer); err != nil {
// Collect RegisterPeerFailureCount metrics.
metrics.RegisterPeerFailureCount.WithLabelValues(priority.String(), peer.Task.Type.String(),
peer.Task.Tag, peer.Task.Application, peer.Host.Type.Name()).Inc()
@ -1302,7 +1302,7 @@ func (v *V2) handleResource(ctx context.Context, stream schedulerv2.Scheduler_An
}
// downloadTaskBySeedPeer downloads task by seed peer.
func (v *V2) downloadTaskBySeedPeer(ctx context.Context, peer *resource.Peer) error {
func (v *V2) downloadTaskBySeedPeer(ctx context.Context, download *commonv2.Download, peer *resource.Peer) error {
// Trigger the first download task based on different priority levels,
// refer to https://github.com/dragonflyoss/api/blob/main/pkg/apis/common/v2/common.proto#L74.
priority := peer.CalculatePriority(v.dynconfig)
@ -1312,7 +1312,7 @@ func (v *V2) downloadTaskBySeedPeer(ctx context.Context, peer *resource.Peer) er
// Super peer is first triggered to download back-to-source.
if v.config.SeedPeer.Enable && !peer.Task.IsSeedPeerFailed() {
go func(ctx context.Context, peer *resource.Peer, hostType types.HostType) {
if err := v.resource.SeedPeer().DownloadTask(context.Background(), peer.Task, hostType); err != nil {
if err := v.resource.SeedPeer().DownloadTask(context.Background(), download); err != nil {
peer.Log.Errorf("%s seed peer downloads task failed %s", hostType.Name(), err.Error())
return
}
@ -1325,7 +1325,7 @@ func (v *V2) downloadTaskBySeedPeer(ctx context.Context, peer *resource.Peer) er
// Strong peer is first triggered to download back-to-source.
if v.config.SeedPeer.Enable && !peer.Task.IsSeedPeerFailed() {
go func(ctx context.Context, peer *resource.Peer, hostType types.HostType) {
if err := v.resource.SeedPeer().DownloadTask(context.Background(), peer.Task, hostType); err != nil {
if err := v.resource.SeedPeer().DownloadTask(context.Background(), download); err != nil {
peer.Log.Errorf("%s seed peer downloads task failed %s", hostType.Name(), err.Error())
return
}
@ -1338,7 +1338,7 @@ func (v *V2) downloadTaskBySeedPeer(ctx context.Context, peer *resource.Peer) er
// Weak peer is first triggered to download back-to-source.
if v.config.SeedPeer.Enable && !peer.Task.IsSeedPeerFailed() {
go func(ctx context.Context, peer *resource.Peer, hostType types.HostType) {
if err := v.resource.SeedPeer().DownloadTask(context.Background(), peer.Task, hostType); err != nil {
if err := v.resource.SeedPeer().DownloadTask(context.Background(), download); err != nil {
peer.Log.Errorf("%s seed peer downloads task failed %s", hostType.Name(), err.Error())
return
}

39
scheduler/service/service_v2_test.go
View File

@ -43,7 +43,6 @@ import (
managertypes "d7y.io/dragonfly/v2/manager/types"
nethttp "d7y.io/dragonfly/v2/pkg/net/http"
"d7y.io/dragonfly/v2/pkg/types"
pkgtypes "d7y.io/dragonfly/v2/pkg/types"
"d7y.io/dragonfly/v2/scheduler/config"
configmocks "d7y.io/dragonfly/v2/scheduler/config/mocks"
@ -3364,13 +3363,13 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any(), types.HostTypeSuperSeed).Do(func(context.Context, *resource.Task, types.HostType) { wg.Done() }).Return(nil).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *commonv2.Download) { wg.Done() }).Return(nil).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL6
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.False(peer.NeedBackToSource.Load())
},
},
@ -3388,13 +3387,13 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any(), types.HostTypeSuperSeed).Do(func(context.Context, *resource.Task, types.HostType) { wg.Done() }).Return(errors.New("foo")).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *commonv2.Download) { wg.Done() }).Return(errors.New("foo")).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL6
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.False(peer.NeedBackToSource.Load())
},
},
@ -3409,7 +3408,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
peer.Priority = commonv2.Priority_LEVEL6
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.True(peer.NeedBackToSource.Load())
},
},
@ -3427,13 +3426,13 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any(), types.HostTypeStrongSeed).Do(func(context.Context, *resource.Task, types.HostType) { wg.Done() }).Return(nil).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *commonv2.Download) { wg.Done() }).Return(nil).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL5
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.False(peer.NeedBackToSource.Load())
},
},
@ -3451,13 +3450,13 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any(), types.HostTypeStrongSeed).Do(func(context.Context, *resource.Task, types.HostType) { wg.Done() }).Return(errors.New("foo")).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *commonv2.Download) { wg.Done() }).Return(errors.New("foo")).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL5
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.False(peer.NeedBackToSource.Load())
},
},
@ -3472,7 +3471,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
peer.Priority = commonv2.Priority_LEVEL5
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.True(peer.NeedBackToSource.Load())
},
},
@ -3490,13 +3489,13 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any(), types.HostTypeWeakSeed).Do(func(context.Context, *resource.Task, types.HostType) { wg.Done() }).Return(nil).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *commonv2.Download) { wg.Done() }).Return(nil).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL4
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.False(peer.NeedBackToSource.Load())
},
},
@ -3514,13 +3513,13 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any(), types.HostTypeWeakSeed).Do(func(context.Context, *resource.Task, types.HostType) { wg.Done() }).Return(errors.New("foo")).Times(1),
ms.DownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *commonv2.Download) { wg.Done() }).Return(errors.New("foo")).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL4
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.False(peer.NeedBackToSource.Load())
},
},
@ -3535,7 +3534,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
peer.Priority = commonv2.Priority_LEVEL4
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.True(peer.NeedBackToSource.Load())
},
},
@ -3550,7 +3549,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
peer.Priority = commonv2.Priority_LEVEL3
assert := assert.New(t)
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), peer))
assert.NoError(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer))
assert.True(peer.NeedBackToSource.Load())
},
},
@ -3565,7 +3564,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
peer.Priority = commonv2.Priority_LEVEL2
assert := assert.New(t)
assert.ErrorIs(svc.downloadTaskBySeedPeer(context.Background(), peer), status.Errorf(codes.NotFound, "%s peer not found candidate peers", commonv2.Priority_LEVEL2.String()))
assert.ErrorIs(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer), status.Errorf(codes.NotFound, "%s peer not found candidate peers", commonv2.Priority_LEVEL2.String()))
},
},
{
@ -3579,7 +3578,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
peer.Priority = commonv2.Priority_LEVEL1
assert := assert.New(t)
assert.ErrorIs(svc.downloadTaskBySeedPeer(context.Background(), peer), status.Errorf(codes.FailedPrecondition, "%s peer is forbidden", commonv2.Priority_LEVEL1.String()))
assert.ErrorIs(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer), status.Errorf(codes.FailedPrecondition, "%s peer is forbidden", commonv2.Priority_LEVEL1.String()))
},
},
{
@ -3593,7 +3592,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
peer.Priority = commonv2.Priority(100)
assert := assert.New(t)
assert.ErrorIs(svc.downloadTaskBySeedPeer(context.Background(), peer), status.Errorf(codes.InvalidArgument, "invalid priority %#v", peer.Priority))
assert.ErrorIs(svc.downloadTaskBySeedPeer(context.Background(), &commonv2.Download{}, peer), status.Errorf(codes.InvalidArgument, "invalid priority %#v", peer.Priority))
},
},
}