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

refactor: scheduler end and begin of piece (#1189) 

Signed-off-by: Gaius <gaius.qi@gmail.com>
This commit is contained in:
Gaius 2022-03-25 14:01:37 +08:00
parent 9d41cee1ec
commit 1939ceff8f
No known key found for this signature in database
GPG Key ID: 8B4E5D1290FA2FFB
3 changed files with 11 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
client/daemon/peer/peertask_conductor.go
View File

@ -1315,7 +1315,7 @@ func (pt *peerTaskConductor) done() {
// send EOF piece result to scheduler // send EOF piece result to scheduler
err := pt.peerPacketStream.Send( err := pt.peerPacketStream.Send(
scheduler.NewEndPieceResult(pt.taskID, pt.peerID, pt.readyPieces.Settled())) schedulerclient.NewEndOfPiece(pt.taskID, pt.peerID, pt.readyPieces.Settled()))
pt.Debugf("end piece result sent: %v, peer task finished", err) pt.Debugf("end piece result sent: %v, peer task finished", err)
err = pt.schedulerClient.ReportPeerResult( err = pt.schedulerClient.ReportPeerResult(
@ -1362,7 +1362,7 @@ func (pt *peerTaskConductor) fail() {
// send EOF piece result to scheduler // send EOF piece result to scheduler
err := pt.peerPacketStream.Send( err := pt.peerPacketStream.Send(
scheduler.NewEndPieceResult(pt.taskID, pt.peerID, pt.readyPieces.Settled())) schedulerclient.NewEndOfPiece(pt.taskID, pt.peerID, pt.readyPieces.Settled()))
pt.Debugf("end piece result sent: %v, peer task finished", err) pt.Debugf("end piece result sent: %v, peer task finished", err)
ctx := trace.ContextWithSpan(context.Background(), trace.SpanFromContext(pt.ctx)) ctx := trace.ContextWithSpan(context.Background(), trace.SpanFromContext(pt.ctx))

2
pkg/rpc/scheduler/client/client.go
View File

@ -158,7 +158,7 @@ func (sc *schedulerClient) ReportPieceResult(ctx context.Context, taskID string,
logger.With("peerId", ptr.PeerId, "errMsg", err).Infof("start to report piece result for taskID: %s", taskID) logger.With("peerId", ptr.PeerId, "errMsg", err).Infof("start to report piece result for taskID: %s", taskID)
// trigger scheduling // trigger scheduling
return pps, pps.Send(scheduler.NewZeroPieceResult(taskID, ptr.PeerId)) return pps, pps.Send(NewBeginOfPiece(taskID, ptr.PeerId))
} }
func (sc *schedulerClient) ReportPeerResult(ctx context.Context, pr *scheduler.PeerResult, opts ...grpc.CallOption) error { func (sc *schedulerClient) ReportPeerResult(ctx context.Context, pr *scheduler.PeerResult, opts ...grpc.CallOption) error {

25
pkg/rpc/scheduler/scheduler.go → pkg/rpc/scheduler/client/piece.go
View File

@ -14,40 +14,31 @@
* limitations under the License. * limitations under the License.
*/ */
package scheduler package client
import ( import (
"d7y.io/dragonfly/v2/pkg/rpc/base" "d7y.io/dragonfly/v2/pkg/rpc/base"
"d7y.io/dragonfly/v2/pkg/rpc/base/common" "d7y.io/dragonfly/v2/pkg/rpc/base/common"
"d7y.io/dragonfly/v2/pkg/rpc/scheduler"
) )
func NewZeroPieceResult(taskID, peerID string) *PieceResult { func NewBeginOfPiece(taskID, peerID string) *scheduler.PieceResult {
return &PieceResult{ return &scheduler.PieceResult{
TaskId: taskID, TaskId: taskID,
SrcPid: peerID, SrcPid: peerID,
PieceInfo: &base.PieceInfo{ PieceInfo: &base.PieceInfo{
PieceNum: common.BeginOfPiece, PieceNum: common.BeginOfPiece,
RangeStart: 0,
RangeSize: 0,
PieceMd5: "",
PieceOffset: 0,
PieceStyle: 0,
}, },
} }
} }
func NewEndPieceResult(taskID, peerID string, finishedCount int32) *PieceResult { func NewEndOfPiece(taskID, peerID string, finishedCount int32) *scheduler.PieceResult {
return &PieceResult{ return &scheduler.PieceResult{
TaskId: taskID, TaskId: taskID,
SrcPid: peerID, SrcPid: peerID,
FinishedCount: finishedCount, FinishedCount: finishedCount,
PieceInfo: &base.PieceInfo{ PieceInfo: &base.PieceInfo{
PieceNum: common.EndOfPiece, PieceNum: common.EndOfPiece,
RangeStart: 0,
RangeSize: 0,
PieceMd5: "",
PieceOffset: 0,
PieceStyle: 0,
}, },
} }
} }