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

feat: optimize trigger download task return empty (#2958) 

Signed-off-by: Gaius <gaius.qi@gmail.com>
This commit is contained in:
Gaius 2023-12-19 18:41:32 +08:00 committed by GitHub
parent ec66fd1229
commit 0c591e46a3
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
10 changed files with 36 additions and 45 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.67
d7y.io/api/v2 v2.0.70
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.67 h1:4fiGXT1WHWgRXSTmnP53MU83Zbf+7i1jYeGNEJWrM7Q=
d7y.io/api/v2 v2.0.67/go.mod h1:ve0R4ePgRYZVdnVyhWTOi2LdP3/gyf21ZwP2xij+3Io=
d7y.io/api/v2 v2.0.70 h1:qZ1HslBwgI24VlgtyA1K3GpS3Mm7wWtgMujtcMCl2TY=
d7y.io/api/v2 v2.0.70/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=

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

@ -87,7 +87,7 @@ type V2 interface {
DownloadPiece(context.Context, *dfdaemonv2.DownloadPieceRequest, ...grpc.CallOption) (*dfdaemonv2.DownloadPieceResponse, error)
// TriggerDownloadTask triggers download task from the other peer.
TriggerDownloadTask(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest, ...grpc.CallOption) (*dfdaemonv2.TriggerDownloadTaskResponse, error)
TriggerDownloadTask(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest, ...grpc.CallOption) error
// Close tears down the ClientConn and all underlying connections.
Close() error
@ -125,13 +125,14 @@ func (v *v2) DownloadPiece(ctx context.Context, req *dfdaemonv2.DownloadPieceReq
}
// TriggerDownloadTask triggers download task from the other peer.
func (v *v2) TriggerDownloadTask(ctx context.Context, req *dfdaemonv2.TriggerDownloadTaskRequest, opts ...grpc.CallOption) (*dfdaemonv2.TriggerDownloadTaskResponse, error) {
func (v *v2) TriggerDownloadTask(ctx context.Context, req *dfdaemonv2.TriggerDownloadTaskRequest, opts ...grpc.CallOption) error {
ctx, cancel := context.WithTimeout(ctx, contextTimeout)
defer cancel()
return v.DfdaemonUploadClient.TriggerDownloadTask(
_, err := v.DfdaemonUploadClient.TriggerDownloadTask(
ctx,
req,
opts...,
)
return err
}

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

@ -95,16 +95,15 @@ func (mr *MockV2MockRecorder) SyncPieces(arg0, arg1 any, arg2 ...any) *gomock.Ca
}
// TriggerDownloadTask mocks base method.
func (m *MockV2) TriggerDownloadTask(arg0 context.Context, arg1 *dfdaemon.TriggerDownloadTaskRequest, arg2 ...grpc.CallOption) (*dfdaemon.TriggerDownloadTaskResponse, error) {
func (m *MockV2) TriggerDownloadTask(arg0 context.Context, arg1 *dfdaemon.TriggerDownloadTaskRequest, 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, "TriggerDownloadTask", varargs...)
ret0, _ := ret[0].(*dfdaemon.TriggerDownloadTaskResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
ret0, _ := ret[0].(error)
return ret0
}
// TriggerDownloadTask indicates an expected call of TriggerDownloadTask.

4
scheduler/resource/seed_peer.go
View File

@ -49,7 +49,7 @@ const (
type SeedPeer interface {
// TriggerDownloadTask triggers the seed peer to download task.
// Used only in v2 version of the grpc.
TriggerDownloadTask(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) (*dfdaemonv2.TriggerDownloadTaskResponse, error)
TriggerDownloadTask(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) error
// TriggerTask triggers the seed peer to download task.
// Used only in v1 version of the grpc.
@ -89,7 +89,7 @@ func newSeedPeer(cfg *config.Config, client SeedPeerClient, peerManager PeerMana
// TriggerDownloadTask triggers the seed peer to download task.
// Used only in v2 version of the grpc.
func (s *seedPeer) TriggerDownloadTask(ctx context.Context, req *dfdaemonv2.TriggerDownloadTaskRequest) (*dfdaemonv2.TriggerDownloadTaskResponse, error) {
func (s *seedPeer) TriggerDownloadTask(ctx context.Context, req *dfdaemonv2.TriggerDownloadTaskRequest) error {
ctx, cancel := context.WithCancel(trace.ContextWithSpan(context.Background(), trace.SpanFromContext(ctx)))
defer cancel()

7
scheduler/resource/seed_peer_client_mock.go
View File

@ -184,16 +184,15 @@ func (mr *MockSeedPeerClientMockRecorder) SyncPieces(arg0, arg1 any, arg2 ...any
}
// TriggerDownloadTask mocks base method.
func (m *MockSeedPeerClient) TriggerDownloadTask(arg0 context.Context, arg1 *dfdaemon.TriggerDownloadTaskRequest, arg2 ...grpc.CallOption) (*dfdaemon.TriggerDownloadTaskResponse, error) {
func (m *MockSeedPeerClient) TriggerDownloadTask(arg0 context.Context, arg1 *dfdaemon.TriggerDownloadTaskRequest, 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, "TriggerDownloadTask", varargs...)
ret0, _ := ret[0].(*dfdaemon.TriggerDownloadTaskResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
ret0, _ := ret[0].(error)
return ret0
}
// TriggerDownloadTask indicates an expected call of TriggerDownloadTask.

7
scheduler/resource/seed_peer_mock.go
View File

@ -70,12 +70,11 @@ func (mr *MockSeedPeerMockRecorder) Stop() *gomock.Call {
}
// TriggerDownloadTask mocks base method.
func (m *MockSeedPeer) TriggerDownloadTask(arg0 context.Context, arg1 *dfdaemon.TriggerDownloadTaskRequest) (*dfdaemon.TriggerDownloadTaskResponse, error) {
func (m *MockSeedPeer) TriggerDownloadTask(arg0 context.Context, arg1 *dfdaemon.TriggerDownloadTaskRequest) error {
m.ctrl.T.Helper()
ret := m.ctrl.Call(m, "TriggerDownloadTask", arg0, arg1)
ret0, _ := ret[0].(*dfdaemon.TriggerDownloadTaskResponse)
ret1, _ := ret[1].(error)
return ret0, ret1
ret0, _ := ret[0].(error)
return ret0
}
// TriggerDownloadTask indicates an expected call of TriggerDownloadTask.

16
scheduler/resource/seed_peer_test.go
View File

@ -61,14 +61,14 @@ func TestSeedPeer_TriggerDownloadTask(t *testing.T) {
tests := []struct {
name string
mock func(mc *MockSeedPeerClientMockRecorder)
expect func(t *testing.T, resp *dfdaemonv2.TriggerDownloadTaskResponse, err error)
expect func(t *testing.T, err error)
}{
{
name: "trigger download task failed",
mock: func(mc *MockSeedPeerClientMockRecorder) {
mc.TriggerDownloadTask(gomock.Any(), gomock.Any()).Return(nil, errors.New("foo")).Times(1)
mc.TriggerDownloadTask(gomock.Any(), gomock.Any()).Return(errors.New("foo")).Times(1)
},
expect: func(t *testing.T, resp *dfdaemonv2.TriggerDownloadTaskResponse, err error) {
expect: func(t *testing.T, err error) {
assert := assert.New(t)
assert.EqualError(err, "foo")
},
@ -76,14 +76,11 @@ func TestSeedPeer_TriggerDownloadTask(t *testing.T) {
{
name: "trigger download task scuccess",
mock: func(mc *MockSeedPeerClientMockRecorder) {
mc.TriggerDownloadTask(gomock.Any(), gomock.Any()).Return(&dfdaemonv2.TriggerDownloadTaskResponse{HostId: mockHostID, TaskId: mockTaskID, PeerId: mockPeerID}, nil).Times(1)
mc.TriggerDownloadTask(gomock.Any(), gomock.Any()).Return(nil).Times(1)
},
expect: func(t *testing.T, resp *dfdaemonv2.TriggerDownloadTaskResponse, err error) {
expect: func(t *testing.T, err error) {
assert := assert.New(t)
assert.NoError(err)
assert.Equal(mockHostID, resp.HostId)
assert.Equal(mockTaskID, resp.TaskId)
assert.Equal(mockPeerID, resp.PeerId)
},
},
}
@ -98,8 +95,7 @@ func TestSeedPeer_TriggerDownloadTask(t *testing.T) {
tc.mock(client.EXPECT())
seedPeer := newSeedPeer(mockConfig, client, peerManager, hostManager)
resp, err := seedPeer.TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{})
tc.expect(t, resp, err)
tc.expect(t, seedPeer.TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{}))
})
}
}

15
scheduler/service/service_v2.go
View File

@ -1313,13 +1313,12 @@ func (v *V2) downloadTaskBySeedPeer(ctx context.Context, download *commonv2.Down
// Super peer is first triggered to download back-to-source.
if v.config.SeedPeer.Enable && !peer.Task.IsSeedPeerFailed() {
go func(ctx context.Context, download *commonv2.Download, hostType types.HostType) {
resp, err := v.resource.SeedPeer().TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{Download: download})
if err != nil {
if err := v.resource.SeedPeer().TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{Download: download}); err != nil {
peer.Log.Errorf("%s seed peer triggers download task failed %s", hostType.Name(), err.Error())
return
}
peer.Log.Infof("%s seed peer triggers download task success, hostID: %s, peerID: %s", hostType.Name(), resp.GetHostId(), resp.GetPeerId())
peer.Log.Infof("%s seed peer triggers download task success", hostType.Name())
}(ctx, download, types.HostTypeSuperSeed)
break
@ -1330,13 +1329,12 @@ func (v *V2) downloadTaskBySeedPeer(ctx context.Context, download *commonv2.Down
// Strong peer is first triggered to download back-to-source.
if v.config.SeedPeer.Enable && !peer.Task.IsSeedPeerFailed() {
go func(ctx context.Context, download *commonv2.Download, hostType types.HostType) {
resp, err := v.resource.SeedPeer().TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{Download: download})
if err != nil {
if err := v.resource.SeedPeer().TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{Download: download}); err != nil {
peer.Log.Errorf("%s seed peer triggers download task failed %s", hostType.Name(), err.Error())
return
}
peer.Log.Infof("%s seed peer triggers download task success, hostID: %s, peerID: %s", hostType.Name(), resp.GetHostId(), resp.GetPeerId())
peer.Log.Infof("%s seed peer triggers download task success", hostType.Name())
}(ctx, download, types.HostTypeSuperSeed)
break
@ -1347,13 +1345,12 @@ func (v *V2) downloadTaskBySeedPeer(ctx context.Context, download *commonv2.Down
// Weak peer is first triggered to download back-to-source.
if v.config.SeedPeer.Enable && !peer.Task.IsSeedPeerFailed() {
go func(ctx context.Context, download *commonv2.Download, hostType types.HostType) {
resp, err := v.resource.SeedPeer().TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{Download: download})
if err != nil {
if err := v.resource.SeedPeer().TriggerDownloadTask(context.Background(), &dfdaemonv2.TriggerDownloadTaskRequest{Download: download}); err != nil {
peer.Log.Errorf("%s seed peer triggers download task failed %s", hostType.Name(), err.Error())
return
}
peer.Log.Infof("%s seed peer triggers download task success, hostID: %s, peerID: %s", hostType.Name(), resp.GetHostId(), resp.GetPeerId())
peer.Log.Infof("%s seed peer triggers download task success", hostType.Name())
}(ctx, download, types.HostTypeSuperSeed)
break

12
scheduler/service/service_v2_test.go
View File

@ -3364,7 +3364,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(&dfdaemonv2.TriggerDownloadTaskResponse{}, nil).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(nil).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL6
@ -3388,7 +3388,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(nil, errors.New("foo")).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(errors.New("foo")).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL6
@ -3427,7 +3427,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(&dfdaemonv2.TriggerDownloadTaskResponse{}, nil).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(nil).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL5
@ -3451,7 +3451,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(nil, errors.New("foo")).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(errors.New("foo")).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL5
@ -3490,7 +3490,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(&dfdaemonv2.TriggerDownloadTaskResponse{}, nil).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(nil).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL4
@ -3514,7 +3514,7 @@ func TestServiceV2_downloadTaskBySeedPeer(t *testing.T) {
gomock.InOrder(
mr.SeedPeer().Return(seedPeerClient).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(nil, errors.New("foo")).Times(1),
ms.TriggerDownloadTask(gomock.All(), gomock.Any()).Do(func(context.Context, *dfdaemonv2.TriggerDownloadTaskRequest) { wg.Done() }).Return(errors.New("foo")).Times(1),
)
peer.Priority = commonv2.Priority_LEVEL4